swarm-controller: move OTEL deps to workspace level, drop jobq_metrics singleton
This commit is contained in:
parent
1f5a7b71ed
commit
621245306f
6 changed files with 81 additions and 67 deletions
15
Cargo.toml
15
Cargo.toml
|
|
@ -127,6 +127,21 @@ reqwest = { version = "0.13", default-features = false, features = [
|
||||||
] }
|
] }
|
||||||
hyper = { version = "1", features = ["client", "http1"] }
|
hyper = { version = "1", features = ["client", "http1"] }
|
||||||
hyper-util = { version = "0.1", features = ["tokio"] }
|
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"
|
http-body-util = "0.1"
|
||||||
# ⚠️ Keep at 0.11.1 or newer, and keep it on the SAME reqwest as everything
|
# ⚠️ 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
|
# else. 0.11.0 links reqwest 0.12 while the workspace is on 0.13, and cargo
|
||||||
|
|
|
||||||
|
|
@ -23,20 +23,12 @@ clap.workspace = true
|
||||||
clap_complete.workspace = true
|
clap_complete.workspace = true
|
||||||
clap-markdown = "0.1"
|
clap-markdown = "0.1"
|
||||||
# OTEL SDK for the per-agent container-resource metrics exporter
|
# OTEL SDK for the per-agent container-resource metrics exporter
|
||||||
# (stats/otel_metrics.rs). Same versions as hive-metric — the blocking OTLP
|
# (stats/otel_metrics.rs). "internal-logs" on top of the workspace base is
|
||||||
# client is deliberate: the metrics SDK's PeriodicReader runs on a background
|
# this crate's own opt-in: it's not a transport, it's in `default`, and
|
||||||
# thread with no Tokio reactor, where the async client panics.
|
# `default-features = false` upstream drops it unless named here.
|
||||||
opentelemetry = "0.32"
|
opentelemetry.workspace = true
|
||||||
opentelemetry_sdk = { version = "0.32", features = ["metrics"] }
|
opentelemetry_sdk.workspace = true
|
||||||
opentelemetry-otlp = { version = "0.32", default-features = false, features = [
|
opentelemetry-otlp = { workspace = true, features = ["internal-logs"] }
|
||||||
# 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",
|
|
||||||
] }
|
|
||||||
indicatif.workspace = true
|
indicatif.workspace = true
|
||||||
hive-core-agent-sock.workspace = true
|
hive-core-agent-sock.workspace = true
|
||||||
hive-sh4re.workspace = true
|
hive-sh4re.workspace = true
|
||||||
|
|
|
||||||
|
|
@ -11,9 +11,8 @@ path = "src/main.rs"
|
||||||
[dependencies]
|
[dependencies]
|
||||||
anyhow.workspace = true
|
anyhow.workspace = true
|
||||||
clap.workspace = true
|
clap.workspace = true
|
||||||
# OTEL SDK — only used by this crate, so kept local rather than in workspace.dependencies.
|
# OTEL SDK, from the workspace base — see the root Cargo.toml's comment for
|
||||||
opentelemetry = "0.32"
|
# the blocking-client rationale shared by every OTEL-pushing crate.
|
||||||
opentelemetry_sdk = { version = "0.32", features = ["metrics"] }
|
|
||||||
# Blocking (not async) reqwest client on purpose: the metrics SDK drives the
|
# Blocking (not async) reqwest client on purpose: the metrics SDK drives the
|
||||||
# OTLP push from a `PeriodicReader` background thread that has no Tokio runtime,
|
# OTLP push from a `PeriodicReader` background thread that has no Tokio runtime,
|
||||||
# so the async client panics there with "no reactor running". The blocking
|
# 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.
|
# records one point then `shutdown()`s, a synchronous send is exactly right.
|
||||||
# No `internal-logs` here, unlike hive-c0re: this binary installs no tracing
|
# No `internal-logs` here, unlike hive-c0re: this binary installs no tracing
|
||||||
# subscriber, so the SDK's diagnostics would have nowhere to go.
|
# subscriber, so the SDK's diagnostics would have nowhere to go.
|
||||||
opentelemetry-otlp = { version = "0.32", default-features = false, features = [
|
opentelemetry.workspace = true
|
||||||
"metrics",
|
opentelemetry_sdk.workspace = true
|
||||||
"http-json",
|
opentelemetry-otlp.workspace = true
|
||||||
"reqwest-blocking-client",
|
|
||||||
"reqwest-rustls",
|
|
||||||
] }
|
|
||||||
|
|
|
||||||
|
|
@ -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.
|
# same shape `hive-c0re/src/job_queue/scheduler.rs` uses over its own graph.
|
||||||
hive-jobq.workspace = true
|
hive-jobq.workspace = true
|
||||||
hive-jobq-wire.workspace = true
|
hive-jobq-wire.workspace = true
|
||||||
# OTEL SDK for the jobq-rollup metrics exporter (`jobq_metrics.rs`). Same
|
# OTEL SDK for the jobq-rollup metrics exporter (`jobq_metrics.rs`), from
|
||||||
# versions + blocking-client rationale as `hive-c0re`/`hive-metric`: the
|
# the workspace base — see the root Cargo.toml's comment for the
|
||||||
# metrics SDK's `PeriodicReader` runs on a background thread with no Tokio
|
# blocking-client rationale shared by every OTEL-pushing crate.
|
||||||
# reactor, where the async client panics. Kept local rather than in
|
opentelemetry.workspace = true
|
||||||
# `workspace.dependencies` — same call `hive-metric` already made, this is
|
opentelemetry_sdk.workspace = true
|
||||||
# still the only other crate that needs it.
|
opentelemetry-otlp.workspace = true
|
||||||
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
|
# 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
|
# 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
|
# with its webhook routes once registration moves here, so the second holder
|
||||||
|
|
|
||||||
|
|
@ -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
|
//! `hive_jobq_wire::state_rollup()` counts `GET /api/jobq/rollup` already
|
||||||
//! serves, ridden out to the collector on a timer instead of only on
|
//! 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
|
//! request. First consumer of a generic "add jobq metrics" ask: [`spawn_exporter`]
|
||||||
//! shape works for any `hive-jobq` instance, this crate's is just the
|
//! is generic over any `hive_jobq::scheduler::Scheduler<N, R>`, not tied to
|
||||||
//! first to wire it.
|
//! 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
|
//! Same OTEL SDK setup as `hive-c0re::stats::otel_metrics` (container
|
||||||
//! stats) and `hive-metric` (the one-shot CLI): the metrics SDK's
|
//! 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
|
//! shared snapshot on an interval, and the (sync) SDK callbacks only ever
|
||||||
//! read that snapshot, never the scheduler directly.
|
//! 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 std::time::Duration;
|
||||||
|
|
||||||
use anyhow::{Context, Result};
|
use anyhow::{Context, Result};
|
||||||
|
|
@ -30,35 +34,39 @@ use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig};
|
||||||
use opentelemetry_sdk::Resource;
|
use opentelemetry_sdk::Resource;
|
||||||
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
|
use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider};
|
||||||
|
|
||||||
use crate::{SwarmNodeKind, SwarmResourceKind};
|
|
||||||
|
|
||||||
/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset.
|
/// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset.
|
||||||
/// Matches `hive-c0re::stats::otel_metrics`'s default — no reason for the
|
/// Matches `hive-c0re::stats::otel_metrics`'s default — no reason for the
|
||||||
/// two exporters to disagree on how fresh "current" means by default.
|
/// two exporters to disagree on how fresh "current" means by default.
|
||||||
const DEFAULT_INTERVAL: Duration = Duration::from_mins(1);
|
const DEFAULT_INTERVAL: Duration = Duration::from_mins(1);
|
||||||
|
|
||||||
/// Latest rollup snapshot: written by the async refresher, read by the sync
|
/// Spawn a jobq-rollup OTEL exporter for `jobq` if OTEL is configured
|
||||||
/// observable-instrument callbacks.
|
/// (`OTEL_EXPORTER_OTLP_ENDPOINT` non-empty) — `None` otherwise, same
|
||||||
static SNAPSHOT: OnceLock<Arc<Mutex<Vec<StateCount>>>> = OnceLock::new();
|
|
||||||
|
|
||||||
/// Keep the provider alive for the process lifetime — the `PeriodicReader`
|
|
||||||
/// exports only while the provider lives.
|
|
||||||
static PROVIDER: OnceLock<SdkMeterProvider> = 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
|
/// graceful-absence shape every other optional wiring in this daemon uses
|
||||||
/// (queue, bridge, forge, webhook secret). Call once at startup.
|
/// (queue, bridge, forge, webhook secret).
|
||||||
pub fn spawn_exporter(
|
///
|
||||||
jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>>,
|
/// **No process-global state.** Earlier drafts of this held the snapshot and
|
||||||
) {
|
/// the `SdkMeterProvider` behind `static OnceLock`s, which quietly made this
|
||||||
let Some(endpoint) = endpoint() else {
|
/// a swarm-controller singleton — a second call (a second graph, a test, a
|
||||||
tracing::debug!("otel jobq-metrics: no endpoint configured, exporter disabled");
|
/// future host with more than one jobq instance) would have silently reused
|
||||||
return;
|
/// 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 —
|
||||||
let snapshot = SNAPSHOT
|
/// **the caller owns it and must keep it alive** (bind it to a named
|
||||||
.get_or_init(|| Arc::new(Mutex::new(Vec::new())))
|
/// variable, not `_`) for as long as export should continue; dropping it
|
||||||
.clone();
|
/// 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<N, R>(
|
||||||
|
jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<N, R>>>,
|
||||||
|
) -> Option<SdkMeterProvider>
|
||||||
|
where
|
||||||
|
N: Send + 'static,
|
||||||
|
R: Clone + Eq + Hash + Send + 'static,
|
||||||
|
{
|
||||||
|
let endpoint = endpoint()?;
|
||||||
|
let snapshot: Arc<Mutex<Vec<StateCount>>> = Arc::new(Mutex::new(Vec::new()));
|
||||||
let interval = interval();
|
let interval = interval();
|
||||||
|
|
||||||
let refresh = snapshot.clone();
|
let refresh = snapshot.clone();
|
||||||
|
|
@ -81,11 +89,12 @@ pub fn spawn_exporter(
|
||||||
|
|
||||||
match build_provider(interval, snapshot) {
|
match build_provider(interval, snapshot) {
|
||||||
Ok(provider) => {
|
Ok(provider) => {
|
||||||
let _ = PROVIDER.set(provider);
|
|
||||||
tracing::info!(%endpoint, ?interval, "otel jobq-metrics: exporter enabled");
|
tracing::info!(%endpoint, ?interval, "otel jobq-metrics: exporter enabled");
|
||||||
|
Some(provider)
|
||||||
}
|
}
|
||||||
Err(e) => {
|
Err(e) => {
|
||||||
tracing::warn!(error = ?e, "otel jobq-metrics: exporter init failed");
|
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 —
|
/// 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
|
/// this daemon has no single "hive" of its own, so nothing is assumed here
|
||||||
/// beyond what the operator supplies).
|
/// 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 {
|
fn resource() -> Resource {
|
||||||
let mut builder = Resource::builder().with_service_name("swarm-controller");
|
let mut builder = Resource::builder().with_service_name("swarm-controller");
|
||||||
for (k, v) in resource_attributes() {
|
for (k, v) in resource_attributes() {
|
||||||
|
|
|
||||||
|
|
@ -917,7 +917,11 @@ async fn main() -> Result<()> {
|
||||||
hive_jobq::resources::ResourceTable::new(),
|
hive_jobq::resources::ResourceTable::new(),
|
||||||
)));
|
)));
|
||||||
spawn_jobq_worker(Arc::clone(&jobq), deps);
|
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.
|
// Same "log and carry on" shape as the queue/bridge/forge wiring above.
|
||||||
// A controller that cannot hold a webhook secret still serves every
|
// A controller that cannot hold a webhook secret still serves every
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue