From f0dd7fce433f5748774dc346326ccb6c7a64f58c Mon Sep 17 00:00:00 2001 From: passionworkeer Date: Sat, 31 Oct 2026 22:24:50 +0800 Subject: [PATCH 1/2] fix(otel): widen the span queue when flush_at exceeds its default --- langfuse/_client/span_processor.py | 9 ++ tests/unit/test_flush_at_batch_size.py | 141 +++++++++++++++++++++++++ 2 files changed, 150 insertions(+) create mode 100644 tests/unit/test_flush_at_batch_size.py diff --git a/langfuse/_client/span_processor.py b/langfuse/_client/span_processor.py index b6ebbbce6..c86c345e1 100644 --- a/langfuse/_client/span_processor.py +++ b/langfuse/_client/span_processor.py @@ -225,10 +225,19 @@ def __init__( mask_otel_spans=mask_otel_spans, ) + # OpenTelemetry rejects a batch size larger than the queue size, and + # flush_at maps onto the batch size. The queue keeps OpenTelemetry's own + # default -- which is configurable through OTEL_BSP_MAX_QUEUE_SIZE -- and + # is only widened when a larger batch was asked for. + default_queue_size = self._default_max_queue_size() + super().__init__( span_exporter=span_exporter, export_timeout_millis=timeout * 1_000 if timeout else None, max_export_batch_size=flush_at, + max_queue_size=max(flush_at, default_queue_size) + if flush_at is not None + else None, schedule_delay_millis=flush_interval * 1_000 if flush_interval is not None else None, diff --git a/tests/unit/test_flush_at_batch_size.py b/tests/unit/test_flush_at_batch_size.py new file mode 100644 index 000000000..9bf534063 --- /dev/null +++ b/tests/unit/test_flush_at_batch_size.py @@ -0,0 +1,141 @@ +"""A flush_at larger than the OpenTelemetry default queue size must be usable.""" + +from typing import Optional, Sequence + +import pytest +from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult +from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + +from langfuse import Langfuse +from langfuse._client.resource_manager import LangfuseResourceManager + + +class NoOpSpanExporter(SpanExporter): + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + return SpanExportResult.SUCCESS + + def shutdown(self) -> None: + pass + + +@pytest.fixture +def isolated_client(): + with LangfuseResourceManager._lock: + LangfuseResourceManager._instances.clear() + + client = Langfuse( + public_key="pk-flush-at", + secret_key="sk-flush-at", + span_exporter=NoOpSpanExporter(), + flush_at=5000, + ) + + yield client + + client.shutdown() + + +@pytest.mark.unit +def test_flush_at_above_the_default_queue_size_is_accepted(isolated_client): + """Configuring a large flush_at must not fail client construction.""" + assert isolated_client is not None + + +@pytest.mark.unit +def test_client_with_large_flush_at_can_flush(isolated_client): + isolated_client.flush() + + +def _queue_capacity(client: Langfuse) -> Optional[int]: + """Best-effort read of the configured queue capacity. + + Returns None when the OpenTelemetry internals this relies on are not present, + so the caller can skip instead of failing on an upstream refactor. + """ + provider = client._resources.tracer_provider + + 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 + + return None + + +@pytest.mark.unit +def test_flush_at_uses_the_effective_default_queue_size(monkeypatch): + """A smaller OTEL_BSP_MAX_QUEUE_SIZE must still be widened when flush_at needs it.""" + monkeypatch.setenv("OTEL_BSP_MAX_QUEUE_SIZE", "1000") + + with LangfuseResourceManager._lock: + LangfuseResourceManager._instances.clear() + + client = Langfuse( + public_key="pk-flush-at", + secret_key="sk-flush-at", + span_exporter=NoOpSpanExporter(), + flush_at=1500, + ) + + try: + client.flush() + finally: + client.shutdown() + + +@pytest.mark.unit +def test_a_larger_default_queue_size_is_not_shrunk(monkeypatch): + """Widening the batch must not quietly narrow an already larger queue.""" + monkeypatch.setenv("OTEL_BSP_MAX_QUEUE_SIZE", "5000") + + with LangfuseResourceManager._lock: + LangfuseResourceManager._instances.clear() + + client = Langfuse( + public_key="pk-flush-at", + secret_key="sk-flush-at", + span_exporter=NoOpSpanExporter(), + flush_at=3000, + ) + + try: + capacity = _queue_capacity(client) + + if capacity is None: + pytest.skip("OpenTelemetry no longer exposes the queue capacity") + + assert capacity == 5000 + finally: + client.shutdown() + + +@pytest.mark.unit +def test_spans_are_still_exported_with_a_large_flush_at(): + """A large flush_at must not silently drop or hold back finished spans.""" + with LangfuseResourceManager._lock: + LangfuseResourceManager._instances.clear() + + exporter = InMemorySpanExporter() + + client = Langfuse( + public_key="pk-flush-at", + secret_key="sk-flush-at", + span_exporter=exporter, + flush_at=5000, + ) + + try: + client.start_observation(name="flush-at-span").end() + + client.flush() + + span_names = [span.name for span in exporter.get_finished_spans()] + assert "flush-at-span" in span_names + finally: + client.shutdown() From 68499cb6e8d23af75e283975ef86aa72dbd124da Mon Sep 17 00:00:00 2001 From: passionworkeer Date: Sun, 1 Nov 2026 12:28:09 +0800 Subject: [PATCH 2/2] test(otel): give the flush_at tests a dedicated TracerProvider --- tests/unit/test_flush_at_batch_size.py | 104 ++++++++++++------------- 1 file changed, 49 insertions(+), 55 deletions(-) diff --git a/tests/unit/test_flush_at_batch_size.py b/tests/unit/test_flush_at_batch_size.py index 9bf534063..53f095d9b 100644 --- a/tests/unit/test_flush_at_batch_size.py +++ b/tests/unit/test_flush_at_batch_size.py @@ -3,12 +3,13 @@ from typing import Optional, Sequence import pytest -from opentelemetry.sdk.trace import ReadableSpan +from opentelemetry.sdk.trace import ReadableSpan, TracerProvider from opentelemetry.sdk.trace.export import SpanExporter, SpanExportResult from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter from langfuse import Langfuse from langfuse._client.resource_manager import LangfuseResourceManager +from langfuse._client.span_processor import LangfuseSpanProcessor class NoOpSpanExporter(SpanExporter): @@ -19,17 +20,50 @@ def shutdown(self) -> None: pass -@pytest.fixture -def isolated_client(): +def _build_client(**kwargs) -> Langfuse: + """Build a client with its own TracerProvider. + + Without this the processor is added to the globally registered provider, where + processors from earlier clients are still attached. + """ with LangfuseResourceManager._lock: LangfuseResourceManager._instances.clear() - client = Langfuse( + return Langfuse( public_key="pk-flush-at", secret_key="sk-flush-at", - span_exporter=NoOpSpanExporter(), - flush_at=5000, + tracer_provider=TracerProvider(), + **kwargs, + ) + + +def _queue_capacity(client: Langfuse) -> Optional[int]: + """Read the queue capacity of this client's span processor. + + Returns None when the OpenTelemetry internals this relies on are not present, + so the caller can skip instead of failing on an upstream refactor. + """ + provider = client._resources.tracer_provider + processors = [ + processor + for processor in provider._active_span_processor._span_processors + if isinstance(processor, LangfuseSpanProcessor) + ] + + assert len(processors) == 1, "the client must own its TracerProvider" + + queue = getattr(processors[0], "_queue", None) or getattr( + processors[0], "_batch_processor", None ) + target = getattr(queue, "_queue", queue) + capacity = getattr(target, "maxlen", None) or getattr(target, "capacity", None) + + return capacity if isinstance(capacity, int) else None + + +@pytest.fixture +def isolated_client(): + client = _build_client(span_exporter=NoOpSpanExporter(), flush_at=5000) yield client @@ -47,44 +81,20 @@ def test_client_with_large_flush_at_can_flush(isolated_client): isolated_client.flush() -def _queue_capacity(client: Langfuse) -> Optional[int]: - """Best-effort read of the configured queue capacity. - - Returns None when the OpenTelemetry internals this relies on are not present, - so the caller can skip instead of failing on an upstream refactor. - """ - provider = client._resources.tracer_provider - - 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 - - return None - - @pytest.mark.unit def test_flush_at_uses_the_effective_default_queue_size(monkeypatch): """A smaller OTEL_BSP_MAX_QUEUE_SIZE must still be widened when flush_at needs it.""" monkeypatch.setenv("OTEL_BSP_MAX_QUEUE_SIZE", "1000") - with LangfuseResourceManager._lock: - LangfuseResourceManager._instances.clear() - - client = Langfuse( - public_key="pk-flush-at", - secret_key="sk-flush-at", - span_exporter=NoOpSpanExporter(), - flush_at=1500, - ) + client = _build_client(span_exporter=NoOpSpanExporter(), flush_at=1500) try: - client.flush() + capacity = _queue_capacity(client) + + if capacity is None: + pytest.skip("OpenTelemetry no longer exposes the queue capacity") + + assert capacity == 1500 finally: client.shutdown() @@ -94,15 +104,7 @@ def test_a_larger_default_queue_size_is_not_shrunk(monkeypatch): """Widening the batch must not quietly narrow an already larger queue.""" monkeypatch.setenv("OTEL_BSP_MAX_QUEUE_SIZE", "5000") - with LangfuseResourceManager._lock: - LangfuseResourceManager._instances.clear() - - client = Langfuse( - public_key="pk-flush-at", - secret_key="sk-flush-at", - span_exporter=NoOpSpanExporter(), - flush_at=3000, - ) + client = _build_client(span_exporter=NoOpSpanExporter(), flush_at=3000) try: capacity = _queue_capacity(client) @@ -118,17 +120,9 @@ def test_a_larger_default_queue_size_is_not_shrunk(monkeypatch): @pytest.mark.unit def test_spans_are_still_exported_with_a_large_flush_at(): """A large flush_at must not silently drop or hold back finished spans.""" - with LangfuseResourceManager._lock: - LangfuseResourceManager._instances.clear() - exporter = InMemorySpanExporter() - client = Langfuse( - public_key="pk-flush-at", - secret_key="sk-flush-at", - span_exporter=exporter, - flush_at=5000, - ) + client = _build_client(span_exporter=exporter, flush_at=5000) try: client.start_observation(name="flush-at-span").end()