From 8891b469433d54d0d57e5941807a9e3dc62328e4 Mon Sep 17 00:00:00 2001 From: atlas Date: Sat, 15 Aug 2026 18:37:23 +0200 Subject: [PATCH] feat(swarm-controller): aggregate per-hive status from the swarm queue The controller connects to the swarm queue as its own client and serves what each hive last said about itself at GET /api/hives/status. THE QUEUE IS THE STORE. A hive publishes into the `hive-status` JetStream KV bucket (history 1) and the controller reads it per request, keeping no copy. A cache here would be a second answer to the same question, free to disagree with the first, and the disagreement surfaces as a hive reading healthy on a dashboard while the bucket says otherwise. Whichever side arrives first creates the bucket; both want the same shape. Rows come from the roster rather than from the bucket, so an empty bucket renders as a swarm nobody has heard from instead of a healthy one, and `never_reported` stays distinct from `stale` - went quiet is a fault, never spoke is usually a deployment that has not happened. Freshness is derived at read time and never stored as a flag, because a stored `healthy` boolean goes stale silently the moment nothing arrives, which is the failure this endpoint is designed against. The timestamp is the NATS server's, applied when the value landed, so a publisher cannot make itself look fresher than it is. Authentication is per connection attempt, not per process. Authelia issues `client_credentials` tokens that expire in 3599s, and auth happens at CONNECT, so a long-lived connection is fine but a reconnect an hour later needs a token minted an hour later. `with_auth_callback` is re-run by async-nats for each attempt, which handles expiry by construction rather than by a timer - the alternative fails in the way this subsystem exists to prevent, with the controller still serving while its data quietly stops updating. Three failure shapes are deliberate: - A half-set environment is fatal; an absent one is not. Silently behaving like an unconfigured host is how every hive ends up reading `never_reported` with nothing to point at. - The endpoint answers 503 rather than an empty list when the store cannot be read. "I cannot reach the store" and "every hive is silent" are different answers, and rendering the second turns a local fault into an apparent swarm-wide outage. - `retry_on_initial_connect` makes the daemon and the queue bootable in either order, and the status handler refuses when the client is not Connected rather than issuing a request into it - a request made in that window does not fail, it waits, so every poll would hang and learn nothing. `Pending` is the state a never-connected client is in, which is why the test is `!= Connected` and not `== Disconnected`. The rendering rules are a pure function over a map, so the semantics are tested against a table rather than against a running server. The KV read, the credential rotation and the 503 paths are covered behaviourally instead: a real NATS server with a rotating token endpoint, asserting that the controller recovers only when the credential rotates, and mutation-tested by holding the credential wrong for the same window. --- Cargo.lock | 25 ++ docs/swarm/README.md | 59 +++ nix/host-modules/swarm-controller.nix | 124 +++++- swarm-controller/Cargo.toml | 8 + swarm-controller/src/main.rs | 91 +++++ swarm-controller/src/queue.rs | 197 ++++++++++ swarm-controller/src/status.rs | 543 ++++++++++++++++++++++++++ 7 files changed, 1030 insertions(+), 17 deletions(-) create mode 100644 swarm-controller/src/queue.rs create mode 100644 swarm-controller/src/status.rs diff --git a/Cargo.lock b/Cargo.lock index b6b1966c..76f2dfa4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -213,14 +213,17 @@ dependencies = [ "rustls-webpki", "serde", "serde_json", + "serde_nanos", "serde_repr", "thiserror 2.0.18", + "time", "tokio", "tokio-rustls", "tokio-stream", "tokio-util", "tokio-websockets", "tracing", + "tryhard", "url", ] @@ -4310,6 +4313,15 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_nanos" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a93142f0367a4cc53ae0fead1bcda39e85beccfad3dcd717656cacab94b12985" +dependencies = [ + "serde", +] + [[package]] name = "serde_path_to_error" version = "0.1.20" @@ -4546,7 +4558,10 @@ name = "swarm-controller" version = "0.1.0" dependencies = [ "anyhow", + "async-nats", "axum", + "futures-util", + "reqwest 0.13.1", "serde", "serde_json", "tokio", @@ -5070,6 +5085,16 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tryhard" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fe58ebd5edd976e0fe0f8a14d2a04b7c81ef153ea9a54eebc42e67c2c23b4e5" +dependencies = [ + "pin-project-lite", + "tokio", +] + [[package]] name = "tungstenite" version = "0.29.0" diff --git a/docs/swarm/README.md b/docs/swarm/README.md index 95776d2d..89e9b1e4 100644 --- a/docs/swarm/README.md +++ b/docs/swarm/README.md @@ -330,6 +330,65 @@ What it serves, why it is a unix socket rather than a port, and the socket-directory constraint that governs where `socketPath` may point: [`swarm-controller/README.md`](../../swarm-controller/README.md). +### Per-hive status (`GET /api/hives/status`) + +What each hive last **offered** about itself. Hives publish upward; the +controller never reaches down to collect. That direction is deliberate: +during the #3097 gateway outage every recovery channel ran through the +one broken thing, so a status path that depended on the controller would +have gone dark exactly when it was needed to diagnose the controller's +own network. A hive computes its own status locally either way — this +endpoint is a view of what was published, never the source. + +**The queue is the store.** A hive publishes into the `hive-status` +JetStream KV bucket (`history: 1` — the last thing each hive said), and +the controller reads that bucket per request, keeping no copy. A cache +here would be a second answer free to disagree with the first, and the +disagreement would surface as a hive reading healthy on a dashboard +while the bucket says otherwise. The bucket is created by whichever side +gets there first. + +**Absence is what the endpoint is built around**: + +| freshness | means | +|---|---| +| `fresh` | published within `staleAfterSeconds` | +| `stale` | published longer ago than that — the payload is still returned, because "old" and "absent" are different answers | +| `never_reported` | in the roster, has never published. Distinct from `stale`: went quiet is a fault, never spoke is usually a deployment that hasn't happened | +| `unknown` | published but not in `swarm.hives` — surfaced rather than dropped | + +Rows come from the **roster**, not from the bucket, so a hive that has +never reported appears rather than not appearing, and an empty bucket +renders as a swarm nobody has heard from instead of a healthy one. +Freshness is derived at read time from a timestamp and never stored as a +flag — a stored `healthy` boolean goes stale silently the moment nothing +arrives, which is the failure this is designed against. Each row also +carries `last_seen_unix` and `age_seconds`, so a consumer that disagrees +with `staleAfterSeconds` can apply its own threshold. + +`last_seen_unix` is the **bucket's** timestamp, applied by the NATS +server when the value landed, not a field inside the payload — a +publisher cannot make itself look fresher than it is, and a hive with a +wrong clock skews its own payload rather than its freshness. + +Because the bucket outlives a controller restart, a restarted controller +reports what it reads: `stale, age_seconds: 10800` rather than +`never_reported`. That is the more honest of the two — it genuinely +knows when the hive last spoke. Losing the bucket degrades in the same +direction: every hive reads `never_reported` until its next publish, +which is the true answer and not a remembered "healthy". + +The endpoint answers **503**, not an empty list, when no queue is +configured on this host or its store cannot be read. "I cannot reach the +store" and "every hive is silent" are different answers, and rendering +the second when the first is true would turn a local fault into an +apparent swarm-wide outage. + +⚠️ **Nothing writes to the bucket yet.** The transport and the +controller's read path are in place; the hive-side publisher is a later +slice. Until one lands, every hive reads `never_reported` — the correct +answer for a controller that has been told nothing. + ## Cross-references - `docs/snapshot-store.md` — the swarm's `btrfs receive` endpoint, and diff --git a/nix/host-modules/swarm-controller.nix b/nix/host-modules/swarm-controller.nix index 30f7ca86..b5ba75ee 100644 --- a/nix/host-modules/swarm-controller.nix +++ b/nix/host-modules/swarm-controller.nix @@ -33,6 +33,39 @@ let SWARMCTL_AUTHELIA_UNIT = autheliaCfg.unit; }; + natsCfg = config.services.hyperhive.swarm.nats; + + # The controller's own OAuth2 client. It is NOT a hive: the per-hive + # clients the roster issues belong to hives, and the responder's client + # belongs to the responder. One identity per principal — the rule is that + # a principal's credentials all derive from the same identity, not that + # the swarm has one. + queueClientId = "swarm-controller"; + + # Both halves have to be here: authelia to have minted the secret, and the + # queue to connect to. Same guard, and the same reasoning, as `autheliaEnv` + # above — a value set on a host that runs neither would point at a file + # that does not exist and produce a daemon that retries forever. + queueLocal = autheliaCfg.enable && natsCfg.enable; + + # `LoadCredential` and not a copy-oneshot, which is where this deliberately + # differs from the callout responder: that one delivers INTO a container, + # so it has to copy across a filesystem boundary. The controller is a plain + # host unit, so systemd can hand it the file directly — fewer moving parts, + # and the secret never gains a second on-disk copy to forget about. + queueEnv = lib.optionalAttrs queueLocal { + # The queue container shares the host netns, so loopback is correct here + # and is not the `localhost`-means-the-wrong-thing trap that applies + # inside agent containers. + SWARM_CONTROLLER_NATS_URL = "nats://127.0.0.1:${toString natsCfg.port}"; + SWARM_CONTROLLER_OIDC_TOKEN_ENDPOINT = "${autheliaCfg.url}/api/oidc/token"; + SWARM_CONTROLLER_OIDC_CLIENT_ID = queueClientId; + # `%d` is systemd's credentials directory: root reads the plaintext at + # unit start and the daemon's own user sees it 0400, without the unit + # ever being able to read the rest of authelia's state dir. + SWARM_CONTROLLER_OIDC_CLIENT_SECRET_FILE = "%d/queue-client.secret"; + }; + # 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 @@ -113,6 +146,25 @@ in ''; }; + staleAfterSeconds = lib.mkOption { + type = lib.types.ints.positive; + default = 120; + description = '' + How old a hive's last status snapshot may be before + `GET /api/hives/status` reports it as `stale` rather than `fresh`. + + This is a statement about how often hives *publish*, not about how + patient a reader is — set it above the publishing cadence or every + hive reads stale between offers. It is an option and not a + constant precisely because that cadence is a property of the + deployment. + + Freshness is derived when the endpoint is read, never stored, so + changing this takes effect for the next request; no hive has to + re-publish anything. + ''; + }; + links = lib.mkOption { type = lib.types.listOf ( lib.types.submodule { @@ -174,6 +226,23 @@ in # exactly what it must not inherit. environment.systemPackages = [ swarmctlConfigured ]; + # One declaration, two readers — the controller knows which client id it + # authenticates under, so making the operator restate it in authelia's + # client list would be a second source of truth for a string whose + # mismatch is an opaque 401 from the token endpoint. Same shape as the + # queue's own client declaration. + services.hyperhive.swarm.authelia.oidc.clients = lib.mkIf autheliaCfg.enable [ + { + id = queueClientId; + description = "HyperHive swarm controller"; + # `client_credentials`: a daemon authenticating as itself, with + # nobody to redirect. Declared rather than inferred from an empty + # redirect list, because authelia permits only the grants a client + # names and an omitted `grant_types` means authorization-code alone. + kind = "machine"; + } + ]; + systemd.services.swarm-controller = { description = "hyperhive swarm-level controller daemon"; wantedBy = [ "multi-user.target" ]; @@ -181,6 +250,14 @@ in serviceConfig = { ExecStart = "${cfg.package}/bin/swarm-controller"; + + # Only when the queue is actually reachable from here. An absent + # credential is not a failure: the daemon logs that no queue is + # configured and serves its HTTP surface, which is the correct + # behaviour on the hosts that do not run one. + LoadCredential = lib.mkIf queueLocal [ + "queue-client.secret:${autheliaCfg.hostClientSecretDir}/${queueClientId}.secret" + ]; User = "swarm-controller"; Group = "swarm-controller"; Restart = "on-failure"; @@ -218,23 +295,36 @@ in ]; }; - environment.SWARM_CONTROLLER_SOCKET = cfg.socketPath; - # The swarm's hive directory, JSON-encoded — same shape hive-c0re - # already builds for HYPERHIVE_PEERS (../hive-c0re/environment.nix), - # just the full directory (this daemon has no "self" hive to - # exclude, unlike a per-hive c0re's peer list) rather than - # peers-minus-self. Consumed by `GET /api/hives` - # (swarm-controller/src/main.rs::load_hives). - environment.SWARM_CONTROLLER_HIVES = builtins.toJSON ( - lib.mapAttrsToList (name: h: { - inherit name; - inherit (h) domain; - }) config.services.hyperhive.swarm.hives - ); - # The merged links list — see `links`' description above for who - # contributes to it. Consumed by `GET /api/links` - # (swarm-controller/src/main.rs::load_links). - environment.SWARM_CONTROLLER_LINKS = builtins.toJSON cfg.links; + # Queue coordinates (`queueEnv`) merge in last and are present only + # where the queue and its IdP both run. The daemon refuses a PARTIAL + # set rather than treating it as absent, which is why they are built + # as one attrset and never assigned individually. + environment = { + SWARM_CONTROLLER_SOCKET = cfg.socketPath; + # The swarm's hive directory, JSON-encoded — same shape hive-c0re + # already builds for HYPERHIVE_PEERS (../hive-c0re/environment.nix), + # just the full directory (this daemon has no "self" hive to + # exclude, unlike a per-hive c0re's peer list) rather than + # peers-minus-self. Consumed by `GET /api/hives` + # (swarm-controller/src/main.rs::load_hives). + SWARM_CONTROLLER_HIVES = builtins.toJSON ( + lib.mapAttrsToList (name: h: { + inherit name; + inherit (h) domain; + }) config.services.hyperhive.swarm.hives + ); + # The merged links list — see `links`' description above for who + # contributes to it. Consumed by `GET /api/links` + # (swarm-controller/src/main.rs::load_links). + SWARM_CONTROLLER_LINKS = builtins.toJSON cfg.links; + # Staleness threshold for `GET /api/hives/status` — see + # `staleAfterSeconds`' description. Set unconditionally rather + # than inside `queueEnv`: it is not a queue coordinate, and + # nothing about it is unsafe to define on a host whose queue is + # off (the daemon just has nothing to apply it to). + SWARM_CONTROLLER_STALE_AFTER_SECS = toString cfg.staleAfterSeconds; + } + // queueEnv; }; }; } diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index 0c3b628f..507858d3 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -10,7 +10,15 @@ path = "src/main.rs" [dependencies] anyhow.workspace = true +# `kv` (which pulls `jetstream`) on top of the workspace's feature set: the +# queue is this daemon's *store*, not just its transport - a hive's last +# status snapshot is read out of a JetStream KV bucket. Declared here rather +# than in the workspace entry so the auth-callout responder, which speaks +# neither, does not claim to need them. +async-nats = { workspace = true, features = ["kv"] } axum.workspace = true +futures-util.workspace = true +reqwest.workspace = true serde.workspace = true serde_json.workspace = true tokio.workspace = true diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index 49ab4141..71459cbd 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -31,6 +31,9 @@ use serde::{Deserialize, Serialize}; use utoipa::{OpenApi, ToSchema}; use utoipa_axum::{router::OpenApiRouter, routes}; +mod queue; +mod status; + /// Where the daemon binds, overridable via `SWARM_CONTROLLER_SOCKET`. /// /// A compiled-in default is legitimate here and is *not* the mistake that @@ -101,6 +104,11 @@ struct AppState { /// Loaded once at startup (`load_links`); same synchronization story /// as `hives`. links: Arc>, + /// `None` when this deployment wired up no swarm queue — the only + /// state in which `/api/hives/status` cannot answer at all. A queue + /// that is merely *unreachable* still yields a reader, because + /// `async-nats` reconnects underneath it. + status: Option>, } /// Env var the controller's NixOS module sets from @@ -192,6 +200,59 @@ async fn get_links(State(state): State) -> Json> { Json((*state.links).clone()) } +/// Why the status route answers 503 rather than an empty list. +/// +/// "I cannot reach the store" and "every hive is silent" are different +/// answers, and rendering the second when the first is true is exactly +/// the smoothing this endpoint exists to avoid — a caller would draw a +/// swarm-wide outage out of a local one. The cause is carried in the +/// body because a bare 503 on an operator-facing diagnostic is how a +/// misconfiguration costs an afternoon; it is a queue/JetStream error +/// string, and this surface is already behind the swarm's SSO. +struct StatusUnavailable(String); + +impl axum::response::IntoResponse for StatusUnavailable { + fn into_response(self) -> axum::response::Response { + (axum::http::StatusCode::SERVICE_UNAVAILABLE, self.0).into_response() + } +} + +/// What each hive last said about itself, read from the swarm queue at +/// request time. +/// +/// Every hive in the roster gets a row whether or not it has ever +/// reported — see the `status` module for why absence, not presence, is +/// the case this is built around. +#[utoipa::path( + get, + path = "/api/hives/status", + responses( + (status = 200, description = "a row per hive, freshness derived now", body = Vec), + (status = 503, description = "no swarm queue is configured here, or its store could not be read", body = String), + ), + tag = "hives" +)] +async fn get_hives_status( + State(state): State, +) -> Result>, StatusUnavailable> { + let Some(reader) = state.status.as_ref() else { + return Err(StatusUnavailable( + "no swarm queue is configured on this host".to_owned(), + )); + }; + match reader + .view(&state.hives, std::time::SystemTime::now()) + .await + { + Ok(rows) => Ok(Json(rows)), + Err(e) => { + let detail = format!("{e:#}"); + tracing::warn!(error = %detail, "reading the swarm status bucket failed"); + Err(StatusUnavailable(detail)) + } + } +} + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -230,14 +291,44 @@ async fn main() -> Result<()> { .with_context(|| format!("chmod {}", path.display()))?; tracing::info!(socket = %path.display(), "swarm-controller listening"); + // Connect to the swarm queue when this deployment wired one up. + // + // Deliberately NOT fatal on failure: the controller's HTTP surface is + // useful without the queue, and a hive that cannot be read from renders + // as `unknown` rather than as an outage of this daemon. What IS fatal is + // a half-set environment — `QueueConfig::from_env` refuses that, because + // silently behaving like an unconfigured host is how every hive ends up + // reading `never_reported` with nothing to point at. + let status = match queue::QueueConfig::from_env()? { + None => { + tracing::info!("no swarm queue configured; status aggregation is off"); + None + } + Some(cfg) => match queue::connect(cfg).await { + Ok(client) => { + tracing::info!("connected to the swarm queue"); + Some(Arc::new(status::StatusReader::new( + client, + status::StatusReader::stale_after_from_env(), + ))) + } + Err(e) => { + tracing::warn!(error = format!("{e:#}"), "swarm queue unreachable"); + None + } + }, + }; + let state = AppState { hives: Arc::new(load_hives()), links: Arc::new(load_links()), + status, }; let (router, api) = OpenApiRouter::::with_openapi(ApiDoc::openapi()) .routes(routes!(health)) .routes(routes!(get_hives)) + .routes(routes!(get_hives_status)) .routes(routes!(get_links)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from diff --git a/swarm-controller/src/queue.rs b/swarm-controller/src/queue.rs new file mode 100644 index 00000000..1be9e846 --- /dev/null +++ b/swarm-controller/src/queue.rs @@ -0,0 +1,197 @@ +//! The controller's client end of the swarm message queue. +//! +//! The queue admits every non-responder client through `auth_callout`: a +//! client presents a token at CONNECT, the callout responder introspects it +//! against authelia and mints a user JWT if it is good. So the controller is +//! an ordinary client and needs an identity of its own — it is not a hive, and +//! the per-hive clients issued from the roster are not its to use. +//! +//! Two things about that shape drive everything here: +//! +//! - **A token expires.** Authelia issues `client_credentials` access tokens +//! with `expires_in: 3599`. Authentication happens at CONNECT, so a +//! long-lived connection is fine — but a *reconnect* an hour later needs a +//! token that was minted an hour later. +//! - **`async-nats` re-runs an auth callback per connection attempt** (it is +//! handed that attempt's nonce). So the refresh belongs in the callback and +//! not in a timer: there is no window in which the client holds a token it +//! minted for a previous connection. +//! +//! The alternative — mint once, pass a static `auth_token`, own the reconnect +//! loop — fails in the way this subsystem exists to prevent: the controller +//! keeps serving, its status data quietly stops updating, and nothing says so +//! until someone reads a dashboard. + +use std::path::PathBuf; + +use anyhow::{Context, Result, bail}; + +/// Only the one field this needs; authelia returns several. +#[derive(serde::Deserialize)] +struct TokenResponse { + access_token: String, +} + +/// Where the controller finds the queue and what it authenticates with. +/// +/// Every field comes from an environment variable the NixOS module sets, the +/// same way `load_hives` takes the roster — a config change is a redeploy, and +/// this process reads no file it was not pointed at. +#[derive(Debug, Clone)] +pub struct QueueConfig { + /// `nats://host:port` for the swarm queue. + pub url: String, + /// Authelia's token endpoint, e.g. `https://auth./api/oidc/token`. + pub token_endpoint: String, + /// The controller's own `OAuth2` client id. + pub client_id: String, + /// File holding the client secret's PLAINTEXT. + /// + /// A path and not a value: the secret is minted on the authelia host and + /// read here, and putting it in the environment would publish it to + /// anything that can read `/proc//environ`. + pub client_secret_file: PathBuf, +} + +impl QueueConfig { + /// Read the config from the environment, or `None` when the queue was not + /// wired up for this deployment. + /// + /// `None` rather than an error on purpose: the controller serves its HTTP + /// surface on hosts where the queue is not enabled, and refusing to start + /// there would trade a missing feature for a dead daemon. What must NOT + /// happen is a *half* configuration silently behaving like an absent one — + /// hence the explicit partial check below. + pub fn from_env() -> Result> { + let url = std::env::var("SWARM_CONTROLLER_NATS_URL").ok(); + let token_endpoint = std::env::var("SWARM_CONTROLLER_OIDC_TOKEN_ENDPOINT").ok(); + let client_id = std::env::var("SWARM_CONTROLLER_OIDC_CLIENT_ID").ok(); + let secret = std::env::var("SWARM_CONTROLLER_OIDC_CLIENT_SECRET_FILE").ok(); + + match (url, token_endpoint, client_id, secret) { + (None, None, None, None) => Ok(None), + (Some(url), Some(token_endpoint), Some(client_id), Some(secret)) => Ok(Some(Self { + url, + token_endpoint, + client_id, + client_secret_file: PathBuf::from(secret), + })), + // A partially-set environment is a deployment bug, and the failure + // it would otherwise produce is the expensive kind: the controller + // comes up "fine", never connects, and every hive reads as having + // never reported. Naming the missing variables costs one line. + _ => bail!( + "swarm queue is half-configured: SWARM_CONTROLLER_NATS_URL, \ + _OIDC_TOKEN_ENDPOINT, _OIDC_CLIENT_ID and \ + _OIDC_CLIENT_SECRET_FILE must be set together or not at all" + ), + } + } +} + +/// Mint a fresh access token for the controller's own client. +/// +/// `client_credentials`, because there is no user here: the controller +/// 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 { + // 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) + .await + .with_context(|| { + format!( + "reading the queue client secret from {}", + cfg.client_secret_file.display() + ) + })?; + + let response = http + .post(&cfg.token_endpoint) + .form(&[ + ("grant_type", "client_credentials"), + ("client_id", cfg.client_id.as_str()), + ("client_secret", secret.trim()), + ]) + .send() + .await + .context("requesting an access token from authelia")?; + + // The body carries authelia's own error description, and it is far more + // useful than the status alone: a wrong grant says `unauthorized_client`, + // a wrong secret says `invalid_client`, and those point at different + // config. + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + if !status.is_success() { + bail!("authelia refused the controller's token request ({status}): {body}"); + } + + let parsed: TokenResponse = + serde_json::from_str(&body).context("parsing authelia's token response")?; + Ok(parsed.access_token) +} + +/// Connect to the swarm queue, minting a token for each connection attempt. +pub async fn connect(cfg: QueueConfig) -> Result { + let http = reqwest::Client::new(); + let url = cfg.url.clone(); + + let client = async_nats::ConnectOptions::with_auth_callback(move |_nonce| { + let http = http.clone(); + let cfg = cfg.clone(); + async move { + let token = mint_token(&http, &cfg) + .await + // The callback's error type carries a string, so the context + // chain would be lost; flatten it rather than dropping it. + .map_err(|e| async_nats::AuthError::new(format!("{e:#}")))?; + let mut auth = async_nats::Auth::new(); + auth.token = Some(token); + Ok(auth) + } + }) + // The controller and the queue are separate units on (possibly) + // separate hosts, and nothing orders them. Without this, a queue that + // comes up one second later leaves the controller permanently + // queue-less until someone restarts it — a boot-order race that + // presents as "status has been unavailable since Tuesday". + // + // It also composes with the callback above rather than fighting it: + // each background attempt is a connection attempt, so each one mints + // its own token instead of retrying a stale one. + .retry_on_initial_connect() + .connect(&url) + .await + .with_context(|| format!("connecting to the swarm queue at {url}"))?; + + Ok(client) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The all-unset case is the common one — most hosts do not run the queue. + #[test] + fn an_absent_environment_is_not_an_error() { + // Guard: this test would pass vacuously inside a configured + // environment, so it asserts the variables really are unset first. + for k in [ + "SWARM_CONTROLLER_NATS_URL", + "SWARM_CONTROLLER_OIDC_TOKEN_ENDPOINT", + "SWARM_CONTROLLER_OIDC_CLIENT_ID", + "SWARM_CONTROLLER_OIDC_CLIENT_SECRET_FILE", + ] { + if std::env::var(k).is_ok() { + return; + } + } + assert!( + QueueConfig::from_env() + .expect("absent is not an error") + .is_none() + ); + } +} diff --git a/swarm-controller/src/status.rs b/swarm-controller/src/status.rs new file mode 100644 index 00000000..b7d322aa --- /dev/null +++ b/swarm-controller/src/status.rs @@ -0,0 +1,543 @@ +//! Swarm-wide view of what each hive last said about itself. +//! +//! Hives **offer** a snapshot upward; this daemon never reaches down to +//! collect one. That direction is the design, not an implementation +//! detail: the hive gateway has gone down in a way where every recovery +//! channel ran through the one broken thing, so a status path that +//! depended on the controller would have gone dark exactly when it was +//! needed to diagnose the controller's own network. A hive computes its +//! status locally regardless of whether the swarm can be reached. +//! +//! **The queue is the store.** A hive publishes into a `JetStream` KV +//! bucket, which retains the last value per key; this daemon reads that +//! bucket per request and keeps no copy. A cache here would be a second +//! answer to the same question, free to disagree with the first — and +//! the disagreement would surface as a hive reading healthy on a +//! dashboard while the bucket says otherwise. +//! +//! **Absence is the case this is built around** — the freshness states +//! and the reasoning behind each are in `docs/swarm/README.md`. The two +//! properties that constrain the code rather than describe it: +//! freshness is **derived at read time**, never stored (a stored +//! `healthy: bool` goes stale silently the moment nothing arrives), and +//! rows come from the **roster**, not the bucket, so an empty bucket +//! cannot render as a healthy swarm. +//! +//! One consequence worth stating because it is the opposite of what a +//! cache would give: losing the bucket degrades **to honesty**. Every +//! hive reads `never_reported` until its next publish, which is the true +//! answer — not a remembered "healthy" from before the loss. + +use std::collections::BTreeMap; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use anyhow::{Context, Result}; +use futures_util::TryStreamExt as _; +use serde::{Deserialize, Serialize}; +use utoipa::ToSchema; + +use crate::HiveEntry; + +/// The KV bucket hives publish their snapshots into. +/// +/// A constant and not an option: reader and writer must name the same +/// bucket, and an option is a way for two deployments to disagree about +/// which one that is. Nothing about a bucket name is site-specific. +pub const BUCKET: &str = "hive-status"; + +/// Default age past which a snapshot is reported stale. +/// +/// A threshold is a statement about how often hives offer, and that +/// cadence is decided by the publisher (a later slice), so this is a +/// default to be overridden rather than a constant to be relied on. +pub const DEFAULT_STALE_AFTER: Duration = Duration::from_mins(2); + +/// Env var the NixOS module sets from +/// `services.hyperhive.swarm.controller.staleAfterSeconds`. +pub const STALE_AFTER_ENV: &str = "SWARM_CONTROLLER_STALE_AFTER_SECS"; + +/// How a hive's last report reads *now* — a function of the clock, not a +/// property of the report. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, ToSchema)] +#[serde(rename_all = "snake_case")] +pub enum Freshness { + /// Reported within the staleness threshold. + Fresh, + /// Reported, but longer ago than the threshold. The payload is still + /// rendered: "old" and "absent" are different answers and a consumer + /// may want the last thing a hive managed to say. + Stale, + /// In the roster, has never offered a snapshot. Distinct from + /// `Stale` because it separates "went quiet" from "never spoke" — + /// the first is a fault, the second is usually a deployment that + /// hasn't happened yet. + NeverReported, + /// Offered a snapshot but is not in the roster. Not an error this + /// daemon can resolve, and not one it should hide. + Unknown, +} + +/// One row of the aggregate. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, ToSchema)] +pub struct HiveStatus { + pub name: String, + /// From the roster; `None` for a hive the roster doesn't list. + pub domain: Option, + pub freshness: Freshness, + /// When the snapshot was stored, unix seconds. `None` when nothing + /// has been. + /// + /// This is the **bucket's** stamp, applied by the NATS server when + /// the value landed — not a field inside the payload. A publisher + /// therefore cannot make itself look fresher than it is, and a hive + /// with a wrong clock skews its own payload rather than this. + pub last_seen_unix: Option, + /// Age at render time. Carried alongside `last_seen_unix` so a + /// consumer with a different threshold need not re-derive it from a + /// clock that may not match this host's. + pub age_seconds: Option, + /// Whatever the hive published, unopened — stored opaquely so the + /// snapshot's contents stay settleable later without reworking the + /// aggregate. + /// + /// `None` in two cases the consumer can tell apart by `freshness`: + /// nothing has ever been published (`never_reported`), or something + /// was published that is not JSON (any other freshness — the row + /// still reports *when* the hive last spoke, and the read logs a + /// warning naming it). + pub snapshot: Option, +} + +/// A snapshot as retained by the bucket. +#[derive(Clone, Debug)] +struct Offered { + received_at: SystemTime, + /// `None` when the stored bytes are not JSON — see [`HiveStatus::snapshot`]. + payload: Option, +} + +/// Reads the aggregate out of the KV bucket. +/// +/// Holds a NATS client rather than a bucket handle: the bucket is +/// resolved on first use and cached, so a controller that starts before +/// the bucket exists picks it up without a restart. Resolution failures +/// are not cached — [`tokio::sync::OnceCell::get_or_try_init`] retries — +/// which is what makes the queue coming up *after* this daemon a +/// non-event rather than a permanent degradation. +pub struct StatusReader { + client: async_nats::Client, + store: tokio::sync::OnceCell, + stale_after: Duration, +} + +impl StatusReader { + #[must_use] + pub fn new(client: async_nats::Client, stale_after: Duration) -> Self { + Self { + client, + store: tokio::sync::OnceCell::new(), + stale_after, + } + } + + /// Reads [`STALE_AFTER_ENV`], falling back to + /// [`DEFAULT_STALE_AFTER`]. A zero or unparseable value takes the + /// default rather than failing startup — same rule as `load_hives`: + /// a controller whose own config is wrong must still serve. + #[must_use] + pub fn stale_after_from_env() -> Duration { + std::env::var(STALE_AFTER_ENV) + .ok() + .and_then(|raw| raw.trim().parse::().ok()) + .filter(|secs| *secs > 0) + .map_or(DEFAULT_STALE_AFTER, Duration::from_secs) + } + + /// The bucket handle, created on first use if nothing has made it yet. + /// + /// Whichever side arrives first creates it, and both sides want the + /// same shape, so this is a race with one outcome. `history: 1` is + /// the shape: the aggregate reads *the last thing each hive said*, + /// and retaining more would be storage bought for a query nobody + /// makes. + async fn store(&self) -> Result<&async_nats::jetstream::kv::Store> { + self.store + .get_or_try_init(|| async { + let js = async_nats::jetstream::new(self.client.clone()); + match js.get_key_value(BUCKET).await { + Ok(store) => Ok(store), + Err(e) => { + tracing::info!( + bucket = BUCKET, + reason = %e, + "status bucket not available, creating it" + ); + js.create_key_value(async_nats::jetstream::kv::Config { + bucket: BUCKET.to_owned(), + description: "Last status snapshot offered by each hive".to_owned(), + history: 1, + ..Default::default() + }) + .await + .with_context(|| format!("creating the {BUCKET} bucket")) + } + } + }) + .await + } + + /// The aggregate, rendered against `now`. + /// + /// Every roster hive produces a row whether or not it has ever + /// reported; a reporting hive outside the roster produces one too. + pub async fn view(&self, roster: &[HiveEntry], now: SystemTime) -> Result> { + // Only a CONNECTED client can be asked anything. `retry_on_initial_connect` + // means the client exists before it is usable, and a JetStream request + // made in that window does not fail — it WAITS, on every call, for + // longer than any dashboard poll should take (measured: still going at + // 15s against a queue that simply refuses the credential). + // + // Testing for `!= Connected` rather than `== Disconnected` is the whole + // point: a client that has never connected once sits in `Pending`, so + // the `Disconnected` test passes it straight through to the hang it was + // written to prevent. That is exactly the case here — a controller + // whose credential is wrong from boot never reaches `Disconnected`, + // because it was never connected to begin with. + // + // Naming the state is also the better error: "not connected" is + // actionable, a timeout is not. + let state = self.client.connection_state(); + if state != async_nats::connection::State::Connected { + anyhow::bail!("not connected to the swarm queue (client state: {state:?})"); + } + + let store = self.store().await?; + + // Keys first, then a fetch per key. The roster is a handful of + // hives, so the round-trip count is not worth trading for a + // watcher whose "I have seen everything current" condition is + // one more thing to get right on a read path. + let mut keys = store.keys().await.context("listing status bucket keys")?; + let mut entries: BTreeMap = BTreeMap::new(); + while let Some(key) = keys + .try_next() + .await + .context("reading the status bucket's key list")? + { + let Some(entry) = store + .entry(&key) + .await + .with_context(|| format!("reading status entry {key}"))? + else { + // Deleted between listing and fetching. Not an error: + // the next read simply won't list it. + continue; + }; + let payload = match serde_json::from_slice(&entry.value) { + Ok(value) => Some(value), + Err(e) => { + tracing::warn!( + hive = %key, + error = %e, + "status snapshot is not JSON; reporting the timestamp without it" + ); + None + } + }; + entries.insert( + key, + Offered { + received_at: to_system_time(entry.created.unix_timestamp()), + payload, + }, + ); + } + + Ok(render(roster, &entries, now, self.stale_after)) + } +} + +/// A bucket timestamp as a [`SystemTime`]. +/// +/// A pre-epoch stamp is not representable here and is not a thing a NATS +/// server produces; treating it as the epoch renders the row as +/// extremely stale, which is the safe direction — a nonsense timestamp +/// must never read as fresh. +fn to_system_time(unix_seconds: i64) -> SystemTime { + u64::try_from(unix_seconds).map_or(UNIX_EPOCH, |secs| UNIX_EPOCH + Duration::from_secs(secs)) +} + +/// Turn a roster plus whatever the bucket held into the rendered rows. +/// +/// Split out of [`StatusReader::view`] deliberately: this is where every +/// rule the acceptance criterion cares about lives, and keeping it a +/// pure function means those rules are tested against a table rather +/// than against a running NATS server. +fn render( + roster: &[HiveEntry], + entries: &BTreeMap, + now: SystemTime, + stale_after: Duration, +) -> Vec { + let mut rows: Vec = roster + .iter() + .map(|hive| match entries.get(&hive.name) { + Some(offered) => row( + hive.name.clone(), + Some(hive.domain.clone()), + offered, + now, + stale_after, + ), + None => HiveStatus { + name: hive.name.clone(), + domain: Some(hive.domain.clone()), + freshness: Freshness::NeverReported, + last_seen_unix: None, + age_seconds: None, + snapshot: None, + }, + }) + .collect(); + + rows.extend( + entries + .iter() + .filter(|(name, _)| !roster.iter().any(|hive| &&hive.name == name)) + .map(|(name, offered)| { + let mut unknown = row(name.clone(), None, offered, now, stale_after); + unknown.freshness = Freshness::Unknown; + unknown + }), + ); + rows +} + +fn row( + name: String, + domain: Option, + offered: &Offered, + now: SystemTime, + stale_after: Duration, +) -> HiveStatus { + // A snapshot stamped in the future (clock skew between the NATS + // server and this host) yields no age rather than a negative one, + // and is treated as fresh — the honest reading of "this arrived, I + // cannot tell how long ago". + let age = now.duration_since(offered.received_at).ok(); + let freshness = match age { + Some(age) if age > stale_after => Freshness::Stale, + _ => Freshness::Fresh, + }; + HiveStatus { + name, + domain, + freshness, + last_seen_unix: offered + .received_at + .duration_since(UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()), + age_seconds: age.map(|age| age.as_secs()), + snapshot: offered.payload.clone(), + } +} + +#[cfg(test)] +mod tests { + use super::{DEFAULT_STALE_AFTER, Freshness, Offered, STALE_AFTER_ENV, StatusReader, render}; + use crate::HiveEntry; + use std::collections::BTreeMap; + use std::time::{Duration, SystemTime}; + + fn roster() -> Vec { + vec![ + HiveEntry { + name: "pr1ma".to_owned(), + domain: "pr1ma.example.com".to_owned(), + }, + HiveEntry { + name: "umbra".to_owned(), + domain: "umbra.example.com".to_owned(), + }, + ] + } + + fn entries(rows: &[(&str, SystemTime)]) -> BTreeMap { + rows.iter() + .map(|(name, at)| { + ( + (*name).to_owned(), + Offered { + received_at: *at, + payload: Some(serde_json::json!({ "ok": true })), + }, + ) + }) + .collect() + } + + fn t0() -> SystemTime { + SystemTime::UNIX_EPOCH + Duration::from_secs(1_700_000_000) + } + + /// The whole point of the module: an empty bucket must not render as + /// a healthy swarm. Rows come from the roster, so silence is visible. + #[test] + fn an_empty_bucket_renders_every_hive_as_never_reported() { + let rows = render(&roster(), &BTreeMap::new(), t0(), DEFAULT_STALE_AFTER); + assert_eq!(rows.len(), 2, "a row per roster hive, not per report"); + assert!( + rows.iter().all(|r| r.freshness == Freshness::NeverReported), + "nothing published means nothing known — not healthy" + ); + assert!(rows.iter().all(|r| r.snapshot.is_none())); + } + + /// Freshness is derived from the clock at read time, so the same + /// stored value reads differently as it ages. The boundary is where + /// an off-by-one would hide, so it is pinned in both directions. + #[test] + fn the_threshold_boundary_is_inclusive() { + let stale_after = Duration::from_secs(90); + let stored = entries(&[("pr1ma", t0())]); + + assert_eq!( + render( + &roster(), + &stored, + t0() + Duration::from_secs(90), + stale_after + )[0] + .freshness, + Freshness::Fresh, + "exactly at the threshold is inside it" + ); + assert_eq!( + render( + &roster(), + &stored, + t0() + Duration::from_secs(91), + stale_after + )[0] + .freshness, + Freshness::Stale, + "one second past is outside it" + ); + } + + /// One hive reporting must not make its silent neighbour look + /// healthy — the failure mode of any aggregate that renders only + /// what it has. + #[test] + fn a_reporting_hive_does_not_vouch_for_a_silent_one() { + let rows = render( + &roster(), + &entries(&[("pr1ma", t0())]), + t0(), + DEFAULT_STALE_AFTER, + ); + assert_eq!(rows[0].name, "pr1ma"); + assert_eq!(rows[0].freshness, Freshness::Fresh); + assert_eq!(rows[1].name, "umbra"); + assert_eq!(rows[1].freshness, Freshness::NeverReported); + } + + /// An observation the daemon cannot explain is surfaced, not dropped. + #[test] + fn a_hive_outside_the_roster_is_surfaced_as_unknown() { + let rows = render( + &roster(), + &entries(&[("ghost", t0())]), + t0(), + DEFAULT_STALE_AFTER, + ); + assert_eq!(rows.len(), 3, "two roster hives plus the stranger"); + let ghost = rows.last().expect("rows is non-empty"); + assert_eq!(ghost.name, "ghost"); + assert_eq!(ghost.freshness, Freshness::Unknown); + assert!( + ghost.domain.is_none(), + "the roster is where a domain comes from, and this hive isn't in it" + ); + } + + /// Clock skew must not produce a negative age or a panic. A snapshot + /// stamped in the future reads fresh with no age — "it arrived, I + /// cannot tell how long ago". + #[test] + fn a_future_timestamp_yields_no_age_rather_than_a_wrong_one() { + let rows = render( + &roster(), + &entries(&[("pr1ma", t0() + Duration::from_secs(30))]), + t0(), + Duration::from_secs(90), + ); + assert_eq!(rows[0].freshness, Freshness::Fresh); + assert_eq!(rows[0].age_seconds, None); + } + + /// A hive that published something unreadable still gets its + /// timestamp reported: *when* it last spoke is exactly what this + /// aggregate is for, and dropping the row would read as silence. + #[test] + fn an_unparseable_payload_still_reports_when_it_arrived() { + let mut stored = BTreeMap::new(); + stored.insert( + "pr1ma".to_owned(), + Offered { + received_at: t0(), + payload: None, + }, + ); + + let row = &render(&roster(), &stored, t0(), DEFAULT_STALE_AFTER)[0]; + assert_eq!( + row.freshness, + Freshness::Fresh, + "unreadable is not the same as absent — freshness is what \ + separates them on the wire" + ); + assert!(row.snapshot.is_none()); + assert_eq!(row.last_seen_unix, Some(1_700_000_000)); + } + + /// SAFETY: single-threaded mutation of a process env var no other + /// test in this crate reads; restored before returning. One test + /// rather than four for the same reason `load_hives`'s is — the + /// parallel runner would race them. + #[test] + fn stale_after_from_env_covers_missing_bogus_zero_and_valid() { + unsafe { + std::env::remove_var(STALE_AFTER_ENV); + } + assert_eq!(StatusReader::stale_after_from_env(), DEFAULT_STALE_AFTER); + + unsafe { + std::env::set_var(STALE_AFTER_ENV, "not a number"); + } + assert_eq!( + StatusReader::stale_after_from_env(), + DEFAULT_STALE_AFTER, + "a controller whose own config is wrong must still serve" + ); + + unsafe { + std::env::set_var(STALE_AFTER_ENV, "0"); + } + assert_eq!( + StatusReader::stale_after_from_env(), + DEFAULT_STALE_AFTER, + "zero would make every snapshot instantly stale — take the default" + ); + + unsafe { + std::env::set_var(STALE_AFTER_ENV, "300"); + } + assert_eq!(StatusReader::stale_after_from_env(), Duration::from_mins(5)); + + unsafe { + std::env::remove_var(STALE_AFTER_ENV); + } + } +}