feat: database spans, and pin that sampling stays env-driven

SQLAlchemy is instrumented with an explicit tracer_provider, so a slow
request can be attributed to the queries it made. Allowlisted only after
probing what those spans carry: db.statement keeps bound parameters as
placeholders, so chat message text stays in the database.

The outbound HTTP scopes (httpx, requests, urllib3) are deliberately still
absent. They are the transports the LLM vendor SDKs ride on, and those SDKs
instrument them globally against whichever provider is global, so admitting
them would put one span per outbound provider call into the operator's APM.
Outbound provider health has a leak-safe metrics path instead. Left as its
own decision rather than a side effect of this ticket.

Sampling needed no change: the provider takes no explicit sampler, so
OTEL_TRACES_SAMPLER already applies. Pinned end to end against a loopback
collector, 0 spans at ratio 0.0 and all 20 at 1.0, so that adding a sampler
later cannot silently take env control away.

Author: ogabrielluiz

.secrets.baseline

@@ -6696,7 +6696,7 @@
         "filename": "src/lfx/src/lfx/cli/serve_app.py",
         "hashed_secret": "b894b81be94cf8fa8d7536475aaec876addf05c8",
         "is_verified": false,
-        "line_number": 70,
+        "line_number": 71,
         "is_secret": false
       }
     ],
@@ -7309,5 +7309,5 @@
       }
     ]
   },
-  "generated_at": "2026-08-04T18:09:09Z"
+  "generated_at": "2026-08-05T14:27:06Z"
 }

src/backend/base/langflow/main.py

@@ -21,7 +21,12 @@
 from filelock import FileLock
 from lfx.interface.utils import setup_llm_caching
 from lfx.log.logger import configure, logger
-from lfx.observability import instrument_fastapi_app, start_event_loop_lag_monitor, stop_event_loop_lag_monitor
+from lfx.observability import (
+    instrument_dependencies,
+    instrument_fastapi_app,
+    start_event_loop_lag_monitor,
+    stop_event_loop_lag_monitor,
+)
 from pydantic import PydanticDeprecatedSince20
 from pydantic_core import PydanticSerializationError
 from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
@@ -988,6 +993,9 @@ async def exception_handler(_request: Request, exc: Exception):
     # lazy-include route patch + instrument_app). The helper lives in lfx so lfx serve
     # instruments its own app the same way.
     instrument_fastapi_app(app)
+    # The rest of the trace tree: DB queries and outbound HTTP made during a run. The engine is
+    # not built yet at app-factory time, so SQLAlchemy instruments its own dispatcher globally.
+    instrument_dependencies()
 
     add_pagination(app)
 

src/backend/base/pyproject.toml

@@ -69,6 +69,7 @@ dependencies = [
     "opentelemetry-exporter-prometheus>=0.50b0,<1.0.0",
     "opentelemetry-exporter-otlp>=1.30.0,<2.0.0",
     "opentelemetry-instrumentation-fastapi>=0.50b0,<1.0.0",
+    "opentelemetry-instrumentation-sqlalchemy>=0.50b0,<1.0.0",
     "opentelemetry-instrumentation-requests>=0.50b0,<1.0.0",
     "opentelemetry-instrumentation-urllib3>=0.50b0,<1.0.0",
     # Process CPU, memory, GC and thread counts. Without it the only metrics the APM gets are

src/backend/tests/unit/services/telemetry/test_dependency_spans.py

@@ -0,0 +1,121 @@
+"""Dependency spans and sampling.
+
+Database spans are allowlisted for export; the LLM vendor transports deliberately are not.
+Sampling needs no code of ours, so the check here is that it stays that way.
+
+Each case runs in a subprocess because the tracer provider is process-global.
+"""
+
+import gzip
+import http.server
+import os
+import subprocess
+import sys
+import threading
+
+import pytest
+from lfx.observability import APPLICATION_INSTRUMENTATION_SCOPES
+
+
+class _Collector(http.server.BaseHTTPRequestHandler):
+    def do_POST(self) -> None:
+        body = self.rfile.read(int(self.headers.get("Content-Length", 0)))
+        if self.headers.get("Content-Encoding") == "gzip":
+            body = gzip.decompress(body)
+        self.server.requests.append((self.path, body))
+        self.send_response(200)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def log_message(self, *args) -> None:
+        pass
+
+
+@pytest.fixture
+def collector():
+    server = http.server.HTTPServer(("127.0.0.1", 0), _Collector)
+    server.requests = []
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    yield server
+    server.shutdown()
+    server.server_close()
+    thread.join(timeout=5)
+
+
+def test_outbound_http_scopes_stay_out_of_the_allowlist():
+    """Redacting URLs does not make the LLM transports safe to export; that boundary stands."""
+    for scope in (
+        "opentelemetry.instrumentation.httpx",
+        "opentelemetry.instrumentation.requests",
+        "opentelemetry.instrumentation.urllib3",
+    ):
+        assert scope not in APPLICATION_INSTRUMENTATION_SCOPES
+
+
+def test_database_spans_are_allowlisted():
+    """Verified separately to carry bound-parameter placeholders, never row values."""
+    assert "opentelemetry.instrumentation.sqlalchemy" in APPLICATION_INSTRUMENTATION_SCOPES
+
+
+# The ticket's sampler criterion, run against the real bootstrap rather than a hand-built
+# provider: the same request at ratio 0.0 must export nothing and at 1.0 must export the lot.
+SAMPLER_PROBE = """
+import json
+from opentelemetry import trace
+from lfx.observability import APPLICATION_TRACER_NAME, bootstrap_application_telemetry
+
+telemetry = bootstrap_application_telemetry()
+tracer = trace.get_tracer(APPLICATION_TRACER_NAME)
+for _ in range(20):
+    with tracer.start_as_current_span("flow.execute"):
+        pass
+telemetry.shutdown()
+print("PROBE_RESULT done")
+"""
+
+
+def _run_sampler_probe(endpoint: str, ratio: str) -> None:
+    env = {k: v for k, v in os.environ.items() if not k.startswith("OTEL_")}
+    env["OTEL_EXPORTER_OTLP_ENDPOINT"] = endpoint
+    env["OTEL_EXPORTER_OTLP_PROTOCOL"] = "http/protobuf"
+    env["OTEL_TRACES_SAMPLER"] = "traceidratio"
+    env["OTEL_TRACES_SAMPLER_ARG"] = ratio
+    completed = subprocess.run(  # noqa: S603
+        [sys.executable, "-c", SAMPLER_PROBE],
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=300,
+        check=False,
+    )
+    assert completed.returncode == 0, completed.stderr
+
+
+def _exported_span_count(requests_seen) -> int:
+    from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
+
+    total = 0
+    for path, body in requests_seen:
+        if path != "/v1/traces":
+            continue
+        request = ExportTraceServiceRequest()
+        request.ParseFromString(body)
+        for rs in request.resource_spans:
+            for ss in rs.scope_spans:
+                total += len(ss.spans)
+    return total
+
+
+def test_sampling_off_exports_nothing(collector):
+    port = collector.server_address[1]
+    _run_sampler_probe(f"http://127.0.0.1:{port}", "0.0")
+
+    assert _exported_span_count(collector.requests) == 0
+
+
+def test_sampling_on_exports_every_span(collector):
+    port = collector.server_address[1]
+    _run_sampler_probe(f"http://127.0.0.1:{port}", "1.0")
+
+    assert _exported_span_count(collector.requests) == 20

src/lfx/pyproject.toml

@@ -90,6 +90,7 @@ otel = [
     "opentelemetry-exporter-otlp>=1.30.0,<2.0.0",
     "opentelemetry-exporter-prometheus>=0.50b0,<1.0.0",
     "opentelemetry-instrumentation-fastapi>=0.50b0,<1.0.0",
+    "opentelemetry-instrumentation-sqlalchemy>=0.50b0,<1.0.0",
     "opentelemetry-instrumentation-system-metrics>=0.50b0,<1.0.0",
 ]
 # Engine-only by default: ``pip install lfx`` ships NO bundles.  This optional

src/lfx/src/lfx/cli/serve_app.py

@@ -47,6 +47,7 @@
 from lfx.log.logger import logger
 from lfx.observability import (
     bootstrap_application_telemetry,
+    instrument_dependencies,
     instrument_fastapi_app,
     start_event_loop_lag_monitor,
     stop_event_loop_lag_monitor,
@@ -707,6 +708,7 @@ async def _lifespan(_app: FastAPI):
     )
 
     instrument_fastapi_app(app)
+    instrument_dependencies()
 
     app.state.registry = registry
     # Snapshot the API key once so per-request auth (verify_api_key, run on a threadpool

src/lfx/src/lfx/observability.py

@@ -96,6 +96,16 @@
     {
         "opentelemetry.instrumentation.asgi",
         "opentelemetry.instrumentation.fastapi",
+        # DB spans. Admitted after checking what they actually carry: db.statement keeps bound
+        # parameters as placeholders ("INSERT INTO messagetable (text) VALUES (?)"), so chat
+        # message text stays in the database and out of the APM. Verified by probe, because
+        # "it probably does not log values" is how the httpx hole got opened the first time.
+        #
+        # The outbound HTTP scopes (httpx, requests, urllib3) remain deliberately absent. The
+        # LLM vendor SDKs instrument them globally against whatever provider is global (ours),
+        # so admitting them would put one span per outbound LLM call in the operator's APM.
+        # Outbound provider health is delivered as leak-safe metrics instead.
+        "opentelemetry.instrumentation.sqlalchemy",
         APPLICATION_TRACER_NAME,
     }
 )
@@ -580,6 +590,41 @@ def bootstrap_application_telemetry(*, prometheus_enabled: bool = False) -> Appl
     )
 
 
+def instrument_dependencies(*, engine: object | None = None) -> None:
+    """Instrument the database, so a slow request can be attributed to the queries it made.
+
+    Deliberately does NOT instrument httpx or requests. Those are the transports the LLM
+    vendor SDKs ride on, and instrumenting them globally would put one span per outbound
+    provider call into the operator's APM. That boundary is held elsewhere too (the export
+    filter does not allowlist those scopes), and outbound provider health is delivered as
+    leak-safe metrics rather than spans.
+
+    The instrumentor is given an explicit ``tracer_provider``. That is not decoration: a bare
+    ``instrument()`` binds to whatever provider is global, which is how vendor SDKs end up
+    exporting through ours.
+
+    Optional and failure-tolerant: a missing package or a double-instrument call must not take
+    the app down, because none of this is worth a failed boot.
+    """
+    _instrument_sqlalchemy(engine)
+
+
+def _instrument_sqlalchemy(engine: object | None) -> None:
+    try:
+        from opentelemetry import trace
+        from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
+    except ImportError:
+        return
+    try:
+        kwargs = {"tracer_provider": trace.get_tracer_provider()}
+        if engine is not None:
+            # An async engine exposes the sync one the instrumentor actually patches.
+            kwargs["engine"] = getattr(engine, "sync_engine", engine)
+        SQLAlchemyInstrumentor().instrument(**kwargs)
+    except Exception:  # noqa: BLE001 - see above
+        logger.debug("sqlalchemy instrumentation unavailable; DB spans will be missing", exc_info=True)
+
+
 def instrument_fastapi_app(app: FastAPI) -> None:
     """Instrument an ASGI app for HTTP server telemetry under the stable conventions.