mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
feat(relay): add OpenTelemetry tracing, keep Prometheus metrics (#1398)
Signed-off-by: Will Pfleger <pfleger.will@gmail.com> Co-authored-by: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 <dcfd242e557282d7a1e2cf2e6877522682f1e5c6156dc92ca7d90eaedd3b0f95@sprout-oss.stage.blox.sqprod.co>
This commit is contained in:
Will Pfleger
120
Cargo.lock
generated
120
Cargo.lock
generated
@@ -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",
|
||||
@@ -1075,6 +1078,7 @@ dependencies = [
|
||||
"tower",
|
||||
"tower-http",
|
||||
"tracing",
|
||||
"tracing-opentelemetry",
|
||||
"tracing-subscriber",
|
||||
"url",
|
||||
"uuid",
|
||||
@@ -4252,9 +4256,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",
|
||||
@@ -5427,6 +5431,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"
|
||||
@@ -5436,7 +5454,7 @@ dependencies = [
|
||||
"async-trait",
|
||||
"bytes",
|
||||
"http",
|
||||
"opentelemetry",
|
||||
"opentelemetry 0.31.0",
|
||||
"reqwest 0.12.28",
|
||||
]
|
||||
|
||||
@@ -5447,15 +5465,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"
|
||||
@@ -5464,8 +5499,8 @@ checksum = "a7175df06de5eaee9909d4805a3d07e28bb752c34cab57fa9cff549da596b30f"
|
||||
dependencies = [
|
||||
"base64",
|
||||
"const-hex",
|
||||
"opentelemetry",
|
||||
"opentelemetry_sdk",
|
||||
"opentelemetry 0.31.0",
|
||||
"opentelemetry_sdk 0.31.0",
|
||||
"prost",
|
||||
"serde",
|
||||
"serde_json",
|
||||
@@ -5473,6 +5508,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"
|
||||
@@ -5482,12 +5530,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"
|
||||
@@ -6302,9 +6368,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",
|
||||
]
|
||||
@@ -7397,7 +7463,7 @@ dependencies = [
|
||||
"futures-util",
|
||||
"libc",
|
||||
"openai-frontend",
|
||||
"opentelemetry-proto",
|
||||
"opentelemetry-proto 0.31.0",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"sha2 0.10.9",
|
||||
@@ -8259,6 +8325,7 @@ dependencies = [
|
||||
"socket2",
|
||||
"sync_wrapper",
|
||||
"tokio",
|
||||
"tokio-rustls",
|
||||
"tokio-stream",
|
||||
"tower",
|
||||
"tower-layer",
|
||||
@@ -8277,6 +8344,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"
|
||||
@@ -8394,6 +8472,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"
|
||||
|
||||
@@ -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"
|
||||
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -131,6 +131,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.
|
||||
@@ -203,12 +216,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.
|
||||
@@ -221,6 +240,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.
|
||||
@@ -2563,7 +2595,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 {
|
||||
|
||||
@@ -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 }
|
||||
|
||||
@@ -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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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<ConnectionState>, 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;
|
||||
|
||||
@@ -38,6 +38,7 @@ pub fn extract_auth_tag_json(event: &nostr::Event) -> Option<String> {
|
||||
/// 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<ConnectionState>, state: Arc<AppState>) {
|
||||
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<ConnectionState>, 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.
|
||||
|
||||
@@ -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<AppState>, 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<ConnectionState>, state: Arc<AppState>) {
|
||||
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);
|
||||
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
@@ -36,11 +37,33 @@ async fn main() -> anyhow::Result<()> {
|
||||
.expect("failed to install rustls crypto provider");
|
||||
|
||||
// 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| {
|
||||
@@ -710,6 +733,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::<u64>().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
|
||||
@@ -719,6 +770,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(())
|
||||
}
|
||||
|
||||
|
||||
@@ -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,
|
||||
|
||||
214
crates/buzz-relay/src/telemetry.rs
Normal file
214
crates/buzz-relay/src/telemetry.rs
Normal file
@@ -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:<exe>` 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<SpanExporter, ExporterBuildError>` 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<opentelemetry_otlp::SpanExporter, ExporterBuildError>,
|
||||
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<String> {
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user