🎨 Tracing: instrument logs and ensure traces are propagated through asyncio.Queues (#6857)

Author: bisgaard-itis

packages/aws-library/requirements/_base.txt

@@ -142,6 +142,7 @@ opentelemetry-api==1.27.0
     #   opentelemetry-exporter-otlp-proto-http
     #   opentelemetry-instrumentation
     #   opentelemetry-instrumentation-botocore
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
     #   opentelemetry-propagator-aws-xray
@@ -160,10 +161,13 @@ opentelemetry-exporter-otlp-proto-http==1.27.0
 opentelemetry-instrumentation==0.48b0
     # via
     #   opentelemetry-instrumentation-botocore
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
 opentelemetry-instrumentation-botocore==0.48b0
     # via -r requirements/_base.in
+opentelemetry-instrumentation-logging==0.48b0
+    # via -r requirements/../../../packages/service-library/requirements/_base.in
 opentelemetry-instrumentation-redis==0.48b0
     # via -r requirements/../../../packages/service-library/requirements/_base.in
 opentelemetry-instrumentation-requests==0.48b0

packages/service-library/requirements/_base.in

@@ -20,6 +20,7 @@ opentelemetry-api
 opentelemetry-exporter-otlp
 opentelemetry-instrumentation-requests
 opentelemetry-instrumentation-redis
+opentelemetry-instrumentation-logging
 opentelemetry-sdk
 psutil
 pydantic

packages/service-library/requirements/_base.txt

@@ -101,6 +101,7 @@ opentelemetry-api==1.27.0
     #   opentelemetry-exporter-otlp-proto-grpc
     #   opentelemetry-exporter-otlp-proto-http
     #   opentelemetry-instrumentation
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
     #   opentelemetry-sdk
@@ -117,8 +118,11 @@ opentelemetry-exporter-otlp-proto-http==1.27.0
     # via opentelemetry-exporter-otlp
 opentelemetry-instrumentation==0.48b0
     # via
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
+opentelemetry-instrumentation-logging==0.48b0
+    # via -r requirements/_base.in
 opentelemetry-instrumentation-redis==0.48b0
     # via -r requirements/_base.in
 opentelemetry-instrumentation-requests==0.48b0

packages/service-library/src/servicelib/async_utils.py

@@ -6,6 +6,7 @@
 from functools import wraps
 from typing import TYPE_CHECKING, Any, Awaitable, Callable, Deque
 
+from . import tracing
 from .utils_profiling_middleware import dont_profile, is_profiling, profile_context
 
 logger = logging.getLogger(__name__)
@@ -34,6 +35,7 @@ class Context:
 
 @dataclass
 class QueueElement:
+    tracing_context: tracing.TracingContext
     do_profile: bool = False
     input: Awaitable | None = None
     output: Any | None = None
@@ -162,16 +164,17 @@ async def worker(in_q: Queue[QueueElement], out_q: Queue) -> None:
                     while True:
                         element = await in_q.get()
                         in_q.task_done()
-                        # check if requested to shutdown
-                        try:
-                            do_profile = element.do_profile
-                            awaitable = element.input
-                            if awaitable is None:
-                                break
-                            with profile_context(do_profile):
-                                result = await awaitable
-                        except Exception as e:  # pylint: disable=broad-except
-                            result = e
+                        with tracing.use_tracing_context(element.tracing_context):
+                            # check if requested to shutdown
+                            try:
+                                do_profile = element.do_profile
+                                awaitable = element.input
+                                if awaitable is None:
+                                    break
+                                with profile_context(do_profile):
+                                    result = await awaitable
+                            except Exception as e:  # pylint: disable=broad-except
+                                result = e
                         await out_q.put(result)
 
                     logging.info(
@@ -189,6 +192,7 @@ async def worker(in_q: Queue[QueueElement], out_q: Queue) -> None:
                 queue_input = QueueElement(
                     input=decorated_function(*args, **kwargs),
                     do_profile=is_profiling(),
+                    tracing_context=tracing.get_context(),
                 )
                 await context.in_queue.put(queue_input)
                 wrapped_result = await context.out_queue.get()

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

@@ -70,7 +70,6 @@ def setup_tracing(
     ):
         _logger.warning("Skipping opentelemetry tracing setup")
         return
-
     # Set up the tracer provider
     resource = Resource(attributes={"service.name": service_name})
     trace.set_tracer_provider(TracerProvider(resource=resource))

packages/service-library/src/servicelib/logging_utils.py

@@ -16,7 +16,10 @@
 from pathlib import Path
 from typing import Any, NotRequired, TypeAlias, TypedDict, TypeVar
 
+from settings_library.tracing import TracingSettings
+
 from .logging_utils_filtering import GeneralLogFilter, LoggerName, MessageSubstring
+from .tracing import setup_log_tracing
 from .utils_secrets import mask_sensitive_data
 
 _logger = logging.getLogger(__name__)
@@ -121,20 +124,41 @@ def config_all_loggers(
     *,
     log_format_local_dev_enabled: bool,
     logger_filter_mapping: dict[LoggerName, list[MessageSubstring]],
+    tracing_settings: TracingSettings | None,
 ) -> None:
     """
     Applies common configuration to ALL registered loggers
     """
-    fmt = DEFAULT_FORMATTING
     the_manager: logging.Manager = logging.Logger.manager
     root_logger = logging.getLogger()
 
     loggers = [root_logger] + [
         logging.getLogger(name) for name in the_manager.loggerDict
     ]
 
+    fmt = DEFAULT_FORMATTING
+    if tracing_settings is not None:
+        fmt = (
+            "log_level=%(levelname)s "
+            "| log_timestamp=%(asctime)s "
+            "| log_source=%(name)s:%(funcName)s(%(lineno)d) "
+            "| log_uid=%(log_uid)s "
+            "| log_oec=%(log_oec)s"
+            "| log_trace_id=%(otelTraceID)s "
+            "| log_span_id=%(otelSpanID)s "
+            "| log_resource.service.name=%(otelServiceName)s "
+            "| log_trace_sampled=%(otelTraceSampled)s] "
+            "| log_msg=%(message)s"
+        )
+        setup_log_tracing(tracing_settings=tracing_settings)
     if log_format_local_dev_enabled:
         fmt = LOCAL_FORMATTING
+        if tracing_settings is not None:
+            fmt = (
+                "%(levelname)s: [%(asctime)s/%(processName)s] "
+                "[log_trace_id=%(otelTraceID)s log_span_id=%(otelSpanID)s log_resource.service.name=%(otelServiceName)s log_trace_sampled=%(otelTraceSampled)s] "
+                "[%(name)s:%(funcName)s(%(lineno)d)] -  %(message)s"
+            )
 
     for logger in loggers:
         _set_logging_handler(

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

@@ -0,0 +1,35 @@
+from contextlib import contextmanager
+from typing import TypeAlias
+
+from opentelemetry import context as otcontext
+from opentelemetry import trace
+from opentelemetry.instrumentation.logging import LoggingInstrumentor
+from settings_library.tracing import TracingSettings
+
+TracingContext: TypeAlias = otcontext.Context | None
+
+
+def _is_tracing() -> bool:
+    return trace.get_current_span().is_recording()
+
+
+def get_context() -> TracingContext:
+    if not _is_tracing():
+        return None
+    return otcontext.get_current()
+
+
+@contextmanager
+def use_tracing_context(context: TracingContext):
+    if context is not None:
+        otcontext.attach(context)
+    try:
+        yield
+    finally:
+        if context is not None:
+            otcontext.detach(context)
+
+
+def setup_log_tracing(tracing_settings: TracingSettings):
+    _ = tracing_settings
+    LoggingInstrumentor().instrument(set_logging_format=False)

packages/simcore-sdk/requirements/_base.txt

@@ -165,6 +165,7 @@ opentelemetry-api==1.27.0
     #   opentelemetry-instrumentation-aiopg
     #   opentelemetry-instrumentation-asyncpg
     #   opentelemetry-instrumentation-dbapi
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
     #   opentelemetry-sdk
@@ -184,6 +185,7 @@ opentelemetry-instrumentation==0.48b0
     #   opentelemetry-instrumentation-aiopg
     #   opentelemetry-instrumentation-asyncpg
     #   opentelemetry-instrumentation-dbapi
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
 opentelemetry-instrumentation-aiopg==0.48b0
@@ -192,6 +194,8 @@ opentelemetry-instrumentation-asyncpg==0.48b0
     # via -r requirements/../../../packages/postgres-database/requirements/_base.in
 opentelemetry-instrumentation-dbapi==0.48b0
     # via opentelemetry-instrumentation-aiopg
+opentelemetry-instrumentation-logging==0.48b0
+    # via -r requirements/../../../packages/service-library/requirements/_base.in
 opentelemetry-instrumentation-redis==0.48b0
     # via -r requirements/../../../packages/service-library/requirements/_base.in
 opentelemetry-instrumentation-requests==0.48b0

services/agent/requirements/_base.txt

@@ -157,6 +157,7 @@ opentelemetry-api==1.28.2
     #   opentelemetry-instrumentation-asgi
     #   opentelemetry-instrumentation-fastapi
     #   opentelemetry-instrumentation-httpx
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
     #   opentelemetry-sdk
@@ -176,6 +177,7 @@ opentelemetry-instrumentation==0.49b2
     #   opentelemetry-instrumentation-asgi
     #   opentelemetry-instrumentation-fastapi
     #   opentelemetry-instrumentation-httpx
+    #   opentelemetry-instrumentation-logging
     #   opentelemetry-instrumentation-redis
     #   opentelemetry-instrumentation-requests
 opentelemetry-instrumentation-asgi==0.49b2
@@ -184,7 +186,9 @@ opentelemetry-instrumentation-fastapi==0.49b2
     # via -r requirements/../../../packages/service-library/requirements/_fastapi.in
 opentelemetry-instrumentation-httpx==0.49b2
     # via -r requirements/../../../packages/service-library/requirements/_fastapi.in
-opentelemetry-instrumentation-redis==0.49b2
+opentelemetry-instrumentation-logging==0.48b0
+    # via -r requirements/../../../packages/service-library/requirements/_base.in
+opentelemetry-instrumentation-redis==0.48b0
     # via -r requirements/../../../packages/service-library/requirements/_base.in
 opentelemetry-instrumentation-requests==0.49b2
     # via -r requirements/../../../packages/service-library/requirements/_base.in

services/agent/src/simcore_service_agent/core/application.py

@@ -33,6 +33,7 @@ def _setup_logger(settings: ApplicationSettings):
     config_all_loggers(
         log_format_local_dev_enabled=settings.AGENT_VOLUMES_LOG_FORMAT_LOCAL_DEV_ENABLED,
         logger_filter_mapping=settings.AGENT_VOLUMES_LOG_FILTER_MAPPING,
+        tracing_settings=settings.AGENT_TRACING,
     )