hyperhive/swarm-ci-metrics/src/otlp.rs

//! The OTLP/HTTP JSON body for a batch of runs.
//!
//! Two gauges, one data point per run each, stamped with the run's `stopped`.
//! Victoriametrics stores an OTLP point at its `timeUnixNano`, so a run
//! finished months ago lands at that time. With `run` among the labels every
//! run is a series of its own, and a query reads them with `last_over_time`,
//! which a re-pushed identical point does not change.

use serde_json::{Value, json};

use crate::sample::{DURATION_METRIC, Labels, QUEUE_WAIT_METRIC, RunSamples};

/// The `service.name` resource attribute, which becomes a label.
pub const SERVICE_NAME: &str = "swarm-ci-metrics";

fn attribute(key: &str, value: &str) -> Value {
    json!({ "key": key, "value": { "stringValue": value } })
}

fn attributes(l: &Labels) -> Value {
    json!([
        attribute("repo", &l.repo),
        attribute("run", &l.run),
        attribute("workflow", &l.workflow),
        attribute("pr", &l.pr),
        attribute("sha", &l.sha),
        attribute("outcome", &l.outcome),
        attribute("event", &l.event),
    ])
}

/// OTLP JSON carries a 64-bit integer as a decimal string.
fn point(l: &Labels, at_unix_nanos: i128, value: f64) -> Value {
    json!({
        "attributes": attributes(l),
        "timeUnixNano": at_unix_nanos.to_string(),
        "asDouble": value,
    })
}

fn gauge(name: &str, points: &[Value]) -> Value {
    json!({ "name": name, "unit": "s", "gauge": { "dataPoints": points } })
}

/// The request body for `runs`.
pub fn body(runs: &[RunSamples]) -> Value {
    let durations: Vec<Value> = runs
        .iter()
        .map(|r| point(&r.labels, r.at_unix_nanos, r.duration_s))
        .collect();
    let waits: Vec<Value> = runs
        .iter()
        .filter_map(|r| r.queue_wait_s.map(|w| point(&r.labels, r.at_unix_nanos, w)))
        .collect();
    json!({
        "resourceMetrics": [{
            "resource": { "attributes": [attribute("service.name", SERVICE_NAME)] },
            "scopeMetrics": [{
                "scope": { "name": SERVICE_NAME },
                "metrics": [gauge(DURATION_METRIC, &durations), gauge(QUEUE_WAIT_METRIC, &waits)],
            }],
        }],
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn samples(sha: &str, queue_wait_s: Option<f64>) -> RunSamples {
        RunSamples {
            labels: Labels {
                repo: "hyperhive/hyperhive".to_owned(),
                run: "5638".to_owned(),
                workflow: "ci.yml".to_owned(),
                pr: "5372".to_owned(),
                sha: sha.to_owned(),
                outcome: "success".to_owned(),
                event: "pull_request".to_owned(),
            },
            at_unix_nanos: 1_792_163_610_000_000_000,
            duration_s: 1093.0,
            queue_wait_s,
        }
    }

    #[test]
    fn each_run_is_one_point_per_gauge_at_its_stopped_time() {
        let b = body(&[samples("aaa", Some(279.0)), samples("bbb", None)]);
        let rm = &b["resourceMetrics"][0];
        assert_eq!(
            rm["resource"]["attributes"][0],
            json!({"key": "service.name", "value": {"stringValue": "swarm-ci-metrics"}})
        );
        let metrics = &rm["scopeMetrics"][0]["metrics"];
        assert_eq!(metrics[0]["name"], DURATION_METRIC);
        assert_eq!(metrics[0]["unit"], "s");
        assert_eq!(metrics[1]["name"], QUEUE_WAIT_METRIC);

        let durations = metrics[0]["gauge"]["dataPoints"]
            .as_array()
            .expect("points");
        assert_eq!(durations.len(), 2);
        assert_eq!(durations[0]["timeUnixNano"], "1792163610000000000");
        assert_eq!(durations[0]["asDouble"], 1093.0);
        let labels: Vec<(&str, &str)> = durations[0]["attributes"]
            .as_array()
            .expect("attributes")
            .iter()
            .map(|a| {
                (
                    a["key"].as_str().expect("key"),
                    a["value"]["stringValue"].as_str().expect("value"),
                )
            })
            .collect();
        assert_eq!(
            labels,
            [
                ("repo", "hyperhive/hyperhive"),
                ("run", "5638"),
                ("workflow", "ci.yml"),
                ("pr", "5372"),
                ("sha", "aaa"),
                ("outcome", "success"),
                ("event", "pull_request"),
            ]
        );

        // The run with no wait has a duration and no wait point.
        let waits = metrics[1]["gauge"]["dataPoints"]
            .as_array()
            .expect("points");
        assert_eq!(waits.len(), 1);
        assert_eq!(waits[0]["asDouble"], 279.0);
        assert_eq!(waits[0]["timeUnixNano"], "1792163610000000000");
    }
}