swarm-queue-client: add an authenticated HttpClient for OTLP exporters (otel-auth feature)

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",

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

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

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.
 ///

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<String>,
+    ) -> 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<Bytes>) -> Result<Response<Bytes>, 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)
+    }
+}