From e30f5849d59c1e4718d3d6d5962d405dc567346d Mon Sep 17 00:00:00 2001 From: atlas Date: Wed, 4 Nov 2026 23:13:53 +0100 Subject: [PATCH] swarm-builder: export build counts, durations, queue wait and publish failures through the host's collector The builder pushes its build metrics (builds by kind and outcome, build duration, queue depth and wait, reply publish failures) to its forwarder's OTLP receiver, which relays them to the swarm collector. It also registers the standard store client-certificate gauges (bao.client_cert.*, cert=builder) on the same exporter. They are the builder's liveness signal: exported every minute whether or not a build runs. The cert-metrics table in docs/scheduler/observability.md gains the builder row. --- Cargo.lock | 3 + docs/scheduler/observability.md | 39 ++- docs/swarm/builder.md | 3 + nix/host-modules/swarm-builder.nix | 8 + nix/module-eval/builder.nix | 13 + swarm-builder/Cargo.toml | 8 +- swarm-builder/README.md | 7 + swarm-builder/src/main.rs | 399 +++++++++++++++++++++++++++-- swarm-builder/src/metrics.rs | 292 +++++++++++++++++++++ 9 files changed, 740 insertions(+), 32 deletions(-) create mode 100644 swarm-builder/src/metrics.rs diff --git a/Cargo.lock b/Cargo.lock index a32c8981..ba83fbab 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4930,6 +4930,9 @@ dependencies = [ "futures-util", "hive-log", "mockall", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", "serde_json", "swarm-queue-client", "swarm-secret-client", diff --git a/docs/scheduler/observability.md b/docs/scheduler/observability.md index 85a920e7..1be92cf8 100644 --- a/docs/scheduler/observability.md +++ b/docs/scheduler/observability.md @@ -597,6 +597,38 @@ rules file, at most 10 per repo. The `review-classifier` dashboard in Grafana charts these. +## Swarm builder metrics + +`swarm-builder` exports what it builds through the OTLP receiver of its +container's own forwarder, which pushes to the swarm collector as the builder's +client. `nix/host-modules/swarm-builder.nix` sets +`OTEL_EXPORTER_OTLP_ENDPOINT` and `OTEL_RESOURCE_ATTRIBUTES` for the daemon. The exporter is the same setup as the review +classifier's: HTTP/JSON, cumulative, one export a minute, and nothing exported +without `OTEL_EXPORTER_OTLP_ENDPOINT`. + +| Metric | Unit | Kind | Attributes | +| ------------------------------------------ | ---- | --------- | ---------------------------------------------------------------- | +| `hyperhive.swarm_builder.builds` | — | counter | `kind`: `pr`, `meta`; `outcome`: `success`, `failure`, `refused` | +| `hyperhive.swarm_builder.build.duration` | `s` | histogram | `kind`, `outcome`: `success`, `failure` | +| `hyperhive.swarm_builder.queue.depth` | — | gauge | none | +| `hyperhive.swarm_builder.queue.wait` | `s` | histogram | `kind` | +| `hyperhive.swarm_builder.publish.failures` | — | counter | none | + +`kind` `pr` is an agent config PR's head and `meta` an agent built from the +swarm meta flake. Lock and archetype requests aren't counted. `refused` is a +request that waited out the one-build slot, which the builder then answers busy +without building, so it has a `builds` point and no `build.duration` point. A build that +can't start (no forge token) counts as a `failure` with a duration; a payload +that doesn't decode counts as a `failure` in `builds` only. + +`build.duration` runs from holding the slot to the reply, including the cache +push of a meta build. `queue.wait` runs from the request's arrival to the slot, +or to the refusal. `queue.depth` is the requests waiting for the slot now, +exported every minute, 0 when idle, so it doubles as the liveness signal; the +`bao.client_cert.*` gauges below ride the same exporter. +`publish.failures` counts replies the queue refused to take, which the +controller sees as a missing answer. + ## Store client-certificate metrics Every service that logs in to the secret store with a client certificate can @@ -609,7 +641,7 @@ data point each, unix seconds: | `bao.client_cert.not_after` | `s` | gauge | `cert` | `cert` names the identity (`client`, `controller`, `agent`, `review-classifier`, -`ci-metrics`); the resource labels (`service.name`, `hive`, `agent`) say who +`ci-metrics`, `builder`); the resource labels (`service.name`, `hive`, `agent`) say who holds it. `swarm_secret_client::cert_metrics` (cargo feature `otel`) registers both gauges on the service's existing exporter and reads `BAO_CLIENT_CERT` at every collection, so a renewed leaf shows up on the next export. A file that's missing, or that isn't a @@ -623,9 +655,10 @@ certificate yields no data point and a debug log line. The parser is | `swarm-controller` | `controller` | the job-graph exporter | | `swarm-review-classifier` | `review-classifier` | the cost exporter, built on its first recording | | `swarm-ci-metrics` | `ci-metrics` | a push of its own at the start of each timer pass | +| `swarm-builder` | `builder` | the build exporter, started with the daemon | -Not covered yet: the other seven Rust services that hold a store certificate -(`swarm-builder`, `swarm-claude-login`, `swarm-notify-router`, +Not covered yet: the other six Rust services that hold a store certificate +(`swarm-claude-login`, `swarm-notify-router`, `swarm-nats-auth`, `swarm-logs-auth`, `swarm-metrics-auth`, `swarm-matrix-ctl`) have no exporter, and the ten identities that only shell units use report nothing. Neither does any CA certificate. Each leaf's own diff --git a/docs/swarm/builder.md b/docs/swarm/builder.md index 0bade76f..e2ca37e1 100644 --- a/docs/swarm/builder.md +++ b/docs/swarm/builder.md @@ -563,6 +563,9 @@ request for the forge's host: any other host gets nothing. The builder holds no status with its own forge token. Details per path: [`credentials-reference.md`](credentials-reference.md). +The builder exports build counts, durations and queue wait; see +[Swarm builder metrics](../scheduler/observability.md#swarm-builder-metrics). + ## Isolation The container has its own network namespace on the bridge, as the CI runner diff --git a/nix/host-modules/swarm-builder.nix b/nix/host-modules/swarm-builder.nix index 2ab058e5..4bc4d218 100644 --- a/nix/host-modules/swarm-builder.nix +++ b/nix/host-modules/swarm-builder.nix @@ -51,6 +51,9 @@ let # Loopback in the container's own netns, and distinct from the other service # containers' forwarders all the same. forwarderTelemetryPort = 8892; + # The standard OTLP/HTTP port, on the container's loopback: where the daemon + # pushes its build metrics, which the forwarder relays to the collector. + forwarderOtlpPort = 4318; rootfsCheck = import ./lib/container-rootfs-check.nix { inherit config; }; @@ -242,6 +245,7 @@ in inherit clientId; tokenUrl = "${autheliaCfg.url}/api/oidc/token"; telemetryPort = forwarderTelemetryPort; + otlpMetricsPort = forwarderOtlpPort; store = { package = baoDeploy.package; addr = "https://${baoCfg.domain}:${toString baoCfg.port}"; @@ -281,6 +285,10 @@ in SWARM_BUILDER_NIXPKGS_FLAKE = nixpkgsFlake; SWARM_BUILDER_FORGE_URL = "https://${forgeCfg.domain}"; SWARM_BUILDER_CONFIG_ORG = "agent-configs"; + OTEL_EXPORTER_OTLP_ENDPOINT = "http://127.0.0.1:${toString forwarderOtlpPort}"; + OTEL_RESOURCE_ATTRIBUTES = "service.name=hyperhive-swarm-builder,swarm=${ + if hyperhiveCfg.swarm.name == null then "unknown" else hyperhiveCfg.swarm.name + }"; } // lib.optionalAttrs pushesToCache { SWARM_BUILDER_ATTIC_DOMAIN = atticCfg.domain; diff --git a/nix/module-eval/builder.nix b/nix/module-eval/builder.nix index 938ae150..ec48573b 100644 --- a/nix/module-eval/builder.nix +++ b/nix/module-eval/builder.nix @@ -188,6 +188,19 @@ let && svc.serviceConfig.DynamicUser && !(lib.any (lib.hasSuffix "TOKEN") (lib.attrNames env)); } + { + # The daemon pushes its build metrics to the forwarder's OTLP receiver on + # the container's loopback, which relays them to the swarm collector under + # the builder's own client. + name = "the builder exports its metrics through the forwarder's OTLP receiver on the container's loopback"; + ok = + let + f = inner.services.hyperhive.swarmJournalForwarder; + in + f.otlpMetricsPort != null + && env.OTEL_EXPORTER_OTLP_ENDPOINT == "http://127.0.0.1:${toString f.otlpMetricsPort}" + && lib.hasInfix "service.name=hyperhive-swarm-builder" env.OTEL_RESOURCE_ATTRIBUTES; + } { # The closure is pushed with `attic` from the unit, whose config file holds # the push token: it goes in a runtime directory only the dynamic user can diff --git a/swarm-builder/Cargo.toml b/swarm-builder/Cargo.toml index 69d2a943..3487af8f 100644 --- a/swarm-builder/Cargo.toml +++ b/swarm-builder/Cargo.toml @@ -14,10 +14,14 @@ async-nats.workspace = true # StreamExt::next on the subscription. futures-util.workspace = true hive-log.workspace = true +# The build metrics, exported over OTLP like the review classifier's. +opentelemetry.workspace = true +opentelemetry-otlp.workspace = true +opentelemetry_sdk.workspace = true serde_json.workspace = true swarm-queue-client.workspace = true # The forge token and queue client secret paths, and the store login. -swarm-secret-client.workspace = true +swarm-secret-client = { workspace = true, features = ["otel"] } # The build subject and the shapes the controller sends and reads back. swarm-wire.workspace = true tokio.workspace = true @@ -27,6 +31,8 @@ tracing.workspace = true # out the build slot's wait limit on paused time. [dev-dependencies] mockall.workspace = true +# `testing`: the in-memory exporter the metric tests read. +opentelemetry_sdk = { workspace = true, features = ["testing"] } tokio = { workspace = true, features = ["test-util"] } tracing-subscriber.workspace = true diff --git a/swarm-builder/README.md b/swarm-builder/README.md index 6449986a..1a1e5b83 100644 --- a/swarm-builder/README.md +++ b/swarm-builder/README.md @@ -36,6 +36,13 @@ to the swarm's own cache), it first pushes a green build's closure to that cache whether that worked. A failed push leaves the build green. See [`docs/swarm/builder.md`](../docs/swarm/builder.md#binary-cache). +## Metrics + +The service exports build counts, durations, queue depth and wait, and reply +publish failures through its container's forwarder to the swarm collector. The +names and attributes are in +[`docs/scheduler/observability.md`](../docs/scheduler/observability.md#swarm-builder-metrics). + ## Credentials The container holds one credential: its own store certificate, which nspawn diff --git a/swarm-builder/src/main.rs b/swarm-builder/src/main.rs index 678ddb47..7dda6326 100644 --- a/swarm-builder/src/main.rs +++ b/swarm-builder/src/main.rs @@ -19,6 +19,9 @@ //! When a cache is set (see [`push`]), a green meta build is pushed to it before //! the reply, and the reply says whether that worked. //! +//! Build counts, durations, queue depth and wait, and reply publish failures +//! are exported by [`metrics`]. +//! //! The store identity comes from the `BAO_*` environment. The queue client //! secret is read at start and again when authelia refuses it, the forge token //! once per build, both from the store under that identity. @@ -31,6 +34,7 @@ use std::time::Duration; use anyhow::{Context, Result, anyhow, bail}; use async_nats::{Message, Subject}; use futures_util::{Stream, StreamExt as _}; +use metrics::{Judged, Kind, Metrics, Outcome}; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::forge::Credential; use swarm_secret_client::stored_secret::StoredSecret; @@ -41,6 +45,7 @@ use swarm_wire::builder::{ }; mod archetypes; +mod metrics; mod nix_build; mod push; @@ -131,7 +136,12 @@ async fn main() -> Result<()> { .await .with_context(|| format!("subscribing to {SUBJECT}"))?; tracing::info!(subject = SUBJECT, "builder: awaiting builds"); - let publisher = Queue(queue.clone()); + // Kept alive: dropping the provider stops the export. + let (_metrics_provider, metrics) = metrics::start(); + let publisher = Metered { + inner: Queue(queue.clone()), + metrics: metrics.clone(), + }; let evaluations = queue .queue_subscribe(ARCHETYPES_SUBJECT, QUEUE_GROUP.to_owned()) .await @@ -160,9 +170,9 @@ async fn main() -> Result<()> { // refuses one that waited too long behind another. Meta builds take the // same slot as PR checks. let slot = Arc::new(tokio::sync::Semaphore::new(1)); - serve_meta_builds(&queue, &publisher, Store, &builder, &slot).await?; + serve_meta_builds(&queue, &publisher, Store, &builder, &slot, &metrics).await?; - answer_builds(requests, publisher, Store, builder, slot).await; + answer_builds(requests, publisher, Store, builder, slot, metrics).await; bail!("the subscription to {SUBJECT} ended") } @@ -176,6 +186,23 @@ trait Publisher: Clone + Send + Sync + 'static { ) -> impl Future> + Send; } +/// A [`Publisher`] that counts the replies its inner publisher fails to send. +#[derive(Clone)] +struct Metered

{ + inner: P, + metrics: Metrics, +} + +impl Publisher for Metered

{ + async fn publish(&self, subject: Subject, body: Vec) -> Result<(), String> { + let sent = self.inner.publish(subject, body).await; + if sent.is_err() { + self.metrics.publish_failed(); + } + sent + } +} + /// The secrets a request reads from the store, once per request. [`Store`] in /// production; tests answer with canned replies. trait Tokens: Clone + Send + Sync + 'static { @@ -237,6 +264,7 @@ async fn answer_builds( tokens: impl Tokens, builder: Arc, slot: Arc, + metrics: Metrics, ) { while let Some(msg) = requests.next().await { let received = tokio::time::Instant::now(); @@ -244,14 +272,16 @@ async fn answer_builds( tracing::warn!("a build arrived with no reply subject; dropping it"); continue; }; - let (publisher, tokens, builder, slot) = ( + let (publisher, tokens, builder, slot, metrics) = ( publisher.clone(), tokens.clone(), Arc::clone(&builder), Arc::clone(&slot), + metrics.clone(), ); tokio::spawn(async move { - let Some(reply) = build_reply(&msg.payload, received, &slot, &builder, &tokens).await + let Some(reply) = + build_reply(&msg.payload, received, &slot, &builder, &tokens, &metrics).await else { return; }; @@ -274,10 +304,21 @@ async fn build_reply( slot: &tokio::sync::Semaphore, builder: &nix_build::Builder, tokens: &impl Tokens, + metrics: &Metrics, ) -> Option { match serde_json::from_slice::(payload) { - Ok(job) => in_slot(slot, received, || busy(&job), handle(&job, builder, tokens)).await, + Ok(job) => { + in_slot( + slot, + received, + (metrics, Kind::Pr), + || busy(&job), + handle(&job, builder, tokens), + ) + .await + } Err(e) => { + metrics.count(Kind::Pr, Outcome::Failure); tracing::warn!(error = %e, "an undecodable build; answering with the error"); Some(Reply::Failed { error_tail: format!("the builder could not decode the request: {e}"), @@ -374,6 +415,7 @@ async fn serve_meta_builds( tokens: impl Tokens, builder: &Arc, slot: &Arc, + metrics: &Metrics, ) -> Result<()> { let requests = queue .queue_subscribe(META_SUBJECT, QUEUE_GROUP.to_owned()) @@ -385,6 +427,7 @@ async fn serve_meta_builds( tokens, Arc::clone(builder), Arc::clone(slot), + metrics.clone(), )); Ok(()) } @@ -397,6 +440,7 @@ async fn answer_meta_builds( tokens: impl Tokens, builder: Arc, slot: Arc, + metrics: Metrics, ) { while let Some(msg) = requests.next().await { let received = tokio::time::Instant::now(); @@ -404,14 +448,16 @@ async fn answer_meta_builds( tracing::warn!("a meta build arrived with no reply subject; dropping it"); continue; }; - let (publisher, tokens, builder, slot) = ( + let (publisher, tokens, builder, slot, metrics) = ( publisher.clone(), tokens.clone(), Arc::clone(&builder), Arc::clone(&slot), + metrics.clone(), ); tokio::spawn(async move { - let Some(reply) = meta_reply(&msg.payload, received, &slot, &builder, &tokens).await + let Some(reply) = + meta_reply(&msg.payload, received, &slot, &builder, &tokens, &metrics).await else { return; }; @@ -435,12 +481,14 @@ async fn meta_reply( slot: &tokio::sync::Semaphore, builder: &nix_build::Builder, tokens: &impl Tokens, + metrics: &Metrics, ) -> Option { match serde_json::from_slice::(payload) { Ok(job) => { in_slot( slot, received, + (metrics, Kind::Meta), || busy_meta(&job), handle_meta( &job, @@ -453,6 +501,7 @@ async fn meta_reply( } Err(e) => { tracing::warn!(error = %e, "an undecodable meta build; answering with the error"); + metrics.count(Kind::Meta, Outcome::Failure); Some(FromMetaReply::Failed { error_tail: format!("the builder could not decode the request: {e}"), }) @@ -463,17 +512,47 @@ async fn meta_reply( /// `build`'s reply once a request holds the build slot, or `refuse`'s, without /// running `build`, once it has waited [`MAX_SLOT_WAIT`] since `received`. /// `None` only when the slot is closed. -async fn in_slot( +/// +/// Records the wait, and the build's count and duration, on `metrics` as +/// `kind`; a refusal counts as [`Outcome::Refused`]. +async fn in_slot( slot: &tokio::sync::Semaphore, received: tokio::time::Instant, + (metrics, kind): (&Metrics, Kind), refuse: impl FnOnce() -> T, build: impl Future, ) -> Option { - let Ok(permit) = tokio::time::timeout_at(received + MAX_SLOT_WAIT, slot.acquire()).await else { + let waiting = metrics.start_waiting(); + let acquired = tokio::time::timeout_at(received + MAX_SLOT_WAIT, slot.acquire()).await; + metrics.record_wait(kind, received.elapsed()); + drop(waiting); + let Ok(permit) = acquired else { + metrics.count(kind, Outcome::Refused); return Some(refuse()); }; let _permit = permit.ok()?; - Some(build.await) + let started = tokio::time::Instant::now(); + let reply = build.await; + metrics.record_build(kind, reply.outcome(), started.elapsed()); + Some(reply) +} + +impl Judged for Reply { + fn outcome(&self) -> Outcome { + match self { + Self::Built { .. } => Outcome::Success, + Self::Failed { .. } => Outcome::Failure, + } + } +} + +impl Judged for FromMetaReply { + fn outcome(&self) -> Outcome { + match self { + Self::Built { .. } => Outcome::Success, + Self::Failed { .. } | Self::Unavailable { .. } => Outcome::Failure, + } + } } /// The refusal text of a request that waited [`MAX_SLOT_WAIT`] for the slot. @@ -727,7 +806,14 @@ mod tests { out_path: "/nix/store/x".to_owned(), } }; - let reply = in_slot(&slot, received, || busy(&job()), build).await; + let reply = in_slot( + &slot, + received, + (&Metrics::noop(), Kind::Pr), + || busy(&job()), + build, + ) + .await; (reply, ran.load(Ordering::SeqCst), received.elapsed()) } @@ -762,14 +848,20 @@ mod tests { let slot = Arc::new(Semaphore::new(1)); let check = Arc::clone(&slot).acquire_owned().await.expect("open"); let received = Instant::now(); - let refused = in_slot(&slot, received, || busy_meta(&meta), async { - FromMetaReply::Built { - out_path: "/nix/store/x".to_owned(), - system: "s".to_owned(), - cached: false, - private_users: false, - } - }) + let refused = in_slot( + &slot, + received, + (&Metrics::noop(), Kind::Meta), + || busy_meta(&meta), + async { + FromMetaReply::Built { + out_path: "/nix/store/x".to_owned(), + system: "s".to_owned(), + cached: false, + private_users: false, + } + }, + ) .await; assert!( matches!(&refused, Some(FromMetaReply::Unavailable { reason }) if reason.contains("busy")), @@ -778,14 +870,20 @@ mod tests { assert_eq!(received.elapsed(), MAX_SLOT_WAIT); // Control: once the check lets go, the same request builds. drop(check); - let built = in_slot(&slot, Instant::now(), || busy_meta(&meta), async { - FromMetaReply::Built { - out_path: "/nix/store/x".to_owned(), - system: "s".to_owned(), - cached: false, - private_users: false, - } - }) + let built = in_slot( + &slot, + Instant::now(), + (&Metrics::noop(), Kind::Meta), + || busy_meta(&meta), + async { + FromMetaReply::Built { + out_path: "/nix/store/x".to_owned(), + system: "s".to_owned(), + cached: false, + private_users: false, + } + }, + ) .await; assert!( matches!(built, Some(FromMetaReply::Built { .. })), @@ -1083,6 +1181,7 @@ mod tests { env.tokens, builder, Arc::new(Semaphore::new(1)), + Metrics::noop(), ) .await; } @@ -1094,6 +1193,7 @@ mod tests { env.tokens, builder, Arc::new(Semaphore::new(1)), + Metrics::noop(), ) .await; } @@ -1308,6 +1408,7 @@ mod tests { &slot, &fake.builder(), &env.tokens, + &Metrics::noop(), ) .await; assert!( @@ -1330,6 +1431,7 @@ mod tests { &slot, &fake.builder(), &env.tokens, + &Metrics::noop(), ) .await; assert!(reply.is_none(), "{reply:?}"); @@ -1495,6 +1597,7 @@ mod tests { &slot, &fake.builder(), &env.tokens, + &Metrics::noop(), ) .await; assert!( @@ -1648,4 +1751,244 @@ mod tests { ); } } + + // --- metrics --- + + use crate::metrics::probe::{Point, Probe}; + + const BUILDS: &str = "hyperhive.swarm_builder.builds"; + const DURATION: &str = "hyperhive.swarm_builder.build.duration"; + const WAIT: &str = "hyperhive.swarm_builder.queue.wait"; + const DEPTH: &str = "hyperhive.swarm_builder.queue.depth"; + const PUBLISH_FAILURES: &str = "hyperhive.swarm_builder.publish.failures"; + + /// `(labels, value)` of every point of `name`. + fn counts(probe: &Probe, name: &str) -> Vec<(String, i64)> { + probe + .read(name) + .into_iter() + .map(|Point { labels, value, .. }| (labels, value)) + .collect() + } + + async fn pr_reply( + fake: &Fake, + env: &Rig, + slot: &Semaphore, + metrics: &Metrics, + ) -> Option { + build_reply( + build_json("atlas", REV, 3).as_bytes(), + Instant::now(), + slot, + &fake.builder(), + &env.tokens, + metrics, + ) + .await + } + + #[tokio::test] + async fn a_green_pr_build_counts_one_success_with_a_duration_and_a_wait() { + let probe = Probe::new(); + let fake = Fake::new(&format!("echo {OUT}")); + let (env, _sent) = env(); + let reply = pr_reply(&fake, &env, &Semaphore::new(1), &probe.metrics).await; + assert!(matches!(reply, Some(Reply::Built { .. })), "{reply:?}"); + assert_eq!( + counts(&probe, BUILDS), + [("kind=pr,outcome=success".to_owned(), 1)] + ); + assert_eq!( + counts(&probe, DURATION), + [("kind=pr,outcome=success".to_owned(), 1)] + ); + assert_eq!(counts(&probe, WAIT), [("kind=pr".to_owned(), 1)]); + } + + #[tokio::test] + async fn a_red_pr_build_counts_a_failure_and_no_success() { + let probe = Probe::new(); + let fake = Fake::new("echo 'error: boom' >&2\nexit 1"); + let (env, _sent) = env(); + pr_reply(&fake, &env, &Semaphore::new(1), &probe.metrics).await; + assert_eq!( + counts(&probe, BUILDS), + [("kind=pr,outcome=failure".to_owned(), 1)] + ); + assert_eq!( + counts(&probe, DURATION), + [("kind=pr,outcome=failure".to_owned(), 1)] + ); + } + + #[tokio::test] + async fn a_pr_build_whose_token_cannot_be_read_counts_a_failure() { + let probe = Probe::new(); + let fake = Fake::new(&format!("echo {OUT}")); + let (env, _sent) = rig_with(unreadable("no token"), token("p"), 0); + pr_reply(&fake, &env, &Semaphore::new(1), &probe.metrics).await; + assert_eq!( + counts(&probe, BUILDS), + [("kind=pr,outcome=failure".to_owned(), 1)] + ); + } + + #[tokio::test] + async fn an_undecodable_request_counts_a_failure_of_its_kind_and_no_duration() { + let probe = Probe::new(); + let fake = Fake::new(&format!("echo {OUT}")); + let (env, _sent) = env(); + let slot = Semaphore::new(1); + build_reply( + b"{", + Instant::now(), + &slot, + &fake.builder(), + &env.tokens, + &probe.metrics, + ) + .await; + meta_reply( + b"{", + Instant::now(), + &slot, + &fake.builder(), + &env.tokens, + &probe.metrics, + ) + .await; + assert_eq!( + counts(&probe, BUILDS), + [ + ("kind=meta,outcome=failure".to_owned(), 1), + ("kind=pr,outcome=failure".to_owned(), 1), + ] + ); + assert_eq!(probe.read(DURATION), Vec::::new()); + } + + async fn meta_outcome(fake: &Fake, env: &Rig, metrics: &Metrics) { + meta_reply( + meta_json("atlas", REV).as_bytes(), + Instant::now(), + &Semaphore::new(1), + &fake.builder(), + &env.tokens, + metrics, + ) + .await; + } + + #[tokio::test] + async fn a_meta_build_is_counted_under_its_own_kind_for_success_and_for_failure() { + let probe = Probe::new(); + let (env, _sent) = env(); + meta_outcome(&meta_nix("echo true"), &env, &probe.metrics).await; + assert_eq!( + counts(&probe, BUILDS), + [("kind=meta,outcome=success".to_owned(), 1)] + ); + let failing = Fake::new("exit 1"); + meta_outcome(&failing, &env, &probe.metrics).await; + assert_eq!( + counts(&probe, BUILDS), + [ + ("kind=meta,outcome=failure".to_owned(), 1), + ("kind=meta,outcome=success".to_owned(), 1), + ] + ); + assert_eq!( + counts(&probe, DURATION), + [ + ("kind=meta,outcome=failure".to_owned(), 1), + ("kind=meta,outcome=success".to_owned(), 1), + ] + ); + } + + #[tokio::test] + async fn a_meta_build_with_no_forge_token_counts_a_failure() { + let probe = Probe::new(); + let (env, _sent) = rig_with(unreadable("no token"), token("p"), 0); + meta_outcome(&meta_nix("echo true"), &env, &probe.metrics).await; + assert_eq!( + counts(&probe, BUILDS), + [("kind=meta,outcome=failure".to_owned(), 1)] + ); + } + + #[tokio::test(start_paused = true)] + async fn a_refused_request_counts_refused_records_its_whole_wait_and_no_duration() { + let probe = Probe::new(); + let fake = Fake::new(&format!("echo {OUT}")); + let (env, _sent) = env(); + let slot = Semaphore::new(1); + let _held = slot.acquire().await.expect("open"); + pr_reply(&fake, &env, &slot, &probe.metrics).await; + assert_eq!( + counts(&probe, BUILDS), + [("kind=pr,outcome=refused".to_owned(), 1)] + ); + assert_eq!(probe.read(DURATION), Vec::::new()); + let wait = probe.read(WAIT); + assert_eq!(wait.len(), 1, "{wait:?}"); + assert!( + (wait[0].sum - MAX_SLOT_WAIT.as_secs_f64()).abs() < 1e-6, + "{wait:?}" + ); + } + + #[tokio::test(start_paused = true)] + async fn the_queue_depth_counts_requests_waiting_for_the_slot_and_returns_to_zero() { + let probe = Probe::new(); + let slot = Arc::new(Semaphore::new(1)); + let held = Arc::clone(&slot).acquire_owned().await.expect("open"); + let waiter = { + let (slot, metrics) = (Arc::clone(&slot), probe.metrics.clone()); + tokio::spawn(async move { + in_slot( + &slot, + Instant::now(), + (&metrics, Kind::Pr), + || Reply::Failed { + error_tail: "busy".to_owned(), + }, + async { + Reply::Built { + out_path: "/nix/store/x".to_owned(), + } + }, + ) + .await + }) + }; + tokio::time::sleep(Duration::from_secs(1)).await; + assert_eq!(counts(&probe, DEPTH), [(String::new(), 1)]); + drop(held); + waiter.await.expect("joined"); + assert_eq!(counts(&probe, DEPTH), [(String::new(), 0)]); + } + + #[tokio::test] + async fn a_reply_the_queue_refuses_counts_one_publish_failure_and_a_sent_one_none() { + let probe = Probe::new(); + let mut sink = MockPublisher::new(); + let mut calls = 0; + sink.expect_publish().returning(move |_, _| { + calls += 1; + let ok = calls > 1; + Box::pin(async move { if ok { Ok(()) } else { Err("closed".to_owned()) } }) + }); + let metered = Metered { + inner: Arc::new(sink), + metrics: probe.metrics.clone(), + }; + let failed = metered.publish(Subject::from("a"), vec![]).await; + assert!(failed.is_err()); + assert_eq!(counts(&probe, PUBLISH_FAILURES), [(String::new(), 1)]); + let sent = metered.publish(Subject::from("b"), vec![]).await; + assert!(sent.is_ok()); + assert_eq!(counts(&probe, PUBLISH_FAILURES), [(String::new(), 1)]); + } } diff --git a/swarm-builder/src/metrics.rs b/swarm-builder/src/metrics.rs new file mode 100644 index 00000000..da93d84d --- /dev/null +++ b/swarm-builder/src/metrics.rs @@ -0,0 +1,292 @@ +//! What the builder exports about its builds, through the OTLP endpoint the +//! standard `OTEL_*` environment names: the container's own forwarder, which +//! pushes to the swarm collector as the builder's client. +//! +//! | metric | unit | kind | attributes | +//! | --- | --- | --- | --- | +//! | `hyperhive.swarm_builder.builds` | | counter | `kind`, `outcome` | +//! | `hyperhive.swarm_builder.build.duration` | `s` | histogram | `kind`, `outcome` | +//! | `hyperhive.swarm_builder.queue.depth` | | gauge | | +//! | `hyperhive.swarm_builder.queue.wait` | `s` | histogram | `kind` | +//! | `hyperhive.swarm_builder.publish.failures` | | counter | | +//! +//! `kind` is [`Kind`], `outcome` is [`Outcome`]. A build's duration runs from +//! holding the build slot to the reply, and the wait from the request's arrival +//! to the slot, or to the refusal when the wait limit passes first. +//! +//! The metrics are created on a no-op meter when no endpoint is configured, so +//! recording is always safe. + +use std::sync::Arc; +use std::sync::atomic::{AtomicI64, Ordering}; +use std::time::Duration; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider as _}; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; + +/// How often the metrics are exported. +const INTERVAL: Duration = Duration::from_mins(1); + +/// Bucket bounds in seconds, from a cached build to the slot's wait limit. +const SECONDS: [f64; 11] = [ + 1.0, 5.0, 15.0, 30.0, 60.0, 120.0, 300.0, 600.0, 1200.0, 1800.0, 3600.0, +]; + +/// The kind of request a build answers. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Kind { + /// An agent config PR's head. + Pr, + /// An agent from the swarm meta flake. + Meta, +} + +impl Kind { + const fn label(self) -> &'static str { + match self { + Self::Pr => "pr", + Self::Meta => "meta", + } + } +} + +/// How a request ended. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Outcome { + /// The build produced a closure. + Success, + /// The build failed, was undecodable, or could not run (no forge token). + Failure, + /// The request waited out the slot and was refused without building. + Refused, +} + +impl Outcome { + const fn label(self) -> &'static str { + match self { + Self::Success => "success", + Self::Failure => "failure", + Self::Refused => "refused", + } + } +} + +/// A reply that says whether the build behind it worked. +pub trait Judged { + fn outcome(&self) -> Outcome; +} + +/// The instruments. Clones share them. +#[derive(Clone)] +pub struct Metrics { + builds: Counter, + duration: Histogram, + wait: Histogram, + publish_failures: Counter, + waiting: Arc, +} + +/// A request counted in the queue depth until dropped. +pub struct Waiting(Arc); + +impl Drop for Waiting { + fn drop(&mut self) { + self.0.fetch_sub(1, Ordering::Relaxed); + } +} + +impl Metrics { + /// The instruments on `meter`. + #[must_use] + pub fn new(meter: &Meter) -> Self { + let waiting = Arc::new(AtomicI64::new(0)); + let depth = Arc::clone(&waiting); + meter + .i64_observable_gauge("hyperhive.swarm_builder.queue.depth") + .with_callback(move |obs| obs.observe(depth.load(Ordering::Relaxed), &[])) + .build(); + Self { + builds: meter.u64_counter("hyperhive.swarm_builder.builds").build(), + duration: meter + .f64_histogram("hyperhive.swarm_builder.build.duration") + .with_unit("s") + .with_boundaries(SECONDS.to_vec()) + .build(), + wait: meter + .f64_histogram("hyperhive.swarm_builder.queue.wait") + .with_unit("s") + .with_boundaries(SECONDS.to_vec()) + .build(), + publish_failures: meter + .u64_counter("hyperhive.swarm_builder.publish.failures") + .build(), + waiting, + } + } + + /// Metrics that go nowhere, for a builder with no collector. + #[must_use] + pub fn noop() -> Self { + Self::new(&opentelemetry::global::meter("hyperhive.swarm_builder")) + } + + /// Counts a request as waiting for the build slot until the guard drops. + pub fn start_waiting(&self) -> Waiting { + self.waiting.fetch_add(1, Ordering::Relaxed); + Waiting(Arc::clone(&self.waiting)) + } + + /// Records how long a `kind` request waited for the slot. + pub fn record_wait(&self, kind: Kind, waited: Duration) { + self.wait + .record(waited.as_secs_f64(), &[KeyValue::new("kind", kind.label())]); + } + + /// Counts a `kind` request that ended `outcome`. + pub fn count(&self, kind: Kind, outcome: Outcome) { + self.builds.add(1, &labels(kind, outcome)); + } + + /// Counts a build that ended `outcome` and records how long it ran. + pub fn record_build(&self, kind: Kind, outcome: Outcome, ran: Duration) { + self.count(kind, outcome); + self.duration + .record(ran.as_secs_f64(), &labels(kind, outcome)); + } + + /// Counts a reply the queue refused to take. + pub fn publish_failed(&self) { + self.publish_failures.add(1, &[]); + } +} + +fn labels(kind: Kind, outcome: Outcome) -> [KeyValue; 2] { + [ + KeyValue::new("kind", kind.label()), + KeyValue::new("outcome", outcome.label()), + ] +} + +/// The provider to keep alive while the metrics should export, with the +/// metrics on it; `None` for the provider when no endpoint is configured or +/// the exporter failed, and the metrics then go nowhere. +#[must_use] +pub fn start() -> (Option, Metrics) { + if !std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok_and(|s| !s.trim().is_empty()) { + return (None, Metrics::noop()); + } + // The blocking client: `PeriodicReader` exports from a thread of its own + // with no Tokio reactor. The resource comes from `OTEL_RESOURCE_ATTRIBUTES`. + let exporter = match MetricExporter::builder() + .with_http() + .with_protocol(Protocol::HttpJson) + .build() + { + Ok(exporter) => exporter, + Err(e) => { + tracing::warn!(error = %e, "building the metric exporter failed; no metrics are exported"); + return (None, Metrics::noop()); + } + }; + let reader = PeriodicReader::builder(exporter) + .with_interval(INTERVAL) + .build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let meter = provider.meter("hyperhive.swarm_builder"); + swarm_secret_client::cert_metrics::register(&meter, "builder"); + let metrics = Metrics::new(&meter); + (Some(provider), metrics) +} + +/// An in-memory exporter behind a [`Metrics`], for the tests of the code that +/// records. +#[cfg(test)] +pub mod probe { + use super::*; + use opentelemetry_sdk::metrics::InMemoryMetricExporter; + use opentelemetry_sdk::metrics::data::{ + AggregatedMetrics, MetricData, ResourceMetrics, ScopeMetrics, + }; + + /// One data point: its attributes as `k=v,k=v`, its value (a sum, a + /// gauge's reading or a histogram's count), and a histogram's sum. + #[derive(Debug, PartialEq)] + pub struct Point { + pub labels: String, + pub value: i64, + pub sum: f64, + } + + pub struct Probe { + provider: SdkMeterProvider, + exporter: InMemoryMetricExporter, + pub metrics: Metrics, + } + + impl Probe { + pub fn new() -> Self { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let metrics = Metrics::new(&provider.meter("test")); + Self { + provider, + exporter, + metrics, + } + } + + /// The data points of metric `name` now, sorted by labels; empty when + /// nothing was recorded under it. + pub fn read(&self, name: &str) -> Vec { + self.exporter.reset(); + self.provider.force_flush().expect("flush"); + let finished = self.exporter.get_finished_metrics().expect("metrics"); + let Some(metric) = finished + .iter() + .flat_map(ResourceMetrics::scope_metrics) + .flat_map(ScopeMetrics::metrics) + .find(|m| m.name() == name) + else { + return Vec::new(); + }; + let label = |attrs: &mut dyn Iterator| { + let mut all: Vec = + attrs.map(|kv| format!("{}={}", kv.key, kv.value)).collect(); + all.sort(); + all.join(",") + }; + let mut points: Vec = match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => sum + .data_points() + .map(|p| Point { + labels: label(&mut p.attributes()), + value: i64::try_from(p.value()).expect("fits"), + sum: 0.0, + }) + .collect(), + AggregatedMetrics::I64(MetricData::Gauge(gauge)) => gauge + .data_points() + .map(|p| Point { + labels: label(&mut p.attributes()), + value: p.value(), + sum: 0.0, + }) + .collect(), + AggregatedMetrics::F64(MetricData::Histogram(hist)) => hist + .data_points() + .map(|p| Point { + labels: label(&mut p.attributes()), + value: i64::try_from(p.count()).expect("fits"), + sum: p.sum(), + }) + .collect(), + other => panic!("unexpected aggregation: {other:?}"), + }; + points.sort_by(|a, b| a.labels.cmp(&b.labels)); + points + } + } +}