From 7e73db738855a9e615a0a82951ef775aab85c0a1 Mon Sep 17 00:00:00 2001 From: Chet Nichols III Date: Mon, 13 Jul 2026 22:34:12 -0700 Subject: [PATCH] feat: add alternate-prefix mirroring and external cancellation to the metrics endpoint The shared `metrics-endpoint` crate is how a binary serves its Prometheus `/metrics` page. Two capabilities still lived only in api-core's own copy of that server. The first mirrors a metric under a second name: during a rename like `carbide_` -> `nico_`, the endpoint can expose every `carbide_*` series *also* as `nico_*`, so a scraper or dashboard pointed at either name keeps working through the migration. The second lets the caller stop the `/metrics` listener on a `CancellationToken` for a coordinated graceful shutdown, rather than it only stopping when the process exits. Folding both into the shared crate lets every binary serve `/metrics` through one implementation -- and lets api-core delete its copy. - `MetricsEndpointConfig` gains an `additional_prefix: Option` -- a named `{ old, new }` prefix pair. When set, the exposition re-emits each family whose name starts with `old` under `new`, so a scrape sees both the original and the aliased series. `None` -- every current caller -- is byte-for-byte the previous output. - `run_metrics_endpoint_with_cancellation` binds the address and serves until a caller-owned `CancellationToken` fires; `run_metrics_endpoint` now delegates to it with a fresh, never-cancelled token, so its behavior is unchanged. - The five existing callers (bmc-proxy, dhcp-server, dhcp, dns, dsx-exchange-consumer) pass `additional_prefix: None`, keeping their `/metrics` identical. Tests cover all three paths: `None` matches a raw text-encode of the registry, `Some` puts each matching family under both prefixes and leaves non-matching families alone, and the cancellation entry point returns once its token fires. This supports https://github.com/NVIDIA/infra-controller/issues/3180 --- crates/bmc-proxy/src/metrics.rs | 1 + crates/dhcp-server/src/main.rs | 1 + crates/dhcp/src/metrics.rs | 1 + crates/dns/src/lib.rs | 1 + crates/dsx-exchange-consumer/src/lib.rs | 1 + crates/metrics-endpoint/src/lib.rs | 202 +++++++++++++++++++++++- 6 files changed, 200 insertions(+), 7 deletions(-) diff --git a/crates/bmc-proxy/src/metrics.rs b/crates/bmc-proxy/src/metrics.rs index e1049094e1..ba3282a65a 100644 --- a/crates/bmc-proxy/src/metrics.rs +++ b/crates/bmc-proxy/src/metrics.rs @@ -48,6 +48,7 @@ pub async fn start( address, registry: metrics_setup.registry, health_controller: Some(metrics_setup.health_controller), + additional_prefix: None, }, cancellation_token, listener, diff --git a/crates/dhcp-server/src/main.rs b/crates/dhcp-server/src/main.rs index 22b6e20491..cc1a4e7c8e 100644 --- a/crates/dhcp-server/src/main.rs +++ b/crates/dhcp-server/src/main.rs @@ -503,6 +503,7 @@ async fn main() -> Result<(), Box> { address: metrics_listen_addr, registry: metrics_setup.registry, health_controller: Some(metrics_setup.health_controller), + additional_prefix: None, }; // The endpoint's /health and /ready report process liveness (the // default HealthController state), not packet-serving readiness -- diff --git a/crates/dhcp/src/metrics.rs b/crates/dhcp/src/metrics.rs index b26757c8f7..4b3d6e4e4c 100644 --- a/crates/dhcp/src/metrics.rs +++ b/crates/dhcp/src/metrics.rs @@ -314,6 +314,7 @@ pub fn metrics_server() { address: metrics_endpoint, registry: mconf.registry, health_controller: Some(health_controller), + additional_prefix: None, }) .await { diff --git a/crates/dns/src/lib.rs b/crates/dns/src/lib.rs index 1085fafb81..c231f116df 100644 --- a/crates/dns/src/lib.rs +++ b/crates/dns/src/lib.rs @@ -575,6 +575,7 @@ impl DnsServer { address: config.metrics_listen_address, registry: metrics_setup.registry, health_controller: Some(metrics_setup.health_controller), + additional_prefix: None, }; tokio::spawn(async move { diff --git a/crates/dsx-exchange-consumer/src/lib.rs b/crates/dsx-exchange-consumer/src/lib.rs index 41c2e75d7b..f590a0d38e 100644 --- a/crates/dsx-exchange-consumer/src/lib.rs +++ b/crates/dsx-exchange-consumer/src/lib.rs @@ -70,6 +70,7 @@ pub async fn run_service(config: Config) -> Result<(), DsxConsumerError> { address: metrics_endpoint, registry, health_controller: Some(metrics_setup.health_controller), + additional_prefix: None, }; let join_listener = tokio::spawn(async move { metrics_endpoint::run_metrics_endpoint(&metrics_config).await }); diff --git a/crates/metrics-endpoint/src/lib.rs b/crates/metrics-endpoint/src/lib.rs index 45d31f57ba..fc85aca4ca 100644 --- a/crates/metrics-endpoint/src/lib.rs +++ b/crates/metrics-endpoint/src/lib.rs @@ -30,6 +30,7 @@ use opentelemetry::KeyValue; use opentelemetry::metrics::{Meter, MeterProvider}; use opentelemetry_sdk::metrics::SdkMeterProvider; use opentelemetry_semantic_conventions as semconv; +use prometheus::proto::MetricFamily; use prometheus::{Encoder, TextEncoder}; use tokio::net::TcpListener; use tokio_util::sync::CancellationToken; @@ -86,6 +87,16 @@ pub struct MetricsSetup { struct MetricsHandlerState { registry: prometheus::Registry, health_controller: HealthController, + additional_prefix: Option, +} + +/// An old/new prefix pair: metric families whose name starts with `old` are +/// re-exposed under `new` as well, so scrapers on either name keep working +/// through a rename migration. +#[derive(Debug, Clone)] +pub struct PrefixMigration { + pub old: String, + pub new: String, } /// Configuration for the metrics endpoint @@ -93,6 +104,13 @@ pub struct MetricsEndpointConfig { pub address: SocketAddr, pub registry: prometheus::Registry, pub health_controller: Option, + /// When set, the `/metrics` exposition additionally emits every metric family + /// whose name starts with `.old` under a copy renamed to use `.new` in place of + /// that prefix, so the same series appear under both names. This supports a + /// gradual metric-rename migration where series are published under both an old + /// and a new prefix for a time. Defaults to `None`, in which case the exposition + /// is unchanged. + pub additional_prefix: Option, } pub fn new_metrics_setup( @@ -156,8 +174,23 @@ fn create_metric_view_for_retry_histograms( ) } -/// Start a HTTP endpoint which exposes metrics using the provided configuration +/// Start a HTTP endpoint which exposes metrics using the provided configuration. +/// +/// The endpoint runs until the process exits. Callers that need to stop it (for +/// example on graceful shutdown) should use [`run_metrics_endpoint_with_cancellation`]. pub async fn run_metrics_endpoint(config: &MetricsEndpointConfig) -> Result<(), std::io::Error> { + run_metrics_endpoint_with_cancellation(config, CancellationToken::new()).await +} + +/// Start a HTTP endpoint which exposes metrics and runs until `cancel_token` is +/// cancelled. +/// +/// This binds `config.address`; callers that have already bound a listener can use +/// [`run_metrics_endpoint_with_listener`] directly. +pub async fn run_metrics_endpoint_with_cancellation( + config: &MetricsEndpointConfig, + cancel_token: CancellationToken, +) -> Result<(), std::io::Error> { let listener = TcpListener::bind(&config.address).await?; tracing::info!( @@ -165,7 +198,7 @@ pub async fn run_metrics_endpoint(config: &MetricsEndpointConfig) -> Result<(), "Starting metrics listener" ); - run_metrics_endpoint_with_listener(config, CancellationToken::new(), listener).await; + run_metrics_endpoint_with_listener(config, cancel_token, listener).await; Ok(()) } @@ -178,6 +211,7 @@ pub async fn run_metrics_endpoint_with_listener( let handler_state = Arc::new(MetricsHandlerState { registry: config.registry.clone(), health_controller: config.health_controller.clone().unwrap_or_default(), + additional_prefix: config.additional_prefix.clone(), }); while let Some(result) = cancel_token.run_until_cancelled(listener.accept()).await { @@ -210,6 +244,43 @@ pub async fn run_metrics_endpoint_with_listener( } } +/// Encode the registry's metric families in the Prometheus text exposition format. +/// +/// When `additional_prefix` is `Some(PrefixMigration { old, new })`, every gathered +/// family whose name starts with `old` is additionally emitted under a copy whose name +/// has that prefix replaced by `new`, so the same series appear under both names. When +/// `None`, the output is exactly `TextEncoder` over `registry.gather()`. +fn encode_metrics( + registry: &prometheus::Registry, + additional_prefix: Option<&PrefixMigration>, +) -> Vec { + let mut buffer = vec![]; + let encoder = TextEncoder::new(); + let mut metric_families = registry.gather(); + + if let Some(PrefixMigration { old, new }) = additional_prefix { + let alt_name_families: Vec = metric_families + .iter() + .filter_map(|family| { + if !family.name().starts_with(old) { + return None; + } + + let mut alt_name_family = family.clone(); + alt_name_family.set_name(family.name().replacen(old, new, 1)); + Some(alt_name_family) + }) + .collect(); + + if !alt_name_families.is_empty() { + metric_families.extend(alt_name_families); + } + } + + encoder.encode(&metric_families, &mut buffer).unwrap(); + buffer +} + /// Metrics request handler fn handle_metrics_request( req: Request, @@ -217,14 +288,11 @@ fn handle_metrics_request( ) -> Result>, hyper::Error> { let response = match (req.method(), req.uri().path()) { (&Method::GET, "/metrics") => { - let mut buffer = vec![]; - let encoder = TextEncoder::new(); - let metric_families = state.registry.gather(); - encoder.encode(&metric_families, &mut buffer).unwrap(); + let buffer = encode_metrics(&state.registry, state.additional_prefix.as_ref()); Response::builder() .status(200) - .header(CONTENT_TYPE, encoder.format_type()) + .header(CONTENT_TYPE, TextEncoder::new().format_type()) .header(CONTENT_LENGTH, buffer.len()) .body(Full::new(Bytes::from(buffer))) .unwrap() @@ -356,4 +424,124 @@ mod tests { assert!(controller.is_ready()); assert!(controller.is_healthy()); } + + /// Builds a deterministic registry with two `carbide_`-prefixed families plus + /// one unprefixed family, for exercising the `/metrics` exposition. + fn sample_registry() -> prometheus::Registry { + let registry = prometheus::Registry::new(); + + let requests = prometheus::Counter::with_opts(prometheus::Opts::new( + "carbide_requests_total", + "Total number of requests", + )) + .unwrap(); + requests.inc_by(3.0); + registry.register(Box::new(requests)).unwrap(); + + let queue_depth = prometheus::Gauge::with_opts(prometheus::Opts::new( + "carbide_queue_depth", + "Current queue depth", + )) + .unwrap(); + queue_depth.set(7.0); + registry.register(Box::new(queue_depth)).unwrap(); + + let other = prometheus::Counter::with_opts(prometheus::Opts::new( + "other_events_total", + "Other events", + )) + .unwrap(); + other.inc(); + registry.register(Box::new(other)).unwrap(); + + registry + } + + /// With `additional_prefix` unset, the exposition must be byte-for-byte what + /// the pre-change handler produced: a plain `TextEncoder` over `gather()`. + #[test] + fn test_additional_prefix_none_is_byte_identical() { + let registry = sample_registry(); + + let mut expected = vec![]; + TextEncoder::new() + .encode(®istry.gather(), &mut expected) + .unwrap(); + + let actual = encode_metrics(®istry, None); + + assert_eq!( + actual, expected, + "None must reproduce the pre-change exposition exactly" + ); + } + + /// With `additional_prefix` set, each matching family is emitted under both the + /// original and the alternate prefix; non-matching families are left untouched. + #[test] + fn test_additional_prefix_duplicates_matching_families() { + let registry = sample_registry(); + let prefixes = PrefixMigration { + old: "carbide_".to_string(), + new: "nico_".to_string(), + }; + + let out = String::from_utf8(encode_metrics(®istry, Some(&prefixes))).unwrap(); + + // Original families remain, unchanged. + assert!(out.contains("# HELP carbide_requests_total Total number of requests")); + assert!(out.contains("# TYPE carbide_requests_total counter")); + assert!(out.contains("\ncarbide_requests_total 3")); + assert!(out.contains("# TYPE carbide_queue_depth gauge")); + assert!(out.contains("\ncarbide_queue_depth 7")); + + // Alternate-prefixed copies carry identical HELP/TYPE/value. + assert!(out.contains("# HELP nico_requests_total Total number of requests")); + assert!(out.contains("# TYPE nico_requests_total counter")); + assert!(out.contains("\nnico_requests_total 3")); + assert!(out.contains("# TYPE nico_queue_depth gauge")); + assert!(out.contains("\nnico_queue_depth 7")); + + // A family that does not match the old prefix is not duplicated. + assert!(!out.contains("nico_other_events_total")); + assert_eq!(out.matches("other_events_total 1").count(), 1); + + // Two matching families -> two extra HELP lines (3 original + 2 alternate). + assert_eq!(out.matches("# HELP ").count(), 5); + + // The alternate copies are appended after the originals. + assert!( + out.find("carbide_requests_total 3").unwrap() + < out.find("nico_requests_total 3").unwrap() + ); + } + + /// The cancel-token entry point binds, serves, and then returns promptly once + /// the token is cancelled. + #[tokio::test] + async fn test_cancellation_shuts_down_server() { + let config = MetricsEndpointConfig { + address: "127.0.0.1:0".parse().unwrap(), + registry: prometheus::Registry::new(), + health_controller: None, + additional_prefix: None, + }; + + let cancel_token = CancellationToken::new(); + let server_token = cancel_token.clone(); + let server = tokio::spawn(async move { + run_metrics_endpoint_with_cancellation(&config, server_token).await + }); + + // Let the server bind and start accepting before we cancel it. + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + cancel_token.cancel(); + + let joined = tokio::time::timeout(std::time::Duration::from_secs(5), server).await; + let result = joined.expect("server did not shut down within timeout"); + assert!( + result.expect("server task panicked").is_ok(), + "cancelled server should return Ok" + ); + } }