swarm-controller: emit vcs commit/push otel counters from an instance-wide forge webhook

Cargo.lock

@@ -4582,6 +4582,9 @@ dependencies = [
  "hive-jobq-wire",
  "hive-types",
  "hmac 0.13.0",
+ "opentelemetry",
+ "opentelemetry-otlp",
+ "opentelemetry_sdk",
  "problem_details",
  "reqwest",
  "serde",

swarm-controller/Cargo.toml

@@ -44,6 +44,14 @@ hive-jobq-wire.workspace = true
 # (e.g. hive-c0re, for its own per-hive job graph) doesn't have to depend on
 # this whole binary to reuse it.
 hive-jobq-metrics.workspace = true
+# Direct OTEL SDK use in `vcs_metrics.rs` — sync counters recorded off
+# webhook deliveries, a different shape from `hive-jobq-metrics`'s
+# observable-gauge rollup, so it isn't a fit for that crate's API and lives
+# here instead. Same three crates that pairing already pulls in transitively,
+# named directly since this module builds its own `SdkMeterProvider`.
+opentelemetry.workspace = true
+opentelemetry_sdk.workspace = true
+opentelemetry-otlp.workspace = true
 # The forge webhook HMAC (`webhook.rs`). Kept in this crate rather than
 # shared with hive-c0re's equivalent: c0re's copy is scheduled to be deleted
 # with its webhook routes once registration moves here, so the second holder

swarm-controller/src/forge.rs

@@ -557,6 +557,15 @@ impl Client {
                     },
                 ),
                 DeliveryKind::ConfigPr => ("pull_request", HookScope::Org { org: CONFIG_ORG }),
+                // Instance-wide: commit/push activity is not scoped to one
+                // org (unlike the two above) — it needs observing across
+                // every repo the forge hosts, per the "no metric exists for
+                // commits or pushes" tracker issue. Forgejo's
+                // "global (system) webhook" (`admin_create_hook`) is exactly
+                // this — the only scope in this API that fires for a repo
+                // in an org created after this hook was registered, with no
+                // per-org registration to keep in sync as orgs come and go.
+                DeliveryKind::VcsActivity => ("push", HookScope::Instance),
             };
             self.ensure_hook(&scope, &target_url, event, secret)
                 .await
@@ -624,8 +633,18 @@ impl Client {
 /// theirs — a hook on the wrong scope would never fire, and forgejo would
 /// report that as a perfectly healthy hook with no deliveries.
 enum HookScope<'a> {
-    Repo { org: &'a str, repo: &'a str },
-    Org { org: &'a str },
+    Repo {
+        org: &'a str,
+        repo: &'a str,
+    },
+    Org {
+        org: &'a str,
+    },
+    /// Forgejo's "global (system) webhook" — fires for every repo in the
+    /// instance, in every org, present or future. The `admin_*` API
+    /// namespace; needs the same `write:admin` scope
+    /// `Client::ensure_agent_user` already requires on this token.
+    Instance,
 }
 
 impl HookScope<'_> {
@@ -634,6 +653,7 @@ impl HookScope<'_> {
         let hooks = match self {
             Self::Repo { org, repo } => api.repo_list_hooks(org, repo).all().await?,
             Self::Org { org } => api.org_list_hooks(org).send().await?,
+            Self::Instance => api.admin_list_hooks().await?,
         };
         Ok(hooks
             .iter()
@@ -645,6 +665,7 @@ impl HookScope<'_> {
         match self {
             Self::Repo { org, repo } => api.repo_create_hook(org, repo, hook).await.map(drop),
             Self::Org { org } => api.org_create_hook(org, hook).await.map(drop),
+            Self::Instance => api.admin_create_hook(hook).await.map(drop),
         }
     }
 }

swarm-controller/src/main.rs

@@ -44,6 +44,7 @@ mod auth;
 mod config_pr;
 mod forge;
 mod status;
+mod vcs_metrics;
 mod webhook;
 
 /// Node payload for the swarm-level job graph. Named `Swarm*` rather than

swarm-controller/src/vcs_metrics.rs

@@ -0,0 +1,238 @@
+//! OTEL export of commit/push activity across every forge repo in the
+//! instance — the metric forgejo's own native `/metrics` endpoint (see
+//! `docs/forge.md`'s note on the survey that shipped it) does not carry.
+//! 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.
+//!
+//! Same SDK setup as `hive_jobq_metrics` (this daemon's other OTEL
+//! exporter): blocking OTLP client on the `PeriodicReader`'s own background
+//! thread, `service.name` + `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` for
+//! the resource, `OTEL_EXPORTER_OTLP_ENDPOINT` as the sole enable signal,
+//! `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` for cadence. Recording is
+//! synchronous (`counter.add`, called at the event site) rather than
+//! observable/polled — see `hive-agent::otel_turn_metrics`'s doc comment
+//! for why an occurrence, unlike a live gauge, has nothing to poll between
+//! events.
+
+use std::sync::OnceLock;
+use std::time::Duration;
+
+use anyhow::{Context, Result};
+use opentelemetry::KeyValue;
+use opentelemetry::metrics::{Counter, MeterProvider as _};
+use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig};
+use opentelemetry_sdk::Resource;
+use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
+
+/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is
+/// unset. Matches every other OTEL exporter in this tree.
+const DEFAULT_INTERVAL: Duration = Duration::from_mins(1);
+
+/// One push delivery, as parsed off the webhook body — just the two 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 {
+    /// `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.
+    pub(super) repo: String,
+    /// `commits.len()` from the payload. A push with zero commits is real
+    /// (a branch delete, a force-push moving the ref backward) and still
+    /// counted as a push — just with a commit delta of zero, not skipped.
+    pub(super) commit_count: u64,
+}
+
+/// Parse a Forgejo `push` webhook body into a [`PushActivity`]. `None` on
+/// anything unparseable — a malformed body is logged and dropped by the
+/// caller, not a reason to fail the delivery Forgejo already got a 200 for.
+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();
+    Some(PushActivity {
+        repo,
+        commit_count: u64::try_from(commit_count).unwrap_or(u64::MAX),
+    })
+}
+
+struct Instruments {
+    _provider: SdkMeterProvider,
+    commit_count: Counter<u64>,
+    push_count: Counter<u64>,
+}
+
+static INSTRUMENTS: OnceLock<Option<Instruments>> = OnceLock::new();
+
+/// Record one push delivery. A no-op when OTEL isn't configured
+/// (`OTEL_EXPORTER_OTLP_ENDPOINT` unset) — the same graceful-absence shape
+/// every other exporter in this tree uses, so a deployment with no
+/// collector pays nothing beyond the one-time `OnceLock` check.
+pub(super) fn record_push(activity: &PushActivity) {
+    let Some(inst) = INSTRUMENTS.get_or_init(build).as_ref() else {
+        return;
+    };
+    let attrs = [KeyValue::new("repo", activity.repo.clone())];
+    inst.push_count.add(1, &attrs);
+    if activity.commit_count > 0 {
+        inst.commit_count.add(activity.commit_count, &attrs);
+    }
+}
+
+fn build() -> Option<Instruments> {
+    let endpoint = endpoint()?;
+    let interval = interval();
+    match build_provider(interval) {
+        Ok(provider) => {
+            tracing::info!(%endpoint, ?interval, "otel vcs-metrics: exporter enabled");
+            let meter = provider.meter("hyperhive.vcs");
+            Some(Instruments {
+                commit_count: meter
+                    .u64_counter("hyperhive.vcs.commit.count")
+                    .with_description("commits observed across every forge push delivery")
+                    .build(),
+                push_count: meter
+                    .u64_counter("hyperhive.vcs.push.count")
+                    .with_description("push deliveries observed, one per webhook event")
+                    .build(),
+                _provider: provider,
+            })
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "otel vcs-metrics: exporter init failed");
+            None
+        }
+    }
+}
+
+fn build_provider(interval: Duration) -> Result<SdkMeterProvider> {
+    // Same http/json, endpoint-from-env-only construction as
+    // `hive_jobq_metrics::build_provider` — see that module's doc comment
+    // for why `with_endpoint` is deliberately never called here.
+    let exporter = MetricExporter::builder()
+        .with_http()
+        .with_protocol(Protocol::HttpJson)
+        .build()
+        .context("build OTLP metric exporter")?;
+    let reader = PeriodicReader::builder(exporter)
+        .with_interval(interval)
+        .build();
+    Ok(SdkMeterProvider::builder()
+        .with_reader(reader)
+        .with_resource(resource())
+        .build())
+}
+
+/// `service.name = swarm-controller` plus whatever the operator set in
+/// `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` — same channel
+/// `hive_jobq_metrics::resource` reads, since both live in this one binary.
+fn resource() -> Resource {
+    let mut builder = Resource::builder().with_service_name("swarm-controller");
+    for (k, v) in resource_attributes() {
+        builder = builder.with_attribute(KeyValue::new(k, v));
+    }
+    builder.build()
+}
+
+fn endpoint() -> Option<String> {
+    std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
+        .ok()
+        .map(|s| s.trim().to_owned())
+        .filter(|s| !s.is_empty())
+}
+
+fn interval() -> Duration {
+    std::env::var("HYPERHIVE_OTEL_METRIC_INTERVAL_MS")
+        .ok()
+        .and_then(|s| s.trim().parse::<u64>().ok())
+        .filter(|ms| *ms > 0)
+        .map_or(DEFAULT_INTERVAL, Duration::from_millis)
+}
+
+fn resource_attributes() -> Vec<(String, String)> {
+    std::env::var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES")
+        .ok()
+        .map(|s| parse_kv(&s))
+        .unwrap_or_default()
+}
+
+/// Byte-identical logic to `hive_jobq_metrics::parse_kv` — small enough
+/// that sharing it isn't worth a dependency edge, kept in lockstep on
+/// purpose.
+fn parse_kv(s: &str) -> Vec<(String, String)> {
+    s.split([',', '\n'])
+        .filter_map(|pair| {
+            let pair = pair.trim();
+            if pair.is_empty() {
+                return None;
+            }
+            let (k, v) = pair.split_once('=')?;
+            let k = k.trim();
+            if k.is_empty() {
+                None
+            } else {
+                Some((k.to_owned(), v.trim().to_owned()))
+            }
+        })
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::parse_push;
+
+    /// A realistic (trimmed) Forgejo push payload — two commits on one ref.
+    #[test]
+    fn parses_repo_and_commit_count_from_a_real_shaped_payload() {
+        let body = br#"{
+            "ref": "refs/heads/main",
+            "repository": { "full_name": "hyperhive/hyperhive" },
+            "commits": [
+                {"id": "abc123", "message": "first"},
+                {"id": "def456", "message": "second"}
+            ]
+        }"#;
+        let activity = parse_push(body).expect("valid push payload parses");
+        assert_eq!(activity.repo, "hyperhive/hyperhive");
+        assert_eq!(activity.commit_count, 2);
+    }
+
+    /// A branch delete or a force-push landing on an already-known tip both
+    /// arrive with an empty `commits` array — still a real push, at zero
+    /// commits, not something to reject as malformed.
+    #[test]
+    fn a_push_with_zero_commits_still_parses() {
+        let body = br#"{
+            "repository": { "full_name": "agents/test-1" },
+            "commits": []
+        }"#;
+        let activity = parse_push(body).expect("empty commits is still a valid push");
+        assert_eq!(activity.repo, "agents/test-1");
+        assert_eq!(activity.commit_count, 0);
+    }
+
+    /// Missing `repository.full_name` or a non-array `commits` must not
+    /// panic or silently default — the caller needs to know parsing failed
+    /// so it can log and drop rather than emit a mislabeled data point.
+    #[test]
+    fn malformed_or_missing_fields_refuse_rather_than_default() {
+        assert!(
+            parse_push(b"{}").is_none(),
+            "empty object has neither field"
+        );
+        assert!(
+            parse_push(br#"{"repository": {}, "commits": []}"#).is_none(),
+            "missing full_name must refuse"
+        );
+        assert!(
+            parse_push(br#"{"repository": {"full_name": "a/b"}, "commits": "oops"}"#).is_none(),
+            "commits must be an array, not any other JSON type"
+        );
+        assert!(parse_push(b"not json at all").is_none());
+    }
+}

swarm-controller/src/webhook.rs

@@ -173,6 +173,13 @@ pub(super) enum DeliveryKind {
     Knowledge,
     /// `pull_request` events on the agent-config repos.
     ConfigPr,
+    /// `push` events on every repo in the instance — see
+    /// `crate::vcs_metrics`'s doc comment for why this exists as its own
+    /// kind rather than folding into [`Self::Knowledge`]'s already-`push`
+    /// hook: that one is repo-scoped to the knowledge repo alone, and this
+    /// one is instance-wide (registered via `admin_create_hook`, not
+    /// `repo_create_hook`) — different scope, same event name.
+    VcsActivity,
 }
 
 /// The route prefix a registered `target_url` must point at.
@@ -192,7 +199,7 @@ impl DeliveryKind {
     /// Every kind, for the registration sweep. An array rather than a
     /// hand-written list at the call site, so adding a kind cannot leave one
     /// hook unregistered.
-    pub(super) const ALL: [Self; 2] = [Self::Knowledge, Self::ConfigPr];
+    pub(super) const ALL: [Self; 3] = [Self::Knowledge, Self::ConfigPr, Self::VcsActivity];
 
     /// The `target_url` to register with Forgejo for this kind, given the
     /// swarm's public base URL.
@@ -217,6 +224,7 @@ impl DeliveryKind {
         match segment {
             "knowledge" => Some(Self::Knowledge),
             "config-pr" => Some(Self::ConfigPr),
+            "vcs-activity" => Some(Self::VcsActivity),
             _ => None,
         }
     }
@@ -233,6 +241,7 @@ impl DeliveryKind {
         match self {
             Self::Knowledge => "knowledge",
             Self::ConfigPr => "config-pr",
+            Self::VcsActivity => "vcs-activity",
         }
     }
 }
@@ -383,6 +392,13 @@ pub(super) async fn post_webhook_forge(
                 cache.apply_webhook_delivery(&body);
             }
         }
+        DeliveryKind::VcsActivity => {
+            if let Some(activity) = crate::vcs_metrics::parse_push(&body) {
+                crate::vcs_metrics::record_push(&activity);
+            } else {
+                tracing::warn!("webhook: vcs-activity delivery did not parse as a push payload");
+            }
+        }
     }
 
     (StatusCode::OK, "ok").into_response()