fix(fastapi): preserve OTel context from sync route handlers to BackgroundTasks (#3586)

.changelog/4724.fixed

@@ -0,0 +1 @@
+`opentelemetry-instrumentation-fastapi`: Fix OpenTelemetry context (e.g. values set via `context.attach`) not propagating from a synchronous route handler to its `BackgroundTasks`, since sync handlers run in a worker thread and the resulting context mutation was previously lost once the thread returned.

instrumentation/opentelemetry-instrumentation-fastapi/src/opentelemetry/instrumentation/fastapi/__init__.py

@@ -172,18 +172,21 @@ def client_response_hook(span: Span, scope: dict[str, Any], message: dict[str, A
 from __future__ import annotations
 
 import functools
+import inspect
 import logging
 import types
 from typing import Any, Collection, Literal
 from weakref import WeakSet as _WeakSet
 
 import fastapi
 from starlette.applications import Starlette
-from starlette.background import BackgroundTask
+from starlette.background import BackgroundTask, BackgroundTasks
 from starlette.middleware.errors import ServerErrorMiddleware
 from starlette.routing import Match, Route
 from starlette.types import ASGIApp, Receive, Scope, Send
 
+from opentelemetry import context as otel_context
+from opentelemetry.context import set_value
 from opentelemetry.instrumentation._semconv import (
     _get_schema_url,
     _OpenTelemetrySemanticConventionStability,
@@ -398,6 +401,83 @@ async def traced_call(self):
                         return await BackgroundTask._otel_original_call(self)
 
                 BackgroundTask.__call__ = traced_call
+                if not hasattr(BackgroundTasks, "_otel_original_add_task"):
+                    BackgroundTasks._otel_original_add_task = (
+                        BackgroundTasks.add_task
+                    )
+
+                    def context_preserving_add_task(
+                        self, func, *args, **kwargs
+                    ):
+                        """Snapshot the OpenTelemetry context at the moment a
+                        background task is scheduled and merge it into the
+                        context active when the task actually executes.
+
+                        This matters specifically for sync route handlers: they
+                        run inside a worker thread via anyio.to_thread.run_sync,
+                        which copies context into the thread but does not
+                        propagate any mutations (e.g. context.attach calls) back
+                        out once the thread returns. By the time the background
+                        task executes, any OTel context set during the route
+                        handler (such as baggage or values attached via
+                        context.attach) is already gone from the ambient
+                        context. Capturing it here -- inside the same thread
+                        where the mutation happened, at add_task() time --
+                        preserves it for the background task.
+
+                        The captured context is merged into the ambient context
+                        at execution time rather than replacing it outright, so
+                        this does not clobber context set up afterwards, such as
+                        the BackgroundTask span this instrumentation starts
+                        around each task's execution: keys already present in
+                        the ambient context (e.g. the current span) take
+                        precedence over the captured ones.
+                        """
+                        captured_otel_context = otel_context.get_current()
+
+                        def _merge_and_attach():
+                            ambient_otel_context = otel_context.get_current()
+                            merged_otel_context = ambient_otel_context
+                            for (
+                                key,
+                                value,
+                            ) in captured_otel_context.items():
+                                if key not in ambient_otel_context:
+                                    merged_otel_context = set_value(
+                                        key, value, context=merged_otel_context
+                                    )
+                            return otel_context.attach(merged_otel_context)
+
+                        if inspect.iscoroutinefunction(func):
+
+                            @functools.wraps(func)
+                            async def context_preserving_wrapper(
+                                *call_args, **call_kwargs
+                            ):
+                                token = _merge_and_attach()
+                                try:
+                                    return await func(
+                                        *call_args, **call_kwargs
+                                    )
+                                finally:
+                                    otel_context.detach(token)
+                        else:
+
+                            @functools.wraps(func)
+                            def context_preserving_wrapper(
+                                *call_args, **call_kwargs
+                            ):
+                                token = _merge_and_attach()
+                                try:
+                                    return func(*call_args, **call_kwargs)
+                                finally:
+                                    otel_context.detach(token)
+
+                        return BackgroundTasks._otel_original_add_task(
+                            self, context_preserving_wrapper, *args, **kwargs
+                        )
+
+                    BackgroundTasks.add_task = context_preserving_add_task
 
             app._is_instrumented_by_opentelemetry = True
             if app not in _InstrumentedFastAPI._instrumented_fastapi_apps:
@@ -421,6 +501,10 @@ def uninstrument_app(app: fastapi.FastAPI):
             BackgroundTask.__call__ = BackgroundTask._otel_original_call
             del BackgroundTask._otel_original_call
 
+        if hasattr(BackgroundTasks, "_otel_original_add_task"):
+            BackgroundTasks.add_task = BackgroundTasks._otel_original_add_task
+            del BackgroundTasks._otel_original_add_task
+
         app._is_instrumented_by_opentelemetry = False
 
         # Remove the app from the set of instrumented apps to avoid calling uninstrument twice

instrumentation/opentelemetry-instrumentation-fastapi/tests/test_fastapi_instrumentation.py

@@ -25,7 +25,9 @@
 from starlette.types import Receive, Scope, Send
 
 import opentelemetry.instrumentation.fastapi as otel_fastapi
+from opentelemetry import context as otel_context
 from opentelemetry import trace
+from opentelemetry.context import set_value
 from opentelemetry.instrumentation._semconv import (
     OTEL_SEMCONV_STABILITY_OPT_IN,
     _OpenTelemetrySemanticConventionStability,
@@ -529,6 +531,106 @@ async def checkout(background_tasks: BackgroundTasks):
         )
         otel_fastapi.FastAPIInstrumentor().uninstrument_app(app)
 
+    def test_background_task_inherits_sync_route_context(self):
+        """Regression test for #3586: context values attached inside a
+        *sync* route handler must be visible to background tasks scheduled
+        from that handler, matching the behavior already seen for async
+        route handlers.
+
+        Sync routes run inside a worker thread (via
+        anyio.to_thread.run_sync), which copies context into the thread but
+        does not propagate mutations made there back out once the thread
+        returns. Without explicitly preserving it, any context attached
+        during the sync route handler is lost by the time its background
+        tasks run.
+        """
+        self.memory_exporter.clear()
+        app = fastapi.FastAPI()
+        self._instrumentor.instrument_app(app)
+
+        results = {}
+
+        def record_context(label):
+            results[label] = dict(otel_context.get_current())
+
+        def attach_test_value():
+            ctx = set_value("test_key", "test_value")
+            otel_context.attach(ctx)
+
+        @app.get("/sync-endpoint")
+        def sync_endpoint(background_tasks: BackgroundTasks):
+            attach_test_value()
+            background_tasks.add_task(record_context, "sync")
+            return "Hello, World!"
+
+        @app.get("/async-endpoint")
+        async def async_endpoint(background_tasks: BackgroundTasks):
+            attach_test_value()
+            background_tasks.add_task(record_context, "async")
+            return "Hello, World!"
+
+        with TestClient(app) as client:
+            sync_response = client.get("/sync-endpoint")
+            async_response = client.get("/async-endpoint")
+
+        self.assertEqual(200, sync_response.status_code)
+        self.assertEqual(200, async_response.status_code)
+        self.assertIn("test_key", results["sync"])
+        self.assertEqual(results["sync"]["test_key"], "test_value")
+        self.assertIn("test_key", results["async"])
+        self.assertEqual(results["async"]["test_key"], "test_value")
+        otel_fastapi.FastAPIInstrumentor().uninstrument_app(app)
+
+    def test_background_task_context_does_not_clobber_span_parenting(self):
+        """A background task's preserved route-handler context must not
+        override the dedicated BackgroundTask span (see #4251) that this
+        instrumentation starts around the task's execution."""
+        self.memory_exporter.clear()
+        app = fastapi.FastAPI()
+        self._instrumentor.instrument_app(app)
+        tracer = self.tracer_provider.get_tracer(__name__)
+
+        def attach_test_value():
+            ctx = set_value("test_key", "test_value")
+            otel_context.attach(ctx)
+
+        def background_notify():
+            with tracer.start_as_current_span("inside-background-task"):
+                pass
+
+        @app.get("/sync-checkout")
+        def sync_checkout(background_tasks: BackgroundTasks):
+            attach_test_value()
+            background_tasks.add_task(background_notify)
+            return {"status": "processing"}
+
+        with TestClient(app) as client:
+            response = client.get("/sync-checkout")
+        self.assertEqual(200, response.status_code)
+        spans = self.memory_exporter.get_finished_spans()
+        request_span = next(
+            span for span in spans if span.name == "GET /sync-checkout"
+        )
+        background_span = next(
+            span
+            for span in spans
+            if span.name == "BackgroundTask background_notify"
+        )
+        inner_span = next(
+            span for span in spans if span.name == "inside-background-task"
+        )
+        self.assertIsNotNone(background_span.parent)
+        self.assertEqual(
+            background_span.parent.span_id,
+            request_span.context.span_id,
+        )
+        self.assertIsNotNone(inner_span.parent)
+        self.assertEqual(
+            inner_span.parent.span_id,
+            background_span.context.span_id,
+        )
+        otel_fastapi.FastAPIInstrumentor().uninstrument_app(app)
+
     def test_fastapi_route_attribute_added(self):
         """Ensure that fastapi routes are used as the span name."""
         self._client.get("/user/123")