100 lines
4.7 KiB
Rust
100 lines
4.7 KiB
Rust
//! 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)
|
|
}
|
|
}
|