diff --git a/docs/4-implementation/scheduler/observability.md b/docs/4-implementation/scheduler/observability.md index f9a6faed..9377b994 100644 --- a/docs/4-implementation/scheduler/observability.md +++ b/docs/4-implementation/scheduler/observability.md @@ -580,6 +580,27 @@ store either a commit or a push anywhere as a row to count. | `hyperhive.vcs.commit.count` | — | counter | `repo` (`org/repo`) | | `hyperhive.vcs.push.count` | — | counter | `repo` (`org/repo`) | +Three more counters count what lands on each repo's default branch. A push +to any other ref adds nothing to them, and neither does deleting the default +branch. + +| Metric | Unit | Kind | Attributes | +| -------------------------------------------- | ---- | ------- | ------------------- | +| `hyperhive.vcs.default_branch.merges` | — | counter | `repo` (`org/repo`) | +| `hyperhive.vcs.default_branch.lines_added` | — | counter | `repo` (`org/repo`) | +| `hyperhive.vcs.default_branch.lines_deleted` | — | counter | `repo` (`org/repo`) | + +`merges` counts every such push. The line counters sum the per-commit stats +of the forge's compare of the push's `before...after`, so a line one commit +adds and a later one removes counts in both. A push that creates the branch +has no `before` and adds to `merges` only. When the compare fails, +`swarm-controller` logs a warning and the push adds to `merges` only. A +force-push counts the commits the compare returns, which are those reachable +from the new tip and not the old one; the commits it discards aren't +subtracted. The compare runs after the webhook has answered, so a slow forge +never delays a delivery. The counters start at zero on deploy and aren't +backfilled. The forge dashboard's content tab charts them per repo. + A push with zero commits (a branch delete, or a force-push that doesn't add new commits) still increments `push.count`; `commit.count` only advances when the delivery actually carries commits. Same enable signal (`OTEL_EXPORTER_OTLP_ENDPOINT`), fixed 60s cadence diff --git a/nix/host-modules/swarm-grafana/dashboards/forge.json b/nix/host-modules/swarm-grafana/dashboards/forge.json index e97bee54..aa2dc3d9 100644 --- a/nix/host-modules/swarm-grafana/dashboards/forge.json +++ b/nix/host-modules/swarm-grafana/dashboards/forge.json @@ -2115,6 +2115,228 @@ } } } + }, + "panel-24": { + "kind": "Panel", + "spec": { + "id": 24, + "title": "Lines added on default branches", + "description": "Lines added on each repo's default branch, per interval.", + "links": [], + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "query": { + "kind": "DataQuery", + "group": "prometheus", + "version": "v0", + "datasource": { + "name": "@datasourceUid@" + }, + "spec": { + "expr": "sum by (repo) (increase({__name__=\"hyperhive.vcs.default_branch.lines_added\", \"service.name\"=\"swarm-controller\"}[$__interval]))", + "legendFormat": "{{repo}}" + } + }, + "refId": "A", + "hidden": false + } + } + ], + "transformations": [], + "queryOptions": {} + } + }, + "vizConfig": { + "kind": "VizConfig", + "group": "timeseries", + "version": "", + "spec": { + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 0, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0 + } + }, + "overrides": [] + } + } + } + } + }, + "panel-25": { + "kind": "Panel", + "spec": { + "id": 25, + "title": "Lines deleted on default branches", + "description": "Lines deleted on each repo's default branch, per interval.", + "links": [], + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "query": { + "kind": "DataQuery", + "group": "prometheus", + "version": "v0", + "datasource": { + "name": "@datasourceUid@" + }, + "spec": { + "expr": "sum by (repo) (increase({__name__=\"hyperhive.vcs.default_branch.lines_deleted\", \"service.name\"=\"swarm-controller\"}[$__interval]))", + "legendFormat": "{{repo}}" + } + }, + "refId": "A", + "hidden": false + } + } + ], + "transformations": [], + "queryOptions": {} + } + }, + "vizConfig": { + "kind": "VizConfig", + "group": "timeseries", + "version": "", + "spec": { + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 0, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0 + } + }, + "overrides": [] + } + } + } + } + }, + "panel-26": { + "kind": "Panel", + "spec": { + "id": 26, + "title": "Average change size", + "description": "Lines added plus deleted per default-branch push, by repo.", + "links": [], + "data": { + "kind": "QueryGroup", + "spec": { + "queries": [ + { + "kind": "PanelQuery", + "spec": { + "query": { + "kind": "DataQuery", + "group": "prometheus", + "version": "v0", + "datasource": { + "name": "@datasourceUid@" + }, + "spec": { + "expr": "sum by (repo) (rate({__name__=\"hyperhive.vcs.default_branch.lines_added\", \"service.name\"=\"swarm-controller\"}[$__rate_interval]) + rate({__name__=\"hyperhive.vcs.default_branch.lines_deleted\", \"service.name\"=\"swarm-controller\"}[$__rate_interval])) / sum by (repo) (rate({__name__=\"hyperhive.vcs.default_branch.merges\", \"service.name\"=\"swarm-controller\"}[$__rate_interval]))", + "legendFormat": "{{repo}}" + } + }, + "refId": "A", + "hidden": false + } + } + ], + "transformations": [], + "queryOptions": {} + } + }, + "vizConfig": { + "kind": "VizConfig", + "group": "timeseries", + "version": "", + "spec": { + "options": { + "legend": { + "displayMode": "list", + "placement": "bottom", + "showLegend": true, + "calcs": [] + }, + "tooltip": { + "mode": "multi", + "sort": "desc" + } + }, + "fieldConfig": { + "defaults": { + "unit": "short", + "decimals": 0, + "color": { + "mode": "palette-classic" + }, + "custom": { + "drawStyle": "line", + "lineWidth": 1, + "fillOpacity": 0, + "showPoints": "never", + "spanNulls": false, + "axisSoftMin": 0 + } + }, + "overrides": [] + } + } + } + } } }, "layout": { @@ -2271,6 +2493,45 @@ "name": "panel-11" } } + }, + { + "kind": "GridLayoutItem", + "spec": { + "x": 0, + "y": 35, + "width": 12, + "height": 9, + "element": { + "kind": "ElementReference", + "name": "panel-24" + } + } + }, + { + "kind": "GridLayoutItem", + "spec": { + "x": 12, + "y": 35, + "width": 12, + "height": 9, + "element": { + "kind": "ElementReference", + "name": "panel-25" + } + } + }, + { + "kind": "GridLayoutItem", + "spec": { + "x": 0, + "y": 44, + "width": 24, + "height": 9, + "element": { + "kind": "ElementReference", + "name": "panel-26" + } + } } ] } diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 82122095..521e1a85 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -132,6 +132,28 @@ fn merge_base_of(commits: &[forgejo_api::structs::Commit]) -> Option { } } +/// 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| -> 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) -> Vec { @@ -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 { + 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::(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`). diff --git a/swarm-controller/src/vcs_metrics.rs b/swarm-controller/src/vcs_metrics.rs index cb877de4..4a78d626 100644 --- a/swarm-controller/src/vcs_metrics.rs +++ b/swarm-controller/src/vcs_metrics.rs @@ -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 { 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, push_count: Counter, + merges: Counter, + lines_added: Counter, + lines_deleted: Counter, } static INSTRUMENTS: OnceLock> = 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, +} + +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( + activity: &PushActivity, + compare: F, +) -> Option +where + F: FnOnce(String, String) -> Fut, + Fut: Future>, +{ + 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>, +) { + 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 { let endpoint = endpoint()?; let interval = EXPORT_INTERVAL; @@ -99,6 +209,18 @@ fn build() -> Option { .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, + ) -> Option { + 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)); + } +} diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index 7ba56c55..e0a016f9 100644 --- a/swarm-controller/src/webhook.rs +++ b/swarm-controller/src/webhook.rs @@ -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"); }