From 7913be5435d461eee104aaa9d1c3a74efac1ca28 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 26 Aug 2026 22:38:18 +0200 Subject: [PATCH] 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) + } +}