Phase 5j: Prometheus /metrics endpoint
Adds a tiny axum-served HTTP endpoint that exposes the same CacheMetrics counters that back GetMetrics + gossip, in Prometheus text exposition format (v0.0.4). Enable per node by setting `cluster.prom_bind` in the daemon config. * metrics.rs: MetricsReply::to_prometheus() emits one HELP + TYPE + sample line per counter. started_unix is a gauge; everything else is a counter. Preallocates ~1 KiB so no reallocs mid-format. * prom.rs (new): PromServer::bind spins up axum on a TcpListener, serves GET /metrics, returns 404 elsewhere. Graceful shutdown via oneshot channel; abort() variant for the sync-drop path in ClusterServices. Snapshots on every scrape (no cache) — Relaxed atomic loads are cheap enough that even 1 Hz is sub-microsecond. * config.rs: new optional `cluster.prom_bind: SocketAddr` field. Default None means no server; typical value is 127.0.0.1:7702. * services.rs: wires PromServer into ClusterServices when both a router and prom_bind exist. Warns (doesn't fail) if prom_bind is set without RPC — nothing would ever change on a scrape. +8 tests: - metrics: to_prometheus emits every counter with correct type; zeros still produce valid exposition (fresh daemon scrape) - prom: content-type is text/plain; version=0.0.4; live updates between requests (no cache); unknown paths 404; shutdown stops serving - services: end-to-end scrape returns router-driven counters; prom_addr() is None when unconfigured (no accidentally-leaked port) Raw-TCP HTTP client in tests instead of pulling in reqwest — 30 lines of tokio::net + string split for GET / read-to-close is small enough to justify not adding a dep. 249 tests pass (+8 from Phase 5i). Pre-existing macOS `du -sb` failure unchanged.
This commit is contained in:
Omar Sobh
@@ -0,0 +1,272 @@
|
||||
//! Prometheus `/metrics` endpoint (Phase 5j).
|
||||
//!
|
||||
//! Exposes the same [`CacheMetrics`] counters that back `GetMetrics`
|
||||
//! and gossip, but in the Prometheus text exposition format so any
|
||||
//! existing scraper can consume them without knowing anything about
|
||||
//! QUIC or chitchat.
|
||||
//!
|
||||
//! # Wire shape
|
||||
//!
|
||||
//! One GET route:
|
||||
//!
|
||||
//! ```text
|
||||
//! GET /metrics
|
||||
//! → 200 OK, content-type text/plain; version=0.0.4
|
||||
//! <the counters>
|
||||
//! ```
|
||||
//!
|
||||
//! Every other path returns 404. The scrape is stateless — no auth,
|
||||
//! no session — because the endpoint is intended for `127.0.0.1`
|
||||
//! binding by default, or a Tailscale-only bind at operator choice.
|
||||
//! A reverse proxy in front is the recommended pattern for
|
||||
//! internet-exposed scrapers.
|
||||
|
||||
use crate::cluster::metrics::CacheMetrics;
|
||||
use anyhow::{Context, Result};
|
||||
use axum::{extract::State, http::header, response::IntoResponse, routing::get, Router};
|
||||
use std::net::SocketAddr;
|
||||
use std::sync::Arc;
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::sync::oneshot;
|
||||
use tokio::task::JoinHandle;
|
||||
|
||||
/// Prometheus text exposition v0.0.4 content-type. Every scraper in
|
||||
/// the wild parses this MIME correctly — a plain `text/plain` also
|
||||
/// works but the versioned string is the canonical value.
|
||||
const PROM_CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";
|
||||
|
||||
/// Running Prometheus HTTP server. Drop / [`PromServer::shutdown`] cancels
|
||||
/// the underlying accept loop and closes the listening socket.
|
||||
pub struct PromServer {
|
||||
/// Address we actually bound. Copies of `SocketAddr` are cheap;
|
||||
/// exposing it lets callers log or tests connect without a race.
|
||||
bind_addr: SocketAddr,
|
||||
task: JoinHandle<()>,
|
||||
shutdown_tx: Option<oneshot::Sender<()>>,
|
||||
}
|
||||
|
||||
impl std::fmt::Debug for PromServer {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
f.debug_struct("PromServer")
|
||||
.field("bind_addr", &self.bind_addr)
|
||||
.finish_non_exhaustive()
|
||||
}
|
||||
}
|
||||
|
||||
impl PromServer {
|
||||
/// Bind the HTTP server, spawn the accept loop, return. The
|
||||
/// returned handle is the sole owner of the background task; drop
|
||||
/// it (or call [`Self::shutdown`]) to stop serving.
|
||||
///
|
||||
/// `metrics` is shared with the RPC router so both endpoints
|
||||
/// observe the same counters — no risk of drift between the
|
||||
/// `GetMetrics` RPC and a scrape landing at the same instant.
|
||||
pub async fn bind(addr: SocketAddr, metrics: Arc<CacheMetrics>) -> Result<Self> {
|
||||
let listener = TcpListener::bind(addr)
|
||||
.await
|
||||
.with_context(|| format!("binding prometheus listener at {addr}"))?;
|
||||
let bind_addr = listener
|
||||
.local_addr()
|
||||
.context("reading prometheus listener bound address")?;
|
||||
let router = Router::new()
|
||||
.route("/metrics", get(metrics_handler))
|
||||
.with_state(metrics);
|
||||
let (tx, rx) = oneshot::channel::<()>();
|
||||
let task = tokio::spawn(async move {
|
||||
let shutdown = async move {
|
||||
let _ = rx.await;
|
||||
};
|
||||
if let Err(e) = axum::serve(listener, router)
|
||||
.with_graceful_shutdown(shutdown)
|
||||
.await
|
||||
{
|
||||
tracing::warn!(error = %e, "prometheus server exited with error");
|
||||
}
|
||||
});
|
||||
Ok(Self {
|
||||
bind_addr,
|
||||
task,
|
||||
shutdown_tx: Some(tx),
|
||||
})
|
||||
}
|
||||
|
||||
/// The address the server actually bound to. Differs from the
|
||||
/// requested `addr` only when the caller passed a `0` port
|
||||
/// (OS-assigned) — useful in tests.
|
||||
pub fn local_addr(&self) -> SocketAddr {
|
||||
self.bind_addr
|
||||
}
|
||||
|
||||
/// Fire the graceful-shutdown signal and await the accept task.
|
||||
/// Idempotent — a second call after the receiver's already been
|
||||
/// dropped is a no-op.
|
||||
pub async fn shutdown(mut self) {
|
||||
if let Some(tx) = self.shutdown_tx.take() {
|
||||
let _ = tx.send(());
|
||||
}
|
||||
// Best-effort await. If it panicked we've already logged.
|
||||
let _ = self.task.await;
|
||||
}
|
||||
|
||||
/// Abort the accept task without waiting. Used by `ClusterServices`
|
||||
/// during a synchronous drop where awaiting isn't an option.
|
||||
pub fn abort(self) {
|
||||
if let Some(tx) = self.shutdown_tx {
|
||||
let _ = tx.send(());
|
||||
}
|
||||
self.task.abort();
|
||||
}
|
||||
}
|
||||
|
||||
/// `GET /metrics` handler — snapshot the counters, format, return.
|
||||
///
|
||||
/// Snapshotting on every scrape (rather than caching for N ms) keeps
|
||||
/// counters as fresh as possible; the underlying `Ordering::Relaxed`
|
||||
/// atomic loads are cheap enough that even a 1 Hz scrape cadence is
|
||||
/// well under a microsecond of CPU per request.
|
||||
async fn metrics_handler(State(metrics): State<Arc<CacheMetrics>>) -> impl IntoResponse {
|
||||
let body = metrics.snapshot().to_prometheus();
|
||||
([(header::CONTENT_TYPE, PROM_CONTENT_TYPE)], body)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpStream;
|
||||
|
||||
/// Raw HTTP/1.1 client: writes one GET, reads the entire response
|
||||
/// until connection close, and splits headers from body on the
|
||||
/// double CRLF. Returns `(status, content_type, body)`.
|
||||
///
|
||||
/// The server closes each response's TCP connection (default for a
|
||||
/// non-keep-alive request), so a bounded read-to-end is safe here.
|
||||
async fn scrape(addr: SocketAddr, path: &str) -> (u16, String, String) {
|
||||
let mut sock = TcpStream::connect(addr).await.unwrap();
|
||||
let req = format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n");
|
||||
sock.write_all(req.as_bytes()).await.unwrap();
|
||||
let mut buf = Vec::with_capacity(4096);
|
||||
sock.read_to_end(&mut buf).await.unwrap();
|
||||
let raw = String::from_utf8_lossy(&buf).to_string();
|
||||
let (head, body) = raw.split_once("\r\n\r\n").unwrap_or((&raw, ""));
|
||||
let status = head
|
||||
.lines()
|
||||
.next()
|
||||
.and_then(|l| l.split_whitespace().nth(1))
|
||||
.and_then(|s| s.parse().ok())
|
||||
.unwrap_or(0);
|
||||
let ct = head
|
||||
.lines()
|
||||
.find_map(|l| {
|
||||
let (k, v) = l.split_once(':')?;
|
||||
if k.eq_ignore_ascii_case("content-type") {
|
||||
Some(v.trim().to_string())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
.unwrap_or_default();
|
||||
(status, ct, body.to_string())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scrape_returns_prometheus_text_with_correct_content_type() {
|
||||
let m = Arc::new(CacheMetrics::new());
|
||||
for _ in 0..3 {
|
||||
m.record_get_ref_hit();
|
||||
}
|
||||
m.record_get_ref_miss();
|
||||
m.record_blob_get_bytes(1024);
|
||||
let bind = "127.0.0.1:0".parse().unwrap();
|
||||
let server = PromServer::bind(bind, m.clone()).await.unwrap();
|
||||
let addr = server.local_addr();
|
||||
assert_ne!(addr.port(), 0, "kernel must assign a real port");
|
||||
|
||||
let (status, ct, body) = scrape(addr, "/metrics").await;
|
||||
assert_eq!(status, 200);
|
||||
assert!(ct.starts_with("text/plain"), "unexpected content-type {ct}");
|
||||
// Exposition body carries the counters at the values we set.
|
||||
// Body may be chunked (transfer-encoding: chunked) — check via
|
||||
// substring rather than exact equality.
|
||||
assert!(
|
||||
body.contains("clawstor_cache_get_ref_hits_total 3"),
|
||||
"body missing get_ref_hits=3:\n{body}"
|
||||
);
|
||||
assert!(
|
||||
body.contains("clawstor_cache_get_ref_misses_total 1"),
|
||||
"body missing get_ref_misses=1:\n{body}"
|
||||
);
|
||||
assert!(
|
||||
body.contains("clawstor_cache_blob_get_bytes_total 1024"),
|
||||
"body missing blob_get_bytes=1024:\n{body}"
|
||||
);
|
||||
server.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn scrape_shows_live_updates_between_requests() {
|
||||
// Prove there's no cache — a second scrape after an increment
|
||||
// must show the new value.
|
||||
let m = Arc::new(CacheMetrics::new());
|
||||
let bind = "127.0.0.1:0".parse().unwrap();
|
||||
let server = PromServer::bind(bind, m.clone()).await.unwrap();
|
||||
let addr = server.local_addr();
|
||||
|
||||
let (_, _, first) = scrape(addr, "/metrics").await;
|
||||
assert!(first.contains("clawstor_cache_get_ref_hits_total 0"));
|
||||
|
||||
for _ in 0..5 {
|
||||
m.record_get_ref_hit();
|
||||
}
|
||||
let (_, _, second) = scrape(addr, "/metrics").await;
|
||||
assert!(
|
||||
second.contains("clawstor_cache_get_ref_hits_total 5"),
|
||||
"expected 5 after increments:\n{second}"
|
||||
);
|
||||
server.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unknown_paths_return_404() {
|
||||
// Only /metrics is a valid path; anything else is a 404 (no
|
||||
// catch-all router that would tempt someone into treating this
|
||||
// as a general HTTP server).
|
||||
let m = Arc::new(CacheMetrics::new());
|
||||
let bind = "127.0.0.1:0".parse().unwrap();
|
||||
let server = PromServer::bind(bind, m.clone()).await.unwrap();
|
||||
let addr = server.local_addr();
|
||||
let (status, _, _) = scrape(addr, "/not-a-thing").await;
|
||||
assert_eq!(status, 404);
|
||||
server.shutdown().await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn shutdown_stops_serving() {
|
||||
// After shutdown() the port must refuse new connections.
|
||||
let m = Arc::new(CacheMetrics::new());
|
||||
let bind = "127.0.0.1:0".parse().unwrap();
|
||||
let server = PromServer::bind(bind, m.clone()).await.unwrap();
|
||||
let addr = server.local_addr();
|
||||
server.shutdown().await;
|
||||
// Give the OS a beat to release the socket.
|
||||
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
|
||||
// Either the connect fails or the connection is closed with 0
|
||||
// bytes read — both count as "not serving".
|
||||
let res = tokio::time::timeout(
|
||||
std::time::Duration::from_millis(500),
|
||||
TcpStream::connect(addr),
|
||||
)
|
||||
.await;
|
||||
match res {
|
||||
Err(_) => { /* timeout — good */ }
|
||||
Ok(Err(_)) => { /* connect refused — good */ }
|
||||
Ok(Ok(mut sock)) => {
|
||||
// Connected — but a read must return 0 bytes because
|
||||
// the accept loop is gone.
|
||||
let mut buf = [0u8; 16];
|
||||
let n = sock.read(&mut buf).await.unwrap_or(0);
|
||||
assert_eq!(n, 0, "server accepted a connection after shutdown");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user