feat(otel): warn when mask cannot cover third-party span attributes

langfuse/_client/client.py

@@ -217,7 +217,7 @@ class Langfuse:
         release (Optional[str]): Release version/hash of your application. Used for grouping analytics by release.
         media_upload_thread_count (Optional[int]): Number of background threads for handling media uploads. Defaults to 1. Can also be set via LANGFUSE_MEDIA_UPLOAD_THREAD_COUNT environment variable.
         sample_rate (Optional[float]): Sampling rate for traces (0.0 to 1.0). Defaults to 1.0 (100% of traces are sampled). Can also be set via LANGFUSE_SAMPLE_RATE environment variable.
-        mask (Optional[MaskFunction]): Function to mask sensitive data synchronously when Langfuse SDK attributes are created. This applies only to data set through Langfuse SDK APIs such as `start_observation()`, `update()`, and `set_trace_io()`.
+        mask (Optional[MaskFunction]): Function to mask sensitive data synchronously when Langfuse SDK attributes are created. This applies only to data set through Langfuse SDK APIs such as `start_observation()`, `update()`, and `set_trace_io()`. If `mask_otel_spans` is not set and exported spans carry input or output attributes from other OpenTelemetry instrumentations, a one-time warning is logged.
         mask_otel_spans (Optional[MaskOtelSpansFunction]): Synchronous export-stage hook for masking raw OpenTelemetry span attributes before this Langfuse client sends them to Langfuse. Use this for spans created by third-party OpenTelemetry instrumentations, or when you need to inspect final span attributes after export filtering and Langfuse media handling. It does not modify spans already exported through other OpenTelemetry exporters.
 
             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.

langfuse/_client/resource_manager.py

@@ -260,6 +260,7 @@ def _initialize_instance(
                 additional_headers=additional_headers,
                 span_exporter=span_exporter,
                 media_manager=self._media_manager,
+                mask=mask,
                 mask_otel_spans=mask_otel_spans,
             )
             tracer_provider.add_span_processor(langfuse_processor)

langfuse/_client/span_exporter.py

@@ -94,16 +94,21 @@
 class LangfuseTransformingSpanExporter(SpanExporter):
     """Apply Langfuse export-stage transformations before delegating export."""
 
+    _MAX_UNMASKED_ATTRIBUTE_KEYS_IN_WARNING = 10
+
     def __init__(
         self,
         *,
         exporter: SpanExporter,
         media_manager: Optional[MediaManager],
         mask_otel_spans: Optional[MaskOtelSpansFunction],
+        mask_configured: bool = False,
     ) -> None:
         self._exporter = exporter
         self._media_manager = media_manager
         self._mask_otel_spans = mask_otel_spans
+        self._mask_configured = mask_configured
+        self._mask_coverage_warning_logged = False
 
     def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         span_attributes = [
@@ -126,6 +131,8 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
                 return SpanExportResult.SUCCESS
 
             span_attributes = masked_span_attributes
+        elif self._mask_configured:
+            self._warn_if_mask_misses_attributes(span_attributes=span_attributes)
 
         transformed_spans = [
             self._clone_span(span=span, attributes=attributes)
@@ -137,6 +144,36 @@ def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
 
         return self._exporter.export(transformed_spans)
 
+    def _warn_if_mask_misses_attributes(
+        self,
+        *,
+        span_attributes: Sequence[tuple[ReadableSpan, Dict[str, AttributeValue]]],
+    ) -> None:
+        if self._mask_coverage_warning_logged:
+            return
+
+        unmasked_keys = sorted(
+            {
+                key
+                for _, attributes in span_attributes
+                for key, value in attributes.items()
+                if _is_unmasked_io_attribute(key=key, value=value)
+            }
+        )
+
+        if not unmasked_keys:
+            return
+
+        self._mask_coverage_warning_logged = True
+        langfuse_logger.warning(
+            "Masking warning: `mask` only applies to data set through Langfuse SDK APIs "
+            "and does not inspect attributes set by other OpenTelemetry instrumentations. "
+            "Exported spans contain input or output attributes that `mask` did not "
+            "process: %s. Configure `mask_otel_spans` to mask them before export. "
+            "This warning is only logged once.",
+            ", ".join(unmasked_keys[: self._MAX_UNMASKED_ATTRIBUTE_KEYS_IN_WARNING]),
+        )
+
     def shutdown(self) -> None:
         self._exporter.shutdown()
 
@@ -664,6 +701,21 @@ def _media_field_for_attribute(
     return "metadata"
 
 
+def _is_unmasked_io_attribute(*, key: str, value: AttributeValue) -> bool:
+    if key.startswith("langfuse."):
+        return False
+
+    if _media_field_for_attribute(key) == "metadata":
+        return False
+
+    if isinstance(value, str):
+        return True
+
+    return _is_attribute_sequence(value) and any(
+        isinstance(item, str) for item in cast(Sequence[Any], value)
+    )
+
+
 def _get_trace_id(span: ReadableSpan) -> str:
     if span.context is None:
         return ""

langfuse/_client/span_processor.py

@@ -44,7 +44,7 @@
 from langfuse._task_manager.media_manager import MediaManager
 from langfuse._version import __version__ as langfuse_version
 from langfuse.logger import langfuse_logger
-from langfuse.types import MaskOtelSpansFunction
+from langfuse.types import MaskFunction, MaskOtelSpansFunction
 
 
 class LangfuseSpanProcessor(BatchSpanProcessor):
@@ -78,6 +78,7 @@ def __init__(
         additional_headers: Optional[Dict[str, str]] = None,
         span_exporter: Optional[SpanExporter] = None,
         media_manager: Optional[MediaManager] = None,
+        mask: Optional[MaskFunction] = None,
         mask_otel_spans: Optional[MaskOtelSpansFunction] = None,
     ):
         self.public_key = public_key
@@ -129,11 +130,12 @@ def __init__(
                 timeout=timeout,
             )
 
-        if media_manager is not None or mask_otel_spans is not None:
+        if media_manager is not None or mask_otel_spans is not None or mask is not None:
             span_exporter = LangfuseTransformingSpanExporter(
                 exporter=span_exporter,
                 media_manager=media_manager,
                 mask_otel_spans=mask_otel_spans,
+                mask_configured=mask is not None,
             )
 
         super().__init__(

tests/unit/test_mask_coverage_warning.py

@@ -0,0 +1,194 @@
+import logging
+from typing import Any, Optional
+
+import pytest
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+from opentelemetry.sdk.trace.export import SpanExporter
+
+from langfuse._client.client import Langfuse
+from langfuse._client.constants import LANGFUSE_TRACER_NAME
+from langfuse._client.span_processor import LangfuseSpanProcessor
+from langfuse.types import MaskFunction, MaskOtelSpansFunction, MaskOtelSpansParams
+
+COVERAGE_WARNING = (
+    "does not inspect attributes set by other OpenTelemetry instrumentations"
+)
+RAW_PROMPT = "raw prompt that names a client"
+
+
+def _mask(*, data: Any, **kwargs: Any) -> Any:
+    return "masked"
+
+
+def _tracer_provider(
+    *,
+    exporter: SpanExporter,
+    mask: Optional[MaskFunction] = None,
+    mask_otel_spans: Optional[MaskOtelSpansFunction] = None,
+) -> TracerProvider:
+    provider = TracerProvider(resource=Resource.create({"service.name": "test"}))
+    provider.add_span_processor(
+        LangfuseSpanProcessor(
+            public_key="test-public-key",
+            secret_key="test-secret-key",
+            base_url="http://localhost:3000",
+            flush_at=10,
+            flush_interval=1,
+            span_exporter=exporter,
+            mask=mask,
+            mask_otel_spans=mask_otel_spans,
+        )
+    )
+
+    return provider
+
+
+def _coverage_warnings(caplog: pytest.LogCaptureFixture) -> list[logging.LogRecord]:
+    return [
+        record for record in caplog.records if COVERAGE_WARNING in record.getMessage()
+    ]
+
+
+def test_mask_warns_once_when_third_party_spans_carry_input_attributes(
+    memory_exporter, caplog
+):
+    provider = _tracer_provider(exporter=memory_exporter, mask=_mask)
+    tracer = provider.get_tracer("openinference.instrumentation.openai")
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        for name in ("first-llm-call", "second-llm-call"):
+            with tracer.start_as_current_span(name) as span:
+                span.set_attribute("input.value", RAW_PROMPT)
+                span.set_attribute("llm.token_count.prompt", 12)
+
+        provider.force_flush()
+
+    warnings = _coverage_warnings(caplog)
+
+    assert len(warnings) == 1
+    assert "input.value" in warnings[0].getMessage()
+    assert "mask_otel_spans" in warnings[0].getMessage()
+    assert "llm.token_count.prompt" not in warnings[0].getMessage()
+
+
+def test_mask_coverage_warning_never_logs_attribute_values(memory_exporter, caplog):
+    provider = _tracer_provider(exporter=memory_exporter, mask=_mask)
+    tracer = provider.get_tracer("openinference.instrumentation.openai")
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        with tracer.start_as_current_span("third-party-llm-call") as span:
+            span.set_attribute("input.value", RAW_PROMPT)
+
+        provider.force_flush()
+
+    assert _coverage_warnings(caplog)
+    assert all(RAW_PROMPT not in record.getMessage() for record in caplog.records)
+
+
+def test_mask_coverage_warning_does_not_change_exported_attributes(memory_exporter):
+    provider = _tracer_provider(exporter=memory_exporter, mask=_mask)
+    tracer = provider.get_tracer("openinference.instrumentation.openai")
+
+    with tracer.start_as_current_span("third-party-llm-call") as span:
+        span.set_attribute("input.value", RAW_PROMPT)
+
+    provider.force_flush()
+
+    exported_span = memory_exporter.get_finished_spans()[0]
+
+    assert exported_span.attributes["input.value"] == RAW_PROMPT
+
+
+def test_mask_warns_for_third_party_attributes_on_langfuse_tracer_spans(
+    memory_exporter, caplog
+):
+    # Instrumentations wired through the Langfuse tracer, e.g.
+    # openlit.init(tracer=langfuse._otel_tracer), keep the langfuse-sdk scope,
+    # so detection has to key on attributes rather than instrumentation scope.
+    provider = _tracer_provider(exporter=memory_exporter, mask=_mask)
+    tracer = provider.get_tracer(
+        LANGFUSE_TRACER_NAME, attributes={"public_key": "test-public-key"}
+    )
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        with tracer.start_as_current_span("third-party-llm-call") as span:
+            span.set_attribute("input.value", RAW_PROMPT)
+
+        provider.force_flush()
+
+    assert len(_coverage_warnings(caplog)) == 1
+
+
+def test_mask_does_not_warn_for_native_langfuse_input_attributes(
+    memory_exporter, caplog
+):
+    provider = _tracer_provider(exporter=memory_exporter, mask=_mask)
+    tracer = provider.get_tracer(
+        LANGFUSE_TRACER_NAME, attributes={"public_key": "test-public-key"}
+    )
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        with tracer.start_as_current_span("langfuse-span") as span:
+            span.set_attribute("langfuse.observation.type", "span")
+            span.set_attribute("langfuse.observation.input", '"masked"')
+
+        provider.force_flush()
+
+    assert _coverage_warnings(caplog) == []
+
+
+def test_no_coverage_warning_when_mask_otel_spans_is_configured(
+    memory_exporter, caplog
+):
+    def mask_otel_spans(*, params: MaskOtelSpansParams):
+        return None
+
+    provider = _tracer_provider(
+        exporter=memory_exporter, mask=_mask, mask_otel_spans=mask_otel_spans
+    )
+    tracer = provider.get_tracer("openinference.instrumentation.openai")
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        with tracer.start_as_current_span("third-party-llm-call") as span:
+            span.set_attribute("input.value", RAW_PROMPT)
+
+        provider.force_flush()
+
+    assert _coverage_warnings(caplog) == []
+
+
+def test_no_coverage_warning_when_mask_is_not_configured(memory_exporter, caplog):
+    provider = _tracer_provider(exporter=memory_exporter)
+    tracer = provider.get_tracer("openinference.instrumentation.openai")
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        with tracer.start_as_current_span("third-party-llm-call") as span:
+            span.set_attribute("input.value", RAW_PROMPT)
+
+        provider.force_flush()
+
+    assert _coverage_warnings(caplog) == []
+
+
+def test_langfuse_client_mask_reaches_export_stage_warning(memory_exporter, caplog):
+    tracer_provider = TracerProvider(resource=Resource.create({"service.name": "test"}))
+    client = Langfuse(
+        public_key="test-public-key",
+        secret_key="test-secret-key",
+        base_url="http://localhost:3000",
+        mask=_mask,
+        span_exporter=memory_exporter,
+        tracer_provider=tracer_provider,
+    )
+    tracer = tracer_provider.get_tracer("openinference.instrumentation.openai")
+
+    with caplog.at_level(logging.WARNING, logger="langfuse"):
+        with tracer.start_as_current_span("third-party-llm-call") as span:
+            span.set_attribute("input.value", RAW_PROMPT)
+
+        client.flush()
+
+    client.shutdown()
+
+    assert len(_coverage_warnings(caplog)) == 1