swarm-otel/swarm-controller: authenticate + route the swarm-tier OTLP producer through the gateway

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",
@@ -4573,8 +4575,10 @@ version = "0.1.0"
 dependencies = [
  "anyhow",
  "async-nats",
+ "async-trait",
  "axum",
  "base64",
+ "bytes",
  "forgejo-api",
  "futures-util",
  "hive-jobq",
@@ -4582,7 +4586,9 @@ dependencies = [
  "hive-jobq-wire",
  "hive-types",
  "hmac 0.13.0",
+ "http",
  "opentelemetry",
+ "opentelemetry-http",
  "opentelemetry-otlp",
  "opentelemetry_sdk",
  "problem_details",

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-controller`'s own `otel_http_client` module, 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

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 own
+/// `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 its own `otel_http_client` 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)
+        .with_protocol(Protocol::HttpJson);
-        .build()
+    // Only when the caller supplied one — see `spawn_exporter`'s doc for
-        .context("build OTLP metric exporter")?;
+    // 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();

nix/host-modules/swarm-controller.nix

@@ -149,6 +149,47 @@ let
     SWARM_CONTROLLER_NAME = config.services.hyperhive.swarm.name;
   };
 
+  # `vcs_metrics`/`hive_jobq_metrics` (this daemon's OTLP push exporters)
+  # read the standard `OTEL_EXPORTER_OTLP_ENDPOINT` var and are silent,
+  # graceful no-ops without it — same shape as `forgeEnv`/`authBridgeEnv`
+  # above, genuinely optional and gated on the option resolving rather
+  # than assumed.
+  #
+  # By NAME through the gateway, never a loopback literal or a bare
+  # `producerPort` — mara's ruling on this exact module: "the swarm
+  # services dont have to run on the same host as the swarm controller."
+  # A controller and the swarm's collector are two independently-placed
+  # swarm services, exactly like a hive and the collector already are
+  # (`otel.nix`'s own `endpoint`), so this reaches it the same way: by
+  # its swarm-service name, resolved locally by dnsmasq on a co-located
+  # host and over the real network otherwise — no loopback-vs-remote
+  # knob to get wrong, and nothing here to gate on co-location because
+  # there is no co-location requirement left to check.
+  #
+  # No trailing slash: `opentelemetry_otlp`'s http/json exporter appends
+  # `/v1/metrics` to this base itself, and the swarm-otel gateway route
+  # (`swarm-otel.nix`'s nginx `locations."/${producerName}/"`) strips the
+  # `/${producerName}` prefix before the request reaches the receiver —
+  # so the two ends agree on exactly one path, `/v1/metrics`, arrived at
+  # from opposite directions.
+  #
+  # Gated on `otelSwarmCfg.enable` alone (a swarm-wide fact, not "enabled
+  # HERE"): the collector is reachable by name wherever it runs, so a
+  # controller not co-located with it now exports over https instead of
+  # exporting nothing — the graceful-absence case left is "no swarm-otel
+  # anywhere in this swarm at all," which `forgeEnv` already models the
+  # same way.
+  otelSwarmCfg = config.services.hyperhive.swarm.otel;
+  otelEnv = lib.optionalAttrs otelSwarmCfg.enable {
+    OTEL_EXPORTER_OTLP_ENDPOINT = "https://${otelSwarmCfg.domain}/${otelSwarmCfg.producerName}";
+    # Checked by `swarm-otel.nix`'s `oidc/${producerName}` authenticator
+    # against exactly this controller's own registered audience (see the
+    # `audience = [ queueClientId ]` client entry below) — the two have
+    # to name the same value or the exporter mints a token the receiver
+    # was never told to accept.
+    SWARM_CONTROLLER_OTEL_AUDIENCE = queueClientId;
+  };
+
   # Wrapped rather than documented: every one of these values is derived
   # from an option this deployment already set, so making the operator
   # re-supply them on the command line would be asking them to repeat the
@@ -476,6 +517,17 @@ in
         # redirect list, because authelia permits only the grants a client
         # names and an omitted `grant_types` means authorization-code alone.
         kind = "machine";
+        # Its own id as its own permitted audience — same "client's own id
+        # is the value it may ask for" shape `swarm-authelia.nix`'s
+        # `hiveClients` uses, just self-referential instead of prefixed:
+        # this controller is not a hive (see `queueClientId`'s own doc
+        # comment), so it has no `hiveClientPrefix` name to reuse. This is
+        # what lets `vcs_metrics::authenticated_http_client()` mint a token
+        # for `SWARM_CONTROLLER_OTEL_AUDIENCE` and have `swarm-otel.nix`'s
+        # `oidc/swarm` authenticator (which checks for this same id) accept
+        # it — registering here without requesting there, or the reverse,
+        # both fail as a healthy-looking 401 several layers from this line.
+        audience = [ queueClientId ];
       }
     ];
 
@@ -676,7 +728,8 @@ in
       // forgeEnv
       // webhookEnv
       // authBridgeEnv
-      // swarmNameEnv;
+      // swarmNameEnv
+      // otelEnv;
     };
 
     # A systemd credential is a SNAPSHOT: it is materialised into `%d` once,

nix/host-modules/swarm-otel.nix

@@ -54,7 +54,11 @@ let
   # that reserves it is checking the same string the config emits. A literal
   # repeated at each site would let the guard and the config drift apart, which
   # is the failure this guard exists to prevent.
-  swarmTierName = "swarm";
+  # Read-only option below, not a bare literal — `swarm-controller.nix` needs
+  # the identical string to build the same audience/endpoint, and a value
+  # bound once here (rather than copy-pasted at both sites) is the only way
+  # the two can't drift apart.
+  swarmTierName = cfg.producerName;
   # Every `<owner>` no hive may take. Read from `nix/reserved-names.nix`, the
   # same file the daemons are handed as `HIVE_RESERVED_NAMES`, because agent
   # names and hive names are ONE namespace going forward — a locally-owned
@@ -63,7 +67,8 @@ let
   #
   # The assertion below still checks the string this module emits: the file
   # is asserted to CONTAIN `swarmTierName`, so a rename that dropped it from
-  # the file would be an eval error rather than a silently missing guard.
+  # the file (or a `producerName` override the file was never updated for)
+  # would be an eval error rather than a silently missing guard.
   reservedOwners = import ../reserved-names.nix;
 
   # A published target is declared as ONE url, because that url is also the
@@ -249,6 +254,57 @@ in
       '';
     };
 
+    producerPort = lib.mkOption {
+      type = lib.types.port;
+      default = 4390;
+      description = ''
+        Port the swarm tier's own OTLP/HTTP receiver for a
+        **swarm-level** producer (`swarm-controller`'s vcs/jobq counters
+        today) listens on, at `127.0.0.1`. Reached through the gateway at
+        `https://''${domain}/''${producerName}/`, same shape as a hive's
+        own receiver — a swarm-level producer may not share a host with
+        this collector (mara, on the swarm-controller topology: "the
+        swarm services dont have to run on the same host as the swarm
+        controller"), so loopback-only reachability is not a supported
+        shape here any more than it is for a hive.
+
+        Authenticated by `oidc/''${producerName}`, checking for the
+        audience `services.hyperhive.swarm.controller.queueClientId`
+        requests — not the per-hive `hiveClientPrefix` namespace, because
+        a swarm-level producer is explicitly not a hive (see that
+        option's own doc comment). The `swarm` label this receiver's
+        samples get is still the one the receiver's own *existence*
+        establishes, same as before — there is no hive identity being
+        forged or attributed here, just a caller proving it is the one
+        principal allowed to push into this pipeline.
+
+        ⚠️ Deliberately NOT derived from `port + (number of hives)`: that
+        range grows every time a hive is added, and a fixed offset from
+        it would silently start colliding once the hive count caught up.
+        Kept as its own reserved value instead, and the assertion below
+        still catches a real collision (including one hive growth
+        eventually causes) rather than starting a collector that quietly
+        drops one receiver's samples.
+      '';
+    };
+
+    producerName = lib.mkOption {
+      type = lib.types.str;
+      readOnly = true;
+      default = "swarm";
+      description = ''
+        The swarm-tier producer's own component/path/authenticator name
+        — the same reserved literal this module's collector components
+        (`otlp/swarm`, `oidc/swarm`, `resource/swarm`, `metrics/swarm`)
+        are already named after internally. Published as an option so a
+        sibling module (`swarm-controller.nix`) can address the receiver
+        by name — `https://''${domain}/''${producerName}/` — instead of
+        repeating the string. Read-only for the same reason
+        `clientId`/`machine`/`domain` are: two spellings of a name that
+        has to match on both ends is a mismatch waiting to happen.
+      '';
+    };
+
     domain = lib.mkOption {
       type = lib.types.str;
       default = "otel.${domainBase}";
@@ -423,9 +479,15 @@ in
           h: p: lib.nameValuePair "/${h}/" { proxyPass = "http://127.0.0.1:${toString p}/"; }
         ) hivePorts
         // {
+          # The swarm-tier producer's own route — same trailing-slash
+          # shape as the per-hive locations above, and load-bearing for
+          # the same reason: it strips `/${producerName}` before the
+          # request reaches the receiver, which knows nothing about the
+          # path it was found at.
+          "/${swarmTierName}/".proxyPass = "http://127.0.0.1:${toString cfg.producerPort}/";
           # There is no swarm-wide inbox, and a closed door is the honest
-          # description of that. Every route into this collector belongs to
+          # description of that. Every other route into this collector
-          # exactly one hive.
+          # belongs to exactly one hive or the swarm-tier producer above.
           "/".return = "404";
         };
     };
@@ -715,6 +777,7 @@ in
             derived = lib.attrValues hivePorts;
             others = [
               cfg.telemetryPort
+              cfg.producerPort
               otelCfg.collector.port
             ]
             ++ lib.optional vmCfg.enable vmCfg.port;
@@ -846,6 +909,17 @@ in
                     };
                   }
                 ) hivePorts
+                # The swarm tier's OWN receiver — authenticated exactly
+                # like a hive's, just against a different identity (the
+                # swarm-level producer is not a hive; see
+                # `producerPort`'s description for what its audience
+                # checks and why).
+                // {
+                  "otlp/${swarmTierName}".protocols.http = {
+                    endpoint = "127.0.0.1:${toString cfg.producerPort}";
+                    auth.authenticator = "oidc/${swarmTierName}";
+                  };
+                }
                 # MERGED with the per-hive receivers, never assigned over
                 # them. A plain assignment here would drop every hive's
                 # receiver and still render a valid config that starts
@@ -1020,7 +1094,13 @@ in
               # INERT — the collector starts clean and the receiver
               # naming it authenticates nothing. Derived from the same
               # attrset as the receivers so the two cannot disagree.
-              service.extensions = map (h: "oidc/${h}") (lib.attrNames hivePorts);
+              service.extensions =
+                map (h: "oidc/${h}") (lib.attrNames hivePorts)
+                # The swarm-tier producer's own authenticator — unconditional,
+                # same reasoning as its receiver above: `otlp/${swarmTierName}`
+                # always exists, so the authenticator it names must too, or an
+                # extension-not-listed startup failure follows every deploy.
+                ++ [ "oidc/${swarmTierName}" ];
 
               # Fan-out, not a choice: with both configured the same
               # samples go upstream AND into the swarm's store. The store
@@ -1044,9 +1124,17 @@ in
                 # them would acquire the one label a swarm-level service must
                 # not have. Keeping it out of them makes the absence
                 # structural rather than something to remember to strip.
-                // lib.optionalAttrs (cfg.scrapeTargets != { }) {
+                // {
+                  # `otlp/${swarmTierName}` is unconditional (see the
+                  # receiver above), so this pipeline is too — a swarm-tier
+                  # producer must always have somewhere to land, unlike
+                  # `prometheus`, which only joins the receiver list once
+                  # something has actually declared a scrape target.
                   "metrics/${swarmTierName}" = {
-                    receivers = [ "prometheus" ];
+                    receivers = [
+                      "otlp/${swarmTierName}"
+                    ]
+                    ++ lib.optional (cfg.scrapeTargets != { }) "prometheus";
                     processors = [ "resource/${swarmTierName}" ];
                     exporters = exporterNames;
                   };
@@ -1069,7 +1157,8 @@ in
                 };
             }
             // {
-              extensions = lib.mapAttrs' (
+              extensions =
+                lib.mapAttrs' (
                   h: _:
                   lib.nameValuePair "oidc/${h}" {
                     issuer_url = autheliaCfg.url;
@@ -1107,7 +1196,25 @@ in
                     # worth not re-deriving: `issuer_ca_file`, `ca_file` and
                     # `tls.ca_file` are INVALID KEYS for this extension.)
                   }
-              ) hivePorts;
+                ) hivePorts
+                # The swarm-tier producer's own authenticator. Same extension
+                # type and same `issuer_url`/no-`issuer_ca_path` shape as every
+                # `oidc/${h}` above — only the audience differs, and
+                # deliberately does not reuse `hiveClientPrefix`:
+                # `swarm-controller` is explicitly not a hive (see its own
+                # module's `queueClientId` doc comment), so its audience is
+                # its own client id rather than a value from the per-hive
+                # namespace. Read via the option rather than a literal so a
+                # rename of `queueClientId`'s default cannot silently
+                # desync the two ends — this authenticator and the client
+                # requesting the audience it checks both have to agree, or
+                # a real swarm-controller push gets a healthy-looking 401.
+                // {
+                  "oidc/${swarmTierName}" = {
+                    issuer_url = autheliaCfg.url;
+                    audience = config.services.hyperhive.swarm.controller.queueClientId;
+                  };
+                };
 
               # `upsert`, not `insert`: a sender that stamps its own
               # `hive` must be OVERWRITTEN, not deferred to. This
@@ -1151,12 +1258,14 @@ in
                 # an invented one (a sentinel, the local hive's name) would be
                 # queried as though it meant something.
                 #
-                # ⚠️ Emitted for the LOG pipeline too, not just the scraped one:
+                # ⚠️ Unconditional, not gated on `scrapeTargets != {} ||
-                # a processor a pipeline names but the config does not define is
+                # collectLogs` (as it once was): a processor a pipeline names
-                # a collector that refuses to start, and enabling the log store
+                # but the config does not define is a collector that refuses
-                # without declaring a scrape target is a perfectly ordinary
+                # to start, and `metrics/${swarmTierName}` now ALWAYS exists
-                # config.
+                # (its `otlp/${swarmTierName}` receiver is unconditional too,
-                // lib.optionalAttrs (cfg.scrapeTargets != { } || collectLogs) {
+                # for a swarm-tier producer to push to) — so this processor
+                # has to exist unconditionally right alongside it.
+                // {
                   "resource/${swarmTierName}".attributes = [
                     {
                       # The metric LABEL, a different namespace from the

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,21 @@ 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_http_client.rs`'s `AuthenticatedHttpClient` — an
+# `opentelemetry_http::HttpClient` impl authenticated with this crate's own
+# `swarm-queue-client` identity. Lives in this crate rather than
+# `swarm-queue-client` (mara's own call during review) since this daemon is
+# its only caller; see that module's doc for the full rationale. `opentelemetry-http`
+# is also needed directly (not just transitively) 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`.
+opentelemetry-http.workspace = true
+async-trait.workspace = true
+bytes.workspace = true
+http.workspace = true
 tokio.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true

swarm-controller/src/auth.rs

@@ -98,7 +98,7 @@ impl AuthBridge {
         // configured CA, if any) — deliberately not `self.http`, which is
         // the bridge's own client and has nothing to do with authelia's
         // token endpoint's trust anchors.
-        let token = swarm_queue_client::mint_token_for(&self.queue_cfg)
+        let token = swarm_queue_client::mint_token_for(&self.queue_cfg, None)
             .await
             .context("minting a bearer token for swarm-authelia-bridge")?;
 

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;
@@ -1157,8 +1158,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/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

@@ -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
+    // `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)
+        .with_http_client(authenticated_http_client()?)
         .build()
         .context("build OTLP metric exporter")?;
     let reader = PeriodicReader::builder(exporter)
@@ -128,6 +135,53 @@ fn build_provider(interval: Duration) -> Result<SdkMeterProvider> {
         .build())
 }
 
+/// 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
+/// 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<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(|| {
+            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(crate::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.

swarm-queue-client/Cargo.toml

@@ -35,7 +35,16 @@ notices = ["async-nats/jetstream"]
 # Bare (no `kv`/`jetstream`) unless a consumer opts into the `kv` feature
 # above - the connect itself needs none of them.
 async-nats.workspace = true
-reqwest.workspace = true
+# `blocking` on top of the workspace default (`form`/`json`/`rustls`) —
+# `mint_token_for_blocking` needs `reqwest::blocking::Client` for a caller
+# with no tokio reactor to `.await` an async request on (an OTLP exporter's
+# `HttpClient` impl, see that function's doc). Declared here rather than
+# left to arrive transitively from a consumer that happens to pull in
+# `reqwest`'s blocking feature some other way — this crate already has a
+# recorded case of exactly that kind of accidental compile (see the `kv`
+# 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"] }
 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

@@ -316,7 +316,11 @@ impl QueueConfig {
 /// authenticates as itself. Authelia refuses the `openid` scope for this grant
 /// (a machine client receives an access token and never an id-token), so no
 /// scope is requested.
-async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedToken, Error> {
+async fn mint_token(
+    http: &reqwest::Client,
+    cfg: &QueueConfig,
+    audience: Option<&str>,
+) -> Result<CachedToken, Error> {
     // Read per call rather than caching: the file is small, and a cached
     // secret would survive a rotation that the operator believes took effect.
     let secret = tokio::fs::read_to_string(&cfg.client_secret_file)
@@ -326,19 +330,35 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedT
             source,
         })?;
 
-    let response = token_request(http, cfg, secret.trim())
+    let response = token_request(http, cfg, secret.trim(), audience)
         .send()
         .await
         .map_err(Error::TokenRequest)?;
 
-    // Why the body and not just the status: see `Error::TokenRefused`.
     let status = response.status();
     let body = response.text().await.unwrap_or_default();
+    parse_token_response(status, &body)
+}
+
+/// Turn a raw token-endpoint response into a [`CachedToken`] — the half of
+/// [`mint_token`] that is pure and therefore shared with the blocking sibling
+/// below: neither the success/failure check nor the JSON shape cares which
+/// HTTP client fetched the bytes.
+///
+/// Takes the body already read out, not the response itself: `reqwest`'s
+/// blocking and async response types have no common trait for that, and
+/// forcing one here would mean this function is not actually callable from
+/// both callers, which defeats the reason it exists.
+fn parse_token_response(status: reqwest::StatusCode, body: &str) -> Result<CachedToken, Error> {
+    // Why the body and not just the status: see `Error::TokenRefused`.
     if !status.is_success() {
-        return Err(Error::TokenRefused { status, body });
+        return Err(Error::TokenRefused {
+            status,
+            body: body.to_owned(),
+        });
     }
 
-    let parsed: TokenResponse = serde_json::from_str(&body).map_err(Error::TokenResponse)?;
+    let parsed: TokenResponse = serde_json::from_str(body).map_err(Error::TokenResponse)?;
     Ok(CachedToken {
         token: parsed.access_token,
         expires_at: std::time::Instant::now() + std::time::Duration::from_secs(parsed.expires_in),
@@ -364,14 +384,45 @@ async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result<CachedT
 ///
 /// Split out of [`mint_token`] so the request's *shape* is testable without a
 /// running identity provider — see the tests at the bottom of this file.
+///
+/// `audience` is RFC 8707 resource-indicator territory, and it is opt-in:
+/// omitted, this identity gets whatever audience authelia defaults a
+/// scopeless `client_credentials` grant to (the queue connection's own
+/// case — it has always worked without asking). A caller proving this
+/// identity to a SPECIFIC audience-checked receiver — the swarm-otel
+/// `oidc/swarm` authenticator being the first one — has to ask for it by
+/// name, the same way `swarm-otel.nix`'s own prometheus scrape config
+/// already does per target (`endpoint_params.audience`): a token minted
+/// without asking carries `aud: []`, and an audience-checked receiver
+/// refuses that just as readily as the wrong one.
 fn token_request(
     http: &reqwest::Client,
     cfg: &QueueConfig,
     secret: &str,
+    audience: Option<&str>,
 ) -> reqwest::RequestBuilder {
+    let mut form = vec![("grant_type", "client_credentials")];
+    if let Some(audience) = audience {
+        form.push(("audience", audience));
+    }
     http.post(&cfg.token_endpoint)
         .basic_auth(&cfg.client_id, Some(secret))
-        .form(&[("grant_type", "client_credentials")])
+        .form(&form)
+}
+
+/// Read + parse `path` as a PEM root certificate. Shared by the async and
+/// blocking HTTP client builders below — `reqwest::Certificate::from_pem`
+/// takes the same bytes regardless of which client type ends up trusting
+/// it, so the CA loading has no reason to exist twice.
+fn load_ca_cert(path: &std::path::Path) -> Result<reqwest::Certificate, Error> {
+    let pem = std::fs::read(path).map_err(|source| Error::CaFile {
+        path: path.display().to_string(),
+        source,
+    })?;
+    reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse {
+        path: path.display().to_string(),
+        source,
+    })
 }
 
 /// Build the HTTP client used to reach `cfg.token_endpoint`, trusting
@@ -382,15 +433,7 @@ fn token_request(
 fn build_http_client(cfg: &QueueConfig) -> Result<reqwest::Client, Error> {
     let mut builder = reqwest::Client::builder().timeout(std::time::Duration::from_secs(10));
     if let Some(path) = &cfg.ca_file {
-        let pem = std::fs::read(path).map_err(|source| Error::CaFile {
+        builder = builder.add_root_certificate(load_ca_cert(path)?);
-            path: path.display().to_string(),
-            source,
-        })?;
-        let cert = reqwest::Certificate::from_pem(&pem).map_err(|source| Error::CaParse {
-            path: path.display().to_string(),
-            source,
-        })?;
-        builder = builder.add_root_certificate(cert);
     }
     builder.build().map_err(Error::HttpClient)
 }
@@ -408,9 +451,71 @@ fn build_http_client(cfg: &QueueConfig) -> Result<reqwest::Client, Error> {
 /// outside that loop (e.g. `swarm-controller::auth`'s bridge client) mints
 /// per call, same as this crate did before the reconnect-storm fix added the
 /// cache.
-pub async fn mint_token_for(cfg: &QueueConfig) -> Result<String, Error> {
+///
+/// `audience` is passed straight to the private `token_request` helper —
+/// see its doc for why it is optional and when a caller needs it. `None`
+/// reproduces this
+/// function's behaviour before the parameter existed, so every caller from
+/// before that added it (the queue connect path, `auth.rs`'s bridge client)
+/// is unaffected.
+pub async fn mint_token_for(cfg: &QueueConfig, audience: Option<&str>) -> Result<String, Error> {
     let http = build_http_client(cfg)?;
-    Ok(mint_token(&http, cfg).await?.token)
+    Ok(mint_token(&http, cfg, audience).await?.token)
+}
+
+/// Blocking sibling of [`mint_token_for`], for a caller with no tokio
+/// reactor to `.await` on.
+///
+/// That caller is real, not hypothetical: an OTLP exporter's `HttpClient`
+/// implementation runs on the `PeriodicReader`'s own thread, which this
+/// workspace deliberately keeps reactor-free (`reqwest-blocking-client` was
+/// chosen over the async client for exactly that reason — see
+/// `swarm-controller::vcs_metrics`'s module doc). Calling the async
+/// [`mint_token_for`] from there would need a runtime that thread does not
+/// have; this function uses `reqwest::blocking::Client` end to end instead,
+/// so it never needs one.
+///
+/// Same request shape and same no-caching behaviour as [`mint_token_for`] —
+/// see that function's doc for why both of those are the right call here.
+pub fn mint_token_for_blocking(cfg: &QueueConfig, audience: Option<&str>) -> Result<String, Error> {
+    let http = build_blocking_http_client(cfg)?;
+
+    // Read per call rather than caching — see `mint_token`'s identical
+    // comment on the async path; the reasoning does not change with the
+    // client type.
+    let secret =
+        std::fs::read_to_string(&cfg.client_secret_file).map_err(|source| Error::ClientSecret {
+            path: cfg.client_secret_file.display().to_string(),
+            source,
+        })?;
+
+    let mut form = vec![("grant_type", "client_credentials")];
+    if let Some(audience) = audience {
+        form.push(("audience", audience));
+    }
+    let response = http
+        .post(&cfg.token_endpoint)
+        .basic_auth(&cfg.client_id, Some(secret.trim()))
+        .form(&form)
+        .send()
+        .map_err(Error::TokenRequest)?;
+
+    let status = response.status();
+    let body = response.text().unwrap_or_default();
+    Ok(parse_token_response(status, &body)?.token)
+}
+
+/// Blocking counterpart to [`build_http_client`] — same CA-trust logic
+/// (via the shared [`load_ca_cert`]), a `reqwest::blocking::Client` instead
+/// of the async one. See [`mint_token_for_blocking`] for why a caller needs
+/// this rather than the async builder.
+fn build_blocking_http_client(cfg: &QueueConfig) -> Result<reqwest::blocking::Client, Error> {
+    let mut builder =
+        reqwest::blocking::Client::builder().timeout(std::time::Duration::from_secs(10));
+    if let Some(path) = &cfg.ca_file {
+        builder = builder.add_root_certificate(load_ca_cert(path)?);
+    }
+    builder.build().map_err(Error::HttpClient)
 }
 
 /// Fail fast unless the client is actually connected.
@@ -487,7 +592,7 @@ pub async fn connect(cfg: QueueConfig) -> Result<async_nats::Client, Error> {
             let token = if let Some(token) = reuse {
                 token
             } else {
-                let minted = mint_token(&http, &cfg)
+                let minted = mint_token(&http, &cfg, None)
                     .await
                     // The callback's error type carries a string, so the
                     // source chain would be lost; flatten it rather than
@@ -631,7 +736,7 @@ mod tests {
 
     #[test]
     fn the_token_request_authenticates_with_http_basic() {
-        let req = token_request(&offline_client(), &token_cfg(), "s3cret")
+        let req = token_request(&offline_client(), &token_cfg(), "s3cret", None)
             .build()
             .expect("the token request must build");
 
@@ -667,4 +772,49 @@ mod tests {
             "the client id belongs in the Basic credentials, got: {body}"
         );
     }
+
+    /// The queue/bridge shape (`audience: None`) must stay unchanged by the
+    /// parameter's addition — no `audience` field appears in the body at
+    /// all, not even empty.
+    #[test]
+    fn no_audience_means_no_audience_field() {
+        let req = token_request(&offline_client(), &token_cfg(), "s3cret", None)
+            .build()
+            .expect("the token request must build");
+        let body = std::str::from_utf8(
+            req.body()
+                .and_then(reqwest::Body::as_bytes)
+                .expect("the request has an in-memory body"),
+        )
+        .expect("the body is utf-8");
+        assert!(
+            !body.contains("audience"),
+            "omitting the audience must not even send an empty field, got: {body}"
+        );
+    }
+
+    /// A caller that DOES ask for an audience gets it in the form body,
+    /// verbatim — this is the half `swarm-otel`'s `oidc/<owner>`
+    /// authenticators actually check.
+    #[test]
+    fn an_audience_is_sent_verbatim() {
+        let req = token_request(
+            &offline_client(),
+            &token_cfg(),
+            "s3cret",
+            Some("https://otel.example/swarm"),
+        )
+        .build()
+        .expect("the token request must build");
+        let body = std::str::from_utf8(
+            req.body()
+                .and_then(reqwest::Body::as_bytes)
+                .expect("the request has an in-memory body"),
+        )
+        .expect("the body is utf-8");
+        assert!(
+            body.contains("audience=https%3A%2F%2Fotel.example%2Fswarm"),
+            "the requested audience must reach the form body, got: {body}"
+        );
+    }
 }