From 361396cab881771df5d74c69ff0a788fc557f7e9 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 22:56:44 +0200 Subject: [PATCH] wire the authenticated OTLP push into vcs_metrics and hive-jobq-metrics --- Cargo.lock | 3 ++ hive-jobq-metrics/Cargo.toml | 8 ++++ hive-jobq-metrics/src/lib.rs | 53 +++++++++++++++++++++++--- swarm-controller/Cargo.toml | 20 +++++++++- swarm-controller/src/main.rs | 15 +++++++- swarm-controller/src/vcs_metrics.rs | 58 ++++++++++++++++++++++++++++- 6 files changed, 147 insertions(+), 10 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 99990449..8b7f5a70 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1794,9 +1794,11 @@ name = "hive-jobq-metrics" version = "0.1.0" dependencies = [ "anyhow", + "async-trait", "hive-jobq", "hive-jobq-wire", "opentelemetry", + "opentelemetry-http", "opentelemetry-otlp", "opentelemetry_sdk", "tokio", @@ -4583,6 +4585,7 @@ dependencies = [ "hive-types", "hmac 0.13.0", "opentelemetry", + "opentelemetry-http", "opentelemetry-otlp", "opentelemetry_sdk", "problem_details", diff --git a/hive-jobq-metrics/Cargo.toml b/hive-jobq-metrics/Cargo.toml index e48094e3..a280f7b6 100644 --- a/hive-jobq-metrics/Cargo.toml +++ b/hive-jobq-metrics/Cargo.toml @@ -9,10 +9,18 @@ workspace = true [dependencies] anyhow.workspace = true +# `spawn_exporter`'s optional `http_client` param: an authenticated push +# (`swarm-controller`'s case, wired through `swarm_queue_client:: +# otel_http_client`) needs a caller-supplied `HttpClient` per request, not +# a static header. This crate stays generic over that caller's concrete +# client type — see `BoxedHttpClient`'s own doc — so it depends on the +# trait, never on `swarm-queue-client` itself. +async-trait.workspace = true hive-jobq.workspace = true hive-jobq-wire.workspace = true opentelemetry.workspace = true opentelemetry_sdk.workspace = true opentelemetry-otlp.workspace = true +opentelemetry-http.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/hive-jobq-metrics/src/lib.rs b/hive-jobq-metrics/src/lib.rs index b927ceef..60373b1c 100644 --- a/hive-jobq-metrics/src/lib.rs +++ b/hive-jobq-metrics/src/lib.rs @@ -23,13 +23,38 @@ use std::sync::{Arc, Mutex}; use std::time::Duration; use anyhow::{Context, Result}; +use async_trait::async_trait; use hive_jobq_wire::StateCount; use opentelemetry::KeyValue; use opentelemetry::metrics::MeterProvider as _; -use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; +use opentelemetry_http::{Bytes, HttpClient, HttpError, Request, Response}; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig, WithHttpConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; +/// Type-erases a caller's concrete [`HttpClient`] impl so [`spawn_exporter`] +/// can accept "any client, or none" without this crate depending on any one +/// caller's concrete type (`swarm-controller`'s +/// `swarm_queue_client::otel_http_client::AuthenticatedHttpClient` today, +/// conceivably something else from a future caller) — the same genericity +/// this crate already has over `N`/`R`, applied to the one other caller- +/// supplied thing it touches. +/// +/// A thin delegating newtype rather than a blanket `impl HttpClient for +/// Box`: the orphan rule refuses that blanket impl outright +/// (`Box` and `dyn HttpClient` are both foreign to this crate), so a local +/// wrapper is the standard way around it, not a workaround chosen over a +/// cleaner alternative. +#[derive(Debug)] +struct BoxedHttpClient(Box); + +#[async_trait] +impl HttpClient for BoxedHttpClient { + async fn send_bytes(&self, request: Request) -> Result, HttpError> { + self.0.send_bytes(request).await + } +} + /// Default export cadence when `HYPERHIVE_OTEL_METRIC_INTERVAL_MS` is unset. /// Matches `hive-c0re::stats::otel_metrics`'s default — no reason for the /// exporters across this tree to disagree on how fresh "current" means by @@ -53,9 +78,18 @@ const DEFAULT_INTERVAL: Duration = Duration::from_mins(1); /// 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. +/// +/// `http_client`: `None` builds the exporter with its default (endpoint- +/// only, no per-request auth) transport — the shape every caller used +/// before this parameter existed. `Some(client)` routes every export +/// through that [`HttpClient`] instead, for a caller whose destination +/// checks a bearer token per request rather than a static header +/// (`swarm-controller`'s case — see `swarm_queue_client::otel_http_client`'s +/// module doc for why a static header doesn't fit an expiring token). pub fn spawn_exporter( jobq: Arc>>, service_name: &str, + http_client: Option>, ) -> Option where N: Send + 'static, @@ -83,7 +117,7 @@ where } }); - match build_provider(interval, snapshot, service_name) { + match build_provider(interval, snapshot, service_name, http_client) { Ok(provider) => { tracing::info!(%endpoint, ?interval, %service_name, "otel jobq-metrics: exporter enabled"); Some(provider) @@ -99,15 +133,22 @@ fn build_provider( interval: Duration, snapshot: Arc>>, service_name: &str, + http_client: Option>, ) -> 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() + let mut builder = MetricExporter::builder() .with_http() - .with_protocol(Protocol::HttpJson) - .build() - .context("build OTLP metric exporter")?; + .with_protocol(Protocol::HttpJson); + // Only when the caller supplied one — see `spawn_exporter`'s doc for + // who does and why. Wrapped in `BoxedHttpClient` because + // `with_http_client` wants an owned `T: HttpClient`, not the trait + // object this parameter is typed as (see that newtype's own doc). + if let Some(client) = http_client { + builder = builder.with_http_client(BoxedHttpClient(client)); + } + let exporter = builder.build().context("build OTLP metric exporter")?; let reader = PeriodicReader::builder(exporter) .with_interval(interval) .build(); diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 1bf1a16d..8ea1ec6e 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -65,7 +65,13 @@ hive-types.workspace = true # `auth`'s bridge client — same crate the bridge itself uses to define the # request/response shape, so the two ends cannot drift. `forge.rs` also # uses this directly for `StatusCode` in its error-classification helpers. -reqwest.workspace = true +# +# `blocking` on top of the workspace default: `vcs_metrics`'s OTLP exporter +# runs on a thread with no tokio reactor (see that module's doc), so the +# `reqwest::blocking::Client` it hands to `AuthenticatedHttpClient` has to +# come from the blocking half of this crate, not the async one every other +# consumer here uses. +reqwest = { workspace = true, features = ["blocking"] } serde.workspace = true serde_json.workspace = true swarm-authelia-bridge-sock.workspace = true @@ -77,7 +83,17 @@ swarm-authelia-bridge-sock.workspace = true # `kv` for the same reason one level in: the status bucket's name and # creation config are shared with the hive that writes it, so this end does # not get to declare them privately. -swarm-queue-client = { workspace = true, features = ["kv"] } +# +# `otel-auth` for `vcs_metrics`'s `AuthenticatedHttpClient` — see that +# feature's own comment in `swarm-queue-client`'s Cargo.toml. +swarm-queue-client = { workspace = true, features = ["kv", "otel-auth"] } +# The `HttpClient` trait itself, so `main.rs` can name +# `Box` when handing +# `vcs_metrics::authenticated_http_client()`'s output to +# `hive_jobq_metrics::spawn_exporter` — a transitive dependency (via +# `swarm-queue-client`'s `otel-auth` feature above) isn't enough to `use` +# it directly, Cargo requires a direct entry for that. +opentelemetry-http.workspace = true tokio.workspace = true tracing.workspace = true tracing-subscriber.workspace = true diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index bad9edfe..57bf51cc 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -1157,8 +1157,21 @@ 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). + // Own `AuthenticatedHttpClient` instance, independent of `vcs_metrics`'s + // — both read the same env vars, and a client is cheap enough + // (one `reqwest::blocking::Client`) that sharing one across two + // otherwise-independent exporters isn't worth the plumbing. `Ok` becomes + // `Some`, `Err` becomes `None` — same "log and carry on" shape as the + // rest of this function's optional wiring; a controller that can't + // authenticate its jobq push still serves every other route. + let jobq_http_client = crate::vcs_metrics::authenticated_http_client() + .inspect_err( + |e| tracing::warn!(error = ?e, "otel jobq-metrics: no authenticated http client"), + ) + .ok() + .map(|c| Box::new(c) as Box); let _jobq_metrics_provider = - hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller"); + hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller", jobq_http_client); // Same "log and carry on" shape as the queue/bridge/forge wiring above. // A controller that cannot hold a webhook secret still serves every diff --git a/swarm-controller/src/vcs_metrics.rs b/swarm-controller/src/vcs_metrics.rs index c4265e81..09f169ef 100644 --- a/swarm-controller/src/vcs_metrics.rs +++ b/swarm-controller/src/vcs_metrics.rs @@ -25,7 +25,7 @@ use std::time::Duration; use anyhow::{Context, Result}; use opentelemetry::KeyValue; use opentelemetry::metrics::{Counter, MeterProvider as _}; -use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig, WithHttpConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; @@ -114,9 +114,16 @@ fn build_provider(interval: Duration) -> Result { // Same http/json, endpoint-from-env-only construction as // `hive_jobq_metrics::build_provider` — see that module's doc comment // for why `with_endpoint` is deliberately never called here. + // + // `.with_http_client(..)`: unlike the hive tier's own exporters, this + // one crosses a real trust boundary (the swarm-tier `otlp/swarm` + // receiver checks a bearer token's audience, see + // `swarm_queue_client::otel_http_client`'s module doc) — a plain + // endpoint-only client would be refused at the receiver. let exporter = MetricExporter::builder() .with_http() .with_protocol(Protocol::HttpJson) + .with_http_client(authenticated_http_client()?) .build() .context("build OTLP metric exporter")?; let reader = PeriodicReader::builder(exporter) @@ -128,6 +135,55 @@ fn build_provider(interval: Duration) -> Result { .build()) } +/// Build the [`swarm_queue_client::otel_http_client::AuthenticatedHttpClient`] +/// this exporter pushes through. +/// +/// The queue identity (`SWARM_CONTROLLER_OIDC_*`) is read fresh here rather +/// than threaded in from a caller — by the time this runs, [`endpoint`] has +/// already confirmed OTEL is configured for this host, and `queueEnv` is set +/// **unconditionally** for every `swarm-controller` (the nix module's own +/// words: "a controller without a queue is not a lighter controller, it is +/// a broken one"). So an absent queue identity here is never a supported +/// "OTEL enabled, queue not configured" deployment shape — it is a +/// deployment bug, and gets the same `Err` treatment `build()`'s caller +/// already applies to any other exporter-construction failure. +/// +/// `SWARM_CONTROLLER_OTEL_AUDIENCE` names the audience `otlp/swarm`'s +/// authenticator checks — set by the same nix option that sets +/// `OTEL_EXPORTER_OTLP_ENDPOINT`, so the two are never independently absent. +/// +/// No CA handling here unlike [`swarm_queue_client::QueueConfig::ca_file`] +/// (which `mint_token_for_blocking` already applies to the TOKEN endpoint +/// internally): the metrics endpoint is reached through the gateway with a +/// certificate the swarm CA issued, and this whole process already trusts +/// that CA via `SSL_CERT_FILE` (`hive-ca-trust.nix`, applied host-wide to +/// this unit for the same reason the forge client needs no special +/// handling either) — a second, narrower trust bundle here would just be +/// the same fact stated twice. +pub(crate) fn authenticated_http_client() +-> Result { + let cfg = swarm_queue_client::QueueConfig::from_env("SWARM_CONTROLLER") + .context("reading the queue identity this daemon authenticates its OTLP push with")? + .ok_or_else(|| { + anyhow::anyhow!( + "SWARM_CONTROLLER_NATS_URL and friends are unset, but OTEL_EXPORTER_OTLP_ENDPOINT \ + is — the queue is required for every controller, so this combination is a \ + deployment bug, not a supported partial config" + ) + })?; + let audience = std::env::var("SWARM_CONTROLLER_OTEL_AUDIENCE").context( + "SWARM_CONTROLLER_OTEL_AUDIENCE is unset, but OTEL_EXPORTER_OTLP_ENDPOINT is — the nix \ + module sets both together", + )?; + Ok( + swarm_queue_client::otel_http_client::AuthenticatedHttpClient::new( + reqwest::blocking::Client::new(), + cfg, + audience, + ), + ) +} + /// `service.name = swarm-controller` plus whatever the operator set in /// `HYPERHIVE_OTEL_EXTRA_RESOURCE_ATTRIBUTES` — same channel /// `hive_jobq_metrics::resource` reads, since both live in this one binary.