Bootstrap OpenTelemetry in workers, Batch and Lambda

backend/consultations/apps.py

@@ -9,6 +9,12 @@ class ConsultationsConfig(AppConfig):
 
     def ready(self):
         logger = settings.LOGGER
+
+        if settings.EXECUTION_CONTEXT == "worker":
+            from otel_bootstrap import bootstrap_otel
+
+            bootstrap_otel(service_name="consult-worker")
+
         if settings.ENVIRONMENT.upper() in ["LOCAL", "TEST"]:
             s3_client = s3.get_s3_client()
             buckets = s3_client.list_buckets()["Buckets"]

backend/otel_bootstrap.py

@@ -0,0 +1,91 @@
+"""Process-level OpenTelemetry bootstrap for the RQ worker.
+
+Dormant until OTEL_ENABLED is on, a collector endpoint is set, and the util's
+[otel] extra is installed; a no-op otherwise so the worker keeps running on the
+existing StructuredLogger.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+from collections.abc import Iterator
+from typing import Any
+
+from django.conf import settings
+
+OTEL_ENDPOINT_ENV = "OTEL_EXPORTER_OTLP_ENDPOINT"
+OTEL_ENABLED_ENV = "OTEL_ENABLED"
+_TRACER_NAME = "consult.worker"
+
+
+def otel_requested() -> bool:
+    """Bootstrap only when the flag is on and a collector endpoint is configured."""
+    enabled = os.environ.get(OTEL_ENABLED_ENV, "").strip().lower() == "true"
+    return enabled and bool(os.environ.get(OTEL_ENDPOINT_ENV))
+
+
+def bootstrap_otel(service_name: str) -> None:
+    """Configure OTel for a long-lived worker process. No-op unless requested."""
+    if not otel_requested():
+        return
+
+    logger = settings.LOGGER
+    try:
+        from i_dot_ai_utilities.logging._otel import (
+            configure_otel,
+            ensure_structlog_otel_processors,
+        )
+    except ImportError:
+        logger.warning(
+            "OTel endpoint is set but the i-dot-ai-utilities [otel] extra is not "
+            "installed; telemetry disabled for {service_name}",
+            service_name=service_name,
+        )
+        return
+
+    try:
+        configure_otel(service_name=service_name)
+        ensure_structlog_otel_processors()
+    except RuntimeError:
+        logger.warning(
+            "OTel endpoint is set but the exporter is unavailable; telemetry "
+            "disabled for {service_name}",
+            service_name=service_name,
+        )
+
+
+@contextlib.contextmanager
+def execution_span(name: str, *, context_id: str | None = None, **attributes: Any) -> Iterator[None]:
+    """Wrap a unit of work in a span, carrying context_id so logs correlate.
+
+    A no-op that just runs the body when OTel isn't configured.
+    """
+    if not otel_requested():
+        yield
+        return
+    try:
+        from opentelemetry import trace
+    except ImportError:
+        yield
+        return
+
+    tracer = trace.get_tracer(_TRACER_NAME)
+    with tracer.start_as_current_span(name) as span:
+        if context_id:
+            span.set_attribute("context_id", context_id)
+        for key, value in attributes.items():
+            if value is not None:
+                span.set_attribute(key, value)
+        yield
+
+
+def flush_otel() -> None:
+    """Flush pending telemetry at a job boundary. No-op unless requested."""
+    if not otel_requested():
+        return
+    try:
+        from i_dot_ai_utilities.logging._otel import force_flush_otel
+    except ImportError:
+        return
+    force_flush_otel()

backend/rq_context.py

@@ -3,6 +3,7 @@
 from django_rq import job as _rq_job
 
 from logging_context import get_or_create_context_id, rebind_context
+from otel_bootstrap import execution_span, flush_otel
 
 
 def job(*job_args, **job_kwargs):
@@ -17,7 +18,19 @@ def decorator(func):
         @functools.wraps(func)
         def context_aware(*args, context_id: str | None = None, **kwargs):
             rebind_context(context_id)
-            return func(*args, **kwargs)
+            # Resolve after rebinding so the span carries the same id as the logs.
+            resolved_context_id = get_or_create_context_id()
+            try:
+                with execution_span(
+                    f"rq.job {func.__name__}",
+                    context_id=resolved_context_id,
+                    rq_job=func.__name__,
+                ):
+                    return func(*args, **kwargs)
+            finally:
+                # A worker can idle after a job, so flush at the boundary rather
+                # than wait for the batch processor's timer.
+                flush_otel()
 
         decorated = _rq_job(*job_args, **job_kwargs)(context_aware)
         enqueue_call = decorated.delay  # .delay and .enqueue are the same underlying function

backend/tests/unit/test_otel_bootstrap.py

@@ -0,0 +1,81 @@
+"""The OTel seam stays dormant unless the flag is on and an endpoint is wired."""
+
+import sys
+
+import pytest
+
+import otel_bootstrap
+
+
+@pytest.fixture(autouse=True)
+def _clear_otel_env(monkeypatch):
+    monkeypatch.delenv(otel_bootstrap.OTEL_ENDPOINT_ENV, raising=False)
+    monkeypatch.delenv(otel_bootstrap.OTEL_ENABLED_ENV, raising=False)
+
+
+def _enable(monkeypatch):
+    monkeypatch.setenv(otel_bootstrap.OTEL_ENABLED_ENV, "true")
+    monkeypatch.setenv(otel_bootstrap.OTEL_ENDPOINT_ENV, "http://collector:4318")
+
+
+class TestOtelRequested:
+    def test_false_by_default(self):
+        assert otel_bootstrap.otel_requested() is False
+
+    def test_false_with_endpoint_but_flag_off(self, monkeypatch):
+        monkeypatch.setenv(otel_bootstrap.OTEL_ENDPOINT_ENV, "http://collector:4318")
+        assert otel_bootstrap.otel_requested() is False
+
+    def test_false_with_flag_but_no_endpoint(self, monkeypatch):
+        monkeypatch.setenv(otel_bootstrap.OTEL_ENABLED_ENV, "true")
+        assert otel_bootstrap.otel_requested() is False
+
+    def test_false_when_flag_not_true(self, monkeypatch):
+        monkeypatch.setenv(otel_bootstrap.OTEL_ENABLED_ENV, "false")
+        monkeypatch.setenv(otel_bootstrap.OTEL_ENDPOINT_ENV, "http://collector:4318")
+        assert otel_bootstrap.otel_requested() is False
+
+    def test_true_with_flag_and_endpoint(self, monkeypatch):
+        _enable(monkeypatch)
+        assert otel_bootstrap.otel_requested() is True
+
+
+class TestBootstrapOtel:
+    def test_noop_when_dormant_does_not_import_util(self, monkeypatch):
+        def explode(*_args, **_kwargs):
+            raise AssertionError("must not touch the util when dormant")
+
+        monkeypatch.setattr(otel_bootstrap, "flush_otel", explode)
+        otel_bootstrap.bootstrap_otel(service_name="consult-worker")
+
+    def test_warns_when_requested_but_extra_missing(self, monkeypatch, settings):
+        _enable(monkeypatch)
+        # The [otel] extra is installed in CI, so fake its absence to reach the ImportError branch.
+        monkeypatch.setitem(sys.modules, "i_dot_ai_utilities.logging._otel", None)
+        warnings = []
+        monkeypatch.setattr(
+            settings.LOGGER, "warning", lambda msg, **kw: warnings.append((msg, kw))
+        )
+        otel_bootstrap.bootstrap_otel(service_name="consult-worker")
+        assert warnings
+        assert warnings[0][1]["service_name"] == "consult-worker"
+
+
+class TestFlushOtel:
+    def test_noop_when_dormant(self):
+        otel_bootstrap.flush_otel()
+
+
+class TestExecutionSpan:
+    def test_runs_body_when_dormant(self):
+        ran = False
+        with otel_bootstrap.execution_span("rq.job probe", context_id="abc", rq_job="probe"):
+            ran = True
+        assert ran
+
+    def test_body_exception_propagates(self):
+        with (
+            pytest.raises(ValueError, match="boom"),
+            otel_bootstrap.execution_span("rq.job probe"),
+        ):
+            raise ValueError("boom")

lambda/import_candidate_themes/Makefile

@@ -10,4 +10,5 @@ build:
 	mkdir -p -- ${BUILD_DIR} ${PACKAGES_DIR}/python
 	uv pip install -r code/requirements.txt --target code/package
 	cp code/main.py ${BUILD_DIR}/
+	cp ../shared/otel_bootstrap.py ${BUILD_DIR}/
 	cp -r code/package/* ${PACKAGES_DIR}/python/

lambda/import_candidate_themes/code/main.py

@@ -9,6 +9,7 @@
     ExecutionEnvironmentType,
 )
 from i_dot_ai_utilities.logging.types.log_output_format import LogOutputFormat
+from otel_bootstrap import bootstrap_otel, execution_span, flush_otel
 from rq import Queue
 
 logger = StructuredLogger(
@@ -28,6 +29,8 @@
     )
     logger.info("Sentry initialized")
 
+bootstrap_otel(service_name="consult-import-candidate-themes", logger=logger)
+
 
 def _epoch_ms_to_iso(epoch_ms: int | None) -> str | None:
     """Convert an AWS Batch epoch-millisecond timestamp to an ISO-8601 string."""
@@ -73,52 +76,55 @@ def lambda_handler(event, context):
     )
 
     try:
-        # Connect to Redis
-        redis_host = os.environ.get("REDIS_HOST")
-        if redis_host is None:
-            raise ValueError("REDIS_HOST environment variable is required")
-        redis_port = int(os.environ.get("REDIS_PORT", "6379"))
-
-        logger.info(
-            "Connecting to Redis: {redis_host}:{redis_port}",
-            redis_host=redis_host,
-            redis_port=redis_port,
-        )
-
-        redis_conn = redis.Redis(
-            host=redis_host,
-            port=redis_port,
-            socket_timeout=30,
-            socket_connect_timeout=30,
-        )
-
-        # Test Redis connection
-        ping_result = redis_conn.ping()
-        logger.info("✅ Redis PING result: {ping_result}", ping_result=ping_result)
-
-        # Enqueue the RQ job
-        queue_name = "default"
-        queue = Queue(queue_name, connection=redis_conn)
-        logger.info("Enqueueing RQ job to import candidate themes...")
-        job = queue.enqueue(
-            "data_pipeline.jobs.import_candidate_themes",
-            consultation_code,
-            run_date,
-            user_id,
-            model_name,
+        with execution_span(
+            "lambda.import_candidate_themes",
             context_id=context_id,
-        )
-
-        logger.info(
-            "✅ Successfully queued candidate themes import job {job_id} for: {consultation_code}. "
-            "Job status: {job_status}, queue '{queue_name}' now has {job_count} jobs",
-            job_id=job.id,
             consultation_code=consultation_code,
-            job_status=job.get_status(),
-            queue_name=queue_name,
-            job_count=len(queue),
-        )
-
+        ):
+            redis_host = os.environ.get("REDIS_HOST")
+            if redis_host is None:
+                raise ValueError("REDIS_HOST environment variable is required")
+            redis_port = int(os.environ.get("REDIS_PORT", "6379"))
+
+            logger.info(
+                "Connecting to Redis: {redis_host}:{redis_port}",
+                redis_host=redis_host,
+                redis_port=redis_port,
+            )
+
+            redis_conn = redis.Redis(
+                host=redis_host,
+                port=redis_port,
+                socket_timeout=30,
+                socket_connect_timeout=30,
+            )
+
+            ping_result = redis_conn.ping()
+            logger.info("✅ Redis PING result: {ping_result}", ping_result=ping_result)
+
+            queue_name = "default"
+            queue = Queue(queue_name, connection=redis_conn)
+            logger.info("Enqueueing RQ job to import candidate themes...")
+            job = queue.enqueue(
+                "data_pipeline.jobs.import_candidate_themes",
+                consultation_code,
+                run_date,
+                user_id,
+                model_name,
+                context_id=context_id,
+            )
+
+            logger.info(
+                "✅ Successfully queued candidate themes import job {job_id} for: {consultation_code}. "
+                "Job status: {job_status}, queue '{queue_name}' now has {job_count} jobs",
+                job_id=job.id,
+                consultation_code=consultation_code,
+                job_status=job.get_status(),
+                queue_name=queue_name,
+                job_count=len(queue),
+            )
     except Exception:
         logger.exception("Failed to enqueue candidate themes import job")
         raise
+    finally:
+        flush_otel()

lambda/import_response_annotations/Makefile

@@ -10,4 +10,5 @@ build:
 	mkdir -p -- ${BUILD_DIR} ${PACKAGES_DIR}/python
 	uv pip install -r code/requirements.txt --target code/package
 	cp code/main.py ${BUILD_DIR}/
+	cp ../shared/otel_bootstrap.py ${BUILD_DIR}/
 	cp -r code/package/* ${PACKAGES_DIR}/python/