A runner's own metrics (refs #35) 4be5e4cd0af6

//! The runner's own metrics (oxygenflow#35, oxygenhub#212): a Prometheus

//! text page on an address of the runner's choosing, off unless

//! `metrics = "127.0.0.1:9464"` is set under `[runner]`. With

//! `metrics_token`, a scrape must send it as a bearer token.

use std::io::{BufRead, BufReader, Write};

use std::sync::atomic::{AtomicU64, Ordering::Relaxed};

use std::time::Instant;

/// Jobs this runner claimed.

pub static CLAIMS: AtomicU64 = AtomicU64::new(0);

/// Jobs declined before running (the preflight said no).

pub static DECLINED: AtomicU64 = AtomicU64::new(0);

/// Finished jobs by result.

pub static PASSED: AtomicU64 = AtomicU64::new(0);

pub static FAILED: AtomicU64 = AtomicU64::new(0);

pub static ERRORS: AtomicU64 = AtomicU64::new(0);

pub static CANCELLED: AtomicU64 = AtomicU64::new(0);

/// Milliseconds finished jobs ran, together.

pub static JOB_MS: AtomicU64 = AtomicU64::new(0);

/// Jobs running now.

pub static RUNNING: AtomicU64 = AtomicU64::new(0);

/// Count a job's end: by exit code, its error, and how long it ran.

pub fn finished(exit_code: Option<i32>, error: Option<&str>, duration_ms: u64, cancelled: bool) {

    let counter = if cancelled {

        &CANCELLED

    } else if error.is_some() {

        &ERRORS

    } else if exit_code == Some(0) {

        &PASSED

    } else {

        &FAILED

    };

    counter.fetch_add(1, Relaxed);

    JOB_MS.fetch_add(duration_ms, Relaxed);

}

/// The page.

pub fn render(name: &str, started: Instant) -> String {

    let n = name.replace('\\', "\\\\").replace('"', "\\\"");

    let mut out = String::new();

    let mut family = |metric: &str, help: &str, kind: &str, samples: &[(String, u64)]| {

        out.push_str(&format!("# HELP {metric} {help}\n# TYPE {metric} {kind}\n"));

        for (labels, v) in samples {

            let sep = if labels.is_empty() { "" } else { "," };

            out.push_str(&format!("{metric}{{runner=\"{n}\"{sep}{labels}}} {v}\n"));

        }

    };

    family("flow_runner_build_info", "This runner's version.", "gauge", &[(format!("version=\"{}\"", env!("CARGO_PKG_VERSION")), 1)]);

    family("flow_runner_uptime_seconds", "Seconds since the runner started.", "gauge", &[(String::new(), started.elapsed().as_secs())]);

    family("flow_runner_claims_total", "Jobs claimed from the hub.", "counter", &[(String::new(), CLAIMS.load(Relaxed))]);

    family("flow_runner_declined_total", "Claimed jobs declined before running.", "counter", &[(String::new(), DECLINED.load(Relaxed))]);

    family(

        "flow_runner_jobs_finished_total",

        "Finished jobs by result.",

        "counter",

        &[

            ("result=\"passed\"".into(), PASSED.load(Relaxed)),

            ("result=\"failed\"".into(), FAILED.load(Relaxed)),

            ("result=\"error\"".into(), ERRORS.load(Relaxed)),

            ("result=\"cancelled\"".into(), CANCELLED.load(Relaxed)),

        ],

    );

    family(

        "flow_runner_job_seconds_total",

        "Seconds finished jobs ran, together.",

        "counter",

        &[(String::new(), JOB_MS.load(Relaxed) / 1000)],

    );

    family("flow_runner_jobs_running", "Jobs running now.", "gauge", &[(String::new(), RUNNING.load(Relaxed))]);

    out

}

/// Serve the page at `addr` on a thread of its own; a failure to listen is

/// said and the runner goes on without it.

pub fn serve(addr: &str, token: Option<String>, name: String) {

    let listener = match std::net::TcpListener::bind(addr) {

        Ok(l) => l,

        Err(e) => {

            eprintln!("flow runner {name}: not serving metrics on {addr}: {e}");

            return;

        }

    };

    eprintln!("flow runner {name}: metrics on http://{addr}/metrics");

    let started = Instant::now();

    std::thread::spawn(move || {

        for stream in listener.incoming().flatten() {

            let _ = answer(stream, token.as_deref(), &name, started);

        }

    });

}

fn answer(mut stream: std::net::TcpStream, token: Option<&str>, name: &str, started: Instant) -> std::io::Result<()> {

    stream.set_read_timeout(Some(std::time::Duration::from_secs(5)))?;

    let mut reader = BufReader::new(stream.try_clone()?);

    let mut line = String::new();

    reader.read_line(&mut line)?;

    let path = line.split_whitespace().nth(1).unwrap_or("").to_string();

    let mut authorized = token.is_none();

    loop {

        let mut h = String::new();

        if reader.read_line(&mut h)? == 0 || h.trim().is_empty() {

            break;

        }

        if let (Some(t), Some((k, v))) = (token, h.split_once(':')) {

            if k.trim().eq_ignore_ascii_case("authorization") && v.trim() == format!("Bearer {t}") {

                authorized = true;

            }

        }

    }

    let (status, body) = match (path.as_str(), authorized) {

        ("/metrics", true) => ("200 OK", render(name, started)),

        ("/metrics", false) => ("401 Unauthorized", "a bearer token is needed\n".to_string()),

        _ => ("404 Not Found", "not found\n".to_string()),

    };

    write!(

        stream,

        "HTTP/1.1 {status}\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",

        body.len()

    )

}

#[cfg(test)]

mod tests {

    use super::*;

    use std::io::Read;

    #[test]

    fn a_scrape_needs_the_token_and_reads_the_counters() {

        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();

        let addr = l.local_addr().unwrap().to_string();

        drop(l);

        serve(&addr, Some("t0k".into()), "earth-1".into());

        CLAIMS.fetch_add(1, Relaxed);

        finished(Some(0), None, 2500, false);

        let get = |auth: &str| {

            let mut s = std::net::TcpStream::connect(&addr).unwrap();

            write!(s, "GET /metrics HTTP/1.1\r\nHost: x\r\n{auth}\r\n").unwrap();

            let mut body = String::new();

            s.read_to_string(&mut body).unwrap();

            body

        };

        assert!(get("").starts_with("HTTP/1.1 401"));

        let page = get("Authorization: Bearer t0k\r\n");

        assert!(page.starts_with("HTTP/1.1 200"), "{page}");

        assert!(page.contains("flow_runner_claims_total{runner=\"earth-1\"} ") && page.contains("result=\"passed\"}"), "{page}");

    }

}

- **A runner's own metrics (#35):** `metrics = "127.0.0.1:9464"` under `[runner]` serves Prometheus text at `/metrics`: claims, declines, finished jobs by result, the seconds they ran, the jobs running now, version and uptime, optionally behind `metrics_token`. Off unless set.

    /// Serve this runner's metrics at this address, e.g. `127.0.0.1:9464`

    /// (oxygenflow#35); off unless set.

    #[serde(default)]

    pub metrics: Option<String>,

    /// A bearer token a scrape of the metrics must send.

    #[serde(default)]

    pub metrics_token: Option<String>,

mod metrics;

    if let Some(addr) = &cfg.metrics {

        crate::metrics::serve(addr, cfg.metrics_token.clone(), name.clone());

    }

                    crate::metrics::CLAIMS.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                            crate::metrics::DECLINED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                        crate::metrics::RUNNING.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                        crate::metrics::RUNNING.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);

                        let cancelled = finish.error.as_deref() == Some(CANCELLED);

                        crate::metrics::finished(finish.exit_code, finish.error.as_deref(), finish.duration_ms, cancelled);

                        if !cancelled {

**Metrics** (oxygenflow#35). With `metrics = "127.0.0.1:9464"` under

`[runner]` (off unless set), a runner serves Prometheus text at

`/metrics` on that address, from a thread of its own: its version and

uptime, jobs claimed and declined, finished jobs by result (`passed`,

`failed`, `error`, `cancelled`), the seconds they ran together, and the

jobs running now, each labelled with the runner's name. With

`metrics_token`, a scrape must send it as a bearer token; without one,

bind it to an address only the scraper reaches. The hub's own `/metrics`

has the queue and run times of every runner's jobs.