metrics.rs 92.5%

383 of 414 lines run, 35 of 80 functions · run #537 on main · run missed
//! `/metrics` (oxygenhub#212): the hub's numbers in the Prometheus text
//! format, for a monitoring stack (oxygenhub-gcp's Prometheus and Grafana).
//! Private: served only to a bearer of `[server] metrics_token`, and not
//! at all while that is unset.
//!
//! The process, the state cache, repositories kept in buckets, OxygenFlow's
//! jobs, runs and runners, the transparency log, the database pool, pushes,
//! backups, mail, and HTTP requests by route class and status with latency
//! histograms. The rest the issue lists comes next.

use crate::{db, Shared};
use axum::extract::State;
use axum::http::{header, HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use std::collections::HashMap;
use std::fmt::Write;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;

/// Messages the hub sent, and failed to send, since it started.
pub static MAIL_SENT: AtomicU64 = AtomicU64::new(0);
pub static MAIL_FAILED: AtomicU64 = AtomicU64::new(0);

/// Latency buckets, in seconds.
const BUCKETS: [f64; 12] = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0];
/// At most this many route classes are kept; more go to `other`.
const MAX_ROUTES: usize = 300;

#[derive(Default)]
struct Hist {
    /// Requests by status class (2xx … 5xx).
    by_status: [u64; 4],
    buckets: [u64; BUCKETS.len()],
    sum: f64,
    count: u64,
}

static HTTP: Mutex<Option<HashMap<(String, String), Hist>>> = Mutex::new(None);

/// Bytes the protocol received and sent, by endpoint, counted before the
/// answer is compressed: what pushes and fetches carry.
static PROTOCOL: Mutex<std::collections::BTreeMap<&'static str, (u64, u64)>> = Mutex::new(std::collections::BTreeMap::new());

/// Count a protocol call's request and response bodies.
pub fn protocol_bytes(endpoint: &'static str, received: u64, sent: u64) {
    let mut g = PROTOCOL.lock().unwrap_or_else(|p| p.into_inner());
    let e = g.entry(endpoint).or_default();
    e.0 += received;
    e.1 += sent;
}

/// A request's route class: a few dozen at most, so a label never holds a
/// user's or repository's name.
pub fn route_class(path: &str) -> String {
    let segs: Vec<&str> = path.trim_start_matches('/').split('/').filter(|s| !s.is_empty()).collect();
    let tab = |s: &str| s.chars().all(|c| c.is_ascii_lowercase() || c == '-' || c == '.') && s.len() <= 24;
    match segs.as_slice() {
        [] => "page:home".into(),
        ["static", ..] => "static".into(),
        ["api", "v1", area, ..] if tab(area) => format!("api:{area}"),
        ["api", ..] => "api".into(),
        ["log", ..] => "tlog".into(),
        [".well-known", ..] => "well-known".into(),
        ["live", ..] => "live".into(),
        ["avatar", ..] => "avatar".into(),
        [_, _, "oxy", "v1", endpoint, ..] if tab(endpoint) => format!("protocol:{endpoint}"),
        [_, _, "oxy", ..] => "protocol".into(),
        [_, _, t, ..] if tab(t) => format!("page:{t}"),
        [_, _] => "page:repository".into(),
        [one] if tab(one) => format!("page:{one}"),
        _ => "page:other".into(),
    }
}

/// Count a request (from the router's outermost timing layer).
pub fn observe(method: &str, path: &str, status: u16, seconds: f64) {
    let mut g = HTTP.lock().unwrap_or_else(|p| p.into_inner());
    let map = g.get_or_insert_with(HashMap::new);
    let mut route = route_class(path);
    if !map.contains_key(&(route.clone(), method.to_string())) && map.len() >= MAX_ROUTES {
        route = "other".into();
    }
    let h = map.entry((route, method.to_string())).or_default();
    let class = (status / 100).clamp(2, 5) as usize - 2;
    h.by_status[class] += 1;
    for (i, b) in BUCKETS.iter().enumerate() {
        if seconds <= *b {
            h.buckets[i] += 1;
        }
    }
    h.sum += seconds;
    h.count += 1;
}

fn http_families(out: &mut String) {
    let g = HTTP.lock().unwrap_or_else(|p| p.into_inner());
    let Some(map) = g.as_ref() else { return };
    let mut keys: Vec<&(String, String)> = map.keys().collect();
    keys.sort();
    let _ = writeln!(out, "# HELP oxyhub_http_requests_total HTTP requests by route class, method and status class.");
    let _ = writeln!(out, "# TYPE oxyhub_http_requests_total counter");
    for k in &keys {
        let h = &map[*k];
        for (i, n) in h.by_status.iter().enumerate().filter(|(_, n)| **n > 0) {
            let _ = writeln!(
                out,
                "oxyhub_http_requests_total{{route=\"{}\",method=\"{}\",status=\"{}xx\"}} {n}",
                label(&k.0),
                label(&k.1),
                i + 2
            );
        }
    }
    let _ = writeln!(out, "# HELP oxyhub_http_request_duration_seconds How long requests took, by route class and method.");
    let _ = writeln!(out, "# TYPE oxyhub_http_request_duration_seconds histogram");
    for k in &keys {
        let h = &map[*k];
        let l = format!("route=\"{}\",method=\"{}\"", label(&k.0), label(&k.1));
        for (i, b) in BUCKETS.iter().enumerate() {
            let _ = writeln!(out, "oxyhub_http_request_duration_seconds_bucket{{{l},le=\"{b}\"}} {}", h.buckets[i]);
        }
        let _ = writeln!(out, "oxyhub_http_request_duration_seconds_bucket{{{l},le=\"+Inf\"}} {}", h.count);
        let _ = writeln!(out, "oxyhub_http_request_duration_seconds_sum{{{l}}} {}", h.sum);
        let _ = writeln!(out, "oxyhub_http_request_duration_seconds_count{{{l}}} {}", h.count);
    }
}

/// One metric family: its help, type and samples (label set, value).
fn family(out: &mut String, name: &str, help: &str, kind: &str, samples: &[(String, f64)]) {
    let _ = writeln!(out, "# HELP {name} {help}");
    let _ = writeln!(out, "# TYPE {name} {kind}");
    for (labels, v) in samples {
        if labels.is_empty() {
            let _ = writeln!(out, "{name} {v}");
        } else {
            let _ = writeln!(out, "{name}{{{labels}}} {v}");
        }
    }
}

/// A label value, escaped as the text format asks.
fn label(v: &str) -> String {
    v.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n")
}

/// The process's resident memory, from /proc (Linux), in bytes.
fn resident_bytes() -> Option<f64> {
    let statm = std::fs::read_to_string("/proc/self/statm").ok()?;
    let pages: f64 = statm.split_whitespace().nth(1)?.parse().ok()?;
    Some(pages * 4096.0)
}

/// The metrics as text, for whoever is let in.
pub async fn render(app: &Shared) -> String {
    let mut out = String::new();
    family(
        &mut out,
        "oxyhub_build_info",
        "The hub's version and build.",
        "gauge",
        &[(format!("version=\"{}\",build=\"{}\"", env!("CARGO_PKG_VERSION"), label(env!("OXY_BUILD_ID"))), 1.0)],
    );
    family(
        &mut out,
        "oxyhub_uptime_seconds",
        "Seconds since the hub started.",
        "gauge",
        &[(String::new(), app.started.elapsed().as_secs_f64())],
    );
    if let Some(rss) = resident_bytes() {
        family(&mut out, "oxyhub_resident_memory_bytes", "The process's resident memory.", "gauge", &[(String::new(), rss)]);
    }
    let (states, weight, summaries) = app.repos.cache().stats();
    family(&mut out, "oxyhub_state_cache_states", "States the state cache holds.", "gauge", &[(String::new(), states as f64)]);
    family(
        &mut out,
        "oxyhub_state_cache_bytes",
        "What the state cache's states are estimated to weigh.",
        "gauge",
        &[(String::new(), weight as f64)],
    );
    family(
        &mut out,
        "oxyhub_state_cache_budget_bytes",
        "What the state cache may hold.",
        "gauge",
        &[(String::new(), crate::repos::cache::budget() as f64)],
    );
    family(&mut out, "oxyhub_state_cache_summaries", "Summaries the state cache holds.", "gauge", &[(String::new(), summaries as f64)]);
    let strata = oxy_strata::stats();
    family(
        &mut out,
        "oxyhub_bucket_clients_opened_total",
        "Bucket store clients opened.",
        "counter",
        &[(String::new(), strata.opens as f64)],
    );
    family(
        &mut out,
        "oxyhub_bucket_requests_total",
        "Requests to bucket stores, by kind.",
        "counter",
        &strata.ops().map(|(k, n)| (format!("kind=\"{k}\""), n as f64)).collect::<Vec<_>>(),
    );
    family(
        &mut out,
        "oxyhub_bucket_request_seconds_total",
        "Time spent in requests to bucket stores.",
        "counter",
        &[(String::new(), strata.millis as f64 / 1000.0)],
    );
    let jobs: Vec<(String, i64)> =
        sqlx::query_as("SELECT status, COUNT(*) FROM flow_jobs WHERE status IN ('queued', 'running') GROUP BY status")
            .fetch_all(&app.db)
            .await
            .unwrap_or_default();
    let count = |s: &str| jobs.iter().find(|(k, _)| k == s).map(|(_, n)| *n).unwrap_or(0) as f64;
    let waiting: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM flow_jobs WHERE status = 'queued' AND approval = 'waiting'")
        .fetch_one(&app.db)
        .await
        .unwrap_or(0);
    family(
        &mut out,
        "oxyhub_flow_jobs",
        "OxygenFlow jobs now, by state (waiting: for a person's approval).",
        "gauge",
        &[
            ("state=\"queued\"".into(), count("queued") - waiting as f64),
            ("state=\"waiting\"".into(), waiting as f64),
            ("state=\"running\"".into(), count("running")),
        ],
    );
    let runners = db::runners::list(&app.db).await.unwrap_or_default();
    family(
        &mut out,
        "oxyhub_flow_runners",
        "Registered OxygenFlow runners, by whether they reported in the last minutes.",
        "gauge",
        &[
            ("online=\"true\"".into(), runners.iter().filter(|r| r.online()).count() as f64),
            ("online=\"false\"".into(), runners.iter().filter(|r| !r.online()).count() as f64),
        ],
    );
    let now = db::now_ms();
    family(
        &mut out,
        "oxyhub_flow_runner_last_seen_seconds",
        "Seconds since each registered runner last called the hub.",
        "gauge",
        &runners
            .iter()
            .filter_map(|r| r.last_seen_ms.map(|t| (format!("runner=\"{}\"", label(&r.name)), (now - t).max(0) as f64 / 1000.0)))
            .collect::<Vec<_>>(),
    );
    let since = now - 86_400_000;
    let runs: Vec<(String, i64)> = sqlx::query_as("SELECT status, COUNT(*) FROM flow_runs WHERE finished_ms > $1 GROUP BY status")
        .bind(since)
        .fetch_all(&app.db)
        .await
        .unwrap_or_default();
    family(
        &mut out,
        "oxyhub_flow_runs_last_day",
        "OxygenFlow runs finished in the last 24 hours, by result.",
        "gauge",
        &["passed", "failed", "error"]
            .iter()
            .map(|r| (format!("result=\"{r}\""), runs.iter().find(|(s, _)| s == r).map(|(_, n)| *n).unwrap_or(0) as f64))
            .collect::<Vec<_>>(),
    );
    family(&mut out, "oxyhub_tlog_entries", "Entries in the transparency log.", "gauge", &[(String::new(), app.tlog.size() as f64)]);
    let protocol: Vec<(&'static str, (u64, u64))> =
        PROTOCOL.lock().unwrap_or_else(|p| p.into_inner()).iter().map(|(k, v)| (*k, *v)).collect();
    family(
        &mut out,
        "oxyhub_protocol_bytes_total",
        "Bytes the Oxygen protocol received and sent since the hub started, by endpoint, before compression: push carries pushes, fetch and objects carry fetches.",
        "counter",
        &protocol
            .iter()
            .flat_map(|(e, (rx, tx))| {
                [(format!("endpoint=\"{e}\",direction=\"received\""), *rx as f64), (format!("endpoint=\"{e}\",direction=\"sent\""), *tx as f64)]
            })
            .collect::<Vec<_>>(),
    );
    family(
        &mut out,
        "oxyhub_mail_total",
        "Messages the hub sent and failed to send since it started.",
        "counter",
        &[
            ("result=\"sent\"".into(), MAIL_SENT.load(Ordering::Relaxed) as f64),
            ("result=\"failed\"".into(), MAIL_FAILED.load(Ordering::Relaxed) as f64),
        ],
    );
    family(
        &mut out,
        "oxyhub_db_connections",
        "The database pool's connections, by whether they are idle.",
        "gauge",
        &[
            ("idle=\"true\"".into(), app.db.num_idle() as f64),
            ("idle=\"false\"".into(), (app.db.size() as f64 - app.db.num_idle() as f64).max(0.0)),
        ],
    );
    let (ok, refused): (Option<i64>, Option<i64>) =
        sqlx::query_as("SELECT SUM(CASE WHEN ok THEN 1 ELSE 0 END), SUM(CASE WHEN ok THEN 0 ELSE 1 END) FROM pushes WHERE at_ms > $1")
            .bind(since)
            .fetch_one(&app.db)
            .await
            .unwrap_or((None, None));
    family(
        &mut out,
        "oxyhub_pushes_last_day",
        "Reference updates pushed in the last 24 hours, by whether they were accepted.",
        "gauge",
        &[("result=\"accepted\"".into(), ok.unwrap_or(0) as f64), ("result=\"refused\"".into(), refused.unwrap_or(0) as f64)],
    );
    let backups: Vec<(String, String, i64, Option<i64>)> = sqlx::query_as(
        "SELECT owner, status, started_ms, finished_ms FROM backup_runs WHERE id IN (SELECT MAX(id) FROM backup_runs GROUP BY owner) ORDER BY owner",
    )
    .fetch_all(&app.db)
    .await
    .unwrap_or_default();
    family(
        &mut out,
        "oxyhub_backup_last_run_age_seconds",
        "Seconds since each owner's latest backup run started.",
        "gauge",
        &backups.iter().map(|(o, _, s, _)| (format!("owner=\"{}\"", label(o)), (now - s) as f64 / 1000.0)).collect::<Vec<_>>(),
    );
    family(
        &mut out,
        "oxyhub_backup_last_run_ok",
        "Whether each owner's latest finished backup run succeeded (1), was partial or failed (0); running ones are left out.",
        "gauge",
        &backups
            .iter()
            .filter(|(_, st, _, f)| f.is_some() && st != "running")
            .map(|(o, st, _, _)| (format!("owner=\"{}\"", label(o)), if st == "ok" { 1.0 } else { 0.0 }))
            .collect::<Vec<_>>(),
    );
    flow_timings(app, &mut out, now).await;
    if let Ok((waiting, dead)) = db::queue::counts(&app.db).await {
        family(
            &mut out,
            "oxyhub_queue_items",
            "Items in the durable work queue (#276): waiting (due, backing off or claimed) and dead (failed every attempt).",
            "gauge",
            &[("state=\"waiting\"".into(), waiting as f64), ("state=\"dead\"".into(), dead as f64)],
        );
    }
    http_families(&mut out);
    out
}

/// A finished job's runner, result, note, and when it was ready, started
/// and finished.
type JobTimes = (String, String, String, Option<i64>, Option<i64>, Option<i64>);

/// Bucket bounds for job times, in seconds.
const JOB_SECONDS: [f64; 11] = [1.0, 5.0, 15.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1200.0, 1800.0, 3600.0];

/// Flow's job times over the last day (oxygenflow#35), by runner: how long
/// jobs waited from ready to started (queue time) and ran (run time), and
/// how many finished, by result, and were reused (#265).
async fn flow_timings(app: &Shared, out: &mut String, now: i64) {
    let rows: Vec<JobTimes> = sqlx::query_as(
        "SELECT runner, status, error, COALESCE(ready_ms, queued_ms), started_ms, finished_ms FROM flow_jobs \
         WHERE finished_ms >= $1 AND status IN ('passed', 'failed', 'error', 'cancelled')",
    )
    .bind(now - 86_400_000)
    .fetch_all(&app.db)
    .await
    .unwrap_or_default();
    let mut queue: std::collections::BTreeMap<String, Vec<f64>> = Default::default();
    let mut run: std::collections::BTreeMap<String, Vec<f64>> = Default::default();
    let mut finished: std::collections::BTreeMap<(String, String), f64> = Default::default();
    let mut reused = 0.0;
    for (runner, status, error, ready, started, finished_ms) in &rows {
        if error.starts_with("reused from run #") {
            reused += 1.0;
            continue;
        }
        let r = if runner.is_empty() { "none".to_string() } else { runner.clone() };
        *finished.entry((r.clone(), status.clone())).or_default() += 1.0;
        if let (Some(ready), Some(started)) = (ready, started) {
            queue.entry(r.clone()).or_default().push((started - ready).max(0) as f64 / 1000.0);
        }
        if let (Some(started), Some(f)) = (started, finished_ms) {
            run.entry(r).or_default().push((f - started).max(0) as f64 / 1000.0);
        }
    }
    histogram(
        out,
        "oxyhub_flow_job_queue_seconds",
        "How long finished jobs of the last day waited from ready to started, by runner.",
        &queue,
    );
    histogram(out, "oxyhub_flow_job_run_seconds", "How long finished jobs of the last day ran, by runner.", &run);
    // Where their time went, as their runners reported (oxygenflow#35).
    let phases: Vec<String> = sqlx::query_scalar("SELECT phases FROM flow_jobs WHERE finished_ms >= $1 AND phases != ''")
        .bind(now - 86_400_000)
        .fetch_all(&app.db)
        .await
        .unwrap_or_default();
    let mut by_phase: std::collections::BTreeMap<String, Vec<f64>> = Default::default();
    for json in &phases {
        let m: std::collections::BTreeMap<String, u64> = serde_json::from_str(json).unwrap_or_default();
        for (phase, ms) in m {
            by_phase.entry(phase).or_default().push(ms as f64 / 1000.0);
        }
    }
    histogram_by(
        out,
        "oxyhub_flow_job_phase_seconds",
        "Where finished jobs of the last day spent their time, by phase: image (fetching it), prepare, run, finish.",
        "phase",
        &by_phase,
    );
    family(
        out,
        "oxyhub_flow_jobs_finished_last_day",
        "Jobs that finished in the last day, by runner and result (none: never placed).",
        "gauge",
        &finished.iter().map(|((r, st), n)| (format!("runner=\"{}\",status=\"{}\"", label(r), label(st)), *n)).collect::<Vec<_>>(),
    );
    family(
        out,
        "oxyhub_flow_jobs_reused_last_day",
        "Jobs of the last day that reused a pass on the same state instead of running (#265).",
        "gauge",
        &[(String::new(), reused)],
    );
}

/// A histogram family from samples (seconds) per runner.
fn histogram(out: &mut String, name: &str, help: &str, by_runner: &std::collections::BTreeMap<String, Vec<f64>>) {
    histogram_by(out, name, help, "runner", by_runner)
}

/// A histogram of job seconds, one series per value of label `key`.
fn histogram_by(out: &mut String, name: &str, help: &str, key: &str, by: &std::collections::BTreeMap<String, Vec<f64>>) {
    let _ = writeln!(out, "# HELP {name} {help}");
    let _ = writeln!(out, "# TYPE {name} histogram");
    for (runner, samples) in by {
        let l = format!("{key}=\"{}\"", label(runner));
        for b in JOB_SECONDS {
            let n = samples.iter().filter(|s| **s <= b).count();
            let _ = writeln!(out, "{name}_bucket{{{l},le=\"{b}\"}} {n}");
        }
        let _ = writeln!(out, "{name}_bucket{{{l},le=\"+Inf\"}} {}", samples.len());
        let _ = writeln!(out, "{name}_sum{{{l}}} {}", samples.iter().sum::<f64>());
        let _ = writeln!(out, "{name}_count{{{l}}} {}", samples.len());
    }
}

/// `GET /metrics`.
pub async fn metrics(State(app): State<Shared>, headers: HeaderMap) -> Response {
    let want = &app.config.server.metrics_token;
    if want.is_empty() {
        return (StatusCode::NOT_FOUND, "metrics are off: set [server] metrics_token in hub.toml\n").into_response();
    }
    let given = oxy_server::bearer_token(&headers).unwrap_or("");
    if !crate::auth::mac_equal(given, want) {
        return (StatusCode::UNAUTHORIZED, [(header::WWW_AUTHENTICATE, "Bearer")], "a bearer token is needed\n").into_response();
    }
    ([(header::CONTENT_TYPE, "text/plain; version=0.0.4; charset=utf-8"), (header::CACHE_CONTROL, "no-store")], render(&app).await)
        .into_response()
}

#[cfg(test)]
mod tests {

    #[test]
    fn job_times_are_cumulative_buckets_per_runner() {
        let mut by: std::collections::BTreeMap<String, Vec<f64>> = Default::default();
        by.insert("earth-1".into(), vec![0.5, 20.0, 4000.0]);
        let mut out = String::new();
        super::histogram(&mut out, "x_seconds", "help", &by);
        assert!(out.contains("x_seconds_bucket{runner=\"earth-1\",le=\"1\"} 1"), "{out}");
        assert!(out.contains("x_seconds_bucket{runner=\"earth-1\",le=\"30\"} 2"), "{out}");
        assert!(out.contains("x_seconds_bucket{runner=\"earth-1\",le=\"3600\"} 2"), "{out}");
        assert!(
            out.contains("x_seconds_bucket{runner=\"earth-1\",le=\"+Inf\"} 3") && out.contains("x_seconds_count{runner=\"earth-1\"} 3"),
            "{out}"
        );
    }

    #[test]
    fn route_classes_hold_no_names() {
        for (path, class) in [
            ("/", "page:home"),
            ("/static/hub.js", "static"),
            ("/api/v1/repos/andy/demo/issues", "api:repos"),
            ("/andy/demo/oxy/v1/negotiate", "protocol:negotiate"),
            ("/andy/demo/issue/3", "page:issue"),
            ("/andy/demo", "page:repository"),
            ("/explore", "page:explore"),
            ("/Andy_McCall", "page:other"),
            ("/log/tile/0/000", "tlog"),
        ] {
            assert_eq!(super::route_class(path), class, "{path}");
        }
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/hub/src/metrics.rs 92.5%

Keyboard
