fix(langfuse): retry raised OTLP exports and honor LANGFUSE_TIMEOUT

The OTLP http exporter only retries 429 and 5xx; a connect or read timeout
propagates and BatchSpanProcessor drops the batch. Wrap the exporter in
RetryingSpanExporter (three backoff retries, as the v2 consumer did) and
build it on every path so the default and private-CA deployments share the
same channel, timeout and retry behaviour

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/integrations/langfuse/langfuse_sdk.py

@@ -4,13 +4,15 @@ import os
 import re
 import threading
 from base64 import b64encode
-from collections.abc import Callable, Generator, Mapping
+from collections.abc import Callable, Generator, Mapping, Sequence
 from contextlib import contextmanager
 from contextvars import ContextVar
+from dataclasses import dataclass
 from datetime import datetime
 from hashlib import sha256
 from importlib.metadata import version
 from itertools import chain
+from time import sleep
 from types import MappingProxyType
 from typing import Final
 from weakref import WeakKeyDictionary, WeakSet
@@ -20,16 +22,20 @@ from langfuse import Langfuse, LangfuseGeneration, LangfuseSpan, propagate_attri
 from langfuse._client.resource_manager import LangfuseResourceManager
 from opentelemetry.context import Context
 from opentelemetry.sdk.resources import Resource
-from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
 from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
 from opentelemetry.sdk.trace.id_generator import RandomIdGenerator
 from opentelemetry.sdk.trace.sampling import TraceIdRatioBased
+from requests import RequestException
+
+from litellm._logging import verbose_logger
 
 __all__ = (
     "AS_ROOT_ATTRIBUTE",
     "PUBLIC_ATTRIBUTE",
     "RELEASE_ATTRIBUTE",
     "DiscardingSpanExporter",
+    "RetryingSpanExporter",
     "acquire_langfuse_client",
     "build_isolated_tracer_provider",
     "evict_stale_langfuse_resources",
@@ -247,6 +253,38 @@ class DiscardingSpanExporter(SpanExporter):
         return True
 
 
+@dataclass(frozen=True, slots=True)
+class RetryingSpanExporter(SpanExporter):
+    """Retry a batch whose HTTP round trip raised, as the v2 ingestion consumer did.
+
+    The OTLP http exporter only retries 429 and 5xx responses; a connect or
+    read timeout propagates, and ``BatchSpanProcessor`` drops the whole batch
+    on any exception. A destination that stalls for a few seconds therefore
+    lost every observation in flight, where v2 backed off three times first.
+    """
+
+    exporter: SpanExporter
+    delays: Sequence[float] = (1.0, 2.0, 4.0)
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        for delay in chain(self.delays, (None,)):
+            try:
+                return self.exporter.export(spans)
+            except RequestException as error:
+                if delay is None:
+                    verbose_logger.error("Langfuse export failed after %d retries: %s", len(self.delays), error)
+                    return SpanExportResult.FAILURE
+                verbose_logger.warning("Langfuse export raised %s, retrying in %ss", error, delay)
+                sleep(delay)
+        return SpanExportResult.FAILURE
+
+    def shutdown(self) -> None:
+        self.exporter.shutdown()
+
+    def force_flush(self, timeout_millis: int = 30_000) -> bool:
+        return self.exporter.force_flush(timeout_millis)
+
+
 _LIVE_CLIENTS_LOCK: Final = threading.Lock()
 # litellm clients still using each SDK resource bundle; the bundle is torn down with the last one.
 # Both sides are weak so a throwaway client (a health probe, an alerting lookup) that is simply
@@ -446,17 +484,17 @@ def evict_stale_langfuse_resources(*, public_key: str | None, secret_key: str |
     _retire_orphaned_providers()
 
 
-def _build_verified_span_exporter(*, public_key: object, secret_key: object, base_url: object) -> SpanExporter | None:
-    """Rebuild litellm's TLS material onto the export channel.
+def _build_span_exporter(*, public_key: object, secret_key: object, base_url: object) -> RetryingSpanExporter:
+    """Build the OTLP export channel with litellm's TLS material and v2's retry behaviour.
 
     v2 ingested through the injected httpx client, which carried litellm's CA
     bundle and client certificate; v4 ships every observation through its own
     OTLP exporter, so a private-CA deployment would fail TLS on every export in
     a background thread while ``auth_check`` (still on the httpx client) stays
-    green. Only built when TLS is configured away from the default (a CA bundle,
-    a client certificate, or verification switched off); endpoint and headers
-    mirror ``langfuse._client.span_processor``.
+    green. Endpoint, headers and timeout mirror ``langfuse._client.span_processor``.
     """
+    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
+
     import litellm
     from litellm.llms.custom_httpx.http_handler import get_ssl_verify
 
@@ -464,10 +502,6 @@ def _build_verified_span_exporter(*, public_key: object, secret_key: object, bas
     ca_bundle: Final = ssl_verify if isinstance(ssl_verify, str) and os.path.exists(ssl_verify) else None
     configured_certificate: Final = os.getenv("SSL_CERTIFICATE") or litellm.ssl_certificate
     client_certificate: Final = configured_certificate if isinstance(configured_certificate, str) else None
-    if ssl_verify is not False and ca_bundle is None and client_certificate is None:
-        return None
-    from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
-
     export_path: Final = os.getenv("LANGFUSE_OTEL_TRACES_EXPORT_PATH")
     endpoint: Final = f"{base_url}/{export_path}" if export_path else f"{base_url}/api/public/otel/v1/traces"
     encoded_auth: Final = b64encode(f"{public_key}:{secret_key}".encode()).decode("ascii")
@@ -479,12 +513,13 @@ def _build_verified_span_exporter(*, public_key: object, secret_key: object, bas
             "x-langfuse-sdk-version": version("langfuse"),
             "x-langfuse-public-key": str(public_key),
         },
+        timeout=int(os.getenv("LANGFUSE_TIMEOUT", "5")),
         certificate_file=ca_bundle,
         client_certificate_file=client_certificate,
     )
     if ssl_verify is False:
         exporter._certificate_file = False  # pyright: ignore[reportPrivateUsage]  # the ctor coerces a False certificate_file back to True
-    return exporter
+    return RetryingSpanExporter(exporter)
 
 
 def acquire_langfuse_client(
@@ -508,7 +543,7 @@ def acquire_langfuse_client(
     span_exporter: Final = (
         DiscardingSpanExporter()
         if mock_mode
-        else _build_verified_span_exporter(
+        else _build_span_exporter(
             public_key=public_key,
             secret_key=parameters.get("secret_key"),
             base_url=parameters.get("base_url"),

tests/test_litellm/integrations/langfuse/test_langfuse_sdk.py

@@ -998,35 +998,83 @@ def test_a_sweep_overlapping_registration_and_rotation_keeps_the_live_provider()
     assert _exports(client, exporter, "after-racing-sweep")
 
 
-def test_ssl_exporter_is_only_built_with_custom_tls_material(monkeypatch, tmp_path):
+def test_ssl_exporter_carries_litellm_tls_material(monkeypatch, tmp_path):
     """v4 exports over its own OTLP channel, so litellm's CA bundle must be rebuilt onto it."""
     import litellm
-    from litellm.integrations.langfuse.langfuse_sdk import _build_verified_span_exporter
+    from litellm.integrations.langfuse.langfuse_sdk import _build_span_exporter
 
-    for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"):
+    for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE", "LANGFUSE_TIMEOUT"):
         monkeypatch.delenv(name, raising=False)
     monkeypatch.setattr(litellm, "ssl_verify", True)
     monkeypatch.setattr(litellm, "ssl_certificate", None)
-    assert (
-        _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example") is None
-    )
+    default = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
+    assert default._certificate_file is True
+    assert default._timeout == 5
 
     ca_path = tmp_path / "private-ca.pem"
     ca_path.write_text("dummy")
     monkeypatch.setattr(litellm, "ssl_verify", str(ca_path))
-    exporter = _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
-    assert exporter is not None
+    monkeypatch.setenv("LANGFUSE_TIMEOUT", "20")
+    exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
     assert exporter._endpoint == "https://lf.internal.example/api/public/otel/v1/traces"
     assert exporter._certificate_file == str(ca_path)
+    assert exporter._timeout == 20
     assert exporter._headers["x-langfuse-public-key"] == "pk"
     assert exporter._headers["x-langfuse-sdk-version"] == installed_langfuse_version()
 
 
+def test_retrying_exporter_retries_a_raised_export_and_then_succeeds(monkeypatch):
+    """A read timeout used to drop the batch outright; v2 backed off and re-sent it."""
+    from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+    from requests import ReadTimeout
+
+    from litellm.integrations.langfuse.langfuse_sdk import RetryingSpanExporter
+
+    attempts = []
+    slept = []
+
+    class Flaky(SpanExporter):
+        def export(self, spans):
+            attempts.append(spans)
+            if len(attempts) < 3:
+                raise ReadTimeout("destination stalled")
+            return SpanExportResult.SUCCESS
+
+    monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
+    result = RetryingSpanExporter(Flaky(), delays=(0.5, 1.5, 2.5)).export(("span",))
+
+    assert result is SpanExportResult.SUCCESS
+    assert attempts == [("span",)] * 3
+    assert slept == [0.5, 1.5]
+
+
+def test_retrying_exporter_gives_up_after_the_last_delay(monkeypatch):
+    from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult
+    from requests import ConnectionError as RequestsConnectionError
+
+    from litellm.integrations.langfuse.langfuse_sdk import RetryingSpanExporter
+
+    attempts = []
+    slept = []
+
+    class Down(SpanExporter):
+        def export(self, spans):
+            attempts.append(spans)
+            raise RequestsConnectionError("refused")
+
+    monkeypatch.setattr("litellm.integrations.langfuse.langfuse_sdk.sleep", slept.append)
+    result = RetryingSpanExporter(Down(), delays=(1.0, 2.0)).export(("span",))
+
+    assert result is SpanExportResult.FAILURE
+    assert len(attempts) == 3
+    assert slept == [1.0, 2.0]
+
+
 @pytest.mark.parametrize("switch", ["attribute", "env"])
 def test_ssl_exporter_disables_verification_when_litellm_does(monkeypatch, switch):
     """v2 exported through the httpx client, so ``ssl_verify=False`` reached ingestion; v4's exporter must match."""
     import litellm
-    from litellm.integrations.langfuse.langfuse_sdk import _build_verified_span_exporter
+    from litellm.integrations.langfuse.langfuse_sdk import _build_span_exporter
 
     for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"):
         monkeypatch.delenv(name, raising=False)
@@ -1037,8 +1085,7 @@ def test_ssl_exporter_disables_verification_when_litellm_does(monkeypatch, switc
         monkeypatch.setattr(litellm, "ssl_verify", True)
         monkeypatch.setenv("SSL_VERIFY", "False")
 
-    exporter = _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
-    assert exporter is not None
+    exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
     assert exporter._certificate_file is False
     assert exporter._client_cert is None
 
@@ -1060,7 +1107,7 @@ def test_ssl_exporter_falls_back_to_default_ca_when_the_bundle_path_is_missing(
 ):
     """The httpx client ignores a CA path that does not exist; handing it to requests would fail every export."""
     import litellm
-    from litellm.integrations.langfuse.langfuse_sdk import _build_verified_span_exporter
+    from litellm.integrations.langfuse.langfuse_sdk import _build_span_exporter
 
     for name in ("SSL_CERTIFICATE", "SSL_VERIFY", "SSL_CERT_FILE"):
         monkeypatch.delenv(name, raising=False)
@@ -1070,13 +1117,9 @@ def test_ssl_exporter_falls_back_to_default_ca_when_the_bundle_path_is_missing(
     client_cert.write_text("dummy")
     monkeypatch.setattr(litellm, "ssl_certificate", str(client_cert) if with_client_certificate else None)
 
-    exporter = _build_verified_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example")
-    if not with_client_certificate:
-        assert exporter is None
-        return
-    assert exporter is not None
+    exporter = _build_span_exporter(public_key="pk", secret_key="sk", base_url="https://lf.internal.example").exporter
     assert exporter._certificate_file is True
-    assert exporter._client_cert == str(client_cert)
+    assert exporter._client_cert == (str(client_cert) if with_client_certificate else None)
 
 
 def test_second_client_on_the_same_key_does_not_build_another_provider():