mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

swarm-builder: export build counts, durations, queue wait and publish failures through the host's collector

Browse source 
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.
This commit is contained in:
atlas 2026-11-04 23:13:53 +01:00 • committed by mara
commit e30f5849d5
2 changed files with 740 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

399
swarm-builder/src/main.rs
View file

@ -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<Output = Result<(), String>> + Send;
}
/// A [`Publisher`] that counts the replies its inner publisher fails to send.
#[derive(Clone)]
struct Metered<P> {
inner: P,
metrics: Metrics,
}
impl<P: Publisher> Publisher for Metered<P> {
async fn publish(&self, subject: Subject, body: Vec<u8>) -> 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<nix_build::Builder>,
slot: Arc<tokio::sync::Semaphore>,
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<Reply> {
match serde_json::from_slice::<BuildAgent>(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<nix_build::Builder>,
slot: &Arc<tokio::sync::Semaphore>,
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<nix_build::Builder>,
slot: Arc<tokio::sync::Semaphore>,
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<FromMetaReply> {
match serde_json::from_slice::<BuildFromMeta>(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<T>(
///
/// Records the wait, and the build's count and duration, on `metrics` as
/// `kind`; a refusal counts as [`Outcome::Refused`].
async fn in_slot<T: Judged>(
slot: &tokio::sync::Semaphore,
received: tokio::time::Instant,
(metrics, kind): (&Metrics, Kind),
refuse: impl FnOnce() -> T,
build: impl Future<Output = T>,
) -> Option<T> {
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<Reply> {
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::<Point>::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::<Point>::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)]);
}
}

292
swarm-builder/src/metrics.rs Normal file
View file

@ -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<u64>,
duration: Histogram<f64>,
wait: Histogram<f64>,
publish_failures: Counter<u64>,
waiting: Arc<AtomicI64>,
}
/// A request counted in the queue depth until dropped.
pub struct Waiting(Arc<AtomicI64>);
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<SdkMeterProvider>, 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<Point> {
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<Item = &KeyValue>| {
let mut all: Vec<String> =
attrs.map(|kv| format!("{}={}", kv.key, kv.value)).collect();
all.sort();
all.join(",")
};
let mut points: Vec<Point> = 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
}
}
}