From 7efcfd064492607ef0b9f015726b6f61c25798c7 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 4 Nov 2026 22:40:04 +0100 Subject: [PATCH] swarm-notify-router: export event, todo, review request and consumer metrics --- Cargo.lock | 3 + docs/swarm/notify-router.md | 24 ++ nix/host-modules/swarm-notify-router.nix | 9 + swarm-notify-router/Cargo.toml | 8 +- swarm-notify-router/src/main.rs | 37 ++- swarm-notify-router/src/metrics.rs | 281 ++++++++++++++++++++++ swarm-secret-client/Cargo.toml | 6 +- swarm-secret-client/src/lib.rs | 2 + swarm-secret-client/src/meter_provider.rs | 47 ++++ 9 files changed, 413 insertions(+), 4 deletions(-) create mode 100644 swarm-notify-router/src/metrics.rs create mode 100644 swarm-secret-client/src/meter_provider.rs diff --git a/Cargo.lock b/Cargo.lock index ba83fbab..57a9137f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5156,6 +5156,8 @@ dependencies = [ "async-nats", "futures-util", "hive-log", + "opentelemetry", + "opentelemetry_sdk", "serde", "serde_json", "swarm-queue-client", @@ -5217,6 +5219,7 @@ dependencies = [ "backon", "futures-util", "opentelemetry", + "opentelemetry-otlp", "opentelemetry_sdk", "reqwest", "rustify", diff --git a/docs/swarm/notify-router.md b/docs/swarm/notify-router.md index e8540cf2..c9216ca3 100644 --- a/docs/swarm/notify-router.md +++ b/docs/swarm/notify-router.md @@ -291,3 +291,27 @@ controller asks the forge under the token it already holds. The agent's create s carries the consumer's filter, and the queue server checks it against the request, so an agent can't create a consumer that delivers another agent's todos. + +## Observability + +The router pushes OTLP metrics through its container's forwarder to the swarm +collector, labelled `service.name=hyperhive-notify-router`. With no +`OTEL_EXPORTER_OTLP_ENDPOINT` it exports nothing. The client-certificate +gauges (`bao.client_cert.*`) come with them. + +| Metric | Unit | Kind | Attributes | +| ---------------------------------------------- | ---- | --------- | -------------------------------------------------------------------------------- | +| `hyperhive.notify_router.events` | — | counter | `event_type`; `outcome`: `routed`, `unmatched`, `ignored`, `unusable`, `retried` | +| `hyperhive.notify_router.event.duration` | `s` | histogram | `event_type` | +| `hyperhive.notify_router.todos` | — | counter | `op`: `upsert`, `clear`; `outcome`: `delivered`, `failed` | +| `hyperhive.notify_router.review_requests` | — | counter | `outcome`: `requested`, `skipped`, `failed` | +| `hyperhive.notify_router.pull.failures` | — | counter | — | +| `hyperhive.notify_router.consumer.pending` | — | gauge | events not yet delivered to the router's consumer | +| `hyperhive.notify_router.consumer.ack_pending` | — | gauge | events delivered and not yet acked | +| `hyperhive.notify_router.consumer.redelivered` | — | gauge | events delivered more than once | + +`ignored` is an event that names no issue or pull request. `unmatched` is one +no flow matched. `retried` is one the queue or the store failed on: it comes back +and counts again. A growing `consumer.pending` means the router is falling +behind. The consumer gauges are read at the start of each pull, at most 30 +seconds apart. diff --git a/nix/host-modules/swarm-notify-router.nix b/nix/host-modules/swarm-notify-router.nix index 83e43692..04c498d5 100644 --- a/nix/host-modules/swarm-notify-router.nix +++ b/nix/host-modules/swarm-notify-router.nix @@ -43,6 +43,9 @@ let # Loopback in the container's own netns, and distinct from the other service # containers' forwarders all the same. forwarderTelemetryPort = 8902; + # The standard OTLP/HTTP port, on the container's loopback: the forwarder + # takes the router's metrics there and pushes them to the swarm collector. + forwarderOtlpPort = 4318; rootfsCheck = import ./lib/container-rootfs-check.nix { inherit config; }; @@ -183,6 +186,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}"; @@ -210,6 +214,11 @@ in SWARM_NOTIFY_ROUTER_NATS_URL = "tls://${natsCfg.domain}:${toString natsCfg.port}"; SWARM_NOTIFY_ROUTER_OIDC_TOKEN_ENDPOINT = "${autheliaCfg.url}/api/oidc/token"; SWARM_NOTIFY_ROUTER_OIDC_CLIENT_ID = clientId; + OTEL_EXPORTER_OTLP_ENDPOINT = "http://127.0.0.1:${toString forwarderOtlpPort}"; + OTEL_EXPORTER_OTLP_PROTOCOL = "http/protobuf"; + OTEL_RESOURCE_ATTRIBUTES = "service.name=hyperhive-notify-router,swarm=${ + if hyperhiveCfg.swarm.name == null then "unknown" else hyperhiveCfg.swarm.name + }"; } // lib.optionalAttrs (storeFiles ? notify-router-bao-server-ca) { BAO_CACERT = credential "notify-router-bao-server-ca"; diff --git a/swarm-notify-router/Cargo.toml b/swarm-notify-router/Cargo.toml index ea67415e..57fcef82 100644 --- a/swarm-notify-router/Cargo.toml +++ b/swarm-notify-router/Cargo.toml @@ -15,6 +15,10 @@ async-nats = { workspace = true, features = ["kv"] } # StreamExt / TryStreamExt on the pulled batch and the subject listing. futures-util.workspace = true hive-log.workspace = true +# The metrics the router exports (`metrics`), pushed through the OTLP exporter +# `swarm-secret-client`'s `meter_provider` builds. +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true serde.workspace = true serde_json.workspace = true # The connect, and the two streams' configs it shares with the controller and @@ -24,7 +28,7 @@ swarm-queue-client = { workspace = true, features = [ "router-todo", ] } # Its queue client secret, read with its own store leaf. -swarm-secret-client.workspace = true +swarm-secret-client = { workspace = true, features = ["otel"] } # The stream, subject, bucket and op names. swarm-wire.workspace = true tokio.workspace = true @@ -32,6 +36,8 @@ tracing.workspace = true # A fresh directory per test `nats-server`, for its store and ports file. [dev-dependencies] +# `InMemoryMetricExporter`, to read back what `metrics` records. +opentelemetry_sdk = { workspace = true, features = ["testing"] } tempfile = "3" tracing-subscriber.workspace = true diff --git a/swarm-notify-router/src/main.rs b/swarm-notify-router/src/main.rs index 3758ce06..cdf73e4b 100644 --- a/swarm-notify-router/src/main.rs +++ b/swarm-notify-router/src/main.rs @@ -35,6 +35,7 @@ use std::time::{Duration, Instant}; use anyhow::{Context as _, Result, bail}; use async_nats::jetstream::{self, consumer::pull, kv, message::PublishMessage}; use futures_util::{StreamExt as _, TryStreamExt as _}; +use metrics::Handled; use swarm_queue_client::{ClientSecret, QueueConfig}; use swarm_secret_client::stored_secret::StoredSecret; use swarm_wire::router_api::Kind; @@ -46,6 +47,7 @@ mod dry_run; mod event; mod flow; mod log; +mod metrics; mod route; mod settle; mod stored; @@ -111,6 +113,7 @@ async fn main() -> Result<()> { ); loop { if let Err(e) = router.pull_once().await { + router.metrics.pull_failed(); tracing::warn!( error = format!("{e:#}"), "notify router: pulling events failed; retrying" @@ -127,6 +130,7 @@ struct Router { state: kv::Store, flows: stored::Flows, roster: Option<(Instant, BTreeSet)>, + metrics: metrics::Metrics, } impl Router { @@ -172,11 +176,16 @@ impl Router { state, flows: stored::Flows::builtins(), roster: None, + metrics: metrics::Metrics::from_env(), }) } /// Pull one batch and handle each event in stream order. async fn pull_once(&mut self) -> Result<()> { + if let Ok(info) = self.events.info().await { + self.metrics + .consumer(info.num_pending, info.num_ack_pending, info.num_redelivered); + } let batch = self .events .batch() @@ -189,13 +198,27 @@ impl Router { settle::batch(batch, async |m: &jetstream::Message| self.handle(m).await).await } + /// Handle one event and count how it went. async fn handle(&mut self, message: &jetstream::Message) -> Result<(), Skip> { + let started = Instant::now(); + let result = self.route_event(message).await; + let handled = match &result { + Ok(handled) => *handled, + Err(Skip::Unusable(_)) => Handled::Unusable, + Err(Skip::Retry(_)) => Handled::Retried, + }; + let event_type = forge_events::event_type(&message.subject).unwrap_or("unknown"); + self.metrics.event(event_type, handled, started.elapsed()); + result.map(|_| ()) + } + + async fn route_event(&mut self, message: &jetstream::Message) -> Result { let event_type = forge_events::event_type(&message.subject) .ok_or_else(|| Skip::Unusable("not a forge event subject".to_owned()))? .to_owned(); let Some(delivery) = event::Delivery::parse(&message.payload) else { // A delivery that names no issue or PR has nothing to route. - return Ok(()); + return Ok(Handled::Ignored); }; let seq = message .info() @@ -244,7 +267,11 @@ impl Router { for request in &routed.requests { self.request_review(&cause, request).await; } - Ok(()) + Ok(if flows.is_empty() { + Handled::Unmatched + } else { + Handled::Routed + }) } /// Ask swarm-controller for one review, log what it answered and record @@ -272,6 +299,11 @@ impl Router { }, Err(e) => Err(format!("no answer: {e}")), }; + self.metrics.review_request(match &result { + Ok(done) if done == "requested" => "requested", + Ok(_) => "skipped", + Err(_) => "failed", + }); log::sink(cause, flow, Kind::RequestReviewer, reviewer, &result); let row = cause.row(flow, &request.node, Kind::RequestReviewer, reviewer, result); log::append(&self.js, &row).await; @@ -281,6 +313,7 @@ impl Router { /// record the delivery, failed or not. async fn queue(&self, seq: u64, cause: &log::Cause, action: &flow::Action) -> Result<()> { let sent = self.publish(seq, action).await; + self.metrics.todo(op_name(action.op.op), sent.is_ok()); let result = match &sent { Ok(()) => Ok(format!("{} {}", op_name(action.op.op), action.op.key)), Err(e) => Err(format!("{e:#}")), diff --git a/swarm-notify-router/src/metrics.rs b/swarm-notify-router/src/metrics.rs new file mode 100644 index 00000000..5d98cfcd --- /dev/null +++ b/swarm-notify-router/src/metrics.rs @@ -0,0 +1,281 @@ +//! What the router exports for its dashboard, through the swarm collector. +//! +//! - `hyperhive.notify_router.events`: forge events taken off the stream, by +//! `event_type` and `outcome` (`routed`, `unmatched`, `ignored`, `unusable`, +//! `retried`). +//! - `hyperhive.notify_router.event.duration`: seconds to handle one event, by +//! `event_type`. +//! - `hyperhive.notify_router.todos`: todo ops queued for an agent, by `op` +//! (`upsert`, `clear`) and `outcome` (`delivered`, `failed`). +//! - `hyperhive.notify_router.review_requests`: review requests made to the +//! controller, by `outcome` (`requested`, `skipped`, `failed`). +//! - `hyperhive.notify_router.pull.failures`: pulls from the stream that failed. +//! - `hyperhive.notify_router.consumer.pending`, `.ack_pending`, `.redelivered`: +//! the router's own consumer on the events stream, read at the start of each +//! pull: events not yet delivered to it, delivered and not yet acked, and +//! delivered more than once. A growing `pending` is the router falling behind. +//! +//! The provider is built from the standard `OTEL_*` environment, so without a +//! collector the instruments do nothing. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use opentelemetry::KeyValue; +use opentelemetry::metrics::{Counter, Histogram, Meter, MeterProvider as _}; +use opentelemetry_sdk::metrics::SdkMeterProvider; + +const METER: &str = "hyperhive.notify_router"; + +/// What came of handling one event. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Handled { + /// Names no issue or pull request: nothing to route. + Ignored, + /// No flow matched. + Unmatched, + /// At least one flow matched and its ops are queued. + Routed, + /// Can never be handled; dropped for good. + Unusable, + /// The queue or the store failed; it comes back. + Retried, +} + +impl Handled { + const fn name(self) -> &'static str { + match self { + Self::Ignored => "ignored", + Self::Unmatched => "unmatched", + Self::Routed => "routed", + Self::Unusable => "unusable", + Self::Retried => "retried", + } + } +} + +/// The router's instruments, and the provider that exports them. +pub struct Metrics { + /// Kept alive: it exports only while it lives. + _provider: Option, + events: Counter, + duration: Histogram, + todos: Counter, + reviews: Counter, + pull_failures: Counter, + consumer: Arc, +} + +/// The consumer's counts as of the last read, for the gauges to report. +#[derive(Default)] +struct Consumer { + pending: AtomicU64, + ack_pending: AtomicU64, + redelivered: AtomicU64, +} + +impl Metrics { + /// The instruments on the provider the environment names, or no-op ones. + #[must_use] + pub fn from_env() -> Self { + let provider = swarm_secret_client::meter_provider::from_env(METER, "notify-router"); + let meter = provider + .as_ref() + .map_or_else(|| opentelemetry::global::meter(METER), |p| p.meter(METER)); + Self::on(&meter, provider) + } + + fn on(meter: &Meter, provider: Option) -> Self { + let consumer = Arc::new(Consumer::default()); + for (name, read) in [ + ( + "pending", + (|c: &Consumer| &c.pending) as fn(&Consumer) -> &AtomicU64, + ), + ("ack_pending", |c| &c.ack_pending), + ("redelivered", |c| &c.redelivered), + ] { + let consumer = Arc::clone(&consumer); + meter + .u64_observable_gauge(format!("{METER}.consumer.{name}")) + .with_callback(move |obs| obs.observe(read(&consumer).load(Ordering::Relaxed), &[])) + .build(); + } + Self { + _provider: provider, + events: meter.u64_counter(format!("{METER}.events")).build(), + duration: meter + .f64_histogram(format!("{METER}.event.duration")) + .with_unit("s") + .build(), + todos: meter.u64_counter(format!("{METER}.todos")).build(), + reviews: meter + .u64_counter(format!("{METER}.review_requests")) + .build(), + pull_failures: meter.u64_counter(format!("{METER}.pull.failures")).build(), + consumer, + } + } + + /// One event handled, after `took`. + pub fn event(&self, event_type: &str, handled: Handled, took: Duration) { + let event_type = KeyValue::new("event_type", event_type.to_owned()); + self.events.add( + 1, + &[event_type.clone(), KeyValue::new("outcome", handled.name())], + ); + self.duration.record(took.as_secs_f64(), &[event_type]); + } + + /// One todo op queued, or failed to be. + pub fn todo(&self, op: &'static str, delivered: bool) { + let outcome = if delivered { "delivered" } else { "failed" }; + self.todos.add( + 1, + &[KeyValue::new("op", op), KeyValue::new("outcome", outcome)], + ); + } + + /// One review request answered: `requested`, `skipped` or `failed`. + pub fn review_request(&self, outcome: &'static str) { + self.reviews.add(1, &[KeyValue::new("outcome", outcome)]); + } + + /// One pull from the stream failed. + pub fn pull_failed(&self) { + self.pull_failures.add(1, &[]); + } + + /// The consumer's counts, as the stream reports them. + pub fn consumer(&self, pending: u64, ack_pending: usize, redelivered: usize) { + let as_u64 = |n: usize| u64::try_from(n).unwrap_or(u64::MAX); + self.consumer.pending.store(pending, Ordering::Relaxed); + self.consumer + .ack_pending + .store(as_u64(ack_pending), Ordering::Relaxed); + self.consumer + .redelivered + .store(as_u64(redelivered), Ordering::Relaxed); + } +} + +#[cfg(test)] +mod tests { + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; + use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader}; + + use super::*; + + /// Metrics on an in-memory exporter, and the exporter. + fn recorded() -> (Metrics, SdkMeterProvider, InMemoryMetricExporter) { + let exporter = InMemoryMetricExporter::default(); + let reader = PeriodicReader::builder(exporter.clone()).build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let metrics = Metrics::on(&provider.meter("test"), None); + (metrics, provider, exporter) + } + + /// Every data point of the sum or gauge `name`, as its labels and value. + fn points(exporter: &InMemoryMetricExporter, name: &str) -> Vec<(Vec<(String, String)>, u64)> { + let mut out = Vec::new(); + for resource in exporter.get_finished_metrics().unwrap() { + for scope in resource.scope_metrics() { + for metric in scope.metrics().filter(|m| m.name() == name) { + let labels = |attrs: &mut dyn Iterator| { + attrs + .map(|kv| (kv.key.to_string(), kv.value.to_string())) + .collect::>() + }; + match metric.data() { + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + for p in sum.data_points() { + out.push((labels(&mut p.attributes()), p.value())); + } + } + AggregatedMetrics::U64(MetricData::Gauge(gauge)) => { + for p in gauge.data_points() { + out.push((labels(&mut p.attributes()), p.value())); + } + } + _ => {} + } + } + } + } + out + } + + fn pair(k: &str, v: &str) -> (String, String) { + (k.to_owned(), v.to_owned()) + } + + #[test] + fn an_event_counts_under_its_type_and_outcome() { + let (metrics, provider, exporter) = recorded(); + metrics.event("issues", Handled::Routed, Duration::from_millis(5)); + metrics.event("issues", Handled::Routed, Duration::from_millis(7)); + metrics.event("issues", Handled::Retried, Duration::from_millis(1)); + provider.force_flush().unwrap(); + let mut got = points(&exporter, "hyperhive.notify_router.events"); + got.sort(); + assert_eq!( + got, + vec![ + ( + vec![pair("event_type", "issues"), pair("outcome", "retried")], + 1 + ), + ( + vec![pair("event_type", "issues"), pair("outcome", "routed")], + 2 + ), + ] + ); + } + + #[test] + fn todos_and_review_requests_count_by_outcome() { + let (metrics, provider, exporter) = recorded(); + metrics.todo("upsert", true); + metrics.todo("clear", false); + metrics.review_request("requested"); + metrics.pull_failed(); + provider.force_flush().unwrap(); + let mut todos = points(&exporter, "hyperhive.notify_router.todos"); + todos.sort(); + assert_eq!( + todos, + vec![ + (vec![pair("op", "clear"), pair("outcome", "failed")], 1), + (vec![pair("op", "upsert"), pair("outcome", "delivered")], 1), + ] + ); + assert_eq!( + points(&exporter, "hyperhive.notify_router.review_requests"), + vec![(vec![pair("outcome", "requested")], 1)] + ); + assert_eq!( + points(&exporter, "hyperhive.notify_router.pull.failures"), + vec![(vec![], 1)] + ); + } + + #[test] + fn the_consumer_gauges_report_the_last_counts() { + let (metrics, provider, exporter) = recorded(); + metrics.consumer(12, 3, 1); + metrics.consumer(4, 0, 1); + provider.force_flush().unwrap(); + for (name, want) in [("pending", 4), ("ack_pending", 0), ("redelivered", 1)] { + assert_eq!( + points( + &exporter, + &format!("hyperhive.notify_router.consumer.{name}") + ), + vec![(vec![], want)], + "{name}" + ); + } + } +} diff --git a/swarm-secret-client/Cargo.toml b/swarm-secret-client/Cargo.toml index ab36b333..5904e440 100644 --- a/swarm-secret-client/Cargo.toml +++ b/swarm-secret-client/Cargo.toml @@ -16,6 +16,10 @@ reqwest.workspace = true # `cert_metrics`: the `bao.client_cert.*` gauges, for the services that export # OTLP. Off by default so the other dependents link no OTel. opentelemetry = { workspace = true, optional = true } +# `meter_provider`: the provider those services build from the standard OTEL +# environment, and its OTLP push. +opentelemetry_sdk = { workspace = true, optional = true } +opentelemetry-otlp = { workspace = true, optional = true } # vaultrs's own endpoint machinery, for the one endpoint it does not # implement — see `client::WriteAclPolicy`. rustify.workspace = true @@ -54,4 +58,4 @@ tokio = { workspace = true, features = ["test-util"] } tracing-subscriber.workspace = true [features] -otel = ["dep:opentelemetry"] +otel = ["dep:opentelemetry", "dep:opentelemetry_sdk", "dep:opentelemetry-otlp"] diff --git a/swarm-secret-client/src/lib.rs b/swarm-secret-client/src/lib.rs index 05c05465..9a53d440 100644 --- a/swarm-secret-client/src/lib.rs +++ b/swarm-secret-client/src/lib.rs @@ -45,6 +45,8 @@ pub mod github; pub mod gitlab; pub mod grafana; pub mod matrix; +#[cfg(feature = "otel")] +pub mod meter_provider; pub mod mtls; pub mod observability; pub mod path; diff --git a/swarm-secret-client/src/meter_provider.rs b/swarm-secret-client/src/meter_provider.rs new file mode 100644 index 00000000..d7af95e9 --- /dev/null +++ b/swarm-secret-client/src/meter_provider.rs @@ -0,0 +1,47 @@ +//! The meter provider a swarm service pushes its OTLP metrics through. +//! +//! Built from the standard `OTEL_*` environment: the endpoint from +//! `OTEL_EXPORTER_OTLP_ENDPOINT` and the resource from +//! `OTEL_RESOURCE_ATTRIBUTES`. With no endpoint a service exports nothing, so +//! a run outside the swarm (a test, a dev shell) needs no collector. + +use std::time::Duration; + +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); + +/// The provider for the service `cert` names in the `bao.client_cert.*` gauges +/// it also registers on a meter called `meter`, or `None` when no collector is +/// configured or the exporter failed to build. +/// +/// The caller keeps the provider alive: it exports only while it lives. +#[must_use] +pub fn from_env(meter: &'static str, cert: &'static str) -> Option { + use opentelemetry::metrics::MeterProvider as _; + + if !std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT").is_ok_and(|s| !s.trim().is_empty()) { + return None; + } + // The blocking client: `PeriodicReader` exports from a thread of its own + // with no Tokio reactor. + let exporter = match MetricExporter::builder() + .with_http() + .with_protocol(Protocol::HttpJson) + .build() + { + Ok(exporter) => exporter, + Err(e) => { + tracing::warn!(error = %e, meter, "building the metric exporter failed; no metrics are exported"); + return None; + } + }; + let reader = PeriodicReader::builder(exporter) + .with_interval(INTERVAL) + .build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + crate::cert_metrics::register(&provider.meter(meter), cert); + Some(provider) +}