From f55b6e9983a5e47cb21588f0e00ac45c0cf3b1ab Mon Sep 17 00:00:00 2001 From: vismaytiwari Date: Sat, 11 Jul 2026 18:20:06 +0530 Subject: [PATCH] feat(tracing): make span queue size configurable via max_queue_size MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The OpenTelemetry BatchSpanProcessor drops spans once its in-memory queue is full, which under high concurrency can drop a parent span and leave incomplete traces. flush_at and flush_interval were already exposed as first-class settings, but the queue size — the knob that actually governs drops under load — was not, leaving OTEL's default of 2048 with no Langfuse-native way to raise it. Add a max_queue_size client argument and LANGFUSE_MAX_QUEUE_SIZE env var, threaded through the resource manager and span processor exactly like the existing flush settings. Defaults to None, preserving current behavior (OTEL falls back to OTEL_BSP_MAX_QUEUE_SIZE / 2048). Refs langfuse/langfuse#12974 --- langfuse/_client/client.py | 3 ++ langfuse/_client/environment_variables.py | 11 +++++ langfuse/_client/get_client.py | 1 + langfuse/_client/resource_manager.py | 5 +++ langfuse/_client/span_processor.py | 7 ++++ tests/unit/test_span_processor.py | 51 +++++++++++++++++++++++ 6 files changed, 78 insertions(+) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index 6ce72d27a..9d131113f 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -189,6 +189,7 @@ class Langfuse: tracing_enabled (Optional[bool]): Enable or disable tracing. Defaults to True. Can also be set via LANGFUSE_TRACING_ENABLED environment variable. flush_at (Optional[int]): Number of spans to batch before sending to the API. Defaults to 512. Can also be set via LANGFUSE_FLUSH_AT environment variable. flush_interval (Optional[float]): Time in seconds between batch flushes. Defaults to 5 seconds. Can also be set via LANGFUSE_FLUSH_INTERVAL environment variable. + max_queue_size (Optional[int]): Maximum number of spans buffered in memory before export. When the buffer fills (for example under high concurrency), additional spans are dropped, which can result in incomplete traces; increase this to reduce drops. Must be greater than or equal to flush_at. Defaults to 2048. Can also be set via LANGFUSE_MAX_QUEUE_SIZE environment variable. environment (Optional[str]): Environment name for tracing. Default is 'default'. Can also be set via LANGFUSE_TRACING_ENVIRONMENT environment variable. Can be any lowercase alphanumeric string with hyphens and underscores that does not start with 'langfuse'. release (Optional[str]): Release version/hash of your application. Used for grouping analytics by release. media_upload_thread_count (Optional[int]): Number of background threads for handling media uploads. Defaults to 1. Can also be set via LANGFUSE_MEDIA_UPLOAD_THREAD_COUNT environment variable. @@ -299,6 +300,7 @@ def __init__( tracing_enabled: Optional[bool] = True, flush_at: Optional[int] = None, flush_interval: Optional[float] = None, + max_queue_size: Optional[int] = None, environment: Optional[str] = None, release: Optional[str] = None, media_upload_thread_count: Optional[int] = None, @@ -398,6 +400,7 @@ def __init__( release=release, flush_at=flush_at, flush_interval=flush_interval, + max_queue_size=max_queue_size, httpx_client=httpx_client, media_upload_thread_count=media_upload_thread_count, sample_rate=sample_rate, diff --git a/langfuse/_client/environment_variables.py b/langfuse/_client/environment_variables.py index 0a95d53ba..3ebe55357 100644 --- a/langfuse/_client/environment_variables.py +++ b/langfuse/_client/environment_variables.py @@ -105,6 +105,17 @@ **Default value:** same as OTEL ``OTEL_BSP_SCHEDULE_DELAY`` """ +LANGFUSE_MAX_QUEUE_SIZE = "LANGFUSE_MAX_QUEUE_SIZE" +""" +.. envvar:: LANGFUSE_MAX_QUEUE_SIZE + +Max number of spans buffered in memory before the exporter flushes them. When the +buffer is full (for example under high concurrency), additional spans are dropped, +which can result in incomplete traces. Increase this to reduce drops. Must be +greater than or equal to ``LANGFUSE_FLUSH_AT``. +**Default value:** same as OTEL ``OTEL_BSP_MAX_QUEUE_SIZE`` (``2048``) +""" + LANGFUSE_SAMPLE_RATE = "LANGFUSE_SAMPLE_RATE" """ .. envvar: LANGFUSE_SAMPLE_RATE diff --git a/langfuse/_client/get_client.py b/langfuse/_client/get_client.py index a360430ac..6cd173220 100644 --- a/langfuse/_client/get_client.py +++ b/langfuse/_client/get_client.py @@ -46,6 +46,7 @@ def _create_client_from_instance( timeout=instance.timeout, flush_at=instance.flush_at, flush_interval=instance.flush_interval, + max_queue_size=instance.max_queue_size, release=instance.release, media_upload_thread_count=instance.media_upload_thread_count, sample_rate=instance.sample_rate, diff --git a/langfuse/_client/resource_manager.py b/langfuse/_client/resource_manager.py index 67c44920a..d7bced817 100644 --- a/langfuse/_client/resource_manager.py +++ b/langfuse/_client/resource_manager.py @@ -121,6 +121,7 @@ def __new__( timeout: Optional[int] = None, flush_at: Optional[int] = None, flush_interval: Optional[float] = None, + max_queue_size: Optional[int] = None, httpx_client: Optional[httpx.Client] = None, media_upload_thread_count: Optional[int] = None, sample_rate: Optional[float] = None, @@ -157,6 +158,7 @@ def __new__( release=release, flush_at=flush_at, flush_interval=flush_interval, + max_queue_size=max_queue_size, httpx_client=httpx_client, media_upload_thread_count=media_upload_thread_count, sample_rate=sample_rate, @@ -188,6 +190,7 @@ def _initialize_instance( timeout: Optional[int] = None, flush_at: Optional[int] = None, flush_interval: Optional[float] = None, + max_queue_size: Optional[int] = None, media_upload_thread_count: Optional[int] = None, httpx_client: Optional[httpx.Client] = None, sample_rate: Optional[float] = None, @@ -214,6 +217,7 @@ def _initialize_instance( self.timeout = timeout self.flush_at = flush_at self.flush_interval = flush_interval + self.max_queue_size = max_queue_size self.release = release self.media_upload_thread_count = media_upload_thread_count self.sample_rate = sample_rate @@ -255,6 +259,7 @@ def _initialize_instance( timeout=timeout, flush_at=flush_at, flush_interval=flush_interval, + max_queue_size=max_queue_size, blocked_instrumentation_scopes=blocked_instrumentation_scopes, should_export_span=should_export_span, additional_headers=additional_headers, diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index aee15d17b..16f39cdac 100644 --- a/langfuse/_client/span_processor.py +++ b/langfuse/_client/span_processor.py @@ -27,6 +27,7 @@ from langfuse._client.environment_variables import ( LANGFUSE_FLUSH_AT, LANGFUSE_FLUSH_INTERVAL, + LANGFUSE_MAX_QUEUE_SIZE, LANGFUSE_OTEL_TRACES_EXPORT_PATH, ) from langfuse._client.propagation import ( @@ -68,6 +69,7 @@ def __init__( timeout: Optional[int] = None, flush_at: Optional[int] = None, flush_interval: Optional[float] = None, + max_queue_size: Optional[int] = None, blocked_instrumentation_scopes: Optional[List[str]] = None, should_export_span: Optional[Callable[[ReadableSpan], bool]] = None, additional_headers: Optional[Dict[str, str]] = None, @@ -94,6 +96,10 @@ def __init__( if flush_interval is None and env_flush_interval is not None: flush_interval = float(env_flush_interval) + env_max_queue_size = os.environ.get(LANGFUSE_MAX_QUEUE_SIZE, None) + if max_queue_size is None and env_max_queue_size is not None: + max_queue_size = int(env_max_queue_size) + if span_exporter is None: basic_auth_header = "Basic " + base64.b64encode( f"{public_key}:{secret_key}".encode("utf-8") @@ -134,6 +140,7 @@ def __init__( super().__init__( span_exporter=span_exporter, export_timeout_millis=timeout * 1_000 if timeout else None, + max_queue_size=max_queue_size, max_export_batch_size=flush_at, schedule_delay_millis=flush_interval * 1_000 if flush_interval is not None diff --git a/tests/unit/test_span_processor.py b/tests/unit/test_span_processor.py index 5d12813a0..d0826ef3a 100644 --- a/tests/unit/test_span_processor.py +++ b/tests/unit/test_span_processor.py @@ -6,6 +6,7 @@ from langfuse._client.environment_variables import ( LANGFUSE_FLUSH_AT, LANGFUSE_FLUSH_INTERVAL, + LANGFUSE_MAX_QUEUE_SIZE, ) from langfuse._client.span_processor import LangfuseSpanProcessor @@ -54,3 +55,53 @@ def test_span_processor_uses_env_flush_settings_when_constructor_omits_them( assert processor._batch_processor._schedule_delay_millis == 3250 finally: processor.shutdown() + + +def test_span_processor_uses_constructor_max_queue_size_without_env(monkeypatch): + monkeypatch.delenv(LANGFUSE_MAX_QUEUE_SIZE, raising=False) + processor = LangfuseSpanProcessor( + public_key="pk-test", + secret_key="sk-test", + base_url="http://localhost:3000", + max_queue_size=5000, + span_exporter=NoOpSpanExporter(), + ) + + try: + assert processor._batch_processor._max_queue_size == 5000 + finally: + processor.shutdown() + + +def test_span_processor_uses_env_max_queue_size_when_constructor_omits_it(monkeypatch): + monkeypatch.setenv(LANGFUSE_MAX_QUEUE_SIZE, "4096") + processor = LangfuseSpanProcessor( + public_key="pk-test", + secret_key="sk-test", + base_url="http://localhost:3000", + span_exporter=NoOpSpanExporter(), + ) + + try: + assert processor._batch_processor._max_queue_size == 4096 + finally: + processor.shutdown() + + +def test_span_processor_defaults_to_otel_max_queue_size(monkeypatch): + # When neither the constructor arg nor the Langfuse env var is set, the queue + # size falls back to OpenTelemetry's default (OTEL_BSP_MAX_QUEUE_SIZE, 2048), + # keeping behavior unchanged for existing users. + monkeypatch.delenv(LANGFUSE_MAX_QUEUE_SIZE, raising=False) + monkeypatch.delenv("OTEL_BSP_MAX_QUEUE_SIZE", raising=False) + processor = LangfuseSpanProcessor( + public_key="pk-test", + secret_key="sk-test", + base_url="http://localhost:3000", + span_exporter=NoOpSpanExporter(), + ) + + try: + assert processor._batch_processor._max_queue_size == 2048 + finally: + processor.shutdown()