vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:dd9e0bf0b07b8a0f29b5c7b83b429d5334145f49
Branches Tags
vinzenz:main
..
compare: vinzenz:513554fe9a86d8957d6498bccd2e72a736310292
Branches Tags
vinzenz:main
12 changed files with 76 additions and 706 deletions
Download patch file Download diff file Expand all files Collapse all files

9
docs/networking/gateway.md
View file

@ -169,13 +169,6 @@ The gateway always terminates TLS — self-signed is the implicit floor when
nothing else is configured, so there is no http-only mode. Three modes,
selected by which (if any) external TLS source is set:
⚠️ **One vhost class is exempt, so expect it during a TLS audit.** A vhost bound to loopback for a
local consumer carries a single plain-HTTP listen and no TLS. `grafana-metrics`
(`nix/host-modules/swarm-grafana.nix`) is the only one in the tree today: it listens on `127.0.0.1`
alone and serves one `= /metrics` location from grafana's unix socket, for the collector on this
host to scrape. Nothing off-host can reach it, so TLS there protects nothing. Every vhost with a
routable name is covered by the modes below.
| mode | config | cert source | `.well-known` scheme |
|---|---|---|---|
| self-signed (default) | neither `tls.certDir` nor `tls.acme` set | host hive-CA signs a gateway leaf (RSA-4096) | `https` |
@ -209,7 +202,7 @@ Mutual exclusion: `tls.certDir` set together with `tls.acme.enable = true` fails
### Self-signed TLS (default)
On by default, and listens on `httpsPort` (default 443) on every routable vhost beside the plain-http `port` (default 80). (The loopback-only exception is named under *TLS modes* above.)
On by default, and listens on `httpsPort` (default 443) on every vhost beside the plain-http `port` (default 80).
The issuer is a **host-held hive CA**, not a bare self-signed leaf. A host service (`hive-tls-ca.service`, from the `hive-tls` module) generates a long-lived CA (`services.hyperhive.deploy.hive-controller.tls.caValidityDays`, default ~20y) under `services.hyperhive.deploy.hive-controller.tls.stateDir` (default `/var/lib/hive-tls`), then signs a gateway **leaf** (`leafValidityDays`, default 30d) with it. `hive-gateway-self-signed-cert` then imports the leaf into nginx's state dir (`/var/lib/hive-gateway/tls/{cert,key}.pem`).

15
docs/scheduler/observability.md
View file

@ -1,9 +1,8 @@
# Observability (OpenTelemetry)
hyperhive has built-in support for exporting each agent container's telemetry to
any OTLP-compatible collector: Claude Code statistics — token usage, cost, tool
call counts — via Claude Code's built-in OpenTelemetry integration, and the
container's journal, forwarded by a collector running inside it.
hyperhive has built-in support for exporting per-agent Claude Code statistics —
token usage, cost, tool call counts — to any OTLP-compatible collector via
Claude Code's built-in OpenTelemetry integration.
This is a **hive-wide** setting: one switch in the host NixOS config enables it
for every agent container simultaneously. No per-agent opt-in or opt-out exists.
@ -68,14 +67,6 @@ that.** Treat it as a convenience for grouping dashboards, never as evidence of
which container produced a sample: any agent that can reach this hive's
collector can label its data as any other agent.
**Logs ride the same hop, and a journal carries more than a counter does.** Each
agent container forwards its own journal through this port — every unit in it at
`info` and above, not an allowlist. That's the harness, the MCP daemons and
whatever a tool call spawned, so command lines and error text now leave the
container where before only counts did. The trust boundary is unchanged (same
destination, same credential, and an agent could already send arbitrary OTLP);
what changes is how much detail leaves by default.
Worth spelling out, because two different hops are in play and only one of them
is getting a credential:

10
docs/tools/forge-cli.md
View file

@ -548,16 +548,12 @@ Create a pull request
PR health view: mergeable / CI / reviews
**Usage:** `hive-forge pr status [OPTIONS] [NUMBER]`
###### **Arguments:**
* `<NUMBER>` — PR number — full health view (mergeable, CI, reviews, last comment). Mutually exclusive with `--sha`
**Usage:** `hive-forge pr status [OPTIONS]`
###### **Options:**
* `--pr <PR>` — Same as passing the number positionally. Kept because this verb accepted only this spelling before the positional existed
* `--sha <SHA>` — Explicit commit sha (or ref) — CI-only fast path. Mutually exclusive with a PR number
* `--pr <PR>` — PR number — full health view (mergeable, CI, reviews, last comment). Mutually exclusive with `--sha`
* `--sha <SHA>` — Explicit commit sha (or ref) — CI-only fast path. Mutually exclusive with `--pr`

2
hive-forge-notify/src/notify.rs
View file

@ -18,7 +18,7 @@
//! **Multi-source**: always the internal Forgejo, plus github.com when the
//! agent has a PAT. Each source polls independently behind
//! [`Source`]; everything below is shared. Rationale
//! + host differences: `docs/integrations/github.md::Notifications`.
//! + host differences: `docs/integrations/forge.md::Sources`.
//!
//! Activation gates, self-notification filtering, body excerpt +
//! truncation + heading escape, wrapper formats (comment / review /

89
hive-forge/src/verbs/pr_status.rs
View file

@ -1,12 +1,12 @@
//! `pr status <n>` — one-stop PR health view: mergeable state, CI
//! `pr status --pr <n>` — one-stop PR health view: mergeable state, CI
//! checks, requested reviewers + review verdicts, and the last-comment
//! timestamp. `--sha <sha>` is a CI-only fast path for a raw commit.
//! Removes the need for raw `curl` to the statuses / reviews endpoints,
//! keeping forge access on the single `hive-forge` tool.
//!
//! Exit code is a merge-readiness verdict when given a PR: 0 only when
//! CI is green AND the PR is mergeable AND no review requests changes —
//! so it composes (`hive-forge pr status 42 && …`). `--sha` mirrors the
//! Exit code is a merge-readiness verdict for `--pr`: 0 only when CI is
//! green AND the PR is mergeable AND no review requests changes — so it
//! composes (`hive-forge pr status --pr 42 && …`). `--sha` mirrors the
//! CI verdict alone (0 = success).
use anyhow::{Context, Result, bail};
@ -21,42 +21,26 @@ use crate::verbs::{print_json, rfc3339};
pub struct Args {
/// PR number — full health view (mergeable, CI, reviews, last
/// comment). Mutually exclusive with `--sha`.
#[arg(value_name = "NUMBER", conflicts_with_all = ["pr", "sha"])]
number: Option<u64>,
/// Same as passing the number positionally. Kept because this verb
/// accepted only this spelling before the positional existed.
#[arg(long, conflicts_with = "sha")]
pr: Option<u64>,
/// Explicit commit sha (or ref) — CI-only fast path. Mutually
/// exclusive with a PR number.
/// exclusive with `--pr`.
#[arg(long)]
sha: Option<String>,
}
impl Args {
/// The PR to inspect, whichever spelling asked for it.
///
/// Named rather than inlined so the two spellings can be shown to
/// select the same PR: a test that only parses proves clap accepted
/// the positional, never that anything reads it.
fn target_pr(&self) -> Option<u64> {
self.number.or(self.pr)
}
}
/// # Errors
///
/// Returns an error if neither a PR number nor `--sha` is given, or any
/// forge GET fails (PR lookup, combined status, reviews, comments). A
/// Returns an error if neither `--pr` nor `--sha` is given, or any forge
/// GET fails (PR lookup, combined status, reviews, comments). A
/// not-ready verdict is NOT an error — it's reported and reflected in
/// the process exit code instead.
pub fn run(client: &Client, args: Args) -> Result<()> {
let repo = client.repo()?;
let target = args.target_pr();
match (target, args.sha) {
match (args.pr, args.sha) {
(Some(pr), _) => pr_status(client, repo, pr),
(None, Some(sha)) => sha_status(client, &sha),
(None, None) => bail!("pr status: pass a PR number (`pr status 42`) or --sha <sha>"),
(None, None) => bail!("pr-status: pass one of --pr <n> or --sha <sha>"),
}
}
@ -120,7 +104,7 @@ fn pr_status(client: &Client, repo: &str, pr: u64) -> Result<()> {
.head
.as_ref()
.and_then(|h| h.sha.clone())
.with_context(|| format!("pr status: PR #{pr} has no head.sha"))?;
.with_context(|| format!("pr-status: PR #{pr} has no head.sha"))?;
let requested: Vec<String> = pull
.requested_reviewers
@ -405,59 +389,6 @@ fn combined_json(sha: &str, state: &str, statuses: &[Value]) -> Value {
#[cfg(test)]
mod tests {
use super::*;
use clap::Parser;
/// `Args` is a flattened arg group, not a `Parser`, so parsing it
/// standalone needs a wrapper. Without one these cases could only be
/// written against the whole binary's clap tree.
#[derive(Parser)]
struct Wrap {
#[command(flatten)]
args: Args,
}
fn parse(argv: &[&str]) -> Result<Args, clap::Error> {
Wrap::try_parse_from(argv).map(|w| w.args)
}
/// The property: the number is accepted the way every other
/// `pr <verb>` takes it, and the older flag still works.
#[test]
fn the_pr_number_is_accepted_positionally_and_as_a_flag() {
assert_eq!(parse(&["x", "42"]).unwrap().number, Some(42));
assert_eq!(parse(&["x", "--pr", "42"]).unwrap().pr, Some(42));
}
/// And something reads it. Parsing only proves clap accepted the
/// positional — dropping it from the selection leaves every other
/// case here green while `pr status 42` reports "pass a PR number".
#[test]
fn both_spellings_select_the_same_pr_and_sha_selects_none() {
assert_eq!(parse(&["x", "42"]).unwrap().target_pr(), Some(42));
assert_eq!(parse(&["x", "--pr", "42"]).unwrap().target_pr(), Some(42));
assert_eq!(parse(&["x", "--sha", "abc"]).unwrap().target_pr(), None);
}
/// Control for the case above. Clap accepting an argument and clap
/// ignoring it look identical from a passing parse, so the conflicts
/// have to be shown to bite.
#[test]
fn a_pr_number_conflicts_with_sha_and_with_the_flag() {
assert!(parse(&["x", "42", "--sha", "abc"]).is_err());
assert!(parse(&["x", "42", "--pr", "42"]).is_err());
}
/// Giving neither is a runtime error from `run`, not a parse error —
/// so `--sha` alone has to keep parsing.
#[test]
fn neither_form_parses_and_sha_alone_still_does() {
let empty = parse(&["x"]).unwrap();
assert!(empty.number.is_none() && empty.pr.is_none() && empty.sha.is_none());
assert_eq!(
parse(&["x", "--sha", "abc"]).unwrap().sha.as_deref(),
Some("abc")
);
}
#[test]
fn short_sha_truncates() {

12
hive-priv/src/main.rs
View file

@ -3019,17 +3019,7 @@ async fn sync_agent_tmpfiles(agents: &[AgentTmpfilesEntry]) -> Result<(String, S
// Parent dirs — created with permissive mode so hive-c0re can make subdirs.
// /run/hyperhive itself is also a RuntimeDirectory of hive-c0re.service; the
// tmpfiles.d entry here ensures it exists before hive-c0re starts (boot race).
//
// 0751, not 0750: must match hive-c0re.service's own `RuntimeDirectoryMode`
// and `docs/trust-boundary/boundary.md` — the extra `--x` on `other` is what
// lets a `hive-admin`-only user (no `hive-core` membership) traverse into the
// directory to reach `host.sock`; without it that user gets a permission
// denied opening the socket despite correct group membership on the socket
// itself. A mismatch here isn't just cosmetic: this line is regenerated and
// re-applied on every agent spawn/destroy via `systemd-tmpfiles --create`,
// so a stale `0750` here actively re-asserts the old, wrong mode far more
// often than a reboot does.
content.push_str("d /run/hyperhive 0751 hive-core hive-core -\n");
content.push_str("d /run/hyperhive 0750 hive-core hive-core -\n");
writeln!(content, "d {AGENT_RUNTIME_ROOT} 0755 hive-core hive-core -").ok();
// `hive-core`, not root: c0re does the `create_dir_all` for a new agent's
// subdir itself, so a root-owned parent EACCESes on the first spawn of a

112
nix/agent-modules/otel.nix
View file

@ -1,14 +1,6 @@
# OpenTelemetry wiring for an agent container: the per-agent options the
# meta flake injects (host-driven from `services.hyperhive.otel.*`), the
# environment every OTLP producer inside the container reads, and the
# collector that forwards this container's journal to the hive.
#
# Two directions, one endpoint. Metrics are PUSHED by processes in here
# that speak OTLP themselves; logs are not pushed by anyone, because
# nothing in a journal has an SDK. The forwarder is what turns the second
# into the first, and it has to run inside the container — a hive-level
# reader cannot see this journal at all (the host-side per-container
# directory is an id-mapped bind mount journald never writes into).
# meta flake injects (host-driven from `services.hyperhive.otel.*`), and
# the environment every OTLP producer inside the container reads.
#
# Claude Code is one producer here, not the owner. `hive-metric` — and
# any future in-container exporter — reads the same endpoint, protocol
@ -18,7 +10,6 @@
# options declared below.
{
lib,
pkgs,
config,
...
}:
@ -67,11 +58,6 @@ let
// lib.optionalAttrs (cfg.metricIntervalMs != null) {
OTEL_METRIC_EXPORT_INTERVAL = toString cfg.metricIntervalMs;
};
# The exporter's NAME picks the wire protocol, so `protocol` has to be
# read here as well as handed to the SDKs above — same expression as the
# swarm tier's upstream exporter in nix/host-modules/swarm-otel.nix.
logExporter = if cfg.protocol == "grpc" then "otlp" else "otlphttp";
in
{
# OTEL stats export is configured ONCE at host level via
@ -178,99 +164,5 @@ in
# while `NIX_REMOTE` — the same value, also set globally — is there.
systemd.globalEnvironment = producerEnv;
environment.variables = producerEnv;
# The journal this container writes is reachable from inside it and
# from nowhere else, which is why the forwarder runs here rather than
# the hive reading the container's journal from outside: the host-side
# per-container directory is an id-mapped bind mount and journald
# writes nothing into it.
assertions = [
{
# Sibling of the swarm collector's storage assertion in
# nix/host-modules/swarm-otel.nix, and it exists because the
# failure is silent at every layer: with a volatile journal the
# receiver below finds an empty directory, reads nothing, and the
# collector starts clean and stays healthy forever.
assertion =
!(lib.elem config.services.journald.storage [
"volatile"
"none"
]);
message = ''
hyperhive.otel.enable is on for agent ${userName}, but
services.journald.storage is
"${config.services.journald.storage}" in this container.
The log forwarder reads /var/log/journal, which journald only
writes when it stores persistently: with "volatile" the journal
lives in /run/log/journal and with "none" there is none at all.
Either way the forwarder would ship nothing while looking
healthy.
'';
}
];
services.opentelemetry-collector = {
enable = true;
# Contrib, and not a preference: `journald` is a contrib receiver.
# The upstream default build has no way to read a journal at all.
package = pkgs.opentelemetry-collector-contrib;
# Runs `otelcol validate` at build time. ⚠️ A parser, not a wiring
# check — it accepts a pipeline naming a component the build lacks,
# and the collector then dies at startup. Same caveat as both other
# tiers; see nix/host-modules/otel.nix for the measurement.
validateConfigFile = true;
settings = {
receivers.journald = {
# ⚠️ STATED, and it must stay stated: the receiver's own default
# is the RUNTIME journal (`/run/log/journal`), which in an agent
# container is empty — journald stores persistently here, so
# every entry is under /var/log/journal. Dropping this line
# leaves a collector that validates, starts, reports healthy and
# forwards nothing.
directory = "/var/log/journal";
# No `units` allowlist, unlike the swarm tier's receiver. That
# one needs one because the host's journal also holds an
# operator's own session; a container's journal is the harness
# and what the harness spawns, so there is no foreign traffic to
# filter out and an allowlist would only be a list to forget to
# update.
};
# Identity the hop above cannot supply. A host-side reader can say
# which machine a line came from; only a collector running as the
# agent can say which agent. `hive` and `swarm` are deliberately
# NOT stamped here even though the env above carries them for
# in-process producers: the swarm tier upserts `hive` from the
# receiver that accepted the sample, precisely so the label comes
# from something the sender cannot write.
processors.resource.attributes = [
{
key = "service.name";
value = "hyperhive-agent";
action = "upsert";
}
{
key = "agent";
value = userName;
action = "upsert";
}
];
# The hive's collector, at the same base address every in-process
# producer already exports to. `endpoint` is a BASE the exporter
# appends `/v1/logs` to — the path an OTLP/HTTP receiver serves.
exporters.${logExporter} = {
endpoint = cfg.endpoint;
}
// lib.optionalAttrs (cfg.protocol == "http/json") { encoding = "json"; };
service.pipelines.logs = {
receivers = [ "journald" ];
processors = [ "resource" ];
exporters = [ logExporter ];
};
};
};
};
}

24
nix/host-modules/otel.nix
View file

@ -532,30 +532,6 @@ in
];
exporters = [ swarmName ];
};
# The second signal this tier carries, and the reason it exists
# is that an agent's journal is unreadable from here: the
# host-side per-container journal directory is an id-mapped bind
# mount journald never writes into, so each agent container runs
# its own forwarder (nix/agent-modules/otel.nix) and pushes over
# the `otlp` receiver above. Without this pipeline that receiver
# answers 404 on `/v1/logs` — measured, and indistinguishable
# from a route that was never meant to exist.
#
# One receiver, two pipelines: `otlp` accepts both signals and
# each pipeline takes the one it names. The exporter needs no
# change either — its `endpoint` is a base, and otlphttp appends
# `/v1/logs` to it exactly as it appends `/v1/metrics`.
service.pipelines.logs = {
receivers = [ "otlp" ];
# `resourcedetection` and NOT `deltatocumulative`: that one is
# a metrics-only processor, and naming it here is a startup
# failure rather than a no-op. Host identity applies to both
# signals for the same reason — a line is only comparable
# across hives once it says which machine collected it.
processors = [ "resourcedetection" ];
exporters = [ swarmName ];
};
}
// lib.optionalAttrs senderAuth {
extensions.${authName} = {

22
nix/host-modules/swarm-otel.nix
View file

@ -1255,28 +1255,6 @@ in
exporters = exporterNames;
};
}
# A hive's logs, and the receiver is the SAME one its metrics
# arrive on: a receiver is not per-signal, so `otlp/${h}`
# feeds this pipeline and `metrics/${h}` above with no second
# port, credential or audience. Which is also why the `hive`
# label stays trustworthy for logs without any new mechanism
# — it still comes from which receiver accepted the sample.
#
# The senders are agent containers, one forwarder each
# (nix/agent-modules/otel.nix), via their own hive's collector.
# Without this pipeline that whole path terminates here: the
# receiver accepts the push, answers 200, and the records go
# nowhere.
// lib.optionalAttrs collectLogs (
lib.mapAttrs' (
h: _:
lib.nameValuePair "logs/${h}" {
receivers = [ "otlp/${h}" ];
processors = [ "resource/${h}" ];
exporters = logExporterNames;
}
) hivePorts
)
# Logs fan out exactly as metrics do: the swarm's store when it
# runs, the operator's upstream when one is configured, both
# when both. The local store is a destination rather than the

243
nix/module-eval.nix
View file

@ -58,32 +58,6 @@ let
];
}).config;
# The other half of the tree. `nix/agent-modules/` is evaluated by nothing
# else in this suite — every fixture above is a host — so a rendered
# container config was only ever read by a real deploy. Same entry point
# the meta flake hands a container, so what this evaluates is what an
# agent gets.
#
# Note `hyperhive`, not `services.hyperhive`: an agent container's options
# live at the top level.
agent =
extra:
(nixosSystem {
system = pkgs.stdenv.hostPlatform.system;
modules = [
self.nixosModules.agent-base
{
fileSystems."/" = {
device = "/dev/null";
fsType = "tmpfs";
};
boot.loader.grub.enable = false;
system.stateVersion = "25.11";
hyperhive = lib.recursiveUpdate { user.name = "a1"; } extra;
}
];
}).config;
allLocal = hive { deploy.singleHostSwarm = true; };
bare = hive { };
withCi = hive { deploy.forgejo.ci.enable = true; };
@ -399,45 +373,6 @@ let
swarm.otel.clientSecretFile = "/var/lib/swarm-otel-oidc/by-hand.secret";
};
# The log path's three hops, one fixture each. Nothing carries a journal
# record end to end at eval time, so what these defend is the part no tier
# can check for itself: each hop's output is the next hop's input, and
# every mismatch between them is silent — a push accepted and routed
# nowhere, a receiver pointed at an empty directory, a pipeline that does
# not exist.
agentBridge = "http://10.42.0.1:4318";
agentOtel = agent {
otel.enable = true;
otel.endpoint = agentBridge;
};
# The same agent over the other wire protocol. An exporter's NAME is what
# selects it, so this is where a defined exporter and the pipeline's
# reference to it can drift apart.
agentOtelGrpc = agent {
otel.enable = true;
otel.endpoint = agentBridge;
otel.protocol = "grpc";
};
agentNoOtel = agent { };
agentSettings = machine: machine.services.opentelemetry-collector.settings;
# This hive's own collector, which is a HOST service — unlike the swarm
# tier's, which lives in a container and is read through `otelSettings`.
hiveOtel = hive {
otel.enable = true;
otel.clientSecretFile = "/var/lib/hive-otel-oidc/client.secret";
};
hiveOtelPipelines = hiveOtel.services.opentelemetry-collector.settings.service.pipelines;
# Two hives in the roster, which no other fixture here has: every one of
# them declares `swarm.hives.h1` alone, so a per-hive arm written against
# one of those passes on a hardcoded literal.
otelTwoHives = hive {
deploy.swarm-otel.enable = true;
deploy.authelia.enable = true;
swarm.hives.h2.domain = "h2.t.local";
};
# A priority collision is a property of the *option*, not
# of the merged value's interior — nix throws the moment the value is
# demanded at all, so `seq`-ing each `serviceConfig` value to WHNF is
@ -453,85 +388,8 @@ let
in
builtins.foldl' (acc: v: builtins.seq v acc) true vals;
# The swarm's shared services hosted HERE without the all-local mode — a
# services box with hives elsewhere, the shape ./host-modules/
# swarm-required-services.nix documents the switch for.
swarmServicesHere = hive { deploy.allSwarmServices = true; };
# Same, with one of those services placed on another host. Every derivation
# in that module is `mkDefault` so this stays expressible.
swarmServicesBaoElsewhere = hive {
deploy.allSwarmServices = true;
deploy.bao.enable = false;
};
# The enables that switch owns. ⚠️ `otel` is the per-hive collector's own
# option and is NOT under `deploy` — spelled at the wrong path it would be
# undeclared rather than false, and a roster that quietly loses a member is
# what the count guard in the cases below exists to catch. Its membership
# here is deliberate and was the fix for a gap, not an oversight: the hive
# tier lands wherever the swarm services do.
swarmServiceEnables =
machine:
let
h = machine.services.hyperhive;
in
{
matrix = h.deploy.matrix.enable;
otel = h.otel.enable;
authelia = h.deploy.authelia.enable;
nats = h.deploy.nats.enable;
swarm-otel = h.deploy.swarm-otel.enable;
victoriametrics = h.deploy.victoriametrics.enable;
grafana = h.deploy.grafana.enable;
victorialogs = h.deploy.victorialogs.enable;
bao = h.deploy.bao.enable;
};
# Each case: a name stating the property, and `ok`.
cases = [
{
# `lib.all` over an empty set holds vacuously, so the roster is counted
# before it is read: a helper that lost a member would otherwise turn
# this case green by measuring nothing.
name = "hosting the swarm's shared services turns on every service that switch owns";
ok =
let
es = swarmServiceEnables swarmServicesHere;
in
lib.length (lib.attrNames es) == 9 && lib.all lib.id (lib.attrValues es);
}
{
name = "a hive that does not host the swarm's shared services runs none of them";
ok =
let
es = swarmServiceEnables bare;
in
lib.length (lib.attrNames es) == 9 && !lib.any lib.id (lib.attrValues es);
}
{
# The switch fills in for an operator who has not spoken and yields to
# one who has — that is what keeps a shared service placeable on a host
# of its own. A plain assignment or `mkForce` would satisfy both cases
# above and break this one. `nats` is the control: without it the case
# also passes on a fixture where nothing came on at all.
name = "placing one shared service elsewhere survives the switch that would enable it";
ok =
let
es = swarmServiceEnables swarmServicesBaoElsewhere;
in
!es.bao && es.nats;
}
{
# The controller sits on the OTHER tier: `singleHostSwarm` places it
# (./host-modules/local-defaults.nix) and swarm-ui follows the
# controller. Pinned so that moving a service between tiers is a
# decision someone makes rather than a merge nobody reads.
name = "hosting the swarm's shared services does not make a hive the swarm's control plane";
ok =
!swarmServicesHere.services.hyperhive.deploy.swarm-controller.enable
&& !swarmServicesHere.services.hyperhive.deploy.swarm-ui.enable;
}
{
name = "a hive that has not opted into all-local runs no swarm controller";
ok = !bare.services.hyperhive.deploy.swarm-controller.enable;
@ -1164,107 +1022,6 @@ let
in
named != [ ] && lib.all (a: builtins.elem a s.service.extensions) named;
}
{
# The journald receiver's own default directory is the RUNTIME
# journal, and a container that stores persistently leaves that
# empty. At the default the forwarder validates, starts, reports
# healthy and ships nothing, so this one literal is the difference
# between the path working and silently not.
name = "the agent forwarder reads the persistent journal, not the runtime one";
ok = ((agentSettings agentOtel).receivers.journald.directory or null) == "/var/log/journal";
}
{
# The hop's two ends: what it reads, and where what it reads goes.
# The endpoint is compared against the value the fixture handed the
# option rather than a literal spelled here, so an exporter that
# stopped reading the option fails instead of matching a constant
# that travelled beside it.
name = "the agent forwarder ships the journal to the endpoint its hive gave it";
ok =
let
s = agentSettings agentOtel;
p = s.service.pipelines.logs;
in
p.receivers == [ "journald" ]
&& p.exporters != [ ]
&& lib.all (e: (s.exporters ? ${e}) && s.exporters.${e}.endpoint == agentBridge) p.exporters;
}
{
# Presence control for the two cases above: with the switch off there
# is no collector in the container at all, so their passing is about
# the wiring rather than about a unit that renders regardless.
name = "an agent that has not opted into telemetry runs no collector";
ok = !agentNoOtel.services.opentelemetry-collector.enable;
}
{
# `otlp` and `otlphttp` are different components and the protocol
# option picks which one is defined. A pipeline left naming the other
# is a startup failure; an exporter no pipeline names is silence.
name = "the agent forwarder's exporter and its pipeline agree on the protocol";
ok =
let
s = agentSettings agentOtelGrpc;
in
(s.exporters ? otlp) && s.service.pipelines.logs.exporters == [ "otlp" ];
}
{
# The tier in the middle. Its OTLP receiver takes both signals on one
# port, so without this pipeline an agent's push is answered 404 on
# `/v1/logs` — and a forwarder retrying into a 404 is indistinguish-
# able from one with nothing to send. Compared against the metrics
# pipeline's exporters rather than a name, so the two signals cannot
# drift to different destinations.
name = "the hive collector forwards logs upstream, not only metrics";
ok =
(hiveOtelPipelines ? logs)
&& hiveOtelPipelines.metrics.exporters != [ ]
&& hiveOtelPipelines.logs.receivers == [ "otlp" ]
&& hiveOtelPipelines.logs.exporters == hiveOtelPipelines.metrics.exporters;
}
{
# `deltatocumulative` is metrics-only: naming it in a logs pipeline
# kills the collector at startup rather than doing nothing. The second
# clause is the control — the metrics pipeline still names it, so a
# pass means the two processor lists differ rather than that the
# processor left the module.
name = "the hive collector keeps the metrics-only processor out of its logs pipeline";
ok =
!(builtins.elem "deltatocumulative" hiveOtelPipelines.logs.processors)
&& builtins.elem "deltatocumulative" hiveOtelPipelines.metrics.processors;
}
{
# Read against the roster the fixture declares rather than against
# names spelled here: an arm naming `h1` passes on a single-hive
# config however the mapping is written. The length clause is what
# makes the `all` mean anything — over an empty roster it holds
# vacuously.
name = "the swarm collector routes every hive's logs, not just one";
ok =
let
p = (otelSettings otelTwoHives).service.pipelines;
hives = lib.attrNames otelTwoHives.services.hyperhive.swarm.hives;
in
lib.length hives == 2
&& lib.all (h: (p ? "logs/${h}") && p."logs/${h}".receivers == [ "otlp/${h}" ]) hives;
}
{
# The same split as the metrics case above — defining an exporter and
# naming it are two lists — plus the half one shared list cannot have:
# the metrics store's exporter renders perfectly well inside a logs
# pipeline and posts journal records at an ingest route that is not
# for them.
name = "every logs pipeline sends to the log store and to no metrics one";
ok =
let
s = otelSettings otelTwoHives;
logPipes = lib.filterAttrs (n: _: lib.hasPrefix "logs/" n) s.service.pipelines;
used = lib.unique (lib.concatMap (p: p.exporters) (lib.attrValues logPipes));
in
logPipes != { }
&& builtins.elem "otlphttp/victorialogs" used
&& !(builtins.elem "otlphttp/victoriametrics" used)
&& lib.all (e: s.exporters ? ${e}) used;
}
{
# The store's seal is spread over six gates — the stanza, the
# provisioning unit, two bind mounts, a device and an EnvironmentFile.

10
swarm-controller/src/main.rs
View file

@ -1625,11 +1625,13 @@ async fn main() -> Result<()> {
let hives = load_hives();
// Before serving, because a hive whose role does not exist cannot log in,
// and one whose policy does not exist logs in able to read nothing —
// either way it cannot collect what this daemon writes for it. A store
// that is not up yet is retried in the background instead of being
// abandoned, since the two of them boot together.
// either way it cannot collect what this daemon writes for it. Policy
// first: the role names the policy, so writing it the other way round
// leaves a window where the role points at nothing. Same "log and carry
// on" shape as every connect above.
let hive_names: Vec<String> = hives.iter().map(|h| h.name.clone()).collect();
read_policy::ensure_hive_access(hive_names).await;
read_policy::ensure_hive_policies(&hive_names).await;
read_policy::ensure_hive_roles(&hive_names).await;
let state = AppState {
hives: Arc::new(hives),

234
swarm-controller/src/read_policy.rs
View file

@ -6,167 +6,44 @@
//! without the policy the token it gets may read nothing — either way the read
//! half of a delivery answers 403.
//!
//! A pass runs at startup, and repeats until it reaches the store. The two
//! start together, so the store being the slower one is the ordinary case
//! rather than an edge one, and a pass that gave up there provisions nobody
//! until something restarts this daemon. Per-hive failures are different: they
//! are reported and skipped — one hive that cannot be provisioned should not
//! take down a daemon that serves everything else, and a store that answers is
//! answering the same way next time, so the next start is the retry for those.
//!
//! The policy document is the same for every hive and depends on nothing
//! ([`swarm_secret_client::policy`] explains why it is that wide), and the hive
//! list is loaded once because a config change means a redeploy.
use std::{future::Future, time::Duration};
//! Both are startup passes rather than hooks. The policy document is the same
//! for every hive and depends on nothing ([`swarm_secret_client::policy`]
//! explains why it is that wide), and the hive list is loaded once because a
//! config change means a redeploy. Per-hive failures are reported and skipped —
//! one hive that cannot be provisioned should not take down a daemon that
//! serves everything else, and the next start retries it.
use anyhow::{Context, Result};
use swarm_secret_client::{Error, SecretStore, client::DEFAULT_CERT_MOUNT, policy};
use swarm_secret_client::{SecretStore, client::DEFAULT_CERT_MOUNT, policy};
/// File holding the authority hives are issued from, as
/// `swarm-controller.nix` names it.
pub const ENV_HIVE_CLIENT_CA: &str = "SWARM_CONTROLLER_HIVE_CLIENT_CA_FILE";
/// How long to wait before re-attempting a pass that never reached the store.
/// Give every hive in `hives` the read policy its certificate will carry.
///
/// `nix/host-modules/swarm-bao.nix`'s bootstrap unit retries the other side of
/// this same race on the same cadence: the two halves of one race should not
/// disagree about how long a wait is worth.
const RETRY_INTERVAL: Duration = Duration::from_secs(30);
/// How many times [`RETRY_INTERVAL`] is waited before giving up. A store that
/// has not appeared within [`RETRY_WINDOW`] is absent rather than slow, and a
/// daemon logging into the void forever tells an operator nothing.
const RETRY_ATTEMPTS: u32 = 2_880;
/// What [`RETRY_ATTEMPTS`] at [`RETRY_INTERVAL`] come to. A named third
/// constant because it is the claim the other two make together, and either
/// one of them can be edited alone.
const RETRY_WINDOW: Duration = Duration::from_hours(24);
/// A pass that never got as far as the store, carrying what to say about it.
///
/// Returned rather than logged where it happens: the first attempt deserves a
/// warning and the two-hundredth is noise, and only the caller knows which one
/// this is.
struct Unreachable(String);
/// Give every hive in `hives` the access its certificate needs.
///
/// Returns once the first pass is done, so a store that is already up leaves
/// nothing deferred. When that pass cannot reach the store, it repeats in the
/// background for up to [`RETRY_WINDOW`].
pub async fn ensure_hive_access(hives: Vec<String>) {
if let Err(Unreachable(reason)) = provision(&hives).await {
tracing::warn!(
reason,
retry_in = ?RETRY_INTERVAL,
"hive provisioning could not reach the swarm secret store; retrying in the background"
);
spawn_retry(hives);
}
}
/// Repeat [`provision`] until it reaches the store, or [`RETRY_WINDOW`] passes.
fn spawn_retry(hives: Vec<String>) {
tokio::spawn(async move {
let reached = retry_until_ok(RETRY_INTERVAL, RETRY_ATTEMPTS, move || {
let hives = hives.clone();
async move {
let outcome = provision(&hives).await;
if let Err(Unreachable(reason)) = &outcome {
tracing::debug!(reason, "hive provisioning: store still unreachable");
}
outcome
}
})
.await;
if let Some(attempt) = reached {
tracing::info!(attempt, "hive provisioning reached the store");
} else {
tracing::warn!(
attempts = RETRY_ATTEMPTS,
window = ?RETRY_WINDOW,
"giving up on hive provisioning; no hive can log in to the store until this daemon is restarted"
);
}
});
}
/// Run `attempt` every `interval` until it succeeds, at most `attempts` times.
/// Answers with the attempt that succeeded, or `None` if none did.
///
/// The cadence and the bound are parameters rather than the constants above so
/// that a test can drive this loop without waiting out a real day.
async fn retry_until_ok<F, Fut>(interval: Duration, attempts: u32, mut attempt: F) -> Option<u32>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<(), Unreachable>>,
{
let mut ticker = tokio::time::interval(interval);
// `interval`'s first tick is immediate, and that pass is the one the caller
// already ran before deciding to retry at all.
ticker.tick().await;
for n in 1..=attempts {
ticker.tick().await;
if attempt().await.is_ok() {
return Some(n);
}
}
None
}
/// One pass: log in once, write every hive's policy, then every hive's role.
///
/// Policy first, because the role names the policy — the other order leaves a
/// window in which the role points at nothing.
///
/// # Errors
/// [`Unreachable`] for the states a later attempt can resolve: a store that is
/// not up, and an authority file that is not placed yet. A deployment that is
/// configured for neither is `Ok` — there is nothing to wait for.
async fn provision(hives: &[String]) -> Result<(), Unreachable> {
/// Does nothing when this deployment has no store to reach, which is the
/// shape a controller without a secret store runs in.
pub async fn ensure_hive_policies(hives: &[String]) {
if hives.is_empty() {
return Ok(());
return;
}
let store = match crate::store::connect().await {
Ok(store) => store,
// An unset `BAO_*` variable is a deployment that runs no store, not
// one whose store is late: asking again changes nothing.
Err(Error::MissingEnv(var)) => {
tracing::info!(
var,
"no secret store configured; hive access is not managed here"
);
return Ok(());
Err(e) => {
tracing::info!(reason = %e, "no secret store reachable; hive read policies are not managed here");
return;
}
Err(e) => return Err(Unreachable(e.to_string())),
};
for hive in hives {
if let Err(e) = write_policy_for(&store, hive).await {
if let Err(e) = write_one(&store, hive).await {
tracing::warn!(hive, error = %format!("{e:#}"), "writing this hive's read policy failed");
}
}
let Some(path) = std::env::var_os(ENV_HIVE_CLIENT_CA) else {
tracing::info!("no hive certificate authority configured; hive roles are not managed here");
return Ok(());
};
let path = path.to_string_lossy().into_owned();
// Retryable rather than fatal: the unit that places this file and this
// daemon start together, so "not there yet" is a moment, not a verdict.
let ca = std::fs::read_to_string(&path)
.map_err(|e| Unreachable(format!("reading the hive authority {path}: {e}")))?;
for hive in hives {
if let Err(e) = write_role(&store, hive, &ca).await {
tracing::warn!(hive, error = %format!("{e:#}"), "creating this hive's store role failed");
}
}
Ok(())
}
/// One hive's policy: its own name, the shared document.
async fn write_policy_for(store: &SecretStore, hive: &str) -> Result<()> {
async fn write_one(store: &SecretStore, hive: &str) -> Result<()> {
let name = policy::hive_object_name(hive)?;
store
.write_policy(&name, &policy::render())
@ -176,6 +53,38 @@ async fn write_policy_for(store: &SecretStore, hive: &str) -> Result<()> {
Ok(())
}
/// Create each hive's cert-auth role, so the certificate it already holds logs
/// in and carries the policy written above.
///
/// Does nothing when no authority is configured, which is the deployment that
/// has not onboarded a hive yet.
pub async fn ensure_hive_roles(hives: &[String]) {
let Some(path) = std::env::var_os(ENV_HIVE_CLIENT_CA) else {
tracing::info!("no hive certificate authority configured; hive roles are not managed here");
return;
};
let path = path.to_string_lossy().into_owned();
let ca = match std::fs::read_to_string(&path) {
Ok(ca) => ca,
Err(e) => {
tracing::warn!(path, error = %e, var = ENV_HIVE_CLIENT_CA, "reading the hive authority failed; hive roles are not created");
return;
}
};
let store = match crate::store::connect().await {
Ok(store) => store,
Err(e) => {
tracing::warn!(error = %e, "connecting to the swarm secret store failed; hive roles are not created");
return;
}
};
for hive in hives {
if let Err(e) = write_role(&store, hive, &ca).await {
tracing::warn!(hive, error = %format!("{e:#}"), "creating this hive's store role failed");
}
}
}
/// The role for one hive: its own certificate's subject, its own policy.
async fn write_role(store: &SecretStore, hive: &str, ca: &str) -> Result<()> {
let name = policy::hive_object_name(hive)?;
@ -188,48 +97,3 @@ async fn write_role(store: &SecretStore, hive: &str, ca: &str) -> Result<()> {
tracing::info!(hive, role = %name, "hive store role in place");
Ok(())
}
#[cfg(test)]
mod tests {
use std::{cell::Cell, time::Duration};
use super::{RETRY_ATTEMPTS, RETRY_INTERVAL, RETRY_WINDOW, Unreachable, retry_until_ok};
#[test]
fn the_retry_bound_is_the_window_it_claims() {
assert_eq!(RETRY_INTERVAL * RETRY_ATTEMPTS, RETRY_WINDOW);
}
/// The property this issue is about: a pass that fails because the store is
/// not up yet happens again.
#[tokio::test]
async fn a_pass_is_repeated_until_it_reaches_the_store() {
let calls = Cell::new(0_u32);
let reached = retry_until_ok(Duration::from_millis(1), 10, || {
calls.set(calls.get() + 1);
let still_down = calls.get() < 3;
async move {
if still_down {
Err(Unreachable("store not up".to_owned()))
} else {
Ok(())
}
}
})
.await;
assert_eq!(reached, Some(3));
assert_eq!(calls.get(), 3, "the loop stops at the first success");
}
#[tokio::test]
async fn the_loop_is_bounded_rather_than_endless() {
let calls = Cell::new(0_u32);
let reached = retry_until_ok(Duration::from_millis(1), 5, || {
calls.set(calls.get() + 1);
async { Err(Unreachable("store not up".to_owned())) }
})
.await;
assert_eq!(reached, None, "a store that never appears is given up on");
assert_eq!(calls.get(), 5, "one attempt per interval, and no more");
}
}