Repository navigation
fix(otel): widen the span queue when flush_at exceeds its default - #2086
passionworkeer wants to merge 2 commits into
Conversation
![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}
| for processor in provider._active_span_processor._span_processors: | ||
| queue = getattr(processor, "_queue", None) or getattr( | ||
| processor, "_batch_processor", None | ||
| ) | ||
| target = getattr(queue, "_queue", queue) | ||
| capacity = getattr(target, "maxlen", None) or getattr(target, "capacity", None) | ||
|
|
||
| if isinstance(capacity, int): | ||
| return capacity |
There was a problem hiding this comment.
_queue_capacity() returns the first queued processor on the shared provider, not the processor created by this test. LangfuseResourceManager.reset() clears clients but leaves the global provider and its processors in place. If the smaller-queue test runs first, this test reads its capacity of 1500 and fails despite correctly creating a queue of 5000. Give these tests a fresh TracerProvider, or select the processor belonging to the current client.
Knowledge Base Used: Client initialization and resource management
Prompt To Fix With AI
This is a comment left during a code review.
Path: tests/unit/test_flush_at_batch_size.py
Line: 58-66
Comment:
**Test reads an older queue**
`_queue_capacity()` returns the first queued processor on the shared provider, not the processor created by this test. `LangfuseResourceManager.reset()` clears clients but leaves the global provider and its processors in place. If the smaller-queue test runs first, this test reads its capacity of 1500 and fails despite correctly creating a queue of 5000. Give these tests a fresh `TracerProvider`, or select the processor belonging to the current client.
**Knowledge Base Used:** [Client initialization and resource management](https://app.greptile.com/-/switch-tenant?tenant=personal-org-4986&next=%2Fcustom-context%2Fknowledge-base%2Flangfuse%2Flangfuse-python%2F-%2Fdocs%2Fclient-initialization-and-resources.md)
---
For each issue above, determine whether it is valid and should be fixed. If so, fix it directly.There was a problem hiding this comment.
Fixed in 68499cb — each test now builds its client with its own TracerProvider, so the capacity it reads belongs to that client instead of whichever processor was registered first on the shared provider.
What does this PR do?
LangfuseSpanProcessorforwardsflush_attoBatchSpanProcessorasmax_export_batch_sizeand never setsmax_queue_size. OpenTelemetry keeps its owndefault queue size and rejects any larger batch, so a
flush_atabove the queue sizefailed client construction with
The message never mentions
flush_at, which makes the failure hard to attribute to aLangfuse setting.
This passes
max_queue_size=max(flush_at, default_queue_size), wheredefault_queue_sizeis OpenTelemetry's own resolved default (
BatchSpanProcessor._default_max_queue_size(),which honours
OTEL_BSP_MAX_QUEUE_SIZE). A batch that already fits leaves the queueexactly as it is today; only a batch that does not fit widens it.
Fixes langfuse/langfuse#19539
Type of change
Verification
Regression tests in
tests/unit/test_flush_at_batch_size.py. On the unpatchedmain4 of the 5 fail or error with the
ValueErrorabove; all 5 pass with the fix:flush_at=5000constructs and flushesflush_at=5000OTEL_BSP_MAX_QUEUE_SIZE=1000+flush_at=1500OTEL_BSP_MAX_QUEUE_SIZE=5000+flush_at=3000keeps queue at 5000Queue capacity sweep (probe script, not committed — it reads OpenTelemetry internals):
OTEL_BSP_MAX_QUEUE_SIZEflush_atValueErrorValueErrorValueErrorValueErrorOther checks:
tests/unit/test_span_processor.py+test_resource_manager.py+test_otel.py:145 passed, 2 skipped.
tests/unit: 988 passed, 2 skipped. The remaining 2 failures and 18 errors are allin
tests/unit/test_prompt.py/test_prompt_atexit.pyand reproduce identically on theunpatched
mainin this environment (noLANGFUSE_PUBLIC_KEY, no literalpythononPATH); they are unrelated to this change.
ruff check .clean,ruff format --checkclean on both changed files,mypy langfuseclean.Note for reviewers
_default_max_queue_size()is an OpenTelemetry internal. It is used here because it iswhat determines the constraint being fixed, and because the queue default is configurable
through
OTEL_BSP_MAX_QUEUE_SIZE— hard-coding 2048 would both miss smaller configuredqueues and narrow larger ones. The declared range is
opentelemetry-sdk>=1.45.0,<2; thealternative would be duplicating OpenTelemetry's environment parsing.
An alternative fix would be to validate
flush_atand raise a clear error naming thesetting instead of widening the queue. Happy to switch if you prefer that — this PR picks
widening because it keeps every currently working configuration working.
Checklist
langfuse/api/The PR appears safe to merge, with a non-blocking test isolation issue.
Summary
Widens the span queue when
flush_atexceeds OpenTelemetry's configured queue size.flush_atsettings fit in the span queue.Reviews (1) · Last reviewed commit: "fix(otel): widen the span queue when flu..." · Reviewed by Greptile