wire the authenticated OTLP push into vcs_metrics and hive-jobq-metrics

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

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

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<dyn HttpClient>`: 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<dyn HttpClient>);
+
+#[async_trait]
+impl HttpClient for BoxedHttpClient {
+    async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, 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<N, R>(
     jobq: Arc<Mutex<hive_jobq::scheduler::Scheduler<N, R>>>,
     service_name: &str,
+    http_client: Option<Box<dyn HttpClient>>,
 ) -> Option<SdkMeterProvider>
 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<Mutex<Vec<StateCount>>>,
     service_name: &str,
+    http_client: Option<Box<dyn HttpClient>>,
 ) -> Result<SdkMeterProvider> {
     // 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();

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<dyn opentelemetry_http::HttpClient>` 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

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<dyn opentelemetry_http::HttpClient>);
     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

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<SdkMeterProvider> {
     // 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<SdkMeterProvider> {
         .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<swarm_queue_client::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(|| {
+            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.