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..53f095d9b --- /dev/null +++ b/tests/unit/test_flush_at_batch_size.py @@ -0,0 +1,135 @@ +"""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, 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): + def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult: + return SpanExportResult.SUCCESS + + def shutdown(self) -> None: + pass + + +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() + + return Langfuse( + public_key="pk-flush-at", + secret_key="sk-flush-at", + 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 + + 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() + + +@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") + + client = _build_client(span_exporter=NoOpSpanExporter(), flush_at=1500) + + try: + capacity = _queue_capacity(client) + + if capacity is None: + pytest.skip("OpenTelemetry no longer exposes the queue capacity") + + assert capacity == 1500 + 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") + + client = _build_client(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.""" + exporter = InMemorySpanExporter() + + client = _build_client(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()