Flow's job times in /metrics (refs oxygenflow#35, closes #212) 9e8cbfc3ad00

- **Flow's job times in /metrics (oxygenflow#35, #212):** `oxyhub_flow_job_queue_seconds` (ready to started) and `oxyhub_flow_job_run_seconds` histograms by runner, finished jobs by runner and result, reused passes, and the work queue's waiting and dead items, all over the last day, for the Grafana dashboard.

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

        );

    }

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

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

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

    let _ = writeln!(out, "# TYPE {name} histogram");

    for (runner, samples) in by_runner {

        let l = format!("runner=\"{}\"", 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());

    }

}

    #[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}"

        );

    }

    // Flow's job times (oxygenflow#35) and the work queue (#276).

    for family in [

        "# TYPE oxyhub_flow_job_queue_seconds histogram",

        "# TYPE oxyhub_flow_job_run_seconds histogram",

        "oxyhub_flow_jobs_reused_last_day 0",

        "oxyhub_queue_items{state=\"dead\"} 0",

    ] {

        assert!(body.contains(family), "{family}: {body}");

    }