From e94b9aeda0b2272d36e3744e78680be69295b8b5 Mon Sep 17 00:00:00 2001 From: Dave Grochowski Date: Mon, 27 Jul 2026 23:32:42 -0400 Subject: [PATCH] feat(tracing): add datastore tracing plumbing (#2760) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Configure a dedicated datastore tracing target on the OTLP layer while preserving explicit logging filters and avoiding span overhead when OTLP is disabled. This is in preparation for adding trace spans for datastores used in Buzz ## Update — 2026-07-27 - Export HTTP requests as `INFO` server spans under `buzz_relay`, preserving request parentage for datastore spans. - Configure OTEL span filtering independently with `BUZZ_OTEL_FILTER`, so `RUST_LOG` changes cannot break trace topology. - Verify exported HTTP and datastore spans share a trace ID and have the expected parent/child relationship. Co-authored-by: Amp --- .env.example | 5 +- crates/buzz-relay/Cargo.toml | 1 + crates/buzz-relay/src/main.rs | 64 ++++++++++++++++++++++-- crates/buzz-relay/src/router.rs | 79 ++++++++++++++++++++++++++++-- crates/buzz-relay/src/telemetry.rs | 10 ++++ 5 files changed, 152 insertions(+), 7 deletions(-) diff --git a/.env.example b/.env.example index db5a7ea25c..024175bff0 100644 --- a/.env.example +++ b/.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 diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index 10a83f0324..01f78a2d49 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/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"] } diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 533428a4e8..c1a127d8eb 100644 --- a/crates/buzz-relay/src/main.rs +++ b/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, period: std::time::Duration, diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index 2af036079f..400ed1dfe3 100644 --- a/crates/buzz-relay/src/router.rs +++ b/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) -> 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) -> tracing::Span> { + TraceLayer::new_for_http().make_span_with(make_http_span as fn(&Request) -> tracing::Span) +} + +fn make_http_span(request: &Request) -> 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| 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( diff --git a/crates/buzz-relay/src/telemetry.rs b/crates/buzz-relay/src/telemetry.rs index 8dd305cd42..11c6d03512 100644 --- a/crates/buzz-relay/src/telemetry.rs +++ b/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. ///