From 54a6a5fb3a86039b3f810c9d8083f8ae2d614dfd Mon Sep 17 00:00:00 2001 From: tin-berri Date: Wed, 28 Oct 2026 14:24:34 -0400 Subject: [PATCH] fix(otel): suppress spans for excluded proxy requests (#48720) --- litellm/integrations/otel/mount.py | 42 +++++- litellm/integrations/otel/plumbing/context.py | 15 ++ .../integrations/otel/plumbing/providers.py | 27 +++- .../otel/test_otel_v2_components.py | 22 +++ .../integrations/otel/test_otel_v2_mount.py | 141 ++++++++++++++++++ .../test_mcp_server_tool_calls_and_headers.py | 113 ++++++++++++++ 6 files changed, 354 insertions(+), 6 deletions(-) diff --git a/litellm/integrations/otel/mount.py b/litellm/integrations/otel/mount.py index c0d42eeb5c5..9d6c5fca75c 100644 --- a/litellm/integrations/otel/mount.py +++ b/litellm/integrations/otel/mount.py @@ -22,6 +22,7 @@ from litellm.integrations.otel.model.config import is_otel_v2_enabled if TYPE_CHECKING: from fastapi import FastAPI from opentelemetry.trace import Span, TracerProvider + from starlette.types import ASGIApp, Receive, Scope, Send # Routes excluded from server-span tracing by default: high-frequency pollers and # static UI/docs assets, none of which are LLM traffic. Entries are substring-matched @@ -44,6 +45,33 @@ _DEFAULT_EXCLUDED_ROUTES: Final = ( ) _DEFAULT_EXCLUDED_URLS: Final = ",".join(_DEFAULT_EXCLUDED_ROUTES) + +class _RequestTraceExclusionMiddleware: + def __init__(self, app: "ASGIApp", excluded_urls: str) -> None: + from opentelemetry.util.http import parse_excluded_urls + + self.app = app + self.excluded_urls = parse_excluded_urls(excluded_urls) + + async def __call__(self, scope: "Scope", receive: "Receive", send: "Send") -> None: + if scope["type"] not in ("http", "websocket"): + await self.app(scope, receive, send) + return + from opentelemetry.instrumentation.asgi import ( # pyright: ignore[reportMissingTypeStubs] # upstream ships no py.typed marker + get_host_port_url_tuple, # pyright: ignore[reportUnknownVariableType] # upstream URL helper has no annotations + ) + + from litellm.integrations.otel.plumbing.context import ( + request_trace_scope, + request_traces_excluded, + ) + + url: Final = str(get_host_port_url_tuple(scope)[2]) # pyright: ignore[reportUnknownArgumentType] # upstream constructs a str URL but leaves its return type unannotated + excluded: Final = request_traces_excluded() or self.excluded_urls.url_disabled(url) + with request_trace_scope(excluded): + await self.app(scope, receive, send) + + # Passthrough routes are catch-alls (e.g. "/openai/{endpoint:path}"), so the # default OTel server-span name "{method} {route}" collapses every upstream # endpoint into "POST /openai/{endpoint:path}". The hook below renames those spans @@ -129,6 +157,8 @@ def instrument_fastapi_app(app: "FastAPI", tracer_provider: "TracerProvider | No if app.middleware_stack is not None: verbose_logger.debug("Skipping OTel V2 FastAPI instrumentation: the app has already started") return + if getattr(app, "_is_instrumented_by_opentelemetry", False): + return # Lazy: only the V2-enabled path needs the optional # ``opentelemetry-instrumentation-fastapi`` package, which is not part of the @@ -151,11 +181,7 @@ def instrument_fastapi_app(app: "FastAPI", tracer_provider: "TracerProvider | No claim_server_span(span) _passthrough_span_name_hook(span, scope) - excluded_urls: Final = ( - os.environ.get("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS") - if "OTEL_PYTHON_FASTAPI_EXCLUDED_URLS" in os.environ - else _DEFAULT_EXCLUDED_URLS - ) + excluded_urls: Final = os.environ.get("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS", _DEFAULT_EXCLUDED_URLS) FastAPIInstrumentor.instrument_app( app, tracer_provider=tracer_provider, @@ -166,6 +192,12 @@ def instrument_fastapi_app(app: "FastAPI", tracer_provider: "TracerProvider | No # template in their name, which can't be rewritten from a hook. exclude_spans=["receive", "send"], ) + instrumented_stack: Final = app.build_middleware_stack + + def build_middleware_stack() -> "ASGIApp": + return _RequestTraceExclusionMiddleware(instrumented_stack(), excluded_urls) + + app.build_middleware_stack = build_middleware_stack # rebind-ok: wrap the complete ASGI stack if get_global_response_propagator() is None: set_global_response_propagator(UpstreamFirstTraceResponsePropagator()) except Exception as e: diff --git a/litellm/integrations/otel/plumbing/context.py b/litellm/integrations/otel/plumbing/context.py index d0c419abe87..b328b281dde 100644 --- a/litellm/integrations/otel/plumbing/context.py +++ b/litellm/integrations/otel/plumbing/context.py @@ -30,6 +30,21 @@ if TYPE_CHECKING: _PROPAGATOR: Final = TraceContextTextMapPropagator() _W3C_TRACE_HEADERS: Final = frozenset(("traceparent", "tracestate")) +_request_traces_excluded: Final[ContextVar[bool]] = ContextVar("litellm_otel_request_traces_excluded", default=False) + + +def request_traces_excluded() -> bool: + return _request_traces_excluded.get() + + +@contextmanager +def request_trace_scope(excluded: bool) -> Generator[None]: + token: Final = _request_traces_excluded.set(excluded) + try: + yield + finally: + _request_traces_excluded.reset(token) + # The request's root span — the FastAPI-owned SERVER span — captured ONCE when the # proxy first resolves it, so request-level spans (the LLM call, guardrails) can diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index 9736192a816..a369e352609 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -35,7 +35,8 @@ from opentelemetry.sdk.trace.export import ( from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, ) -from opentelemetry.trace import Span, SpanContext, SpanKind, Status, StatusCode, Tracer +from opentelemetry.sdk.trace.sampling import Decision, Sampler, SamplingResult +from opentelemetry.trace import Link, Span, SpanContext, SpanKind, Status, StatusCode, Tracer, TraceState from opentelemetry.util.re import parse_env_headers from opentelemetry.util.types import Attributes, AttributeValue @@ -68,6 +69,7 @@ from litellm.integrations.otel.model.semconv import ( from litellm.integrations.otel.model.spans import LiteLLMSpanKind from litellm.integrations.otel.plumbing.context import ( request_destinations, + request_traces_excluded, suppressed_backends, ) from litellm.integrations.otel.plumbing.otlp_tls import resolve_otlp_http_tls @@ -1499,6 +1501,28 @@ def _spec_processor( return _MessageContentFilter(scoped) if strip_message_content else scoped +class _RequestExclusionSampler(Sampler): + def __init__(self, delegate: Sampler) -> None: + self.delegate = delegate + + def should_sample( + self, + parent_context: Context | None, + trace_id: int, + name: str, + kind: SpanKind | None = None, + attributes: Attributes = None, + links: Sequence[Link] | None = None, + trace_state: TraceState | None = None, + ) -> SamplingResult: + if request_traces_excluded(): + return SamplingResult(Decision.DROP) + return self.delegate.should_sample(parent_context, trace_id, name, kind, attributes, links, trace_state) + + def get_description(self) -> str: + return f"RequestExclusion({self.delegate.get_description()})" + + def build_tracer_provider( config: OpenTelemetryV2Config, exporter: SpanExporter | None = None, @@ -1531,6 +1555,7 @@ def build_tracer_provider( destination asked for. """ provider: Final = TracerProvider(resource=build_resource(config)) + provider.sampler = _RequestExclusionSampler(provider.sampler) if baggage_processor is None: baggage_processor = LiteLLMBaggageSpanProcessor(allowed_keys=config.baggage_promoted_keys) provider.add_span_processor(baggage_processor) diff --git a/tests/unit/integrations/otel/test_otel_v2_components.py b/tests/unit/integrations/otel/test_otel_v2_components.py index 8af13fd6e7e..aad06d33edb 100644 --- a/tests/unit/integrations/otel/test_otel_v2_components.py +++ b/tests/unit/integrations/otel/test_otel_v2_components.py @@ -1076,6 +1076,28 @@ def test_build_resource_includes_deployment_environment(): assert resource.attributes["deployment.environment"] == "prod" +@pytest.mark.parametrize("sampler", ["always_on", "always_off", "parentbased_always_on"]) +def test_request_exclusion_preserves_configured_sampling(monkeypatch: pytest.MonkeyPatch, sampler: str) -> None: + from opentelemetry.trace import NonRecordingSpan, SpanContext, set_span_in_context + + monkeypatch.setenv("OTEL_TRACES_SAMPLER", sampler) + provider: Final = providers.build_tracer_provider(OpenTelemetryV2Config(exporter="in_memory")) + tracer: Final = provider.get_tracer("request-exclusions") + sampled_parent: Final = set_span_in_context( + NonRecordingSpan(SpanContext(trace_id=1, span_id=2, is_remote=True, trace_flags=TraceFlags(1))) + ) + try: + with ctx_mod.request_trace_scope(True): + excluded: Final = tracer.start_span("excluded", context=sampled_parent) + assert not excluded.is_recording() + excluded.end() + included: Final = tracer.start_span("included", context=sampled_parent) + assert included.is_recording() == (sampler != "always_off") + included.end() + finally: + provider.shutdown() + + def test_build_tracer_provider_processor_selection(): cfg = OpenTelemetryV2Config(exporter="in_memory") simple = providers.build_tracer_provider(cfg, exporter=InMemorySpanExporter()) diff --git a/tests/unit/integrations/otel/test_otel_v2_mount.py b/tests/unit/integrations/otel/test_otel_v2_mount.py index 61f8b9ae0d9..32e846cc1b9 100644 --- a/tests/unit/integrations/otel/test_otel_v2_mount.py +++ b/tests/unit/integrations/otel/test_otel_v2_mount.py @@ -5,8 +5,11 @@ surface and the server-span + shared-provider behavior it produces. """ +import asyncio import contextvars +from contextlib import nullcontext from datetime import datetime, timezone +from typing import Final import pytest @@ -16,6 +19,7 @@ pytest.importorskip("opentelemetry.instrumentation.fastapi") fastapi = pytest.importorskip("fastapi") from fastapi.middleware.cors import CORSMiddleware # noqa: E402 +from fastapi import FastAPI, HTTPException, Request, Response # noqa: E402 # optional dependencies are skipped first from fastapi.responses import StreamingResponse # noqa: E402 from fastapi.testclient import TestClient # noqa: E402 from starlette.routing import Route # noqa: E402 @@ -36,6 +40,7 @@ from opentelemetry.sdk.trace.sampling import ALWAYS_OFF # noqa: E402 from opentelemetry.trace import SpanKind # noqa: E402 from litellm.constants import LITELLM_UI_ALLOW_HEADERS # noqa: E402 +from litellm.types.services import ServiceLoggerPayload, ServiceTypes # noqa: E402 # optional dependencies are skipped first from litellm.integrations.otel.model.config import ( # noqa: E402 OpenTelemetryV2Config, @@ -548,3 +553,139 @@ def test_the_unhandled_error_helper_names_the_span_it_is_given_in_any_context(mo headers = contextvars.Context().run(trace_response_headers, span) assert headers["traceresponse"] == f"00-{context.trace_id:032x}-{context.span_id:016x}-{context.trace_flags:02x}" + + +@pytest.mark.parametrize( + ("excluded_urls", "path", "root_path", "status", "excluded"), + [ + ("/health,/model/info", "/model/info", "", 200, True), + ("/model/info", "/model/info", "", 401, True), + ("/model/info", "/v1/models", "", 200, False), + (None, "/health/readiness", "", 200, True), + (None, "/model/info", "", 200, False), + ("", "/health/readiness", "", 200, False), + (r"^http://testserver/proxy/model/info$", "/model/info", "/proxy", 200, True), + (r"^http://other/model/info$", "/model/info", "", 200, False), + ], +) +@pytest.mark.parametrize("active_parent", [False, True]) +def test_url_exclusions_cover_auth_and_service_spans( + monkeypatch: pytest.MonkeyPatch, excluded_urls: str | None, path: str, + root_path: str, status: int, excluded: bool, active_parent: bool, +) -> None: + monkeypatch.setenv("LITELLM_OTEL_V2", "1") + if excluded_urls is None: + monkeypatch.delenv("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS", raising=False) + else: + monkeypatch.setenv("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS", excluded_urls) + app: Final = FastAPI(root_path=root_path) + logger: Final = OpenTelemetryV2(config=OpenTelemetryV2Config(exporter="in_memory")) + exporter: Final = InMemorySpanExporter() + logger.tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + + @app.get(path) + async def endpoint() -> Response: + with logger.start_phase_span(f"auth {path}"), logger.tracer.start_as_current_span("direct SDK span"): + await logger.async_service_success_hook( + payload=ServiceLoggerPayload( + is_error=False, service=ServiceTypes.DB, duration=1, + call_type="get_data", event_metadata=None, + ), + start_time=1, end_time=2, + ) + if status == 401: + raise HTTPException(status_code=status) + return Response(status_code=status) + + instrument_fastapi_app(app, tracer_provider=logger.tracer_provider) + with ( + logger.tracer.start_as_current_span("caller") if active_parent else nullcontext(), + TestClient(app) as client, + ): + response: Final = client.get(f"{root_path}{path}?poll=1") + + names: Final = tuple(span.name for span in exporter.get_finished_spans()) + expected_request: Final = () if excluded else ( + "postgres get_data", "direct SDK span", f"auth {path}", f"GET {path}", + ) + assert response.status_code == status + assert names == expected_request + (("caller",) if active_parent else ()) + + +async def test_exclusion_survives_deferred_work_without_suppressing_other_requests( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("LITELLM_OTEL_V2", "1") + monkeypatch.setenv("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS", "/model/info") + app: Final = FastAPI() + logger: Final = OpenTelemetryV2(config=OpenTelemetryV2Config(exporter="in_memory")) + exporter: Final = InMemorySpanExporter() + logger.tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + entered: Final = asyncio.Event() + release: Final = asyncio.Event() + finish: Final = asyncio.Event() + deferred: Final[asyncio.Queue[asyncio.Task[None]]] = asyncio.Queue() + + async def after_response(path: str) -> None: + await finish.wait() + with logger.start_phase_span(f"deferred {path}"): + pass + + @app.get("/{path:path}") + async def endpoint(path: str) -> Response: + with logger.start_phase_span(f"auth {path}"): + if path == "model/info": + entered.set() + await release.wait() + await deferred.put(asyncio.create_task(after_response(path))) + return Response(status_code=200) + + instrument_fastapi_app(app, tracer_provider=logger.tracer_provider) + async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://testserver") as client: + excluded: Final = asyncio.create_task(client.get("/model/info")) + await entered.wait() + included: Final = await client.get("/v1/models") + release.set() + assert (await excluded).status_code == included.status_code == 200 + finish.set() + await asyncio.gather(await deferred.get(), await deferred.get()) + with logger.start_phase_span("independent SDK work"): + pass + + assert sorted(span.name for span in exporter.get_finished_spans()) == [ + "GET /{path:path}", "auth v1/models", "deferred v1/models", "independent SDK work", + ] + + +@pytest.mark.parametrize("path", ["/excluded", "/included"]) +def test_url_exclusions_cover_unhandled_error_handlers_after_repeated_mount( + monkeypatch: pytest.MonkeyPatch, path: str, +) -> None: + monkeypatch.setenv("LITELLM_OTEL_V2", "1") + monkeypatch.setenv("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS", "/excluded") + app: Final = FastAPI() + logger: Final = OpenTelemetryV2(config=OpenTelemetryV2Config(exporter="in_memory")) + exporter: Final = InMemorySpanExporter() + logger.tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + + async def error_handler(_request: Request, _exc: Exception) -> Response: + with logger.start_phase_span("error handler"), logger.tracer.start_as_current_span("error SDK span"): + return Response("handled", status_code=500) + + app.add_exception_handler(Exception, error_handler) + + @app.get(path) + async def endpoint() -> None: + with logger.start_phase_span("auth"): + raise ValueError("unhandled") + + instrument_fastapi_app(app, tracer_provider=logger.tracer_provider) + instrument_fastapi_app(app, tracer_provider=logger.tracer_provider) + with TestClient(app, raise_server_exceptions=False) as client: + response: Final = client.get(path) + + assert response.status_code == 500 + assert response.text == "handled" + assert tuple(span.name for span in exporter.get_finished_spans()) == ( + () if path == "/excluded" else ("auth", "error SDK span", "error handler", "GET /included") + ) diff --git a/tests/unit/proxy/mcp_server/test_mcp_server_tool_calls_and_headers.py b/tests/unit/proxy/mcp_server/test_mcp_server_tool_calls_and_headers.py index ccd5f7f2fea..78c37184d6c 100644 --- a/tests/unit/proxy/mcp_server/test_mcp_server_tool_calls_and_headers.py +++ b/tests/unit/proxy/mcp_server/test_mcp_server_tool_calls_and_headers.py @@ -9209,6 +9209,119 @@ class TestMCPMetaTraceCarrier: assert _mcp_meta_trace_carrier(SimpleNamespace(meta=only_progress)) is None +@pytest.mark.asyncio +@pytest.mark.parametrize("initialized_excluded", [False, True]) +async def test_stateful_mcp_tracing_follows_each_message_exclusion( + monkeypatch: pytest.MonkeyPatch, initialized_excluded: bool +) -> None: + import httpx2 + from fastapi import FastAPI + from mcp.server.context import HandlerResult + from mcp.server.lowlevel import Server + from mcp.server.streamable_http_manager import StreamableHTTPSessionManager + from mcp.shared import _otel + from mcp.types import CallToolRequestParams, NotificationParams + from mcp_types import CLIENT_CAPABILITIES_META_KEY, PROTOCOL_VERSION_META_KEY + from opentelemetry.instrumentation.propagators import get_global_response_propagator, set_global_response_propagator + + from litellm.integrations.otel.model.config import is_otel_v2_enabled + from litellm.integrations.otel.mount import instrument_fastapi_app + from litellm.integrations.otel.plumbing.providers import in_memory_provider + from litellm.proxy import proxy_server + from litellm.proxy.mcp_server import server as gateway + + original_propagator: Final = get_global_response_propagator() + provider, exporter = in_memory_provider() + tracer: Final = provider.get_tracer("mcp-exclusion-test") + monkeypatch.setattr(_otel, "_tracer", tracer) + monkeypatch.setenv("LITELLM_OTEL_V2", "true") + monkeypatch.setenv("OTEL_PYTHON_FASTAPI_EXCLUDED_URLS", "excluded.example") + monkeypatch.setitem(proxy_server.general_settings, "mcp_advertised_versions", ["2025-11-25", "2026-07-28"]) + is_otel_v2_enabled.cache_clear() + notified: Final = asyncio.Event() + + async def call_tool( + ctx: ServerRequestContext[object, object], params: CallToolRequestParams + ) -> CallToolResult: + with tracer.start_as_current_span("tool-handler"): + return CallToolResult(content=[TextContent(type="text", text="done")]) + + async def notification( + ctx: ServerRequestContext[object, object], params: NotificationParams + ) -> HandlerResult: + with tracer.start_as_current_span("notification-handler"): + notified.set() + return None + + protocol: Final = Server[object]("trace-exclusion", on_call_tool=call_tool) + protocol.middleware[:] = gateway.server.middleware + protocol.add_notification_handler("notifications/exclusion-probe", NotificationParams, notification) + manager: Final = StreamableHTTPSessionManager(protocol, stateless=False) + app: Final = FastAPI() + app.mount("/mcp", manager.handle_request) + instrument_fastapi_app(app, tracer_provider=provider) + initial_url: Final = f"http://{'excluded' if initialized_excluded else 'included'}.example/mcp/" + current_url: Final = f"http://{'included' if initialized_excluded else 'excluded'}.example/mcp/" + headers: Final = {"Accept": "application/json, text/event-stream", "MCP-Protocol-Version": "2025-11-25"} + try: + async with manager.run(): + async with httpx2.AsyncClient(transport=httpx2.ASGITransport(app=app)) as client: + initialized: Final = await client.post(initial_url, headers=headers, json={ + "jsonrpc": "2.0", "id": 0, "method": "initialize", "params": { + "protocolVersion": "2025-11-25", "capabilities": {}, + "clientInfo": {"name": "trace-exclusion", "version": "1"}, + }, + }) + assert initialized.status_code == 200, initialized.text + assert any( + span.name == "initialize" for span in exporter.get_finished_spans() + ) is not initialized_excluded + exporter.clear() + session_headers: Final = {**headers, "MCP-Session-Id": initialized.headers["mcp-session-id"]} + ready: Final = await client.post(initial_url, headers=session_headers, + json={"jsonrpc": "2.0", "method": "notifications/initialized"}) + assert ready.status_code == 202, ready.text + response: Final = await client.post(current_url, headers=session_headers, json={ + "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { + "name": "probe", "arguments": {}, "_meta": { + "traceparent": "00-11111111111111111111111111111111-2222222222222222-01", + }, + }, + }) + assert response.status_code == 200 and '"text":"done"' in response.text, response.text + names: Final = {span.name for span in exporter.get_finished_spans()} + assert ("tools/call probe" in names) is initialized_excluded + assert ("tool-handler" in names) is initialized_excluded + exporter.clear() + notification_response: Final = await client.post(current_url, headers=session_headers, + json={"jsonrpc": "2.0", "method": "notifications/exclusion-probe"}) + assert notification_response.status_code == 202, notification_response.text + await notified.wait() + assert any( + span.name == "notification-handler" for span in exporter.get_finished_spans() + ) is initialized_excluded + exporter.clear() + modern_response: Final = await client.post(current_url, headers={ + "Accept": "application/json, text/event-stream", "MCP-Protocol-Version": "2026-07-28", + "MCP-Method": "tools/call", "MCP-Name": "probe", + }, json={ + "jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": { + "name": "probe", "arguments": {}, "_meta": { + PROTOCOL_VERSION_META_KEY: "2026-07-28", CLIENT_CAPABILITIES_META_KEY: {}, + }, + }, + }) + assert modern_response.status_code == 200, modern_response.text + assert '"text":"done"' in modern_response.text, modern_response.text + modern_names: Final = {span.name for span in exporter.get_finished_spans()} + assert ("tools/call probe" in modern_names) is initialized_excluded + assert ("tool-handler" in modern_names) is initialized_excluded + finally: + provider.shutdown() + set_global_response_propagator(original_propagator) + is_otel_v2_enabled.cache_clear() + + @pytest.mark.asyncio async def test_stateful_mcp_tool_call_uses_current_requests_otel_destinations(_mcp_request_ctx) -> None: from types import SimpleNamespace