Add loopback health and Prometheus metrics endpoint
All checks were successful
CI / check (push) Successful in 4m22s
All checks were successful
CI / check (push) Successful in 4m22s
This commit is contained in:
parent
381f706634
commit
ad8a3fe065
6 changed files with 180 additions and 0 deletions
|
|
@ -36,6 +36,13 @@ protocol = 1206 # InspIRCd link protocol version (1206 = insp4, 1205
|
||||||
# cert = "certs/node.crt"
|
# cert = "certs/node.crt"
|
||||||
# key = "certs/node.key"
|
# 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
|
# 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
|
# (every service comes up by default). List names here to run a subset; add
|
||||||
# "example" to also start the example template service.
|
# "example" to also start the example template service.
|
||||||
|
|
|
||||||
|
|
@ -20,6 +20,10 @@ pub struct Config {
|
||||||
// Absent = it does not start. Bind to localhost; a token is required.
|
// Absent = it does not start. Bind to localhost; a token is required.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub jsonrpc: Option<JsonRpc>,
|
pub jsonrpc: Option<JsonRpc>,
|
||||||
|
// 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<Health>,
|
||||||
// Which service modules to start. Absent = the full standard suite (all the
|
// Which service modules to start. Absent = the full standard suite (all the
|
||||||
// pseudo-clients); listing it trims that set. Every service is first-class.
|
// pseudo-clients); listing it trims that set. Every service is first-class.
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
|
|
@ -277,6 +281,14 @@ pub struct JsonRpc {
|
||||||
pub tls: Option<ServerTls>,
|
pub tls: Option<ServerTls>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[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)]
|
#[derive(Debug, Deserialize, Clone)]
|
||||||
pub struct Email {
|
pub struct Email {
|
||||||
// Sender address stamped on outgoing mail.
|
// Sender address stamped on outgoing mail.
|
||||||
|
|
|
||||||
|
|
@ -935,6 +935,12 @@ impl EventLog {
|
||||||
self.entries.len()
|
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
|
// Rewrite the log to a minimal snapshot: one event per live account and
|
||||||
// channel, authored under our origin at fresh sequence numbers. Peers
|
// channel, authored under our origin at fresh sequence numbers. Peers
|
||||||
// re-converge because the register events overwrite and cert replay is
|
// re-converge because the register events overwrite and cert replay is
|
||||||
|
|
@ -1366,6 +1372,12 @@ impl Db {
|
||||||
self.log.len()
|
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`]).
|
/// The events appended since `mark` (a value from an earlier [`log_len`]).
|
||||||
pub fn events_since(&self, mark: usize) -> Vec<Event> {
|
pub fn events_since(&self, mark: usize) -> Vec<Event> {
|
||||||
self.log.events_from(mark)
|
self.log.events_from(mark)
|
||||||
|
|
|
||||||
|
|
@ -295,6 +295,18 @@ impl Engine {
|
||||||
m
|
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<String, u64> {
|
||||||
|
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
|
// Snapshot the accumulated stat counters to the log (periodically and on
|
||||||
// shutdown) so StatServ / the Stats API keep their history across a restart.
|
// 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.
|
// Only the real counters are persisted; the live gauges above are re-derived.
|
||||||
|
|
|
||||||
131
src/health.rs
Normal file
131
src/health.rs
Normal file
|
|
@ -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<Mutex<Engine>>;
|
||||||
|
|
||||||
|
#[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<AppState>) -> Json<Value> {
|
||||||
|
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_<name> <value>`, 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<AppState>) -> 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}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -6,6 +6,7 @@ mod dict;
|
||||||
mod engine;
|
mod engine;
|
||||||
mod gossip;
|
mod gossip;
|
||||||
mod grpc;
|
mod grpc;
|
||||||
|
mod health;
|
||||||
mod jsonrpc;
|
mod jsonrpc;
|
||||||
mod keycard;
|
mod keycard;
|
||||||
mod link;
|
mod link;
|
||||||
|
|
@ -252,6 +253,11 @@ async fn main() -> Result<()> {
|
||||||
tokio::spawn(jsonrpc::run(engine.clone(), jsonrpc_cfg));
|
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.
|
// Periodically fold log churn into a snapshot when it grows past the accounts.
|
||||||
{
|
{
|
||||||
let engine = engine.clone();
|
let engine = engine.clone();
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue