Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
bc594a36ef | ||
|
|
c67100588e | ||
|
|
c2f8ee225d | ||
|
|
8c8f0e2758 | ||
|
|
016ab70e31 | ||
|
|
32a353a9c5 | ||
|
|
602ae92a3e | ||
|
|
465d68d5a7 | ||
|
|
8891b46943 |
21 changed files with 1099 additions and 349 deletions
25
Cargo.lock
generated
25
Cargo.lock
generated
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -200,8 +200,6 @@ nginx obtains and auto-renews certs via the ACME HTTP-01 challenge on `port` (de
|
|||
|
||||
Mutual exclusion: `tls.certDir` set together with `tls.acme.enable = true` fails an assertion — pick one external TLS source (or neither, for the self-signed default).
|
||||
|
||||
**Swarm peers**: CA-signed certs are trusted by default — this hive's entry in `swarm.hives` needs no `certFingerprint`.
|
||||
|
||||
### Self-signed TLS (default)
|
||||
|
||||
On by default, and listens on `httpsPort` (default 443) on every vhost beside the plain-http `port` (default 80).
|
||||
|
|
@ -252,12 +250,6 @@ security.acme.certs."example.com".group = "nginx";
|
|||
|
||||
or make the key world-readable (`0644`) if your threat model allows it. nginx errors out at startup on a key it can't read — the error is explicit in the journal, not a silent failure.
|
||||
|
||||
**Swarm directory entry**: when using a CA-signed cert, this hive's entry needs no `certFingerprint` — the standard CA bundle validates:
|
||||
|
||||
```nix
|
||||
services.hyperhive.swarm.hives.example = { domain = "example.com"; }; # no certFingerprint needed
|
||||
```
|
||||
|
||||
### Fronting with an external TLS terminator
|
||||
|
||||
There is no http-only mode (see [TLS modes](#tls-modes) above). Two paths
|
||||
|
|
|
|||
|
|
@ -107,9 +107,8 @@ swarm service name has to be wired into: [`ui.md`](ui.md).
|
|||
|
||||
```nix
|
||||
services.hyperhive.swarm.hives = {
|
||||
pr1ma = { domain = "pr1ma.example.com"; }; # this host, per hiveName
|
||||
lab = { domain = "lab.example.com"; }; # CA-trusted (Let's Encrypt etc.)
|
||||
edge = { domain = "edge.corp"; certFingerprint = "sha256:…"; }; # self-signed leaf, pinned
|
||||
pr1ma = { domain = "pr1ma.example.com"; }; # this host, per hiveName
|
||||
lab = { domain = "lab.example.com"; }; # a second hive in the swarm
|
||||
};
|
||||
```
|
||||
|
||||
|
|
@ -128,76 +127,30 @@ contain you derives every hive as a peer and you peer with yourself.
|
|||
conventionally `<name>.<swarm.domain>`, but a wrong domain that
|
||||
evaluates cleanly points at a real machine that isn't the one you meant.
|
||||
|
||||
**`certFingerprint`** (`"sha256:…"`, optional) pins that hive's TLS
|
||||
_leaf_. Scopes **only** to hive-c0re's own peer HTTPS checks (the P33RS
|
||||
dashboard links + agent peer discovery below); matrix federation never
|
||||
consults it. Omit it for any hive under the swarm root CA or a public
|
||||
CA — which is the normal case.
|
||||
|
||||
> **There is no per-hive CA field.** Trust inside a swarm comes from the
|
||||
> swarm root ([`ca.md`](ca.md)): every hive chains to it, so one anchor
|
||||
> replaces the O(n²) pinning. What that genuinely drops is trusting a
|
||||
> hive whose root this swarm does *not* own — another swarm's, or one
|
||||
> keeping its own CA. That is a cross-swarm problem and wants a
|
||||
> mechanism designed for it, not a field that happened to work.
|
||||
|
||||
### Fingerprint format
|
||||
|
||||
The value is the string `sha256:` followed by exactly 64 hexadecimal
|
||||
digits — the SHA-256 digest of the peer's DER-encoded TLS leaf
|
||||
certificate. The hex is case-insensitive (upper or lower both parse),
|
||||
carries no colon separators between bytes, and any value not matching
|
||||
this shape is ignored with a warning rather than weakening trust.
|
||||
|
||||
```
|
||||
sha256:b1946ac92492d2347c6235b4d2611184a3f5b6cae6c19d6e3c2f0a8e7d4c9f12
|
||||
```
|
||||
|
||||
Generate it from the peer's certificate with openssl. The
|
||||
`-fingerprint -sha256` output is uppercase and colon-separated, so
|
||||
strip the colons, lowercase, and prepend the `sha256:` prefix:
|
||||
|
||||
```sh
|
||||
# from a PEM/CRT file
|
||||
openssl x509 -in peer.crt -noout -fingerprint -sha256 \
|
||||
| sed 's/^.*=//; s/://g' | tr 'A-Z' 'a-z' | sed 's/^/sha256:/'
|
||||
|
||||
# straight from the live endpoint (port 443)
|
||||
echo | openssl s_client -connect peer.example.com:443 -servername peer.example.com 2>/dev/null \
|
||||
| openssl x509 -noout -fingerprint -sha256 \
|
||||
| sed 's/^.*=//; s/://g' | tr 'A-Z' 'a-z' | sed 's/^/sha256:/'
|
||||
```
|
||||
|
||||
Pin the leaf certificate, not an intermediate or the CA — the
|
||||
digest must match the exact cert the peer serves on its HTTPS
|
||||
endpoint. When the peer rotates its cert, update the pin to the new
|
||||
fingerprint (or switch the peer to a CA-trusted cert and drop the
|
||||
field).
|
||||
|
||||
The nix module serialises the attrset to a `HYPERHIVE_PEERS` JSON
|
||||
array (`[{ domain, cert_fingerprint }]`) injected into the c0re
|
||||
environment and forwarded to agent containers.
|
||||
> **There is no per-hive CA field, and no per-hive cert pinning.** Trust
|
||||
> inside a swarm comes from the swarm root ([`ca.md`](ca.md)): every
|
||||
> hive chains to it, so one anchor replaces per-hive pinning entirely.
|
||||
> What that genuinely drops is trusting a hive whose root this swarm
|
||||
> does *not* own — another swarm's, or one keeping its own CA. That is
|
||||
> a cross-swarm problem and wants a mechanism designed for it. (An
|
||||
> earlier `certFingerprint` field existed for exactly that gap, pinning
|
||||
> a peer's TLS leaf for hive-c0re's own peer HTTPS checks — removed
|
||||
> along with the dashboard feature it existed to serve, since nothing
|
||||
> else ever consumed it.)
|
||||
|
||||
## What the config does at runtime
|
||||
|
||||
1. **Dashboard P33RS tab** — hive-c0re reads `HYPERHIVE_PEERS` and
|
||||
surfaces it as the peer list in the dashboard's state API. The
|
||||
dashboard shows a P33RS tab (hidden when the list is empty) with a
|
||||
card per peer linking to `https://{domain}/`. Wire format + module
|
||||
pointer: `docs/web-ui/dashboard.md` § P33RS tab.
|
||||
1. **Swarm-wide hive roster** — swarm-controller reads this same
|
||||
directory and serves it at `GET /api/hives`; `swarm-ui`'s overview
|
||||
page renders it (`docs/swarm/ui.md`). This is the operator-facing
|
||||
"what hives exist" surface — a per-hive dashboard "peer hives"
|
||||
display existed here once and was removed in favour of it.
|
||||
|
||||
2. **Agent identity** — the same `HYPERHIVE_PEERS` env var is
|
||||
forwarded to agent containers, so agent code can discover peer
|
||||
hives and address them with qualified names (`agent@domain`). See
|
||||
`hive-agent/src/identity.rs`'s module doc for the label/domain
|
||||
helpers.
|
||||
|
||||
3. **Matrix federation** — when `matrix.enable` is on, tuwunel
|
||||
2. **Matrix federation** — when `matrix.enable` is on, tuwunel
|
||||
federates with the peer's matrix server (discovered via the peer's
|
||||
`.well-known/matrix/server` delegation, which the gateway serves).
|
||||
Federation validates the peer's TLS certificate against the matrix
|
||||
**container's** trust bundle — independently of `certFingerprint`,
|
||||
which it never consults.
|
||||
**container's** trust bundle, independent of this directory.
|
||||
|
||||
⚠️ **That container currently trusts no swarm-internal CA**, so a
|
||||
self-signed gateway certificate does not federate. The swarm root
|
||||
|
|
@ -208,6 +161,10 @@ environment and forwarded to agent containers.
|
|||
issue. Until then, federation needs CA-issued certs (ACME). See
|
||||
`docs/matrix.md` for federation firewall + TLS requirements.
|
||||
|
||||
3. **WireGuard mesh** (optional) — `swarm.wireguard.enable` reads each
|
||||
entry's `wireguardPublicKey`/`wireguardEndpoint`/`wireguardAddress`
|
||||
to configure `wg-hive`. See "WireGuard inter-hive mesh" below.
|
||||
|
||||
## One directory, not a bilateral declaration
|
||||
|
||||
Both hives hold the **same** `hives` attrset; neither declares the
|
||||
|
|
@ -264,7 +221,6 @@ services.hyperhive = {
|
|||
};
|
||||
edge = {
|
||||
domain = "edge.corp";
|
||||
certFingerprint = "sha256:…"; # TLS trust (unchanged)
|
||||
wireguardPublicKey = "base64keyB=";
|
||||
wireguardEndpoint = "203.0.113.42:51820";
|
||||
wireguardAddress = "10.100.0.2/32";
|
||||
|
|
@ -290,9 +246,10 @@ services.hyperhive = {
|
|||
(not inside agent containers; containers reach peers via the host's
|
||||
routing table).
|
||||
- UDP port 51820 (or `listenPort`) is opened on the host firewall.
|
||||
- `HYPERHIVE_PEERS` gains a `wireguard_address` field for each mesh
|
||||
peer so hive-c0re can reach intra-swarm services without a public
|
||||
DNS round-trip.
|
||||
- `swarm-wireguard.nix` reads each entry's `wireguardAddress` directly
|
||||
from `services.hyperhive.swarm.peerHives` to build `wg-hive`'s
|
||||
`allowedIPs`, so intra-swarm traffic can route over the mesh address
|
||||
rather than the public domain.
|
||||
- `persistentKeepalive = 25` is set by default; override or null to
|
||||
disable (not needed when both sides have public IPs and no NAT).
|
||||
|
||||
|
|
@ -330,6 +287,39 @@ 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`)
|
||||
|
||||
One row per hive in `swarm.hives`, saying when it last reported and what
|
||||
it said. Hives publish upward through the swarm queue; the controller
|
||||
never reaches down to collect, so a hive that cannot reach the swarm
|
||||
still knows its own state — you just cannot see it from here.
|
||||
|
||||
⚠️ **Nothing publishes yet.** The read path is in place; the hive-side
|
||||
publisher lands in a later change. Until it does, every hive reads
|
||||
`never_reported`.
|
||||
|
||||
| freshness | what to do about it |
|
||||
|---|---|
|
||||
| `fresh` | nothing — reported within `staleAfterSeconds` |
|
||||
| `stale` | the hive stopped reporting. Its last payload is still shown, so check `age_seconds` and the payload for what it managed to say |
|
||||
| `never_reported` | this hive has never reported at all — normally a deployment that hasn't happened, not an outage |
|
||||
| `unknown` | something is publishing under a name that is not in `swarm.hives` — a typo in the roster, or a hive that was removed and is still running |
|
||||
|
||||
Every row also carries `last_seen_unix` and `age_seconds` if you want to
|
||||
apply your own threshold. The timestamp is the one the queue recorded on
|
||||
arrival, not one the hive put in its own payload.
|
||||
|
||||
Set `services.hyperhive.swarm.controller.staleAfterSeconds` (default
|
||||
`120`) **above the rate hives publish at**, or everything reads `stale`
|
||||
between reports. It takes effect on the next request; nothing has to
|
||||
re-publish.
|
||||
|
||||
The endpoint answers **503** when this host has no swarm queue
|
||||
configured, or has one and cannot read it — deliberately not an empty
|
||||
list, which would look like a silent swarm rather than a controller that
|
||||
cannot see. The body says which. Status survives a controller restart:
|
||||
it is stored in the queue, not in the daemon.
|
||||
|
||||
## Cross-references
|
||||
|
||||
- `docs/snapshot-store.md` — the swarm's `btrfs receive` endpoint, and
|
||||
|
|
@ -337,6 +327,8 @@ socket-directory constraint that governs where `socketPath` may point:
|
|||
- `docs/conventions.md` § Hive identity — env vars, qualified labels
|
||||
- `docs/matrix.md` — matrix federation, TLS cert auto-generation,
|
||||
firewall posture
|
||||
- `docs/web-ui/dashboard.md` § P33RS tab — dashboard surface
|
||||
- `docs/swarm/ui.md` — the swarm-wide hive roster, now the operator
|
||||
surface for "what hives exist" (superseded the per-hive dashboard's
|
||||
old "peer hives" display)
|
||||
- `docs/gateway.md` — nginx vhosts and the `.well-known/matrix/`
|
||||
auto-discovery scheme
|
||||
|
|
|
|||
|
|
@ -18,9 +18,7 @@ The **dashboard** itself (`/dashboard.html`) is where you'll spend most
|
|||
of your time. It's a single page with exactly four tabs:
|
||||
|
||||
- **SW4RM** — every agent, live. This is the default tab and the one
|
||||
you'll check most. When the hive has peer hives configured, they show
|
||||
up here too, as a card list under the main container list — not a
|
||||
separate tab.
|
||||
you'll check most.
|
||||
- **Y3R C4LL** — anything waiting on *you*: pending approvals and
|
||||
agent questions. If an agent needs a decision from you, it's here.
|
||||
- **P3RM1SS10NS** — what tools and system-level access each agent has.
|
||||
|
|
|
|||
|
|
@ -7,7 +7,7 @@
|
|||
The dashboard is served at `/dashboard.html` (with the home page at `/`).
|
||||
It has a fixed chrome header at the top and a `<main>` that shows exactly
|
||||
one tab pane at a time. The URL hash (`#swarm`, `#call`, `#system`,
|
||||
`#permissions`, `#schedules`, `#peers`, `#settings`) drives which pane is
|
||||
`#permissions`, `#schedules`, `#settings`) drives which pane is
|
||||
active; hash changes don't reload the page. FL0W, L0GS, and the optional
|
||||
M4TR1X client are separate pages reachable from the H0M3 hub at `/`, not
|
||||
from the dashboard tab strip.
|
||||
|
|
@ -19,11 +19,10 @@ from the dashboard tab strip.
|
|||
`◆ SCH3DUL3S ◆`. In-page tabs only — the SYST3M panels moved to the
|
||||
standalone **C0R3** page (`/core.html`), and FL0W / L0GS / ST4TS /
|
||||
S3TT1NGS / M4TR1X live on their own pages too, all reachable from the
|
||||
H0M3 hub (not the tab strip). Peer hives render as a headline under
|
||||
SW4RM rather than a tab. Count pills on SW4RM
|
||||
H0M3 hub (not the tab strip). Count pills on SW4RM
|
||||
(container count), Y3R C4LL (pending approvals + questions + unread
|
||||
operator messages), and SCH3DUL3S (active schedules); P33RS and
|
||||
S3TT1NGS have no count.
|
||||
operator messages), and SCH3DUL3S (active schedules); S3TT1NGS has no
|
||||
count.
|
||||
- **Banner-thin** (`░▒▓█▓▒░ HYPERHIVE / HIVE-C0RE / WE ARE THE WIRED ░▒▓█▓▒░`)
|
||||
— sits below the tab strip.
|
||||
- **Server-warnings banner** — a generic, sticky top-of-page strip shown
|
||||
|
|
@ -642,44 +641,6 @@ the model id, longest match wins) mapping to
|
|||
`{ input, output, cache_read, cache_write }` USD-per-million-token
|
||||
prices. Models not covered fall back to hive-c0re's built-in estimate.
|
||||
|
||||
## P33R H1V3S (within the SW4RM tab)
|
||||
|
||||
Peer hives in this swarm. Not its own tab — a headline block
|
||||
(`#peers-block`) rendered under the SW4RM tab's container list (see
|
||||
"Chrome header" above: "Peer hives render as a headline under SW4RM
|
||||
rather than a tab"). The block is hidden when the
|
||||
`state.peer_hives` array from `/api/state` is empty — i.e. when
|
||||
`services.hyperhive.swarm.hives` holds no hive other than this one.
|
||||
When at least one peer is present the `hidden` attribute is removed
|
||||
and the cards render.
|
||||
|
||||
**P33R H1V3S** — each peer renders as a card row: a hexagon icon
|
||||
(`⬡`), the peer's DNS domain as the primary name, and the peer
|
||||
dashboard HTTPS URL as a clickable secondary link. Clicking the
|
||||
URL opens the peer hive's dashboard in a new tab.
|
||||
|
||||
### Backend wiring
|
||||
|
||||
The host daemon reads `services.hyperhive.swarm.peerHives` from the
|
||||
nix config (the `swarm.hives` directory minus this hive), serialises
|
||||
each entry as `{ name, url }` into `state.peer_hives: Vec<PeerHiveView>`,
|
||||
and includes the field in the `/api/state` snapshot. `tabs.js`
|
||||
reads `state.peer_hives` on every `refreshState` call and calls
|
||||
`swarm.js::renderPeerHives(peers)`, which rebuilds the `#peers-section`
|
||||
div from scratch.
|
||||
|
||||
The `name` field is the peer's DNS domain (its entry's `domain`, not
|
||||
the attrset key — the key is the hive's name); `url`
|
||||
is `https://{domain}/`. Both are derived from the env var
|
||||
`HYPERHIVE_PEERS` (a JSON array of `{ domain, cert_fingerprint }`
|
||||
objects) that the nix module writes into the c0re container
|
||||
environment. `cert_fingerprint` is null for CA-trusted (e.g.
|
||||
Let's Encrypt) peers and non-null to pin a self-signed cert.
|
||||
`parse_peer_hives()` in `hive-c0re/src/dashboard/state_snapshot.rs`
|
||||
converts each entry to the
|
||||
`PeerHiveView { name: domain, url: "https://domain/" }` shape the
|
||||
frontend reads.
|
||||
|
||||
## S3TT1NGS tab
|
||||
|
||||
Operator-local preferences. State lives in the browser's
|
||||
|
|
|
|||
|
|
@ -100,15 +100,6 @@
|
|||
<div id="containers-section">
|
||||
<p class="meta">loading…</p>
|
||||
</div>
|
||||
<!-- Peer hives (federated swarms). The headline block is hidden
|
||||
entirely when nothing is federated; renderPeerHives then drops
|
||||
a quiet "no peer hives configured" note into #peers-section. -->
|
||||
<div id="peers-block" hidden>
|
||||
<h2>◆ P33R H1V3S ◆</h2>
|
||||
<div class="divider">══════════════════════════════════════════════════════════════</div>
|
||||
<p class="meta">peer hives in this swarm. click a card to open that hive's dashboard.</p>
|
||||
</div>
|
||||
<div id="peers-section"></div>
|
||||
</section>
|
||||
|
||||
<!-- Y3R C4LL: things blocked on operator decision. Approvals +
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// SW4RM (containers) domain — extracted from tabs.js.
|
||||
// Agent topology, container-row rendering, selection bar, peer-hives block,
|
||||
// and all live-update apply handlers for container-state, transient, and
|
||||
// Agent topology, container-row rendering, selection bar, and all
|
||||
// live-update apply handlers for container-state, transient, and
|
||||
// job-queue-rollup ops. See docs/web-ui.md::Container row for the
|
||||
// rendering contract.
|
||||
|
||||
|
|
@ -1049,43 +1049,6 @@ function addBulkButton(parent, btnClass, label, enabled, selected, opts) {
|
|||
parent.append(btn);
|
||||
}
|
||||
|
||||
// ─── peer hives ──────────────────────────────────────────────────────────────
|
||||
|
||||
// Peer hives: render link cards as a headline section under SW4RM
|
||||
// (state.peer_hives). Called on every state refresh. When nothing is
|
||||
// federated, the "P33R H1V3S" headline block is hidden entirely and a
|
||||
// quiet grey note replaces the cards.
|
||||
export function renderPeerHives(peers) {
|
||||
const block = $('peers-block');
|
||||
const root = $('peers-section');
|
||||
if (!root) return;
|
||||
root.replaceChildren();
|
||||
const hasPeers = Array.isArray(peers) && peers.length > 0;
|
||||
if (block) block.hidden = !hasPeers;
|
||||
if (!hasPeers) {
|
||||
root.append(el('p', { class: 'empty' }, 'no peer hives configured'));
|
||||
return;
|
||||
}
|
||||
const ul = el('ul', { class: 'containers' });
|
||||
for (const p of peers) {
|
||||
const li = el('li', { class: 'container-row' });
|
||||
const head = el('div', { class: 'head' });
|
||||
const icon = el('span', { class: 'container-icon' }, '⬡');
|
||||
const nameEl = el('span', { class: 'name' }, p.name || p.url);
|
||||
const linkEl = el('a', {
|
||||
href: p.url,
|
||||
target: '_blank',
|
||||
rel: 'noopener noreferrer',
|
||||
class: 'meta',
|
||||
title: 'open ' + p.name + ' dashboard',
|
||||
}, p.url);
|
||||
head.append(icon, nameEl);
|
||||
li.append(head, el('div', { class: 'meta' }, linkEl));
|
||||
ul.append(li);
|
||||
}
|
||||
root.append(ul);
|
||||
}
|
||||
|
||||
// ─── tickers ─────────────────────────────────────────────────────────────────
|
||||
|
||||
// 30s ticker for agent status-age chips. Renderers stamp `data-set-at`
|
||||
|
|
|
|||
|
|
@ -46,7 +46,7 @@ import {
|
|||
applyRebuildQueueChanged, applyContainerStateChanged, applyContainerRemoved,
|
||||
applyTransientSet, applyTransientCleared,
|
||||
renderContainers, renderContainersFromState,
|
||||
renderSelectionBar, renderPeerHives,
|
||||
renderSelectionBar,
|
||||
} from './swarm.js';
|
||||
|
||||
// mdNode (in common.js) reads `window.marked` for the markdown side
|
||||
|
|
@ -201,9 +201,6 @@ window.marked = marked;
|
|||
// synchronous read (e.g. the compose autocomplete pulls agent
|
||||
// names from here instead of refetching on every keystroke).
|
||||
window.__hyperhive_state = s;
|
||||
// Peer hives render as a headline section under SW4RM (no longer
|
||||
// a tab); renderPeerHives shows/hides its own headline block.
|
||||
renderPeerHives(s.peer_hives || []);
|
||||
renderServerWarnings(s.server_warnings);
|
||||
// (The M4TR1X surface is reachable from the H0M3 hub now, not the
|
||||
// dashboard tab strip — home.js gates its tile on matrix_gui_enabled.)
|
||||
|
|
|
|||
|
|
@ -120,12 +120,6 @@ pub(super) struct StateSnapshot {
|
|||
/// var, set from `services.hyperhive.swarm.name`. `None` when
|
||||
/// unset — chrome omits the swarm segment of the breadcrumb.
|
||||
swarm_name: Option<String>,
|
||||
/// Peer hives in the same swarm. Parsed from `HYPERHIVE_PEERS`
|
||||
/// (JSON array of `{domain,cert_fingerprint}` objects, emitted by
|
||||
/// the c0re NixOS module from `services.hyperhive.swarm.peerHives`
|
||||
/// — the swarm's `hives` directory minus this hive).
|
||||
/// Empty on single-hive deploys. Feeds the P33RS dashboard tab.
|
||||
peer_hives: Vec<PeerHiveView>,
|
||||
/// Server-level warnings for the dashboard's top-of-page banner
|
||||
/// (currently host disk-pressure; more producers can be added
|
||||
/// backend-side). Empty when all clear. Built by
|
||||
|
|
@ -161,18 +155,6 @@ async fn infra_container_views() -> Vec<InfraContainerView> {
|
|||
infra_containers
|
||||
}
|
||||
|
||||
/// One peer hive for the P33RS dashboard tab. Derived from
|
||||
/// `HYPERHIVE_PEERS` env; `url` is the peer's HTTPS dashboard root.
|
||||
/// `cert_fingerprint` is `Some("sha256:<hex64>")` when the operator
|
||||
/// pinned the peer's leaf in `services.hyperhive.swarm.hives`, which a
|
||||
/// hive under the swarm root CA does not need.
|
||||
#[derive(Serialize)]
|
||||
struct PeerHiveView {
|
||||
name: String,
|
||||
url: String,
|
||||
cert_fingerprint: Option<String>,
|
||||
}
|
||||
|
||||
/// `OpQuestion` + computed `question_refs` / `answer_refs`. Built
|
||||
/// from the snapshot read; the live channel attaches the same
|
||||
/// fields directly on `QuestionAdded` / `QuestionResolved`.
|
||||
|
|
@ -442,64 +424,11 @@ pub(super) async fn api_state(
|
|||
swarm_name: std::env::var("HYPERHIVE_SWARM_NAME")
|
||||
.ok()
|
||||
.filter(|s| !s.is_empty()),
|
||||
peer_hives: parse_peer_hives(),
|
||||
server_warnings,
|
||||
infra_containers,
|
||||
})
|
||||
}
|
||||
|
||||
/// Parse `HYPERHIVE_PEERS` env var into dashboard-ready `PeerHiveView`
|
||||
/// entries. The env var is a JSON array of `{domain, cert_fingerprint}`
|
||||
/// objects emitted by the c0re NixOS module from
|
||||
/// `services.hyperhive.swarm.peerHives`. Each entry becomes
|
||||
/// `{ name: domain, url: "https://domain/" }` for the P33RS tab.
|
||||
/// Returns empty vec when unset (single-hive deploy).
|
||||
fn parse_peer_hives() -> Vec<PeerHiveView> {
|
||||
#[derive(serde::Deserialize)]
|
||||
struct Raw {
|
||||
domain: String,
|
||||
cert_fingerprint: Option<String>,
|
||||
}
|
||||
let Ok(json) = std::env::var("HYPERHIVE_PEERS") else {
|
||||
return Vec::new();
|
||||
};
|
||||
let Ok(raw): Result<Vec<Raw>, _> = serde_json::from_str(&json) else {
|
||||
tracing::warn!("HYPERHIVE_PEERS is not valid JSON; ignoring");
|
||||
return Vec::new();
|
||||
};
|
||||
raw.into_iter()
|
||||
.map(|r| {
|
||||
let cert_fingerprint = r.cert_fingerprint.and_then(|fp| {
|
||||
if validate_cert_fingerprint(&fp) {
|
||||
Some(fp)
|
||||
} else {
|
||||
tracing::warn!(
|
||||
domain = %r.domain,
|
||||
fingerprint = %fp,
|
||||
"HYPERHIVE_PEERS: invalid cert_fingerprint format \
|
||||
(expected `sha256:<64 hex chars>`); ignoring fingerprint"
|
||||
);
|
||||
None
|
||||
}
|
||||
});
|
||||
PeerHiveView {
|
||||
name: r.domain.clone(),
|
||||
url: format!("https://{}/", r.domain),
|
||||
cert_fingerprint,
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Validate a TLS certificate fingerprint string from `HYPERHIVE_PEERS`.
|
||||
/// Accepts `sha256:<64 hex chars>` (upper or lower case).
|
||||
fn validate_cert_fingerprint(fp: &str) -> bool {
|
||||
let Some(hex) = fp.strip_prefix("sha256:") else {
|
||||
return false;
|
||||
};
|
||||
hex.len() == 64 && hex.chars().all(|c| c.is_ascii_hexdigit())
|
||||
}
|
||||
|
||||
/// Group live containers by their assigned web UI port; clusters with
|
||||
/// more than one member are port-hash collisions the operator needs
|
||||
/// to resolve by renaming. Manager (fixed at 8000) and sub-agents
|
||||
|
|
|
|||
|
|
@ -703,7 +703,6 @@ const FORWARDED_VARS: &[&str] = &[
|
|||
"HIVE_FORGE_URL",
|
||||
"HIVE_FORGE_PUBLIC_URL",
|
||||
"HIVE_MATRIX_URL",
|
||||
"HYPERHIVE_PEERS",
|
||||
"HYPERHIVE_HIVE_DOMAIN",
|
||||
"HYPERHIVE_HIVE_NAME",
|
||||
"HYPERHIVE_SWARM_NAME",
|
||||
|
|
|
|||
|
|
@ -209,25 +209,3 @@ in
|
|||
in
|
||||
"${s.address}:${toString s.port}";
|
||||
}
|
||||
// lib.optionalAttrs (config.services.hyperhive.swarm.peerHives != { }) {
|
||||
# Peer hives serialised as a JSON array of {domain, cert_fingerprint,
|
||||
# wireguard_address?} objects. Consumed by hive-agent::identity::peers()
|
||||
# + the dashboard's peer_hives StateSnapshot field (P33RS tab).
|
||||
# `cert_fingerprint` is null for CA-trusted hives; `wireguard_address`
|
||||
# is omitted when not part of the mesh.
|
||||
#
|
||||
# Reads `peerHives` — `swarm.hives` minus this hive — so the "not me"
|
||||
# filter is the one derived in ../swarm.nix rather than a fifth copy.
|
||||
HYPERHIVE_PEERS = builtins.toJSON (
|
||||
lib.mapAttrsToList (
|
||||
_name: p:
|
||||
{
|
||||
inherit (p) domain;
|
||||
cert_fingerprint = p.certFingerprint;
|
||||
}
|
||||
// lib.optionalAttrs (p.wireguardAddress != null) {
|
||||
wireguard_address = p.wireguardAddress;
|
||||
}
|
||||
) config.services.hyperhive.swarm.peerHives
|
||||
);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -849,7 +849,24 @@ in
|
|||
};
|
||||
# `FORGEJO_CUSTOM` (not `GITEA_CUSTOM` — forgejo renamed it)
|
||||
# is how the CLI finds the app.ini upstream's module wrote.
|
||||
environment.FORGEJO_CUSTOM = "/var/lib/forgejo/custom";
|
||||
#
|
||||
# `SSL_CERT_FILE` for the same reason `forgejo.service` has it,
|
||||
# and it was missing here: registering the login source makes an
|
||||
# **outbound HTTPS call** — the CLI fetches
|
||||
# `<issuer>/.well-known/openid-configuration` to validate the
|
||||
# provider before writing the row. That URL is a swarm service
|
||||
# name served under the swarm CA, which the default system store
|
||||
# has never heard of, so without this the unit fails every single
|
||||
# time with `x509: certificate signed by unknown authority` and no
|
||||
# restart can help it.
|
||||
#
|
||||
# The trust belongs to every process that makes the call, not to
|
||||
# the service that happens to be the obvious consumer. Same
|
||||
# binary, same host, different unit — and only one of them had it.
|
||||
environment = {
|
||||
FORGEJO_CUSTOM = "/var/lib/forgejo/custom";
|
||||
}
|
||||
// lib.optionalAttrs useSelfSigned { SSL_CERT_FILE = forgeCaBundle; };
|
||||
path = [
|
||||
cfg.package
|
||||
pkgs.coreutils
|
||||
|
|
|
|||
|
|
@ -242,10 +242,6 @@ in
|
|||
config.security.acme.certs."example.com".directory;
|
||||
```
|
||||
|
||||
When using an external CA cert, other hives can declare this
|
||||
one in `services.hyperhive.swarm.hives` without
|
||||
`certFingerprint` — the standard CA bundle validates.
|
||||
|
||||
Mutual exclusion with `tls.acme.enable` — set one or the other,
|
||||
not both.
|
||||
'';
|
||||
|
|
@ -301,10 +297,6 @@ in
|
|||
};
|
||||
};
|
||||
```
|
||||
|
||||
After enabling, this hive's entry in `swarm.hives` can omit
|
||||
`certFingerprint` — Let's Encrypt certs are CA-trusted
|
||||
by default.
|
||||
'';
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -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,35 @@ 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 — the full directory
|
||||
# (this daemon has no "self" hive to exclude, unlike
|
||||
# `swarm.peerHives`, `swarm.hives` minus this hive) 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;
|
||||
};
|
||||
};
|
||||
}
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ in
|
|||
|
||||
services.hyperhive.swarm.hives.<name> = {
|
||||
domain = "<the old attrset key>";
|
||||
# certFingerprint / wireguard* carry over unchanged
|
||||
# wireguard* carries over unchanged
|
||||
};
|
||||
|
||||
Still set: ${lib.concatStringsSep ", " (lib.attrNames peers)}
|
||||
|
|
@ -61,9 +61,7 @@ in
|
|||
(services.hyperhive.swarm.ca — see docs/swarm/ca.md): every hive
|
||||
under it chains to it, so a per-hive CA is dead weight. What this
|
||||
genuinely drops is trusting a hive whose root this swarm does NOT
|
||||
own — another swarm's, or one keeping its own CA. certFingerprint
|
||||
does not cover that: it pins a leaf for hive-c0re's own HTTPS
|
||||
checks and does not reach Matrix federation.
|
||||
own — another swarm's, or one keeping its own CA.
|
||||
|
||||
Still set on: ${lib.concatStringsSep ", " withCaCert}
|
||||
'';
|
||||
|
|
|
|||
|
|
@ -1,9 +1,9 @@
|
|||
# The WireGuard inter-hive mesh for the local host. Split out of
|
||||
# ./swarm.nix because the two are different concerns with different
|
||||
# audiences: that file declares WHO the peers are (data hive-c0re
|
||||
# serialises into HYPERHIVE_PEERS and the dashboard renders), while
|
||||
# this one is plain host networking that a machine which runs no hive
|
||||
# at all --- the snapshot store, for one --- still needs.
|
||||
# audiences: that file declares WHO the peers are (consumed by
|
||||
# swarm-controller's hive roster and, here, the mesh), while this one
|
||||
# is plain host networking that a machine which runs no hive at all
|
||||
# --- the snapshot store, for one --- still needs.
|
||||
#
|
||||
# The two stay coupled by data, not by structure: the per-peer
|
||||
# `wireguard*` fields live on the peer submodule in ./swarm.nix, since
|
||||
|
|
@ -80,9 +80,9 @@
|
|||
# networking, not a c0re feature: a swarm host that runs no hive —
|
||||
# the snapshot store, for one — still has to join the mesh, and under
|
||||
# the old `c0re.enable` gate it silently got no `wg-hive` interface
|
||||
# at all. Nothing below is c0re-specific; the peer data
|
||||
# c0re consumes (HYPERHIVE_PEERS / HIVE_PEER_CA_PATHS) is rendered in
|
||||
# ./hive-c0re and stays gated there.
|
||||
# at all. Nothing below is c0re-specific; the peer data c0re consumes
|
||||
# (HIVE_PEER_CA_PATHS) is rendered in ./hive-c0re and stays gated
|
||||
# there.
|
||||
config = lib.mkIf config.services.hyperhive.swarm.wireguard.enable {
|
||||
assertions = [
|
||||
{
|
||||
|
|
|
|||
|
|
@ -10,11 +10,11 @@
|
|||
# could hold different endpoints for the same third hive and nothing
|
||||
# detected it. One entry per hive makes that unrepresentable.
|
||||
#
|
||||
# Consumed by hive-c0re's environment (HYPERHIVE_PEERS — see
|
||||
# ./hive-c0re), identity.rs + the dashboard's P33RS tab, and the mesh in
|
||||
# ./swarm-wireguard.nix. The mesh lives there rather than here because
|
||||
# bringing up an interface is host networking rather than swarm
|
||||
# bookkeeping, and a host that runs no hive still needs it.
|
||||
# Consumed by swarm-controller's own hive directory (its `/api/hives`,
|
||||
# swarm-controller.nix) and the mesh in ./swarm-wireguard.nix. The mesh
|
||||
# lives there rather than here because bringing up an interface is
|
||||
# host networking rather than swarm bookkeeping, and a host that runs
|
||||
# no hive still needs it.
|
||||
{
|
||||
lib,
|
||||
config,
|
||||
|
|
@ -75,8 +75,10 @@ in
|
|||
defaultText = lib.literalExpression ''"''${name}.''${services.hyperhive.swarm.domain}"'';
|
||||
example = "lab.example.com";
|
||||
description = ''
|
||||
Public DNS domain this hive occupies — used for dashboard
|
||||
links, peer HTTPS checks and Matrix federation discovery.
|
||||
Public DNS domain this hive occupies — used for
|
||||
swarm-controller's hive roster, agent identity
|
||||
(qualified `agent@domain` labels), and Matrix federation
|
||||
discovery.
|
||||
|
||||
Defaults to `<name>.<swarm.domain>`, the convention every
|
||||
hive in a swarm follows, so a conventional directory is
|
||||
|
|
@ -85,33 +87,6 @@ in
|
|||
'';
|
||||
};
|
||||
|
||||
certFingerprint = lib.mkOption {
|
||||
type = lib.types.nullOr lib.types.str;
|
||||
default = null;
|
||||
example = "sha256:b1946ac92492d2347c6235b4d2611184a3f5b6cae6c19d6e3c2f0a8e7d4c9f12";
|
||||
description = ''
|
||||
Expected TLS certificate fingerprint for this hive's HTTPS
|
||||
endpoint. Null = trust the CA bundle — which for a hive
|
||||
inside the swarm CA hierarchy is the normal case, since
|
||||
every hive under the swarm root already chains to it.
|
||||
Set it to pin a leaf that no CA in the bundle vouches for.
|
||||
|
||||
Format: the literal `sha256:` followed by exactly 64
|
||||
hex digits (case-insensitive, no colon separators) — the
|
||||
SHA-256 digest of the DER-encoded leaf certificate.
|
||||
Generate with `openssl x509 -noout -fingerprint -sha256`,
|
||||
then strip the colons and prepend `sha256:`. A malformed
|
||||
value is ignored with a warning rather than weakening
|
||||
trust. See docs/swarm/README.md for the full recipe.
|
||||
|
||||
Scopes only to hive-c0re's own peer HTTPS checks — it does
|
||||
NOT help Matrix federation, which validates against the
|
||||
container's trust bundle. There is no per-hive CA field to
|
||||
cover that case any more: the swarm root is the trust path
|
||||
(see ./swarm-ca.nix).
|
||||
'';
|
||||
};
|
||||
|
||||
wireguardPublicKey = lib.mkOption {
|
||||
type = lib.types.nullOr lib.types.str;
|
||||
default = null;
|
||||
|
|
@ -146,9 +121,9 @@ in
|
|||
description = ''
|
||||
IP address (with prefix) of this hive's host on the
|
||||
WireGuard mesh. Used as the `allowedIPs` for its
|
||||
WireGuard config entry and injected into `HYPERHIVE_PEERS`
|
||||
so hive-c0re can route intra-swarm traffic to the mesh
|
||||
address rather than the public domain. Required to include
|
||||
WireGuard config entry (`./swarm-wireguard.nix`), so
|
||||
intra-swarm traffic can route over the mesh address
|
||||
rather than the public domain. Required to include
|
||||
a hive in the mesh (entries missing this field are
|
||||
silently excluded from `wg-hive`).
|
||||
'';
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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,13 +104,17 @@ struct AppState {
|
|||
/// Loaded once at startup (`load_links`); same synchronization story
|
||||
/// as `hives`.
|
||||
links: Arc<Vec<ServiceLink>>,
|
||||
/// `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<Arc<status::StatusReader>>,
|
||||
}
|
||||
|
||||
/// Env var the controller's NixOS module sets from
|
||||
/// `services.hyperhive.swarm.hives`, JSON-encoded — same shape hive-c0re
|
||||
/// already builds for `HYPERHIVE_PEERS`
|
||||
/// (nix/host-modules/hive-c0re/environment.nix), just the full directory
|
||||
/// (this daemon has no "self" to exclude) rather than peers-minus-self.
|
||||
/// `services.hyperhive.swarm.hives`, JSON-encoded — the full directory
|
||||
/// (this daemon has no "self" to exclude) rather than peers-minus-self
|
||||
/// (`services.hyperhive.swarm.peerHives`, which other consumers use).
|
||||
const HIVES_ENV: &str = "SWARM_CONTROLLER_HIVES";
|
||||
|
||||
/// Parses [`HIVES_ENV`] into the swarm's hive directory. Unset or
|
||||
|
|
@ -192,6 +199,59 @@ async fn get_links(State(state): State<AppState>) -> Json<Vec<ServiceLink>> {
|
|||
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::HiveStatus>),
|
||||
(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<AppState>,
|
||||
) -> Result<Json<Vec<status::HiveStatus>>, 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 +290,51 @@ 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) => {
|
||||
// NOT "connected": `retry_on_initial_connect` returns a client
|
||||
// before any connection has been established, so claiming a
|
||||
// connection here would put "connected to the swarm queue" in
|
||||
// the journal moments before every request 503s with "not
|
||||
// connected" — and a reader would rightly distrust the second
|
||||
// line rather than the first. The connection's real state is
|
||||
// reported by the status endpoint, which checks it per request.
|
||||
tracing::info!("swarm queue configured; connecting in the background");
|
||||
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::<AppState>::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
|
||||
|
|
|
|||
206
swarm-controller/src/queue.rs
Normal file
206
swarm-controller/src/queue.rs
Normal file
|
|
@ -0,0 +1,206 @@
|
|||
//! 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.<swarm>/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/<pid>/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<Option<Self>> {
|
||||
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<String> {
|
||||
// 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<async_nats::Client> {
|
||||
// A timeout, because this client runs INSIDE the auth callback: a token
|
||||
// endpoint that accepts the connection and then never answers would hang
|
||||
// the callback, and with it the connection attempt that invoked it, with
|
||||
// no retry and nothing in the log to say why. Failing fast lets
|
||||
// `async-nats` do what it already does well — back off and try again.
|
||||
// 10s is generous for a form POST to a local IdP.
|
||||
let http = reqwest::Client::builder()
|
||||
.timeout(std::time::Duration::from_secs(10))
|
||||
.build()
|
||||
.context("building the token-endpoint HTTP client")?;
|
||||
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()
|
||||
);
|
||||
}
|
||||
}
|
||||
543
swarm-controller/src/status.rs
Normal file
543
swarm-controller/src/status.rs
Normal file
|
|
@ -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<String>,
|
||||
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<i64>,
|
||||
/// 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<u64>,
|
||||
/// 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<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// 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<serde_json::Value>,
|
||||
}
|
||||
|
||||
/// 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<async_nats::jetstream::kv::Store>,
|
||||
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::<u64>().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<Vec<HiveStatus>> {
|
||||
// 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<String, Offered> = 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<String, Offered>,
|
||||
now: SystemTime,
|
||||
stale_after: Duration,
|
||||
) -> Vec<HiveStatus> {
|
||||
let mut rows: Vec<HiveStatus> = 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<String>,
|
||||
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<HiveEntry> {
|
||||
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<String, Offered> {
|
||||
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);
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue