500 lines
20 KiB
Rust
500 lines
20 KiB
Rust
//! OTEL export of commit/push activity across every forge repo in the
|
|
//! instance — the metric forgejo's own native `/metrics` endpoint (see
|
|
//! `docs/2-handbook/swarm/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: 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
|
|
//! thread, `service.name` + `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` for
|
|
//! the resource, `OTEL_EXPORTER_OTLP_ENDPOINT` as the sole enable signal,
|
|
//! a fixed 60 s export 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::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 _};
|
|
use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig, WithHttpConfig};
|
|
use opentelemetry_sdk::Resource;
|
|
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 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.
|
|
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();
|
|
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),
|
|
})
|
|
}
|
|
|
|
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();
|
|
|
|
/// 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);
|
|
}
|
|
}
|
|
|
|
/// 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;
|
|
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(),
|
|
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,
|
|
})
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "otel vcs-metrics: exporter init failed");
|
|
None
|
|
}
|
|
}
|
|
}
|
|
|
|
pub(crate) 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()
|
|
}
|
|
|
|
pub(crate) fn endpoint() -> Option<String> {
|
|
std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT")
|
|
.ok()
|
|
.map(|s| s.trim().to_owned())
|
|
.filter(|s| !s.is_empty())
|
|
}
|
|
|
|
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());
|
|
}
|
|
}
|
|
|
|
#[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));
|
|
}
|
|
}
|