Merge branch 'feature/RES-844-fix-tracing' into 'master'

RES-844: fix opentelemtry tracing

See merge request knecon/research/pyinfra!98

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):
@@ -37,7 +38,7 @@ class JsonSpanExporter(SpanExporter):
 def setup_trace(settings: Dynaconf, service_name: str = None, exporter: SpanExporter = None):
     tracing_type = settings.tracing.type
     if tracing_type == "azure_monitor":
-        # Configure OpenTelemetry to use Azure Monitor with the 
+        # Configure OpenTelemetry to use Azure Monitor with the
         # APPLICATIONINSIGHTS_CONNECTION_STRING environment variable.
         try:
             configure_azure_monitor()
@@ -84,8 +85,11 @@ def get_exporter(settings: Dynaconf):
         )
 
 
-def instrument_pika():
-    PikaInstrumentor().instrument()
+def instrument_pika(dynamic_queues: bool):
+    if dynamic_queues:
+        AioPikaInstrumentor().instrument()
+    else:
+        PikaInstrumentor().instrument()
 
 
 def instrument_app(app: FastAPI, excluded_urls: str = "/health,/ready,/prometheus"):

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyinfra"
-version = "3.2.8"
+version = "3.2.9"
 description = ""
 authors = ["Team Research <research@knecon.com>"]
 license = "All rights reseverd"
@@ -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"