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

src/backend/base/langflow/main.py

@@ -21,7 +21,11 @@
 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_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

src/backend/base/langflow/services/database/service.py

@@ -16,6 +16,7 @@
 from alembic import command, util
 from alembic.config import Config
 from lfx.log.logger import logger
+from lfx.observability import instrument_database
 from lfx.services.deps import session_scope
 from sqlalchemy import event, inspect
 from sqlalchemy.dialects import sqlite as dialect_sqlite
@@ -280,6 +281,12 @@ def __init__(self, settings_service: SettingsService):
             self.engine = self._create_engine_with_retry()
         else:
             self.engine = self._create_engine()
+        # Here rather than in the app factory: this is an AsyncEngine, and the instrumentor
+        # patches the sync engine underneath it. Instrumenting globally instead (no engine
+        # argument) attaches to pool events only, which yields a connect span per checkout and
+        # not a single query span. Verified against a live backend: 13 connect spans and zero
+        # db.statement for one API request.
+        instrument_database(self.engine)
 
         # Create async session maker for efficient session creation
         # This is the recommended SQLAlchemy 2.0+ pattern

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/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,
     }
 )
@@ -617,6 +627,41 @@ def bootstrap_application_telemetry(*, prometheus_enabled: bool = False) -> Appl
     )
 
 
+def instrument_database(engine: object) -> None:
+    """Instrument a SQLAlchemy engine, so a slow request can be attributed to its queries.
+
+    Takes the engine rather than instrumenting globally. Langflow's engine is an AsyncEngine and
+    the instrumentor patches the sync engine underneath it; a global ``instrument()`` with no
+    engine attaches to pool events only, which produces a connect span per checkout and no query
+    spans at all. Verified against a live backend: 13 connect spans and zero db.statement for one
+    API request. That is worse than nothing, because it looks like DB visibility.
+
+    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.
+
+    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:
+    try:
+        from opentelemetry import trace
+        from opentelemetry.instrumentation.sqlalchemy import SQLAlchemyInstrumentor
+    except ImportError:
+        return
+    try:
+        SQLAlchemyInstrumentor().instrument(
+            # An AsyncEngine exposes the sync one the instrumentor actually patches.
+            engine=getattr(engine, "sync_engine", engine),
+            tracer_provider=trace.get_tracer_provider(),
+        )
+    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.