hyperhive/swarm-controller/src/held_deploys.rs

//! The `swarm.deploys.held` gauge: how many agents have a deploy held back by
//! the build gate, exported on the jobq exporter. Grafana alerts when it stays
//! above zero (`hh-deploys-held`, `nix/host-modules/swarm-grafana.nix`).
//!
//! An agent counts when its latest `BuildMeta` node is `Failed`, once per agent
//! under one of two `reason` labels:
//! - `red`: `nix` exited with an error twice. Nothing retries it; a config merge
//!   builds again.
//! - `retrying`: the build gave no verdict or met its first `nix` error, so the
//!   ten-minute timer builds it again.
//!
//! A pending, running or finished build does not count, so the value falls to
//! zero when the builder answers.

use std::sync::{Arc, Mutex};

use hive_jobq::State;
use opentelemetry::KeyValue;
use opentelemetry::metrics::MeterProvider as _;
use opentelemetry_sdk::metrics::SdkMeterProvider;

use crate::jobq::{SwarmNodeKind, SwarmResourceKind};
use crate::meta_build;

/// The gauge's name. The Grafana rule selects it by this name.
pub(crate) const METRIC: &str = "swarm.deploys.held";

/// Held agents as `(red, retrying)`.
fn count(graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>) -> (u64, u64) {
    let mut latest = std::collections::BTreeMap::new();
    for n in graph.nodes() {
        if let SwarmNodeKind::BuildMeta { agent, .. } = &n.payload
            && latest
                .get(agent)
                .is_none_or(|l: &&hive_jobq::Node<_, _>| l.id < n.id)
        {
            latest.insert(agent, n);
        }
    }
    latest
        .into_values()
        .filter(|n| n.state == State::Failed)
        .fold((0, 0), |(red, retrying), n| {
            if n.error.as_deref().is_some_and(meta_build::is_retried) {
                (red, retrying + 1)
            } else {
                (red + 1, retrying)
            }
        })
}

/// Register the gauge on `provider`, reading `jobq` at each export. Nothing when
/// no exporter is configured.
pub(crate) fn register(
    provider: Option<&SdkMeterProvider>,
    jobq: &Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
) {
    let Some(provider) = provider else {
        return;
    };
    let jobq = Arc::clone(jobq);
    provider
        .meter("hyperhive.deploy_gate")
        .u64_observable_gauge(METRIC)
        .with_description("agents whose deploy the build gate is holding back, by reason")
        .with_callback(move |obs| {
            let (red, retrying) = count(
                jobq.lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .graph(),
            );
            obs.observe(red, &[KeyValue::new("reason", "red")]);
            obs.observe(retrying, &[KeyValue::new("reason", "retrying")]);
        })
        .build();
}

#[cfg(test)]
mod tests {
    use hive_jobq::scheduler::{Outcome, Scheduler};
    use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics};
    use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader};

    use super::*;

    type Jobq = Arc<Mutex<Scheduler<SwarmNodeKind, SwarmResourceKind>>>;

    /// A scheduler where each of `builds` (agent, outcome) ran once, in order.
    async fn after(builds: &[(&str, Outcome)]) -> Jobq {
        let jobq: Jobq = Arc::new(Mutex::new(Scheduler::new(
            hive_jobq::Graph::new(),
            hive_jobq::resources::ResourceTable::new(),
        )));
        for (agent, outcome) in builds {
            jobq.lock()
                .unwrap()
                .append(
                    SwarmNodeKind::BuildMeta {
                        agent: (*agent).to_owned(),
                        attempt: 0,
                    },
                    Vec::new(),
                    None,
                )
                .expect("insert");
            let outcome = outcome.clone();
            Scheduler::claim_next(&jobq, move |_, _, b| async move { (b, outcome) })
                .expect("runnable")
                .await
                .1
                .expect("no growth");
        }
        jobq
    }

    fn red() -> Outcome {
        Outcome::Failed("error: attribute missing".to_owned())
    }

    fn retrying() -> Outcome {
        Outcome::Failed(format!("{}no builder", meta_build::RETRY_PREFIX))
    }

    #[tokio::test]
    async fn a_failed_latest_build_holds_its_agent_by_reason() {
        let jobq = after(&[
            ("atlas", red()),
            ("argus", retrying()),
            ("iris", Outcome::Done),
        ])
        .await;
        assert_eq!(count(jobq.lock().unwrap().graph()), (1, 1));
    }

    /// Only an agent's latest build counts: a green build after a red one
    /// releases it, and a red one after a green one holds it.
    #[tokio::test]
    async fn only_the_latest_build_of_an_agent_counts() {
        let released = after(&[("atlas", red()), ("atlas", Outcome::Done)]).await;
        assert_eq!(count(released.lock().unwrap().graph()), (0, 0));
        let held = after(&[("atlas", Outcome::Done), ("atlas", red())]).await;
        assert_eq!(count(held.lock().unwrap().graph()), (1, 0));
    }

    /// The gauge reads the graph at export time and labels both reasons, zero
    /// included, so a rule sees the series before anything is held.
    #[tokio::test]
    async fn the_gauge_exports_both_reasons() {
        let jobq = after(&[("atlas", red())]).await;
        let exporter = InMemoryMetricExporter::default();
        let reader = PeriodicReader::builder(exporter.clone()).build();
        let provider = SdkMeterProvider::builder().with_reader(reader).build();
        register(Some(&provider), &jobq);
        provider.force_flush().expect("flush");
        let all = exporter.get_finished_metrics().expect("metrics");
        let mut points = Vec::new();
        for m in all
            .last()
            .into_iter()
            .flat_map(ResourceMetrics::scope_metrics)
            .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics)
            .filter(|m| m.name() == METRIC)
        {
            let AggregatedMetrics::U64(MetricData::Gauge(g)) = m.data() else {
                panic!("not a u64 gauge: {:?}", m.data());
            };
            for p in g.data_points() {
                let reason = p
                    .attributes()
                    .find(|kv| kv.key.as_str() == "reason")
                    .map(|kv| kv.value.to_string())
                    .expect("a reason");
                points.push((reason, p.value()));
            }
        }
        points.sort();
        assert_eq!(points, [("red".to_owned(), 1), ("retrying".to_owned(), 0)]);
    }
}