fix(otel v2): bound the teardown that closes a destination, not the one that never blocks

The five-second bound guarded the wait for spans still inside on_end, but a
batching processor's on_end only queues the span and returns, so that counter is
empty and the bound engaged against nothing. The blocking half was the serial
close, which flushes over the network and joins the SDK's own worker thread with
no timeout of its own, so a single tenant collector that answers and never
finishes held process teardown open for as long as it liked.

Hand every close to the drain, whose workers are daemons, and give the whole
teardown one deadline.

litellm/integrations/otel/plumbing/providers.py

@@ -2,6 +2,7 @@
 
 import queue
 import threading
+import time
 from collections import OrderedDict
 from collections.abc import Callable, Iterable, Mapping
 from types import MappingProxyType
@@ -250,10 +251,12 @@ class _DrainPool:
         self._lock: Final = threading.Lock()
         self._closed = False
         self._pending: Final[queue.Queue[SpanProcessor | None]] = pending if pending is not None else queue.Queue()
-        for _ in range(workers):
-            threading.Thread(
-                target=self._drain_until_closed, daemon=True, name="litellm-otel-destination-drain"
-            ).start()
+        self._threads: Final = tuple(
+            threading.Thread(target=self._drain_until_closed, daemon=True, name="litellm-otel-destination-drain")
+            for _ in range(workers)
+        )
+        for worker in self._threads:
+            worker.start()
 
     def submit(self, processor: SpanProcessor) -> None:
         """Queue ``processor`` for closing, or close it here once the pool is retired.
@@ -268,11 +271,15 @@ class _DrainPool:
                 return
         _shutdown_quietly(processor)
 
-    def close(self) -> None:
+    def close(self, timeout: float | None = None) -> None:
         """Retire the workers once they have closed everything already queued.
 
         A proxy that rebuilds its telemetry builds another fan-out, so workers that
         outlive the one that started them are two more threads per reload, forever.
+
+        ``timeout`` bounds how long the caller waits for that draining to finish. The
+        workers are daemons, so whatever is still flushing when it expires is dropped
+        by the interpreter rather than holding it open.
         """
         with self._lock:
             if self._closed:
@@ -280,6 +287,11 @@ class _DrainPool:
             self._closed = True
             for _ in range(self._workers):
                 self._pending.put(None)
+        if timeout is None:
+            return
+        deadline: Final = time.monotonic() + timeout
+        for worker in self._threads:
+            worker.join(timeout=max(0.0, deadline - time.monotonic()))
 
     def _drain_until_closed(self) -> None:
         while True:
@@ -388,12 +400,18 @@ class TenantFanOutSpanProcessor(SpanProcessor):
         while the SDK is tearing the provider down, so closing blind would drop a
         trace mid-forward and would hand the next caller a fresh exporter nothing
         will ever close. Refusing new work and then waiting out the in-flight ones
-        keeps both from happening. The wait has to be bounded, or one destination
-        whose collector stopped answering would hold the proxy open on the way down,
-        so a straggler past the bound is retired instead of closed: the thread still
-        exporting it closes it through the drain as soon as its export returns, and
-        no span is ever dropped mid-forward.
+        keeps both from happening. A straggler past the bound is retired instead of
+        closed: the thread still exporting it closes it through the drain as soon as
+        its export returns, so no span is dropped mid-forward.
+
+        Every close then goes to the drain rather than running here. Closing a
+        destination processor flushes it over the network and the SDK joins its own
+        worker with no timeout of its own, so one tenant collector that answers but
+        never finishes a response would otherwise hold process teardown open for as
+        long as it likes. The drain's workers are daemons, and the whole teardown
+        shares one deadline.
         """
+        deadline: Final = time.monotonic() + self._drain_seconds
         with self._lock:
             self._closed = True
             self._lock.wait_for(lambda: not self._exporting, timeout=self._drain_seconds)
@@ -404,11 +422,8 @@ class TenantFanOutSpanProcessor(SpanProcessor):
                 (ident, p) for ident, p in live if ident in self._exporting
             )
         for processor in closing:
-            try:
-                processor.shutdown()
-            except Exception as exc:  # noqa: BLE001  # one processor's shutdown must not abort the rest
-                verbose_logger.debug("OTel V2 fan-out: processor shutdown failed: %s", exc)
-        self._drain.close()
+            self._drain.submit(processor)
+        self._drain.close(timeout=max(0.0, deadline - time.monotonic()))
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         results: Final = tuple(self._flush_one(processor, timeout_millis) for processor in self._snapshot())

tests/test_litellm/integrations/otel/test_otel_v2_destinations.py

@@ -1246,6 +1246,34 @@ class TestEvictionSafety:
         assert built[0].shutdown_calls == 1, "the exporter outlived the fan-out"
         assert fan_out._processors == {}, "an exporter was left in a cleared cache"
 
+    def test_shutdown_returns_when_a_destination_never_finishes_closing(self):
+        """Closing an exporter flushes over the network and the SDK joins its own
+        worker with no timeout, so a tenant collector that answers but never finishes
+        a response would hold process teardown open for as long as it likes."""
+        import threading
+
+        never = threading.Event()
+
+        class Stuck(self.Recording):
+            def shutdown(self):
+                never.wait()
+
+        built = []
+
+        def factory(_destination):
+            built.append(Stuck())
+            return built[-1]
+
+        fan_out = TenantFanOutSpanProcessor(processor_factory=factory, shutdown_drain_seconds=0.3)
+        fan_out._release(fan_out._acquire(self._dest(0)))
+        returned = threading.Event()
+        threading.Thread(target=lambda: (fan_out.shutdown(), returned.set()), daemon=True).start()
+
+        came_back = returned.wait(timeout=8)
+        never.set()
+
+        assert came_back, "shutdown never returned while a collector held its exporter open"
+
     def test_a_cold_cache_met_by_a_burst_builds_one_processor_per_destination(self):
         """Building outside the cache lock let every thread of the burst construct its
         own exporter, each with a batch thread and a connection pool, and shed all but