From 560f2fc93b5b710d2c4159867b9c97c1dd34a970 Mon Sep 17 00:00:00 2001 From: Amir Hatamzad Date: Fri, 17 Jul 2026 12:16:53 -0700 Subject: [PATCH 1/3] feat(metrics-endpoint): count /metrics scrape failures by outcome The shared metrics-endpoint handler can fail a scrape five distinct ways (encoder queue full -> 503; encoder gone, encode error, encoder panic, dropped reply -> 500). Each was only logged. Add a carbide_metrics_scrape_failures_total counter, labeled by a bounded ScrapeOutcome enum, emitted at each failure branch alongside the existing logs. This meta-observability surfaces a busy or erroring encoder on the next successful scrape, where aggregatable/alertable counters beat logs alone. Closes #3546 --- Cargo.lock | 1 + crates/metrics-endpoint/Cargo.toml | 4 + crates/metrics-endpoint/src/lib.rs | 127 +++++++++++++++++++++++++++-- docs/observability/core_metrics.md | 1 + 4 files changed, 128 insertions(+), 5 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 82c70ff738..db9e54a5f4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7025,6 +7025,7 @@ name = "metrics-endpoint" version = "0.1.0" dependencies = [ "bytes", + "carbide-instrument", "carbide-metrics-utils", "eyre", "http-body-util", diff --git a/crates/metrics-endpoint/Cargo.toml b/crates/metrics-endpoint/Cargo.toml index 36f3404847..a322ed844e 100644 --- a/crates/metrics-endpoint/Cargo.toml +++ b/crates/metrics-endpoint/Cargo.toml @@ -24,6 +24,7 @@ authors.workspace = true [dependencies] # [local-dependencies] carbide-metrics-utils = { path = "../metrics-utils" } +carbide-instrument = { path = "../instrument" } # [end local-dependencies] hyper = { workspace = true } @@ -50,5 +51,8 @@ opentelemetry_sdk = { workspace = true, features = [ "spec_unstable_metrics_views", ] } +[dev-dependencies] +carbide-instrument = { path = "../instrument", features = ["test-support"] } + [lints] workspace = true diff --git a/crates/metrics-endpoint/src/lib.rs b/crates/metrics-endpoint/src/lib.rs index faed8b7113..42ed2f5671 100644 --- a/crates/metrics-endpoint/src/lib.rs +++ b/crates/metrics-endpoint/src/lib.rs @@ -20,6 +20,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{SyncSender, TrySendError, sync_channel}; use bytes::Bytes; +use carbide_instrument::{Event, LabelValue, emit}; use carbide_metrics_utils::OtelView; use http_body_util::Full; use hyper::header::{CONTENT_LENGTH, CONTENT_TYPE}; @@ -113,6 +114,40 @@ enum EncodeError { Panicked, } +/// Why a `/metrics` scrape failed to return an exposition, as a bounded low-cardinality +/// label for [`MetricsScrapeFailed`]. Each variant maps to one failure branch of the +/// handler (and its HTTP status): `EncoderBusy` → 503, all others → 500. +#[derive(Debug, Clone, Copy, PartialEq, Eq, LabelValue)] +enum ScrapeOutcome { + /// The encoder queue was full; the scrape was shed (503). + EncoderBusy, + /// The encoder thread was gone, so nothing could be enqueued (500). + EncoderGone, + /// The Prometheus text encoder returned an error (500). + EncodeFailed, + /// A collector/observable callback panicked during gather/encode (500). + EncoderPanicked, + /// The encoder dropped the reply without answering, e.g. during shutdown (500). + ReplyDropped, +} + +/// Counts `/metrics` scrape failures by outcome, alongside the existing per-branch logs. +/// Meta-observability: it surfaces a busy or erroring encoder on the next successful +/// scrape, where aggregatable/alertable counters beat logs alone. +#[derive(Event)] +#[event( + event_name = "metrics_scrape_failed", + metric_name = "carbide_metrics_scrape_failures_total", + component = "metrics-endpoint", + metric = counter, + log = off, + describe = "Number of /metrics scrape failures, by outcome." +)] +struct MetricsScrapeFailed { + #[label] + outcome: ScrapeOutcome, +} + /// The reply channel a scrape hands to the encoder thread. The thread sends the encoded /// exposition back through it, or an [`EncodeError`] describing why it could not. type EncodeReply = oneshot::Sender, EncodeError>>; @@ -448,6 +483,9 @@ async fn handle_metrics_request( // The single encoder is already busy with a backlog; shed this // scrape rather than pile on more work. tracing::warn!("metrics encoder busy; shedding scrape with 503"); + emit(MetricsScrapeFailed { + outcome: ScrapeOutcome::EncoderBusy, + }); return Ok(Response::builder() .status(503) .body(Full::new(Bytes::from("metrics encoder busy"))) @@ -455,6 +493,9 @@ async fn handle_metrics_request( } Err(TrySendError::Disconnected(_)) => { tracing::error!("metrics encoder thread is gone; cannot encode metrics"); + emit(MetricsScrapeFailed { + outcome: ScrapeOutcome::EncoderGone, + }); return Ok(Response::builder() .status(500) .body(Full::new(Bytes::from("Failed to encode metrics"))) @@ -471,6 +512,9 @@ async fn handle_metrics_request( .unwrap(), Ok(Err(EncodeError::Encode(err))) => { tracing::error!(error = %err, "failed to encode metrics"); + emit(MetricsScrapeFailed { + outcome: ScrapeOutcome::EncodeFailed, + }); Response::builder() .status(500) .body(Full::new(Bytes::from("Failed to encode metrics"))) @@ -478,6 +522,9 @@ async fn handle_metrics_request( } Ok(Err(EncodeError::Panicked)) => { tracing::error!("metrics encoder caught a panic while encoding"); + emit(MetricsScrapeFailed { + outcome: ScrapeOutcome::EncoderPanicked, + }); Response::builder() .status(500) .body(Full::new(Bytes::from("Failed to encode metrics"))) @@ -488,6 +535,9 @@ async fn handle_metrics_request( // to stop (shutdown) or otherwise went away. Distinct from a busy or // already-gone encoder above. tracing::error!("metrics encoder dropped the reply without responding"); + emit(MetricsScrapeFailed { + outcome: ScrapeOutcome::ReplyDropped, + }); Response::builder() .status(500) .body(Full::new(Bytes::from("Failed to encode metrics"))) @@ -530,12 +580,35 @@ async fn handle_metrics_request( mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; + use carbide_instrument::testing::MetricsCapture; use opentelemetry::KeyValue; use opentelemetry_sdk::metrics; use prometheus::{Encoder, TextEncoder}; use super::*; + /// The `EncodeFailed` outcome (a `TextEncoder` error) cannot be triggered + /// deterministically through the handler — the text encoder does not error on a valid + /// registry, and there is no seam to inject a failure — so its emit wiring is covered + /// here directly. The other four outcomes are driven through `handle_metrics_request` + /// in the tests below. + #[test] + fn encode_failed_outcome_emits_counter() { + let metrics = MetricsCapture::start(); + + emit(MetricsScrapeFailed { + outcome: ScrapeOutcome::EncodeFailed, + }); + + assert_eq!( + metrics.counter_delta( + "carbide_metrics_scrape_failures_total", + &[("outcome", "encode_failed")] + ), + 1.0, + ); + } + /// This test mostly mimics the test setup above and checks whether /// the prometheus opentelemetry stack will only report the most recent /// values for gauges and not cached values that are not important anymore @@ -1019,6 +1092,8 @@ mod tests { run_metrics_endpoint_with_listener(&config, server_token, listener).await }); + let metrics = MetricsCapture::start(); + // First scrape trips the collector panic. The encoder thread catches the unwind // and answers 500 instead of dying. let (headers, _body) = scrape_metrics(addr).await; @@ -1026,6 +1101,15 @@ mod tests { headers.starts_with("http/1.1 500"), "panicking scrape should be 500, got headers:\n{headers}" ); + // The panic branch records the `encoder_panicked` outcome before replying. + assert_eq!( + metrics.counter_delta( + "carbide_metrics_scrape_failures_total", + &[("outcome", "encoder_panicked")] + ), + 1.0, + "panicking scrape should increment the encoder_panicked counter once" + ); // The thread survived, so a subsequent scrape on the same server succeeds. let (headers, body) = scrape_metrics(addr).await; @@ -1045,19 +1129,23 @@ mod tests { .expect("metrics server exited cleanly"); } - /// The `/metrics` handler maps a full queue to `503` and a disconnected encoder to - /// `500`, exercised directly against `handle_metrics_request` with a channel we control - /// — no live server or encoder thread. + /// Drive `handle_metrics_request` through its channel-controllable failure branches with + /// a queue we manipulate — no live server or encoder thread — and assert each maps to + /// the expected status *and* increments `carbide_metrics_scrape_failures_total` once + /// under the matching `outcome` label. This guards the handler's emit wiring, not just + /// that `emit` works. (`EncodeFailed` is covered separately; see + /// `encode_failed_outcome_emits_counter`.) #[tokio::test] - async fn test_handle_metrics_request_queue_backpressure() { + async fn test_handle_metrics_request_failure_outcomes() { // The encoder sender under test, plus any receiver that must stay alive for the - // call (`None` means it was dropped, disconnecting the channel). + // call (`None` means it was dropped or handed to a drainer thread). type EncoderChannel = (SyncSender, Option>); struct Case { name: &'static str, setup: fn() -> EncoderChannel, expected_status: u16, + outcome: &'static str, } let cases = [ @@ -1070,6 +1158,7 @@ mod tests { (tx, Some(rx)) }, expected_status: 503, + outcome: "encoder_busy", }, Case { name: "encoder gone -> 500", @@ -1079,9 +1168,28 @@ mod tests { (tx, None) }, expected_status: 500, + outcome: "encoder_gone", + }, + Case { + name: "reply dropped -> 500", + setup: || { + let (tx, rx) = sync_channel::(1); + // Accept the enqueued Work and drop its reply channel without answering, + // so the handler's `reply_rx.await` resolves to `Err` (reply dropped). + std::thread::spawn(move || { + if let Ok(Item::Work(reply_tx)) = rx.recv() { + drop(reply_tx); + } + }); + (tx, None) + }, + expected_status: 500, + outcome: "reply_dropped", }, ]; + let metrics = MetricsCapture::start(); + for case in cases { let (encoder_tx, _rx_guard) = (case.setup)(); let state = Arc::new(MetricsHandlerState { @@ -1104,6 +1212,15 @@ mod tests { "case: {}", case.name ); + assert_eq!( + metrics.counter_delta( + "carbide_metrics_scrape_failures_total", + &[("outcome", case.outcome)] + ), + 1.0, + "case: {} should increment its outcome counter exactly once", + case.name + ); // `_rx_guard` (kept for the full-queue case) stays alive until here so the // channel reads as full rather than disconnected. } diff --git a/docs/observability/core_metrics.md b/docs/observability/core_metrics.md index 2b1cbb361f..f9eab5c96d 100644 --- a/docs/observability/core_metrics.md +++ b/docs/observability/core_metrics.md @@ -135,6 +135,7 @@ This file contains a list of metrics exported by NVIDIA Infra Controller (NICo). carbide_measured_boot_machines_totalgaugeNumber of machines reporting measurements. carbide_measured_boot_profiles_totalgaugeNumber of measured boot profiles. carbide_measured_boot_verification_failures_totalcounterNumber of measured boot verification failures, across quote verification and attestation handling, by cause +carbide_metrics_scrape_failures_totalcounterNumber of /metrics scrape failures, by outcome. carbide_mqtt_dispatch_dropped_totalcounterNumber of received MQTT messages dropped because the handler concurrency semaphore could not be acquired carbide_mqtt_reconnects_totalcounterNumber of times an MQTT client re-established its broker connection after the initial connect carbide_network_segments_enqueuer_iteration_latency_millisecondshistogramThe overall time it took to enqueue state handling tasks for all carbide_network_segments in the system From 2f080dfa63c1112669519f202fb345af254877e9 Mon Sep 17 00:00:00 2001 From: Amir Hatamzad Date: Tue, 21 Jul 2026 14:07:48 -0700 Subject: [PATCH 2/3] refactor(metrics-endpoint): use dynamic log/message for MetricsScrapeFailed --- crates/metrics-endpoint/src/lib.rs | 31 +++++++++++++++++++++++------- 1 file changed, 24 insertions(+), 7 deletions(-) diff --git a/crates/metrics-endpoint/src/lib.rs b/crates/metrics-endpoint/src/lib.rs index 42ed2f5671..7a0ff0d270 100644 --- a/crates/metrics-endpoint/src/lib.rs +++ b/crates/metrics-endpoint/src/lib.rs @@ -20,7 +20,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc::{SyncSender, TrySendError, sync_channel}; use bytes::Bytes; -use carbide_instrument::{Event, LabelValue, emit}; +use carbide_instrument::{DynamicLog, DynamicMessage, Event, LabelValue, LogAt, emit}; use carbide_metrics_utils::OtelView; use http_body_util::Full; use hyper::header::{CONTENT_LENGTH, CONTENT_TYPE}; @@ -140,7 +140,8 @@ enum ScrapeOutcome { metric_name = "carbide_metrics_scrape_failures_total", component = "metrics-endpoint", metric = counter, - log = off, + log = dynamic, + message = dynamic, describe = "Number of /metrics scrape failures, by outcome." )] struct MetricsScrapeFailed { @@ -148,6 +149,27 @@ struct MetricsScrapeFailed { outcome: ScrapeOutcome, } +impl DynamicLog for MetricsScrapeFailed { + fn log_at(&self) -> LogAt { + match self.outcome { + ScrapeOutcome::EncoderBusy => LogAt::Level(tracing::Level::WARN), + _ => LogAt::Level(tracing::Level::ERROR), + } + } +} + +impl DynamicMessage for MetricsScrapeFailed { + fn message(&self) -> &'static str { + match self.outcome { + ScrapeOutcome::EncoderBusy => "metrics encoder busy; shedding scrape with 503", + ScrapeOutcome::EncoderGone => "metrics encoder thread is gone; cannot encode metrics", + ScrapeOutcome::EncodeFailed => "failed to encode metrics", + ScrapeOutcome::EncoderPanicked => "metrics encoder caught a panic while encoding", + ScrapeOutcome::ReplyDropped => "metrics encoder dropped the reply without responding", + } + } +} + /// The reply channel a scrape hands to the encoder thread. The thread sends the encoded /// exposition back through it, or an [`EncodeError`] describing why it could not. type EncodeReply = oneshot::Sender, EncodeError>>; @@ -482,7 +504,6 @@ async fn handle_metrics_request( Err(TrySendError::Full(_)) => { // The single encoder is already busy with a backlog; shed this // scrape rather than pile on more work. - tracing::warn!("metrics encoder busy; shedding scrape with 503"); emit(MetricsScrapeFailed { outcome: ScrapeOutcome::EncoderBusy, }); @@ -492,7 +513,6 @@ async fn handle_metrics_request( .unwrap()); } Err(TrySendError::Disconnected(_)) => { - tracing::error!("metrics encoder thread is gone; cannot encode metrics"); emit(MetricsScrapeFailed { outcome: ScrapeOutcome::EncoderGone, }); @@ -511,7 +531,6 @@ async fn handle_metrics_request( .body(Full::new(Bytes::from(buffer))) .unwrap(), Ok(Err(EncodeError::Encode(err))) => { - tracing::error!(error = %err, "failed to encode metrics"); emit(MetricsScrapeFailed { outcome: ScrapeOutcome::EncodeFailed, }); @@ -521,7 +540,6 @@ async fn handle_metrics_request( .unwrap() } Ok(Err(EncodeError::Panicked)) => { - tracing::error!("metrics encoder caught a panic while encoding"); emit(MetricsScrapeFailed { outcome: ScrapeOutcome::EncoderPanicked, }); @@ -534,7 +552,6 @@ async fn handle_metrics_request( // The encoder thread dropped the reply without answering — it was told // to stop (shutdown) or otherwise went away. Distinct from a busy or // already-gone encoder above. - tracing::error!("metrics encoder dropped the reply without responding"); emit(MetricsScrapeFailed { outcome: ScrapeOutcome::ReplyDropped, }); From 22f2a84635ed11e1e7b6567e9204cef196a4f805 Mon Sep 17 00:00:00 2001 From: Amir Hatamzad Date: Tue, 21 Jul 2026 14:32:22 -0700 Subject: [PATCH 3/3] refactor(metrics-endpoint): use dynamic log/message for MetricsScrapeFailed --- crates/metrics-endpoint/src/lib.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/crates/metrics-endpoint/src/lib.rs b/crates/metrics-endpoint/src/lib.rs index 7a0ff0d270..8565a61a26 100644 --- a/crates/metrics-endpoint/src/lib.rs +++ b/crates/metrics-endpoint/src/lib.rs @@ -161,11 +161,11 @@ impl DynamicLog for MetricsScrapeFailed { impl DynamicMessage for MetricsScrapeFailed { fn message(&self) -> &'static str { match self.outcome { - ScrapeOutcome::EncoderBusy => "metrics encoder busy; shedding scrape with 503", - ScrapeOutcome::EncoderGone => "metrics encoder thread is gone; cannot encode metrics", - ScrapeOutcome::EncodeFailed => "failed to encode metrics", + ScrapeOutcome::EncoderBusy => "metrics encoder busy; shedding scrape with 503", + ScrapeOutcome::EncoderGone => "metrics encoder thread is gone; cannot encode metrics", + ScrapeOutcome::EncodeFailed => "failed to encode metrics", ScrapeOutcome::EncoderPanicked => "metrics encoder caught a panic while encoding", - ScrapeOutcome::ReplyDropped => "metrics encoder dropped the reply without responding", + ScrapeOutcome::ReplyDropped => "metrics encoder dropped the reply without responding", } } } @@ -530,7 +530,7 @@ async fn handle_metrics_request( .header(CONTENT_LENGTH, buffer.len()) .body(Full::new(Bytes::from(buffer))) .unwrap(), - Ok(Err(EncodeError::Encode(err))) => { + Ok(Err(EncodeError::Encode(_err))) => { emit(MetricsScrapeFailed { outcome: ScrapeOutcome::EncodeFailed, });