From 2aa33f12d2e2dbc8995513d9866ff3d0eb89b12b Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 21:36:45 +0200 Subject: [PATCH 1/6] swarm-otel: add an unauthenticated swarm-tier OTLP receiver for swarm-controller to push to --- nix/host-modules/swarm-controller.nix | 36 ++++++++++++++++- nix/host-modules/swarm-otel.nix | 56 +++++++++++++++++++++++---- 2 files changed, 83 insertions(+), 9 deletions(-) diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 5ba54251..672481ff 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -149,6 +149,35 @@ let SWARM_CONTROLLER_NAME = config.services.hyperhive.swarm.name; }; + # `vcs_metrics`/`hive_jobq_metrics` (this daemon's OTLP push exporters) + # read the standard `OTEL_EXPORTER_OTLP_ENDPOINT` var and are silent, + # graceful no-ops without it — same shape as `forgeEnv`/`authBridgeEnv` + # above, genuinely optional and gated on the option resolving rather + # than assumed. + # + # Gated on THIS HOST running `swarm-otel`, not merely on it existing + # somewhere in the swarm: its receiver for a swarm-tier producer binds + # `127.0.0.1` only (see `swarm-otel.nix::producerPort`), so reaching it + # needs co-location, and this checks that directly rather than assuming + # it — the unasserted-co-location gap flagged elsewhere in this codebase + # doesn't apply here because there is nothing to assert: a controller on + # a host that doesn't run the collector simply exports nothing, the same + # graceful absence `forgeEnv` already models. + otelSwarmCfg = config.services.hyperhive.swarm.otel; + otelEnv = lib.optionalAttrs otelSwarmCfg.enable { + OTEL_EXPORTER_OTLP_ENDPOINT = "http://${otelSwarmCfg.domain}:${toString otelSwarmCfg.producerPort}"; + }; + + # Only when THIS host also runs `swarm-otel` — same reasoning and same + # shape as `hive-forge/default.nix`'s `ssoLocal`-gated entry: a raw host + # systemd unit does not get the bridge's dnsmasq resolution containers + # get, so the name that reaches a co-located collector over `otelEnv` + # above needs an explicit loopback alias here, or it resolves however + # (or however it fails to) off-host DNS says on this host. + otelHostsEntry = lib.optionalAttrs otelSwarmCfg.enable { + "127.0.0.1" = [ otelSwarmCfg.domain ]; + }; + # Wrapped rather than documented: every one of these values is derived # from an option this deployment already set, so making the operator # re-supply them on the command line would be asking them to repeat the @@ -450,6 +479,10 @@ in "swarm-controller-credential" ]; + # See `otelHostsEntry`'s own comment: only present, and only ever + # `{ "127.0.0.1" = [ ... ]; }`, when this host also runs `swarm-otel`. + networking.hosts = otelHostsEntry; + users.users.swarm-controller = { isSystemUser = true; group = "swarm-controller"; @@ -676,7 +709,8 @@ in // forgeEnv // webhookEnv // authBridgeEnv - // swarmNameEnv; + // swarmNameEnv + // otelEnv; }; # A systemd credential is a SNAPSHOT: it is materialised into `%d` once, diff --git a/nix/host-modules/swarm-otel.nix b/nix/host-modules/swarm-otel.nix index 9d5ebd00..e037c448 100644 --- a/nix/host-modules/swarm-otel.nix +++ b/nix/host-modules/swarm-otel.nix @@ -249,6 +249,29 @@ in ''; }; + producerPort = lib.mkOption { + type = lib.types.port; + default = 4390; + description = '' + Port the swarm tier's own unauthenticated OTLP/HTTP receiver + listens on, at `127.0.0.1` — for a **swarm-level** producer + (`swarm-controller`'s vcs/jobq counters today) to push to, on the + same host this collector runs on. No authenticator, deliberately: + unlike the per-hive receivers, the label this receiver's samples + get (`swarm`) is the one the receiver's own *existence* + establishes — there is no hive identity to forge or attribute, + so the per-hive attribution mechanism does not apply here. + + ⚠️ Deliberately NOT derived from `port + (number of hives)`: that + range grows every time a hive is added, and a fixed offset from + it would silently start colliding once the hive count caught up. + Kept as its own reserved value instead, and the assertion below + still catches a real collision (including one hive growth + eventually causes) rather than starting a collector that quietly + drops one receiver's samples. + ''; + }; + domain = lib.mkOption { type = lib.types.str; default = "otel.${domainBase}"; @@ -715,6 +738,7 @@ in derived = lib.attrValues hivePorts; others = [ cfg.telemetryPort + cfg.producerPort otelCfg.collector.port ] ++ lib.optional vmCfg.enable vmCfg.port; @@ -846,6 +870,12 @@ in }; } ) hivePorts + # The swarm tier's OWN receiver, unauthenticated on + # purpose — see `producerPort`'s description for why no + # per-hive-shaped attribution applies here. + // { + "otlp/${swarmTierName}".protocols.http.endpoint = "127.0.0.1:${toString cfg.producerPort}"; + } # MERGED with the per-hive receivers, never assigned over # them. A plain assignment here would drop every hive's # receiver and still render a valid config that starts @@ -1044,9 +1074,17 @@ in # them would acquire the one label a swarm-level service must # not have. Keeping it out of them makes the absence # structural rather than something to remember to strip. - // lib.optionalAttrs (cfg.scrapeTargets != { }) { + // { + # `otlp/${swarmTierName}` is unconditional (see the + # receiver above), so this pipeline is too — a swarm-tier + # producer must always have somewhere to land, unlike + # `prometheus`, which only joins the receiver list once + # something has actually declared a scrape target. "metrics/${swarmTierName}" = { - receivers = [ "prometheus" ]; + receivers = [ + "otlp/${swarmTierName}" + ] + ++ lib.optional (cfg.scrapeTargets != { }) "prometheus"; processors = [ "resource/${swarmTierName}" ]; exporters = exporterNames; }; @@ -1151,12 +1189,14 @@ in # an invented one (a sentinel, the local hive's name) would be # queried as though it meant something. # - # ⚠️ Emitted for the LOG pipeline too, not just the scraped one: - # a processor a pipeline names but the config does not define is - # a collector that refuses to start, and enabling the log store - # without declaring a scrape target is a perfectly ordinary - # config. - // lib.optionalAttrs (cfg.scrapeTargets != { } || collectLogs) { + # ⚠️ Unconditional, not gated on `scrapeTargets != {} || + # collectLogs` (as it once was): a processor a pipeline names + # but the config does not define is a collector that refuses + # to start, and `metrics/${swarmTierName}` now ALWAYS exists + # (its `otlp/${swarmTierName}` receiver is unconditional too, + # for a swarm-tier producer to push to) — so this processor + # has to exist unconditionally right alongside it. + // { "resource/${swarmTierName}".attributes = [ { # The metric LABEL, a different namespace from the From e8584e595b7a18b36df5bd054d69ad5d89433539 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 22:03:23 +0200 Subject: [PATCH 2/6] swarm-queue-client: audience-scoped tokens + a blocking mint for a non-reactor caller --- swarm-controller/src/auth.rs | 2 +- swarm-queue-client/Cargo.toml | 11 +- swarm-queue-client/src/lib.rs | 187 ++++++++++++++++++++++++++++++---- 3 files changed, 179 insertions(+), 21 deletions(-) diff --git a/swarm-controller/src/auth.rs b/swarm-controller/src/auth.rs index f08cbb7a..e53634a0 100644 --- a/swarm-controller/src/auth.rs +++ b/swarm-controller/src/auth.rs @@ -98,7 +98,7 @@ impl AuthBridge { // configured CA, if any) — deliberately not `self.http`, which is // the bridge's own client and has nothing to do with authelia's // token endpoint's trust anchors. - let token = swarm_queue_client::mint_token_for(&self.queue_cfg) + let token = swarm_queue_client::mint_token_for(&self.queue_cfg, None) .await .context("minting a bearer token for swarm-authelia-bridge")?; diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index b0b63588..3798b01d 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -35,7 +35,16 @@ notices = ["async-nats/jetstream"] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature # above - the connect itself needs none of them. async-nats.workspace = true -reqwest.workspace = true +# `blocking` on top of the workspace default (`form`/`json`/`rustls`) — +# `mint_token_for_blocking` needs `reqwest::blocking::Client` for a caller +# with no tokio reactor to `.await` an async request on (an OTLP exporter's +# `HttpClient` impl, see that function's doc). Declared here rather than +# left to arrive transitively from a consumer that happens to pull in +# `reqwest`'s blocking feature some other way — this crate already has a +# recorded case of exactly that kind of accidental compile (see the `kv` +# feature's comment above), and `cargo check -p swarm-queue-client` alone +# must not depend on what else is in the build. +reqwest = { workspace = true, features = ["blocking"] } serde.workspace = true serde_json.workspace = true # A library, so its errors are a matchable enum rather than an opaque diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 5e5d956e..1165142e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -316,7 +316,11 @@ impl QueueConfig { /// authenticates as itself. Authelia refuses the `openid` scope for this grant /// (a machine client receives an access token and never an id-token), so no /// scope is requested. -async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result { +async fn mint_token( + http: &reqwest::Client, + cfg: &QueueConfig, + audience: Option<&str>, +) -> Result { // Read per call rather than caching: the file is small, and a cached // secret would survive a rotation that the operator believes took effect. let secret = tokio::fs::read_to_string(&cfg.client_secret_file) @@ -326,19 +330,35 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result Result { + // Why the body and not just the status: see `Error::TokenRefused`. if !status.is_success() { - return Err(Error::TokenRefused { status, body }); + return Err(Error::TokenRefused { + status, + body: body.to_owned(), + }); } - let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?; + let parsed: TokenResponse = serde_json::from_str(body).map_err(Error::TokenResponse)?; Ok(CachedToken { token: parsed.access_token, expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in), @@ -364,14 +384,45 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result, ) -> reqwest::RequestBuilder { + let mut form = vec![("grant_type", "client_credentials")]; + if let Some(audience) = audience { + form.push(("audience", audience)); + } http.post(&cfg.token_endpoint) .basic_auth(&cfg.client_id, Some(secret)) - .form(&[("grant_type", "client_credentials")]) + .form(&form) +} + +/// Read + parse `path` as a PEM root certificate. Shared by the async and +/// blocking HTTP client builders below — `reqwest::Certificate::from_pem` +/// takes the same bytes regardless of which client type ends up trusting +/// it, so the CA loading has no reason to exist twice. +fn load_ca_cert(path: &std::path::Path) -> Result { + let pem = std::fs::read(path).map_err(|source| Error::CaFile { + path: path.display().to_string(), + source, + })?; + reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse { + path: path.display().to_string(), + source, + }) } /// Build the HTTP client used to reach `cfg.token_endpoint`, trusting @@ -382,15 +433,7 @@ fn token_request( fn build_http_client(cfg: &QueueConfig) -> Result { let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10)); if let Some(path) = &cfg.ca_file { - let pem = std::fs::read(path).map_err(|source| Error::CaFile { - path: path.display().to_string(), - source, - })?; - let cert = reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse { - path: path.display().to_string(), - source, - })?; - builder = builder.add_root_certificate(cert); + builder = builder.add_root_certificate(load_ca_cert(path)?); } builder.build().map_err(Error::HttpClient) } @@ -408,9 +451,70 @@ fn build_http_client(cfg: &QueueConfig) -> Result { /// outside that loop (e.g. `swarm-controller::auth`'s bridge client) mints /// per call, same as this crate did before the reconnect-storm fix added the /// cache. -pub async fn mint_token_for(cfg: &QueueConfig) -> Result { +/// +/// `audience` is passed straight to [`token_request`] — see its doc for why +/// it is optional and when a caller needs it. `None` reproduces this +/// function's behaviour before the parameter existed, so every caller from +/// before that added it (the queue connect path, `auth.rs`'s bridge client) +/// is unaffected. +pub async fn mint_token_for(cfg: &QueueConfig, audience: Option<&str>) -> Result { let http = build_http_client(cfg)?; - Ok(mint_token(&http, cfg).await?.token) + Ok(mint_token(&http, cfg, audience).await?.token) +} + +/// Blocking sibling of [`mint_token_for`], for a caller with no tokio +/// reactor to `.await` on. +/// +/// That caller is real, not hypothetical: an OTLP exporter's `HttpClient` +/// implementation runs on the `PeriodicReader`'s own thread, which this +/// workspace deliberately keeps reactor-free (`reqwest-blocking-client` was +/// chosen over the async client for exactly that reason — see +/// `swarm-controller::vcs_metrics`'s module doc). Calling the async +/// [`mint_token_for`] from there would need a runtime that thread does not +/// have; this function uses `reqwest::blocking::Client` end to end instead, +/// so it never needs one. +/// +/// Same request shape and same no-caching behaviour as [`mint_token_for`] — +/// see that function's doc for why both of those are the right call here. +pub fn mint_token_for_blocking(cfg: &QueueConfig, audience: Option<&str>) -> Result { + let http = build_blocking_http_client(cfg)?; + + // Read per call rather than caching — see `mint_token`'s identical + // comment on the async path; the reasoning does not change with the + // client type. + let secret = + std::fs::read_to_string(&cfg.client_secret_file).map_err(|source| Error::ClientSecret { + path: cfg.client_secret_file.display().to_string(), + source, + })?; + + let mut form = vec![("grant_type", "client_credentials")]; + if let Some(audience) = audience { + form.push(("audience", audience)); + } + let response = http + .post(&cfg.token_endpoint) + .basic_auth(&cfg.client_id, Some(secret.trim())) + .form(&form) + .send() + .map_err(Error::TokenRequest)?; + + let status = response.status(); + let body = response.text().unwrap_or_default(); + Ok(parse_token_response(status, &body)?.token) +} + +/// Blocking counterpart to [`build_http_client`] — same CA-trust logic +/// (via the shared [`load_ca_cert`]), a `reqwest::blocking::Client` instead +/// of the async one. See [`mint_token_for_blocking`] for why a caller needs +/// this rather than the async builder. +fn build_blocking_http_client(cfg: &QueueConfig) -> Result { + let mut builder = + reqwest::blocking::Client::builder().timeout(std::time::Duration::from_secs(10)); + if let Some(path) = &cfg.ca_file { + builder = builder.add_root_certificate(load_ca_cert(path)?); + } + builder.build().map_err(Error::HttpClient) } /// Fail fast unless the client is actually connected. @@ -487,7 +591,7 @@ pub async fn connect(cfg: QueueConfig) -> Result { let token = if let Some(token) = reuse { token } else { - let minted = mint_token(&http, &cfg) + let minted = mint_token(&http, &cfg, None) .await // The callback's error type carries a string, so the // source chain would be lost; flatten it rather than @@ -631,7 +735,7 @@ mod tests { #[test] fn the_token_request_authenticates_with_http_basic() { - let req = token_request(&offline_client(), &token_cfg(), "s3cret") + let req = token_request(&offline_client(), &token_cfg(), "s3cret", None) .build() .expect("the token request must build"); @@ -667,4 +771,49 @@ mod tests { "the client id belongs in the Basic credentials, got: {body}" ); } + + /// The queue/bridge shape (`audience: None`) must stay unchanged by the + /// parameter's addition — no `audience` field appears in the body at + /// all, not even empty. + #[test] + fn no_audience_means_no_audience_field() { + let req = token_request(&offline_client(), &token_cfg(), "s3cret", None) + .build() + .expect("the token request must build"); + let body = std::str::from_utf8( + req.body() + .and_then(reqwest::Body::as_bytes) + .expect("the request has an in-memory body"), + ) + .expect("the body is utf-8"); + assert!( + !body.contains("audience"), + "omitting the audience must not even send an empty field, got: {body}" + ); + } + + /// A caller that DOES ask for an audience gets it in the form body, + /// verbatim — this is the half `swarm-otel`'s `oidc/` + /// authenticators actually check. + #[test] + fn an_audience_is_sent_verbatim() { + let req = token_request( + &offline_client(), + &token_cfg(), + "s3cret", + Some("https://otel.example/swarm"), + ) + .build() + .expect("the token request must build"); + let body = std::str::from_utf8( + req.body() + .and_then(reqwest::Body::as_bytes) + .expect("the request has an in-memory body"), + ) + .expect("the body is utf-8"); + assert!( + body.contains("audience=https%3A%2F%2Fotel.example%2Fswarm"), + "the requested audience must reach the form body, got: {body}" + ); + } } From 7913be5435d461eee104aaa9d1c3a74efac1ca28 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 22:38:18 +0200 Subject: [PATCH 3/6] swarm-queue-client: add an authenticated HttpClient for OTLP exporters (otel-auth feature) --- Cargo.lock | 4 + Cargo.toml | 23 ++++++ swarm-queue-client/Cargo.toml | 18 +++++ swarm-queue-client/src/lib.rs | 8 ++ swarm-queue-client/src/otel_http_client.rs | 94 ++++++++++++++++++++++ 5 files changed, 147 insertions(+) create mode 100644 swarm-queue-client/src/otel_http_client.rs diff --git a/Cargo.lock b/Cargo.lock index 80a70420..99990449 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4626,6 +4626,10 @@ name = "swarm-queue-client" version = "0.1.0" dependencies = [ "async-nats", + "async-trait", + "bytes", + "http", + "opentelemetry-http", "reqwest", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index f2c58eb9..e20a2254 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -144,6 +144,29 @@ opentelemetry-otlp = { version = "0.32", default-features = false, features = [ "reqwest-blocking-client", "reqwest-rustls", ] } +# The trait an OTLP exporter's HTTP transport is built on +# (`opentelemetry_otlp::WithHttpConfig::with_http_client`) — pulled in +# directly (not just transitively via `opentelemetry-otlp` above) by +# `swarm-queue-client`'s `otel-auth` feature, which implements the trait +# itself rather than using the crate's own blanket `reqwest`/ +# `reqwest-blocking` impls (this workspace's implementation needs to mint +# a fresh bearer token per request, which no blanket impl can do). No +# extra cargo features requested here for exactly that reason — the +# blanket impls live behind `reqwest`/`reqwest-blocking` features this +# workspace never turns on for this crate. +opentelemetry-http = "0.32" +# `async-trait`: `opentelemetry_http::HttpClient` predates stable +# `async fn` in traits and is still expressed with this macro upstream — +# implementing the trait (rather than only consuming the crate's own +# blanket impls) means writing `#[async_trait::async_trait]` on the impl +# block to match its expansion. +async-trait = "0.1" +# `bytes`/`http`: the payload and request/response types +# `opentelemetry_http::HttpClient::send_bytes` is expressed in terms of — +# needed to implement the trait, not just call something that already +# does. +bytes = "1" +http = "1" 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/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 3798b01d..e7640599 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -30,6 +30,18 @@ kv = ["async-nats/kv"] # `cargo check -p swarm-nats-auth` — no `kv` anywhere in that build — # surfaced it as `cannot find jetstream in async_nats`). notices = ["async-nats/jetstream"] +# OFF by default, same reasoning as `kv`/`notices` above: only a caller +# wiring an OTLP exporter's `HttpClient` seam to this crate's identity +# (today, `swarm-controller`'s `vcs_metrics`/`hive_jobq_metrics`) needs +# `opentelemetry_http`/`async-trait`/`bytes`/`http` pulled in — a plain +# queue-connect-only consumer (the auth-callout responder, a hive +# publishing its status) has no reason to carry them. +otel-auth = [ + "dep:opentelemetry-http", + "dep:async-trait", + "dep:bytes", + "dep:http", +] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature @@ -45,6 +57,12 @@ async-nats.workspace = true # feature's comment above), and `cargo check -p swarm-queue-client` alone # must not depend on what else is in the build. reqwest = { workspace = true, features = ["blocking"] } +# All four `optional = true`, gated behind the `otel-auth` feature above — +# see that feature's own comment for why. +opentelemetry-http = { workspace = true, optional = true } +async-trait = { workspace = true, optional = true } +bytes = { workspace = true, optional = true } +http = { workspace = true, optional = true } serde.workspace = true serde_json.workspace = true # A library, so its errors are a matchable enum rather than an opaque diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 1165142e..1a7acbbe 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -164,6 +164,14 @@ pub fn chain(error: &dyn std::error::Error) -> String { /// which is the disagreement this module exists to prevent. pub mod status; +/// An [`opentelemetry_http::HttpClient`] impl that authenticates every +/// request with a fresh token from this crate's identity — see the +/// module's own doc for the full rationale. Behind the `otel-auth` +/// feature, same reasoning as `status`/`kv` above: only a caller wiring an +/// OTLP exporter to this identity needs the extra dependencies it pulls in. +#[cfg(feature = "otel-auth")] +pub mod otel_http_client; + /// The subject the swarm controller publishes on when the hive-wide knowledge /// repository has changed. One writer, many readers — every hive subscribes. /// diff --git a/swarm-queue-client/src/otel_http_client.rs b/swarm-queue-client/src/otel_http_client.rs new file mode 100644 index 00000000..ce526637 --- /dev/null +++ b/swarm-queue-client/src/otel_http_client.rs @@ -0,0 +1,94 @@ +//! An [`opentelemetry_http::HttpClient`] that authenticates every request +//! with a fresh, audience-scoped bearer token minted from this crate's own +//! [`QueueConfig`] identity — "one identity per principal" applied to a +//! third destination (the queue connection and `swarm-authelia-bridge`'s +//! client, [`crate::mint_token_for`], are the other two). +//! +//! Exists for exactly one shape of caller: an OTLP metric exporter whose +//! `PeriodicReader` drives export from a background thread with **no tokio +//! reactor** (`opentelemetry-otlp`'s `reqwest-blocking-client` feature is +//! chosen everywhere in this workspace for that reason — see +//! `swarm-controller::vcs_metrics`'s module doc). That constraint is why +//! this mints via [`crate::mint_token_for_blocking`] rather than the async +//! [`crate::mint_token_for`]: the latter needs a runtime to `.await` on, +//! which the calling thread does not have. +//! +//! Per-request, not per-tick or cached: `opentelemetry-otlp` 0.32 exposes +//! `WithHttpConfig::with_http_client`, a seam for a caller-supplied +//! [`opentelemetry_http::HttpClient`] — strictly better than minting once +//! and setting a static header (`WithHttpConfig::with_headers`), because a +//! tick that runs long, retries, or fires late can outlive a token minted +//! before it started, and that failure is intermittent — the worst kind to +//! debug through a metrics pipeline that is itself the thing being fixed. + +use async_trait::async_trait; +use bytes::Bytes; +use http::{Request, Response}; +use opentelemetry_http::{HttpClient, HttpError}; + +use crate::{QueueConfig, mint_token_for_blocking}; + +/// See the module doc. Wraps a plain `reqwest::blocking::Client` — deliberately +/// not `opentelemetry_http`'s own blanket impl for that type (behind its +/// `reqwest-blocking` cargo feature, unused here): that impl has no seam to +/// mint a token per call, which is this type's entire reason to exist. +#[derive(Debug)] +pub struct AuthenticatedHttpClient { + inner: reqwest::blocking::Client, + cfg: QueueConfig, + /// The audience requested at the token endpoint — see + /// [`crate::mint_token_for_blocking`]'s own doc (via + /// [`crate::mint_token_for`]'s) for why this has to be asked for + /// explicitly rather than left to whatever the identity's default is. + audience: String, +} + +impl AuthenticatedHttpClient { + /// `inner` is the caller's own client (already configured with + /// whatever timeout/TLS trust the destination needs) — this type only + /// adds the auth header, it does not build a client of its own, so a + /// caller reusing an existing `reqwest::blocking::Client` doesn't pay + /// for a second connection pool. + #[must_use] + pub fn new( + inner: reqwest::blocking::Client, + cfg: QueueConfig, + audience: impl Into, + ) -> Self { + Self { + inner, + cfg, + audience: audience.into(), + } + } +} + +#[async_trait] +impl HttpClient for AuthenticatedHttpClient { + /// `async fn` in signature, synchronous in body — matching + /// `opentelemetry_http`'s own `reqwest::blocking::Client` impl exactly + /// (see that impl's source): the trait is async because some callers + /// need that, not because this one does, and the module doc explains + /// why this specific caller deliberately runs on a thread with nothing + /// to yield to. + async fn send_bytes(&self, mut request: Request) -> Result, HttpError> { + let token = mint_token_for_blocking(&self.cfg, Some(&self.audience))?; + request.headers_mut().insert( + http::header::AUTHORIZATION, + http::HeaderValue::from_str(&format!("Bearer {token}"))?, + ); + + let req = request.try_into()?; + let mut response = self.inner.execute(req)?.error_for_status()?; + // Same header/body reassembly `opentelemetry_http`'s own blocking + // impl does — `reqwest::blocking::Response` and `http::Response` + // are different types, so this is the required conversion, not an + // extra step invented here. + let headers = std::mem::take(response.headers_mut()); + let mut http_response = Response::builder() + .status(response.status()) + .body(response.bytes()?)?; + *http_response.headers_mut() = headers; + Ok(http_response) + } +} From 361396cab881771df5d74c69ff0a788fc557f7e9 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 22:56:44 +0200 Subject: [PATCH 4/6] 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. From f3be08f6b690df79e5953708f11ad64a9ba94508 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 23:38:20 +0200 Subject: [PATCH 5/6] move otel_http_client from swarm-queue-client into swarm-controller --- Cargo.lock | 7 ++-- Cargo.toml | 4 +-- hive-jobq-metrics/src/lib.rs | 10 +++--- swarm-controller/Cargo.toml | 18 +++++++---- swarm-controller/src/main.rs | 1 + .../src/otel_http_client.rs | 32 +++++++++++-------- swarm-controller/src/vcs_metrics.rs | 24 +++++++------- swarm-queue-client/Cargo.toml | 18 ----------- swarm-queue-client/src/lib.rs | 8 ----- 9 files changed, 52 insertions(+), 70 deletions(-) rename {swarm-queue-client => swarm-controller}/src/otel_http_client.rs (76%) diff --git a/Cargo.lock b/Cargo.lock index 8b7f5a70..88c8eb6c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4575,8 +4575,10 @@ version = "0.1.0" dependencies = [ "anyhow", "async-nats", + "async-trait", "axum", "base64", + "bytes", "forgejo-api", "futures-util", "hive-jobq", @@ -4584,6 +4586,7 @@ dependencies = [ "hive-jobq-wire", "hive-types", "hmac 0.13.0", + "http", "opentelemetry", "opentelemetry-http", "opentelemetry-otlp", @@ -4629,10 +4632,6 @@ name = "swarm-queue-client" version = "0.1.0" dependencies = [ "async-nats", - "async-trait", - "bytes", - "http", - "opentelemetry-http", "reqwest", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index e20a2254..4733a5dc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -147,8 +147,8 @@ opentelemetry-otlp = { version = "0.32", default-features = false, features = [ # The trait an OTLP exporter's HTTP transport is built on # (`opentelemetry_otlp::WithHttpConfig::with_http_client`) — pulled in # directly (not just transitively via `opentelemetry-otlp` above) by -# `swarm-queue-client`'s `otel-auth` feature, which implements the trait -# itself rather than using the crate's own blanket `reqwest`/ +# `swarm-controller`'s own `otel_http_client` module, which implements the +# trait itself rather than using the crate's own blanket `reqwest`/ # `reqwest-blocking` impls (this workspace's implementation needs to mint # a fresh bearer token per request, which no blanket impl can do). No # extra cargo features requested here for exactly that reason — the diff --git a/hive-jobq-metrics/src/lib.rs b/hive-jobq-metrics/src/lib.rs index 60373b1c..9c45d453 100644 --- a/hive-jobq-metrics/src/lib.rs +++ b/hive-jobq-metrics/src/lib.rs @@ -34,9 +34,9 @@ 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 +/// caller's concrete type (`swarm-controller`'s own +/// `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. /// @@ -84,8 +84,8 @@ const DEFAULT_INTERVAL: Duration = Duration::from_mins(1); /// 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). +/// (`swarm-controller`'s case — see its own `otel_http_client` module doc +/// for why a static header doesn't fit an expiring token). pub fn spawn_exporter( jobq: Arc>>, service_name: &str, diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 8ea1ec6e..32d2ce10 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -84,16 +84,20 @@ swarm-authelia-bridge-sock.workspace = true # creation config are shared with the hive that writes it, so this end does # not get to declare them privately. # -# `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 +swarm-queue-client = { workspace = true, features = ["kv"] } +# `otel_http_client.rs`'s `AuthenticatedHttpClient` — an +# `opentelemetry_http::HttpClient` impl authenticated with this crate's own +# `swarm-queue-client` identity. Lives in this crate rather than +# `swarm-queue-client` (mara's own call during review) since this daemon is +# its only caller; see that module's doc for the full rationale. `opentelemetry-http` +# is also needed directly (not just transitively) 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. +# `hive_jobq_metrics::spawn_exporter`. opentelemetry-http.workspace = true +async-trait.workspace = true +bytes.workspace = true +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 57bf51cc..ea09dfc9 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -43,6 +43,7 @@ use utoipa_axum::{router::OpenApiRouter, routes}; mod auth; mod config_pr; mod forge; +mod otel_http_client; mod status; mod vcs_metrics; mod webhook; diff --git a/swarm-queue-client/src/otel_http_client.rs b/swarm-controller/src/otel_http_client.rs similarity index 76% rename from swarm-queue-client/src/otel_http_client.rs rename to swarm-controller/src/otel_http_client.rs index ce526637..3429cba8 100644 --- a/swarm-queue-client/src/otel_http_client.rs +++ b/swarm-controller/src/otel_http_client.rs @@ -1,17 +1,24 @@ //! An [`opentelemetry_http::HttpClient`] that authenticates every request -//! with a fresh, audience-scoped bearer token minted from this crate's own -//! [`QueueConfig`] identity — "one identity per principal" applied to a -//! third destination (the queue connection and `swarm-authelia-bridge`'s -//! client, [`crate::mint_token_for`], are the other two). +//! with a fresh, audience-scoped bearer token minted from this daemon's own +//! [`swarm_queue_client::QueueConfig`] identity — "one identity per +//! principal" applied to a third destination (the queue connection and +//! `auth.rs`'s bridge client, `swarm_queue_client::mint_token_for`, are the +//! other two). +//! +//! Lives here, not in `swarm-queue-client` (mara's own call during review: +//! it does not belong in the queue-connect crate) — this daemon is the only +//! caller, and the module only *reuses* that crate's +//! `QueueConfig`/`mint_token_for_blocking`, it doesn't extend the +//! queue-connect contract those exist for. //! //! Exists for exactly one shape of caller: an OTLP metric exporter whose //! `PeriodicReader` drives export from a background thread with **no tokio //! reactor** (`opentelemetry-otlp`'s `reqwest-blocking-client` feature is //! chosen everywhere in this workspace for that reason — see -//! `swarm-controller::vcs_metrics`'s module doc). That constraint is why -//! this mints via [`crate::mint_token_for_blocking`] rather than the async -//! [`crate::mint_token_for`]: the latter needs a runtime to `.await` on, -//! which the calling thread does not have. +//! `vcs_metrics`'s module doc). That constraint is why this mints via +//! [`swarm_queue_client::mint_token_for_blocking`] rather than the async +//! `swarm_queue_client::mint_token_for`: the latter needs a runtime to +//! `.await` on, which the calling thread does not have. //! //! Per-request, not per-tick or cached: `opentelemetry-otlp` 0.32 exposes //! `WithHttpConfig::with_http_client`, a seam for a caller-supplied @@ -25,8 +32,7 @@ use async_trait::async_trait; use bytes::Bytes; use http::{Request, Response}; use opentelemetry_http::{HttpClient, HttpError}; - -use crate::{QueueConfig, mint_token_for_blocking}; +use swarm_queue_client::{QueueConfig, mint_token_for_blocking}; /// See the module doc. Wraps a plain `reqwest::blocking::Client` — deliberately /// not `opentelemetry_http`'s own blanket impl for that type (behind its @@ -37,9 +43,9 @@ pub struct AuthenticatedHttpClient { inner: reqwest::blocking::Client, cfg: QueueConfig, /// The audience requested at the token endpoint — see - /// [`crate::mint_token_for_blocking`]'s own doc (via - /// [`crate::mint_token_for`]'s) for why this has to be asked for - /// explicitly rather than left to whatever the identity's default is. + /// [`swarm_queue_client::mint_token_for_blocking`]'s own doc (via + /// `swarm_queue_client::mint_token_for`'s) for why this has to be asked + /// for explicitly rather than left to whatever the identity's default is. audience: String, } diff --git a/swarm-controller/src/vcs_metrics.rs b/swarm-controller/src/vcs_metrics.rs index 09f169ef..10d51723 100644 --- a/swarm-controller/src/vcs_metrics.rs +++ b/swarm-controller/src/vcs_metrics.rs @@ -118,8 +118,8 @@ fn build_provider(interval: Duration) -> Result { // `.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. + // `crate::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) @@ -135,8 +135,8 @@ fn build_provider(interval: Duration) -> Result { .build()) } -/// Build the [`swarm_queue_client::otel_http_client::AuthenticatedHttpClient`] -/// this exporter pushes through. +/// Build the [`crate::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 @@ -160,8 +160,8 @@ fn build_provider(interval: Duration) -> Result { /// 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 { +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(|| { @@ -175,13 +175,11 @@ pub(crate) fn authenticated_http_client() "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, - ), - ) + Ok(crate::otel_http_client::AuthenticatedHttpClient::new( + reqwest::blocking::Client::new(), + cfg, + audience, + )) } /// `service.name = swarm-controller` plus whatever the operator set in diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index e7640599..3798b01d 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -30,18 +30,6 @@ kv = ["async-nats/kv"] # `cargo check -p swarm-nats-auth` — no `kv` anywhere in that build — # surfaced it as `cannot find jetstream in async_nats`). notices = ["async-nats/jetstream"] -# OFF by default, same reasoning as `kv`/`notices` above: only a caller -# wiring an OTLP exporter's `HttpClient` seam to this crate's identity -# (today, `swarm-controller`'s `vcs_metrics`/`hive_jobq_metrics`) needs -# `opentelemetry_http`/`async-trait`/`bytes`/`http` pulled in — a plain -# queue-connect-only consumer (the auth-callout responder, a hive -# publishing its status) has no reason to carry them. -otel-auth = [ - "dep:opentelemetry-http", - "dep:async-trait", - "dep:bytes", - "dep:http", -] [dependencies] # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature @@ -57,12 +45,6 @@ async-nats.workspace = true # feature's comment above), and `cargo check -p swarm-queue-client` alone # must not depend on what else is in the build. reqwest = { workspace = true, features = ["blocking"] } -# All four `optional = true`, gated behind the `otel-auth` feature above — -# see that feature's own comment for why. -opentelemetry-http = { workspace = true, optional = true } -async-trait = { workspace = true, optional = true } -bytes = { workspace = true, optional = true } -http = { workspace = true, optional = true } serde.workspace = true serde_json.workspace = true # A library, so its errors are a matchable enum rather than an opaque diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 1a7acbbe..1165142e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -164,14 +164,6 @@ pub fn chain(error: &dyn std::error::Error) -> String { /// which is the disagreement this module exists to prevent. pub mod status; -/// An [`opentelemetry_http::HttpClient`] impl that authenticates every -/// request with a fresh token from this crate's identity — see the -/// module's own doc for the full rationale. Behind the `otel-auth` -/// feature, same reasoning as `status`/`kv` above: only a caller wiring an -/// OTLP exporter to this identity needs the extra dependencies it pulls in. -#[cfg(feature = "otel-auth")] -pub mod otel_http_client; - /// The subject the swarm controller publishes on when the hive-wide knowledge /// repository has changed. One writer, many readers — every hive subscribes. /// From 350b4e9fb77d0092c266231aaa04a8752a7c64ee Mon Sep 17 00:00:00 2001 From: damocles Date: Thu, 27 Aug 2026 00:24:19 +0200 Subject: [PATCH 6/6] swarm-otel/swarm-controller: authenticate + route the swarm-tier OTLP producer through the gateway --- nix/host-modules/swarm-controller.nix | 65 +++++---- nix/host-modules/swarm-otel.nix | 181 ++++++++++++++++++-------- swarm-queue-client/src/lib.rs | 5 +- 3 files changed, 170 insertions(+), 81 deletions(-) diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 672481ff..cd100f3e 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -155,27 +155,39 @@ let # above, genuinely optional and gated on the option resolving rather # than assumed. # - # Gated on THIS HOST running `swarm-otel`, not merely on it existing - # somewhere in the swarm: its receiver for a swarm-tier producer binds - # `127.0.0.1` only (see `swarm-otel.nix::producerPort`), so reaching it - # needs co-location, and this checks that directly rather than assuming - # it — the unasserted-co-location gap flagged elsewhere in this codebase - # doesn't apply here because there is nothing to assert: a controller on - # a host that doesn't run the collector simply exports nothing, the same - # graceful absence `forgeEnv` already models. + # By NAME through the gateway, never a loopback literal or a bare + # `producerPort` — mara's ruling on this exact module: "the swarm + # services dont have to run on the same host as the swarm controller." + # A controller and the swarm's collector are two independently-placed + # swarm services, exactly like a hive and the collector already are + # (`otel.nix`'s own `endpoint`), so this reaches it the same way: by + # its swarm-service name, resolved locally by dnsmasq on a co-located + # host and over the real network otherwise — no loopback-vs-remote + # knob to get wrong, and nothing here to gate on co-location because + # there is no co-location requirement left to check. + # + # No trailing slash: `opentelemetry_otlp`'s http/json exporter appends + # `/v1/metrics` to this base itself, and the swarm-otel gateway route + # (`swarm-otel.nix`'s nginx `locations."/${producerName}/"`) strips the + # `/${producerName}` prefix before the request reaches the receiver — + # so the two ends agree on exactly one path, `/v1/metrics`, arrived at + # from opposite directions. + # + # Gated on `otelSwarmCfg.enable` alone (a swarm-wide fact, not "enabled + # HERE"): the collector is reachable by name wherever it runs, so a + # controller not co-located with it now exports over https instead of + # exporting nothing — the graceful-absence case left is "no swarm-otel + # anywhere in this swarm at all," which `forgeEnv` already models the + # same way. otelSwarmCfg = config.services.hyperhive.swarm.otel; otelEnv = lib.optionalAttrs otelSwarmCfg.enable { - OTEL_EXPORTER_OTLP_ENDPOINT = "http://${otelSwarmCfg.domain}:${toString otelSwarmCfg.producerPort}"; - }; - - # Only when THIS host also runs `swarm-otel` — same reasoning and same - # shape as `hive-forge/default.nix`'s `ssoLocal`-gated entry: a raw host - # systemd unit does not get the bridge's dnsmasq resolution containers - # get, so the name that reaches a co-located collector over `otelEnv` - # above needs an explicit loopback alias here, or it resolves however - # (or however it fails to) off-host DNS says on this host. - otelHostsEntry = lib.optionalAttrs otelSwarmCfg.enable { - "127.0.0.1" = [ otelSwarmCfg.domain ]; + OTEL_EXPORTER_OTLP_ENDPOINT = "https://${otelSwarmCfg.domain}/${otelSwarmCfg.producerName}"; + # Checked by `swarm-otel.nix`'s `oidc/${producerName}` authenticator + # against exactly this controller's own registered audience (see the + # `audience = [ queueClientId ]` client entry below) — the two have + # to name the same value or the exporter mints a token the receiver + # was never told to accept. + SWARM_CONTROLLER_OTEL_AUDIENCE = queueClientId; }; # Wrapped rather than documented: every one of these values is derived @@ -479,10 +491,6 @@ in "swarm-controller-credential" ]; - # See `otelHostsEntry`'s own comment: only present, and only ever - # `{ "127.0.0.1" = [ ... ]; }`, when this host also runs `swarm-otel`. - networking.hosts = otelHostsEntry; - users.users.swarm-controller = { isSystemUser = true; group = "swarm-controller"; @@ -509,6 +517,17 @@ in # redirect list, because authelia permits only the grants a client # names and an omitted `grant_types` means authorization-code alone. kind = "machine"; + # Its own id as its own permitted audience — same "client's own id + # is the value it may ask for" shape `swarm-authelia.nix`'s + # `hiveClients` uses, just self-referential instead of prefixed: + # this controller is not a hive (see `queueClientId`'s own doc + # comment), so it has no `hiveClientPrefix` name to reuse. This is + # what lets `vcs_metrics::authenticated_http_client()` mint a token + # for `SWARM_CONTROLLER_OTEL_AUDIENCE` and have `swarm-otel.nix`'s + # `oidc/swarm` authenticator (which checks for this same id) accept + # it — registering here without requesting there, or the reverse, + # both fail as a healthy-looking 401 several layers from this line. + audience = [ queueClientId ]; } ]; diff --git a/nix/host-modules/swarm-otel.nix b/nix/host-modules/swarm-otel.nix index e037c448..b39e66bd 100644 --- a/nix/host-modules/swarm-otel.nix +++ b/nix/host-modules/swarm-otel.nix @@ -54,7 +54,11 @@ let # that reserves it is checking the same string the config emits. A literal # repeated at each site would let the guard and the config drift apart, which # is the failure this guard exists to prevent. - swarmTierName = "swarm"; + # Read-only option below, not a bare literal — `swarm-controller.nix` needs + # the identical string to build the same audience/endpoint, and a value + # bound once here (rather than copy-pasted at both sites) is the only way + # the two can't drift apart. + swarmTierName = cfg.producerName; # Every `` no hive may take. Read from `nix/reserved-names.nix`, the # same file the daemons are handed as `HIVE_RESERVED_NAMES`, because agent # names and hive names are ONE namespace going forward — a locally-owned @@ -63,7 +67,8 @@ let # # The assertion below still checks the string this module emits: the file # is asserted to CONTAIN `swarmTierName`, so a rename that dropped it from - # the file would be an eval error rather than a silently missing guard. + # the file (or a `producerName` override the file was never updated for) + # would be an eval error rather than a silently missing guard. reservedOwners = import ../reserved-names.nix; # A published target is declared as ONE url, because that url is also the @@ -253,14 +258,25 @@ in type = lib.types.port; default = 4390; description = '' - Port the swarm tier's own unauthenticated OTLP/HTTP receiver - listens on, at `127.0.0.1` — for a **swarm-level** producer - (`swarm-controller`'s vcs/jobq counters today) to push to, on the - same host this collector runs on. No authenticator, deliberately: - unlike the per-hive receivers, the label this receiver's samples - get (`swarm`) is the one the receiver's own *existence* - establishes — there is no hive identity to forge or attribute, - so the per-hive attribution mechanism does not apply here. + Port the swarm tier's own OTLP/HTTP receiver for a + **swarm-level** producer (`swarm-controller`'s vcs/jobq counters + today) listens on, at `127.0.0.1`. Reached through the gateway at + `https://''${domain}/''${producerName}/`, same shape as a hive's + own receiver — a swarm-level producer may not share a host with + this collector (mara, on the swarm-controller topology: "the + swarm services dont have to run on the same host as the swarm + controller"), so loopback-only reachability is not a supported + shape here any more than it is for a hive. + + Authenticated by `oidc/''${producerName}`, checking for the + audience `services.hyperhive.swarm.controller.queueClientId` + requests — not the per-hive `hiveClientPrefix` namespace, because + a swarm-level producer is explicitly not a hive (see that + option's own doc comment). The `swarm` label this receiver's + samples get is still the one the receiver's own *existence* + establishes, same as before — there is no hive identity being + forged or attributed here, just a caller proving it is the one + principal allowed to push into this pipeline. ⚠️ Deliberately NOT derived from `port + (number of hives)`: that range grows every time a hive is added, and a fixed offset from @@ -272,6 +288,23 @@ in ''; }; + producerName = lib.mkOption { + type = lib.types.str; + readOnly = true; + default = "swarm"; + description = '' + The swarm-tier producer's own component/path/authenticator name + — the same reserved literal this module's collector components + (`otlp/swarm`, `oidc/swarm`, `resource/swarm`, `metrics/swarm`) + are already named after internally. Published as an option so a + sibling module (`swarm-controller.nix`) can address the receiver + by name — `https://''${domain}/''${producerName}/` — instead of + repeating the string. Read-only for the same reason + `clientId`/`machine`/`domain` are: two spellings of a name that + has to match on both ends is a mismatch waiting to happen. + ''; + }; + domain = lib.mkOption { type = lib.types.str; default = "otel.${domainBase}"; @@ -446,9 +479,15 @@ in h: p: lib.nameValuePair "/${h}/" { proxyPass = "http://127.0.0.1:${toString p}/"; } ) hivePorts // { + # The swarm-tier producer's own route — same trailing-slash + # shape as the per-hive locations above, and load-bearing for + # the same reason: it strips `/${producerName}` before the + # request reaches the receiver, which knows nothing about the + # path it was found at. + "/${swarmTierName}/".proxyPass = "http://127.0.0.1:${toString cfg.producerPort}/"; # There is no swarm-wide inbox, and a closed door is the honest - # description of that. Every route into this collector belongs to - # exactly one hive. + # description of that. Every other route into this collector + # belongs to exactly one hive or the swarm-tier producer above. "/".return = "404"; }; }; @@ -870,11 +909,16 @@ in }; } ) hivePorts - # The swarm tier's OWN receiver, unauthenticated on - # purpose — see `producerPort`'s description for why no - # per-hive-shaped attribution applies here. + # The swarm tier's OWN receiver — authenticated exactly + # like a hive's, just against a different identity (the + # swarm-level producer is not a hive; see + # `producerPort`'s description for what its audience + # checks and why). // { - "otlp/${swarmTierName}".protocols.http.endpoint = "127.0.0.1:${toString cfg.producerPort}"; + "otlp/${swarmTierName}".protocols.http = { + endpoint = "127.0.0.1:${toString cfg.producerPort}"; + auth.authenticator = "oidc/${swarmTierName}"; + }; } # MERGED with the per-hive receivers, never assigned over # them. A plain assignment here would drop every hive's @@ -1050,7 +1094,13 @@ in # INERT — the collector starts clean and the receiver # naming it authenticates nothing. Derived from the same # attrset as the receivers so the two cannot disagree. - service.extensions = map (h: "oidc/${h}") (lib.attrNames hivePorts); + service.extensions = + map (h: "oidc/${h}") (lib.attrNames hivePorts) + # The swarm-tier producer's own authenticator — unconditional, + # same reasoning as its receiver above: `otlp/${swarmTierName}` + # always exists, so the authenticator it names must too, or an + # extension-not-listed startup failure follows every deploy. + ++ [ "oidc/${swarmTierName}" ]; # Fan-out, not a choice: with both configured the same # samples go upstream AND into the swarm's store. The store @@ -1107,45 +1157,64 @@ in }; } // { - extensions = lib.mapAttrs' ( - h: _: - lib.nameValuePair "oidc/${h}" { - issuer_url = autheliaCfg.url; - # The audience this hive's client is registered to - # request, and the reason one hive's token is refused by - # another hive's receiver. Same expression authelia - # registers it under — a second spelling here would deny - # every hive, as a 401 that blames the token. - audience = "${autheliaCfg.hiveClientPrefix}${h}"; - # ⛔ DO NOT ADD `issuer_ca_path` HERE. It took this - # collector down for forty minutes once, and the failure - # is invisible to every check we have. - # - # It loads only the FIRST certificate in the file it - # names. The bundle assembled for this container is - # `system CAs ++ hive trust bundle`, so the anchor sits - # ~123rd and is never in the pool: the extension then - # cannot verify authelia and the whole collector exits - # `x509: certificate signed by unknown authority`, on - # every start, with 125 valid certificates in the file. - # - # Leaving it unset makes the extension use the process - # trust store, which `trustBundle` already populates via - # `SSL_CERT_FILE` — and *that* consumer reads every - # certificate regardless of order. One file, two - # consumers, opposite parsing: the fix is to stop naming - # it twice, not to reorder the bundle. - # - # ⚠️ Nor is pointing it at the hive trust bundle a fix: - # `hive-tls.nix` writes that leading with the *hive* CA - # (`nameConstraints` = this hive's domain), which cannot - # issue a swarm-level name at all. - # - # (Kept from the original note, still true and still - # worth not re-deriving: `issuer_ca_file`, `ca_file` and - # `tls.ca_file` are INVALID KEYS for this extension.) - } - ) hivePorts; + extensions = + lib.mapAttrs' ( + h: _: + lib.nameValuePair "oidc/${h}" { + issuer_url = autheliaCfg.url; + # The audience this hive's client is registered to + # request, and the reason one hive's token is refused by + # another hive's receiver. Same expression authelia + # registers it under — a second spelling here would deny + # every hive, as a 401 that blames the token. + audience = "${autheliaCfg.hiveClientPrefix}${h}"; + # ⛔ DO NOT ADD `issuer_ca_path` HERE. It took this + # collector down for forty minutes once, and the failure + # is invisible to every check we have. + # + # It loads only the FIRST certificate in the file it + # names. The bundle assembled for this container is + # `system CAs ++ hive trust bundle`, so the anchor sits + # ~123rd and is never in the pool: the extension then + # cannot verify authelia and the whole collector exits + # `x509: certificate signed by unknown authority`, on + # every start, with 125 valid certificates in the file. + # + # Leaving it unset makes the extension use the process + # trust store, which `trustBundle` already populates via + # `SSL_CERT_FILE` — and *that* consumer reads every + # certificate regardless of order. One file, two + # consumers, opposite parsing: the fix is to stop naming + # it twice, not to reorder the bundle. + # + # ⚠️ Nor is pointing it at the hive trust bundle a fix: + # `hive-tls.nix` writes that leading with the *hive* CA + # (`nameConstraints` = this hive's domain), which cannot + # issue a swarm-level name at all. + # + # (Kept from the original note, still true and still + # worth not re-deriving: `issuer_ca_file`, `ca_file` and + # `tls.ca_file` are INVALID KEYS for this extension.) + } + ) hivePorts + # The swarm-tier producer's own authenticator. Same extension + # type and same `issuer_url`/no-`issuer_ca_path` shape as every + # `oidc/${h}` above — only the audience differs, and + # deliberately does not reuse `hiveClientPrefix`: + # `swarm-controller` is explicitly not a hive (see its own + # module's `queueClientId` doc comment), so its audience is + # its own client id rather than a value from the per-hive + # namespace. Read via the option rather than a literal so a + # rename of `queueClientId`'s default cannot silently + # desync the two ends — this authenticator and the client + # requesting the audience it checks both have to agree, or + # a real swarm-controller push gets a healthy-looking 401. + // { + "oidc/${swarmTierName}" = { + issuer_url = autheliaCfg.url; + audience = config.services.hyperhive.swarm.controller.queueClientId; + }; + }; # `upsert`, not `insert`: a sender that stamps its own # `hive` must be OVERWRITTEN, not deferred to. This diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index 1165142e..c7a9143f 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -452,8 +452,9 @@ fn build_http_client(cfg: &QueueConfig) -> Result { /// per call, same as this crate did before the reconnect-storm fix added the /// cache. /// -/// `audience` is passed straight to [`token_request`] — see its doc for why -/// it is optional and when a caller needs it. `None` reproduces this +/// `audience` is passed straight to the private `token_request` helper — +/// see its doc for why it is optional and when a caller needs it. `None` +/// reproduces this /// function's behaviour before the parameter existed, so every caller from /// before that added it (the queue connect path, `auth.rs`'s bridge client) /// is unaffected.