amadan
host: status loop — inert until configured, one publisher via the lease
Wires internal/status's CollectBox/WriteBox/BuildFleet/PublishFleet into `carlos agent` as a 5-minute timer beside adoptLoop. statusLoop is inert (one log line, no goroutine work) until CARLOS_STATUS_BUCKET is set — the same litestream/adoptLoop convention every other optional subsystem in this file uses. The pure per-tick logic is statusTick(ctx, store, reg, probes, publish), store-agnostic and lease-agnostic, so host_test.go exercises it against a DirStore + fixture registry with no S3, systemd, or timers involved. Each tick self-reports unconditionally (every box writes its own status/boxes/<label>.json snapshot every 5 minutes) and only merges + publishes the fleet-wide feed when it wins that tick's status-lease — lease.HeldError is normal contention and is swallowed silently, any other acquire error is logged like any other tick failure. Box-side probes (probes.go) read /etc/machine-id, /proc/meminfo, and /proc/net/unix — never dialing a hibernating tenant's socket, which would wake it — with every failure path returning nil rather than a guessed zero, matching status.Probes' "unknown is null, never a fabricated number" contract. There's no existing disk-floor config to mirror (checked ops/litestream), so DiskFrac is the plain 1 - avail/total fraction.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
3 files changed,
+350
−4
internal/host/agent.go+154 −4internal/host/host_test.go+61 −0internal/host/probes.go+135 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go| index cee0d6f36..a892dfb25 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -1,8 +1,38 @@ |
| // The `carlos agent` verb — the single-box role: edge (TLS + host routing) + |
| -// hibernation activator + hourly release adoption composed in one process. |
| -// The composition is new here, but each part is a port: the edge from |
| -// Eleven's router.go, the activator from lchat activator.go, and the adopt |
| -// loop is release.Pass on the cadence of Eleven's updater timer. |
| +// hibernation activator + hourly release adoption + the fleet status page's |
| +// 5-minute self-report composed in one process. The composition is new |
| +// here, but each part is a port: the edge from Eleven's router.go, the |
| +// activator from lchat activator.go, the adopt loop is release.Pass on the |
| +// cadence of Eleven's updater timer, and the status loop is |
| +// internal/status's CollectBox/WriteBox/BuildFleet/PublishFleet run on a |
| +// timer with one publisher elected via internal/lease. |
| +// |
| +// Fleet status page config surface (every env var any part of this file |
| +// reads for it, gathered in one place since no single function owns the |
| +// whole contract): |
| +// |
| +// - CARLOS_STATUS_BUCKET / CARLOS_STATUS_PREFIX / CARLOS_STATUS_REGION — |
| +// read by statusLoop. CARLOS_STATUS_BUCKET names the S3 bucket each |
| +// box writes its own status/boxes/<label>.json snapshot to and (when |
| +// it wins the lease) publishes status/fleet.json + the viewer assets |
| +// to; unset -> the loop never starts, logging one line and returning — |
| +// the same "inert until configured" convention adoptLoop uses for |
| +// CARLOS_DEPLOYMENT_BUCKET/_DIR, and litestream's own env-gated |
| +// replication before that. CARLOS_STATUS_PREFIX and CARLOS_STATUS_REGION |
| +// feed lease.Acquire (the single-writer election for the publish half |
| +// of each tick); both may be empty (region falls back to the AWS |
| +// default chain). |
| +// - STATUS_BOX_REGION — an operator-set, coarse region word (e.g. |
| +// "EU · Ireland") carried into status.Probes.Region and published |
| +// verbatim on this box's snapshot; "" omits the field. Deliberately |
| +// free text chosen by the operator, never derived from cloud metadata — |
| +// the public commitment is a human-readable region, not a specific |
| +// datacenter. |
| +// - STATUS_HOST / STATUS_PUBLIC — NOT read anywhere in this file. They |
| +// gate whether ops/tofu fronts the bucket's published status/ prefix |
| +// with a public hostname; recorded here only so the fleet status |
| +// page's entire config surface is documented in one place rather than |
| +// split across this file and the tofu module. |
| package host |
| import ( |
| @@ -20,8 +50,10 @@ import ( |
| "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/edge" |
| + "github.com/carlosframework/platform/internal/lease" |
| "github.com/carlosframework/platform/internal/registry" |
| "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/status" |
| ) |
| // dataDir resolves the fleet data directory (CARLOS_DATA_DIR, default /data). |
| @@ -72,6 +104,7 @@ func RunAgent(args []string) { |
| } |
| go adoptLoop(reg) |
| + go statusLoop(reg) |
| // The same serving loop as `carlos edge`, with the activator injected. |
| edge.RunWithActivator(args, act) |
| @@ -134,6 +167,123 @@ func adoptLoop(reg *registry.Registry) { |
| } |
| } |
| +// statusLoop runs the fleet status page's 5-minute tick, forever, once armed |
| +// — see the package doc comment above for the full env contract. Missing |
| +// CARLOS_STATUS_BUCKET disables the loop with one log line, the same |
| +// "a box with no status store is still a valid deployment" shape adoptLoop |
| +// uses for its own missing config. |
| +// |
| +// Each tick: gather fresh probes (small OS-reading funcs in probes.go, none |
| +// of which this loop's own tests call — statusTick below takes a |
| +// status.Probes value, so tests inject fixtures instead of touching /proc or |
| +// a syscall), then contend for the cross-box status-lease. Acquire succeeding |
| +// means this box is the one publisher for this tick: statusTick both writes |
| +// this box's own snapshot AND merges+publishes the fleet-wide feed, and the |
| +// lease is released again right after (held only for the publish, not across |
| +// the 5-minute gap — the next tick, possibly on a different box, re-acquires |
| +// fresh). A *lease.HeldError means another box already won this tick's |
| +// election: completely normal contention, not a failure, so it's swallowed |
| +// without a log line — every box hits this on every tick it doesn't win, and |
| +// logging that every 5 minutes forever would bury the log in noise. Any |
| +// OTHER acquire error (a real S3/credential problem) is logged, same as any |
| +// other tick failure. |
| +func statusLoop(reg *registry.Registry) { |
| + ctx := context.Background() |
| + bucket := os.Getenv("CARLOS_STATUS_BUCKET") |
| + if bucket == "" { |
| + log.Printf("agent: status loop disabled: CARLOS_STATUS_BUCKET not set") |
| + return |
| + } |
| + prefix := os.Getenv("CARLOS_STATUS_PREFIX") |
| + region := os.Getenv("CARLOS_STATUS_REGION") |
| + boxRegion := os.Getenv("STATUS_BOX_REGION") |
| + |
| + s, err := release.NewS3Store(ctx, bucket) |
| + if err != nil { |
| + log.Printf("agent: status loop disabled: %v", err) |
| + return |
| + } |
| + log.Printf("agent: status loop armed (5min + jitter)") |
| + |
| + // Jittered on the same shape as adoptLoop's delay: every box on the same |
| + // 5-minute cadence would all hit the lease and the bucket in the same |
| + // second otherwise. |
| + delay := time.Minute |
| + for { |
| + time.Sleep(delay) |
| + delay = 5*time.Minute + rand.N(time.Minute) |
| + |
| + p := status.Probes{ |
| + MachineID: readMachineID(), |
| + Region: boxRegion, |
| + MemFrac: readMemFrac(), |
| + DiskFrac: readDiskFrac(), |
| + Listening: listeningProbe, |
| + // LagSecs stays nil: litestream's per-DB replication lag isn't |
| + // surfaced up to the agent process anywhere yet — the |
| + // replication machinery (internal/lease's Replicator) lives |
| + // inside the activator, one level down, and doesn't currently |
| + // export a "seconds since last successful sync" reading this |
| + // loop could poll. Publishing a guess (0? the box's own clock |
| + // skew?) would be worse than the honest null CollectBox already |
| + // knows how to carry — wiring this up is a seam for later, not |
| + // a gap to paper over now. |
| + } |
| + |
| + held, acqErr := lease.Acquire(ctx, bucket, prefix+"/status-lease", region) |
| + publish := acqErr == nil |
| + if acqErr != nil { |
| + var heldErr *lease.HeldError |
| + if !errors.As(acqErr, &heldErr) { |
| + log.Printf("agent: status tick: lease acquire: %v", acqErr) |
| + } |
| + } |
| + |
| + if err := statusTick(ctx, s, reg, p, publish); err != nil { |
| + log.Printf("agent: status tick: %v", err) |
| + } |
| + |
| + if publish { |
| + if err := held.Release(ctx); err != nil { |
| + log.Printf("agent: status tick: lease release: %v", err) |
| + } |
| + } |
| + } |
| +} |
| + |
| +// statusTick is the fleet status page's pure per-tick work, extracted out of |
| +// statusLoop so it needs no systemd, S3, or timers to test: given an |
| +// already-built Store (DirStore in tests, the real S3Store in production) |
| +// and an already-decided publish flag (statusLoop's lease result; a fixed |
| +// true/false in tests), it does exactly what one tick does and nothing else. |
| +// The box's own snapshot is written unconditionally — every box self-reports |
| +// every tick regardless of who's publishing — and BuildFleet+PublishFleet |
| +// only run when publish is true, which for a DirStore-backed dev/test caller |
| +// with no lease in play at all just means "always true, no election needed" |
| +// (a lone box is its own only publisher). |
| +func statusTick(ctx context.Context, s release.Store, reg *registry.Registry, p status.Probes, publish bool) error { |
| + routes, err := reg.List() |
| + if err != nil { |
| + return err |
| + } |
| + now := time.Now() |
| + |
| + box := status.CollectBox(routes, p, now) |
| + if err := status.WriteBox(ctx, s, box); err != nil { |
| + return err |
| + } |
| + if !publish { |
| + return nil |
| + } |
| + |
| + const fleetPeriod = 5 * time.Minute // matches statusLoop's own tick cadence |
| + fleet, err := status.BuildFleet(ctx, s, routes, now, fleetPeriod) |
| + if err != nil { |
| + return err |
| + } |
| + return status.PublishFleet(ctx, s, fleet) |
| +} |
| + |
| // restartRoute cycles a route's process onto its freshly-repointed live |
| // symlink. Only systemd-managed routes need anything done here: hibernating |
| // hosts never reach this (release.Pass gives them the symlink-only path), and |
diff --git a/internal/host/host_test.go b/internal/host/host_test.go| index 4d38fafdb..d31937844 100644 |
| --- a/internal/host/host_test.go |
| +++ b/internal/host/host_test.go |
| @@ -1,11 +1,16 @@ |
| package host |
| import ( |
| + "context" |
| "net" |
| "os" |
| "path/filepath" |
| "strings" |
| "testing" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/status" |
| ) |
| // expectedUnits is the contract for what bootstrap installs — a new unit file |
| @@ -152,6 +157,62 @@ func TestBootstrapWritesUnitsToStagingDir(t *testing.T) { |
| } |
| } |
| +// ---- Task 8: statusTick — the loop's pure part ----------------------------- |
| + |
| +// TestStatusTickWritesBoxAndPublishesOnlyWhenAsked is the plan's Step-1 test: |
| +// statusTick against a DirStore + fixture registry must always land the |
| +// box's own snapshot under status/boxes/, but only land the merged |
| +// status/fleet.json (+ the embedded page assets PublishFleet ships beside |
| +// it) when publish=true — publish=false is the "another box holds the |
| +// lease" tick, which must still self-report without touching the fleet-wide |
| +// objects. |
| +func TestStatusTickWritesBoxAndPublishesOnlyWhenAsked(t *testing.T) { |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.Upsert(registry.Route{Host: "a.example.com", Kind: "static"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + store := release.NewDirStore(t.TempDir()) |
| + probes := status.Probes{MachineID: "test-machine"} |
| + ctx := context.Background() |
| + |
| + if err := statusTick(ctx, store, reg, probes, false); err != nil { |
| + t.Fatalf("statusTick (publish=false): %v", err) |
| + } |
| + boxes, err := store.List(ctx, "status/boxes/") |
| + if err != nil { |
| + t.Fatalf("List status/boxes/: %v", err) |
| + } |
| + if len(boxes) != 1 { |
| + t.Fatalf("status/boxes/ = %v, want exactly one snapshot", boxes) |
| + } |
| + if _, _, err := store.Get(ctx, "status/fleet.json"); err == nil { |
| + t.Error("publish=false must not write status/fleet.json") |
| + } |
| + |
| + if err := statusTick(ctx, store, reg, probes, true); err != nil { |
| + t.Fatalf("statusTick (publish=true): %v", err) |
| + } |
| + for _, key := range []string{"status/fleet.json", "status/index.html", "status/status.css"} { |
| + if _, _, err := store.Get(ctx, key); err != nil { |
| + t.Errorf("publish=true must write %s: %v", key, err) |
| + } |
| + } |
| + // The box snapshot count must stay at one: the second tick republishes |
| + // the SAME box's key (WriteBox is last-writer-wins on the label), not a |
| + // second file. |
| + boxes, err = store.List(ctx, "status/boxes/") |
| + if err != nil { |
| + t.Fatalf("List status/boxes/ after publish tick: %v", err) |
| + } |
| + if len(boxes) != 1 { |
| + t.Errorf("status/boxes/ = %v after second tick, want still exactly one snapshot", boxes) |
| + } |
| +} |
| + |
| func TestBootstrapIdempotent(t *testing.T) { |
| root := t.TempDir() |
| if err := bootstrap(root); err != nil { |
diff --git a/internal/host/probes.go b/internal/host/probes.go| new file mode 100644 |
| index 000000000..eb03b5433 |
| --- /dev/null |
| +++ b/internal/host/probes.go |
| @@ -0,0 +1,135 @@ |
| +// The box-side half of the fleet status page's Probes contract |
| +// (internal/status.Probes): every function here reads local kernel/OS state |
| +// and MUST fail toward nil, never a guessed zero — status.CollectBox already |
| +// treats a nil probe as "unknown, publish null" everywhere, so a probe that |
| +// invents a number on failure would silently turn an honest unknown into a |
| +// false reading on the public page. Deliberately small, single-purpose |
| +// functions (rather than one combined "gather everything" call) so |
| +// host_test.go's statusTick tests never have to touch /proc or a live |
| +// syscall — they inject a status.Probes fixture directly and never call |
| +// anything in this file. |
| +package host |
| + |
| +import ( |
| + "bufio" |
| + "os" |
| + "strconv" |
| + "strings" |
| + "syscall" |
| +) |
| + |
| +// readMachineID reads /etc/machine-id, trimmed. Unreadable (missing file, |
| +// permission denied — plausible inside some containers) -> "", NOT a |
| +// hash-of-hostname fallback. A hostname-derived id would look exactly like a |
| +// real one to status.Label, silently manufacturing a stable-looking label |
| +// for a box that never actually reported a machine id; "" is honest instead |
| +// — it still hashes to a deterministic Label(""), it just always means "this |
| +// probe never worked," which is the true state of affairs. |
| +func readMachineID() string { |
| + b, err := os.ReadFile("/etc/machine-id") |
| + if err != nil { |
| + return "" |
| + } |
| + return strings.TrimSpace(string(b)) |
| +} |
| + |
| +// readMemFrac reads /proc/meminfo's MemAvailable/MemTotal (both in kB) and |
| +// returns the used fraction (1 - available/total). MemAvailable (not |
| +// MemFree) is the kernel's own "how much could a new process actually get |
| +// without swapping" estimate — it already accounts for reclaimable |
| +// page/slab cache, which MemFree does not, so MemFree would read as |
| +// permanently "critical" on any box with a healthy page cache. Either field |
| +// missing or unparsable -> nil. |
| +func readMemFrac() *float64 { |
| + f, err := os.Open("/proc/meminfo") |
| + if err != nil { |
| + return nil |
| + } |
| + defer f.Close() |
| + |
| + var totalKB, availKB int64 |
| + haveTotal, haveAvail := false, false |
| + scanner := bufio.NewScanner(f) |
| + for scanner.Scan() { |
| + fields := strings.Fields(scanner.Text()) |
| + if len(fields) < 2 { |
| + continue |
| + } |
| + v, err := strconv.ParseInt(fields[1], 10, 64) |
| + if err != nil { |
| + continue |
| + } |
| + switch fields[0] { |
| + case "MemTotal:": |
| + totalKB, haveTotal = v, true |
| + case "MemAvailable:": |
| + availKB, haveAvail = v, true |
| + } |
| + } |
| + if err := scanner.Err(); err != nil || !haveTotal || !haveAvail || totalKB <= 0 { |
| + return nil |
| + } |
| + frac := float64(totalKB-availKB) / float64(totalKB) |
| + return &frac |
| +} |
| + |
| +// readDiskFrac reports how full the data volume is (1 - available/total), |
| +// via syscall.Statfs on dataDir() — /data by default, CARLOS_DATA_DIR when |
| +// set, matching every other on-box path in this package. There is no |
| +// existing "floor" config to mirror: ops/litestream (grepped before writing |
| +// this) carries no reserved-space setting, so this is the plain fraction, |
| +// not a fraction-of-headroom-above-a-floor. Statfs failure (unmounted, |
| +// missing dir) -> nil; a zero-block filesystem (shouldn't happen, but would |
| +// divide by zero) also -> nil rather than a crash. |
| +func readDiskFrac() *float64 { |
| + var st syscall.Statfs_t |
| + if err := syscall.Statfs(dataDir(), &st); err != nil { |
| + return nil |
| + } |
| + if st.Blocks == 0 { |
| + return nil |
| + } |
| + total := float64(st.Blocks) * float64(st.Bsize) |
| + avail := float64(st.Bavail) * float64(st.Bsize) |
| + frac := 1 - avail/total |
| + return &frac |
| +} |
| + |
| +// listeningProbe answers status.Probes.Listening by reading /proc/net/unix |
| +// — the kernel's own table of bound unix-domain sockets — rather than |
| +// dialing sock. Dialing would itself be a connection attempt on a |
| +// hibernating tenant's socket, which is exactly the event that wakes it: the |
| +// probe would falsify the very thing it's trying to observe. Presence of |
| +// sock's path in the table is the awake signal (an exec-backed hibernating |
| +// route's socket only exists in the kernel while the spawned child holds it |
| +// open; asleep, the child is gone and the entry is gone with it); absence is |
| +// asleep. Only a failure to read the table at all -> nil ("unknown"), which |
| +// is what makes CollectBox null the whole Awake/Asleep pair for this box |
| +// rather than reporting a guess. |
| +func listeningProbe(sock string) *bool { |
| + f, err := os.Open("/proc/net/unix") |
| + if err != nil { |
| + return nil |
| + } |
| + defer f.Close() |
| + |
| + found := false |
| + scanner := bufio.NewScanner(f) |
| + scanner.Scan() // header line ("Num RefCount Protocol Flags Type St Inode Path") — not a data row |
| + for scanner.Scan() { |
| + fields := strings.Fields(scanner.Text()) |
| + if len(fields) == 0 { |
| + continue |
| + } |
| + // Path, when present, is always the last column; sockets with no |
| + // Path (anonymous socketpairs) simply never match sock. |
| + if fields[len(fields)-1] == sock { |
| + found = true |
| + break |
| + } |
| + } |
| + if err := scanner.Err(); err != nil { |
| + return nil |
| + } |
| + return &found |
| +} |