diff --git a/litellm/integrations/otel/plumbing/providers.py b/litellm/integrations/otel/plumbing/providers.py index df8e449bc42..0c95fcb3d61 100644 --- a/litellm/integrations/otel/plumbing/providers.py +++ b/litellm/integrations/otel/plumbing/providers.py @@ -542,7 +542,7 @@ class TenantFanOutSpanProcessor(SpanProcessor): with self._lock: if self._closed: return False - built: Final = self._cached_or_built_locked(destination) + built: Final = self._cached_or_built_locked(destination, anchored=False) drained: Final = self._drainable_locked() for shed in drained: self._drain.submit(shed) @@ -559,7 +559,7 @@ class TenantFanOutSpanProcessor(SpanProcessor): with self._lock: if self._closed: return None - processor: Final = self._cached_or_built_locked(destination) + processor: Final = self._cached_or_built_locked(destination, anchored=True) if processor is None: return None self._exporting[id(processor)] = self._exporting.get(id(processor), 0) + 1 @@ -568,24 +568,29 @@ class TenantFanOutSpanProcessor(SpanProcessor): self._drain.submit(shed) return processor - def _cached_or_built_locked(self, destination: "OtelDestination") -> SpanProcessor | None: + def _cached_or_built_locked(self, destination: "OtelDestination", *, anchored: bool) -> SpanProcessor | None: + """The cached processor for ``destination``, or a new one if the drain can take it. + + Every build past the cache cap sheds one processor into the drain, so while the + shed ones are stuck closing against a collector that stopped answering, a + destination that is not yet anchored is refused rather than parked behind them: + ``deliverable`` then leaves its spans with the operator's exporter until the + drain catches up. One the request already anchored is rebuilt regardless. The + operator's exporter has stood down for it, so refusing here would drop the span, + and other tenants' auths can evict it in the meantime, with that eviction being + what tips the drain over. Those rebuilds are bounded by the requests in flight, + since anchoring itself stops once the drain is full. + """ key: Final = destination.cache_key() if (cached := self._processors.get(key)) is not None: self._processors.move_to_end(key) return cached + if not anchored and self._drain.saturated(): + verbose_logger.debug("OTel V2 fan-out: drain saturated, not building for %s", destination.endpoint) + return None return self._build_locked(destination, key) def _build_locked(self, destination: "OtelDestination", key: object) -> SpanProcessor | None: - """Build and cache a processor for ``destination``, unless the drain is saturated. - - Every build past the cache cap sheds one processor into the drain, so while the - shed ones are stuck closing against a collector that stopped answering, a new - destination is refused rather than parked behind them: ``deliverable`` then - leaves its spans with the operator's exporter until the drain catches up. - """ - if self._drain.saturated(): - verbose_logger.debug("OTel V2 fan-out: drain saturated, not building for %s", destination.endpoint) - return None built: Final = self._build(destination) if built is None: return None diff --git a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py index 0f74cb4d9d0..54946febf1d 100644 --- a/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py +++ b/tests/test_litellm/integrations/otel/test_otel_v2_destinations.py @@ -1760,12 +1760,12 @@ class TestEvictionSafety: fan_out = TenantFanOutSpanProcessor(processor_factory=factory, pending_drains=3) try: - for index in range(_MAX_CACHED_DESTINATION_PROCESSORS + 40): - processor = fan_out._acquire(self._dest(index)) - if processor is not None: - fan_out._release(processor) + anchored = tuple( + fan_out.deliverable((self._dest(index),)) for index in range(_MAX_CACHED_DESTINATION_PROCESSORS + 40) + ) assert len(built) == _MAX_CACHED_DESTINATION_PROCESSORS + 3, "a processor per request during the outage" + assert sum(1 for accepted in anchored if accepted) == len(built), "anchored what it could not build" assert fan_out.deliverable((self._dest(999),)) == (), "the span would vanish instead of staying with the operator" finally: release.set() @@ -1776,6 +1776,58 @@ class TestEvictionSafety: assert fan_out.deliverable((self._dest(999),)) == (self._dest(999),), "the fan-out never recovered" + def test_an_anchored_destination_evicted_under_a_saturated_drain_still_gets_the_span(self): + """``deliverable`` accepted the destination, so the operator's exporter has stood + down for it. Other tenants' auths can then evict it, and the eviction is what + tips the drain into saturation, so refusing the rebuild at ``on_end`` would drop + the span outright.""" + import threading + + from litellm.integrations.otel.plumbing.providers import _MAX_CACHED_DESTINATION_PROCESSORS + + release = threading.Event() + + class Blocking(self.Recording): + def shutdown(self): + release.wait(timeout=10) + super().shutdown() + + built = [] + + def factory(_destination): + built.append(Blocking()) + return built[-1] + + fan_out = TenantFanOutSpanProcessor( + processor_factory=factory, pending_drains=_MAX_CACHED_DESTINATION_PROCESSORS + 1 + ) + provider = TracerProvider() + provider.add_span_processor(fan_out) + tracer = get_tracer(provider, "litellm") + anchored = self._dest(0) + try: + for index in range(1, _MAX_CACHED_DESTINATION_PROCESSORS + 1): + assert fan_out.deliverable((self._dest(index),)) + assert fan_out.deliverable((anchored,)) == (anchored,) + first = built[-1] + for index in range(_MAX_CACHED_DESTINATION_PROCESSORS + 1, 2 * _MAX_CACHED_DESTINATION_PROCESSORS + 1): + assert fan_out.deliverable((self._dest(index),)) + assert fan_out._drain.saturated(), "the anchored destination's own eviction saturates the drain" + assert first not in fan_out._processors.values(), "the anchored destination was not evicted" + + def run(): + set_request_destinations((anchored,)) + with tracer.start_as_current_span("chat anthropic"): + pass + + before = len(built) + in_fresh_context(run) + assert len(built) == before + 1, "the anchored destination was not rebuilt, so its span went nowhere" + assert [span.name for span in built[-1].span_exporter.get_finished_spans()] == ["chat anthropic"] + assert first.span_exporter.get_finished_spans() == (), "the shed processor was handed out again" + finally: + release.set() + def test_concurrent_eviction_cannot_build_between_retirement_and_drain_submission(self): """A second request cannot build while the first eviction is being handed to the drain, or concurrent churn can outrun the pending-drain limit."""