From 1f5a7b71ed807960b386e1de96175e06c0f52d4e Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 19 Aug 2026 17:22:55 +0200 Subject: [PATCH 1/3] swarm-controller: export jobq state rollup as OTEL gauges --- Cargo.lock | 3 + swarm-controller/Cargo.toml | 14 ++ swarm-controller/src/jobq_metrics.rs | 259 +++++++++++++++++++++++++++ swarm-controller/src/main.rs | 2 + 4 files changed, 278 insertions(+) create mode 100644 swarm-controller/src/jobq_metrics.rs diff --git a/Cargo.lock b/Cargo.lock index a3710f58..bb9d6699 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4561,6 +4561,9 @@ dependencies = [ "hive-jobq-wire", "hive-types", "hmac 0.13.0", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", "problem_details", "reqwest", "serde", diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index cb7e0419..72a6f361 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -38,6 +38,20 @@ problem_details = { version = "0.9.0", features = ["axum"] } # same shape `hive-c0re/src/job_queue/scheduler.rs` uses over its own graph. hive-jobq.workspace = true hive-jobq-wire.workspace = true +# OTEL SDK for the jobq-rollup metrics exporter (`jobq_metrics.rs`). Same +# versions + blocking-client rationale as `hive-c0re`/`hive-metric`: the +# metrics SDK's `PeriodicReader` runs on a background thread with no Tokio +# reactor, where the async client panics. Kept local rather than in +# `workspace.dependencies` — same call `hive-metric` already made, this is +# still the only other crate that needs it. +opentelemetry = "0.32" +opentelemetry_sdk = { version = "0.32", features = ["metrics"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = [ + "metrics", + "http-json", + "reqwest-blocking-client", + "reqwest-rustls", +] } # The forge webhook HMAC (`webhook.rs`). Kept in this crate rather than # shared with hive-c0re's equivalent: c0re's copy is scheduled to be deleted # with its webhook routes once registration moves here, so the second holder diff --git a/swarm-controller/src/jobq_metrics.rs b/swarm-controller/src/jobq_metrics.rs new file mode 100644 index 00000000..a734bba5 --- /dev/null +++ b/swarm-controller/src/jobq_metrics.rs @@ -0,0 +1,259 @@ +//! OTEL export of the swarm-level job graph's state rollup — the same +//! `hive_jobq_wire::state_rollup()` counts `GET /api/jobq/rollup` already +//! serves, ridden out to the collector on a timer instead of only on +//! request. First consumer of a generic "add jobq metrics" ask: the same +//! shape works for any `hive-jobq` instance, this crate's is just the +//! first to wire it. +//! +//! Same OTEL SDK setup as `hive-c0re::stats::otel_metrics` (container +//! stats) and `hive-metric` (the one-shot CLI): the metrics SDK's +//! `PeriodicReader` drives its export from a background thread with no +//! Tokio reactor, so the blocking OTLP client is deliberate here too, not +//! an oversight. +//! +//! Bridging async→sync: reading `state_rollup()` needs the scheduler's +//! `std::sync::Mutex`, which is fine to lock briefly from a sync callback — +//! but holding it for the whole OTLP export would block every request +//! handler that also locks it for however long the export takes. So, same +//! pattern as the container-stats exporter: an async task refreshes a +//! shared snapshot on an interval, and the (sync) SDK callbacks only ever +//! read that snapshot, never the scheduler directly. + +use std::sync::{Arc, Mutex, OnceLock}; +use std::time::Duration; + +use anyhow::{Context, Result}; +use hive_jobq_wire::StateCount; +use opentelemetry::KeyValue; +use opentelemetry::metrics::MeterProvider as _; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; +use opentelemetry_sdk::Resource; +use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; + +use crate::{SwarmNodeKind, SwarmResourceKind}; + +/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset. +/// Matches `hive-c0re::stats::otel_metrics`'s default — no reason for the +/// two exporters to disagree on how fresh "current" means by default. +const DEFAULT_INTERVAL: Duration = Duration::from_mins(1); + +/// Latest rollup snapshot: written by the async refresher, read by the sync +/// observable-instrument callbacks. +static SNAPSHOT: OnceLock>>> = OnceLock::new(); + +/// Keep the provider alive for the process lifetime — the `PeriodicReader` +/// exports only while the provider lives. +static PROVIDER: OnceLock = OnceLock::new(); + +/// Spawn the jobq-rollup OTEL exporter if OTEL is configured +/// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty). No-op otherwise — same +/// graceful-absence shape every other optional wiring in this daemon uses +/// (queue, bridge, forge, webhook secret). Call once at startup. +pub fn spawn_exporter( + jobq: Arc>>, +) { + let Some(endpoint) = endpoint() else { + tracing::debug!("otel jobq-metrics: no endpoint configured, exporter disabled"); + return; + }; + let snapshot = SNAPSHOT + .get_or_init(|| Arc::new(Mutex::new(Vec::new()))) + .clone(); + let interval = interval(); + + let refresh = snapshot.clone(); + tokio::spawn(async move { + loop { + let fresh = { + let sched = jobq + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let graph = sched.graph(); + let roots: Vec = graph.roots().map(|n| n.id).collect(); + hive_jobq_wire::state_rollup(graph, roots) + }; + if let Ok(mut g) = refresh.lock() { + *g = fresh; + } + tokio::time::sleep(interval).await; + } + }); + + match build_provider(interval, snapshot) { + Ok(provider) => { + let _ = PROVIDER.set(provider); + tracing::info!(%endpoint, ?interval, "otel jobq-metrics: exporter enabled"); + } + Err(e) => { + tracing::warn!(error = ?e, "otel jobq-metrics: exporter init failed"); + } + } +} + +fn build_provider( + interval: Duration, + snapshot: Arc>>, +) -> Result { + // Same http/json, endpoint-from-env-only construction as + // `hive-c0re::stats::otel_metrics` — see that module's doc comment for + // why `with_endpoint` is deliberately never called here. + let exporter = MetricExporter::builder() + .with_http() + .with_protocol(Protocol::HttpJson) + .build() + .context("build OTLP metric exporter")?; + let reader = PeriodicReader::builder(exporter) + .with_interval(interval) + .build(); + let provider = SdkMeterProvider::builder() + .with_reader(reader) + .with_resource(resource()) + .build(); + register_instruments(&provider, snapshot); + Ok(provider) +} + +/// Register the two observable gauges — `jobq.nodes` (every node at any +/// depth) and `jobq.roots` (only the root groups), each with a `state` +/// attribute — mirroring `StateCount`'s own two-count shape (see its doc +/// comment for why both numbers are kept rather than collapsed to one). +fn register_instruments(provider: &SdkMeterProvider, snapshot: Arc>>) { + let meter = provider.meter("hyperhive.jobq"); + + let snap = snapshot.clone(); + meter + .u64_observable_gauge("jobq.nodes") + .with_description("hive-jobq nodes by lifecycle state, at any depth") + .with_callback(move |obs| { + if let Ok(g) = snap.lock() { + for c in g.iter() { + obs.observe(c.nodes, &[state_attr(c)]); + } + } + }) + .build(); + + let snap = snapshot; + meter + .u64_observable_gauge("jobq.roots") + .with_description("hive-jobq root groups by lifecycle state") + .with_callback(move |obs| { + if let Ok(g) = snap.lock() { + for c in g.iter() { + obs.observe(c.roots, &[state_attr(c)]); + } + } + }) + .build(); +} + +/// The `state` attribute for one data point, spelled from `Debug` — the +/// `State` enum has no `Display`, and `Debug` on a fieldless variant is +/// exactly the variant name (`Pending`, `Running`, ...), so this costs no +/// new formatting code to keep in sync with the enum. +fn state_attr(c: &StateCount) -> KeyValue { + KeyValue::new("state", format!("{:?}", c.state)) +} + +/// Resource: `service.name = swarm-controller` plus whatever the operator +/// set in `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` (same channel every +/// other OTEL exporter in this tree reads its `hive`/`swarm` labels from — +/// this daemon has no single "hive" of its own, so nothing is assumed here +/// beyond what the operator supplies). +fn resource() -> Resource { + let mut builder = Resource::builder().with_service_name("swarm-controller"); + for (k, v) in resource_attributes() { + builder = builder.with_attribute(KeyValue::new(k, v)); + } + builder.build() +} + +/// `OTEL_EXPORTER_OTLP_ENDPOINT`, non-empty. The enable signal — and the +/// very variable the SDK reads to build the exporter's URL, so "configured" +/// and "where it goes" cannot disagree. See +/// `hive-c0re::stats::otel_metrics::endpoint`'s doc comment for why this is +/// the only variable allowed to gate this exporter. +fn endpoint() -> Option { + std::env::var("OTEL_EXPORTER_OTLP_ENDPOINT") + .ok() + .map(|s| s.trim().to_owned()) + .filter(|s| !s.is_empty()) +} + +/// Export cadence from `HYPERHIVE_OTEL_METRIC_INTERVAL_MS`, else the +/// default. Same variable `hive-c0re::stats::otel_metrics` reads — one +/// cadence knob for every OTEL exporter in the tree, not a new one per +/// module. +fn interval() -> Duration { + std::env::var("HYPERHIVE_OTEL_METRIC_INTERVAL_MS") + .ok() + .and_then(|s| s.trim().parse::().ok()) + .filter(|ms| *ms > 0) + .map_or(DEFAULT_INTERVAL, Duration::from_millis) +} + +/// Extra resource attributes from `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` +/// (`key=value,key=value`), same env `hive-c0re::stats::otel_metrics` reads. +fn resource_attributes() -> Vec<(String, String)> { + std::env::var("HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES") + .ok() + .map(|s| parse_kv(&s)) + .unwrap_or_default() +} + +/// Parse `key=value` pairs separated by commas and/or newlines. The value +/// keeps any `=` after the first (so `Authorization=Bearer x=y` → `Bearer +/// x=y`). Byte-identical logic to `hive-c0re::stats::otel_metrics::parse_kv` +/// — small enough that a shared crate for it would cost more than it saves, +/// but kept in lockstep on purpose. +fn parse_kv(s: &str) -> Vec<(String, String)> { + s.split([',', '\n']) + .filter_map(|pair| { + let pair = pair.trim(); + if pair.is_empty() { + return None; + } + let (k, v) = pair.split_once('=')?; + let k = k.trim(); + if k.is_empty() { + None + } else { + Some((k.to_owned(), v.trim().to_owned())) + } + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use hive_jobq::State; + + #[test] + fn parse_kv_keeps_bearer_value() { + assert_eq!( + parse_kv("Authorization=Bearer abc=def"), + vec![("Authorization".to_owned(), "Bearer abc=def".to_owned())] + ); + assert_eq!( + parse_kv("a=1,\n b=2 ,=bad,"), + vec![ + ("a".to_owned(), "1".to_owned()), + ("b".to_owned(), "2".to_owned()) + ] + ); + } + + /// `state_attr` must spell the exact variant name — a Grafana query + /// filtering `state="Running"` should match what this daemon actually + /// emits, not a re-cased or re-worded version of it. + #[test] + fn state_attr_spells_the_variant_name() { + let c = StateCount { + state: State::Running, + nodes: 3, + roots: 1, + }; + assert_eq!(state_attr(&c), KeyValue::new("state", "Running")); + } +} diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index d1423759..bc86bc5a 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -37,6 +37,7 @@ use utoipa_axum::{router::OpenApiRouter, routes}; mod auth; mod forge; +mod jobq_metrics; mod status; mod webhook; @@ -916,6 +917,7 @@ async fn main() -> Result<()> { hive_jobq::resources::ResourceTable::new(), ))); spawn_jobq_worker(Arc::clone(&jobq), deps); + jobq_metrics::spawn_exporter(Arc::clone(&jobq)); // Same "log and carry on" shape as the queue/bridge/forge wiring above. // A controller that cannot hold a webhook secret still serves every From 621245306f3a86f1da96335be37942cc90fde371 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 19 Aug 2026 17:32:38 +0200 Subject: [PATCH 2/3] swarm-controller: move OTEL deps to workspace level, drop jobq_metrics singleton --- Cargo.toml | 15 ++++++ hive-c0re/Cargo.toml | 20 +++----- hive-metric/Cargo.toml | 14 ++---- swarm-controller/Cargo.toml | 20 +++----- swarm-controller/src/jobq_metrics.rs | 73 +++++++++++++++++----------- swarm-controller/src/main.rs | 6 ++- 6 files changed, 81 insertions(+), 67 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index f7445c9a..571b3455 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -127,6 +127,21 @@ reqwest = { version = "0.13", default-features = false, features = [ ] } hyper = { version = "1", features = ["client", "http1"] } hyper-util = { version = "0.1", features = ["tokio"] } +# OTEL SDK, shared by every crate that pushes metrics (hive-c0re, hive-metric, +# swarm-controller). The blocking OTLP client is deliberate everywhere it's +# used: the metrics SDK's `PeriodicReader` drives its export from a +# background thread with no Tokio reactor, where the async client panics. +# `default-features = false` on the OTLP crate drops its own diagnostics +# ("internal-logs") unless a member opts back in — hive-c0re does, via +# `{ workspace = true, features = ["internal-logs"] }`. +opentelemetry = "0.32" +opentelemetry_sdk = { version = "0.32", features = ["metrics"] } +opentelemetry-otlp = { version = "0.32", default-features = false, features = [ + "metrics", + "http-json", + "reqwest-blocking-client", + "reqwest-rustls", +] } http-body-util = "0.1" # ⚠️ Keep at 0.11.1 or newer, and keep it on the SAME reqwest as everything # else. 0.11.0 links reqwest 0.12 while the workspace is on 0.13, and cargo diff --git a/hive-c0re/Cargo.toml b/hive-c0re/Cargo.toml index 8f23a07a..6ffc9549 100644 --- a/hive-c0re/Cargo.toml +++ b/hive-c0re/Cargo.toml @@ -23,20 +23,12 @@ clap.workspace = true clap_complete.workspace = true clap-markdown = "0.1" # OTEL SDK for the per-agent container-resource metrics exporter -# (stats/otel_metrics.rs). Same versions as hive-metric — the blocking OTLP -# client is deliberate: the metrics SDK's PeriodicReader runs on a background -# thread with no Tokio reactor, where the async client panics. -opentelemetry = "0.32" -opentelemetry_sdk = { version = "0.32", features = ["metrics"] } -opentelemetry-otlp = { version = "0.32", default-features = false, features = [ - # Not a transport: it is in `default`, so `default-features = false` drops the - # exporter's own diagnostics unless it is named here. - "internal-logs", - "metrics", - "http-json", - "reqwest-blocking-client", - "reqwest-rustls", -] } +# (stats/otel_metrics.rs). "internal-logs" on top of the workspace base is +# this crate's own opt-in: it's not a transport, it's in `default`, and +# `default-features = false` upstream drops it unless named here. +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true +opentelemetry-otlp = { workspace = true, features = ["internal-logs"] } indicatif.workspace = true hive-core-agent-sock.workspace = true hive-sh4re.workspace = true diff --git a/hive-metric/Cargo.toml b/hive-metric/Cargo.toml index e066970c..c6f6e045 100644 --- a/hive-metric/Cargo.toml +++ b/hive-metric/Cargo.toml @@ -11,9 +11,8 @@ path = "src/main.rs" [dependencies] anyhow.workspace = true clap.workspace = true -# OTEL SDK — only used by this crate, so kept local rather than in workspace.dependencies. -opentelemetry = "0.32" -opentelemetry_sdk = { version = "0.32", features = ["metrics"] } +# OTEL SDK, from the workspace base — see the root Cargo.toml's comment for +# the blocking-client rationale shared by every OTEL-pushing crate. # Blocking (not async) reqwest client on purpose: the metrics SDK drives the # OTLP push from a `PeriodicReader` background thread that has no Tokio runtime, # so the async client panics there with "no reactor running". The blocking @@ -21,9 +20,6 @@ opentelemetry_sdk = { version = "0.32", features = ["metrics"] } # records one point then `shutdown()`s, a synchronous send is exactly right. # No `internal-logs` here, unlike hive-c0re: this binary installs no tracing # subscriber, so the SDK's diagnostics would have nowhere to go. -opentelemetry-otlp = { version = "0.32", default-features = false, features = [ - "metrics", - "http-json", - "reqwest-blocking-client", - "reqwest-rustls", -] } +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true +opentelemetry-otlp.workspace = true diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 72a6f361..b1dba490 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -38,20 +38,12 @@ problem_details = { version = "0.9.0", features = ["axum"] } # same shape `hive-c0re/src/job_queue/scheduler.rs` uses over its own graph. hive-jobq.workspace = true hive-jobq-wire.workspace = true -# OTEL SDK for the jobq-rollup metrics exporter (`jobq_metrics.rs`). Same -# versions + blocking-client rationale as `hive-c0re`/`hive-metric`: the -# metrics SDK's `PeriodicReader` runs on a background thread with no Tokio -# reactor, where the async client panics. Kept local rather than in -# `workspace.dependencies` — same call `hive-metric` already made, this is -# still the only other crate that needs it. -opentelemetry = "0.32" -opentelemetry_sdk = { version = "0.32", features = ["metrics"] } -opentelemetry-otlp = { version = "0.32", default-features = false, features = [ - "metrics", - "http-json", - "reqwest-blocking-client", - "reqwest-rustls", -] } +# OTEL SDK for the jobq-rollup metrics exporter (`jobq_metrics.rs`), from +# the workspace base — see the root Cargo.toml's comment for the +# blocking-client rationale shared by every OTEL-pushing crate. +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true +opentelemetry-otlp.workspace = true # The forge webhook HMAC (`webhook.rs`). Kept in this crate rather than # shared with hive-c0re's equivalent: c0re's copy is scheduled to be deleted # with its webhook routes once registration moves here, so the second holder diff --git a/swarm-controller/src/jobq_metrics.rs b/swarm-controller/src/jobq_metrics.rs index a734bba5..8dcc185f 100644 --- a/swarm-controller/src/jobq_metrics.rs +++ b/swarm-controller/src/jobq_metrics.rs @@ -1,9 +1,12 @@ -//! OTEL export of the swarm-level job graph's state rollup — the same +//! OTEL export of a job graph's state rollup — the same //! `hive_jobq_wire::state_rollup()` counts `GET /api/jobq/rollup` already //! serves, ridden out to the collector on a timer instead of only on -//! request. First consumer of a generic "add jobq metrics" ask: the same -//! shape works for any `hive-jobq` instance, this crate's is just the -//! first to wire it. +//! request. First consumer of a generic "add jobq metrics" ask: [`spawn_exporter`] +//! is generic over any `hive_jobq::scheduler::Scheduler`, not tied to +//! this crate's own `SwarmNodeKind`/`SwarmResourceKind` — swarm-controller's +//! is just the first instance to wire it. No process-global state either +//! (see [`spawn_exporter`]'s doc comment): call it once per graph you want +//! exported, from as many call sites as you like. //! //! Same OTEL SDK setup as `hive-c0re::stats::otel_metrics` (container //! stats) and `hive-metric` (the one-shot CLI): the metrics SDK's @@ -19,7 +22,8 @@ //! shared snapshot on an interval, and the (sync) SDK callbacks only ever //! read that snapshot, never the scheduler directly. -use std::sync::{Arc, Mutex, OnceLock}; +use std::hash::Hash; +use std::sync::{Arc, Mutex}; use std::time::Duration; use anyhow::{Context, Result}; @@ -30,35 +34,39 @@ use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; -use crate::{SwarmNodeKind, SwarmResourceKind}; - /// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset. /// Matches `hive-c0re::stats::otel_metrics`'s default — no reason for the /// two exporters to disagree on how fresh "current" means by default. const DEFAULT_INTERVAL: Duration = Duration::from_mins(1); -/// Latest rollup snapshot: written by the async refresher, read by the sync -/// observable-instrument callbacks. -static SNAPSHOT: OnceLock>>> = OnceLock::new(); - -/// Keep the provider alive for the process lifetime — the `PeriodicReader` -/// exports only while the provider lives. -static PROVIDER: OnceLock = OnceLock::new(); - -/// Spawn the jobq-rollup OTEL exporter if OTEL is configured -/// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty). No-op otherwise — same +/// Spawn a jobq-rollup OTEL exporter for `jobq` if OTEL is configured +/// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty) — `None` otherwise, same /// graceful-absence shape every other optional wiring in this daemon uses -/// (queue, bridge, forge, webhook secret). Call once at startup. -pub fn spawn_exporter( - jobq: Arc>>, -) { - let Some(endpoint) = endpoint() else { - tracing::debug!("otel jobq-metrics: no endpoint configured, exporter disabled"); - return; - }; - let snapshot = SNAPSHOT - .get_or_init(|| Arc::new(Mutex::new(Vec::new()))) - .clone(); +/// (queue, bridge, forge, webhook secret). +/// +/// **No process-global state.** Earlier drafts of this held the snapshot and +/// the `SdkMeterProvider` behind `static OnceLock`s, which quietly made this +/// a swarm-controller singleton — a second call (a second graph, a test, a +/// future host with more than one jobq instance) would have silently reused +/// the first call's state instead of exporting its own. The snapshot is now +/// a plain local `Arc`, and the provider is returned rather than stashed — +/// **the caller owns it and must keep it alive** (bind it to a named +/// variable, not `_`) for as long as export should continue; dropping it +/// stops the `PeriodicReader`. +/// +/// Generic over `N`/`R` (the same parameters `Scheduler` itself takes) +/// rather than this crate's own `SwarmNodeKind`/`SwarmResourceKind` — the +/// whole point of the ask was a shape any `hive-jobq` host can plug in, not +/// one hardcoded to this crate's node/resource vocabulary. +pub fn spawn_exporter( + jobq: Arc>>, +) -> Option +where + N: Send + 'static, + R: Clone + Eq + Hash + Send + 'static, +{ + let endpoint = endpoint()?; + let snapshot: Arc>> = Arc::new(Mutex::new(Vec::new())); let interval = interval(); let refresh = snapshot.clone(); @@ -81,11 +89,12 @@ pub fn spawn_exporter( match build_provider(interval, snapshot) { Ok(provider) => { - let _ = PROVIDER.set(provider); tracing::info!(%endpoint, ?interval, "otel jobq-metrics: exporter enabled"); + Some(provider) } Err(e) => { tracing::warn!(error = ?e, "otel jobq-metrics: exporter init failed"); + None } } } @@ -160,6 +169,12 @@ fn state_attr(c: &StateCount) -> KeyValue { /// other OTEL exporter in this tree reads its `hive`/`swarm` labels from — /// this daemon has no single "hive" of its own, so nothing is assumed here /// beyond what the operator supplies). +/// +/// Hardcodes `swarm-controller` even though [`spawn_exporter`] is generic — +/// a future non-swarm-controller caller of this same function would want a +/// different `service.name`, but that's a real parameter to add when a +/// second caller actually exists, not a guess to make now for one that +/// doesn't. fn resource() -> Resource { let mut builder = Resource::builder().with_service_name("swarm-controller"); for (k, v) in resource_attributes() { diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index bc86bc5a..247f85a2 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -917,7 +917,11 @@ async fn main() -> Result<()> { hive_jobq::resources::ResourceTable::new(), ))); spawn_jobq_worker(Arc::clone(&jobq), deps); - jobq_metrics::spawn_exporter(Arc::clone(&jobq)); + // Bound to a named variable, not `_` — dropping the provider stops its + // `PeriodicReader`, so it must live as long as `main` does (which it + // does here: this binding never goes out of scope before the process + // exits via `axum::serve(...).await` below). + let _jobq_metrics_provider = jobq_metrics::spawn_exporter(Arc::clone(&jobq)); // Same "log and carry on" shape as the queue/bridge/forge wiring above. // A controller that cannot hold a webhook secret still serves every From f5ff8698f33f476974e3703d4f3a4120502d88fa Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 19 Aug 2026 17:52:12 +0200 Subject: [PATCH 3/3] swarm-controller: extract jobq_metrics into its own hive-jobq-metrics crate --- Cargo.lock | 18 +++- Cargo.toml | 2 + hive-jobq-metrics/Cargo.toml | 18 ++++ hive-jobq-metrics/README.md | 20 +++++ .../src/lib.rs | 85 ++++++++----------- swarm-controller/Cargo.toml | 12 +-- swarm-controller/src/main.rs | 4 +- 7 files changed, 100 insertions(+), 59 deletions(-) create mode 100644 hive-jobq-metrics/Cargo.toml create mode 100644 hive-jobq-metrics/README.md rename swarm-controller/src/jobq_metrics.rs => hive-jobq-metrics/src/lib.rs (71%) diff --git a/Cargo.lock b/Cargo.lock index bb9d6699..d95ae5b5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1785,6 +1785,20 @@ dependencies = [ "uuid", ] +[[package]] +name = "hive-jobq-metrics" +version = "0.1.0" +dependencies = [ + "anyhow", + "hive-jobq", + "hive-jobq-wire", + "opentelemetry", + "opentelemetry-otlp", + "opentelemetry_sdk", + "tokio", + "tracing", +] + [[package]] name = "hive-jobq-wire" version = "0.1.0" @@ -4558,12 +4572,10 @@ dependencies = [ "forgejo-api", "futures-util", "hive-jobq", + "hive-jobq-metrics", "hive-jobq-wire", "hive-types", "hmac 0.13.0", - "opentelemetry", - "opentelemetry-otlp", - "opentelemetry_sdk", "problem_details", "reqwest", "serde", diff --git a/Cargo.toml b/Cargo.toml index 571b3455..f2c58eb9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "hive-forge-notify", "hive-host-sock", "hive-jobq", + "hive-jobq-metrics", "hive-jobq-wire", "hive-matrix-mcp", "hive-metric", @@ -80,6 +81,7 @@ indicatif = "0.18" hive-sh4re = { path = "hive-sh4re" } hive-agent-sock = { path = "hive-agent-sock" } hive-jobq = { path = "hive-jobq" } +hive-jobq-metrics = { path = "hive-jobq-metrics" } hive-jobq-wire = { path = "hive-jobq-wire" } hive-core-agent-sock = { path = "hive-core-agent-sock" } hive-claude = "0.1" diff --git a/hive-jobq-metrics/Cargo.toml b/hive-jobq-metrics/Cargo.toml new file mode 100644 index 00000000..e48094e3 --- /dev/null +++ b/hive-jobq-metrics/Cargo.toml @@ -0,0 +1,18 @@ +[package] +name = "hive-jobq-metrics" +edition.workspace = true +version.workspace = true +readme = "README.md" + +[lints] +workspace = true + +[dependencies] +anyhow.workspace = true +hive-jobq.workspace = true +hive-jobq-wire.workspace = true +opentelemetry.workspace = true +opentelemetry_sdk.workspace = true +opentelemetry-otlp.workspace = true +tokio.workspace = true +tracing.workspace = true diff --git a/hive-jobq-metrics/README.md b/hive-jobq-metrics/README.md new file mode 100644 index 00000000..4996f34f --- /dev/null +++ b/hive-jobq-metrics/README.md @@ -0,0 +1,20 @@ +# hive-jobq-metrics + +OTEL export of a [`hive-jobq`](../hive-jobq) graph's state rollup, using the +same [`hive-jobq-wire`](../hive-jobq-wire) `state_rollup()` counts a viewer's +`/rollup` endpoint would serve — ridden out to the collector on a timer +instead of only on request. + +**Why this is not part of `hive-jobq` or `hive-jobq-wire`.** Both of those +crates are dependency-light on purpose (no `tokio`, no HTTP client) — the +scheduler is logic, the wire crate is presentation, and neither wants to drag +the OTEL SDK, an async runtime, and an OTLP HTTP client into every consumer +that just wants to run a graph or serialize one to JSON. Metrics export is a +third concern with its own weight, so it gets its own crate rather than +bloating either of theirs. + +`spawn_exporter` is generic over `hive_jobq::scheduler::Scheduler` — any +host's jobq instance can call it, not just one hardcoded caller. It holds no +process-global state: call it once per graph you want exported, and keep the +returned `SdkMeterProvider` alive for as long as export should continue +(dropping it stops the `PeriodicReader`). diff --git a/swarm-controller/src/jobq_metrics.rs b/hive-jobq-metrics/src/lib.rs similarity index 71% rename from swarm-controller/src/jobq_metrics.rs rename to hive-jobq-metrics/src/lib.rs index 8dcc185f..b927ceef 100644 --- a/swarm-controller/src/jobq_metrics.rs +++ b/hive-jobq-metrics/src/lib.rs @@ -1,12 +1,8 @@ -//! OTEL export of a job graph's state rollup — the same -//! `hive_jobq_wire::state_rollup()` counts `GET /api/jobq/rollup` already -//! serves, ridden out to the collector on a timer instead of only on -//! request. First consumer of a generic "add jobq metrics" ask: [`spawn_exporter`] -//! is generic over any `hive_jobq::scheduler::Scheduler`, not tied to -//! this crate's own `SwarmNodeKind`/`SwarmResourceKind` — swarm-controller's -//! is just the first instance to wire it. No process-global state either -//! (see [`spawn_exporter`]'s doc comment): call it once per graph you want -//! exported, from as many call sites as you like. +//! OTEL export of a [`hive_jobq`] graph's state rollup — the same +//! [`hive_jobq_wire::state_rollup()`] counts a viewer's rollup endpoint +//! already serves, ridden out to the collector on a timer instead of only +//! on request. See the crate README for why this lives in its own crate +//! rather than inside `hive-jobq` or `hive-jobq-wire`. //! //! Same OTEL SDK setup as `hive-c0re::stats::otel_metrics` (container //! stats) and `hive-metric` (the one-shot CLI): the metrics SDK's @@ -17,10 +13,10 @@ //! Bridging async→sync: reading `state_rollup()` needs the scheduler's //! `std::sync::Mutex`, which is fine to lock briefly from a sync callback — //! but holding it for the whole OTLP export would block every request -//! handler that also locks it for however long the export takes. So, same -//! pattern as the container-stats exporter: an async task refreshes a -//! shared snapshot on an interval, and the (sync) SDK callbacks only ever -//! read that snapshot, never the scheduler directly. +//! handler that also locks it for however long the export takes. So an +//! async task refreshes a shared snapshot on an interval, and the (sync) +//! SDK callbacks only ever read that snapshot, never the scheduler +//! directly. use std::hash::Hash; use std::sync::{Arc, Mutex}; @@ -36,30 +32,30 @@ use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; /// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset. /// Matches `hive-c0re::stats::otel_metrics`'s default — no reason for the -/// two exporters to disagree on how fresh "current" means by default. +/// exporters across this tree to disagree on how fresh "current" means by +/// default. const DEFAULT_INTERVAL: Duration = Duration::from_mins(1); /// Spawn a jobq-rollup OTEL exporter for `jobq` if OTEL is configured /// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty) — `None` otherwise, same -/// graceful-absence shape every other optional wiring in this daemon uses -/// (queue, bridge, forge, webhook secret). +/// graceful-absence shape optional OTEL wiring uses elsewhere in this tree. +/// `service_name` becomes the exported resource's `service.name` — the +/// caller's own binary name, since this crate has no opinion on who's +/// calling it. /// -/// **No process-global state.** Earlier drafts of this held the snapshot and -/// the `SdkMeterProvider` behind `static OnceLock`s, which quietly made this -/// a swarm-controller singleton — a second call (a second graph, a test, a -/// future host with more than one jobq instance) would have silently reused -/// the first call's state instead of exporting its own. The snapshot is now -/// a plain local `Arc`, and the provider is returned rather than stashed — -/// **the caller owns it and must keep it alive** (bind it to a named -/// variable, not `_`) for as long as export should continue; dropping it -/// stops the `PeriodicReader`. +/// **No process-global state.** The snapshot is a plain local `Arc`, and +/// the provider is returned rather than stashed in a static — **the caller +/// owns it and must keep it alive** (bind it to a named variable, not `_`) +/// for as long as export should continue; dropping it stops the +/// `PeriodicReader`. Call this once per graph you want exported, from as +/// many call sites as you like — nothing here assumes there's only one. /// -/// Generic over `N`/`R` (the same parameters `Scheduler` itself takes) -/// rather than this crate's own `SwarmNodeKind`/`SwarmResourceKind` — the -/// whole point of the ask was a shape any `hive-jobq` host can plug in, not -/// one hardcoded to this crate's node/resource vocabulary. +/// Generic over `N`/`R` (the same parameters [`hive_jobq::scheduler::Scheduler`] +/// itself takes) rather than any one host's node/resource vocabulary — the +/// whole point of this crate is a shape any `hive-jobq` host can plug in. pub fn spawn_exporter( jobq: Arc>>, + service_name: &str, ) -> Option where N: Send + 'static, @@ -87,9 +83,9 @@ where } }); - match build_provider(interval, snapshot) { + match build_provider(interval, snapshot, service_name) { Ok(provider) => { - tracing::info!(%endpoint, ?interval, "otel jobq-metrics: exporter enabled"); + tracing::info!(%endpoint, ?interval, %service_name, "otel jobq-metrics: exporter enabled"); Some(provider) } Err(e) => { @@ -102,6 +98,7 @@ where fn build_provider( interval: Duration, snapshot: Arc>>, + service_name: &str, ) -> Result { // Same http/json, endpoint-from-env-only construction as // `hive-c0re::stats::otel_metrics` — see that module's doc comment for @@ -116,7 +113,7 @@ fn build_provider( .build(); let provider = SdkMeterProvider::builder() .with_reader(reader) - .with_resource(resource()) + .with_resource(resource(service_name)) .build(); register_instruments(&provider, snapshot); Ok(provider) @@ -164,19 +161,11 @@ fn state_attr(c: &StateCount) -> KeyValue { KeyValue::new("state", format!("{:?}", c.state)) } -/// Resource: `service.name = swarm-controller` plus whatever the operator -/// set in `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` (same channel every -/// other OTEL exporter in this tree reads its `hive`/`swarm` labels from — -/// this daemon has no single "hive" of its own, so nothing is assumed here -/// beyond what the operator supplies). -/// -/// Hardcodes `swarm-controller` even though [`spawn_exporter`] is generic — -/// a future non-swarm-controller caller of this same function would want a -/// different `service.name`, but that's a real parameter to add when a -/// second caller actually exists, not a guess to make now for one that -/// doesn't. -fn resource() -> Resource { - let mut builder = Resource::builder().with_service_name("swarm-controller"); +/// Resource: `service.name = service_name` plus whatever the operator set +/// in `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` (same channel every other +/// OTEL exporter in this tree reads its `hive`/`swarm` labels from). +fn resource(service_name: &str) -> Resource { + let mut builder = Resource::builder().with_service_name(service_name.to_owned()); for (k, v) in resource_attributes() { builder = builder.with_attribute(KeyValue::new(k, v)); } @@ -219,8 +208,8 @@ fn resource_attributes() -> Vec<(String, String)> { /// Parse `key=value` pairs separated by commas and/or newlines. The value /// keeps any `=` after the first (so `Authorization=Bearer x=y` → `Bearer /// x=y`). Byte-identical logic to `hive-c0re::stats::otel_metrics::parse_kv` -/// — small enough that a shared crate for it would cost more than it saves, -/// but kept in lockstep on purpose. +/// — small enough that sharing it isn't worth a dependency edge, but kept +/// in lockstep on purpose. fn parse_kv(s: &str) -> Vec<(String, String)> { s.split([',', '\n']) .filter_map(|pair| { @@ -260,7 +249,7 @@ mod tests { } /// `state_attr` must spell the exact variant name — a Grafana query - /// filtering `state="Running"` should match what this daemon actually + /// filtering `state="Running"` should match what this crate actually /// emits, not a re-cased or re-worded version of it. #[test] fn state_attr_spells_the_variant_name() { diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index b1dba490..9716f4e8 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -38,12 +38,12 @@ problem_details = { version = "0.9.0", features = ["axum"] } # same shape `hive-c0re/src/job_queue/scheduler.rs` uses over its own graph. hive-jobq.workspace = true hive-jobq-wire.workspace = true -# OTEL SDK for the jobq-rollup metrics exporter (`jobq_metrics.rs`), from -# the workspace base — see the root Cargo.toml's comment for the -# blocking-client rationale shared by every OTEL-pushing crate. -opentelemetry.workspace = true -opentelemetry_sdk.workspace = true -opentelemetry-otlp.workspace = true +# The jobq-rollup OTEL exporter, wired up in `main` via +# `hive_jobq_metrics::spawn_exporter` — moved to its own crate (rather than +# living here as `jobq_metrics.rs`) specifically so a future second caller +# (e.g. hive-c0re, for its own per-hive job graph) doesn't have to depend on +# this whole binary to reuse it. +hive-jobq-metrics.workspace = true # The forge webhook HMAC (`webhook.rs`). Kept in this crate rather than # shared with hive-c0re's equivalent: c0re's copy is scheduled to be deleted # with its webhook routes once registration moves here, so the second holder diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 247f85a2..b3a84a69 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -37,7 +37,6 @@ use utoipa_axum::{router::OpenApiRouter, routes}; mod auth; mod forge; -mod jobq_metrics; mod status; mod webhook; @@ -921,7 +920,8 @@ async fn main() -> Result<()> { // `PeriodicReader`, so it must live as long as `main` does (which it // does here: this binding never goes out of scope before the process // exits via `axum::serve(...).await` below). - let _jobq_metrics_provider = jobq_metrics::spawn_exporter(Arc::clone(&jobq)); + let _jobq_metrics_provider = + hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller"); // Same "log and carry on" shape as the queue/bridge/forge wiring above. // A controller that cannot hold a webhook secret still serves every