From f5ff8698f33f476974e3703d4f3a4120502d88fa Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 19 Aug 2026 17:52:12 +0200 Subject: [PATCH] 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