move otel_http_client from swarm-queue-client into swarm-controller

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;

swarm-controller/src/otel_http_client.rs

@@ -0,0 +1,100 @@
+//! An [`opentelemetry_http::HttpClient`] that authenticates every request
+//! 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
+//! `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
+//! [`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 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
+/// `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
+    /// [`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,
+}
+
+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)
+    }
+}

swarm-controller/src/vcs_metrics.rs

@@ -118,8 +118,8 @@ fn build_provider(interval: Duration) -> Result<SdkMeterProvider> {
     // `.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<SdkMeterProvider> {
         .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<SdkMeterProvider> {
 /// 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<swarm_queue_client::otel_http_client::AuthenticatedHttpClient> {
+pub(crate) fn authenticated_http_client() -> Result<crate::otel_http_client::AuthenticatedHttpClient>
+{
     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