fix(otel): rebuild an anchored destination's processor past drain saturation

A destination deliverable() accepted at auth can be evicted by other tenants' auths
before its request's spans end, and that eviction is what tips the drain over. The
saturation gate then refused the rebuild at on_end, and with the operator's exporter
already stood down for that backend the span went nowhere. The gate now applies only
while a request decides whether to anchor

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

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."""