fix: filter Traceloop's own processor instead of patching the provider

Two problems with patching the global provider's add_span_processor.

When no OTLP endpoint is configured the bootstrap installs nothing and the
global provider is still a proxy, so there was nothing to patch and the filter
was skipped. Traceloop then registers its own concrete provider, the proxy
resolves onto it, and the service's spans reach the vendor by the same route.
That is the ordinary Traceloop-without-an-APM setup, not an edge case.

The provider is also shared: a processor another integration registered while
init was running would have been wrapped too, and would then have had this
filter's boundary applied backwards.

Wrap the SDK's get_default_span_processor for the duration of init instead. It
applies whichever provider Traceloop ends up with, and touches nothing outside
the SDK. As a side effect the vendor-first ordering no longer leaks either: the
APM still receives nothing, but the span is dropped rather than exported.

Author: ogabrielluiz

src/backend/base/langflow/services/tracing/traceloop.py

@@ -84,32 +84,38 @@ def _application_telemetry_withheld():
     attaches to, and its exporter then sees every span on it -- including the service's own
     HTTP and flow spans, which the operator pointed at their APM and not at Traceloop.
 
-    Wrapping ``add_span_processor`` for the duration of ``init`` rather than passing
-    ``processor=`` leaves the SDK's own pipeline alone. Supplying a processor also turns off
-    its metrics, drops the ``TRACELOOP_HEADERS`` auth the Instana integration depends on,
-    makes ``disable_batch`` inert and skips prompt sync, none of which this needs to touch.
-    It also fails the right way: a span can only reach the vendor via a processor added to
-    our provider, and that is the one call intercepted here.
-
-    Serialised because flows run concurrently and the patch is on a process-wide object: two
+    When no OTLP endpoint is configured the bootstrap installs nothing and the global provider
+    is still a proxy, which does not make the problem go away: the SDK then creates a concrete
+    provider and registers it globally, the proxy resolves onto it, and the service's spans
+    reach the vendor by the same route. So the filter has to apply in both cases.
+
+    It is applied to the SDK's own processor factory rather than to the provider. Patching the
+    provider's ``add_span_processor`` only covers the case where a provider already exists, and
+    it is a shared object -- a processor another integration registers while ``init`` is running
+    would be wrapped too, and would then have this filter's boundary applied backwards.
+
+    Passing ``processor=`` to ``init`` is the other obvious hook and is worse: it turns off the
+    SDK's metrics, drops the ``TRACELOOP_HEADERS`` auth the Instana integration depends on,
+    makes ``disable_batch`` inert and skips prompt sync. Wrapping the factory leaves all of
+    that alone, because from ``init``'s point of view no processor was supplied.
+
+    Serialised because flows run concurrently and the patch is on a module attribute: two
     tracers initialising at once would otherwise have the first one's restore run while the
     second is still inside init, leaving that run's processor unwrapped.
     """
+    from traceloop.sdk.tracing import tracing as traceloop_tracing
+
     with _INIT_LOCK:
-        provider = trace.get_tracer_provider()
-        if not hasattr(provider, "add_span_processor"):
-            # A proxy provider: the SDK builds its own, which carries nothing of ours.
-            yield
-            return
+        original = traceloop_tracing.get_default_span_processor
 
-        def add_span_processor(processor: SpanProcessor) -> None:
-            type(provider).add_span_processor(provider, _ApplicationScopeFilter(processor))
+        def get_default_span_processor(*args, **kwargs) -> SpanProcessor:
+            return _ApplicationScopeFilter(original(*args, **kwargs))
 
-        provider.add_span_processor = add_span_processor
+        traceloop_tracing.get_default_span_processor = get_default_span_processor
         try:
             yield
         finally:
-            del provider.add_span_processor
+            traceloop_tracing.get_default_span_processor = original
 
 
 class TraceloopTracer(BaseTracer):

src/backend/tests/unit/services/tracing/test_traceloop_application_telemetry.py

@@ -106,6 +106,33 @@ def test_application_spans_do_not_reach_the_llm_vendor():
     assert result["traceloop"] == [], "application telemetry leaked to the LLM vendor"
 
 
+def test_application_spans_do_not_reach_the_vendor_without_an_apm():
+    """The filter must also apply when no APM is configured, which is the common setup.
+
+    With no OTLP endpoint the bootstrap installs nothing and the global provider is still a
+    proxy. That does not mean there is nothing to filter: Traceloop then creates the concrete
+    provider and registers it globally, the proxy resolves onto it, and the service's own spans
+    reach the vendor by exactly the same route.
+    """
+    result = _run("""
+        from lfx.observability import bootstrap_application_telemetry, APPLICATION_TRACER_NAME
+        telemetry = bootstrap_application_telemetry(prometheus_enabled=False)
+
+        from langflow.services.tracing.traceloop import TraceloopTracer
+        tracer = TraceloopTracer(
+            trace_name="probe", trace_type="chain", project_name="probe", trace_id=uuid.uuid4()
+        )
+
+        from opentelemetry import trace
+        trace.get_tracer(APPLICATION_TRACER_NAME).start_span("flow.execute").end()
+        report(ready=tracer._ready, installed=telemetry.tracer_provider is not None)
+    """)
+
+    assert result["installed"] is False, "no OTLP endpoint means the bootstrap installs no provider"
+    assert result["ready"] is True, "the vendor integration must still initialise"
+    assert result["traceloop"] == [], "application telemetry leaked to the LLM vendor"
+
+
 def test_vendor_spans_still_reach_the_vendor():
     """Filtering our telemetry out must not cost Traceloop its own spans.
 
@@ -135,16 +162,19 @@ def test_vendor_spans_still_reach_the_vendor():
 
 
 def test_vendor_first_leaves_the_operator_without_application_telemetry():
-    """Characterises the reverse order, which this change does not fix.
+    """Characterises the reverse order: no leak, but no application telemetry either.
 
     When Traceloop initialises before the bootstrap it claims the global provider, and OTel's
     set_tracer_provider is one-shot, so enabling OTLP afterwards in the same process cannot
-    install ours. The filter goes on our provider, so with Traceloop owning the global one
-    there is nothing to put it on and application spans still reach the vendor.
+    install ours. The operator's APM therefore receives nothing, and that part is a real
+    limitation rather than something this change fixes.
+
+    What it does fix is the direction that matters: the filter is on Traceloop's own processor,
+    so it applies whichever provider Traceloop ended up with. The application span is dropped
+    rather than shipped to the vendor.
 
     In practice the bootstrap runs at app startup and Traceloop only on the first flow run, so
-    our provider wins. This pins the limitation rather than asserting it is acceptable: if the
-    ordering is ever made robust, traceloop becomes empty and this test should say so.
+    this ordering does not arise in a deployed process.
     """
     result = _run("""
         from langflow.services.tracing.traceloop import TraceloopTracer
@@ -162,4 +192,4 @@ def test_vendor_first_leaves_the_operator_without_application_telemetry():
 
     assert result["installed"] is False, "the bootstrap must decline rather than fight for the global"
     assert result["apm"] == [], "known limitation: the APM receives nothing when the vendor initialises first"
-    assert result["traceloop"] == ["flow.execute"], "known limitation: the vendor receives it instead"
+    assert result["traceloop"] == [], "but application telemetry must still not reach the vendor"