mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-12 08:42:18 +00:00
fix(otel): root project-routed Phoenix spans in their own trace
Phoenix assigns a whole trace to one project by whichever span arrives first. The request's auth/db/root spans always export through the default provider without the project header, so a project-routed LLM span parented into that trace got dragged back into the default project and the header did nothing (verified against a live Phoenix instance). Detach the routed span into its own trace with a link back to the request trace, mirroring how the v1 Phoenix logger exported each request under its own local parent.
This commit is contained in:
parent
fad0cd515a
commit
3bbbe893a2
4 changed files with 87 additions and 37 deletions
|
|
@ -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,
|
||||
)
|
||||
|
||||
# ====================================================================== #
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue