feat: add AioPikaInstrumentor

pyinfra/examples.py

@@ -89,7 +89,7 @@ def start_standard_queue_consumer(
     if settings.tracing.enabled:
         setup_trace(settings)
 
-        instrument_pika()
+        instrument_pika(dynamic_queues=settings.dynamic_tenant_queues.enabled)
         instrument_app(app)
 
     if settings.dynamic_tenant_queues.enabled:

pyinfra/utils/opentelemetry.py

@@ -3,8 +3,10 @@ import json
 from azure.monitor.opentelemetry import configure_azure_monitor
 from dynaconf import Dynaconf
 from fastapi import FastAPI
+from kn_utils.logging import logger
 from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
+from opentelemetry.instrumentation.aio_pika import AioPikaInstrumentor
 from opentelemetry.instrumentation.fastapi import FastAPIInstrumentor
 from opentelemetry.instrumentation.pika import PikaInstrumentor
 from opentelemetry.sdk.resources import Resource
@@ -18,7 +20,6 @@ from opentelemetry.sdk.trace.export import (
 
 from pyinfra.config.loader import validate_settings
 from pyinfra.config.validators import opentelemetry_validators
-from kn_utils.logging import logger
 
 
 class JsonSpanExporter(SpanExporter):
@@ -84,8 +85,11 @@ def get_exporter(settings: Dynaconf):
         )
 
 
-def instrument_pika():
+def instrument_pika(dynamic_queues: bool):
-    PikaInstrumentor().instrument()
+    if dynamic_queues:
+        AioPikaInstrumentor().instrument()
+    else:
+        PikaInstrumentor().instrument()
 
 
 def instrument_app(app: FastAPI, excluded_urls: str = "/health,/ready,/prometheus"):

pyproject.toml

@@ -38,6 +38,7 @@ protobuf = "^3.20"
 aio-pika = "^9.4.2"
 aiohttp = "^3.9.5"
 tenacity = "^8.5.0"
+opentelemetry-instrumentation-aio-pika = "0.46b0"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^7"