fix(otel): hold destination eviction while the drain is saturated

An anchored destination evicted by other tenants' auths is rebuilt on its
next span, and that rebuild evicted another anchored one, so with more
destinations in flight than the cache holds every span cost one more
processor, one more batch thread and one more close queued behind a collector
that never answers. Eviction now holds while the drain is saturated, so the
cache keeps one entry per destination in flight and trims back to its cap on
the next hit or build once the drain has room

litellm/integrations/otel/plumbing/providers.py

@@ -578,12 +578,14 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         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.
+        what tips the drain over. Eviction holds while the drain is saturated, so such a
+        rebuild costs the cache one entry rather than shedding another processor, and
+        the total stays at one per destination in flight.
         """
         key: Final = destination.cache_key()
         if (cached := self._processors.get(key)) is not None:
             self._processors.move_to_end(key)
+            self._retire_overflow_locked()
             return cached
         if not anchored and self._drain.saturated():
             verbose_logger.debug("OTel V2 fan-out: drain saturated, not building for %s", destination.endpoint)
@@ -612,8 +614,17 @@ class TenantFanOutSpanProcessor(SpanProcessor):
                 self._drain.submit(retired)
 
     def _retire_overflow_locked(self) -> None:
-        """Move the LRU processor out of the cache once it is past the cap."""
-        if len(self._processors) <= _MAX_CACHED_DESTINATION_PROCESSORS:
+        """Move the LRU processor out of the cache once it is past the cap, drain permitting.
+
+        Eviction is what feeds the drain, and a destination a request already anchored
+        is rebuilt on its next span, which would shed another one. While the shed ones
+        are stuck closing against a collector that stopped answering, evicting would
+        churn the cache at one more processor, and one more batch thread, per span.
+        Holding above the cap instead keeps the total at one processor per destination
+        in flight, since ``deliverable`` anchors no new destination while the drain is
+        saturated. Once it has room again, every hit and build trims one entry.
+        """
+        if len(self._processors) <= _MAX_CACHED_DESTINATION_PROCESSORS or self._drain.saturated():
             return
         _, evicted = self._processors.popitem(last=False)
         self._retired[id(evicted)] = evicted

tests/test_litellm/integrations/otel/test_otel_v2_destinations.py

@@ -1828,6 +1828,86 @@ class TestEvictionSafety:
         finally:
             release.set()
 
+    def _saturated_by_anchoring(self, pending_drains, extra):
+        """A fan-out whose drain ``extra`` anchorings past the cache cap have saturated.
+
+        Returns it with the processors built, the destinations that anchored, and the
+        event that lets the blocked closes finish.
+        """
+        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=pending_drains)
+        destinations = tuple(self._dest(index) for index in range(_MAX_CACHED_DESTINATION_PROCESSORS + extra))
+        anchored = tuple(destination for destination in destinations if fan_out.deliverable((destination,)))
+        assert fan_out._drain.saturated(), "anchoring past the cap did not saturate the drain"
+        assert len(anchored) > _MAX_CACHED_DESTINATION_PROCESSORS, "not enough destinations in flight to churn"
+        return fan_out, built, anchored, release
+
+    def test_anchored_rebuilds_under_a_saturated_drain_do_not_grow_with_the_spans(self):
+        """Every anchored rebuild past the cap evicts another anchored destination, whose
+        next span rebuilds it in turn. With more destinations in flight than the cache
+        holds, each span would then cost one more processor, one more batch thread and
+        one more close queued behind a collector that never answers."""
+        from litellm.integrations.otel.plumbing.providers import _MAX_CACHED_DESTINATION_PROCESSORS
+
+        fan_out, built, anchored, release = self._saturated_by_anchoring(pending_drains=4, extra=8)
+        try:
+            after_anchoring = len(built)
+            for _ in range(5):
+                for destination in anchored:
+                    fan_out._release(fan_out._acquire(destination))
+
+            rebuilt = len(built) - after_anchoring
+            assert rebuilt == len(anchored) - _MAX_CACHED_DESTINATION_PROCESSORS, (
+                f"{rebuilt} rebuilds over 5 rounds of {len(anchored)} anchored destinations: one per evicted one expected"
+            )
+            assert len(fan_out._processors) == len(anchored), "an anchored destination was shed under a saturated drain"
+            assert all(destination in fan_out.deliverable((destination,)) for destination in anchored)
+        finally:
+            release.set()
+
+    def test_the_cache_returns_to_its_cap_once_the_drain_has_room(self):
+        """Holding above the cap is for the outage only: with the drain caught up, the
+        entries kept for the destinations in flight are the ones to shed."""
+        from litellm.integrations.otel.plumbing.providers import _MAX_CACHED_DESTINATION_PROCESSORS
+
+        fan_out, built, anchored, release = self._saturated_by_anchoring(pending_drains=4, extra=8)
+        for destination in anchored:
+            fan_out._release(fan_out._acquire(destination))
+        assert len(fan_out._processors) > _MAX_CACHED_DESTINATION_PROCESSORS
+
+        release.set()
+        for _ in range(500):
+            for destination in anchored[-4:]:
+                fan_out._release(fan_out._acquire(destination))
+            if len(fan_out._processors) <= _MAX_CACHED_DESTINATION_PROCESSORS:
+                break
+            time.sleep(0.02)
+
+        assert len(fan_out._processors) == _MAX_CACHED_DESTINATION_PROCESSORS, "the cache never came back to its cap"
+        shed = len(built) - _MAX_CACHED_DESTINATION_PROCESSORS
+        for _ in range(500):
+            if sum(processor.shutdown_calls for processor in built) == shed:
+                break
+            time.sleep(0.02)
+
+        assert sum(processor.shutdown_calls for processor in built) == shed, "a shed processor was never closed"
+
     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."""