Skip to content

feat(tracing): make span queue size configurable via max_queue_size #1752

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vismaytiwari wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(tracing): make span queue size configurable via max_queue_size #1752

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions langfuse/_client/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
11 changes: 11 additions & 0 deletions langfuse/_client/environment_variables.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions langfuse/_client/get_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
5 changes: 5 additions & 0 deletions langfuse/_client/resource_manager.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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
Expand Down Expand Up @@ -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,
Expand Down
7 changes: 7 additions & 0 deletions langfuse/_client/span_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 (
Expand Down Expand Up @@ -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,
Expand All @@ -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)
Comment on lines +99 to +101

@greptile-apps greptile-apps Bot Jul 11, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 No validation guard for invalid env-var value

int(env_max_queue_size) raises an unhandled ValueError if LANGFUSE_MAX_QUEUE_SIZE is set to a non-integer (e.g. "auto" or "4096.5"). The same pattern exists for flush_at on line 93, so this is a pre-existing gap — but this PR is a good opportunity to add a friendlier parse-or-warn pattern. A failed int() here will surface as a raw Python exception deep in the constructor, with no hint that LANGFUSE_MAX_QUEUE_SIZE is the culprit.

Prompt To Fix With AI 
This is a comment left during a code review.
Path: langfuse/_client/span_processor.py
Line: 99-101

Comment:
**No validation guard for invalid env-var value**

`int(env_max_queue_size)` raises an unhandled `ValueError` if `LANGFUSE_MAX_QUEUE_SIZE` is set to a non-integer (e.g. `"auto"` or `"4096.5"`). The same pattern exists for `flush_at` on line 93, so this is a pre-existing gap — but this PR is a good opportunity to add a friendlier parse-or-warn pattern. A failed `int()` here will surface as a raw Python exception deep in the constructor, with no hint that `LANGFUSE_MAX_QUEUE_SIZE` is the culprit.

How can I resolve this? If you propose a fix, please make it concise.

@vismaytiwari vismaytiwari Jul 11, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good catch. I kept this consistent with the flush_at / flush_interval parsing right above it — same behavior — so the new option isn't a one-off special case. Hardening the env-var parsing (parse-or-warn with a clear message) is worth doing, but I'd rather do all three together in a small follow-up than change only this one here. Happy to open that if you'd like it.


if span_exporter is None:
basic_auth_header = "Basic " + base64.b64encode(
f"{public_key}:{secret_key}".encode("utf-8")
Expand Down Expand Up @@ -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
Expand Down
51 changes: 51 additions & 0 deletions tests/unit/test_span_processor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -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()