diff --git a/Cargo.lock b/Cargo.lock index a769c873..80a70420 100644 --- a/Cargo.lock +++ b/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", diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 9716f4e8..1bf1a16d 100644 --- a/swarm-controller/Cargo.toml +++ b/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 diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index a21fa422..6daa49e0 100644 --- a/swarm-controller/src/forge.rs +++ b/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), } } } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 6d33f80d..7b7290f1 100644 --- a/swarm-controller/src/main.rs +++ b/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 diff --git a/swarm-controller/src/vcs_metrics.rs b/swarm-controller/src/vcs_metrics.rs new file mode 100644 index 00000000..c4265e81 --- /dev/null +++ b/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 { + 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, + push_count: Counter, +} + +static INSTRUMENTS: OnceLock> = 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 { + 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 { + // 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 { + 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::().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()); + } +} diff --git a/swarm-controller/src/webhook.rs b/swarm-controller/src/webhook.rs index b93f272d..fee1ffac 100644 --- a/swarm-controller/src/webhook.rs +++ b/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()