From ad8a3fe0653f39043525c244adc4a9341bc6195d Mon Sep 17 00:00:00 2001 From: Jean Date: Sun, 19 Jul 2026 04:13:18 +0000 Subject: [PATCH] Add loopback health and Prometheus metrics endpoint --- config.example.toml | 7 +++ src/config.rs | 12 ++++ src/engine/db/mod.rs | 12 ++++ src/engine/mod.rs | 12 ++++ src/health.rs | 131 +++++++++++++++++++++++++++++++++++++++++++ src/main.rs | 6 ++ 6 files changed, 180 insertions(+) create mode 100644 src/health.rs diff --git a/config.example.toml b/config.example.toml index 5164586..d5624f5 100644 --- a/config.example.toml +++ b/config.example.toml @@ -36,6 +36,13 @@ protocol = 1206 # InspIRCd link protocol version (1206 = insp4, 1205 # cert = "certs/node.crt" # key = "certs/node.key" +# Liveness + Prometheus metrics (plain HTTP, read-only, unauthenticated). Omit to +# leave it off. Bind to localhost and point a monitor at it: GET /health for a +# liveness probe (JSON), GET /metrics for a scrape (gauges: account/channel/oper +# totals, per-service counters, and event.lamport — a monotonic log-progress gauge). +# [health] +# bind = "127.0.0.1:9099" + # Which service modules to start. Omit this section for the full standard suite # (every service comes up by default). List names here to run a subset; add # "example" to also start the example template service. diff --git a/src/config.rs b/src/config.rs index 03ffba4..2a0e3ec 100644 --- a/src/config.rs +++ b/src/config.rs @@ -20,6 +20,10 @@ pub struct Config { // Absent = it does not start. Bind to localhost; a token is required. #[serde(default)] pub jsonrpc: Option, + // Liveness + Prometheus metrics endpoint (plain HTTP). Absent = it does not + // start. Bind to localhost; unauthenticated by design (read-only gauges). + #[serde(default)] + pub health: Option, // Which service modules to start. Absent = the full standard suite (all the // pseudo-clients); listing it trims that set. Every service is first-class. #[serde(default)] @@ -277,6 +281,14 @@ pub struct JsonRpc { pub tls: Option, } +#[derive(Debug, Deserialize, Clone)] +pub struct Health { + // Address to accept HTTP on, e.g. "127.0.0.1:9099". Keep it on localhost: + // the endpoint is unauthenticated and serves read-only gauges for a monitor + // (Prometheus scrape of /metrics, or /health for a liveness probe). + pub bind: String, +} + #[derive(Debug, Deserialize, Clone)] pub struct Email { // Sender address stamped on outgoing mail. diff --git a/src/engine/db/mod.rs b/src/engine/db/mod.rs index f9d4c91..333c323 100644 --- a/src/engine/db/mod.rs +++ b/src/engine/db/mod.rs @@ -935,6 +935,12 @@ impl EventLog { self.entries.len() } + // The highest Lamport clock seen. Monotonic across compaction (unlike `len`, + // which a snapshot shrinks), so it's the gauge that shows the log advancing. + fn lamport(&self) -> u64 { + self.lamport + } + // Rewrite the log to a minimal snapshot: one event per live account and // channel, authored under our origin at fresh sequence numbers. Peers // re-converge because the register events overwrite and cert replay is @@ -1366,6 +1372,12 @@ impl Db { self.log.len() } + /// The highest Lamport clock committed — a monotonic gauge of log progress + /// (survives compaction), for the health/metrics endpoint. + pub fn lamport(&self) -> u64 { + self.log.lamport() + } + /// The events appended since `mark` (a value from an earlier [`log_len`]). pub fn events_since(&self, mark: usize) -> Vec { self.log.events_from(mark) diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 53f15a4..b0af0d2 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -295,6 +295,18 @@ impl Engine { m } + // The stat snapshot plus event-log position, for the health/metrics endpoint. + // The log gauges (seq/lamport/entries) let a monitor watch replication progress + // and catch a stalled or non-advancing log before users notice. + pub fn health_metrics(&self) -> std::collections::BTreeMap { + let mut m = self.stats_snapshot(); + // lamport is monotonic (survives compaction) — the "log advancing" gauge; + // entries is the current in-memory log size (drops at each compaction). + m.insert("event.lamport".to_string(), self.db.lamport()); + m.insert("log.entries".to_string(), self.db.log_len() as u64); + m + } + // Snapshot the accumulated stat counters to the log (periodically and on // shutdown) so StatServ / the Stats API keep their history across a restart. // Only the real counters are persisted; the live gauges above are re-derived. diff --git a/src/health.rs b/src/health.rs new file mode 100644 index 0000000..94fc1bf --- /dev/null +++ b/src/health.rs @@ -0,0 +1,131 @@ +//! A tiny liveness + metrics endpoint (plain HTTP), for a monitor to watch a +//! live node. Deliberately minimal and safe: +//! - read-only: only gauges/counters, no way to mutate any state; +//! - unauthenticated by design (the data isn't secret) — so bind it to +//! localhost and let a monitor/scraper reach it there; +//! - two GET routes only: `/health` (JSON liveness) and `/metrics` +//! (Prometheus text exposition). +//! +//! The gauges come straight from the engine: the stat counters, the live +//! account/channel/bot/oper totals, and the event-log position (seq / lamport / +//! entries) so a stalled or non-advancing log is visible to alerting. + +use std::net::SocketAddr; +use std::sync::Arc; +use std::time::Instant; + +use axum::extract::State; +use axum::http::header; +use axum::response::IntoResponse; +use axum::routing::get; +use axum::{Json, Router}; +use serde_json::{json, Value}; +use tokio::sync::Mutex; + +use crate::config::Health as HealthCfg; +use crate::engine::Engine; + +type Shared = Arc>; + +#[derive(Clone)] +struct AppState { + engine: Shared, + started: Instant, +} + +// Start the endpoint, if configured. Absent [health] in config.toml = no-op. +pub async fn run(engine: Shared, cfg: HealthCfg) { + let addr: SocketAddr = match cfg.bind.parse() { + Ok(a) => a, + Err(e) => return tracing::error!(%e, bind = %cfg.bind, "bad health bind address"), + }; + let state = AppState { engine, started: Instant::now() }; + let app = Router::new() + .route("/health", get(health)) + .route("/metrics", get(metrics)) + .with_state(state); + let listener = match tokio::net::TcpListener::bind(addr).await { + Ok(l) => l, + Err(e) => return tracing::error!(%e, %addr, "health bind failed"), + }; + tracing::info!(%addr, "health/metrics endpoint listening"); + if let Err(e) = axum::serve(listener, app).await { + tracing::error!(%e, "health server exited"); + } +} + +// Liveness probe: 200 + a compact JSON body. Reaching it at all means the tokio +// runtime is up and the engine lock is obtainable (not deadlocked). +async fn health(State(state): State) -> Json { + let gauges = state.engine.lock().await.health_metrics(); + Json(json!({ + "status": "ok", + "uptime_s": state.started.elapsed().as_secs(), + "gauges": gauges, + })) +} + +// Prometheus text exposition. Each gauge becomes `echo_ `, with the +// dotted keys mapped to Prometheus-legal underscores (accounts.total -> +// echo_accounts_total). Plus a synthetic uptime gauge. +async fn metrics(State(state): State) -> impl IntoResponse { + let gauges = state.engine.lock().await.health_metrics(); + let mut body = String::with_capacity(64 * (gauges.len() + 1)); + body.push_str("# HELP echo_uptime_seconds Seconds since this node started.\n"); + body.push_str("# TYPE echo_uptime_seconds gauge\n"); + body.push_str(&format!("echo_uptime_seconds {}\n", state.started.elapsed().as_secs())); + for (key, value) in &gauges { + let metric = sanitize(key); + body.push_str(&format!("# TYPE echo_{metric} gauge\n")); + body.push_str(&format!("echo_{metric} {value}\n")); + } + ([(header::CONTENT_TYPE, "text/plain; version=0.0.4")], body) +} + +// Map a gauge key to a Prometheus-legal metric name: anything not [a-z0-9_] +// becomes '_' (dots, dashes). Keys are ASCII service/stat names, so this is +// enough — no leading-digit case to worry about (all keys start with a letter). +fn sanitize(key: &str) -> String { + key.chars().map(|c| if c.is_ascii_alphanumeric() { c } else { '_' }).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::engine::db::Db; + use axum::http::StatusCode; + + fn state(tag: &str) -> AppState { + let path = std::env::temp_dir().join(format!("echo-health-{tag}.jsonl")); + let _ = std::fs::remove_file(&path); + let mut e = Engine::new(vec![], Db::open(&path, "42S")); + e.bump("botserv.messages"); + AppState { engine: Arc::new(Mutex::new(e)), started: Instant::now() } + } + + #[tokio::test] + async fn health_reports_ok_and_gauges() { + let Json(body) = health(State(state("live"))).await; + assert_eq!(body["status"], "ok"); + assert!(body["gauges"]["accounts.total"].is_number()); + assert!(body["gauges"]["event.lamport"].is_number(), "log position present: {body}"); + } + + #[tokio::test] + async fn metrics_is_prometheus_text_with_sanitized_names() { + let resp = metrics(State(state("metrics"))).await.into_response(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20).await.unwrap(); + let text = String::from_utf8(bytes.to_vec()).unwrap(); + // Dotted keys are sanitized to underscores, counters are present, and the + // synthetic uptime gauge is emitted. + assert!(text.contains("echo_botserv_messages 1"), "counter: {text}"); + assert!(text.contains("echo_accounts_total "), "gauge: {text}"); + assert!(text.contains("echo_uptime_seconds "), "uptime: {text}"); + // No dotted metric NAME leaks through (HELP prose may contain periods, so + // check only the sample lines, not the comments). + let sample_lines: Vec<&str> = text.lines().filter(|l| l.starts_with("echo_")).collect(); + assert!(!sample_lines.is_empty()); + assert!(sample_lines.iter().all(|l| !l.split(' ').next().unwrap().contains('.')), "no dotted names: {text}"); + } +} diff --git a/src/main.rs b/src/main.rs index 86a4d11..9ef0ab6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -6,6 +6,7 @@ mod dict; mod engine; mod gossip; mod grpc; +mod health; mod jsonrpc; mod keycard; mod link; @@ -252,6 +253,11 @@ async fn main() -> Result<()> { tokio::spawn(jsonrpc::run(engine.clone(), jsonrpc_cfg)); } + // Liveness + Prometheus metrics endpoint (plain HTTP, localhost). + if let Some(health_cfg) = cfg.health.clone() { + tokio::spawn(health::run(engine.clone(), health_cfg)); + } + // Periodically fold log churn into a snapshot when it grows past the accounts. { let engine = engine.clone();