Added OTEL tracing

pyproject.toml

@@ -20,7 +20,7 @@ classifiers = [
 ]
 license = { text = "BSD-3-Clause" }
 requires-python = ">=3.10"
-dependencies = ["bidict", "pika", "setuptools", "stomp-py>=7"]
+dependencies = ["bidict", "pika", "setuptools", "stomp-py>=7", "opentelemetry-api==1.20.0", "opentelemetry-sdk==1.20.0", "opentelemetry-exporter-otlp-proto-http==1.20.0" ]
 
 [project.urls]
 Download = "https://github.com/DiamondLightSource/python-workflows/releases"
@@ -53,6 +53,7 @@ OfflineTransport = "workflows.transport.offline_transport:OfflineTransport"
 pika = "workflows.util.zocalo.configuration:Pika"
 stomp = "workflows.util.zocalo.configuration:Stomp"
 transport = "workflows.util.zocalo.configuration:DefaultTransport"
+opentelemetry = "workflows.util.zocalo.configuration:OTEL"
 
 [project.scripts]
 "workflows.validate_recipe" = "workflows.recipe.validate:main"

requirements_dev.txt

@@ -7,3 +7,7 @@ pytest-mock==3.14.0
 pytest-timeout==2.3.1
 stomp-py==8.1.2
 websocket-client==1.8.0
+opentelemetry-api==1.20.0
+opentelemetry-sdk==1.20.0
+opentelemetry-exporter-otlp-proto-http==1.20.0
+marshmallow

src/workflows/recipe/__init__.py

@@ -3,8 +3,11 @@
 import functools
 import logging
 from collections.abc import Callable
+from contextlib import ExitStack
 from typing import Any
 
+from opentelemetry import trace
+
 from workflows.recipe.recipe import Recipe
 from workflows.recipe.validate import validate_recipe
 from workflows.recipe.wrapper import RecipeWrapper
@@ -68,11 +71,52 @@ def unwrap_recipe(header, message):
         if mangle_for_receiving:
             message = mangle_for_receiving(message)
         if header.get("workflows-recipe") in {True, "True", "true", 1}:
+            otel_logs = None
             rw = RecipeWrapper(message=message, transport=transport_layer)
-            if log_extender and rw.environment and rw.environment.get("ID"):
-                with log_extender("recipe_ID", rw.environment["ID"]):
+
+            if hasattr(rw, "environment") and rw.environment.get("ID"):
+                # Extract recipe ID from environment and add to current span
+                span = trace.get_current_span()
+                recipe_id = rw.environment.get("ID")
+
+                if recipe_id:
+                    span.set_attribute("recipe_id", recipe_id)
+
+                # Extract span_id and trace_id for logging
+                span_context = span.get_span_context()
+                if span_context and span_context.is_valid:
+                    span_id = span_context.span_id
+                    trace_id = span_context.trace_id
+
+                    otel_logs = {
+                        "span_id": span_id,
+                        "trace_id": trace_id,
+                    }
+
+                    if recipe_id:
+                        otel_logs["recipe_id"] = recipe_id
+
+                with ExitStack() as stack:
+                    # Configure the context depending on if service is emitting spans
+                    if (
+                        otel_logs
+                        and log_extender
+                        and rw.environment
+                        and rw.environment.get("ID")
+                    ):
+                        stack.enter_context(
+                            log_extender("recipe_ID", rw.environment.get("ID"))
+                        )
+                        stack.enter_context(log_extender("otel_logs", otel_logs))
+                    elif log_extender and rw.environment and rw.environment.get("ID"):
+                        stack.enter_context(
+                            log_extender("recipe_ID", rw.environment.get("ID"))
+                        )
+
                     return callback(rw, header, message.get("payload"))
+
             return callback(rw, header, message.get("payload"))
+
         if allow_non_recipe_messages:
             return callback(None, header, message)
         #   self.log.warning('Discarding non-recipe message:\n' + \

src/workflows/services/common_service.py

@@ -9,8 +9,15 @@
 import time
 from typing import Any
 
+from opentelemetry import trace
+from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+from opentelemetry.sdk.resources import SERVICE_NAME, Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import BatchSpanProcessor
+
 import workflows
 import workflows.logging
+from workflows.transport.middleware.otel_tracing import OTELTracingMiddleware
 
 
 class Status(enum.Enum):
@@ -185,6 +192,40 @@ def start_transport(self):
             self.transport.subscription_callback_set_intercept(
                 self._transport_interceptor
             )
+
+            # Configure OTELTracing if configuration is available
+            otel_config = (
+                self.config._opentelemetry
+                if self.config and hasattr(self.config, "opentelemetry")
+                else None
+            )
+            if otel_config:
+                # Configure OTELTracing
+                resource = Resource.create(
+                    {
+                        SERVICE_NAME: self._service_name,
+                    }
+                )
+
+                self.log.debug("Configuring OTELTracing")
+                provider = TracerProvider(resource=resource)
+                trace.set_tracer_provider(provider)
+
+                # Configure BatchProcessor and OTLPSpanExporter using config values
+                otlp_exporter = OTLPSpanExporter(
+                    endpoint=otel_config["endpoint"],
+                    timeout=otel_config.get("timeout", 10),
+                )
+                span_processor = BatchSpanProcessor(otlp_exporter)
+                provider.add_span_processor(span_processor)
+
+                # Add OTELTracingMiddleware to the transport layer
+                tracer = trace.get_tracer(__name__)
+                otel_middleware = OTELTracingMiddleware(
+                    tracer, service_name=self._service_name
+                )
+                self._transport.add_middleware(otel_middleware)
+
             metrics = self._environment.get("metrics")
             if metrics:
                 import prometheus_client

src/workflows/transport/middleware/__init__.py

@@ -233,6 +233,10 @@ def wrapped_callback(header, message):
 
 
 def wrap(f: Callable):
+    # debugging
+    if f.__name__ == "send":
+        print("we are wrapping send now")
+
     @functools.wraps(f)
     def wrapper(self, *args, **kwargs):
         return functools.reduce(
@@ -243,4 +247,5 @@ def wrapper(self, *args, **kwargs):
             lambda *args, **kwargs: f(self, *args, **kwargs),
         )(*args, **kwargs)
 
+    print(wrapper.__wrapped__)
     return wrapper