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`, or when its
//! latest `TriggerDeploy` is `Skipped` behind a `CopyAtticFacts` that `Failed`.
//! It counts once, under one of two `reason` labels:
//! - `failed`: `nix` failed, its timeout included.
//! - `unavailable`: the build got no verdict (no builder, a forge error, no
//!   `internal/meta` head, a missing read grant, a closure the builder built but
//!   didn't push to the cache), or the cache facts couldn't be copied to the hive.
//!
//! The ten-minute timer declares either again. A pending, running or finished
//! build, or a deploy past its copy, does not count, so the value falls to zero
//! when the retry succeeds.

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

/// The latest node of each agent whose payload `pick` returns the agent of.
fn latest_per_agent(
    graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
    pick: impl Fn(&SwarmNodeKind) -> Option<&String>,
) -> std::collections::BTreeMap<&String, &hive_jobq::Node<SwarmNodeKind, SwarmResourceKind>> {
    let mut latest = std::collections::BTreeMap::new();
    for n in graph.nodes() {
        if let Some(agent) = pick(&n.payload)
            && latest
                .get(agent)
                .is_none_or(|l: &&hive_jobq::Node<_, _>| l.id < n.id)
        {
            latest.insert(agent, n);
        }
    }
    latest
}

/// Held agents as `(failed, unavailable)`.
fn count(graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>) -> (u64, u64) {
    let builds = latest_per_agent(graph, |p| match p {
        SwarmNodeKind::BuildMeta { agent, .. } => Some(agent),
        _ => None,
    });
    let deploys = latest_per_agent(graph, |p| match p {
        SwarmNodeKind::TriggerDeploy { agent, .. } => Some(agent),
        _ => None,
    });
    let copy_failed = |deploy: &hive_jobq::Node<SwarmNodeKind, SwarmResourceKind>| {
        deploy.state == State::Skipped
            && deploy.deps.iter().any(|d| match d {
                hive_jobq::Dep::Node { id, .. } => graph.node(*id).is_some_and(|dep| {
                    matches!(dep.payload, SwarmNodeKind::CopyAtticFacts { .. })
                        && dep.state == State::Failed
                }),
                hive_jobq::Dep::Resource { .. } => false,
            })
    };
    let (mut failed, mut unavailable) = (0, 0);
    for (agent, build) in &builds {
        if build.state == State::Failed {
            if build
                .error
                .as_deref()
                .is_some_and(meta_build::is_no_verdict)
            {
                unavailable += 1;
            } else {
                failed += 1;
            }
        } else if deploys.get(agent).is_some_and(|d| copy_failed(d)) {
            unavailable += 1;
        }
    }
    (failed, unavailable)
}

/// 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 (failed, unavailable) = count(
                jobq.lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .graph(),
            );
            obs.observe(failed, &[KeyValue::new("reason", "failed")]);
            obs.observe(unavailable, &[KeyValue::new("reason", "unavailable")]);
        })
        .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(),
                        pr: None,
                    },
                    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 failed() -> Outcome {
        Outcome::Failed("error: attribute missing".to_owned())
    }

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

    #[tokio::test]
    async fn a_failed_latest_build_holds_its_agent_by_reason() {
        let jobq = after(&[
            ("atlas", failed()),
            ("argus", unavailable()),
            ("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", failed()), ("atlas", Outcome::Done)]).await;
        assert_eq!(count(released.lock().unwrap().graph()), (0, 0));
        let held = after(&[("atlas", Outcome::Done), ("atlas", failed())]).await;
        assert_eq!(count(held.lock().unwrap().graph()), (1, 0));
    }

    /// A scheduler where one declaration of `agent`'s build, facts copy and
    /// deploy ran, the build and the copy ending as given.
    async fn declared(jobq: &Jobq, agent: &str, build: Outcome, copy: Outcome) {
        jobq.lock()
            .unwrap()
            .insert_job(None, |b| {
                let build = b.node(SwarmNodeKind::BuildMeta {
                    agent: agent.to_owned(),
                    pr: None,
                });
                let copy = b.node(SwarmNodeKind::CopyAtticFacts {
                    hive: "pr1ma".to_owned(),
                });
                let deploy = b
                    .node(SwarmNodeKind::TriggerDeploy {
                        hive: "pr1ma".to_owned(),
                        agent: agent.to_owned(),
                        rev: None,
                    })
                    .after_ok(build)
                    .after_ok(copy);
                vec![deploy.guid()]
            })
            .expect("insert");
        while let Some(runner) = Scheduler::claim_next(jobq, {
            let (build, copy) = (build.clone(), copy.clone());
            move |_, kind, b| async move {
                let outcome = match kind {
                    SwarmNodeKind::BuildMeta { .. } => build,
                    SwarmNodeKind::CopyAtticFacts { .. } => copy,
                    _ => Outcome::Done,
                };
                (b, outcome)
            }
        }) {
            runner.await.1.expect("no growth");
        }
    }

    fn empty_jobq() -> Jobq {
        Arc::new(Mutex::new(Scheduler::new(
            hive_jobq::Graph::new(),
            hive_jobq::resources::ResourceTable::new(),
        )))
    }

    /// A deploy the facts copy held back counts as unavailable, and a green
    /// build with a green copy does not (the control that the deploy runs).
    #[tokio::test]
    async fn a_failed_facts_copy_holds_its_agent_as_unavailable() {
        let jobq = empty_jobq();
        declared(
            &jobq,
            "atlas",
            Outcome::Done,
            Outcome::Failed("denied".into()),
        )
        .await;
        declared(&jobq, "argus", Outcome::Done, Outcome::Done).await;
        assert_eq!(count(jobq.lock().unwrap().graph()), (0, 1));
    }

    /// An agent whose build and copy both failed counts once, under the build's
    /// reason; a retry that goes through releases it.
    #[tokio::test]
    async fn a_failed_build_and_copy_count_once_and_a_good_retry_releases() {
        let jobq = empty_jobq();
        declared(&jobq, "atlas", failed(), Outcome::Failed("denied".into())).await;
        assert_eq!(count(jobq.lock().unwrap().graph()), (1, 0));
        declared(
            &jobq,
            "atlas",
            Outcome::Done,
            Outcome::Failed("denied".into()),
        )
        .await;
        assert_eq!(count(jobq.lock().unwrap().graph()), (0, 1));
        declared(&jobq, "atlas", Outcome::Done, Outcome::Done).await;
        assert_eq!(count(jobq.lock().unwrap().graph()), (0, 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", failed())]).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,
            [("failed".to_owned(), 1), ("unavailable".to_owned(), 0)]
        );
    }
}