From e49d8abb4c6321429ad1a4f94cc09573998abccf Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 6 Aug 2026 10:58:48 +0200 Subject: [PATCH 1/5] fix(otel): keep last duplicate span in masking batches --- langfuse/_client/client.py | 2 +- langfuse/_client/span_exporter.py | 38 ++++++++------ langfuse/types.py | 4 +- tests/unit/test_mask_otel_spans.py | 79 ++++++++++++++++++++++++++++++ 4 files changed, 107 insertions(+), 16 deletions(-) diff --git a/langfuse/_client/client.py b/langfuse/_client/client.py index d97708efd..2b2889545 100644 --- a/langfuse/_client/client.py +++ b/langfuse/_client/client.py @@ -222,7 +222,7 @@ class Langfuse: The hook receives one OpenTelemetry export batch. A batch is not guaranteed to contain a complete trace, request, or Langfuse observation tree. The hook usually runs on the OpenTelemetry batch span processor worker thread; during `flush()` and shutdown it may run on the caller thread. Keep it synchronous, deterministic, and fast. - Return `None` to leave the batch unchanged. Return `MaskOtelSpansResult` with `OtelSpanPatch` values to delete or replace attributes on selected spans. If the hook raises or returns an invalid batch result, Langfuse drops the whole export batch. If one returned span patch is invalid, Langfuse drops only that span from the Langfuse export. + Return `None` to leave the batch unchanged. Return `MaskOtelSpansResult` with `OtelSpanPatch` values to delete or replace attributes on selected spans. If a batch contains duplicate trace and span identifiers, Langfuse keeps only the last matching span. If the hook raises or returns an invalid batch result, Langfuse drops the whole export batch. If one returned span patch is invalid, Langfuse drops only that span from the Langfuse export. Example: ```python diff --git a/langfuse/_client/span_exporter.py b/langfuse/_client/span_exporter.py index 582c4574c..256787658 100644 --- a/langfuse/_client/span_exporter.py +++ b/langfuse/_client/span_exporter.py @@ -264,10 +264,10 @@ def _apply_mask_otel_spans( span_attributes: Sequence[tuple[ReadableSpan, Dict[str, AttributeValue]]], ) -> Optional[list[tuple[ReadableSpan, Dict[str, AttributeValue]]]]: mask_otel_spans = cast(MaskOtelSpansFunction, self._mask_otel_spans) - maskable_span_attributes: list[ - tuple[ReadableSpan, Dict[str, AttributeValue]] - ] = [] - span_data_by_identifier: Dict[OtelSpanIdentifier, OtelSpanData] = {} + span_attributes_by_identifier: Dict[ + OtelSpanIdentifier, tuple[ReadableSpan, Dict[str, AttributeValue]] + ] = {} + duplicate_span_count = 0 for span, attributes in span_attributes: if not _has_valid_span_context(span): @@ -279,22 +279,32 @@ def _apply_mask_otel_spans( identifier = _create_otel_span_identifier(span) - if identifier in span_data_by_identifier: - langfuse_logger.error( - "Masking error: mask_otel_spans received duplicate span identifiers. " - "Dropping export batch. " - f"trace_id='{identifier.trace_id}' span_id='{identifier.span_id}'" - ) - return None + if identifier in span_attributes_by_identifier: + duplicate_span_count += 1 + span_attributes_by_identifier.pop(identifier) - span_data_by_identifier[identifier] = _create_otel_span_data( - span=span, attributes=attributes, identifier=identifier + span_attributes_by_identifier[identifier] = (span, attributes) + + if duplicate_span_count: + langfuse_logger.warning( + "Masking warning: mask_otel_spans received duplicate span identifiers. " + "Keeping the last span for each identifier. " + f"duplicate_span_count={duplicate_span_count} " + f"remaining_span_count={len(span_attributes_by_identifier)}" ) - maskable_span_attributes.append((span, attributes)) + + maskable_span_attributes = list(span_attributes_by_identifier.values()) if not maskable_span_attributes: return [] + span_data_by_identifier = { + identifier: _create_otel_span_data( + span=span, attributes=attributes, identifier=identifier + ) + for identifier, (span, attributes) in span_attributes_by_identifier.items() + } + try: result: Any = mask_otel_spans( params=MaskOtelSpansParams( diff --git a/langfuse/types.py b/langfuse/types.py index 023709765..357438d56 100644 --- a/langfuse/types.py +++ b/langfuse/types.py @@ -126,7 +126,9 @@ class MaskOtelSpansParams: A single call receives one OpenTelemetry export batch, not necessarily a complete trace, request, or Langfuse observation tree. Batch contents depend on OpenTelemetry span processor settings such as `flush_at`, - `flush_interval`, explicit `flush()`, and shutdown. + `flush_interval`, explicit `flush()`, and shutdown. If a batch contains + duplicate trace and span identifiers, Langfuse keeps only the last matching + span before calling the masking function. Example: ```python diff --git a/tests/unit/test_mask_otel_spans.py b/tests/unit/test_mask_otel_spans.py index 4834a84d0..5f11e7c4c 100644 --- a/tests/unit/test_mask_otel_spans.py +++ b/tests/unit/test_mask_otel_spans.py @@ -574,6 +574,85 @@ def mask_otel_spans(*, params: MaskOtelSpansParams): ) +def test_mask_otel_spans_keeps_last_duplicate_without_dropping_batch(caplog): + exporter = InMemorySpanExporter() + seen_params: list[MaskOtelSpansParams] = [] + + def mask_otel_spans(*, params: MaskOtelSpansParams): + seen_params.append(params) + duplicate_identifier = next( + identifier + for identifier, span in params.spans.items() + if span.name == "duplicate-last" + ) + + return MaskOtelSpansResult( + span_patches={ + duplicate_identifier: OtelSpanPatch( + set_attributes={"masking.applied": True} + ) + } + ) + + transforming_exporter = span_exporter_module.LangfuseTransformingSpanExporter( + exporter=exporter, + media_manager=None, + mask_otel_spans=mask_otel_spans, + ) + duplicate_context = SpanContext( + trace_id=1, + span_id=2, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + trace_state=TraceState(), + ) + unrelated_context = SpanContext( + trace_id=3, + span_id=4, + is_remote=False, + trace_flags=TraceFlags(TraceFlags.SAMPLED), + trace_state=TraceState(), + ) + spans = [ + ReadableSpan( + name="duplicate-first", + context=duplicate_context, + attributes={"attempt": 1}, + ), + ReadableSpan( + name="unrelated", + context=unrelated_context, + attributes={"unrelated": True}, + ), + ReadableSpan( + name="duplicate-last", + context=duplicate_context, + attributes={"attempt": 2}, + ), + ] + + with caplog.at_level(logging.WARNING, logger="langfuse"): + result = transforming_exporter.export(spans) + + assert result == SpanExportResult.SUCCESS + assert len(seen_params) == 1 + assert [span.name for span in seen_params[0].spans.values()] == [ + "unrelated", + "duplicate-last", + ] + + exported_spans = exporter.get_finished_spans() + assert [span.name for span in exported_spans] == [ + "unrelated", + "duplicate-last", + ] + assert exported_spans[1].attributes == { + "attempt": 2, + "masking.applied": True, + } + assert "duplicate_span_count=1 remaining_span_count=2" in caplog.text + + def test_exporter_exception_does_not_stop_background_export_thread(): exporter = FailsOnceSpanExporter() media_manager, _ = _media_manager() From ade6251cd975bf9ad2c08285cfa58ba65e971ce4 Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 6 Aug 2026 11:07:14 +0200 Subject: [PATCH 2/5] push --- langfuse/_client/span_exporter.py | 8 -------- 1 file changed, 8 deletions(-) diff --git a/langfuse/_client/span_exporter.py b/langfuse/_client/span_exporter.py index 256787658..c42054863 100644 --- a/langfuse/_client/span_exporter.py +++ b/langfuse/_client/span_exporter.py @@ -285,14 +285,6 @@ def _apply_mask_otel_spans( span_attributes_by_identifier[identifier] = (span, attributes) - if duplicate_span_count: - langfuse_logger.warning( - "Masking warning: mask_otel_spans received duplicate span identifiers. " - "Keeping the last span for each identifier. " - f"duplicate_span_count={duplicate_span_count} " - f"remaining_span_count={len(span_attributes_by_identifier)}" - ) - maskable_span_attributes = list(span_attributes_by_identifier.values()) if not maskable_span_attributes: From 4d81406eea71e6225048b107311d3d15bd8ada58 Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 6 Aug 2026 11:07:58 +0200 Subject: [PATCH 3/5] push --- langfuse/_client/span_exporter.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/langfuse/_client/span_exporter.py b/langfuse/_client/span_exporter.py index c42054863..bbdc569cf 100644 --- a/langfuse/_client/span_exporter.py +++ b/langfuse/_client/span_exporter.py @@ -267,7 +267,6 @@ def _apply_mask_otel_spans( span_attributes_by_identifier: Dict[ OtelSpanIdentifier, tuple[ReadableSpan, Dict[str, AttributeValue]] ] = {} - duplicate_span_count = 0 for span, attributes in span_attributes: if not _has_valid_span_context(span): @@ -280,7 +279,6 @@ def _apply_mask_otel_spans( identifier = _create_otel_span_identifier(span) if identifier in span_attributes_by_identifier: - duplicate_span_count += 1 span_attributes_by_identifier.pop(identifier) span_attributes_by_identifier[identifier] = (span, attributes) From 2eeb949ec108b63369967e1e6c11a0cd1111a239 Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 6 Aug 2026 11:08:46 +0200 Subject: [PATCH 4/5] push --- tests/unit/test_mask_otel_spans.py | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/tests/unit/test_mask_otel_spans.py b/tests/unit/test_mask_otel_spans.py index 5f11e7c4c..8b695e5d1 100644 --- a/tests/unit/test_mask_otel_spans.py +++ b/tests/unit/test_mask_otel_spans.py @@ -574,7 +574,7 @@ def mask_otel_spans(*, params: MaskOtelSpansParams): ) -def test_mask_otel_spans_keeps_last_duplicate_without_dropping_batch(caplog): +def test_mask_otel_spans_keeps_last_duplicate_without_dropping_batch(): exporter = InMemorySpanExporter() seen_params: list[MaskOtelSpansParams] = [] @@ -631,9 +631,6 @@ def mask_otel_spans(*, params: MaskOtelSpansParams): ), ] - with caplog.at_level(logging.WARNING, logger="langfuse"): - result = transforming_exporter.export(spans) - assert result == SpanExportResult.SUCCESS assert len(seen_params) == 1 assert [span.name for span in seen_params[0].spans.values()] == [ From c5636b323fe6bef50e30ba03cf4339ac167b881b Mon Sep 17 00:00:00 2001 From: Hassieb Pakzad <68423100+hassiebp@users.noreply.github.com> Date: Thu, 6 Aug 2026 11:10:28 +0200 Subject: [PATCH 5/5] test(otel): fix duplicate masking regression test --- tests/unit/test_mask_otel_spans.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/unit/test_mask_otel_spans.py b/tests/unit/test_mask_otel_spans.py index 8b695e5d1..5d1d52ee4 100644 --- a/tests/unit/test_mask_otel_spans.py +++ b/tests/unit/test_mask_otel_spans.py @@ -631,6 +631,8 @@ def mask_otel_spans(*, params: MaskOtelSpansParams): ), ] + result = transforming_exporter.export(spans) + assert result == SpanExportResult.SUCCESS assert len(seen_params) == 1 assert [span.name for span in seen_params[0].spans.values()] == [ @@ -647,7 +649,6 @@ def mask_otel_spans(*, params: MaskOtelSpansParams): "attempt": 2, "masking.applied": True, } - assert "duplicate_span_count=1 remaining_span_count=2" in caplog.text def test_exporter_exception_does_not_stop_background_export_thread():