diff --git a/litellm/integrations/otel/logger.py b/litellm/integrations/otel/logger.py index 4ecde6f998b..dde9c33bfa6 100644 --- a/litellm/integrations/otel/logger.py +++ b/litellm/integrations/otel/logger.py @@ -137,13 +137,24 @@ class _LLMCallSpan: own (worker-copied) ambient context using ``start_time_ns``. The presence of a carrier for a call at all is the proof that ``pre_call`` ran, i.e. that an upstream call was actually attempted. + + ``provider`` is the routed provider the live span was opened on (``None`` on + the default route or when creation was deferred). It is held in the tenant + cache while the span is open so LRU eviction can't shut the provider down + under it, and must be released exactly once when the carrier is removed. """ - __slots__ = ("span", "start_time_ns") + __slots__ = ("provider", "span", "start_time_ns") - def __init__(self, span: "Span | None", start_time_ns: int | None) -> None: + def __init__( + self, + span: "Span | None", + start_time_ns: int | None, + provider: "TracerProvider | None" = None, + ) -> None: self.span = span self.start_time_ns = start_time_ns + self.provider = provider class OpenTelemetryV2(CustomLogger): @@ -288,12 +299,19 @@ class OpenTelemetryV2(CustomLogger): 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) + if span is not None and route.provider is not None: + self._tenant_tracers.hold(route.provider) + self._open_llm_calls[call_id] = _LLMCallSpan( + span=span, + start_time_ns=start_time_ns, + provider=route.provider if span is not None else None, + ) # Evict the oldest open call if the map is over budget. A call that opens # but never closes (a stream that only fires stream events) would linger # otherwise; the evicted span is simply dropped (never exported). if len(self._open_llm_calls) > _OPEN_CALLS_MAX: - self._open_llm_calls.popitem(last=False) + _, evicted = self._open_llm_calls.popitem(last=False) + self._release_carrier(evicted) async def async_log_success_event(self, kwargs, response_obj, start_time, end_time): if self._emit_mcp_tool_call(kwargs, start_time, end_time): @@ -387,7 +405,7 @@ class OpenTelemetryV2(CustomLogger): # otherwise linger until evicted; drop it so it's neither leaked nor closed # as a phantom LLM span. if data.identity.call_id: - self._open_llm_calls.pop(data.identity.call_id, None) + self._release_carrier(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() seeded: Final = self._seed_identity_baggage(data.identity, None, parent_context) @@ -425,7 +443,7 @@ class OpenTelemetryV2(CustomLogger): payload, capture_content=self.config.capture_span_content ) if data.identity.call_id: - self._open_llm_calls.pop(data.identity.call_id, None) + self._release_carrier(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() seeded: Final = self._seed_identity_baggage(data.identity, None, parent_context) @@ -459,6 +477,24 @@ class OpenTelemetryV2(CustomLogger): carrier: Final = self._open_llm_calls.pop(call_id, None) if call_id else None if carrier is None: return None + try: + return self._finish_carrier(carrier, call, end_time) + finally: + # After the span has ended, so a release-triggered provider shutdown + # force-flushes it out rather than racing its enqueue. + self._release_carrier(carrier) + + def _release_carrier(self, carrier: "_LLMCallSpan | None") -> None: + """Release the routed provider a removed carrier was holding open.""" + if carrier is not None and carrier.provider is not None: + self._tenant_tracers.release(carrier.provider) + + def _finish_carrier( + self, + carrier: _LLMCallSpan, + call: LLMCallEvent, + end_time: datetime | float | None, + ) -> Span | None: payload: Final = call.payload if payload is None: if carrier.span is not None: diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 2377cac3d97..3a440bdb90c 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -92,6 +92,10 @@ class TenantRoute: tracer: Tracer detached: bool + #: The provider ``tracer`` came from, or ``None`` on the default route. A + #: caller that keeps a span open across callbacks must ``hold``/``release`` + #: it so LRU eviction can't shut the provider down under the open span. + provider: TracerProvider | None = None class TenantTracerCache: @@ -107,12 +111,43 @@ class TenantTracerCache: self._callback_name = callback_name self._tracer_name = tracer_name self._providers: OrderedDict[tuple[_HeaderItems, _HeaderItems], TracerProvider] = OrderedDict() + self._open_span_counts: dict[TracerProvider, int] = {} # mutable-ok: live refcount state + self._retired: set[TracerProvider] = set() # mutable-ok: evicted providers draining open spans self._project_routable = any( spec.owner == callback_name and spec.kind.lower() not in (*_NON_OTLP_KINDS, *_GRPC_KINDS) for spec in config.exporters ) self._warned_project_unroutable = False + def hold(self, provider: TracerProvider) -> None: + """Count an open span on ``provider`` so eviction defers its shutdown.""" + self._open_span_counts[provider] = self._open_span_counts.get(provider, 0) + 1 + + def release(self, provider: TracerProvider) -> None: + """Drop one open-span count; shut a retired provider down once drained.""" + remaining: Final = self._open_span_counts.get(provider, 0) - 1 + if remaining > 0: + self._open_span_counts[provider] = remaining + return + self._open_span_counts.pop(provider, None) + if provider in self._retired: + self._retired.discard(provider) + _shutdown_provider(provider) + + def _retire(self, provider: TracerProvider) -> None: + """Shut an evicted provider down, or defer while spans are still open. + + Shutting down immediately would stop the provider's processors while a + span opened at ``pre_call`` is still live, silently dropping it at end + instead of exporting it. A retired provider is pinned only by its open + spans, and the logger's open-call map is itself bounded, so a request + that never reaches a close callback still releases it eventually. + """ + if self._open_span_counts.get(provider, 0) > 0: + self._retired.add(provider) + return + _shutdown_provider(provider) + def route_for( self, default: Tracer, @@ -142,8 +177,12 @@ class TenantTracerCache: self._providers[cache_key] = provider if len(self._providers) > _MAX_CACHED_PROVIDERS: _, evicted = self._providers.popitem(last=False) - _shutdown_provider(evicted) - return TenantRoute(tracer=get_tracer(provider, self._tracer_name), detached=bool(project_headers)) + self._retire(evicted) + return TenantRoute( + tracer=get_tracer(provider, self._tracer_name), + detached=bool(project_headers), + provider=provider, + ) 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 7e7af9c725a..8d14a58e352 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py @@ -298,3 +298,40 @@ def test_grpc_exporter_gets_no_project_routing(): assert cache.route_for(default, None, {"phoenix_project_name": "proj"}).tracer is default assert cache._providers == {} assert cache._warned_project_unroutable is True + + +def test_eviction_defers_shutdown_while_a_span_is_open(monkeypatch): + # An LLM span opened at pre_call stays open until the close callback; LRU + # eviction in that window must not stop the provider's processors, or the + # span is silently dropped at end instead of exported. + from litellm.integrations.otel.plumbing import routing as routing_mod + + monkeypatch.setattr(routing_mod, "_MAX_CACHED_PROVIDERS", 1) + shut_down = [] + monkeypatch.setattr( + routing_mod, "_shutdown_provider", lambda p: shut_down.append(p) + ) + cache = _cache("arize") + default = NoOpTracer() + + route_a = cache.route_for(default, {"arize_space_id": "A", "arize_api_key": "K"}) + assert route_a.provider is not None + cache.hold(route_a.provider) + cache.route_for(default, {"arize_space_id": "B", "arize_api_key": "K"}) # evicts A + assert shut_down == [] # deferred: A still has an open span + cache.release(route_a.provider) + assert shut_down == [route_a.provider] + + +def test_release_without_eviction_keeps_provider_alive(monkeypatch): + from litellm.integrations.otel.plumbing import routing as routing_mod + + shut_down = [] + monkeypatch.setattr( + routing_mod, "_shutdown_provider", lambda p: shut_down.append(p) + ) + cache = _cache("arize") + route = cache.route_for(NoOpTracer(), {"arize_space_id": "A", "arize_api_key": "K"}) + cache.hold(route.provider) + cache.release(route.provider) + assert shut_down == [] # still cached, never retired 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 de05b90676f..4a03fbda7e4 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_logger.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_logger.py @@ -2392,3 +2392,50 @@ def test_project_routing_resolves_at_pre_call_before_payload_exists(): assert all( s.name != "chat gpt-4o" for s in default_exporter.get_finished_spans() ) + + +def test_evicted_provider_still_exports_span_opened_before_eviction(monkeypatch): + """LRU eviction while a routed span is still open must defer the provider + shutdown: the span opened at ``pre_call`` closes at the later success + callback and would otherwise be silently dropped instead of exported.""" + from litellm.integrations.otel.plumbing import routing as routing_mod + + monkeypatch.setattr(routing_mod, "_MAX_CACHED_PROVIDERS", 1) + logger, _default_exporter, captured = _phoenix_routing_logger("capture_evict") + md_a = {"user_api_key_auth_metadata": {"phoenix_project_name": "proj-a"}} + server = logger._emitter.start_span( + SpanRole.PROXY_REQUEST, LITELLM_PROXY_REQUEST_SPAN_NAME + ) + with trace.use_span(server, end_on_exit=False): + logger.log_pre_api_call( + model="gpt-4o", + messages=[], + kwargs={"litellm_call_id": "call_a", "litellm_params": {"metadata": md_a}}, + ) + + # A second project's full call overflows the size-1 LRU and evicts proj-a's + # provider while call_a's span is still open. + md_b = {"user_api_key_auth_metadata": {"phoenix_project_name": "proj-b"}} + _emit_llm( + logger, + { + "standard_logging_object": _payload(litellm_call_id="call_b", metadata=md_b), + "litellm_params": {"metadata": md_b}, + }, + ) + + asyncio.run( + logger.async_log_success_event( + { + "standard_logging_object": _payload(litellm_call_id="call_a", metadata=md_a), + "litellm_params": {"metadata": md_a}, + }, + None, + None, + None, + ) + ) + server.end() + + headers_a = next(h for h in captured if "proj-a" in h) + assert [s.name for s in captured[headers_a].get_finished_spans()] == ["chat gpt-4o"]