feat(otel): cap oversized export batches via LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES

langfuse/_client/client.py

@@ -268,7 +268,7 @@ def mask_otel_spans(
         additional_headers (Optional[Dict[str, str]]): Additional headers to include in all API requests and in the default OTLPSpanExporter requests. These headers will be merged with default headers. Note: If httpx_client is provided, additional_headers must be set directly on your custom httpx_client as well. If `span_exporter` is provided, these headers are not wired into that exporter and must be configured on the exporter instance directly.
         tracer_provider(Optional[TracerProvider]): OpenTelemetry TracerProvider to use for Langfuse. This can be useful to set to have disconnected tracing between Langfuse and other OpenTelemetry-span emitting libraries. Note: To track active spans, the context is still shared between TracerProviders. This may lead to broken trace trees.
         id_generator (Optional[IdGenerator]): OpenTelemetry ID generator to use when Langfuse creates its own TracerProvider. If omitted, the OpenTelemetry SDK default is used. If `tracer_provider` is provided, or an OpenTelemetry TracerProvider is already registered globally, configure the ID generator on that provider instead.
-        span_exporter (Optional[SpanExporter]): Custom OpenTelemetry span exporter for the Langfuse span processor. If omitted, Langfuse creates an OTLPSpanExporter pointed at the Langfuse OTLP endpoint. If provided, Langfuse does not wire `base_url`, exporter headers, exporter auth, or exporter timeout into it. Configure endpoint, headers, and timeout on the exporter instance directly. If you are sending spans to Langfuse v4 or using Langfuse Cloud Fast Preview, include `x-langfuse-ingestion-version=4` on the exporter to enable real time processing of exported spans.
+        span_exporter (Optional[SpanExporter]): Custom OpenTelemetry span exporter for the Langfuse span processor. If omitted, Langfuse creates an OTLPSpanExporter pointed at the Langfuse OTLP endpoint. If provided, Langfuse does not wire `base_url`, exporter headers, exporter auth, exporter timeout, or the `LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES` request size limit into it. Configure endpoint, headers, and timeout on the exporter instance directly. If you are sending spans to Langfuse v4 or using Langfuse Cloud Fast Preview, include `x-langfuse-ingestion-version=4` on the exporter to enable real time processing of exported spans.
 
     Example:
         ```python

langfuse/_client/environment_variables.py

@@ -62,6 +62,20 @@
 **Default value:** ``/api/public/otel/v1/traces``
 """
 
+LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES = "LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES"
+"""
+.. envvar:: LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES
+
+Maximum serialized size in bytes (before compression) of a span batch exported by the
+default OTLP exporter. Oversized batches are dropped before sending and logged as a
+failed export. Must be a positive integer. Custom span exporters are not affected.
+
+Requires ``opentelemetry-exporter-otlp-proto-http>=1.45.0``; older versions do not
+enforce a limit.
+
+**Default value:** ``67108864`` (64 MiB)
+"""
+
 LANGFUSE_DEBUG = "LANGFUSE_DEBUG"
 """
 .. envvar:: LANGFUSE_DEBUG

langfuse/_client/span_processor.py

@@ -12,10 +12,11 @@
 """
 
 import base64
+import inspect
 import logging
 import os
 import threading
-from typing import Callable, Dict, List, Optional, cast
+from typing import Any, Callable, Dict, List, Optional, cast
 
 from opentelemetry import context as context_api
 from opentelemetry.context import Context
@@ -28,6 +29,7 @@
 from langfuse._client.environment_variables import (
     LANGFUSE_FLUSH_AT,
     LANGFUSE_FLUSH_INTERVAL,
+    LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES,
     LANGFUSE_OTEL_TRACES_EXPORT_PATH,
 )
 from langfuse._client.propagation import (
@@ -47,6 +49,32 @@
 from langfuse.types import MaskOtelSpansFunction
 
 
+def _resolve_max_batch_size_bytes() -> Optional[int]:
+    """Return the configured batch byte limit, or None to keep the exporter default."""
+    raw_value = os.environ.get(LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES, "").strip()
+    if not raw_value:
+        return None
+
+    if raw_value.isascii() and raw_value.isdigit() and int(raw_value) > 0:
+        return int(raw_value)
+
+    langfuse_logger.warning(
+        "Invalid %s=%r. Expected a positive integer. Using the default limit.",
+        LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES,
+        raw_value,
+    )
+    return None
+
+
+def _otlp_exporter_supports_max_request_size() -> bool:
+    try:
+        parameters = inspect.signature(OTLPSpanExporter.__init__).parameters
+    except (TypeError, ValueError):
+        return False
+
+    return "max_request_size" in parameters
+
+
 class LangfuseSpanProcessor(BatchSpanProcessor):
     """OpenTelemetry span processor that exports spans to the Langfuse API.
 
@@ -123,10 +151,23 @@ def __init__(
                 else f"{base_url}/api/public/otel/v1/traces"
             )
 
+            exporter_kwargs: Dict[str, Any] = {}
+            max_request_size = _resolve_max_batch_size_bytes()
+            if max_request_size is not None:
+                if _otlp_exporter_supports_max_request_size():
+                    exporter_kwargs["max_request_size"] = max_request_size
+                else:
+                    langfuse_logger.warning(
+                        "%s is set but not enforced. It requires "
+                        "opentelemetry-exporter-otlp-proto-http>=1.45.0.",
+                        LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES,
+                    )
+
             span_exporter = OTLPSpanExporter(
                 endpoint=endpoint,
                 headers=headers,
                 timeout=timeout,
+                **exporter_kwargs,
             )
 
         if media_manager is not None or mask_otel_spans is not None:

tests/unit/test_span_processor.py

@@ -10,6 +10,7 @@
 from langfuse._client.environment_variables import (
     LANGFUSE_FLUSH_AT,
     LANGFUSE_FLUSH_INTERVAL,
+    LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES,
 )
 from langfuse._client.span_processor import LangfuseSpanProcessor
 
@@ -60,6 +61,100 @@ def test_span_processor_uses_env_flush_settings_when_constructor_omits_them(
         processor.shutdown()
 
 
+class RecordingOTLPSpanExporter(NoOpSpanExporter):
+    init_kwargs: dict = {}
+
+    def __init__(self, *, endpoint=None, headers=None, timeout=None, **kwargs):
+        type(self).init_kwargs = kwargs
+
+
+class RecordingOTLPSpanExporterWithRequestLimit(RecordingOTLPSpanExporter):
+    def __init__(
+        self,
+        *,
+        endpoint=None,
+        headers=None,
+        timeout=None,
+        max_request_size=None,
+    ):
+        super().__init__(max_request_size=max_request_size)
+
+
+def _build_default_exporter_processor(monkeypatch, exporter_class):
+    exporter_class.init_kwargs = {}
+    monkeypatch.setattr(span_processor_module, "OTLPSpanExporter", exporter_class)
+    return LangfuseSpanProcessor(
+        public_key="pk-test",
+        secret_key="sk-test",
+        base_url="http://localhost:3000",
+    )
+
+
+def test_default_exporter_receives_configured_max_batch_size_bytes(monkeypatch):
+    monkeypatch.setenv(LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES, " 1024 ")
+    processor = _build_default_exporter_processor(
+        monkeypatch, RecordingOTLPSpanExporterWithRequestLimit
+    )
+
+    try:
+        assert RecordingOTLPSpanExporterWithRequestLimit.init_kwargs == {
+            "max_request_size": 1024
+        }
+    finally:
+        processor.shutdown()
+
+
+def test_default_exporter_keeps_upstream_limit_when_env_unset(monkeypatch):
+    monkeypatch.delenv(LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES, raising=False)
+    processor = _build_default_exporter_processor(
+        monkeypatch, RecordingOTLPSpanExporterWithRequestLimit
+    )
+
+    try:
+        assert RecordingOTLPSpanExporterWithRequestLimit.init_kwargs == {
+            "max_request_size": None
+        }
+    finally:
+        processor.shutdown()
+
+
+@pytest.mark.parametrize("raw_value", ["0", "-5", "abc", "1.5"])
+def test_invalid_max_batch_size_bytes_falls_back_to_upstream_limit(
+    monkeypatch, caplog, raw_value
+):
+    monkeypatch.setenv(LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES, raw_value)
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        processor = _build_default_exporter_processor(
+            monkeypatch, RecordingOTLPSpanExporterWithRequestLimit
+        )
+
+    try:
+        assert RecordingOTLPSpanExporterWithRequestLimit.init_kwargs == {
+            "max_request_size": None
+        }
+        assert LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES in caplog.text
+    finally:
+        processor.shutdown()
+
+
+def test_max_batch_size_bytes_warns_when_exporter_lacks_request_limit(
+    monkeypatch, caplog
+):
+    monkeypatch.setenv(LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES, "1024")
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        processor = _build_default_exporter_processor(
+            monkeypatch, RecordingOTLPSpanExporter
+        )
+
+    try:
+        assert RecordingOTLPSpanExporter.init_kwargs == {}
+        assert "opentelemetry-exporter-otlp-proto-http>=1.45.0" in caplog.text
+    finally:
+        processor.shutdown()
+
+
 @pytest.fixture
 def tracer_with_processor():
     processor = LangfuseSpanProcessor(