mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-07 07:58:09 +00:00
fix(otel): drain in-flight spans before shutting down evicted providers
LRU eviction shut a routed provider down immediately, but an LLM span opened at pre_call stays open until the later success or failure callback; with more than 256 overlapping credential/project routes that in-flight span was silently dropped instead of exported. Refcount open spans per provider (hold at span open, release when the carrier is removed on close, carrier-map eviction, or MCP stray-carrier cleanup) and defer a retired provider's shutdown until its last open span closes.
This commit is contained in:
parent
3bbbe893a2
commit
cece0789ea
4 changed files with 167 additions and 8 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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.
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue