vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

push ci run metrics to victoriametrics, with a grafana board

Browse source 
A swarm-ci-metrics host unit on a 5-minute timer lists every forge
repo's Actions runs as the new read-only ci-metrics account and pushes
each finished run's duration and queue wait over OTLP, stamped at the
run's stop time. The first pass backfills from run 1. swarm-controller
mints the account's token into bao and keeps a ci-metrics team with
read on the Actions unit in every organisation.

Closes #5376
This commit is contained in:
atlas 2026-10-16 18:23:59 +02:00 • committed by mara
commit b072fbf4d6
36 changed files with 2849 additions and 56 deletions
Download patch file Download diff file Expand all files Collapse all files

157
swarm-ci-metrics/src/cursor.rs Normal file
View file

@ -0,0 +1,157 @@
//! Which runs of each repo are already pushed, kept in one file in the unit's
//! state directory.
//!
//! Per repo, a run number `low` below which every run is handled, and the
//! handled run numbers at or above it. A run is handled once it is pushed or
//! skipped. A run still going holds `low` at or below it, so a run that
//! finishes after a later one is still pushed.
//!
//! The file is replaced whole after every push the collector accepted, and
//! never before. A crash between that push and the write re-pushes those runs
//! on the next pass; nothing is pushed only to be forgotten, and nothing is
//! marked handled before it is pushed. A missing file is a cursor at run 0,
//! so the first pass is the backfill.
use std::collections::{BTreeMap, BTreeSet};
use std::io::Write as _;
use std::path::Path;
use anyhow::{Context as _, Result};
use serde::{Deserialize, Serialize};
/// One repo's cursor.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepoCursor {
/// Every run numbered below this is handled.
pub low: u64,
/// Handled runs numbered `low` or above.
pub handled: BTreeSet<u64>,
}
impl RepoCursor {
/// Whether the walk down the newest-first run list has passed every run
/// it still has to look at.
pub fn is_below(&self, run: u64) -> bool {
run < self.low
}
pub fn is_handled(&self, run: u64) -> bool {
self.is_below(run) || self.handled.contains(&run)
}
pub fn mark(&mut self, runs: impl IntoIterator<Item = u64>) {
self.handled.extend(runs);
}
/// Move `low` up once the pass has handled what it could: to the lowest
/// run still going, or past the newest run seen when none is. Never down.
pub fn advance(&mut self, lowest_pending: Option<u64>, newest_seen: Option<u64>) {
let to = match (lowest_pending, newest_seen) {
(Some(pending), _) => pending,
(None, Some(newest)) => newest + 1,
(None, None) => return,
};
self.low = self.low.max(to);
let low = self.low;
self.handled.retain(|&run| run >= low);
}
}
/// Every repo's cursor, keyed by `owner/name`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct State {
pub repos: BTreeMap<String, RepoCursor>,
}
impl State {
/// The state at `path`, or an empty one when there is no file.
pub fn load(path: &Path) -> Result<Self> {
match std::fs::read(path) {
Ok(bytes) => serde_json::from_slice(&bytes)
.with_context(|| format!("parsing the cursor at {}", path.display())),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(Self::default()),
Err(e) => Err(e).with_context(|| format!("reading the cursor at {}", path.display())),
}
}
/// Replace the file at `path`: a reader sees the old state or the new one,
/// never a mix.
pub fn save(&self, path: &Path) -> Result<()> {
let dir = path.parent().context("the cursor path has no directory")?;
let tmp = path.with_extension("tmp");
let mut file =
std::fs::File::create(&tmp).with_context(|| format!("creating {}", tmp.display()))?;
file.write_all(&serde_json::to_vec(self)?)?;
file.sync_all()?;
std::fs::rename(&tmp, path).with_context(|| format!("replacing {}", path.display()))?;
std::fs::File::open(dir)?.sync_all()?;
Ok(())
}
pub fn repo(&mut self, repo: &str) -> &mut RepoCursor {
self.repos.entry(repo.to_owned()).or_default()
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_missing_file_is_a_cursor_at_run_zero() {
let dir = tempfile::tempdir().expect("a scratch dir");
let mut state = State::load(&dir.path().join("cursor.json")).expect("no file is fine");
let c = state.repo("o/r");
assert_eq!(c.low, 0);
assert!(!c.is_handled(1));
}
#[test]
fn a_run_still_going_holds_low_and_one_finished_above_it_stays_handled() {
let mut c = RepoCursor::default();
// 7 is running; 8 and 9 finished first.
c.mark([5, 6, 8, 9]);
c.advance(Some(7), Some(9));
assert_eq!(c.low, 7);
assert!(c.is_handled(6));
assert!(!c.is_handled(7));
assert!(c.is_handled(8));
// 7 finishes.
c.mark([7]);
c.advance(None, Some(9));
assert_eq!(c.low, 10);
assert!(c.handled.is_empty());
}
#[test]
fn low_never_moves_down() {
let mut c = RepoCursor {
low: 10,
handled: BTreeSet::new(),
};
c.advance(Some(4), Some(12));
assert_eq!(c.low, 10);
c.advance(None, None);
assert_eq!(c.low, 10);
}
#[test]
fn a_saved_state_reads_back_and_leaves_no_temp_file() {
let dir = tempfile::tempdir().expect("a scratch dir");
let path = dir.path().join("cursor.json");
let mut state = State::default();
state.repo("o/r").mark([3, 4]);
state.repo("o/r").advance(Some(3), Some(4));
state.save(&path).expect("save");
assert_eq!(State::load(&path).expect("load"), state);
assert!(!path.with_extension("tmp").exists());
}
#[test]
fn a_corrupt_file_is_an_error_not_a_fresh_backfill() {
let dir = tempfile::tempdir().expect("a scratch dir");
let path = dir.path().join("cursor.json");
std::fs::write(&path, b"{").expect("write");
assert!(State::load(&path).is_err());
}
}

271
swarm-ci-metrics/src/main.rs Normal file
View file

@ -0,0 +1,271 @@
//! The swarm's CI run poller: one pass per start, run by a timer.
//!
//! Reads every repo with Actions that the `ci-metrics` forge account can see,
//! lists each repo's runs newest first down to its cursor, and pushes each
//! newly finished run as two OTLP gauges (`sample`) to the swarm collector,
//! which forwards them to the metrics store. With no cursor the walk reaches
//! run 1, so the first pass is the backfill; a page is read at most every
//! [`PAGE_DELAY`].
//!
//! The store identity comes from the `BAO_*` environment. Under it this reads
//! the forge token swarm-controller mints for the account, and the OIDC
//! client secret the collector's route for this service checks, once per pass.
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::time::Duration;
use anyhow::{Context as _, Result, bail};
use forgejo_api::structs::{ActionRun, ListActionRunsQuery, RepoSearchQuery};
use forgejo_api::{Auth, Forgejo};
use swarm_queue_client::stored_secret::StoredSecret;
use swarm_queue_client::{ClientSecret, QueueConfig};
use swarm_secret_client::SecretStore;
use swarm_secret_client::ci_metrics::{ClientSecret as StoredClientSecret, client_secret_path};
use swarm_secret_client::forge::Credential;
mod cursor;
mod otlp;
mod pass;
mod sample;
/// The cert-auth role this service logs in to the store as. `swarm-bao.nix`
/// writes it, matching the leaf's subject.
const CERT_ROLE: &str = "swarm-ci-metrics";
/// The environment prefix `swarm-ci-metrics.nix` sets.
const PREFIX: &str = "SWARM_CI_METRICS";
/// Runs and repos per forge page: the forge's own cap.
const PAGE: u32 = 50;
/// The pause before each further page of one listing.
const PAGE_DELAY: Duration = Duration::from_secs(1);
struct Config {
forge_url: url::Url,
/// `…/v1/metrics` on this service's collector route.
metrics_url: String,
queue: QueueConfig,
state_file: PathBuf,
}
#[tokio::main]
async fn main() -> Result<()> {
hive_log::init();
swarm_queue_client::install_crypto_provider();
let var = |name: &str| {
let full = format!("{PREFIX}_{name}");
std::env::var(&full).with_context(|| format!("{full} is unset"))
};
let client_id = var("OIDC_CLIENT_ID")?;
let secret = client_secret(&client_id).await?;
let reread_id = client_id.clone();
let secret = StoredSecret::new(client_id.clone(), secret, move || {
let client_id = reread_id.clone();
async move { client_secret(&client_id).await.map_err(Into::into) }
});
let state_dir = std::env::var("STATE_DIRECTORY").context("STATE_DIRECTORY is unset")?;
let cfg = Config {
forge_url: var("FORGE_URL")?.parse().context("parsing the forge URL")?,
metrics_url: format!("{}/v1/metrics", var("OTLP_ENDPOINT")?.trim_end_matches('/')),
queue: QueueConfig {
// `mint_token_for` reads only the token endpoint and the client.
url: String::new(),
token_endpoint: var("OIDC_TOKEN_ENDPOINT")?,
client_id,
client_secret: ClientSecret::Stored(secret),
ca_file: None,
},
state_file: PathBuf::from(state_dir).join("cursor.json"),
};
let forge = Forgejo::new(Auth::Token(&forge_token().await?), cfg.forge_url.clone())
.context("building the forge client")?;
let http = reqwest::Client::builder()
.timeout(Duration::from_secs(30))
.build()
.context("building the OTLP client")?;
let mut state = cursor::State::load(&cfg.state_file)?;
let repos = repos(&forge).await?;
tracing::info!(repos = repos.len(), "ci metrics: pass starting");
let mut failed = 0;
for repo in &repos {
if let Err(e) = repo_pass(&cfg, &forge, &http, &mut state, repo).await {
tracing::warn!(repo, error = %format!("{e:#}"), "ci metrics: repo pass failed");
failed += 1;
}
}
if failed > 0 {
bail!(
"{failed} of {} repos failed; their cursors stay where they were",
repos.len()
);
}
Ok(())
}
/// One repo: walk, plan, push, move the cursor.
async fn repo_pass(
cfg: &Config,
forge: &Forgejo,
http: &reqwest::Client,
state: &mut cursor::State,
repo: &str,
) -> Result<()> {
let runs = walk(forge, repo, state.repo(repo)).await?;
let plan = pass::plan(state, repo, &runs, time::OffsetDateTime::now_utc());
let mut skipped: BTreeMap<sample::Skip, usize> = BTreeMap::new();
for (_, why) in &plan.skip {
*skipped.entry(*why).or_default() += 1;
}
let pushing = plan.push.len();
let pending = plan.lowest_pending;
let bearer = if pushing > 0 {
swarm_queue_client::mint_token_for(&cfg.queue, Some(&cfg.queue.client_id), None)
.await
.context("minting a token for the collector")?
} else {
String::new()
};
pass::apply(
state,
repo,
plan,
async |samples| push(http, &cfg.metrics_url, &bearer, samples).await,
|s| s.save(&cfg.state_file),
)
.await?;
tracing::info!(
repo,
read = runs.len(),
pushed = pushing,
?skipped,
?pending,
"ci metrics: repo done"
);
Ok(())
}
/// Every repo the account sees that has Actions and is neither archived nor a
/// mirror, as `owner/name`.
async fn repos(forge: &Forgejo) -> Result<Vec<String>> {
let mut out = Vec::new();
for page in 1.. {
let found = forge
.repo_search(RepoSearchQuery::default())
.page(page)
.page_size(PAGE)
.send()
.await
.context("searching the forge's repos")?
.data
.unwrap_or_default();
let last = found.len() < PAGE as usize;
out.extend(found.into_iter().filter_map(|r| {
let wanted =
r.has_actions == Some(true) && r.archived != Some(true) && r.mirror != Some(true);
r.full_name.filter(|_| wanted)
}));
if last {
break;
}
tokio::time::sleep(PAGE_DELAY).await;
}
Ok(out)
}
/// The runs of `repo`, newest first, down to the first page that reaches below
/// the cursor.
async fn walk(forge: &Forgejo, repo: &str, cursor: &cursor::RepoCursor) -> Result<Vec<ActionRun>> {
let (owner, name) = repo
.split_once('/')
.with_context(|| format!("{repo:?} is not owner/name"))?;
let mut runs = Vec::new();
for page in 1.. {
let found = forge
.list_action_runs(owner, name, ListActionRunsQuery::default())
.page(page)
.page_size(PAGE)
.send()
.await
.with_context(|| format!("listing the runs of {repo}, page {page}"))?
.workflow_runs
.unwrap_or_default();
let done = found.len() < PAGE as usize
|| found.iter().any(|r| {
r.index_in_repo
.and_then(|n| u64::try_from(n).ok())
.is_some_and(|n| cursor.is_below(n))
});
runs.extend(found);
if done {
break;
}
tokio::time::sleep(PAGE_DELAY).await;
}
Ok(runs)
}
/// POST `samples` to the collector; an error unless it answers 2xx.
async fn push(
http: &reqwest::Client,
url: &str,
bearer: &str,
samples: &[sample::RunSamples],
) -> Result<()> {
let resp = http
.post(url)
.bearer_auth(bearer)
.json(&otlp::body(samples))
.send()
.await
.with_context(|| format!("pushing to {url}"))?;
let status = resp.status();
if !status.is_success() {
let body = resp.text().await.unwrap_or_default();
bail!(
"{url} answered {status}: {}",
body.chars().take(300).collect::<String>()
);
}
Ok(())
}
/// The `ci-metrics` account's forge token, which swarm-controller mints.
async fn forge_token() -> Result<String> {
let path = swarm_secret_client::ci_metrics::forge_token_path()?;
let stored: Option<Credential> = login()
.await?
.read_optional(&path)
.await
.with_context(|| format!("reading {path}"))?;
match stored {
Some(c) if !c.value.trim().is_empty() => Ok(c.value.trim().to_owned()),
_ => bail!("nothing at {path}: swarm-controller has not minted the ci-metrics forge token"),
}
}
/// This service's OIDC client secret, which the secret publisher writes.
async fn client_secret(client_id: &str) -> Result<String> {
let path = client_secret_path(client_id)?;
let stored: Option<StoredClientSecret> = login()
.await?
.read_optional(&path)
.await
.with_context(|| format!("reading {path}"))?;
match stored {
Some(s) if !s.value.trim().is_empty() => Ok(s.value.trim().to_owned()),
_ => bail!(
"nothing at {path}: swarm-secret-publish, on the host that runs authelia, has not \
published this service's client secret"
),
}
}
async fn login() -> Result<SecretStore> {
SecretStore::from_env(CERT_ROLE)
.await
.context("logging in to the secret store")
}

138
swarm-ci-metrics/src/otlp.rs Normal file
View file

@ -0,0 +1,138 @@
//! 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");
}
}

258
swarm-ci-metrics/src/pass.rs Normal file
View file

@ -0,0 +1,258 @@
//! One repo's pass: which of the runs read from the forge to push, and the
//! order of pushes and cursor writes that keeps a crash from losing a run.
use std::collections::BTreeMap;
use anyhow::Result;
use forgejo_api::structs::ActionRun;
use time::OffsetDateTime;
use crate::cursor::State;
use crate::sample::{RunSamples, Skip, Verdict, classify};
/// Runs per OTLP request: two points each, well under any body limit on the
/// way to the collector.
pub const CHUNK: usize = 100;
/// What one pass does with a repo's runs.
#[derive(Debug, Default)]
pub struct Plan {
/// Finished runs not yet pushed, oldest first, by run number.
pub push: Vec<(u64, RunSamples)>,
/// Finished runs that become no sample, by run number.
pub skip: Vec<(u64, Skip)>,
/// The lowest run number still going.
pub lowest_pending: Option<u64>,
/// The highest run number read.
pub newest_seen: Option<u64>,
}
/// Sort `runs` of `repo` against its cursor in `state`. A run listed twice
/// (the list shifts while it is paged) counts once; a run with no number is
/// left out.
pub fn plan(state: &State, repo: &str, runs: &[ActionRun], now: OffsetDateTime) -> Plan {
let cursor = state.repos.get(repo).cloned().unwrap_or_default();
let by_number: BTreeMap<u64, &ActionRun> = runs
.iter()
.filter_map(|r| {
let n = u64::try_from(r.index_in_repo?).ok()?;
Some((n, r))
})
.collect();
let mut plan = Plan {
newest_seen: by_number.keys().next_back().copied(),
..Plan::default()
};
for (&n, run) in &by_number {
if cursor.is_handled(n) {
continue;
}
match classify(repo, run, now) {
Verdict::Push(s) => plan.push.push((n, *s)),
Verdict::Skip(why) => plan.skip.push((n, why)),
Verdict::Pending => {
plan.lowest_pending.get_or_insert(n);
}
}
}
plan
}
/// Push `plan` for `repo` in chunks of [`CHUNK`] runs, marking each chunk in
/// `state` and saving it only after `push` accepted that chunk; then move the
/// cursor and save once more. The skips are marked with the first save.
///
/// A failed push returns before anything it carried is marked. A crash after
/// a push and before its save leaves those runs unmarked, so they go again.
pub async fn apply(
state: &mut State,
repo: &str,
plan: Plan,
mut push: impl AsyncFnMut(&[RunSamples]) -> Result<()>,
mut save: impl FnMut(&State) -> Result<()>,
) -> Result<()> {
state.repo(repo).mark(plan.skip.iter().map(|(n, _)| *n));
for chunk in plan.push.chunks(CHUNK) {
let samples: Vec<RunSamples> = chunk.iter().map(|(_, s)| s.clone()).collect();
push(&samples).await?;
state.repo(repo).mark(chunk.iter().map(|(n, _)| *n));
save(state)?;
}
state
.repo(repo)
.advance(plan.lowest_pending, plan.newest_seen);
save(state)
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
fn now() -> OffsetDateTime {
OffsetDateTime::parse(
"2026-10-16T18:00:00+02:00",
&time::format_description::well_known::Rfc3339,
)
.expect("a timestamp")
}
fn run(n: i64, status: &str) -> ActionRun {
let mut v = json!({
"index_in_repo": n,
"status": status,
"event": "pull_request",
"prettyref": format!("#{n}"),
"workflow_id": "ci.yml",
"commit_sha": format!("{n:040}"),
"created": "2026-10-16T16:50:38+02:00",
"started": "2026-10-16T16:55:17+02:00",
"stopped": "2026-10-16T17:13:30+02:00",
});
let obj = v.as_object_mut().expect("an object");
for key in ["updated", "html_url"] {
obj.insert(key.to_owned(), serde_json::Value::Null);
}
serde_json::from_value(v).expect("an ActionRun")
}
/// `count` finished runs numbered from 1, newest first as the forge lists.
fn history(count: i64) -> Vec<ActionRun> {
(1..=count).rev().map(|n| run(n, "success")).collect()
}
fn pushed_numbers(samples: &[RunSamples]) -> Vec<String> {
samples.iter().map(|s| s.labels.pr.clone()).collect()
}
#[tokio::test]
async fn the_first_pass_pushes_every_finished_run_and_the_next_pushes_none() {
let mut state = State::default();
let runs = history(250);
let p = plan(&state, "o/r", &runs, now());
assert_eq!(p.push.len(), 250);
let mut requests = 0;
apply(
&mut state,
"o/r",
p,
async |_| {
requests += 1;
Ok(())
},
|_| Ok(()),
)
.await
.expect("apply");
assert_eq!(requests, 3);
assert_eq!(state.repos["o/r"].low, 251);
let again = plan(&state, "o/r", &runs, now());
assert!(again.push.is_empty() && again.skip.is_empty());
}
#[tokio::test]
async fn a_failed_push_marks_nothing_it_carried_and_the_next_pass_resumes() {
let mut state = State::default();
let runs = history(250);
let mut saved = None;
let mut calls = 0;
let p = plan(&state, "o/r", &runs, now());
let err = apply(
&mut state,
"o/r",
p,
async |_| {
calls += 1;
if calls == 2 {
anyhow::bail!("the collector answered 503");
}
Ok(())
},
|s| {
saved = Some(s.clone());
Ok(())
},
)
.await;
assert!(err.is_err());
// What survives the crash is the last save: the first chunk.
let state = saved.expect("the first chunk was saved");
assert_eq!(state.repos["o/r"].handled.len(), CHUNK);
assert_eq!(state.repos["o/r"].low, 0);
let resumed = plan(&state, "o/r", &runs, now());
let numbers: Vec<u64> = resumed.push.iter().map(|(n, _)| *n).collect();
assert_eq!(numbers, (101..=250).collect::<Vec<_>>());
}
#[tokio::test]
async fn a_crash_between_a_push_and_its_save_pushes_those_runs_again() {
let mut state = State::default();
let runs = history(3);
let mut first = Vec::new();
let p = plan(&state, "o/r", &runs, now());
let err = apply(
&mut state,
"o/r",
p,
async |s| {
first.extend(pushed_numbers(s));
Ok(())
},
|_| Err(anyhow::anyhow!("disk full")),
)
.await;
assert!(err.is_err());
// The file on disk is still the empty state.
let on_disk = State::default();
let again = plan(&on_disk, "o/r", &runs, now());
let second: Vec<String> = again
.push
.iter()
.map(|(_, s)| s.labels.pr.clone())
.collect();
assert_eq!(first, second);
}
#[tokio::test]
async fn a_run_finishing_after_a_newer_one_is_still_pushed() {
let mut state = State::default();
let mut runs = vec![run(3, "success"), run(2, "running"), run(1, "success")];
let mut pushed = Vec::new();
let p = plan(&state, "o/r", &runs, now());
apply(
&mut state,
"o/r",
p,
async |s| {
pushed.extend(pushed_numbers(s));
Ok(())
},
|_| Ok(()),
)
.await
.expect("apply");
assert_eq!(pushed, ["1", "3"]);
assert_eq!(state.repos["o/r"].low, 2);
runs[1] = run(2, "failure");
let next = plan(&state, "o/r", &runs, now());
let numbers: Vec<u64> = next.push.iter().map(|(n, _)| *n).collect();
assert_eq!(numbers, [2]);
}
#[test]
fn skips_are_counted_not_pushed_and_a_relisted_run_counts_once() {
let state = State::default();
let runs = vec![
run(3, "cancelled"),
run(2, "success"),
run(2, "success"),
run(1, "skipped"),
];
let p = plan(&state, "o/r", &runs, now());
assert_eq!(p.push.len(), 1);
assert_eq!(p.skip, [(1, Skip::NotRun), (3, Skip::Cancelled)]);
assert_eq!(p.newest_seen, Some(3));
}
}

274
swarm-ci-metrics/src/sample.rs Normal file
View file

@ -0,0 +1,274 @@
//! One finished forge Actions run to the samples it becomes, or the reason it
//! becomes none.
//!
//! The span is `stopped − started`. The run's `duration` field is not used: it
//! is in nanoseconds, absent on some finished runs, and set on running ones.
use forgejo_api::structs::ActionRun;
use time::OffsetDateTime;
/// The run's wall time, `stopped − started`, in seconds.
pub const DURATION_METRIC: &str = "hyperhive.ci.run.duration";
/// The time the run waited for a runner, `started − created`, in seconds.
pub const QUEUE_WAIT_METRIC: &str = "hyperhive.ci.run.queue_wait";
/// How long a run that has not finished keeps the cursor from passing it.
/// Past this the run is counted as [`Skip::Abandoned`] and not looked at
/// again, so one run the forge never closes cannot make every pass re-read
/// the whole history above it.
pub const ABANDON_AFTER: time::Duration = time::Duration::days(1);
/// The label set both samples of a run carry.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Labels {
/// `owner/name`, as `hyperhive.vcs.*` spells it.
pub repo: String,
/// The run number. A rerun of one commit is a run of its own, so with
/// this label every run is a series of its own.
pub run: String,
/// The workflow file, e.g. `ci.yml`.
pub workflow: String,
/// The PR number for a PR run, otherwise the ref the run ran on.
pub pr: String,
/// The first 12 characters of the commit.
pub sha: String,
/// `success` or `failure`.
pub outcome: String,
/// The webhook event, e.g. `pull_request`.
pub event: String,
}
/// What one finished run is pushed as.
#[derive(Debug, Clone, PartialEq)]
pub struct RunSamples {
pub labels: Labels,
/// `stopped`, the timestamp of both samples.
pub at_unix_nanos: i128,
pub duration_s: f64,
/// `None` when the run has no `created`, or one after `started`.
pub queue_wait_s: Option<f64>,
}
/// Why a run is pushed as nothing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
pub enum Skip {
/// Cancelled: its span is not a run time.
Cancelled,
/// The forge skipped the run; nothing ran.
NotRun,
/// `started` absent or at the epoch, which the forge writes for a run
/// cancelled before it started.
NoStart,
/// `stopped` absent, or not after `started`.
BadSpan,
/// Not finished after [`ABANDON_AFTER`].
Abandoned,
}
/// What [`classify`] makes of one run.
#[derive(Debug, Clone, PartialEq)]
pub enum Verdict {
Push(Box<RunSamples>),
Skip(Skip),
/// Not finished yet: the cursor stays at or below it.
Pending,
}
/// The `pr` label: the number of a `#N` ref, else the ref as the forge
/// prints it (`main`, a branch, a tag).
pub fn pr_label(prettyref: &str) -> String {
match prettyref.strip_prefix('#') {
Some(n) if !n.is_empty() && n.bytes().all(|b| b.is_ascii_digit()) => n.to_owned(),
_ => prettyref.to_owned(),
}
}
/// What `run` of `repo` becomes, judged at `now`.
pub fn classify(repo: &str, run: &ActionRun, now: OffsetDateTime) -> Verdict {
let outcome = match run.status.as_deref() {
Some(s @ ("success" | "failure")) => s,
Some("cancelled") => return Verdict::Skip(Skip::Cancelled),
Some("skipped") => return Verdict::Skip(Skip::NotRun),
_ => {
return match run.created {
Some(created) if now - created > ABANDON_AFTER => Verdict::Skip(Skip::Abandoned),
_ => Verdict::Pending,
};
}
};
let Some(started) = run.started.filter(|s| s.unix_timestamp() > 0) else {
return Verdict::Skip(Skip::NoStart);
};
let Some(stopped) = run.stopped.filter(|s| *s > started) else {
return Verdict::Skip(Skip::BadSpan);
};
let queue_wait_s = run
.created
.filter(|c| *c <= started)
.map(|c| (started - c).as_seconds_f64());
let sha = run.commit_sha.as_deref().unwrap_or_default();
Verdict::Push(Box::new(RunSamples {
labels: Labels {
repo: repo.to_owned(),
run: run.index_in_repo.map(|n| n.to_string()).unwrap_or_default(),
workflow: run.workflow_id.clone().unwrap_or_default(),
pr: pr_label(run.prettyref.as_deref().unwrap_or_default()),
sha: sha.chars().take(12).collect(),
outcome: outcome.to_owned(),
event: run.event.clone().unwrap_or_default(),
},
at_unix_nanos: stopped.unix_timestamp_nanos(),
duration_s: (stopped - started).as_seconds_f64(),
queue_wait_s,
}))
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
pub(crate) fn run(v: serde_json::Value) -> ActionRun {
let mut v = v;
let obj = v.as_object_mut().expect("an object");
for key in ["created", "started", "stopped", "updated", "html_url"] {
obj.entry(key).or_insert(serde_json::Value::Null);
}
serde_json::from_value(v).expect("an ActionRun")
}
fn now() -> OffsetDateTime {
OffsetDateTime::parse(
"2026-10-16T18:00:00+02:00",
&time::format_description::well_known::Rfc3339,
)
.expect("a timestamp")
}
/// A finished PR run, shaped as the forge returns one.
fn finished(status: &str) -> serde_json::Value {
json!({
"index_in_repo": 5638,
"status": status,
"event": "pull_request",
"prettyref": "#7",
"workflow_id": "ci.yml",
"commit_sha": "d0ab2edd05bcce7a35ea15bbc2ef5a366fc93d18",
"created": "2026-10-16T16:50:38+02:00",
"started": "2026-10-16T16:55:17+02:00",
"stopped": "2026-10-16T17:13:30+02:00",
"duration": 1_093_000_000_000_i64,
})
}
#[test]
fn a_finished_run_is_its_span_and_its_wait_stamped_at_stopped() {
let Verdict::Push(s) = classify("hyperhive/hyperhive", &run(finished("success")), now())
else {
panic!("a finished run is pushed");
};
assert_eq!(
s.labels,
Labels {
repo: "hyperhive/hyperhive".to_owned(),
run: "5638".to_owned(),
workflow: "ci.yml".to_owned(),
pr: "7".to_owned(),
sha: "d0ab2edd05bc".to_owned(),
outcome: "success".to_owned(),
event: "pull_request".to_owned(),
}
);
// 16:55:17 → 17:13:30, not the `duration` field's 1093 s.
assert!((s.duration_s - 1093.0).abs() < f64::EPSILON);
// 16:50:38 → 16:55:17.
assert_eq!(s.queue_wait_s, Some(279.0));
// 2026-10-16T15:13:30Z.
assert_eq!(s.at_unix_nanos, 1_792_163_610_000_000_000);
}
#[test]
fn the_span_comes_from_the_timestamps_not_the_duration_field() {
let mut v = finished("failure");
v["duration"] = json!(5);
let Verdict::Push(s) = classify("o/r", &run(v), now()) else {
panic!("a failed run is pushed");
};
assert!((s.duration_s - 1093.0).abs() < f64::EPSILON);
assert_eq!(s.labels.outcome, "failure");
}
#[test]
fn cancelled_skipped_and_unstarted_runs_are_skips() {
assert_eq!(
classify("o/r", &run(finished("cancelled")), now()),
Verdict::Skip(Skip::Cancelled)
);
assert_eq!(
classify("o/r", &run(finished("skipped")), now()),
Verdict::Skip(Skip::NotRun)
);
let mut epoch = finished("failure");
epoch["started"] = json!("1970-01-01T01:00:00+01:00");
assert_eq!(
classify("o/r", &run(epoch), now()),
Verdict::Skip(Skip::NoStart)
);
let mut absent = finished("success");
absent["started"] = serde_json::Value::Null;
assert_eq!(
classify("o/r", &run(absent), now()),
Verdict::Skip(Skip::NoStart)
);
}
#[test]
fn a_span_that_does_not_move_forward_is_a_skip() {
let mut equal = finished("success");
equal["stopped"] = equal["started"].clone();
assert_eq!(
classify("o/r", &run(equal), now()),
Verdict::Skip(Skip::BadSpan)
);
let mut absent = finished("success");
absent["stopped"] = serde_json::Value::Null;
assert_eq!(
classify("o/r", &run(absent), now()),
Verdict::Skip(Skip::BadSpan)
);
}
#[test]
fn an_unfinished_run_waits_until_it_is_abandoned() {
let mut running = finished("running");
running["created"] = json!("2026-10-16T17:30:00+02:00");
assert_eq!(classify("o/r", &run(running), now()), Verdict::Pending);
let mut stale = finished("waiting");
stale["created"] = json!("2026-10-15T17:30:00+02:00");
assert_eq!(
classify("o/r", &run(stale), now()),
Verdict::Skip(Skip::Abandoned)
);
}
#[test]
fn a_created_after_started_drops_the_wait_not_the_run() {
let mut v = finished("success");
v["created"] = json!("2026-10-16T17:00:00+02:00");
let Verdict::Push(s) = classify("o/r", &run(v), now()) else {
panic!("the span is still good");
};
assert_eq!(s.queue_wait_s, None);
}
#[test]
fn the_pr_label_is_the_number_or_the_ref() {
assert_eq!(pr_label("#7"), "7");
assert_eq!(pr_label("main"), "main");
assert_eq!(pr_label("2100-hive-metric"), "2100-hive-metric");
assert_eq!(pr_label("#"), "#");
assert_eq!(pr_label("#12x"), "#12x");
assert_eq!(pr_label(""), "");
}
}