diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 70a6a051ec5..4ecde6f998b 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -9,7 +9,15 @@ from typing import TYPE_CHECKING, Any, Final, cast from opentelemetry.context import Context, attach, get_current from opentelemetry.sdk._logs import LoggerProvider from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.trace import Span, Tracer, get_current_span, use_span +from opentelemetry.trace import ( + INVALID_SPAN, + Link, + Span, + Tracer, + get_current_span, + set_span_in_context, + use_span, +) import litellm from litellm._logging import verbose_logger @@ -114,6 +122,12 @@ _OTEL_MODULES: Final = ( _OPEN_CALLS_MAX: Final = 10_000 +def _request_trace_links(context: Context | None) -> tuple[Link, ...] | None: + """A link back to the request trace, for a span detached into its own trace.""" + anchor: Final = get_current_span(context).get_span_context() + return (Link(anchor),) if anchor.is_valid else None + + class _LLMCallSpan: """The state carried from the ``pre_call`` boundary to span close. @@ -259,14 +273,20 @@ class OpenTelemetryV2(CustomLogger): # falling back to ambient on the SDK path. Open the span live only when # that resolves to a recordable parent; otherwise defer to the close # callback (the thread-pool case, where the anchor isn't visible here). + # A detached route roots its own trace instead (linked to the request + # trace) — see ``TenantRoute.detached``. + route: Final = self._tenant_tracers.route_for(self.tracer, call.dynamic_params, call.auth_metadata) parent_context: Final = resolve_request_span_context() if is_recordable_span(get_current_span(parent_context)): span = self._emitter.start_span( SpanRole.LLM_CALL, call.provisional_span_name, - parent_context=parent_context, + parent_context=( + set_span_in_context(INVALID_SPAN, parent_context) if route.detached else parent_context + ), start_time_ns=start_time_ns, - tracer=self._tenant_tracers.tracer_for(self.tracer, call.dynamic_params, call.auth_metadata), + tracer=route.tracer, + links=_request_trace_links(parent_context) if route.detached else None, ) self._open_llm_calls[call_id] = _LLMCallSpan(span=span, start_time_ns=start_time_ns) # Evict the oldest open call if the map is over budget. A call that opens @@ -368,16 +388,17 @@ class OpenTelemetryV2(CustomLogger): # as a phantom LLM span. if data.identity.call_id: self._open_llm_calls.pop(data.identity.call_id, None) + route: Final = self._tenant_tracers.route_for(self.tracer, None, auth_metadata(payload, kwargs)) parent_context, links = resolve_mcp_span_context() - parent_context = self._seed_identity_baggage(data.identity, None, parent_context) + seeded: Final = self._seed_identity_baggage(data.identity, None, parent_context) self._emitter.emit( SpanRole.MCP_TOOL_CALL, data, - parent_context=parent_context, + parent_context=(set_span_in_context(INVALID_SPAN, seeded) if route.detached else seeded), start_time_ns=to_ns(start_time), end_time_ns=to_ns(end_time), - links=links, - tracer=self._tenant_tracers.tracer_for(self.tracer, None, auth_metadata(payload, kwargs)), + links=((*(links or ()), *(_request_trace_links(seeded) or ())) if route.detached else links), + tracer=route.tracer, ) return True @@ -405,16 +426,17 @@ class OpenTelemetryV2(CustomLogger): ) if data.identity.call_id: self._open_llm_calls.pop(data.identity.call_id, None) + route: Final = self._tenant_tracers.route_for(self.tracer, None, auth_metadata(payload, kwargs)) parent_context, links = resolve_mcp_span_context() - parent_context = self._seed_identity_baggage(data.identity, None, parent_context) + seeded: Final = self._seed_identity_baggage(data.identity, None, parent_context) self._emitter.emit( SpanRole.MCP_LIST_TOOLS, data, - parent_context=parent_context, + parent_context=(set_span_in_context(INVALID_SPAN, seeded) if route.detached else seeded), start_time_ns=to_ns(start_time), end_time_ns=to_ns(end_time), - links=links, - tracer=self._tenant_tracers.tracer_for(self.tracer, None, auth_metadata(payload, kwargs)), + links=((*(links or ()), *(_request_trace_links(seeded) or ())) if route.detached else links), + tracer=route.tracer, ) return True @@ -460,15 +482,17 @@ class OpenTelemetryV2(CustomLogger): # The worker copied the request task's context, which carries the anchored # root span — parent to it (ambient fallback on the SDK path). Seed identity # Baggage so the span — and the SDK path, which has none — is labeled - # consistently. + # consistently. A detached route roots its own trace instead, linked back. + route: Final = self._tenant_tracers.route_for(self.tracer, call.dynamic_params, call.auth_metadata) parent_ctx = self._seed_identity_baggage(data.identity, data.request_model, resolve_request_span_context()) return self._emitter.emit( SpanRole.LLM_CALL, data, - parent_context=parent_ctx, + parent_context=(set_span_in_context(INVALID_SPAN, parent_ctx) if route.detached else parent_ctx), start_time_ns=carrier.start_time_ns, end_time_ns=end_time_ns, - tracer=self._tenant_tracers.tracer_for(self.tracer, call.dynamic_params, call.auth_metadata), + tracer=route.tracer, + links=_request_trace_links(parent_ctx) if route.detached else None, ) # ====================================================================== # diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index f236343fae4..2377cac3d97 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -12,6 +12,7 @@ needing a logger per tenant. from collections import OrderedDict from collections.abc import Mapping +from dataclasses import dataclass from types import MappingProxyType from typing import Any, Final, TypeAlias from urllib.parse import quote @@ -76,6 +77,23 @@ def _encoded_header_string(headers: Mapping[str, str]) -> str: return ",".join(f"{key}={quote(value, safe='')}" for key, value in headers.items()) +@dataclass(frozen=True, slots=True) +class TenantRoute: + """The tracer to create a span on, plus whether it must root its own trace. + + ``detached`` is True when project routing engaged. Phoenix assigns a whole + trace to one project by whichever of its spans arrives first, so a + project-routed span parented into the request trace gets dragged into the + project of the default-exported request spans and the header does nothing. + The span must therefore start a fresh trace (with a link back to the + request trace for correlation) — which is also how the v1 Phoenix logger + behaved, exporting each request under its own Phoenix-local parent span. + """ + + tracer: Tracer + detached: bool + + class TenantTracerCache: """Credential/project-scoped ``TracerProvider`` cache keyed by the routing headers.""" @@ -95,13 +113,13 @@ class TenantTracerCache: ) self._warned_project_unroutable = False - def tracer_for( + def route_for( self, default: Tracer, dynamic_params: Any, auth_metadata: Mapping[str, str] | None = None, - ) -> Tracer: - """Return the tracer for this request. + ) -> TenantRoute: + """Return the tracer (and trace-detachment flag) for this request. Use ``default`` unless the request's dynamic credentials or its key/team project require a scoped tracer, in which case build (or reuse) one. The @@ -111,7 +129,7 @@ class TenantTracerCache: credential_headers: Final = dynamic_otlp_headers(self._callback_name, dynamic_params) or _NO_HEADERS project_headers: Final = self._project_headers(auth_metadata) if not credential_headers and not project_headers: - return default + return TenantRoute(tracer=default, detached=False) cache_key: Final = ( tuple(sorted(credential_headers.items())), tuple(sorted(project_headers.items())), @@ -125,7 +143,7 @@ class TenantTracerCache: if len(self._providers) > _MAX_CACHED_PROVIDERS: _, evicted = self._providers.popitem(last=False) _shutdown_provider(evicted) - return get_tracer(provider, self._tracer_name) + return TenantRoute(tracer=get_tracer(provider, self._tracer_name), detached=bool(project_headers)) def _project_headers(self, auth_metadata: Mapping[str, str] | None) -> Mapping[str, str]: """The per-request project-routing headers, if this cache can apply them. diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py index 41a5e38be18..7e7af9c725a 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py @@ -75,10 +75,10 @@ def test_provider_cached_per_credential_set(): creds_a = {"arize_space_id": "S", "arize_api_key": "K"} creds_b = {"arize_space_id": "S2", "arize_api_key": "K2"} - cache.tracer_for(default, creds_a) - cache.tracer_for(default, creds_a) # same set → reuse, no new provider + cache.route_for(default, creds_a) + cache.route_for(default, creds_a) # same set → reuse, no new provider assert len(cache._providers) == 1 - cache.tracer_for(default, creds_b) # new set → new provider + cache.route_for(default, creds_b) # new set → new provider assert len(cache._providers) == 2 @@ -101,10 +101,10 @@ def test_provider_cache_is_bounded_and_evicts_lru(monkeypatch): def creds(space): return {"arize_space_id": space, "arize_api_key": "K"} - cache.tracer_for(default, creds("1")) - cache.tracer_for(default, creds("2")) - cache.tracer_for(default, creds("1")) # touch "1" → "2" is now LRU - cache.tracer_for(default, creds("3")) # overflow → evict "2" + cache.route_for(default, creds("1")) + cache.route_for(default, creds("2")) + cache.route_for(default, creds("1")) # touch "1" → "2" is now LRU + cache.route_for(default, creds("3")) # overflow → evict "2" assert len(cache._providers) == 2 assert len(shut_down) == 1 # exactly the evicted provider was shut down @@ -113,14 +113,14 @@ def test_provider_cache_is_bounded_and_evicts_lru(monkeypatch): def test_no_dynamic_params_uses_default_tracer(): cache = _cache("arize") default = NoOpTracer() - assert cache.tracer_for(default, {}) is default + assert cache.route_for(default, {}).tracer is default assert cache._providers == {} def test_non_participating_callback_uses_default_tracer(): cache = _cache("arize_phoenix") default = NoOpTracer() - assert cache.tracer_for(default, {"arize_api_key": "K"}) is default + assert cache.route_for(default, {"arize_api_key": "K"}).tracer is default assert cache._providers == {} @@ -258,11 +258,12 @@ def test_project_header_does_not_touch_other_exporters(): def test_provider_cached_per_project(): cache = _phoenix_cache() default = NoOpTracer() - routed = cache.tracer_for(default, None, {"phoenix_project_name": "proj-a"}) - assert routed is not default - cache.tracer_for(default, None, {"phoenix_project_name": "proj-a"}) + routed = cache.route_for(default, None, {"phoenix_project_name": "proj-a"}) + assert routed.tracer is not default + assert routed.detached is True # project spans must root their own trace + cache.route_for(default, None, {"phoenix_project_name": "proj-a"}) assert len(cache._providers) == 1 - cache.tracer_for(default, None, {"phoenix_project_name": "proj-b"}) + cache.route_for(default, None, {"phoenix_project_name": "proj-b"}) assert len(cache._providers) == 2 for provider in cache._providers.values(): provider.shutdown() @@ -274,9 +275,9 @@ def test_client_dynamic_params_cannot_choose_phoenix_project(): # config (the ``auth_metadata`` argument). cache = _phoenix_cache() default = NoOpTracer() - assert cache.tracer_for(default, {"phoenix_project_name": "attacker"}) is default + assert cache.route_for(default, {"phoenix_project_name": "attacker"}).tracer is default assert ( - cache.tracer_for(default, {"phoenix_project_name_override": "attacker"}) + cache.route_for(default, {"phoenix_project_name_override": "attacker"}).tracer is default ) assert cache._providers == {} @@ -285,7 +286,7 @@ def test_client_dynamic_params_cannot_choose_phoenix_project(): def test_auth_metadata_without_project_uses_default_tracer(): cache = _phoenix_cache() default = NoOpTracer() - assert cache.tracer_for(default, None, {"logging_setting": "x"}) is default + assert cache.route_for(default, None, {"logging_setting": "x"}).tracer is default assert cache._providers == {} @@ -294,6 +295,6 @@ def test_grpc_exporter_gets_no_project_routing(): # gRPC-only Phoenix exporter stays on the default project (warned once). cache = _phoenix_cache(kind="otlp_grpc") default = NoOpTracer() - assert cache.tracer_for(default, None, {"phoenix_project_name": "proj"}) is default + assert cache.route_for(default, None, {"phoenix_project_name": "proj"}).tracer is default assert cache._providers == {} assert cache._warned_project_unroutable is True diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_logger.py b/tests/test_litellm/integrations/otel/test_otel_v2_logger.py index f077f58f209..de05b90676f 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_logger.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_logger.py @@ -2335,6 +2335,7 @@ def test_key_team_auth_metadata_routes_llm_span_to_phoenix_project(): assert parsed["authorization"] == "Bearer phoenix-key" routed_spans = captured[headers].get_finished_spans() assert len(routed_spans) == 1 + assert routed_spans[0].parent is None # own trace, so Phoenix can route it def test_client_request_metadata_cannot_route_phoenix_project(): @@ -2381,7 +2382,13 @@ def test_project_routing_resolves_at_pre_call_before_payload_exists(): asyncio.run(logger.async_log_success_event(close_kwargs, None, None, None)) (headers,) = captured - assert [s.name for s in captured[headers].get_finished_spans()] == ["chat gpt-4o"] + (routed_span,) = captured[headers].get_finished_spans() + assert routed_span.name == "chat gpt-4o" + # Phoenix pins a whole trace to one project by its first-arriving span, so + # the routed span must root its OWN trace, linked back to the request trace. + assert routed_span.parent is None + (link,) = routed_span.links + assert link.context.span_id == server.get_span_context().span_id assert all( s.name != "chat gpt-4o" for s in default_exporter.get_finished_spans() )