feat(tracing): add export-stage otel span masking

Author: hassiebp

langfuse/__init__.py

@@ -36,6 +36,14 @@
     is_known_llm_instrumentor,
     is_langfuse_span,
 )
+from .types import (
+    MaskOtelSpansFunction,
+    MaskOtelSpansParams,
+    MaskOtelSpansResult,
+    OtelSpanData,
+    OtelSpanIdentifier,
+    OtelSpanPatch,
+)
 
 Langfuse = _client_module.Langfuse
 
@@ -71,6 +79,12 @@
     "is_genai_span",
     "is_known_llm_instrumentor",
     "KNOWN_LLM_INSTRUMENTATION_SCOPE_PREFIXES",
+    "MaskOtelSpansFunction",
+    "MaskOtelSpansParams",
+    "MaskOtelSpansResult",
+    "OtelSpanData",
+    "OtelSpanIdentifier",
+    "OtelSpanPatch",
     "experiment",
     "api",
 ]

langfuse/_client/client.py

@@ -132,7 +132,13 @@
     PromptClient,
     TextPromptClient,
 )
-from langfuse.types import MaskFunction, ScoreDataType, SpanLevel, TraceContext
+from langfuse.types import (
+    MaskFunction,
+    MaskOtelSpansFunction,
+    ScoreDataType,
+    SpanLevel,
+    TraceContext,
+)
 
 
 class Langfuse:
@@ -170,6 +176,7 @@ class Langfuse:
         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 in traces before sending to the API.
+        mask_otel_spans (Optional[MaskOtelSpansFunction]): Synchronous export-stage hook that receives a batch of OpenTelemetry span snapshots and can return sparse attribute patches before spans are sent to Langfuse. If the hook raises or returns an invalid batch result, the whole export batch is dropped. During flush and shutdown, this hook may run on the caller thread.
         blocked_instrumentation_scopes (Optional[List[str]]): Deprecated. Use `should_export_span` instead. Equivalent behavior:
             ```python
             from langfuse.span_filter import is_default_export_span
@@ -226,6 +233,7 @@ class Langfuse:
 
     _resources: Optional[LangfuseResourceManager] = None
     _mask: Optional[MaskFunction] = None
+    _mask_otel_spans: Optional[MaskOtelSpansFunction] = None
     _otel_tracer: otel_trace_api.Tracer
 
     def __init__(
@@ -246,6 +254,7 @@ def __init__(
         media_upload_thread_count: Optional[int] = None,
         sample_rate: Optional[float] = None,
         mask: Optional[MaskFunction] = None,
+        mask_otel_spans: Optional[MaskOtelSpansFunction] = None,
         blocked_instrumentation_scopes: Optional[List[str]] = None,
         should_export_span: Optional[Callable[[ReadableSpan], bool]] = None,
         additional_headers: Optional[Dict[str, str]] = None,
@@ -342,6 +351,7 @@ def __init__(
             media_upload_thread_count=media_upload_thread_count,
             sample_rate=sample_rate,
             mask=mask,
+            mask_otel_spans=mask_otel_spans,
             tracing_enabled=self._tracing_enabled,
             blocked_instrumentation_scopes=blocked_instrumentation_scopes,
             should_export_span=should_export_span,
@@ -350,6 +360,7 @@ def __init__(
             span_exporter=span_exporter,
         )
         self._mask = self._resources.mask
+        self._mask_otel_spans = self._resources.mask_otel_spans
 
         self._otel_tracer = (
             self._resources.tracer

langfuse/_client/get_client.py

@@ -50,6 +50,7 @@ def _create_client_from_instance(
         media_upload_thread_count=instance.media_upload_thread_count,
         sample_rate=instance.sample_rate,
         mask=instance.mask,
+        mask_otel_spans=instance.mask_otel_spans,
         blocked_instrumentation_scopes=instance.blocked_instrumentation_scopes,
         should_export_span=instance.should_export_span,
         additional_headers=instance.additional_headers,

langfuse/_client/resource_manager.py

@@ -45,7 +45,7 @@
 from langfuse._utils.request import LangfuseClient
 from langfuse.api import AsyncLangfuseAPI, LangfuseAPI
 from langfuse.logger import langfuse_logger
-from langfuse.types import MaskFunction
+from langfuse.types import MaskFunction, MaskOtelSpansFunction
 
 from .._version import __version__ as langfuse_version
 
@@ -94,6 +94,7 @@ def __new__(
         media_upload_thread_count: Optional[int] = None,
         sample_rate: Optional[float] = None,
         mask: Optional[MaskFunction] = None,
+        mask_otel_spans: Optional[MaskOtelSpansFunction] = None,
         tracing_enabled: Optional[bool] = None,
         blocked_instrumentation_scopes: Optional[List[str]] = None,
         should_export_span: Optional[Callable[[ReadableSpan], bool]] = None,
@@ -128,6 +129,7 @@ def __new__(
                     media_upload_thread_count=media_upload_thread_count,
                     sample_rate=sample_rate,
                     mask=mask,
+                    mask_otel_spans=mask_otel_spans,
                     tracing_enabled=tracing_enabled
                     if tracing_enabled is not None
                     else True,
@@ -157,6 +159,7 @@ def _initialize_instance(
         httpx_client: Optional[httpx.Client] = None,
         sample_rate: Optional[float] = None,
         mask: Optional[MaskFunction] = None,
+        mask_otel_spans: Optional[MaskOtelSpansFunction] = None,
         tracing_enabled: bool = True,
         blocked_instrumentation_scopes: Optional[List[str]] = None,
         should_export_span: Optional[Callable[[ReadableSpan], bool]] = None,
@@ -169,6 +172,7 @@ def _initialize_instance(
         self.tracing_enabled = tracing_enabled
         self.base_url = base_url
         self.mask = mask
+        self.mask_otel_spans = mask_otel_spans
         self.environment = environment
 
         # Store additional client settings for get_client() to use
@@ -184,33 +188,6 @@ def _initialize_instance(
         self.span_exporter = span_exporter
         self.tracer_provider: Optional[TracerProvider] = None
 
-        # OTEL Tracer
-        if tracing_enabled:
-            tracer_provider = tracer_provider or _init_tracer_provider(
-                environment=environment, release=release, sample_rate=sample_rate
-            )
-            self.tracer_provider = tracer_provider
-
-            langfuse_processor = LangfuseSpanProcessor(
-                public_key=self.public_key,
-                secret_key=secret_key,
-                base_url=base_url,
-                timeout=timeout,
-                flush_at=flush_at,
-                flush_interval=flush_interval,
-                blocked_instrumentation_scopes=blocked_instrumentation_scopes,
-                should_export_span=should_export_span,
-                additional_headers=additional_headers,
-                span_exporter=span_exporter,
-            )
-            tracer_provider.add_span_processor(langfuse_processor)
-
-            self._otel_tracer = tracer_provider.get_tracer(
-                LANGFUSE_TRACER_NAME,
-                langfuse_version,
-                attributes={"public_key": self.public_key},
-            )
-
         # API Clients
 
         ## API clients must be singletons because the underlying HTTPX clients
@@ -266,6 +243,35 @@ def _initialize_instance(
         )
         self._media_upload_consumers = []
 
+        # OTEL Tracer
+        if tracing_enabled:
+            tracer_provider = tracer_provider or _init_tracer_provider(
+                environment=environment, release=release, sample_rate=sample_rate
+            )
+            self.tracer_provider = tracer_provider
+
+            langfuse_processor = LangfuseSpanProcessor(
+                public_key=self.public_key,
+                secret_key=secret_key,
+                base_url=base_url,
+                timeout=timeout,
+                flush_at=flush_at,
+                flush_interval=flush_interval,
+                blocked_instrumentation_scopes=blocked_instrumentation_scopes,
+                should_export_span=should_export_span,
+                additional_headers=additional_headers,
+                span_exporter=span_exporter,
+                media_manager=self._media_manager,
+                mask_otel_spans=mask_otel_spans,
+            )
+            tracer_provider.add_span_processor(langfuse_processor)
+
+            self._otel_tracer = tracer_provider.get_tracer(
+                LANGFUSE_TRACER_NAME,
+                langfuse_version,
+                attributes={"public_key": self.public_key},
+            )
+
         media_upload_thread_count = media_upload_thread_count or max(
             int(os.getenv(LANGFUSE_MEDIA_UPLOAD_THREAD_COUNT, 1)), 1
         )