hyperhive/swarm-controller/src/vcs_metrics.rs

//! OTEL export of commit/push activity across every forge repo in the
//! instance — the metric forgejo's own native `/metrics` endpoint (see
//! `docs/integrations/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, WithHttpConfig};
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.
    //
    // `.with_http_client(..)`: unlike the hive tier's own exporters, this
    // one crosses a real trust boundary (the swarm-tier `otlp/swarm`
    // receiver checks a bearer token's audience, see
    // `crate::otel_http_client`'s module doc) — a plain endpoint-only
    // client would be refused at the receiver.
    let exporter = MetricExporter::builder()
        .with_http()
        .with_protocol(Protocol::HttpJson)
        .with_http_client(authenticated_http_client()?)
        .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())
}

/// Build the [`crate::otel_http_client::AuthenticatedHttpClient`] this
/// exporter pushes through.
///
/// The queue identity is read from [`crate::queue_identity::get`] rather
/// than threaded in from a caller — by the time this runs, [`endpoint`] has
/// already confirmed OTEL is configured for this host, and `queueEnv` is set
/// **unconditionally** for every `swarm-controller` (the nix module's own
/// words: "a controller without a queue is not a lighter controller, it is
/// a broken one"). So an absent queue identity here is never a supported
/// "OTEL enabled, queue not configured" deployment shape — it is a
/// deployment bug, and gets the same `Err` treatment `build()`'s caller
/// already applies to any other exporter-construction failure.
///
/// `SWARM_CONTROLLER_OTEL_AUDIENCE` names the audience `otlp/swarm`'s
/// authenticator checks — set by the same nix option that sets
/// `OTEL_EXPORTER_OTLP_ENDPOINT`, so the two are never independently absent.
///
/// No CA handling here unlike [`swarm_queue_client::QueueConfig::ca_file`]
/// (which `mint_token_for_blocking` already applies to the TOKEN endpoint
/// internally): the metrics endpoint is reached through the gateway with a
/// certificate the store's `pki` mount issued, and this whole process
/// already trusts that mount's root via `SSL_CERT_FILE` (`hive-ca-trust.nix`,
/// applied host-wide to this unit for the same reason the forge client needs
/// no special handling either) — a second, narrower trust bundle here would
/// just be the same fact stated twice.
pub(crate) fn authenticated_http_client() -> Result<crate::otel_http_client::AuthenticatedHttpClient>
{
    let cfg = crate::queue_identity::get().cloned().ok_or_else(|| {
        anyhow::anyhow!(
            "SWARM_CONTROLLER_NATS_URL and friends are unset, but OTEL_EXPORTER_OTLP_ENDPOINT \
                 is — the queue is required for every controller, so this combination is a \
                 deployment bug, not a supported partial config"
        )
    })?;
    let audience = std::env::var("SWARM_CONTROLLER_OTEL_AUDIENCE").context(
        "SWARM_CONTROLLER_OTEL_AUDIENCE is unset, but OTEL_EXPORTER_OTLP_ENDPOINT is — the nix \
         module sets both together",
    )?;
    Ok(crate::otel_http_client::AuthenticatedHttpClient::new(
        reqwest::blocking::Client::new(),
        cfg,
        audience,
    ))
}

/// `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());
    }
}