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

swarm-controller: count default-branch merges and lines per repo

Browse source
This commit is contained in:
atlas 2026-11-08 12:08:46 +01:00 • committed by mara
commit 2e23f3debf
3 changed files with 576 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

63
swarm-controller/src/forge.rs
View file

@ -132,6 +132,28 @@ fn merge_base_of(commits: &[forgejo_api::structs::Commit]) -> Option<String> {
}
}
/// Lines added and deleted, as [`Client::compare_line_stats`] sums them.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub struct LineStats {
pub added: u64,
pub deleted: u64,
}
/// `commits`' per-commit stats summed; a negative or missing figure is zero.
fn line_stats(commits: &[forgejo_api::structs::Commit]) -> LineStats {
let sum = |pick: fn(&forgejo_api::structs::CommitStats) -> Option<i64>| -> u64 {
commits
.iter()
.filter_map(|c| c.stats.as_ref().and_then(pick))
.map(|n| u64::try_from(n).unwrap_or(0))
.sum()
};
LineStats {
added: sum(|s| s.additions),
deleted: sum(|s| s.deletions),
}
}
/// `pulls` as the dropdown's rows, newest (highest number) first. A PR
/// without a usable number cannot be named, so it is left out.
fn open_pulls(pulls: Vec<forgejo_api::structs::PullRequest>) -> Vec<OpenPull> {
@ -1179,6 +1201,28 @@ impl Client {
Ok(open_pulls(pulls))
}
/// Lines added and deleted by the commits `before..after` of
/// `owner/repo`, summed over the per-commit stats the forge's compare
/// returns. A commit the forge sent no stats for counts as zero.
///
/// The sum is over commits, not a net diff of the two endpoints: a line
/// one commit adds and a later one removes counts once in each total.
pub async fn compare_line_stats(
&self,
owner: &str,
repo: &str,
before: &str,
after: &str,
) -> Result<LineStats> {
let basehead = format!("{before}...{after}");
let compare = self
.api
.call(|api| api.repo_compare_diff(owner, repo, &basehead).into_future())
.await
.with_context(|| format!("compare {basehead} in {owner}/{repo}"))?;
Ok(line_stats(compare.commits.as_deref().unwrap_or_default()))
}
/// Every open issue in `owner/repo`, each with its dependency-derived
/// `blocked` / `depended_on_by_count` pre-resolved server-side — the
/// whole job behind `GET /api/repos/{org}/{repo}/issue-report`. Doing
@ -2228,6 +2272,25 @@ impl Client {
#[cfg(test)]
mod tests {
use super::*;
/// Commit stats sum per side; a commit with none, or a negative figure,
/// adds zero.
#[test]
fn compare_stats_sum_across_commits() {
let commit = |stats: Option<(i64, i64)>| {
serde_json::from_value::<forgejo_api::structs::Commit>(serde_json::json!({
"stats": stats.map(|(a, d)| serde_json::json!({"additions": a, "deletions": d})),
}))
.expect("commit parses")
};
let got = line_stats(&[commit(Some((5, 2))), commit(None), commit(Some((-1, 4)))]);
assert_eq!(
got,
LineStats {
added: 5,
deleted: 6
}
);
}
/// The event list each swarm hook is registered with, as Forgejo names
/// the event types (`modules/webhook/type.go`).

231
swarm-controller/src/vcs_metrics.rs
View file

@ -4,10 +4,13 @@
//! That endpoint gives issue/comment/repo counts because Forgejo tracks
//! those as durable rows it can `SELECT COUNT(*)` on demand; a commit or a
//! push is not stored anywhere as a row to count, only observed in passing
//! as a webhook fires. So this is occurrence-driven, not poll-driven: two
//! sync counters recorded straight from [`record_push`], called once per
//! verified `push` delivery on [`crate::webhook::DeliveryKind::VcsActivity`]
//! — no periodic scan, nothing to refresh.
//! as a webhook fires. So this is occurrence-driven, not poll-driven: the
//! push and commit counters are recorded straight from [`record_push`],
//! called once per verified `push` delivery on
//! [`crate::webhook::DeliveryKind::VcsActivity`]. The default-branch merge
//! and line counters come from [`count_default_branch_push`], which asks the
//! forge to compare the push and runs after the delivery is answered.
//! Nothing is scanned periodically.
//!
//! Same SDK setup as `hive_jobq_metrics` (this daemon's other OTEL
//! exporter): blocking OTLP client on the `PeriodicReader`'s own background
@ -19,9 +22,12 @@
//! for why an occurrence, unlike a live gauge, has nothing to poll between
//! events.
use std::sync::OnceLock;
use std::future::Future;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use crate::forge::LineStats;
use anyhow::{Context, Result};
use opentelemetry::KeyValue;
use opentelemetry::metrics::{Counter, MeterProvider as _};
@ -32,11 +38,20 @@ use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
/// Export cadence. Matches every other OTEL exporter in this tree.
pub(crate) const EXPORT_INTERVAL: Duration = Duration::from_mins(1);
/// One push delivery, as parsed off the webhook body — just the two fields
/// One push delivery, as parsed off the webhook body — just the fields
/// this module needs, not a full mirror of Forgejo's push-event schema.
/// Kept separate from the wire parsing so a test can construct one without
/// a JSON literal.
pub(super) struct PushActivity {
/// The pushed ref, e.g. `refs/heads/main`.
pub(super) git_ref: String,
/// `repository.default_branch`; empty when the payload names none, in
/// which case no push counts as a default-branch push.
pub(super) default_branch: String,
/// The ref's tip before and after the push. All zeros means the ref did
/// not exist before (`before`) or no longer exists (`after`).
pub(super) before: String,
pub(super) after: String,
/// `repository.full_name` (`org/repo`) — the attribute every series
/// this module emits carries, so a dashboard can break activity down
/// per repo without a second join.
@ -54,7 +69,16 @@ pub(super) fn parse_push(body: &[u8]) -> Option<PushActivity> {
let v: serde_json::Value = serde_json::from_slice(body).ok()?;
let repo = v.get("repository")?.get("full_name")?.as_str()?.to_owned();
let commit_count = v.get("commits")?.as_array()?.len();
let text = |v: Option<&serde_json::Value>| {
v.and_then(serde_json::Value::as_str)
.unwrap_or_default()
.to_owned()
};
Some(PushActivity {
git_ref: text(v.get("ref")),
default_branch: text(v.get("repository").and_then(|r| r.get("default_branch"))),
before: text(v.get("before")),
after: text(v.get("after")),
repo,
commit_count: u64::try_from(commit_count).unwrap_or(u64::MAX),
})
@ -64,6 +88,9 @@ struct Instruments {
_provider: SdkMeterProvider,
commit_count: Counter<u64>,
push_count: Counter<u64>,
merges: Counter<u64>,
lines_added: Counter<u64>,
lines_deleted: Counter<u64>,
}
static INSTRUMENTS: OnceLock<Option<Instruments>> = OnceLock::new();
@ -83,6 +110,89 @@ pub(super) fn record_push(activity: &PushActivity) {
}
}
/// What one default-branch push adds to the counters.
#[derive(Debug, PartialEq, Eq)]
pub(super) struct DefaultBranchTally {
/// `None` when the lines are unknown: the push created the branch, or
/// the compare failed. Nothing is added to the line counters then.
pub(super) lines: Option<LineStats>,
}
fn is_null_sha(sha: &str) -> bool {
!sha.is_empty() && sha.bytes().all(|b| b == b'0')
}
/// The counts a push adds, or `None` when it did not move the repo's default
/// branch (another ref, a deleted branch, or a payload naming no default).
///
/// Every other push to the default branch counts one merge. Lines come from
/// `compare(before, after)`, except for a push that created the branch (no
/// `before` to compare against). A failed compare is logged and skips the
/// lines only.
///
/// A force-push counts what the compare returns: the commits reachable from
/// `after` and not from `before`, with their stats. The commits the push
/// discarded are not subtracted, so a rewrite of already-counted history
/// counts those lines a second time.
pub(super) async fn default_branch_tally<F, Fut>(
activity: &PushActivity,
compare: F,
) -> Option<DefaultBranchTally>
where
F: FnOnce(String, String) -> Fut,
Fut: Future<Output = Result<LineStats>>,
{
if activity.default_branch.is_empty()
|| activity.git_ref != format!("refs/heads/{}", activity.default_branch)
|| activity.after.is_empty()
|| is_null_sha(&activity.after)
{
return None;
}
if activity.before.is_empty() || is_null_sha(&activity.before) {
return Some(DefaultBranchTally { lines: None });
}
let lines = match compare(activity.before.clone(), activity.after.clone()).await {
Ok(lines) => Some(lines),
Err(e) => {
tracing::warn!(repo = %activity.repo, error = ?e, "vcs-metrics: compare failed; push counted without lines");
None
}
};
Some(DefaultBranchTally { lines })
}
/// Count `activity` on the default-branch counters, comparing through
/// `forge`. Runs after the webhook has answered; nothing here can fail it.
pub(super) async fn count_default_branch_push(
activity: PushActivity,
forge: Option<Arc<crate::forge::Client>>,
) {
let Some((owner, name)) = activity.repo.split_once('/') else {
return;
};
let tally = default_branch_tally(&activity, |before, after| async move {
let forge = forge.context("no forge client")?;
forge.compare_line_stats(owner, name, &before, &after).await
})
.await;
if let Some(tally) = tally {
record_default_branch(&activity.repo, &tally);
}
}
fn record_default_branch(repo: &str, tally: &DefaultBranchTally) {
let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
return;
};
let attrs = [KeyValue::new("repo", repo.to_owned())];
inst.merges.add(1, &attrs);
if let Some(lines) = tally.lines {
inst.lines_added.add(lines.added, &attrs);
inst.lines_deleted.add(lines.deleted, &attrs);
}
}
fn build() -> Option<Instruments> {
let endpoint = endpoint()?;
let interval = EXPORT_INTERVAL;
@ -99,6 +209,18 @@ fn build() -> Option<Instruments> {
.u64_counter("hyperhive.vcs.push.count")
.with_description("push deliveries observed, one per webhook event")
.build(),
merges: meter
.u64_counter("hyperhive.vcs.default_branch.merges")
.with_description("pushes that moved a repo's default branch")
.build(),
lines_added: meter
.u64_counter("hyperhive.vcs.default_branch.lines_added")
.with_description("lines added by commits pushed to a repo's default branch")
.build(),
lines_deleted: meter
.u64_counter("hyperhive.vcs.default_branch.lines_deleted")
.with_description("lines deleted by commits pushed to a repo's default branch")
.build(),
_provider: provider,
})
}
@ -279,3 +401,100 @@ mod tests {
assert!(parse_push(b"not json at all").is_none());
}
}
#[cfg(test)]
mod default_branch_tests {
use super::{DefaultBranchTally, PushActivity, default_branch_tally};
use crate::forge::LineStats;
const SHA_A: &str = "1111111111111111111111111111111111111111";
const SHA_B: &str = "2222222222222222222222222222222222222222";
const ZEROS: &str = "0000000000000000000000000000000000000000";
fn push(git_ref: &str, before: &str, after: &str) -> PushActivity {
PushActivity {
git_ref: git_ref.to_owned(),
default_branch: "main".to_owned(),
before: before.to_owned(),
after: after.to_owned(),
repo: "hyperhive/hyperhive".to_owned(),
commit_count: 1,
}
}
async fn tally(
activity: &PushActivity,
compare: anyhow::Result<LineStats>,
) -> Option<DefaultBranchTally> {
default_branch_tally(activity, |_, _| async move { compare }).await
}
#[tokio::test]
async fn a_default_branch_push_counts_the_merge_and_the_compared_lines() {
let lines = LineStats {
added: 12,
deleted: 3,
};
let got = tally(&push("refs/heads/main", SHA_A, SHA_B), Ok(lines)).await;
assert_eq!(got, Some(DefaultBranchTally { lines: Some(lines) }));
}
#[tokio::test]
async fn a_push_to_another_branch_counts_nothing() {
let lines = LineStats {
added: 12,
deleted: 3,
};
let got = tally(&push("refs/heads/feature", SHA_A, SHA_B), Ok(lines)).await;
assert_eq!(got, None);
}
#[tokio::test]
async fn deleting_the_default_branch_counts_nothing() {
let got = tally(
&push("refs/heads/main", SHA_A, ZEROS),
Ok(LineStats::default()),
)
.await;
assert_eq!(got, None);
}
#[tokio::test]
async fn a_new_default_branch_counts_the_merge_and_no_lines() {
let called = std::sync::atomic::AtomicBool::new(false);
let got = default_branch_tally(&push("refs/heads/main", ZEROS, SHA_B), |_, _| async {
called.store(true, std::sync::atomic::Ordering::SeqCst);
Ok(LineStats::default())
})
.await;
assert_eq!(got, Some(DefaultBranchTally { lines: None }));
assert!(!called.load(std::sync::atomic::Ordering::SeqCst));
}
#[tokio::test]
async fn a_compare_error_counts_the_merge_and_skips_the_lines() {
let got = tally(
&push("refs/heads/main", SHA_A, SHA_B),
Err(anyhow::anyhow!("forge answered 500")),
)
.await;
assert_eq!(got, Some(DefaultBranchTally { lines: None }));
}
#[test]
fn push_fields_parse_off_the_payload() {
let body = br#"{
"ref": "refs/heads/main",
"before": "1111111111111111111111111111111111111111",
"after": "2222222222222222222222222222222222222222",
"repository": { "full_name": "a/b", "default_branch": "main" },
"commits": []
}"#;
let a = super::parse_push(body).expect("parses");
assert_eq!(
(a.git_ref.as_str(), a.default_branch.as_str()),
("refs/heads/main", "main")
);
assert_eq!((a.before.as_str(), a.after.as_str()), (SHA_A, SHA_B));
}
}

6
swarm-controller/src/webhook.rs
View file

@ -641,6 +641,12 @@ pub(super) async fn post_webhook_forge(
DeliveryKind::VcsActivity => {
if let Some(activity) = crate::vcs_metrics::parse_push(&body) {
crate::vcs_metrics::record_push(&activity);
// The compare runs after the delivery is answered, so a slow
// or failing forge never delays or fails the webhook.
let forge = state.forge.get().map(Arc::clone);
tokio::spawn(crate::vcs_metrics::count_default_branch_push(
activity, forge,
));
} else {
tracing::warn!("webhook: vcs-activity delivery did not parse as a push payload");
}