diff --git a/Cargo.lock b/Cargo.lock index 88c8eb6c..80a70420 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1794,11 +1794,9 @@ 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", @@ -4575,10 +4573,8 @@ version = "0.1.0" dependencies = [ "anyhow", "async-nats", - "async-trait", "axum", "base64", - "bytes", "forgejo-api", "futures-util", "hive-jobq", @@ -4586,9 +4582,7 @@ dependencies = [ "hive-jobq-wire", "hive-types", "hmac 0.13.0", - "http", "opentelemetry", - "opentelemetry-http", "opentelemetry-otlp", "opentelemetry_sdk", "problem_details", diff --git a/Cargo.toml b/Cargo.toml index 4733a5dc..f2c58eb9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -144,29 +144,6 @@ 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 diff --git a/hive-jobq-metrics/Cargo.toml b/hive-jobq-metrics/Cargo.toml index a280f7b6..e48094e3 100644 --- a/hive-jobq-metrics/Cargo.toml +++ b/hive-jobq-metrics/Cargo.toml @@ -9,18 +9,10 @@ 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 diff --git a/hive-jobq-metrics/src/lib.rs b/hive-jobq-metrics/src/lib.rs index 9c45d453..b927ceef 100644 --- a/hive-jobq-metrics/src/lib.rs +++ b/hive-jobq-metrics/src/lib.rs @@ -23,38 +23,13 @@ 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_http::{Bytes, HttpClient, HttpError, Request, Response}; -use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig, WithHttpConfig}; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; 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`: 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); - -#[async_trait] -impl HttpClient for BoxedHttpClient { - async fn send_bytes(&self, request: Request) -> Result, 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 @@ -78,18 +53,9 @@ 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( jobq: Arc>>, service_name: &str, - http_client: Option>, ) -> Option where N: Send + 'static, @@ -117,7 +83,7 @@ where } }); - match build_provider(interval, snapshot, service_name, http_client) { + match build_provider(interval, snapshot, service_name) { Ok(provider) => { tracing::info!(%endpoint, ?interval, %service_name, "otel jobq-metrics: exporter enabled"); Some(provider) @@ -133,22 +99,15 @@ fn build_provider( interval: Duration, snapshot: Arc>>, service_name: &str, - http_client: Option>, ) -> Result { // 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 mut builder = MetricExporter::builder() + let exporter = MetricExporter::builder() .with_http() - .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")?; + .with_protocol(Protocol::HttpJson) + .build() + .context("build OTLP metric exporter")?; let reader = PeriodicReader::builder(exporter) .with_interval(interval) .build(); diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index cd100f3e..5ba54251 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -149,47 +149,6 @@ 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 @@ -517,17 +476,6 @@ 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 ]; } ]; @@ -728,8 +676,7 @@ in // forgeEnv // webhookEnv // authBridgeEnv - // swarmNameEnv - // otelEnv; + // swarmNameEnv; }; # A systemd credential is a SNAPSHOT: it is materialised into `%d` once, diff --git a/nix/host-modules/swarm-otel.nix b/nix/host-modules/swarm-otel.nix index b39e66bd..9d5ebd00 100644 --- a/nix/host-modules/swarm-otel.nix +++ b/nix/host-modules/swarm-otel.nix @@ -54,11 +54,7 @@ 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. - # 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; + swarmTierName = "swarm"; # Every `` 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 @@ -67,8 +63,7 @@ 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 (or a `producerName` override the file was never updated for) - # would be an eval error rather than a silently missing guard. + # the file 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 @@ -254,57 +249,6 @@ 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}"; @@ -479,15 +423,9 @@ 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 other route into this collector - # belongs to exactly one hive or the swarm-tier producer above. + # description of that. Every route into this collector belongs to + # exactly one hive. "/".return = "404"; }; }; @@ -777,7 +715,6 @@ in derived = lib.attrValues hivePorts; others = [ cfg.telemetryPort - cfg.producerPort otelCfg.collector.port ] ++ lib.optional vmCfg.enable vmCfg.port; @@ -909,17 +846,6 @@ 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 @@ -1094,13 +1020,7 @@ 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) - # 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}" ]; + service.extensions = map (h: "oidc/${h}") (lib.attrNames hivePorts); # Fan-out, not a choice: with both configured the same # samples go upstream AND into the swarm's store. The store @@ -1124,17 +1044,9 @@ 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. - // { - # `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. + // lib.optionalAttrs (cfg.scrapeTargets != { }) { "metrics/${swarmTierName}" = { - receivers = [ - "otlp/${swarmTierName}" - ] - ++ lib.optional (cfg.scrapeTargets != { }) "prometheus"; + receivers = [ "prometheus" ]; processors = [ "resource/${swarmTierName}" ]; exporters = exporterNames; }; @@ -1157,64 +1069,45 @@ in }; } // { - extensions = - lib.mapAttrs' ( - h: _: - lib.nameValuePair "oidc/${h}" { - issuer_url = autheliaCfg.url; - # The audience this hive's client is registered to - # request, and the reason one hive's token is refused by - # another hive's receiver. Same expression authelia - # registers it under — a second spelling here would deny - # every hive, as a 401 that blames the token. - audience = "${autheliaCfg.hiveClientPrefix}${h}"; - # ⛔ DO NOT ADD `issuer_ca_path` HERE. It took this - # collector down for forty minutes once, and the failure - # is invisible to every check we have. - # - # It loads only the FIRST certificate in the file it - # names. The bundle assembled for this container is - # `system CAs ++ hive trust bundle`, so the anchor sits - # ~123rd and is never in the pool: the extension then - # cannot verify authelia and the whole collector exits - # `x509: certificate signed by unknown authority`, on - # every start, with 125 valid certificates in the file. - # - # Leaving it unset makes the extension use the process - # trust store, which `trustBundle` already populates via - # `SSL_CERT_FILE` — and *that* consumer reads every - # certificate regardless of order. One file, two - # consumers, opposite parsing: the fix is to stop naming - # it twice, not to reorder the bundle. - # - # ⚠️ Nor is pointing it at the hive trust bundle a fix: - # `hive-tls.nix` writes that leading with the *hive* CA - # (`nameConstraints` = this hive's domain), which cannot - # issue a swarm-level name at all. - # - # (Kept from the original note, still true and still - # worth not re-deriving: `issuer_ca_file`, `ca_file` and - # `tls.ca_file` are INVALID KEYS for this extension.) - } - ) 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; - }; - }; + extensions = lib.mapAttrs' ( + h: _: + lib.nameValuePair "oidc/${h}" { + issuer_url = autheliaCfg.url; + # The audience this hive's client is registered to + # request, and the reason one hive's token is refused by + # another hive's receiver. Same expression authelia + # registers it under — a second spelling here would deny + # every hive, as a 401 that blames the token. + audience = "${autheliaCfg.hiveClientPrefix}${h}"; + # ⛔ DO NOT ADD `issuer_ca_path` HERE. It took this + # collector down for forty minutes once, and the failure + # is invisible to every check we have. + # + # It loads only the FIRST certificate in the file it + # names. The bundle assembled for this container is + # `system CAs ++ hive trust bundle`, so the anchor sits + # ~123rd and is never in the pool: the extension then + # cannot verify authelia and the whole collector exits + # `x509: certificate signed by unknown authority`, on + # every start, with 125 valid certificates in the file. + # + # Leaving it unset makes the extension use the process + # trust store, which `trustBundle` already populates via + # `SSL_CERT_FILE` — and *that* consumer reads every + # certificate regardless of order. One file, two + # consumers, opposite parsing: the fix is to stop naming + # it twice, not to reorder the bundle. + # + # ⚠️ Nor is pointing it at the hive trust bundle a fix: + # `hive-tls.nix` writes that leading with the *hive* CA + # (`nameConstraints` = this hive's domain), which cannot + # issue a swarm-level name at all. + # + # (Kept from the original note, still true and still + # worth not re-deriving: `issuer_ca_file`, `ca_file` and + # `tls.ca_file` are INVALID KEYS for this extension.) + } + ) hivePorts; # `upsert`, not `insert`: a sender that stamps its own # `hive` must be OVERWRITTEN, not deferred to. This @@ -1258,14 +1151,12 @@ in # an invented one (a sentinel, the local hive's name) would be # queried as though it meant something. # - # ⚠️ Unconditional, not gated on `scrapeTargets != {} || - # collectLogs` (as it once was): a processor a pipeline names - # but the config does not define is a collector that refuses - # to start, and `metrics/${swarmTierName}` now ALWAYS exists - # (its `otlp/${swarmTierName}` receiver is unconditional too, - # for a swarm-tier producer to push to) — so this processor - # has to exist unconditionally right alongside it. - // { + # ⚠️ Emitted for the LOG pipeline too, not just the scraped one: + # a processor a pipeline names but the config does not define is + # a collector that refuses to start, and enabling the log store + # without declaring a scrape target is a perfectly ordinary + # config. + // lib.optionalAttrs (cfg.scrapeTargets != { } || collectLogs) { "resource/${swarmTierName}".attributes = [ { # The metric LABEL, a different namespace from the diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 32d2ce10..1bf1a16d 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -65,13 +65,7 @@ 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. -# -# `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"] } +reqwest.workspace = true serde.workspace = true serde_json.workspace = true swarm-authelia-bridge-sock.workspace = true @@ -83,21 +77,7 @@ 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` 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 diff --git a/swarm-controller/src/auth.rs b/swarm-controller/src/auth.rs index e53634a0..f08cbb7a 100644 --- a/swarm-controller/src/auth.rs +++ b/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, None) + let token = swarm_queue_client::mint_token_for(&self.queue_cfg) .await .context("minting a bearer token for swarm-authelia-bridge")?; diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index ea09dfc9..bad9edfe 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -43,7 +43,6 @@ use utoipa_axum::{router::OpenApiRouter, routes}; mod auth; mod config_pr; mod forge; -mod otel_http_client; mod status; mod vcs_metrics; mod webhook; @@ -1158,21 +1157,8 @@ 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); let _jobq_metrics_provider = - hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller", jobq_http_client); + hive_jobq_metrics::spawn_exporter(Arc::clone(&jobq), "swarm-controller"); // Same "log and carry on" shape as the queue/bridge/forge wiring above. // A controller that cannot hold a webhook secret still serves every diff --git a/swarm-controller/src/otel_http_client.rs b/swarm-controller/src/otel_http_client.rs deleted file mode 100644 index 3429cba8..00000000 --- a/swarm-controller/src/otel_http_client.rs +++ /dev/null @@ -1,100 +0,0 @@ -//! 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, - ) -> 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) -> Result, 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) - } -} diff --git a/swarm-controller/src/vcs_metrics.rs b/swarm-controller/src/vcs_metrics.rs index 10d51723..c4265e81 100644 --- a/swarm-controller/src/vcs_metrics.rs +++ b/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, WithHttpConfig}; +use opentelemetry_otlp::{MetricExporter, Protocol, WithExportConfig}; use opentelemetry_sdk::Resource; use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider}; @@ -114,16 +114,9 @@ fn build_provider(interval: Duration) -> Result { // 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) @@ -135,53 +128,6 @@ fn build_provider(interval: Duration) -> Result { .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 -{ - 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. diff --git a/swarm-queue-client/Cargo.toml b/swarm-queue-client/Cargo.toml index 3798b01d..b0b63588 100644 --- a/swarm-queue-client/Cargo.toml +++ b/swarm-queue-client/Cargo.toml @@ -35,16 +35,7 @@ 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 -# `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"] } +reqwest.workspace = true serde.workspace = true serde_json.workspace = true # A library, so its errors are a matchable enum rather than an opaque diff --git a/swarm-queue-client/src/lib.rs b/swarm-queue-client/src/lib.rs index c7a9143f..5e5d956e 100644 --- a/swarm-queue-client/src/lib.rs +++ b/swarm-queue-client/src/lib.rs @@ -316,11 +316,7 @@ 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, - audience: Option<&str>, -) -> Result { +async fn mint_token(http: &reqwest::Client, cfg: &QueueConfig) -> Result { // 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) @@ -330,35 +326,19 @@ async fn mint_token( source, })?; - let response = token_request(http, cfg, secret.trim(), audience) + let response = token_request(http, cfg, secret.trim()) .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 { - // Why the body and not just the status: see `Error::TokenRefused`. if !status.is_success() { - return Err(Error::TokenRefused { - status, - body: body.to_owned(), - }); + return Err(Error::TokenRefused { status, body }); } - 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), @@ -384,45 +364,14 @@ fn parse_token_response(status: reqwest::StatusCode, body: &str) -> Result, ) -> 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(&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 { - 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, - }) + .form(&[("grant_type", "client_credentials")]) } /// Build the HTTP client used to reach `cfg.token_endpoint`, trusting @@ -433,7 +382,15 @@ fn load_ca_cert(path: &std::path::Path) -> Result { fn build_http_client(cfg: &QueueConfig) -> Result { let mut builder = reqwest::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)?); + let pem = std::fs::read(path).map_err(|source| Error::CaFile { + 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) } @@ -451,71 +408,9 @@ fn build_http_client(cfg: &QueueConfig) -> Result { /// 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. -/// -/// `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 { +pub async fn mint_token_for(cfg: &QueueConfig) -> Result { let http = build_http_client(cfg)?; - 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 { - 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 { - 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) + Ok(mint_token(&http, cfg).await?.token) } /// Fail fast unless the client is actually connected. @@ -592,7 +487,7 @@ pub async fn connect(cfg: QueueConfig) -> Result { let token = if let Some(token) = reuse { token } else { - let minted = mint_token(&http, &cfg, None) + let minted = mint_token(&http, &cfg) .await // The callback's error type carries a string, so the // source chain would be lost; flatten it rather than @@ -736,7 +631,7 @@ mod tests { #[test] fn the_token_request_authenticates_with_http_basic() { - let req = token_request(&offline_client(), &token_cfg(), "s3cret", None) + let req = token_request(&offline_client(), &token_cfg(), "s3cret") .build() .expect("the token request must build"); @@ -772,49 +667,4 @@ 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/` - /// 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}" - ); - } }