//! 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 } } }