diff --git a/.changelog/4724.fixed b/.changelog/4724.fixed new file mode 100644 index 0000000000..89984469f4 --- /dev/null +++ b/.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. diff --git a/instrumentation/opentelemetry-instrumentation-fastapi/src/opentelemetry/instrumentation/fastapi/__init__.py b/instrumentation/opentelemetry-instrumentation-fastapi/src/opentelemetry/instrumentation/fastapi/__init__.py index ab936f0ed5..f1b7f59540 100644 --- a/instrumentation/opentelemetry-instrumentation-fastapi/src/opentelemetry/instrumentation/fastapi/__init__.py +++ b/instrumentation/opentelemetry-instrumentation-fastapi/src/opentelemetry/instrumentation/fastapi/__init__.py @@ -172,6 +172,7 @@ 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 @@ -179,11 +180,13 @@ def client_response_hook(span: Span, scope: dict[str, Any], message: dict[str, A 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 diff --git a/instrumentation/opentelemetry-instrumentation-fastapi/tests/test_fastapi_instrumentation.py b/instrumentation/opentelemetry-instrumentation-fastapi/tests/test_fastapi_instrumentation.py index 991aec902f..9ba46a3a63 100644 --- a/instrumentation/opentelemetry-instrumentation-fastapi/tests/test_fastapi_instrumentation.py +++ b/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")