From db79b78357c4020044719724e6579e5d0d687982 Mon Sep 17 00:00:00 2001 From: Yucheng Zhu Date: Wed, 29 Jul 2026 17:26:51 -0700 Subject: [PATCH] fix(otel/v2): reclaim evicted trace exporters and skip idle resolver work Two issues surfaced in review: - TenantTracerCache and the fan-out processor cache dropped the LRU entry on eviction without shutting it down, leaking its BatchSpanProcessor worker thread for the life of the process (306 distinct destinations -> 306 live threads at a 256 cap). Shut the evicted provider/processor down on a background daemon thread, so the worker is reclaimed without blocking the request path. - The request-time destination resolver ran the per-request org/team lookup (_effective_org_id) before checking whether any logging destination exists, so a proxy not using admin-owned destinations still paid a team fetch on the auth path for team-scoped keys. Short-circuit to () when the registry holds no logging credential, before the lookup. --- litellm/integrations/otel/plumbing/routing.py | 30 +++++++-- litellm/proxy/litellm_pre_call_utils.py | 6 ++ .../integrations/otel/test_otel_v2_dynamic.py | 28 ++++---- .../proxy/test_litellm_pre_call_utils.py | 65 +++++++++++++++++++ 4 files changed, 114 insertions(+), 15 deletions(-) diff --git a/litellm/integrations/otel/plumbing/routing.py b/litellm/integrations/otel/plumbing/routing.py index 29b6454994f..63057bf6876 100644 --- a/litellm/integrations/otel/plumbing/routing.py +++ b/litellm/integrations/otel/plumbing/routing.py @@ -13,6 +13,7 @@ a backend like Arize selects its project FROM it, so two Arize projects each get tagged span instead of a last-wins merge. Empty destinations -> the logger's default (global only). """ +import threading from collections import OrderedDict from opentelemetry.context import Context @@ -34,6 +35,24 @@ _NON_OTLP_KINDS = ("console", "in_memory", "inmemory", "memory") _MAX_CACHED_PROVIDERS = 256 +def _shutdown_in_background(evicted: "TracerProvider | SpanProcessor") -> None: + """Reclaim an evicted provider/processor's ``BatchSpanProcessor`` worker thread. + + Dropping it from the cache without ``shutdown`` leaves that daemon thread running for + the life of the process (it does not "drain on its own"). ``shutdown`` force-flushes + and can do network I/O, so it runs fire-and-forget on a daemon thread rather than on + the request path; the evicted object is otherwise unreferenced. + """ + + def _run() -> None: + try: + evicted.shutdown() + except Exception as exc: # noqa: BLE001 # a failed shutdown must not surface on the hot path + verbose_logger.debug("OTel V2: background shutdown of evicted %s failed: %s", type(evicted).__name__, exc) + + threading.Thread(target=_run, name="litellm-otel-evict-shutdown", daemon=True).start() + + class TenantTracerCache: """Destination-scoped ``TracerProvider`` cache keyed by endpoint + headers.""" @@ -51,10 +70,12 @@ class TenantTracerCache: ) # mutable-ok: bounded LRU tracer-provider cache def _evict_if_full(self) -> None: - """Drop the least-recently-used provider when over capacity (no synchronous - ``shutdown``; the evicted worker drains on its own and is reclaimed at process exit).""" + """Drop the least-recently-used provider when over capacity, shutting it down off + the hot path so its ``BatchSpanProcessor`` worker thread is reclaimed rather than + leaked for the life of the process.""" if len(self._providers) > _MAX_CACHED_PROVIDERS: - self._providers.popitem(last=False) + _, evicted = self._providers.popitem(last=False) + _shutdown_in_background(evicted) def tracers_for(self, default: Tracer, destinations: "tuple[OtelDestination, ...]") -> "tuple[Tracer, ...]": """The tracers for this request's gen-AI span, one per distinct Resource group. @@ -320,5 +341,6 @@ class TenantFanOutSpanProcessor(SpanProcessor): return None self._processors[key] = processor if len(self._processors) > _MAX_CACHED_PROCESSORS: - self._processors.popitem(last=False) + _, evicted = self._processors.popitem(last=False) + _shutdown_in_background(evicted) return processor diff --git a/litellm/proxy/litellm_pre_call_utils.py b/litellm/proxy/litellm_pre_call_utils.py index 9b8a0f43590..e323f4ff20e 100644 --- a/litellm/proxy/litellm_pre_call_utils.py +++ b/litellm/proxy/litellm_pre_call_utils.py @@ -641,6 +641,12 @@ async def _resolve_logging_exporters( parse_credential_info, ) + if not any( + (info := parse_credential_info(credential.credential_info)) is not None and info.credential_type == "logging" + for credential in litellm.credential_list + ): + return (), () + team_id = user_api_key_dict.team_id org_id = await _effective_org_id(user_api_key_dict) team_ids, org_ids = identity_scope(team_id, org_id) 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 8c7088dde1d..f00a11266c1 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_dynamic.py @@ -105,12 +105,13 @@ def test_destination_set_is_order_independent(): assert len(cache._providers) == 1 -def test_provider_cache_evicts_lru_without_shutdown(monkeypatch): - """The provider cache stays bounded, and eviction drops the LRU provider WITHOUT - a synchronous ``shutdown``. Eviction runs on the request-serving path, and - ``TracerProvider.shutdown`` force-flushes/blocks and can drop a concurrent - request's in-flight spans, so it mirrors ``TenantFanOutSpanProcessor`` and lets the - evicted worker drain on its own. Restoring a shutdown-on-eviction fails this.""" +def test_provider_cache_evicts_lru_and_shuts_it_down_off_hot_path(monkeypatch): + """The provider cache stays bounded, and eviction shuts the LRU provider down so its + ``BatchSpanProcessor`` worker thread is reclaimed instead of leaked for the process + lifetime. ``shutdown`` force-flushes/can block, so it runs on a background daemon + thread rather than the request path: the evicted provider is shut down, the survivors + are not. Dropping the shutdown (the old leak) fails this.""" + import time from unittest.mock import MagicMock from litellm.integrations.otel.plumbing import routing as routing_mod @@ -129,14 +130,19 @@ def test_provider_cache_evicts_lru_without_shutdown(monkeypatch): cache = _cache("langfuse_otel") default = NoOpTracer() - cache.tracer_for(default, (_dest("https://1/v1"),)) - cache.tracer_for(default, (_dest("https://2/v1"),)) + cache.tracer_for(default, (_dest("https://1/v1"),)) # created[0] + cache.tracer_for(default, (_dest("https://2/v1"),)) # created[1] cache.tracer_for(default, (_dest("https://1/v1"),)) # touch "1" -> "2" is LRU - cache.tracer_for(default, (_dest("https://3/v1"),)) # overflow -> evict "2" + cache.tracer_for(default, (_dest("https://3/v1"),)) # created[2]: overflow -> evict "2" assert len(cache._providers) == 2 - for provider in created: - provider.shutdown.assert_not_called() + # eviction shuts down off the hot path, so wait for the background daemon thread + deadline = time.time() + 5 + while not created[1].shutdown.called and time.time() < deadline: + time.sleep(0.02) + created[1].shutdown.assert_called() # the evicted LRU ("2") is reclaimed + created[0].shutdown.assert_not_called() # survivor + created[2].shutdown.assert_not_called() # survivor # --- fan-out: keep the configured exporters, append one per destination ----- # diff --git a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py index 99a70eb24ec..9060e5a7443 100644 --- a/tests/test_litellm/proxy/test_litellm_pre_call_utils.py +++ b/tests/test_litellm/proxy/test_litellm_pre_call_utils.py @@ -5622,3 +5622,68 @@ def test_warn_stale_team_alias_once_evicts_oldest_key_beyond_cap(monkeypatch): pre_call_utils._warn_stale_team_alias_once("key-3", "stale alias") assert list(pre_call_utils._STALE_TEAM_ALIAS_WARNING_KEYS) == ["key-2", "key-3"] + + +@pytest.mark.asyncio +async def test_resolve_logging_exporters_short_circuits_without_destinations(monkeypatch): + """With no logging destination in the registry, the resolver returns empty and does NOT + run the per-request org/team lookup, so a proxy not using admin-owned destinations pays + nothing on the auth path for team-scoped keys.""" + from litellm.proxy import litellm_pre_call_utils as pcu + + monkeypatch.setattr( + litellm, + "credential_list", + [ + CredentialItem( + credential_name="openai", + credential_values={"api_key": "sk"}, + credential_info={"custom_llm_provider": "openai"}, + ) + ], + ) + lookups = {"org": 0} + + async def _spy_effective_org_id(user_api_key_dict): + lookups["org"] += 1 + return None + + monkeypatch.setattr(pcu, "_effective_org_id", _spy_effective_org_id) + + key = UserAPIKeyAuth(api_key="k", team_id="t1") + destinations, backends = await pcu._resolve_logging_exporters(key) + + assert destinations == () and backends == () + assert lookups["org"] == 0 # the org/team lookup was skipped entirely + + +@pytest.mark.asyncio +async def test_resolve_logging_exporters_runs_lookup_when_a_destination_exists(monkeypatch): + """The short-circuit must not skip resolution when a destination exists: a global + destination is still resolved for a team-scoped key, and the org lookup runs.""" + from litellm.proxy import litellm_pre_call_utils as pcu + + monkeypatch.setattr( + litellm, + "credential_list", + [ + CredentialItem( + credential_name="d-global", + credential_values={"otel_endpoint": "https://collector/v1/traces"}, + credential_info={"credential_type": "logging", "description": "generic", "access": {"global": True}}, + ) + ], + ) + lookups = {"org": 0} + + async def _spy_effective_org_id(user_api_key_dict): + lookups["org"] += 1 + return None + + monkeypatch.setattr(pcu, "_effective_org_id", _spy_effective_org_id) + + key = UserAPIKeyAuth(api_key="k", team_id="t1") + destinations, backends = await pcu._resolve_logging_exporters(key) + + assert lookups["org"] == 1 # a destination exists, so the resolver runs the lookup + assert "generic" in backends # global access grants the team key