From f0e5ac2cddf87e07954e2d512b196936d0ea8dde Mon Sep 17 00:00:00 2001 From: Chet Nichols III Date: Tue, 14 Jul 2026 22:46:57 -0700 Subject: [PATCH] perf(metrics-endpoint): encode the registry on a dedicated thread The shared `/metrics` handler ran `registry.gather()` and the text encode inline on the tokio task serving the connection, so a large scrape could stall a worker -- and every binary serves `/metrics` through this crate. Spawning each encode on the blocking pool would cap one encode's latency but not the aggregate CPU: N concurrent scrapes would run N encoders on N cores. Instead this hands the encode to one dedicated thread behind a bounded queue, capping encode CPU to a single core no matter how many scrapes arrive. - `run_metrics_endpoint_with_listener` spawns one `metrics-encoder` thread that owns the registry and prefix and serves encode requests off a small bounded channel; the `/metrics` handler sends a request and awaits the result. The thread exits when the server shuts down and the sender drops. - A full queue returns 503 (backpressure); a failed encode returns 500 carrying the Prometheus error, kept distinct from a dead-thread 500. - `encode_metrics` now returns the Prometheus encode error instead of unwrapping it. The success-path bytes are unchanged, so `/metrics` output stays byte-identical. This supports https://github.com/NVIDIA/infra-controller/issues/3524 --- crates/bmc-proxy/src/metrics.rs | 5 +- crates/metrics-endpoint/src/lib.rs | 527 +++++++++++++++++++++++++++-- 2 files changed, 508 insertions(+), 24 deletions(-) diff --git a/crates/bmc-proxy/src/metrics.rs b/crates/bmc-proxy/src/metrics.rs index ba3282a65a..cbbb9204e2 100644 --- a/crates/bmc-proxy/src/metrics.rs +++ b/crates/bmc-proxy/src/metrics.rs @@ -43,7 +43,7 @@ pub async fn start( .build_task() .name("bmc-proxy metrics service") .spawn(async move { - metrics_endpoint::run_metrics_endpoint_with_listener( + if let Err(e) = metrics_endpoint::run_metrics_endpoint_with_listener( &MetricsEndpointConfig { address, registry: metrics_setup.registry, @@ -54,6 +54,9 @@ pub async fn start( listener, ) .await + { + tracing::error!(error = %e, "metrics endpoint exited with error"); + } }) // Safety: Should only fail if not in a tokio runtime .expect("Error spawning metrics endpoint"); diff --git a/crates/metrics-endpoint/src/lib.rs b/crates/metrics-endpoint/src/lib.rs index fc85aca4ca..a6716423cf 100644 --- a/crates/metrics-endpoint/src/lib.rs +++ b/crates/metrics-endpoint/src/lib.rs @@ -17,6 +17,7 @@ use std::net::SocketAddr; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::mpsc::{RecvTimeoutError, SyncSender, TrySendError, sync_channel}; use bytes::Bytes; use carbide_metrics_utils::OtelView; @@ -33,6 +34,7 @@ use opentelemetry_semantic_conventions as semconv; use prometheus::proto::MetricFamily; use prometheus::{Encoder, TextEncoder}; use tokio::net::TcpListener; +use tokio::sync::oneshot; use tokio_util::sync::CancellationToken; /// Health and readiness controller @@ -83,11 +85,42 @@ pub struct MetricsSetup { pub health_controller: HealthController, } +/// Depth of the bounded queue feeding the dedicated encoder thread. +/// +/// The single encoder thread caps total encode CPU at one core no matter how many +/// scrapes arrive at once. This small queue lets a couple of scrapes wait behind an +/// in-flight encode; a scrape that arrives with the queue already full is shed with +/// `503` rather than allowed to pile up. +const ENCODER_QUEUE_CAPACITY: usize = 4; + +/// How long the encoder thread blocks on the request queue before re-checking the stop +/// flag. It bounds how long shutdown waits for the thread to notice cancellation (on top +/// of any encode already in progress). +const ENCODER_RECV_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(250); + +/// Why the encoder thread could not produce an exposition. Both variants become an HTTP +/// `500` but log distinctly. `Panicked` is isolated per-request via `catch_unwind` so a +/// misbehaving collector or observable callback cannot take down the sole encoder thread. +#[derive(Debug)] +enum EncodeError { + /// The Prometheus text encoder returned an error. + Encode(prometheus::Error), + /// A collector or observable callback panicked during `gather()`/encode; the encoder + /// thread caught the unwind and stayed alive to serve later scrapes. + Panicked, +} + +/// 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>>; + /// The shared state between HTTP requests struct MetricsHandlerState { - registry: prometheus::Registry, + /// Bounded queue to the single dedicated encoder thread. Each `/metrics` scrape + /// enqueues a reply channel and awaits the encoded exposition on it. The thread + /// owns the registry and prefix, so those are not held here. + encoder_tx: SyncSender, health_controller: HealthController, - additional_prefix: Option, } /// An old/new prefix pair: metric families whose name starts with `old` are @@ -198,20 +231,33 @@ pub async fn run_metrics_endpoint_with_cancellation( "Starting metrics listener" ); - run_metrics_endpoint_with_listener(config, cancel_token, listener).await; - Ok(()) + run_metrics_endpoint_with_listener(config, cancel_token, listener).await } -/// Run the metrics service on an existing listener (which allows this function to not return errors.) +/// Run the metrics service on an existing listener. +/// +/// Returns an error only if the dedicated encoder thread cannot be spawned at startup +/// (rare — OS resource exhaustion); once running it serves until `cancel_token` fires. pub async fn run_metrics_endpoint_with_listener( config: &MetricsEndpointConfig, cancel_token: CancellationToken, listener: TcpListener, -) { +) -> Result<(), std::io::Error> { + // Spawn the single dedicated encoder thread. It owns the registry and prefix and + // serializes all `/metrics` encoding onto one core. `should_stop` lets us shut it + // down promptly on cancellation no matter how many connection tasks still hold a + // queue sender (an idle keep-alive socket would otherwise pin the thread), and we + // join the handle below so shutdown is clean. + let should_stop = Arc::new(AtomicBool::new(false)); + let (encoder_tx, encoder_thread) = spawn_encoder_thread( + config.registry.clone(), + config.additional_prefix.clone(), + should_stop.clone(), + )?; + let handler_state = Arc::new(MetricsHandlerState { - registry: config.registry.clone(), + encoder_tx, 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 { @@ -233,7 +279,7 @@ pub async fn run_metrics_endpoint_with_listener( io, service_fn(move |req: Request| { let handler_state = handler_state.clone(); - async move { handle_metrics_request(req, handler_state) } + async move { handle_metrics_request(req, handler_state).await } }), ) .await @@ -242,6 +288,103 @@ pub async fn run_metrics_endpoint_with_listener( } }); } + + // The accept loop only returns once the token is cancelled. Signal the encoder + // thread to stop and join it so shutdown is clean. The thread exits within one + // `ENCODER_RECV_TIMEOUT` (plus any encode already in progress), independent of + // connection tasks that may still hold a queue sender. Join on a blocking thread so + // this brief wait does not stall an async worker. + should_stop.store(true, Ordering::SeqCst); + match tokio::task::spawn_blocking(move || encoder_thread.join()).await { + Ok(Ok(())) => {} + Ok(Err(_panic)) => tracing::error!("metrics encoder thread panicked"), + Err(err) => tracing::error!(error = %err, "failed to join metrics encoder thread"), + } + + Ok(()) +} + +/// Spawn the single dedicated OS thread that owns the registry and serializes all +/// `/metrics` encoding onto one core. +/// +/// The thread owns `registry` (a cheap `Arc`-backed clone) and `additional_prefix` +/// (moved in), and services encode requests arriving on the returned bounded channel: +/// for each reply-sender it encodes the current exposition with [`encode_metrics`] and +/// sends the `Result` back through the oneshot. A requester that already gave up (its +/// receiver is closed) is skipped without encoding, so an abandoned scrape never burns +/// the sole encoder. +/// +/// It blocks on the queue in short [`ENCODER_RECV_TIMEOUT`] slices so it can observe +/// `should_stop`: the server sets that flag once its accept loop ends and then joins +/// this handle, so the thread exits promptly on cancellation — within one timeout plus +/// any in-flight encode — independent of how many connection tasks still hold a queue +/// sender. Using a plain OS thread (not a tokio worker) keeps the blocking `recv_timeout` +/// off the async runtime. +/// +/// Returns an error if the OS refuses to create the thread (e.g. resource exhaustion), so +/// startup can surface it rather than panic. +fn spawn_encoder_thread( + registry: prometheus::Registry, + additional_prefix: Option, + should_stop: Arc, +) -> std::io::Result<(SyncSender, std::thread::JoinHandle<()>)> { + let (encoder_tx, encoder_rx) = sync_channel::(ENCODER_QUEUE_CAPACITY); + + let handle = std::thread::Builder::new() + .name("metrics-encoder".to_string()) + .spawn(move || { + loop { + if should_stop.load(Ordering::SeqCst) { + break; + } + match encoder_rx.recv_timeout(ENCODER_RECV_TIMEOUT) { + Ok(reply_tx) => { + // Cancellation can fire while we are parked in `recv_timeout` and a + // request arrives concurrently (e.g. from an already-accepted + // keep-alive), so the top-of-loop check may have missed it. Recheck + // before starting an expensive encode, otherwise `join` would wait + // for a scrape begun after shutdown and overrun its bound. + if should_stop.load(Ordering::SeqCst) { + break; + } + // A scrape that disconnected or timed out while queued has already + // dropped its receiver; skip the expensive gather+encode so we do + // not burn the sole encoder (and risk 503-ing live scrapes) on + // work nobody is waiting for. + if reply_tx.is_closed() { + continue; + } + // Isolate a panicking collector/observable callback: a panic during + // `gather()` must fail only this one scrape (500), not unwind and + // kill the sole encoder thread — which would then 500 every later + // scrape until the process restarts. This restores the per-request + // isolation the previous `spawn_blocking` gave for free. + let result = + match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { + encode_metrics(®istry, additional_prefix.as_ref()) + })) { + Ok(Ok(buffer)) => Ok(buffer), + Ok(Err(err)) => Err(EncodeError::Encode(err)), + Err(_panic) => { + tracing::error!( + "metrics gather/encode panicked; encoder thread surviving" + ); + Err(EncodeError::Panicked) + } + }; + // The requester may still have gone away between the check and + // here; dropping the result in that case is fine. + let _ = reply_tx.send(result); + } + // Re-check the stop flag on each timeout so shutdown stays prompt. + Err(RecvTimeoutError::Timeout) => {} + // Every sender has been dropped; nothing more can arrive. + Err(RecvTimeoutError::Disconnected) => break, + } + } + })?; + + Ok((encoder_tx, handle)) } /// Encode the registry's metric families in the Prometheus text exposition format. @@ -250,10 +393,13 @@ pub async fn run_metrics_endpoint_with_listener( /// 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()`. +/// +/// Returns the `prometheus::Error` from the text encoder rather than panicking, so the +/// handler can surface an encode failure as `500` instead of tearing down the thread. fn encode_metrics( registry: &prometheus::Registry, additional_prefix: Option<&PrefixMigration>, -) -> Vec { +) -> Result, prometheus::Error> { let mut buffer = vec![]; let encoder = TextEncoder::new(); let mut metric_families = registry.gather(); @@ -277,25 +423,73 @@ fn encode_metrics( } } - encoder.encode(&metric_families, &mut buffer).unwrap(); - buffer + encoder.encode(&metric_families, &mut buffer)?; + Ok(buffer) } /// Metrics request handler -fn handle_metrics_request( +async fn handle_metrics_request( req: Request, state: Arc, ) -> Result>, hyper::Error> { let response = match (req.method(), req.uri().path()) { (&Method::GET, "/metrics") => { - let buffer = encode_metrics(&state.registry, state.additional_prefix.as_ref()); - - Response::builder() - .status(200) - .header(CONTENT_TYPE, TextEncoder::new().format_type()) - .header(CONTENT_LENGTH, buffer.len()) - .body(Full::new(Bytes::from(buffer))) - .unwrap() + // `gather()` + text encoding can briefly stall a tokio worker on a large + // registry, so it runs on the single dedicated encoder thread. Hand it a + // reply channel through the bounded queue and await the encoded exposition. + let (reply_tx, reply_rx) = oneshot::channel(); + match state.encoder_tx.try_send(reply_tx) { + Ok(()) => {} + 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"); + return Ok(Response::builder() + .status(503) + .body(Full::new(Bytes::from("metrics encoder busy"))) + .unwrap()); + } + Err(TrySendError::Disconnected(_)) => { + tracing::error!("metrics encoder thread is gone; cannot encode metrics"); + return Ok(Response::builder() + .status(500) + .body(Full::new(Bytes::from("Failed to encode metrics"))) + .unwrap()); + } + } + + match reply_rx.await { + Ok(Ok(buffer)) => Response::builder() + .status(200) + .header(CONTENT_TYPE, TextEncoder::new().format_type()) + .header(CONTENT_LENGTH, buffer.len()) + .body(Full::new(Bytes::from(buffer))) + .unwrap(), + Ok(Err(EncodeError::Encode(err))) => { + tracing::error!(error = %err, "failed to encode metrics"); + Response::builder() + .status(500) + .body(Full::new(Bytes::from("Failed to encode metrics"))) + .unwrap() + } + Ok(Err(EncodeError::Panicked)) => { + tracing::error!("metrics encoder caught a panic while encoding"); + Response::builder() + .status(500) + .body(Full::new(Bytes::from("Failed to encode metrics"))) + .unwrap() + } + Err(_) => { + // 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"); + Response::builder() + .status(500) + .body(Full::new(Bytes::from("Failed to encode metrics"))) + .unwrap() + } + } } (&Method::GET, "/health") if state.health_controller.is_healthy() => Response::builder() .status(200) @@ -468,7 +662,7 @@ mod tests { .encode(®istry.gather(), &mut expected) .unwrap(); - let actual = encode_metrics(®istry, None); + let actual = encode_metrics(®istry, None).expect("encode succeeds"); assert_eq!( actual, expected, @@ -486,7 +680,9 @@ mod tests { new: "nico_".to_string(), }; - let out = String::from_utf8(encode_metrics(®istry, Some(&prefixes))).unwrap(); + let out = + String::from_utf8(encode_metrics(®istry, Some(&prefixes)).expect("encode succeeds")) + .unwrap(); // Original families remain, unchanged. assert!(out.contains("# HELP carbide_requests_total Total number of requests")); @@ -544,4 +740,289 @@ mod tests { "cancelled server should return Ok" ); } + + /// A live `GET /metrics` round-trip returns `200` with a body byte-for-byte equal to + /// [`encode_metrics`], proving the dedicated encoder thread produces the same + /// exposition the handler used to build inline. + #[tokio::test] + async fn test_metrics_endpoint_serves_encoded_body() { + let registry = sample_registry(); + let expected = encode_metrics(®istry, None).expect("encode succeeds"); + let expected_content_type = TextEncoder::new().format_type().to_ascii_lowercase(); + + // Bind here so we know the address and there is no bind race with the client. + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let config = MetricsEndpointConfig { + address: addr, + registry, + 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_listener(&config, server_token, listener).await + }); + + let (headers, body) = scrape_metrics(addr).await; + + assert!( + headers.starts_with("http/1.1 200"), + "expected a 200 status line, got headers:\n{headers}" + ); + assert!( + headers.contains(&format!("content-type: {expected_content_type}")), + "expected content-type {expected_content_type:?} in headers:\n{headers}" + ); + assert_eq!( + body, expected, + "served body must be byte-identical to encode_metrics output" + ); + + cancel_token.cancel(); + server + .await + .unwrap() + .expect("metrics server exited cleanly"); + } + + /// The dedicated encoder thread answers an encode request with the expected bytes and + /// then exits once the stop flag is set — while a sender is still alive — proving the + /// exit is flag-driven and never gated on connection tasks dropping their senders. + #[test] + fn test_encoder_thread_answers_then_exits_on_stop_flag() { + let registry = sample_registry(); + let expected = encode_metrics(®istry, None).expect("encode succeeds"); + + let should_stop = Arc::new(AtomicBool::new(false)); + let (encoder_tx, handle) = spawn_encoder_thread(registry, None, should_stop.clone()) + .expect("spawn encoder thread"); + + // A normal request is answered with the same bytes as encode_metrics. + let (reply_tx, reply_rx) = oneshot::channel(); + encoder_tx.try_send(reply_tx).expect("queue has room"); + let encoded = reply_rx + .blocking_recv() + .expect("encoder replied") + .expect("encode succeeds"); + assert_eq!(encoded, expected, "encoder bytes match encode_metrics"); + + // Set the stop flag but keep the sender alive: the thread must still exit. + should_stop.store(true, Ordering::SeqCst); + + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + while !handle.is_finished() { + assert!( + std::time::Instant::now() < deadline, + "encoder thread did not exit after the stop flag was set" + ); + std::thread::sleep(std::time::Duration::from_millis(10)); + } + handle.join().expect("encoder thread joined cleanly"); + + // The sender was alive the whole time, so the exit came from the flag, not from + // the queue disconnecting. + drop(encoder_tx); + } + + /// Cancelling the server while a connection (and therefore a queue-sender clone in its + /// handler state) is still alive must still exit and join the encoder thread. The + /// server future only completes after that join, so finishing within the timeout + /// proves the thread did not linger behind the open connection. + #[tokio::test] + async fn test_cancel_exits_encoder_thread_with_live_connection() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + + let config = MetricsEndpointConfig { + address: addr, + registry: sample_registry(), + 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_listener(&config, server_token, listener).await + }); + + // Complete one keep-alive `/health` round-trip BEFORE cancelling, so the connection + // is deterministically accepted and its task holds an `encoder_tx` clone. Without + // `Connection: close` the socket stays open across shutdown. + let mut held = tokio::net::TcpStream::connect(addr).await.unwrap(); + held.write_all(b"GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n") + .await + .unwrap(); + let response = tokio::time::timeout(std::time::Duration::from_secs(5), async { + let mut raw = Vec::new(); + let mut chunk = [0u8; 128]; + loop { + // Read until we have the full headers plus the Content-Length body. + if let Some(end) = raw.windows(4).position(|w| w == b"\r\n\r\n") { + let head = String::from_utf8_lossy(&raw[..end]).to_ascii_lowercase(); + let body_len: usize = head + .lines() + .find_map(|line| line.strip_prefix("content-length:")) + .map_or(0, |value| value.trim().parse().unwrap()); + if raw.len() >= end + 4 + body_len { + break; + } + } + let n = held.read(&mut chunk).await.unwrap(); + assert!(n > 0, "connection closed before a full /health response"); + raw.extend_from_slice(&chunk[..n]); + } + String::from_utf8_lossy(&raw).to_ascii_lowercase() + }) + .await + .expect("/health round-trip timed out"); + assert!( + response.starts_with("http/1.1 200"), + "kept-alive /health should be 200, got:\n{response}" + ); + + // Cancel with the connection still open. The encoder thread must still exit and be + // joined — the server future only completes after that join. + cancel_token.cancel(); + + let joined = tokio::time::timeout(std::time::Duration::from_secs(5), server).await; + joined + .expect("server did not shut down (encoder-thread join hung) within timeout") + .expect("server task panicked") + .expect("metrics server exited cleanly"); + + drop(held); + } + + /// Issue one `GET /metrics` over a fresh connection and return the lowercased header + /// block and the raw body bytes. `Connection: close` lets us read to EOF without + /// parsing `Content-Length`. + async fn scrape_metrics(addr: SocketAddr) -> (String, Vec) { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let exchange = async { + let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap(); + stream + .write_all(b"GET /metrics HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .await + .unwrap(); + let mut raw = Vec::new(); + stream.read_to_end(&mut raw).await.unwrap(); + raw + }; + let raw = tokio::time::timeout(std::time::Duration::from_secs(5), exchange) + .await + .expect("metrics round-trip timed out"); + + let separator = b"\r\n\r\n"; + let split = raw + .windows(separator.len()) + .position(|window| window == separator) + .expect("response has a header/body separator"); + let headers = String::from_utf8_lossy(&raw[..split]).to_ascii_lowercase(); + let body = raw[split + separator.len()..].to_vec(); + (headers, body) + } + + /// A collector that panics the first time it is gathered and behaves thereafter. It + /// lets a test drive one panicking `/metrics` scrape and confirm the encoder thread + /// survives to serve the next. + struct PanicOnceCollector { + desc: prometheus::core::Desc, + panicked: AtomicBool, + } + + impl PanicOnceCollector { + fn new() -> Self { + let desc = prometheus::core::Desc::new( + "panic_probe".to_string(), + "collector that panics on its first gather".to_string(), + vec![], + std::collections::HashMap::new(), + ) + .unwrap(); + Self { + desc, + panicked: AtomicBool::new(false), + } + } + } + + impl prometheus::core::Collector for PanicOnceCollector { + fn desc(&self) -> Vec<&prometheus::core::Desc> { + vec![&self.desc] + } + + fn collect(&self) -> Vec { + if !self.panicked.swap(true, Ordering::SeqCst) { + panic!("collector panic during gather"); + } + vec![] + } + } + + /// A collector that panics during `gather()` must fail only that one scrape (500); the + /// sole encoder thread has to survive so a subsequent scrape on the same server still + /// succeeds. This is the per-request panic isolation the previous `spawn_blocking` + /// provided for free. + #[tokio::test] + async fn test_encoder_survives_collector_panic() { + let registry = prometheus::Registry::new(); + registry + .register(Box::new(PanicOnceCollector::new())) + .unwrap(); + let survives = prometheus::Counter::with_opts(prometheus::Opts::new( + "survives_total", + "present once the encoder survives a panic", + )) + .unwrap(); + survives.inc(); + registry.register(Box::new(survives)).unwrap(); + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + let config = MetricsEndpointConfig { + address: addr, + registry, + 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_listener(&config, server_token, listener).await + }); + + // 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; + assert!( + headers.starts_with("http/1.1 500"), + "panicking scrape should be 500, got headers:\n{headers}" + ); + + // The thread survived, so a subsequent scrape on the same server succeeds. + let (headers, body) = scrape_metrics(addr).await; + assert!( + headers.starts_with("http/1.1 200"), + "post-panic scrape should be 200, got headers:\n{headers}" + ); + assert!( + String::from_utf8_lossy(&body).contains("survives_total"), + "post-panic body should contain the registry's metrics" + ); + + cancel_token.cancel(); + server + .await + .unwrap() + .expect("metrics server exited cleanly"); + } }