Coverage / crates/hub/src/metrics.rs 93.1%

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;

}

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(),

}

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 {

    }

    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);

}

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}");

        }

}

fn label(v: &str) -> String {

    v.replace('\\', "\\\\").replace('"', "\\\"").replace('\n', "\\n")

}

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)

}

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

}

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 {

    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);

    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)],

}

fn histogram(out: &mut String, name: &str, help: &str, by_runner: &std::collections::BTreeMap<String, Vec<f64>>) {

    let _ = writeln!(out, "# HELP {name} {help}");