feat(tracing): add datastore tracing plumbing

.env.example

@@ -105,7 +105,10 @@ RELAY_URL=ws://localhost:3000
 # -----------------------------------------------------------------------------
 # Logging / Tracing
 # -----------------------------------------------------------------------------
-RUST_LOG=buzz_relay=debug,buzz_db=debug,buzz_auth=debug,buzz_pubsub=debug,tower_http=debug
+RUST_LOG=buzz_relay=debug,buzz_datastore=info,buzz_db=debug,buzz_auth=debug,buzz_pubsub=debug,tower_http=debug
+# Optional OpenTelemetry-only target filter. This is deliberately independent
+# from RUST_LOG so log verbosity changes cannot break trace parentage.
+# BUZZ_OTEL_FILTER=buzz_relay=info,buzz_datastore=info
 
 # OTLP tracing endpoint (optional — leave unset to disable)
 # OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317

crates/buzz-relay/Cargo.toml

@@ -92,3 +92,4 @@ reqwest = { workspace = true }
 tokio-tungstenite = { workspace = true }
 futures = "0.3"
 flate2 = "1.1.9"
+opentelemetry_sdk = { workspace = true, features = ["testing"] }

crates/buzz-relay/src/main.rs

@@ -4,6 +4,10 @@ use std::sync::Arc;
 
 use tracing::{error, info, warn};
 use tracing_subscriber::{fmt, prelude::*, EnvFilter};
+
+fn log_env_filter(rust_log: Option<&str>) -> EnvFilter {
+    EnvFilter::new(rust_log.unwrap_or("buzz_relay=info"))
+}
 use uuid::Uuid;
 
 use buzz_audit::AuditService;
@@ -107,9 +111,17 @@ async fn main() -> anyhow::Result<()> {
     };
 
     tracing_subscriber::registry()
-        .with(fmt::layer().json().flatten_event(true))
-        .with(EnvFilter::from_default_env().add_directive("buzz_relay=info".parse()?))
-        .with(otel_layer)
+        .with(
+            fmt::layer()
+                .json()
+                .flatten_event(true)
+                .with_filter(log_env_filter(std::env::var("RUST_LOG").ok().as_deref())),
+        )
+        .with(otel_layer.map(|layer| {
+            layer.with_filter(telemetry::otel_env_filter(
+                std::env::var("BUZZ_OTEL_FILTER").ok().as_deref(),
+            ))
+        }))
         .init();
 
     // Log any exporter-build failure now that the subscriber is installed.
@@ -1060,6 +1072,52 @@ async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[cfg(test)]
+mod env_filter_tests {
+    use super::log_env_filter;
+    use buzz_relay::telemetry::otel_env_filter;
+    use tracing_subscriber::prelude::*;
+
+    #[test]
+    fn unset_enables_datastore_only_for_otel_filter() {
+        let logs = tracing_subscriber::registry().with(log_env_filter(None));
+        tracing::subscriber::with_default(logs, || {
+            assert!(!tracing::enabled!(target: "buzz_datastore", tracing::Level::INFO));
+            assert!(tracing::enabled!(target: "buzz_relay", tracing::Level::INFO));
+        });
+
+        let otel = tracing_subscriber::registry().with(otel_env_filter(None));
+        tracing::subscriber::with_default(otel, || {
+            assert!(tracing::enabled!(target: "buzz_datastore", tracing::Level::INFO));
+        });
+    }
+
+    #[test]
+    fn explicit_datastore_off_is_preserved_alone() {
+        assert_eq!(
+            otel_env_filter(Some("buzz_datastore=off")).to_string(),
+            "buzz_datastore=off"
+        );
+    }
+
+    #[test]
+    fn explicit_datastore_debug_is_preserved_alone() {
+        assert_eq!(
+            otel_env_filter(Some("buzz_datastore=debug")).to_string(),
+            "buzz_datastore=debug"
+        );
+    }
+
+    #[test]
+    fn log_and_otel_filters_are_configured_independently() {
+        assert_eq!(log_env_filter(Some("warn")).to_string(), "warn");
+        assert_eq!(
+            otel_env_filter(Some("buzz_relay=debug")).to_string(),
+            "buzz_relay=debug"
+        );
+    }
+}
+
 async fn run_community_revalidator(
     state: Arc<AppState>,
     period: std::time::Duration,

crates/buzz-relay/src/router.rs

@@ -4,8 +4,9 @@ use std::sync::atomic::Ordering;
 use std::sync::Arc;
 
 use axum::{
+    body::Body,
     extract::{ConnectInfo, FromRequest, State, WebSocketUpgrade},
-    http::{HeaderMap, StatusCode},
+    http::{HeaderMap, Request, StatusCode},
     middleware,
     response::{IntoResponse, Json},
     routing::{get, post, put},
@@ -16,7 +17,7 @@ use tower::ServiceExt;
 use tower_http::cors::{AllowOrigin, CorsLayer};
 use tower_http::limit::RequestBodyLimitLayer;
 use tower_http::services::ServeDir;
-use tower_http::trace::TraceLayer;
+use tower_http::trace::{HttpMakeClassifier, TraceLayer};
 
 use crate::api;
 use crate::audio;
@@ -187,10 +188,23 @@ pub fn build_router(state: Arc<AppState>) -> Router {
 
     merged
         .layer(middleware::from_fn(track_metrics))
-        .layer(TraceLayer::new_for_http())
+        .layer(http_trace_layer())
         .layer(build_cors_layer(&state.config.cors_origins))
 }
 
+fn http_trace_layer() -> TraceLayer<HttpMakeClassifier, fn(&Request<Body>) -> tracing::Span> {
+    TraceLayer::new_for_http().make_span_with(make_http_span as fn(&Request<Body>) -> tracing::Span)
+}
+
+fn make_http_span(request: &Request<Body>) -> tracing::Span {
+    tracing::info_span!(
+        target: "buzz_relay",
+        "http.request",
+        otel.kind = "server",
+        http.request.method = %request.method(),
+    )
+}
+
 fn is_admin_spa_path(path: &str) -> bool {
     path == "/"
         || path == "/reports"
@@ -435,9 +449,14 @@ fn build_cors_layer(cors_origins: &[String]) -> CorsLayer {
 mod tests {
     use axum::{routing::get, Router};
     use futures_util::SinkExt;
+    use opentelemetry::trace::TracerProvider as _;
+    use opentelemetry_sdk::trace::{InMemorySpanExporter, SdkTracerProvider};
     use tokio::net::TcpListener;
     use tokio::sync::mpsc;
     use tokio_tungstenite::{connect_async, tungstenite::Message};
+    use tower::ServiceBuilder;
+    use tracing::Instrument as _;
+    use tracing_subscriber::prelude::*;
 
     use super::*;
 
@@ -471,6 +490,60 @@ mod tests {
         assert!(!should_serve_spa("/arbitrary", true));
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn http_and_datastore_spans_are_exported_in_the_same_trace() {
+        let exporter = InMemorySpanExporter::default();
+        let provider = SdkTracerProvider::builder()
+            .with_simple_exporter(exporter.clone())
+            .build();
+        let subscriber = tracing_subscriber::registry().with(
+            tracing_opentelemetry::layer()
+                .with_tracer(provider.tracer("test"))
+                .with_filter(crate::telemetry::otel_env_filter(None)),
+        );
+        let _subscriber_guard = tracing::subscriber::set_default(subscriber);
+        let service = ServiceBuilder::new()
+            .layer(http_trace_layer())
+            .service(tower::service_fn(
+                |_: axum::http::Request<axum::body::Body>| async {
+                    async {}
+                        .instrument(tracing::info_span!(
+                            target: "buzz_datastore",
+                            "SELECT",
+                            otel.kind = "client",
+                            db.system.name = "postgresql",
+                        ))
+                        .await;
+                    Ok::<_, std::convert::Infallible>(axum::response::Response::new(
+                        axum::body::Body::empty(),
+                    ))
+                },
+            ));
+
+        service
+            .oneshot(
+                axum::http::Request::get("/")
+                    .body(axum::body::Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+
+        provider.force_flush().unwrap();
+        let spans = exporter.get_finished_spans().unwrap();
+        let http = spans
+            .iter()
+            .find(|span| span.name == "http.request")
+            .unwrap();
+        let datastore = spans.iter().find(|span| span.name == "SELECT").unwrap();
+
+        assert_eq!(
+            datastore.span_context.trace_id(),
+            http.span_context.trace_id()
+        );
+        assert_eq!(datastore.parent_span_id, http.span_context.span_id());
+    }
+
     async fn handler_receives_message_with_limit(limit: usize, size: usize) -> bool {
         let (received_tx, mut received_rx) = mpsc::unbounded_channel();
         let app = Router::new().route(

crates/buzz-relay/src/telemetry.rs

@@ -25,6 +25,16 @@
 
 use opentelemetry_otlp::ExporterBuildError;
 use opentelemetry_sdk::{resource::EnvResourceDetector, trace::SdkTracerProvider, Resource};
+use tracing_subscriber::EnvFilter;
+
+/// Build the filter for spans exported through OpenTelemetry.
+///
+/// This is intentionally independent from `RUST_LOG`: changing stdout log
+/// verbosity must not remove parent spans from exported traces. Set
+/// `BUZZ_OTEL_FILTER` to override the default targets.
+pub fn otel_env_filter(configured: Option<&str>) -> EnvFilter {
+    EnvFilter::new(configured.unwrap_or("buzz_relay=info,buzz_datastore=info"))
+}
 
 /// Build the OTEL [`Resource`] used by the trace provider.
 ///