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

swarm-notify-router: export event, todo, review request and consumer metrics

Browse source
This commit is contained in:
damocles 2026-11-04 22:40:04 +01:00 • committed by mara
commit 7efcfd0644
9 changed files with 413 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

3
Cargo.lock generated
View file

@ -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",

24
docs/swarm/notify-router.md
View file

@ -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.

9
nix/host-modules/swarm-notify-router.nix
View file

@ -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";

8
swarm-notify-router/Cargo.toml
View file

@ -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

37
swarm-notify-router/src/main.rs
View file

@ -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<String>)>,
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<Handled, Skip> {
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:#}")),

281
swarm-notify-router/src/metrics.rs Normal file
View file

@ -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<SdkMeterProvider>,
events: Counter<u64>,
duration: Histogram<f64>,
todos: Counter<u64>,
reviews: Counter<u64>,
pull_failures: Counter<u64>,
consumer: Arc<Consumer>,
}
/// 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<SdkMeterProvider>) -> 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<Item = &KeyValue>| {
attrs
.map(|kv| (kv.key.to_string(), kv.value.to_string()))
.collect::<Vec<_>>()
};
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}"
);
}
}
}

6
swarm-secret-client/Cargo.toml
View file

@ -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"]

2
swarm-secret-client/src/lib.rs
View file

@ -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;

47
swarm-secret-client/src/meter_provider.rs Normal file
View file

@ -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<SdkMeterProvider> {
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)
}