From 9063c39b9daba2695b4ba4e7e3b745a272ea6ad7 Mon Sep 17 00:00:00 2001 From: mubashir1osmani Date: Wed, 19 Aug 2026 21:13:16 -0400 Subject: [PATCH] fix(otel): cap retired tenant providers draining open spans --- litellm/integrations/otel/plumbing/routing.py | 33 ++++++++++++----- .../integrations/otel/test_otel_v2_dynamic.py | 35 +++++++++++++++++++ 2 files changed, 59 insertions(+), 9 deletions(-) diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 0d6b007b188..6474d16050c 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -48,6 +48,16 @@ _GRPC_KINDS: Final = ("otlp_grpc", "grpc") # evicted providers so their threads are reclaimed. _MAX_CACHED_PROVIDERS: Final = 256 +# Cap on providers evicted from the cache while still holding open spans, which +# are kept alive to drain instead of being shut down under them. Their only +# other bound is the logger's open-call map (10k), so without this a caller +# cycling unique credential sets across long-lived calls could pin far more +# live providers, and exporter threads, than the cache cap allows. Past this +# many, the stalest retiree is shut down, dropping whatever it was draining: by +# then a span from a route evicted long ago that never closed. Total live +# providers is therefore ``_MAX_CACHED_PROVIDERS + _MAX_RETIRED_PROVIDERS``. +_MAX_RETIRED_PROVIDERS: Final = 64 + _HeaderItems: TypeAlias = tuple[tuple[str, str], ...] _NO_HEADERS: Final[Mapping[str, str]] = MappingProxyType({}) @@ -118,7 +128,8 @@ class TenantTracerCache: self._lock: Final = threading.Lock() 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 + # Oldest-first so an overflow of draining providers sheds the stalest. + self._retired: OrderedDict[TracerProvider, None] = OrderedDict() # mutable-ok: draining evicted providers self._project_routable = any( spec.owner == callback_name and spec.kind.lower() not in (*_NON_OTLP_KINDS, *_GRPC_KINDS) for spec in config.exporters @@ -142,7 +153,7 @@ class TenantTracerCache: return self._open_span_counts.pop(provider, None) drained: Final = provider in self._retired - self._retired.discard(provider) + self._retired.pop(provider, None) if drained: _shutdown_provider(provider) @@ -203,18 +214,22 @@ class TenantTracerCache: A provider with open spans is retired to drain instead: stopping its processors while a span opened at ``pre_call`` is still live would - silently drop that span 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. + silently drop that span at end instead of exporting it. Retirees are + themselves capped, so the stalest one is shut down (and its open-span + count dropped, making its eventual ``release`` a no-op) once too many + pile up rather than letting them accumulate a thread each. """ if len(self._providers) <= _MAX_CACHED_PROVIDERS: return None _, evicted = self._providers.popitem(last=False) - if self._open_span_counts.get(evicted, 0) > 0: - self._retired.add(evicted) + if self._open_span_counts.get(evicted, 0) == 0: + return evicted + self._retired[evicted] = None + if len(self._retired) <= _MAX_RETIRED_PROVIDERS: return None - return evicted + overflowed, _ = self._retired.popitem(last=False) + self._open_span_counts.pop(overflowed, None) + return overflowed 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 408f010dbd7..ae143e81bf0 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py @@ -326,6 +326,41 @@ def test_eviction_defers_shutdown_while_a_span_is_open(monkeypatch): assert shut_down == [route_a.provider] +def test_retired_providers_are_capped(monkeypatch): + # Retiring an evicted provider keeps it, and its exporter thread, alive + # while a span is open, so retirees need a cap of their own: a caller + # cycling unique credential sets across calls that never close would + # otherwise pin one live provider per open call, far past the cache bound. + # Past the cap the stalest retiree is shut down and its later release is a + # no-op, while the ones still within the cap keep draining. + from litellm.integrations.otel.plumbing import routing as routing_mod + + monkeypatch.setattr(routing_mod, "_MAX_CACHED_PROVIDERS", 1) + monkeypatch.setattr(routing_mod, "_MAX_RETIRED_PROVIDERS", 2) + shut_down = [] + monkeypatch.setattr( + routing_mod, "_shutdown_provider", lambda p: shut_down.append(p) + ) + cache = _cache("arize") + default = NoOpTracer() + + # Every route stays held (no release), so each one evicts and retires its + # predecessor instead of shutting it down. + routes = [ + cache.route_for(default, {"arize_space_id": str(i), "arize_api_key": "K"}) + for i in range(5) + ] + + assert len(cache._providers) == 1 + assert len(cache._retired) == 2 # capped, not one retiree per open call + assert shut_down == [routes[0].provider, routes[1].provider] + + cache.release(routes[0].provider) # already shut down: no second shutdown + assert shut_down == [routes[0].provider, routes[1].provider] + cache.release(routes[2].provider) # still draining: drains and shuts down + assert shut_down[-1] is routes[2].provider + + def test_release_without_eviction_keeps_provider_alive(monkeypatch): from litellm.integrations.otel.plumbing import routing as routing_mod