🚨✨ Implement tracing sampling strategy (🚧 devops 🚧) (#8421)

Co-authored-by: Dustin Kaiser <[email protected]>

force merge after tests are green to use beneficial time window

Author: bisgaard-itis

.env-devel

@@ -377,7 +377,7 @@ TRACING_OPENTELEMETRY_COLLECTOR_BATCH_SIZE=2
 TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT=http://opentelemetry-collector
 TRACING_OPENTELEMETRY_COLLECTOR_EXPORTER_ENDPOINT=http://jaeger:4318
 TRACING_OPENTELEMETRY_COLLECTOR_PORT=4318
-TRACING_OPENTELEMETRY_COLLECTOR_SAMPLING_PERCENTAGE=100
+TRACING_OPENTELEMETRY_SAMPLING_PROBABILITY=1.0
 TRAEFIK_SIMCORE_ZONE=internal_simcore_stack
 TRASH_RETENTION_DAYS=7
 TWILIO_ACCOUNT_SID=DUMMY

packages/service-library/src/servicelib/aiohttp/tracing.py

@@ -1,6 +1,7 @@
 """Adds aiohttp middleware for tracing using opentelemetry instrumentation."""
 
 import logging
+import time
 from collections.abc import AsyncIterator, Callable
 from typing import Final
 
@@ -13,18 +14,28 @@
     AioHttpClientInstrumentor,
 )
 from opentelemetry.instrumentation.aiohttp_server import (
-    middleware as aiohttp_server_opentelemetry_middleware,  # pylint:disable=no-name-in-module
+    _parse_active_request_count_attrs,
+    _parse_duration_attrs,
+    collect_request_attributes,
+    get_default_span_details,
+    getter,
+    meter,
+    set_status_code,
 )
-from opentelemetry.sdk.resources import Resource
+from opentelemetry.propagate import extract
 from opentelemetry.sdk.trace import SpanProcessor, TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
+from opentelemetry.semconv.metrics import MetricInstruments
 from settings_library.tracing import TracingSettings
 from yarl import URL
 
 from ..logging_utils import log_context
-from ..tracing import get_trace_id_header
+from ..tracing import TracingConfig, get_trace_id_header
 
 _logger = logging.getLogger(__name__)
+
+TRACING_CONFIG_KEY: Final[str] = "tracing_config"
+
 try:
     from opentelemetry.instrumentation.botocore import (  # type: ignore[import-not-found]
         BotocoreInstrumentor,
@@ -65,6 +76,57 @@
 )
 
 
+@web.middleware
+async def aiohttp_server_opentelemetry_middleware(request: web.Request, handler):
+    """This middleware is extracted from https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/instrumentation/opentelemetry-instrumentation-aiohttp-server/src/opentelemetry/instrumentation/aiohttp_server/__init__.py
+    and adapted to allow passing the tracer provider via the app instead of using the global object. The original code for the function is licensed under https://github.com/open-telemetry/opentelemetry-python-contrib/blob/main/LICENSE.
+    FIXME: I have recorded this limitation in the official source here: https://github.com/open-telemetry/opentelemetry-python-contrib/issues/3801 and plan on providing a fix soon.
+    """
+
+    span_name, additional_attributes = get_default_span_details(request)
+
+    req_attrs = collect_request_attributes(request)
+    duration_attrs = _parse_duration_attrs(req_attrs)
+    active_requests_count_attrs = _parse_active_request_count_attrs(req_attrs)
+
+    duration_histogram = meter.create_histogram(
+        name=MetricInstruments.HTTP_SERVER_DURATION,
+        unit="ms",
+        description="Measures the duration of inbound HTTP requests.",
+    )
+
+    active_requests_counter = meter.create_up_down_counter(
+        name=MetricInstruments.HTTP_SERVER_ACTIVE_REQUESTS,
+        unit="requests",
+        description="measures the number of concurrent HTTP requests those are currently in flight",
+    )
+    tracing_config = request.app[TRACING_CONFIG_KEY]
+    assert isinstance(tracing_config, TracingConfig)  # nosec
+    assert tracing_config.tracer_provider  # nosec
+    tracer = tracing_config.tracer_provider.get_tracer(__name__)
+    with tracer.start_as_current_span(
+        span_name,
+        context=extract(request, getter=getter),
+        kind=trace.SpanKind.SERVER,
+    ) as span:
+        attributes = collect_request_attributes(request)
+        attributes.update(additional_attributes)
+        span.set_attributes(attributes)
+        start = time.perf_counter()
+        active_requests_counter.add(1, active_requests_count_attrs)
+        try:
+            resp = await handler(request)
+            set_status_code(span, resp.status)
+        except web.HTTPException as ex:
+            set_status_code(span, ex.status_code)
+            raise
+        finally:
+            duration = max((time.perf_counter() - start) * 1000, 0)
+            duration_histogram.record(duration, duration_attrs)
+            active_requests_counter.add(-1, active_requests_count_attrs)
+        return resp
+
+
 def _create_span_processor(tracing_destination: str) -> SpanProcessor:
     otlp_exporter = OTLPSpanExporterHTTP(
         endpoint=tracing_destination,
@@ -77,12 +139,12 @@ def _startup(
     app: web.Application,
     tracing_settings: TracingSettings,
     service_name: str,
+    tracer_provider: TracerProvider,
     add_response_trace_id_header: bool = False,
 ) -> None:
     """
     Sets up this service for a distributed tracing system (opentelemetry)
     """
-    _ = app
     opentelemetry_collector_endpoint = (
         f"{tracing_settings.TRACING_OPENTELEMETRY_COLLECTOR_ENDPOINT}"
     )
@@ -99,9 +161,6 @@ def _startup(
             "unset. Provide both or remove both."
         )
         raise RuntimeError(msg)
-    resource = Resource(attributes={"service.name": service_name})
-    trace.set_tracer_provider(TracerProvider(resource=resource))
-    tracer_provider: trace.TracerProvider = trace.get_tracer_provider()
 
     tracing_destination: str = (
         f"{URL(opentelemetry_collector_endpoint).with_port(opentelemetry_collector_port).with_path('/v1/traces')}"
@@ -114,7 +173,7 @@ def _startup(
     )
 
     # Add the span processor to the tracer provider
-    tracer_provider.add_span_processor(_create_span_processor(tracing_destination))  # type: ignore[attr-defined] # https://github.com/open-telemetry/opentelemetry-python/issues/3713
+    tracer_provider.add_span_processor(_create_span_processor(tracing_destination))
     # Instrument aiohttp server
     # Explanation for custom middleware call DK 10/2024:
     # OpenTelemetry Aiohttp autoinstrumentation is meant to be used by only calling `AioHttpServerInstrumentor().instrument()`
@@ -135,43 +194,43 @@ def _startup(
     # - opentelemetry-instrumentation==0.48b0
 
     # Instrument aiohttp client
-    AioHttpClientInstrumentor().instrument()
+    AioHttpClientInstrumentor().instrument(tracer_provider=tracer_provider)
     if HAS_AIOPG:
         with log_context(
             _logger,
             logging.INFO,
             msg="Attempting to add aio-pg opentelemetry autoinstrumentation...",
         ):
-            AiopgInstrumentor().instrument()
+            AiopgInstrumentor().instrument(tracer_provider=tracer_provider)
     if HAS_ASYNCPG:
         with log_context(
             _logger,
             logging.INFO,
             msg="Attempting to add asyncpg opentelemetry autoinstrumentation...",
         ):
-            AsyncPGInstrumentor().instrument()
+            AsyncPGInstrumentor().instrument(tracer_provider=tracer_provider)
     if HAS_BOTOCORE:
         with log_context(
             _logger,
             logging.INFO,
             msg="Attempting to add botocore opentelemetry autoinstrumentation...",
         ):
-            BotocoreInstrumentor().instrument()
+            BotocoreInstrumentor().instrument(tracer_provider=tracer_provider)
     if HAS_REQUESTS:
         with log_context(
             _logger,
             logging.INFO,
             msg="Attempting to add requests opentelemetry autoinstrumentation...",
         ):
-            RequestsInstrumentor().instrument()
+            RequestsInstrumentor().instrument(tracer_provider=tracer_provider)
 
     if HAS_AIO_PIKA:
         with log_context(
             _logger,
             logging.INFO,
             msg="Attempting to add aio_pika opentelemetry autoinstrumentation...",
         ):
-            AioPikaInstrumentor().instrument()
+            AioPikaInstrumentor().instrument(tracer_provider=tracer_provider)
 
 
 @web.middleware
@@ -222,17 +281,24 @@ def _shutdown() -> None:
             _logger.exception("Failed to uninstrument AioPikaInstrumentor")
 
 
-def get_tracing_lifespan(
+def setup_tracing(
     *,
     app: web.Application,
-    tracing_settings: TracingSettings,
-    service_name: str,
+    tracing_config: TracingConfig,
     add_response_trace_id_header: bool = False,
 ) -> Callable[[web.Application], AsyncIterator]:
+
+    if tracing_config.tracing_enabled is False:
+        msg = "Tracing is not enabled"
+        raise ValueError(msg)
+    assert tracing_config.tracer_provider  # nosec
+    assert tracing_config.tracing_settings  # nosec
+
     _startup(
         app=app,
-        tracing_settings=tracing_settings,
-        service_name=service_name,
+        tracing_settings=tracing_config.tracing_settings,
+        tracer_provider=tracing_config.tracer_provider,
+        service_name=tracing_config.service_name,
         add_response_trace_id_header=add_response_trace_id_header,
     )
 

packages/service-library/src/servicelib/fastapi/client_session.py

@@ -2,7 +2,7 @@
 
 import httpx
 from fastapi import FastAPI
-from settings_library.tracing import TracingSettings
+from servicelib.tracing import TracingConfig
 
 from .tracing import setup_httpx_client_tracing
 
@@ -12,16 +12,16 @@ def setup_client_session(
     *,
     default_timeout: datetime.timedelta = datetime.timedelta(seconds=20),
     max_keepalive_connections: int = 20,
-    tracing_settings: TracingSettings | None,
+    tracing_config: TracingConfig | None
 ) -> None:
     async def on_startup() -> None:
         session = httpx.AsyncClient(
             transport=httpx.AsyncHTTPTransport(http2=True),
             limits=httpx.Limits(max_keepalive_connections=max_keepalive_connections),
             timeout=default_timeout.total_seconds(),
         )
-        if tracing_settings:
-            setup_httpx_client_tracing(session)
+        if tracing_config:
+            setup_httpx_client_tracing(session, tracing_config=tracing_config)
         app.state.aiohttp_client_session = session
 
     async def on_shutdown() -> None:

packages/service-library/src/servicelib/fastapi/http_client_thin.py

@@ -8,7 +8,7 @@
 from common_library.errors_classes import OsparcErrorMixin
 from httpx import AsyncClient, ConnectError, HTTPError, PoolTimeout, Response
 from httpx._types import TimeoutTypes, URLTypes
-from settings_library.tracing import TracingSettings
+from servicelib.tracing import TracingConfig
 from tenacity import RetryCallState
 from tenacity.asyncio import AsyncRetrying
 from tenacity.before_sleep import before_sleep_log
@@ -200,7 +200,7 @@ def __init__(
         self,
         *,
         total_retry_interval: float,
-        tracing_settings: TracingSettings | None,
+        tracing_config: TracingConfig,
         base_url: URLTypes | None = None,
         default_http_client_timeout: TimeoutTypes | None = None,
         extra_allowed_method_names: set[str] | None = None,
@@ -224,8 +224,8 @@ def __init__(
             client_args["timeout"] = default_http_client_timeout
 
         client = AsyncClient(**client_args)
-        if tracing_settings:
-            setup_httpx_client_tracing(client)
+        if tracing_config.tracing_enabled:
+            setup_httpx_client_tracing(client, tracing_config=tracing_config)
         super().__init__(client=client)
 
     async def __aenter__(self):

packages/service-library/src/servicelib/fastapi/logging_lifespan.py

@@ -4,7 +4,7 @@
 
 from common_library.logging.logging_utils_filtering import LoggerName, MessageSubstring
 from fastapi import FastAPI
-from settings_library.tracing import TracingSettings
+from servicelib.tracing import TracingConfig
 
 from ..logging_utils import (
     LogLevelInt,
@@ -20,7 +20,7 @@ def create_logging_lifespan(
     *,
     log_format_local_dev_enabled: bool,
     logger_filter_mapping: dict[LoggerName, list[MessageSubstring]],
-    tracing_settings: TracingSettings | None,
+    tracing_config: TracingConfig,
     log_base_level: LogLevelInt,
     noisy_loggers: tuple[str, ...] | None,
 ) -> Lifespan:
@@ -32,7 +32,7 @@ def create_logging_lifespan(
             noisy_loggers=noisy_loggers,
             log_format_local_dev_enabled=log_format_local_dev_enabled,
             logger_filter_mapping=logger_filter_mapping,
-            tracing_settings=tracing_settings,
+            tracing_config=tracing_config,
         )
     )
 
@@ -49,7 +49,7 @@ def create_logging_shutdown_event(
     *,
     log_format_local_dev_enabled: bool,
     logger_filter_mapping: dict[LoggerName, list[MessageSubstring]],
-    tracing_settings: TracingSettings | None,
+    tracing_config: TracingConfig,
     log_base_level: LogLevelInt,
     noisy_loggers: tuple[str, ...] | None,
 ) -> Callable[[], Awaitable[None]]:
@@ -67,7 +67,7 @@ def create_logging_shutdown_event(
             noisy_loggers=noisy_loggers,
             log_format_local_dev_enabled=log_format_local_dev_enabled,
             logger_filter_mapping=logger_filter_mapping,
-            tracing_settings=tracing_settings,
+            tracing_config=tracing_config,
         )
     )