feat(orchestrator): Telemetry - Trace running execution processing time

cloud_pipelines_backend/instrumentation/orchestrator_tracing.py

@@ -0,0 +1,24 @@
+"""OTel spans for the orchestrator processing loop so per-execution processing time is measurable."""
+
+from __future__ import annotations
+
+import collections.abc
+import contextlib
+
+from opentelemetry import trace
+from opentelemetry.trace import StatusCode
+
+_tracer = trace.get_tracer("tangle.orchestrator")
+
+
+@contextlib.contextmanager
+def operation_span(
+    name: str, *, attributes: dict[str, object] | None = None
+) -> collections.abc.Iterator[trace.Span]:
+    with _tracer.start_as_current_span(name, attributes=attributes) as span:
+        try:
+            yield span
+        except Exception as exception:
+            span.set_status(StatusCode.ERROR)
+            span.record_exception(exception)
+            raise

cloud_pipelines_backend/orchestrator_sql.py

@@ -25,6 +25,7 @@
 from .instrumentation import bugsnag_instrumentation
 from .instrumentation import contextual_logging
 from .instrumentation import metrics as app_metrics
+from .instrumentation import orchestrator_tracing
 
 _logger = logging.getLogger(__name__)
 
@@ -229,10 +230,16 @@ def internal_process_running_executions_queue(self, session: orm.Session):
                     f"Before processing running container execution. Queries duration {queries_duration_ms}ms."
                 )
                 try:
-                    self.internal_process_one_running_execution(
-                        session=session,
-                        container_execution=running_container_execution,
-                    )
+                    with orchestrator_tracing.operation_span(
+                        "orchestrator.process_running_execution",
+                        attributes={
+                            "container_execution.id": running_container_execution.id
+                        },
+                    ):
+                        self.internal_process_one_running_execution(
+                            session=session,
+                            container_execution=running_container_execution,
+                        )
                 except Exception as ex:
                     _logger.exception("Error processing running container execution")
                     session.rollback()