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 + } + } +}