diff --git a/Cargo.lock b/Cargo.lock index 09108807c6..f7a2552a7c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1057,6 +1057,9 @@ dependencies = [ "metrics-exporter-prometheus", "moka", "nostr", + "opentelemetry 0.32.0", + "opentelemetry-otlp 0.32.0", + "opentelemetry_sdk 0.32.1", "rand 0.10.1", "redis", "reqwest 0.13.3", @@ -1074,6 +1077,7 @@ dependencies = [ "tower", "tower-http", "tracing", + "tracing-opentelemetry", "tracing-subscriber", "url", "uuid", @@ -4250,9 +4254,9 @@ dependencies = [ "model-resolver", "nostr-sdk", "openai-frontend", - "opentelemetry", - "opentelemetry-otlp", - "opentelemetry_sdk", + "opentelemetry 0.31.0", + "opentelemetry-otlp 0.31.1", + "opentelemetry_sdk 0.31.0", "prost", "rand 0.10.1", "regex-lite", @@ -5425,6 +5429,20 @@ dependencies = [ "tracing", ] +[[package]] +name = "opentelemetry" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682" +dependencies = [ + "futures-core", + "futures-sink", + "js-sys", + "pin-project-lite", + "thiserror 2.0.18", + "tracing", +] + [[package]] name = "opentelemetry-http" version = "0.31.0" @@ -5434,7 +5452,7 @@ dependencies = [ "async-trait", "bytes", "http", - "opentelemetry", + "opentelemetry 0.31.0", "reqwest 0.12.28", ] @@ -5445,15 +5463,32 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1f69cd6acbb9af919df949cd1ec9e5e7fdc2ef15d234b6b795aaa525cc02f71f" dependencies = [ "http", - "opentelemetry", + "opentelemetry 0.31.0", "opentelemetry-http", - "opentelemetry-proto", - "opentelemetry_sdk", + "opentelemetry-proto 0.31.0", + "opentelemetry_sdk 0.31.0", "prost", "reqwest 0.12.28", "thiserror 2.0.18", ] +[[package]] +name = "opentelemetry-otlp" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35" +dependencies = [ + "http", + "opentelemetry 0.32.0", + "opentelemetry-proto 0.32.0", + "opentelemetry_sdk 0.32.1", + "prost", + "thiserror 2.0.18", + "tokio", + "tonic", + "tonic-types", +] + [[package]] name = "opentelemetry-proto" version = "0.31.0" @@ -5462,8 +5497,8 @@ checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f" dependencies = [ "base64", "const-hex", - "opentelemetry", - "opentelemetry_sdk", + "opentelemetry 0.31.0", + "opentelemetry_sdk 0.31.0", "prost", "serde", "serde_json", @@ -5471,6 +5506,19 @@ dependencies = [ "tonic-prost", ] +[[package]] +name = "opentelemetry-proto" +version = "0.32.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638" +dependencies = [ + "opentelemetry 0.32.0", + "opentelemetry_sdk 0.32.1", + "prost", + "tonic", + "tonic-prost", +] + [[package]] name = "opentelemetry_sdk" version = "0.31.0" @@ -5480,12 +5528,30 @@ dependencies = [ "futures-channel", "futures-executor", "futures-util", - "opentelemetry", + "opentelemetry 0.31.0", "percent-encoding", "rand 0.9.4", "thiserror 2.0.18", ] +[[package]] +name = "opentelemetry_sdk" +version = "0.32.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9" +dependencies = [ + "futures-channel", + "futures-executor", + "futures-util", + "opentelemetry 0.32.0", + "percent-encoding", + "portable-atomic", + "rand 0.9.4", + "thiserror 2.0.18", + "tokio", + "tokio-stream", +] + [[package]] name = "option-ext" version = "0.2.0" @@ -6300,9 +6366,9 @@ dependencies = [ [[package]] name = "rapidhash" -version = "4.4.1" +version = "4.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5e48930979c155e2f33aa36ab3119b5ee81332beb6482199a8ecd6029b80b59" +checksum = "32b266a82f4aa99bb5c25e28d11cc44ace63d91adbcbcee4d323e2ae3d49ef37" dependencies = [ "rustversion", ] @@ -7392,7 +7458,7 @@ dependencies = [ "futures-util", "libc", "openai-frontend", - "opentelemetry-proto", + "opentelemetry-proto 0.31.0", "serde", "serde_json", "sha2 0.10.9", @@ -8254,6 +8320,7 @@ dependencies = [ "socket2", "sync_wrapper", "tokio", + "tokio-rustls", "tokio-stream", "tower", "tower-layer", @@ -8272,6 +8339,17 @@ dependencies = [ "tonic", ] +[[package]] +name = "tonic-types" +version = "0.14.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73ab1b02061f83d519bba3caa167f88f261ef05720ab8ebc954ade70de3348e8" +dependencies = [ + "prost", + "prost-types", + "tonic", +] + [[package]] name = "tower" version = "0.5.3" @@ -8389,6 +8467,22 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-opentelemetry" +version = "0.33.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26" +dependencies = [ + "js-sys", + "opentelemetry 0.32.0", + "smallvec", + "tracing", + "tracing-core", + "tracing-log", + "tracing-subscriber", + "web-time", +] + [[package]] name = "tracing-serde" version = "0.2.0" diff --git a/Cargo.toml b/Cargo.toml index 03cba4ee7c..bdfc8d367c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -65,8 +65,12 @@ serde_yaml = "0.9" evalexpr = "11" cron = "0.16" # Observability -tracing = "0.1" -tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } +tracing-opentelemetry = { version = "0.33" } +opentelemetry = { version = "0.32", features = ["trace"] } +opentelemetry_sdk = { version = "0.32", features = ["trace", "rt-tokio"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = ["trace", "grpc-tonic", "tls-ring"] } metrics = "0.24" metrics-exporter-prometheus = "0.18" diff --git a/crates/buzz-db/src/api_token.rs b/crates/buzz-db/src/api_token.rs index f105d681a6..50821743d2 100644 --- a/crates/buzz-db/src/api_token.rs +++ b/crates/buzz-db/src/api_token.rs @@ -350,7 +350,7 @@ mod tests { let pool = PgPool::connect(TEST_DB_URL) .await .expect("connect to test DB"); - Db { pool } + Db::from_pool(pool) } async fn make_community(pool: &PgPool) -> Uuid { diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index f3f0bc25d0..01e29c20ca 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -129,6 +129,19 @@ pub async fn insert_mentions( #[derive(Clone, Debug)] pub struct Db { pub(crate) pool: PgPool, + /// Maximum connections configured for this pool (from [`DbConfig::max_connections`]). + pub(crate) max_connections: u32, +} + +/// Snapshot of Postgres connection pool utilisation. +#[derive(Debug, Clone, Copy)] +pub struct DbPoolStats { + /// Total connections currently in the pool (idle + active). + pub size: u32, + /// Connections available for immediate reuse. + pub idle: u32, + /// Pool ceiling — the `max_connections` value set at construction. + pub max: u32, } /// Configuration for the Postgres connection pool. @@ -201,12 +214,18 @@ impl Db { .idle_timeout(Duration::from_secs(config.idle_timeout_secs)) .connect(&config.database_url) .await?; - Ok(Self { pool }) + Ok(Self { + pool, + max_connections: config.max_connections, + }) } /// Creates a `Db` from an existing `PgPool` (useful in tests). pub fn from_pool(pool: PgPool) -> Self { - Self { pool } + Self { + max_connections: pool.options().get_max_connections(), + pool, + } } /// Run pending database migrations. @@ -219,6 +238,19 @@ impl Db { sqlx::query("SELECT 1").execute(&self.pool).await.is_ok() } + /// Returns pool utilisation stats for metrics emission. + /// + /// `size` — total connections (idle + active) + /// `idle` — connections available for immediate reuse + /// `max` — pool ceiling set at construction + pub fn pool_stats(&self) -> DbPoolStats { + DbPoolStats { + size: self.pool.size(), + idle: self.pool.num_idle() as u32, + max: self.max_connections, + } + } + /// Begin a database transaction for atomic multi-statement operations. /// /// Returns a `'static` transaction because `PgPool` is `Arc`-backed internally. @@ -2517,7 +2549,7 @@ mod tests { let pool = PgPool::connect(TEST_DB_URL) .await .expect("connect to test DB"); - Db { pool } + Db::from_pool(pool) } async fn make_community(pool: &PgPool) -> Uuid { diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index c9b80d53fc..d4cc3664ac 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -36,6 +36,10 @@ serde = { workspace = true } serde_json = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } +tracing-opentelemetry = { workspace = true } +opentelemetry = { workspace = true } +opentelemetry_sdk = { workspace = true } +opentelemetry-otlp = { workspace = true } thiserror = { workspace = true } anyhow = { workspace = true } uuid = { workspace = true } diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 58a183acb5..a3e72414d9 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -10,6 +10,7 @@ use axum::extract::ws::{Message as WsMessage, WebSocket}; use futures_util::{SinkExt, StreamExt}; use tokio::sync::{mpsc, Mutex, RwLock}; use tokio_util::sync::CancellationToken; +use tracing::Instrument as _; use tracing::{debug, info, trace, warn}; use uuid::Uuid; @@ -423,7 +424,11 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar match msg { ClientMessage::Auth(event) => { - handlers::auth::handle_auth(event, Arc::clone(&conn), Arc::clone(&state)).await; + // Auth is synchronous in the WS loop — no span context is lost. + let span = tracing::info_span!("ws.auth", conn_id = %conn.conn_id); + handlers::auth::handle_auth(event, Arc::clone(&conn), Arc::clone(&state)) + .instrument(span) + .await; } ClientMessage::Event(event) => { let conn = Arc::clone(&conn); @@ -437,10 +442,21 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar return; } }; - tokio::spawn(async move { - handlers::event::handle_event(event, conn, state).await; - drop(permit); - }); + // Capture the parent span BEFORE the spawn so it is propagated into + // the spawned future. A bare `tokio::spawn` drops tracing context. + let span = tracing::info_span!( + "ws.event", + conn_id = %conn.conn_id, + event_id = tracing::field::Empty, + kind = tracing::field::Empty, + ); + tokio::spawn( + async move { + handlers::event::handle_event(event, conn, state).await; + drop(permit); + } + .instrument(span), + ); } ClientMessage::Req { sub_id, filters } => { let conn = Arc::clone(&conn); @@ -454,10 +470,14 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar return; } }; - tokio::spawn(async move { - handlers::req::handle_req(sub_id, filters, conn, state).await; - drop(permit); - }); + let span = tracing::info_span!("ws.req", conn_id = %conn.conn_id, sub_id = %sub_id); + tokio::spawn( + async move { + handlers::req::handle_req(sub_id, filters, conn, state).await; + drop(permit); + } + .instrument(span), + ); } ClientMessage::Count { sub_id, filters } => { let conn = Arc::clone(&conn); @@ -471,10 +491,14 @@ async fn handle_text_message(text: String, conn: Arc, state: Ar return; } }; - tokio::spawn(async move { - handlers::count::handle_count(sub_id, filters, conn, state).await; - drop(permit); - }); + let span = tracing::info_span!("ws.count", conn_id = %conn.conn_id, sub_id = %sub_id); + tokio::spawn( + async move { + handlers::count::handle_count(sub_id, filters, conn, state).await; + drop(permit); + } + .instrument(span), + ); } ClientMessage::Close(sub_id) => { handlers::close::handle_close(sub_id, Arc::clone(&conn), Arc::clone(&state)).await; diff --git a/crates/buzz-relay/src/handlers/auth.rs b/crates/buzz-relay/src/handlers/auth.rs index 143dfbee9f..bc356979f5 100644 --- a/crates/buzz-relay/src/handlers/auth.rs +++ b/crates/buzz-relay/src/handlers/auth.rs @@ -38,6 +38,7 @@ pub fn extract_auth_tag_json(event: &nostr::Event) -> Option { /// the connection to authenticated state. /// /// Pure crypto verification — no API tokens, no JWT, no DB token lookups. +#[tracing::instrument(skip_all, fields(event_id, conn_id))] pub async fn handle_auth(event: nostr::Event, conn: Arc, state: Arc) { let event_id_hex = event.id.to_hex(); let (challenge, conn_id) = { @@ -65,6 +66,11 @@ pub async fn handle_auth(event: nostr::Event, conn: Arc, state: } }; + // Record the declared span fields now that we have the values. + tracing::Span::current() + .record("event_id", event_id_hex.as_str()) + .record("conn_id", conn_id.to_string().as_str()); + // Extract the NIP-OA auth tag before verification consumes the event. // The tag is integrity-protected by the event's Schnorr signature — if // tampered, NIP-42 verification will fail before we ever inspect it. diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index 61fd196dbc..f655ae0572 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -186,6 +186,7 @@ pub(crate) async fn fan_out_event_to_local_subscribers( } /// Fan out one event received from Redis pub/sub to this relay's local subscribers. +#[tracing::instrument(skip_all)] pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pubsub::ChannelEvent) { // The Redis topic carries the tenant-local routing scope explicitly: // `Channel(id)` for a per-channel event, `Global` for a channel-less one. @@ -401,11 +402,18 @@ pub(crate) async fn dispatch_persistent_event( /// /// Extracts auth from the WS connection, dispatches ephemeral events locally, /// and delegates persistent events to [`super::ingest::ingest_event`]. +#[tracing::instrument(skip_all, fields(event_id, kind))] pub async fn handle_event(event: Event, conn: Arc, state: Arc) { let start = std::time::Instant::now(); let event_id_hex = event.id.to_hex(); let kind_u32 = event_kind_u32(&event); let kind_str = bounded_kind_label(kind_u32); + + // Record the declared span fields now that we have the values. + tracing::Span::current() + .record("event_id", event_id_hex.as_str()) + .record("kind", kind_u32); + debug!(event_id = %event_id_hex, kind = kind_u32, "EVENT"); metrics::counter!("buzz_events_received_total", "kind" => kind_str.clone()).increment(1); diff --git a/crates/buzz-relay/src/lib.rs b/crates/buzz-relay/src/lib.rs index 17d56dc2f9..088177e4de 100644 --- a/crates/buzz-relay/src/lib.rs +++ b/crates/buzz-relay/src/lib.rs @@ -33,6 +33,8 @@ pub mod router; pub mod state; /// Subscription registry with (channel, kind) fan-out index. pub mod subscription; +/// OpenTelemetry tracing initialisation (tracer provider + OTLP exporter). +pub mod telemetry; /// Row-zero host binding: resolve the request community from the connection host. pub mod tenant; /// Webhook secret generation and constant-time comparison. diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index eaa4841650..6feb907f2f 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -14,6 +14,7 @@ use buzz_relay::config::Config; use buzz_relay::metrics as relay_metrics; use buzz_relay::router::{build_health_router, build_router}; use buzz_relay::state::AppState; +use buzz_relay::telemetry; use buzz_workflow::WorkflowEngine; fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { @@ -28,11 +29,33 @@ fn buzz_auto_migrate_enabled(value: Option<&str>) -> bool { #[tokio::main] async fn main() -> anyhow::Result<()> { // JSON-only structured logs — simple, machine-parseable, CAKE-compatible. + // If OTEL_EXPORTER_OTLP_ENDPOINT is set, also attach an OpenTelemetry tracing + // layer that exports spans via OTLP gRPC alongside the JSON stdout logs. + // + // Build a single shared Resource (service.name=buzz-relay by default, overridable + // via OTEL_SERVICE_NAME) for the trace provider so that Datadog can identify + // spans under the correct service identity. + let resource = telemetry::service_resource(); + let tracer_init = telemetry::try_init_tracer(resource.clone()); + let otel_layer = match &tracer_init { + telemetry::TracerInit::Enabled(p) => { + use opentelemetry::trace::TracerProvider as _; + Some(tracing_opentelemetry::layer().with_tracer(p.tracer("buzz-relay"))) + } + _ => None, + }; + tracing_subscriber::registry() .with(fmt::layer().json().flatten_event(true)) .with(EnvFilter::from_default_env().add_directive("buzz_relay=info".parse()?)) + .with(otel_layer) .init(); + // Log any exporter-build failure now that the subscriber is installed. + if let telemetry::TracerInit::ExporterBuildFailed(ref e) = tracer_init { + warn!(error = %e, "Failed to build OTLP trace exporter; distributed tracing disabled"); + } + info!("Starting buzz-relay"); let config = Config::from_env().map_err(|e| { @@ -702,6 +725,34 @@ async fn main() -> anyhow::Result<()> { let router = build_router(Arc::clone(&state)); let health_router = build_health_router(Arc::clone(&state)); + // Pool metrics: periodic background task polling DB + Redis pool stats. + { + let pool_state = Arc::clone(&state); + let interval_secs = std::env::var("BUZZ_POOL_METRICS_INTERVAL_SECS") + .ok() + .and_then(|v| v.parse::().ok()) + .unwrap_or(10) + .max(1); // tokio::time::interval panics on Duration::ZERO + tokio::spawn(async move { + let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs)); + loop { + interval.tick().await; + let db_stats = pool_state.db.pool_stats(); + let active = db_stats.size.saturating_sub(db_stats.idle); + metrics::gauge!("buzz_db_pool_size").set(db_stats.size as f64); + metrics::gauge!("buzz_db_pool_idle").set(db_stats.idle as f64); + metrics::gauge!("buzz_db_pool_active").set(active as f64); + metrics::gauge!("buzz_db_pool_max").set(db_stats.max as f64); + + let rs = pool_state.redis_pool.status(); + metrics::gauge!("buzz_redis_pool_available").set(rs.available as f64); + metrics::gauge!("buzz_redis_pool_size").set(rs.size as f64); + metrics::gauge!("buzz_redis_pool_max").set(rs.max_size as f64); + metrics::gauge!("buzz_redis_pool_waiting").set(rs.waiting as f64); + } + }); + } + serve(router, health_router, Arc::clone(&state)).await?; // Signal the audit worker to stop accepting, flush buffered entries, and @@ -711,6 +762,13 @@ async fn main() -> anyhow::Result<()> { .drain(std::time::Duration::from_secs(5)) .await; + // Flush pending OTEL spans before exit. + if let telemetry::TracerInit::Enabled(tp) = tracer_init { + if let Err(e) = tp.shutdown() { + tracing::warn!(error = %e, "OTEL tracer provider shutdown error"); + } + } + Ok(()) } diff --git a/crates/buzz-relay/src/subscription.rs b/crates/buzz-relay/src/subscription.rs index a46f593085..419cf503cb 100644 --- a/crates/buzz-relay/src/subscription.rs +++ b/crates/buzz-relay/src/subscription.rs @@ -261,6 +261,7 @@ impl SubscriptionRegistry { /// Return all (conn_id, sub_id) pairs whose filters match the given event in /// one server-resolved community. + #[tracing::instrument(skip_all)] pub fn fan_out_scoped( &self, community_id: CommunityId, diff --git a/crates/buzz-relay/src/telemetry.rs b/crates/buzz-relay/src/telemetry.rs new file mode 100644 index 0000000000..8dd305cd42 --- /dev/null +++ b/crates/buzz-relay/src/telemetry.rs @@ -0,0 +1,214 @@ +//! OpenTelemetry tracing initialisation. +//! +//! ```text +//! ┌────────────────────────────────────────────────────────────────────┐ +//! │ tracing crate (spans + events from #[instrument] and macros) │ +//! │ │ │ +//! │ ├── fmt::layer().json() → stdout (always on) │ +//! │ └── OpenTelemetryLayer (only when endpoint env var set) │ +//! │ ↓ │ +//! │ SdkTracerProvider + OTLP batch exporter │ +//! │ ↓ │ +//! │ OTLP gRPC → collector / Datadog Agent │ +//! └────────────────────────────────────────────────────────────────────┘ +//! ``` +//! +//! When `OTEL_EXPORTER_OTLP_ENDPOINT` is **unset** this module is a no-op: +//! the JSON stdout logs continue to work exactly as before and no OTLP +//! connection is attempted. +//! +//! Standard OTEL env vars honoured: +//! - `OTEL_SERVICE_NAME` (default: `buzz-relay`; read explicitly — not via SDK detector) +//! - `OTEL_RESOURCE_ATTRIBUTES` (overlaid by [`EnvResourceDetector`]) +//! - `OTEL_TRACES_SAMPLER` (default: `parentbased_always_on`) +//! - `OTEL_TRACES_SAMPLER_ARG` + +use opentelemetry_otlp::ExporterBuildError; +use opentelemetry_sdk::{resource::EnvResourceDetector, trace::SdkTracerProvider, Resource}; + +/// Build the OTEL [`Resource`] used by the trace provider. +/// +/// Strategy (priority order): +/// 1. `service.name` in `OTEL_RESOURCE_ATTRIBUTES` — overlaid last by +/// [`EnvResourceDetector`], wins over everything below. +/// 2. `OTEL_SERVICE_NAME` — read explicitly (non-empty wins over the fallback). +/// 3. Hard-coded fallback `buzz-relay`. +/// +/// Note: [`EnvResourceDetector`] only reads `OTEL_RESOURCE_ATTRIBUTES`; it +/// does **not** read `OTEL_SERVICE_NAME`. `SdkProvidedResourceDetector` does +/// read `OTEL_SERVICE_NAME` but always emits a `service.name` key (falling +/// back to `unknown_service:` when unset), which would clobber our +/// `buzz-relay` default. We therefore read `OTEL_SERVICE_NAME` explicitly +/// so the fallback is fully under our control. +/// +/// The tracer provider receives this `Resource` so Datadog can identify +/// spans under the correct `service.name`. +pub fn service_resource() -> Resource { + // Honor OTEL_SERVICE_NAME when set+non-empty; otherwise use buzz-relay. + // EnvResourceDetector overlays OTEL_RESOURCE_ATTRIBUTES last, so an + // explicit service.name there still wins over OTEL_SERVICE_NAME per spec. + let service_name = std::env::var("OTEL_SERVICE_NAME") + .ok() + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "buzz-relay".to_string()); + + Resource::builder_empty() + .with_service_name(service_name) + .with_detector(Box::new(EnvResourceDetector::new())) + .build() +} + +/// Outcome of [`try_init_tracer`], distinguishing the two disabled states so +/// the caller can log appropriately after the tracing subscriber is installed. +pub enum TracerInit { + /// OTLP endpoint configured; provider is live and ready. + Enabled(SdkTracerProvider), + /// `OTEL_EXPORTER_OTLP_ENDPOINT` was unset — no-op, no connection. + Disabled, + /// Endpoint was set but the exporter failed to build. The inner error + /// string is suitable for a `tracing::warn!` call made by the caller + /// **after** `tracing_subscriber::registry()…init()`. + ExporterBuildFailed(String), +} + +/// Build and install the OTEL tracer provider, returning the outcome so the +/// caller can act after the tracing subscriber is ready. +/// +/// Deliberately does **not** call `tracing::warn!` internally — the subscriber +/// may not be installed yet at call time, which would silently drop the event. +/// Callers are responsible for logging [`TracerInit::ExporterBuildFailed`]. +pub fn try_init_tracer(resource: Resource) -> TracerInit { + if std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_err() { + return TracerInit::Disabled; + } + + let result = opentelemetry_otlp::SpanExporter::builder() + .with_tonic() + .build(); + classify_exporter_result(result, resource) +} + +/// Map a `Result` to a [`TracerInit`] variant. +/// +/// Extracted so the `Err → ExporterBuildFailed` classification can be unit-tested +/// deterministically without relying on SDK behaviour for specific URI inputs. +fn classify_exporter_result( + result: Result, + resource: Resource, +) -> TracerInit { + match result { + Ok(exporter) => { + let provider = SdkTracerProvider::builder() + .with_resource(resource) + .with_batch_exporter(exporter) + .build(); + opentelemetry::global::set_tracer_provider(provider.clone()); + TracerInit::Enabled(provider) + } + Err(e) => TracerInit::ExporterBuildFailed(e.to_string()), + } +} + +#[cfg(test)] +mod tests { + use super::*; + use opentelemetry::KeyValue; + use std::sync::Mutex; + + // Env vars are process-global — serialize tests that mutate them to prevent + // cross-test races when the suite runs with multiple threads. + static ENV_LOCK: Mutex<()> = Mutex::new(()); + + // Helper: read service.name from the Resource's schema_url-independent KV list. + fn service_name_from(resource: &Resource) -> Option { + resource + .iter() + .find(|(k, _)| k.as_str() == "service.name") + .map(|(_, v)| v.to_string()) + } + + #[test] + fn test_service_resource_default_when_env_unset() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::remove_var("OTEL_SERVICE_NAME"); + std::env::remove_var("OTEL_RESOURCE_ATTRIBUTES"); + + let r = service_resource(); + assert_eq!( + service_name_from(&r).as_deref(), + Some("buzz-relay"), + "expected buzz-relay fallback when OTEL_SERVICE_NAME is unset" + ); + } + + #[test] + fn test_service_resource_honors_otel_service_name() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("OTEL_SERVICE_NAME", "my-custom-relay"); + std::env::remove_var("OTEL_RESOURCE_ATTRIBUTES"); + + let r = service_resource(); + std::env::remove_var("OTEL_SERVICE_NAME"); + + assert_eq!( + service_name_from(&r).as_deref(), + Some("my-custom-relay"), + "expected OTEL_SERVICE_NAME to be honoured" + ); + } + + #[test] + fn test_service_resource_empty_string_falls_back_to_default() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("OTEL_SERVICE_NAME", ""); + std::env::remove_var("OTEL_RESOURCE_ATTRIBUTES"); + + let r = service_resource(); + std::env::remove_var("OTEL_SERVICE_NAME"); + + assert_eq!( + service_name_from(&r).as_deref(), + Some("buzz-relay"), + "expected buzz-relay fallback when OTEL_SERVICE_NAME is empty" + ); + } + + #[test] + fn test_try_init_tracer_disabled_when_endpoint_unset() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::remove_var("OTEL_EXPORTER_OTLP_ENDPOINT"); + + let resource = Resource::builder_empty() + .with_attribute(KeyValue::new("service.name", "test")) + .build(); + let result = try_init_tracer(resource); + + assert!( + matches!(result, TracerInit::Disabled), + "expected Disabled when OTEL_EXPORTER_OTLP_ENDPOINT is unset" + ); + } + + /// Pin the `Err(ExporterBuildError) → TracerInit::ExporterBuildFailed` mapping + /// deterministically by synthesising an error value directly, without relying on + /// SDK/tonic behaviour for a particular URI at build time. + #[test] + fn test_classify_exporter_result_maps_err_to_exporter_build_failed() { + let resource = Resource::builder_empty() + .with_attribute(KeyValue::new("service.name", "test")) + .build(); + + // Construct a concrete ExporterBuildError variant directly. + // InvalidUri is available under the grpc-tonic feature we already depend on. + let err = ExporterBuildError::InvalidUri( + "bad-scheme://host".to_string(), + "unsupported scheme".to_string(), + ); + let result = classify_exporter_result(Err(err), resource); + + assert!( + matches!(result, TracerInit::ExporterBuildFailed(_)), + "expected ExporterBuildFailed when classify_exporter_result receives an Err" + ); + } +}