feat(tracing): make span queue size configurable via max_queue_size#1752
feat(tracing): make span queue size configurable via max_queue_size#1752vismaytiwari wants to merge 1 commit into
Conversation
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
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| 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) |
There was a problem hiding this comment.
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.There was a problem hiding this comment.
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.
1 participant
What does this PR do?
Refs langfuse/langfuse#12974
Under high concurrency the OpenTelemetry
BatchSpanProcessordrops spans once its in-memory queue fills up. When the dropped span happens to be a parent, the trace ends up broken/incomplete — which is what #12974 describes ("the SDK's flush queue gets overwhelmed … some spans don't get flushed before the trace completes").flush_atandflush_intervalare already exposed as first-class Langfuse settings, butmax_queue_size— the knob that actually governs how many spans can be buffered before drops start — wasn't. That left the queue pinned to OTEL's default of 2048 with no Langfuse-native way to raise it (you'd have to know about the underlyingOTEL_BSP_MAX_QUEUE_SIZEenv var).This adds a
max_queue_sizeclient argument and a matchingLANGFUSE_MAX_QUEUE_SIZEenv var, threaded through the resource manager and span processor exactly like the existing flush settings. It defaults toNone, so behavior is unchanged for existing users (OTEL still falls back toOTEL_BSP_MAX_QUEUE_SIZE/ 2048). Users hitting drops under load can now raise the buffer without reaching into OTEL internals.Note: this doesn't eliminate the bounded-queue drop behavior itself — it gives users a documented, first-class way to size the buffer for their concurrency, consistent with how
flush_at/flush_intervalalready work.Type of change
Verification
List the main commands you ran:
Checklist
code_review.md..env.templateif needed. (Added the client docstring +LANGFUSE_MAX_QUEUE_SIZEenv-var reference; no.env.templatechange needed.)Greptile Summary
This PR exposes
max_queue_sizeas a first-class Langfuse setting (constructor argument + env varLANGFUSE_MAX_QUEUE_SIZE), giving users a documented way to raise the OTELBatchSpanProcessorin-memory buffer beyond its 2 048-span default without reaching into OTEL internals. The change defaults toNoneso existing behavior is fully preserved.max_queue_sizeconsistently throughclient.py→resource_manager.py→span_processor.py→BatchSpanProcessor.__init__, mirroring the existingflush_at/flush_intervalpattern.Confidence Score: 4/5
Safe to merge; all existing behavior is unchanged when max_queue_size is not set, and the new parameter is correctly propagated through every layer.
The change is mechanical and well-tested, following the existing flush_at/flush_interval pattern precisely. Two small gaps in span_processor.py mean a misconfigured value surfaces as a raw OTEL exception rather than a clear Langfuse message, but neither gap causes silent data loss.
langfuse/_client/span_processor.py — env-var parsing and the constraint check before the BatchSpanProcessor call.
Sequence Diagram
%%{init: {'theme': 'neutral'}}%% sequenceDiagram participant User participant Langfuse as Langfuse(client.py) participant RM as LangfuseResourceManager participant SP as LangfuseSpanProcessor participant BSP as BatchSpanProcessor(OTEL) participant Env as Environment Variables User->>Langfuse: "Langfuse(max_queue_size=N)" Langfuse->>RM: "LangfuseResourceManager(max_queue_size=N)" RM->>SP: "LangfuseSpanProcessor(max_queue_size=N)" SP->>Env: os.environ.get(LANGFUSE_MAX_QUEUE_SIZE) Env-->>SP: env value (if set, overrides None only) SP->>BSP: "super().__init__(max_queue_size=N or None)" BSP->>Env: os.environ.get(OTEL_BSP_MAX_QUEUE_SIZE) Env-->>BSP: fallback to 2048 if all None BSP-->>SP: initialized with effective queue size SP-->>RM: span processor ready RM-->>Langfuse: resource manager ready Langfuse-->>User: client initialized%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%% sequenceDiagram participant User participant Langfuse as Langfuse(client.py) participant RM as LangfuseResourceManager participant SP as LangfuseSpanProcessor participant BSP as BatchSpanProcessor(OTEL) participant Env as Environment Variables User->>Langfuse: "Langfuse(max_queue_size=N)" Langfuse->>RM: "LangfuseResourceManager(max_queue_size=N)" RM->>SP: "LangfuseSpanProcessor(max_queue_size=N)" SP->>Env: os.environ.get(LANGFUSE_MAX_QUEUE_SIZE) Env-->>SP: env value (if set, overrides None only) SP->>BSP: "super().__init__(max_queue_size=N or None)" BSP->>Env: os.environ.get(OTEL_BSP_MAX_QUEUE_SIZE) Env-->>BSP: fallback to 2048 if all None BSP-->>SP: initialized with effective queue size SP-->>RM: span processor ready RM-->>Langfuse: resource manager ready Langfuse-->>User: client initializedPrompt To Fix All With AI
Reviews (1): Last reviewed commit: "feat(tracing): make span queue size conf..." | Re-trigger Greptile