amadan
plan: host & process monitoring — 13 tasks, TDD, exact seams
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
1 file changed,
+720
−0
docs/superpowers/plans/2026-08-02-host-process-monitoring.md+720 −0
diff --git a/docs/superpowers/plans/2026-08-02-host-process-monitoring.md b/docs/superpowers/plans/2026-08-02-host-process-monitoring.md| new file mode 100644 |
| index 000000000..a68b35ddc |
| --- /dev/null |
| +++ b/docs/superpowers/plans/2026-08-02-host-process-monitoring.md |
| @@ -0,0 +1,720 @@ |
| +# Host & Process Monitoring Implementation Plan |
| + |
| +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. |
| + |
| +**Goal:** Operator-grade CPU/memory/disk monitoring per host and per systemd unit, with bucket-stored history, on-box alerting (webhook + console-relayed keymail), an operator dashboard, and owner-facing per-app metrics in the console. |
| + |
| +**Architecture:** A new pure package `internal/metrics` (parsers, storage shapes, rollups, alert rules — no OS access) + OS-reading probes and a 60s sampler loop in `internal/host` (same seam as the existing statusTick) + console additions (alert watcher with keymail delivery, operator dashboard, owner app metrics). Storage is all-new bucket prefixes `metrics/` and `alerts/`; one additive method (`Delete`) lands on `release.Store`. |
| + |
| +**Tech Stack:** Go stdlib only (no new deps). `release.Store` for bucket IO, `html/template` console pages, cgroup v2 files for per-unit readings. |
| + |
| +## Global Constraints |
| + |
| +- Spec: `docs/superpowers/specs/2026-08-02-host-process-monitoring-design.md` — read it first. |
| +- All probes/readers **fail toward nil, never a guessed zero** (convention documented in `internal/host/probes.go:1-10`). |
| +- Wire shapes are **additive**: nothing that exists in a bucket today changes shape; all new objects carry `"v": 1`. |
| +- The public status page writer (`internal/status` fleet publish) must never read `metrics/` or `alerts/` — guarded by a contract test (Task 8). |
| +- Console security gates ordered: session → CSRF → owner/operator (byte-identical `http.NotFound`) → `requireFreshAuth` LAST. Operator gate copies `handleFleet` (`internal/console/server.go:888-912`) exactly: every failure path answers plain `http.NotFound`. |
| +- Owner metrics boundary: an owner sees ONLY their own units' cgroup readings — never host totals, other tenants' units, or anything from `alerts/`. |
| +- Vendored assets only: icons are in-code SVG constants (`internal/console/icons.go`), sparklines are server-computed inline SVG. No CDN, no JS libraries. |
| +- Build gate before every commit: `go build ./... && go vet ./... && go test ./... -count=1` all clean, `gofmt -l` clean for touched files. Sandbox: `GOCACHE=/tmp/claude-1001/gocache GOFLAGS=-mod=mod GOPROXY=off`, disable sandbox if the toolchain reports "read-only file system". |
| +- Times in bucket keys are **UTC**: hour key `2006-01-02-15`, day key `2006-01-02`. |
| +- Commit messages follow repo style: lowercase prefix (`metrics:`, `host:`, `console:`, `release:`), what + why. |
| + |
| +--- |
| + |
| +### Task 1: `release.Store` gains `Delete` |
| + |
| +**Files:** |
| +- Modify: `internal/release/store.go` (interface at :34-42, DirStore :52-157, S3Store :164-280) |
| +- Test: `internal/release/store_test.go` (follow existing DirStore test conventions in that file) |
| + |
| +**Interfaces:** |
| +- Consumes: existing `Store` interface `{Get(ctx, key) ([]byte, string, error); Put(ctx, key, body, ifMatch) (string, error); List(ctx, prefix) ([]string, error)}`. |
| +- Produces: `Delete(ctx context.Context, key string) error` on the interface and both impls. Deleting a missing key is **not an error** (idempotent — pruning re-runs must not fail). Any test doubles embedding `release.Store` (e.g. `flakyPutStore` in `internal/status/status_test.go:21-33`) inherit it via embedding — verify with the full-repo build. |
| + |
| +- [ ] **Step 1: Write the failing test** |
| + |
| +```go |
| +func TestDirStoreDelete(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + ctx := context.Background() |
| + if _, err := s.Put(ctx, "metrics/boxes/b/raw/2026-08-01-10.json", []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := s.Delete(ctx, "metrics/boxes/b/raw/2026-08-01-10.json"); err != nil { |
| + t.Fatalf("Delete: %v", err) |
| + } |
| + if _, _, err := s.Get(ctx, "metrics/boxes/b/raw/2026-08-01-10.json"); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("Get after Delete = %v, want ErrNotExist", err) |
| + } |
| + // Idempotent: deleting a key that isn't there is fine. |
| + if err := s.Delete(ctx, "metrics/boxes/b/raw/2026-08-01-10.json"); err != nil { |
| + t.Fatalf("second Delete: %v", err) |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: Run `go test ./internal/release/ -run TestDirStoreDelete -count=1`** — expect FAIL (method undefined). |
| + |
| +- [ ] **Step 3: Implement.** Add `Delete(ctx context.Context, key string) error` to the `Store` interface with a doc comment stating idempotency. DirStore: under `d.mu`, `os.Remove(filepath.Join(d.root, filepath.FromSlash(key)))`, map `os.IsNotExist` to nil. S3Store: `s.Client.DeleteObject(ctx, &awss3.DeleteObjectInput{Bucket: aws.String(s.Bucket), Key: aws.String(key)})` — S3 DeleteObject on a missing key already succeeds. |
| + |
| +- [ ] **Step 4: Run `go build ./... && go test ./internal/release/ -count=1`** — expect PASS, whole repo compiles (any non-embedding Store double in tests will surface here; give it a trivial Delete). |
| + |
| +- [ ] **Step 5: Commit** `release: Store.Delete — additive, idempotent; metrics pruning needs it` |
| + |
| +--- |
| + |
| +### Task 2: `internal/metrics` — parsers (pure, no OS access) |
| + |
| +**Files:** |
| +- Create: `internal/metrics/parse.go`, `internal/metrics/parse_test.go` |
| + |
| +**Interfaces:** |
| +- Produces (consumed by Tasks 6-7): |
| + |
| +```go |
| +package metrics |
| + |
| +// CPUCounters is a monotonic reading of /proc/stat's aggregate cpu line. |
| +type CPUCounters struct{ Busy, Total uint64 } |
| + |
| +func ParseProcStat(text string) *CPUCounters // nil on any parse failure |
| +func CPUFrac(prev, cur *CPUCounters) *float64 // nil unless both non-nil and cur.Total > prev.Total |
| +func ParseLoadAvg(text string) (l1, l5, l15 *float64) // all-nil on failure |
| +func ParseCgroupCPUUsec(text string) *int64 // usage_usec from cpu.stat; nil on failure |
| +func ParseCgroupInt(text string) *int64 // memory.current / memory.peak; nil on failure ("max" -> nil) |
| +func ParseCgroupIO(text string) (rbytes, wbytes *int64) // io.stat rbytes/wbytes summed across devices; both nil on failure |
| +func UnitCPUFrac(prevUsec, curUsec *int64, elapsed time.Duration) *float64 // nil unless both non-nil, cur >= prev, elapsed > 0 |
| +``` |
| + |
| +- [ ] **Step 1: Write failing tests** covering, minimum: |
| + |
| +```go |
| +func TestParseProcStat(t *testing.T) { |
| + // user nice system idle iowait irq softirq steal |
| + c := ParseProcStat("cpu 100 20 30 800 50 5 5 0 0 0\ncpu0 1 2 3 4 5 6 7 8 9 10\n") |
| + if c == nil { |
| + t.Fatal("nil for valid input") |
| + } |
| + // Busy = everything except idle(800) and iowait(50) = 160; Total = 1010. |
| + if c.Busy != 160 || c.Total != 1010 { |
| + t.Fatalf("got busy=%d total=%d, want 160/1010", c.Busy, c.Total) |
| + } |
| + if ParseProcStat("") != nil || ParseProcStat("cpu x y z") != nil { |
| + t.Fatal("garbage must parse to nil") |
| + } |
| +} |
| + |
| +func TestCPUFrac(t *testing.T) { |
| + prev := &CPUCounters{Busy: 100, Total: 1000} |
| + cur := &CPUCounters{Busy: 160, Total: 1100} |
| + f := CPUFrac(prev, cur) |
| + if f == nil || math.Abs(*f-0.6) > 1e-9 { // 60 busy / 100 total |
| + t.Fatalf("got %v, want 0.6", f) |
| + } |
| + if CPUFrac(nil, cur) != nil || CPUFrac(prev, nil) != nil { |
| + t.Fatal("nil input must be nil") |
| + } |
| + if CPUFrac(cur, prev) != nil { // counters went backwards (reboot) — no guess |
| + t.Fatal("backwards counters must be nil") |
| + } |
| +} |
| + |
| +func TestParseCgroupCPUUsec(t *testing.T) { |
| + v := ParseCgroupCPUUsec("usage_usec 4500000\nuser_usec 3000000\nsystem_usec 1500000\n") |
| + if v == nil || *v != 4500000 { |
| + t.Fatalf("got %v, want 4500000", v) |
| + } |
| + if ParseCgroupCPUUsec("") != nil { |
| + t.Fatal("empty must be nil") |
| + } |
| +} |
| + |
| +func TestParseCgroupIO(t *testing.T) { |
| + r, w := ParseCgroupIO("259:0 rbytes=1000 wbytes=2000 rios=10 wios=20 dbytes=0 dios=0\n259:1 rbytes=500 wbytes=100 rios=1 wios=1 dbytes=0 dios=0\n") |
| + if r == nil || w == nil || *r != 1500 || *w != 2100 { |
| + t.Fatalf("got r=%v w=%v, want 1500/2100", r, w) |
| + } |
| +} |
| +``` |
| + |
| +Plus `TestParseLoadAvg` ("0.52 0.41 0.30 1/234 5678" → 0.52/0.41/0.30; garbage → all nil), `TestParseCgroupInt` ("123456\n" → 123456; "max" → nil), `TestUnitCPUFrac` (3_000_000 usec over 60s = 0.05; backwards/nil/zero-elapsed → nil). |
| + |
| +- [ ] **Step 2: Run `go test ./internal/metrics/ -count=1`** — FAIL (package doesn't exist). |
| + |
| +- [ ] **Step 3: Implement `parse.go`.** `/proc/stat` cpu line: fields after "cpu" are user nice system idle iowait irq softirq steal [guest guest_nice]; Total = sum of first 8, Busy = Total − idle − iowait (guest columns are already included in user/nice on modern kernels — sum only the first 8 to avoid double-count). Every parser returns nil on any missing/garbled field, per the package doc comment you write at the top: mirror the "fail toward nil" language of `internal/host/probes.go:1-10`. |
| + |
| +- [ ] **Step 4: Run tests** — PASS. **Step 5: Commit** `metrics: pure parsers for /proc and cgroup v2 readings` |
| + |
| +--- |
| + |
| +### Task 3: `internal/metrics` — samples, chunk keys, chunk write, prune list |
| + |
| +**Files:** |
| +- Create: `internal/metrics/sample.go`, `internal/metrics/chunk.go`, `internal/metrics/chunk_test.go` |
| + |
| +**Interfaces (produces, consumed by Tasks 6-7 and 10-12):** |
| + |
| +```go |
| +type HostSample struct { |
| + CPUFrac *float64 `json:"cpu"` |
| + Load1 *float64 `json:"load1"` |
| + Load5 *float64 `json:"load5"` |
| + Load15 *float64 `json:"load15"` |
| + MemFrac *float64 `json:"mem"` |
| + DiskFrac *float64 `json:"disk"` |
| +} |
| +type UnitSample struct { |
| + CPUFrac *float64 `json:"cpu"` |
| + MemBytes *int64 `json:"mem_bytes"` |
| + MemPeakBytes *int64 `json:"mem_peak_bytes"` |
| + ReadBytes *int64 `json:"read_bytes"` // cumulative |
| + WriteBytes *int64 `json:"write_bytes"` // cumulative |
| +} |
| +type Sample struct { |
| + At time.Time `json:"at"` |
| + Host HostSample `json:"host"` |
| + Units map[string]UnitSample `json:"units"` // key: full unit name, e.g. carlos-app@x.example.com.service |
| +} |
| +type RawChunk struct { |
| + V int `json:"v"` // 1 |
| + Box string `json:"box"` |
| + Hour string `json:"hour"` // UTC, 2006-01-02-15 |
| + Samples []Sample `json:"samples"` |
| +} |
| + |
| +func HourOf(t time.Time) string // t.UTC().Format("2006-01-02-15") |
| +func RawKey(box, hour string) string // "metrics/boxes/" + box + "/raw/" + hour + ".json" |
| +func DayKey(box, date string) string // "metrics/boxes/" + box + "/days/" + date + ".json" |
| +func WriteRaw(ctx context.Context, s release.Store, c RawChunk) error |
| +func StaleRawKeys(keys []string, now time.Time, keep time.Duration) []string |
| +``` |
| + |
| +`WriteRaw` uses the read-etag-then-Put overwrite pattern of `status.WriteBox` (`internal/status/snapshot.go:260-287`): Get current etag (missing → `""`), Put with it, on `release.ErrCASConflict` re-read and retry once, second conflict returns the error. `StaleRawKeys` is pure: parses the trailing `<hour>.json` of each key, returns keys whose hour-start is older than `now.Add(-keep)`; unparsable keys are **skipped, never returned** (never delete what you don't understand). |
| + |
| +- [ ] **Step 1: Failing tests:** round-trip `WriteRaw` to a `release.NewDirStore(t.TempDir())` then Get+Unmarshal and compare; overwrite same hour twice and verify one key, latest content wins; `TestStaleRawKeys`: |
| + |
| +```go |
| +func TestStaleRawKeys(t *testing.T) { |
| + now := time.Date(2026, 8, 2, 22, 30, 0, 0, time.UTC) |
| + keys := []string{ |
| + "metrics/boxes/b/raw/2026-07-31-10.json", // stale |
| + "metrics/boxes/b/raw/2026-08-02-21.json", // fresh |
| + "metrics/boxes/b/raw/garbage.json", // unparsable — never returned |
| + } |
| + got := StaleRawKeys(keys, now, 48*time.Hour) |
| + if len(got) != 1 || got[0] != keys[0] { |
| + t.Fatalf("got %v, want only the stale key", got) |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: Run — FAIL. Step 3: Implement. Step 4: Run — PASS. Step 5: Commit** `metrics: raw chunk shapes, bucket keys, CAS overwrite, prune selection` |
| + |
| +--- |
| + |
| +### Task 4: `internal/metrics` — hourly rollups |
| + |
| +**Files:** |
| +- Create: `internal/metrics/rollup.go`, `internal/metrics/rollup_test.go` |
| + |
| +**Interfaces (produces):** |
| + |
| +```go |
| +type Agg struct { |
| + Min float64 `json:"min"` |
| + Avg float64 `json:"avg"` |
| + Max float64 `json:"max"` |
| + N int `json:"n"` // samples that carried this metric (nil samples excluded) |
| +} |
| +type HourAgg struct { |
| + Host map[string]Agg `json:"host"` // "cpu","mem","disk","load1" |
| + Units map[string]map[string]Agg `json:"units"` // unit -> "cpu","mem_bytes" -> Agg |
| +} |
| +type DayRollup struct { |
| + V int `json:"v"` // 1 |
| + Box string `json:"box"` |
| + Date string `json:"date"` // UTC 2006-01-02 |
| + Hours map[string]HourAgg `json:"hours"` // "00".."23" |
| +} |
| + |
| +func RollupHour(samples []Sample) HourAgg // nil metrics excluded from N; metric absent everywhere -> key absent |
| +func UpsertRollup(ctx context.Context, s release.Store, box string, hour time.Time, agg HourAgg) error |
| +``` |
| + |
| +`UpsertRollup` reads `DayKey(box, hour.UTC().Format("2006-01-02"))` (missing → fresh `DayRollup{V:1, Box: box, Date: ..., Hours: map[...]}`), sets `Hours[hour.UTC().Format("15")]`, writes back with the same CAS-retry-once pattern as `WriteRaw`. Unit rollup metrics: `"cpu"` from `UnitSample.CPUFrac`, `"mem_bytes"` from `MemBytes` (float64-cast). Host: `"cpu"`, `"mem"`, `"disk"` fractions and `"load1"`. |
| + |
| +- [ ] **Step 1: Failing tests:** `RollupHour` over 3 samples where one has nil CPU → `Host["cpu"].N == 2` with correct min/avg/max; a unit present in only 1 sample → its aggs have N=1; all-nil metric → key absent. `UpsertRollup` twice for hours 10 and 11 of the same day → one day object with two hour entries. |
| +- [ ] **Step 2: FAIL. Step 3: Implement. Step 4: PASS. Step 5: Commit** `metrics: hourly min/avg/max rollups into day objects` |
| + |
| +--- |
| + |
| +### Task 5: `internal/metrics` — alert rules, sustain, hysteresis |
| + |
| +**Files:** |
| +- Create: `internal/metrics/alert.go`, `internal/metrics/alert_test.go` |
| + |
| +**Interfaces (produces, consumed by Task 7 and the console in Tasks 9-11):** |
| + |
| +```go |
| +type Rules struct { |
| + DiskFrac float64 // fire >, default 0.90, immediate |
| + MemFrac float64 // default 0.90 |
| + MemSustain time.Duration // default 10 * time.Minute |
| + CPUFrac float64 // default 0.95 |
| + CPUSustain time.Duration // default 15 * time.Minute |
| + LagSecs int64 // default 900, immediate; inert while readings are nil |
| + ClearBy float64 // hysteresis, default 0.05 (fractions); lag clears at LagSecs-300 |
| +} |
| +func DefaultRules() Rules |
| +func RulesFromEnv(getenv func(string) string) Rules |
| +// env overrides (percent integers, matching the spec's host.env surface): |
| +// CARLOS_ALERT_DISK_PCT, CARLOS_ALERT_MEM_PCT, CARLOS_ALERT_CPU_PCT |
| +// unparsable or out-of-range (0..100) values keep the default (log at caller, not here — pure). |
| + |
| +type Observation struct { |
| + At time.Time |
| + CPUFrac *float64 |
| + MemFrac *float64 |
| + DiskFrac *float64 |
| + LagSecs *int64 |
| +} |
| +type RuleState struct { |
| + Firing bool `json:"firing"` |
| + Since time.Time `json:"since,omitempty"` // continuously-above-threshold since (sustain tracking) |
| + FiredAt time.Time `json:"fired_at,omitempty"` |
| + Value float64 `json:"value,omitempty"` // reading at last transition |
| +} |
| +type AlertState struct { |
| + V int `json:"v"` // 1 |
| + Box string `json:"box"` |
| + UpdatedAt time.Time `json:"updated_at"` |
| + Rules map[string]RuleState `json:"rules"` // "disk","mem","cpu","lag" |
| +} |
| +type Event struct { |
| + Rule string `json:"rule"` |
| + Firing bool `json:"firing"` // true=fired, false=cleared |
| + Value float64 `json:"value"` |
| + Box string `json:"box"` |
| +} |
| +func Evaluate(r Rules, st AlertState, obs Observation) (AlertState, []Event) |
| +func AlertKey(box string) string // "alerts/boxes/" + box + ".json" |
| +``` |
| + |
| +**Evaluate semantics (encode exactly):** |
| +- nil reading: state unchanged for that rule — no data never fires AND never clears (clearing needs evidence). |
| +- Not firing, reading > threshold: set `Since` if zero; fire (emit Event) when `obs.At.Sub(Since) >= sustain` (immediate rules: sustain 0 → fire same tick). Reading back at/below threshold before sustain elapses → reset `Since` to zero. |
| +- Firing, reading < threshold−ClearBy: clear (emit Event), zero `Since`/`FiredAt`. |
| +- Firing, reading between clear-line and threshold: stays firing (hysteresis band). |
| + |
| +- [ ] **Step 1: Failing tests** — table-driven walk of ticks: |
| + |
| +```go |
| +func TestEvaluateMemSustainAndHysteresis(t *testing.T) { |
| + r := DefaultRules() |
| + st := AlertState{V: 1, Box: "b", Rules: map[string]RuleState{}} |
| + t0 := time.Date(2026, 8, 2, 10, 0, 0, 0, time.UTC) |
| + obs := func(m float64, at time.Time) Observation { return Observation{At: at, MemFrac: &m} } |
| + |
| + st, ev := Evaluate(r, st, obs(0.95, t0)) // above, sustain starts — no event yet |
| + if len(ev) != 0 || st.Rules["mem"].Firing { |
| + t.Fatalf("must not fire before sustain: ev=%v firing=%v", ev, st.Rules["mem"].Firing) |
| + } |
| + st, ev = Evaluate(r, st, obs(0.95, t0.Add(10*time.Minute))) // sustain met — fires |
| + if len(ev) != 1 || !ev[0].Firing || ev[0].Rule != "mem" { |
| + t.Fatalf("want fire event, got %v", ev) |
| + } |
| + st, ev = Evaluate(r, st, obs(0.88, t0.Add(11*time.Minute))) // hysteresis band (0.85..0.90) — stays firing |
| + if len(ev) != 0 || !st.Rules["mem"].Firing { |
| + t.Fatal("must hold inside hysteresis band") |
| + } |
| + st, ev = Evaluate(r, st, obs(0.80, t0.Add(12*time.Minute))) // below clear line — clears |
| + if len(ev) != 1 || ev[0].Firing { |
| + t.Fatalf("want clear event, got %v", ev) |
| + } |
| +} |
| +``` |
| + |
| +Plus: disk fires immediately at first tick > 0.90; sustain resets when a dip below threshold interrupts it; nil reading holds a firing alert firing and a quiet rule quiet; lag nil forever → no lag event ever; `RulesFromEnv` with a getenv built from a map (the `env := func(m map[string]string) func(string) string {...}` helper pattern at `internal/host/host_test.go:170-172`): `{"CARLOS_ALERT_DISK_PCT": "95"}` → DiskFrac 0.95, `"nonsense"` → default kept. |
| + |
| +- [ ] **Step 2: FAIL. Step 3: Implement. Step 4: PASS. Step 5: Commit** `metrics: alert rules — sustain windows, hysteresis, nil never fires or clears` |
| + |
| +--- |
| + |
| +### Task 6: `internal/host` — OS readers for the sampler |
| + |
| +**Files:** |
| +- Create: `internal/host/metrics_probes.go` |
| +- Test: parsing already covered in Task 2; this file is thin OS glue, untested by the same convention as `probes.go` readers (its doc comment must say so). |
| + |
| +**Interfaces (produces, consumed by Task 7):** |
| + |
| +```go |
| +// All readers fail toward nil / empty, mirroring probes.go. |
| +func readProcStat() *metrics.CPUCounters // os.ReadFile("/proc/stat") + metrics.ParseProcStat |
| +func readLoadAvg() (l1, l5, l15 *float64) // os.ReadFile("/proc/loadavg") + metrics.ParseLoadAvg |
| +const cgroupSliceDir = "/sys/fs/cgroup/system.slice" |
| +func listCarlosUnits() []string // ReadDir(cgroupSliceDir), names with prefix "carlos-" and suffix ".service"; nil on error |
| +type unitRaw struct { |
| + cpuUsec *int64 |
| + memBytes *int64 |
| + memPeakBytes *int64 |
| + readBytes *int64 |
| + writeBytes *int64 |
| +} |
| +func readUnitRaw(unit string) unitRaw // reads cpu.stat, memory.current, memory.peak, io.stat under cgroupSliceDir/<unit>/ |
| +``` |
| + |
| +- [ ] **Step 1: Implement** (no TDD cycle — parsers are tested; this is `os.ReadFile` + `filepath.Join` glue; each file read failure yields nil for that field only). |
| +- [ ] **Step 2: `go build ./... && go vet ./...`** — clean. |
| +- [ ] **Step 3: Commit** `host: cgroup and /proc readers for the metrics sampler` |
| + |
| +--- |
| + |
| +### Task 7: `internal/host` — sampler loop, flush, rollup, prune, alerting, webhook |
| + |
| +**Files:** |
| +- Create: `internal/host/metrics_loop.go` |
| +- Modify: `internal/host/agent.go:147` (add `go metricsLoop(reg)` beside the other loops; reg is unused v1 but keeps the signature uniform — actually pass nothing: `go metricsLoop()`) |
| +- Test: `internal/host/metrics_loop_test.go` |
| + |
| +**Interfaces:** |
| +- Consumes: Tasks 1-6 (`metrics.*`, `release.Store`, readers), `status.Label(machineID)` (`internal/status/snapshot.go:157`) for the box name, `readMachineID()`/`readMemFrac()`/`readDiskFrac()` (`internal/host/probes.go`), `resolveStatusBucket` (`internal/host/agent.go:425`). |
| +- Produces (pure, the test seam): |
| + |
| +```go |
| +// sampleProbes is the injectable fixture bundle, like status.Probes. |
| +type sampleProbes struct { |
| + procStat func() *metrics.CPUCounters |
| + loadavg func() (l1, l5, l15 *float64) |
| + memFrac func() *float64 |
| + diskFrac func() *float64 |
| + units func() map[string]unitRaw |
| + now func() time.Time |
| +} |
| +// sampler carries the delta state between ticks. |
| +type sampler struct { |
| + prevCPU *metrics.CPUCounters |
| + prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| + prevAt time.Time |
| + buf []metrics.Sample |
| +} |
| +func (sm *sampler) tick(p sampleProbes) // appends one metrics.Sample to buf |
| +// metricsFlush writes buf as the current hour's chunk; on hour rollover it also |
| +// rolls up the PREVIOUS hour's samples, upserts the day rollup, prunes raw |
| +// chunks older than 48h (List + metrics.StaleRawKeys + Delete), and drops |
| +// flushed samples from buf. |
| +func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, now time.Time) error |
| +// alertTick evaluates rules against the latest sample, persists state to |
| +// <dataDir>/monitor/alerts.json, publishes metrics.AlertKey(box) when state |
| +// changed, and posts each event to webhookURL ("" = skip). |
| +func alertTick(ctx context.Context, s release.Store, box string, rules metrics.Rules, |
| + st *metrics.AlertState, latest metrics.Sample, statePath, webhookURL string, |
| + post func(url string, body []byte) error) error |
| +``` |
| + |
| +`metricsLoop()` itself is thin and untested (convention of `statusLoop`): resolve bucket via `resolveStatusBucket(os.Getenv)` (unset → one log line, return — inert until configured); `release.NewS3Store`; box = `status.Label(readMachineID())`; load persisted `AlertState` from `<dataDir()>/monitor/alerts.json` (missing/corrupt → fresh state, never crash); `metrics.RulesFromEnv(os.Getenv)`; webhook = `os.Getenv("CARLOS_ALERT_WEBHOOK")`; then a 60s ticker calling `sm.tick(realProbes())` + `alertTick(...)` every tick and `metricsFlush(...)` every 5th. The webhook `post` is a 5s-timeout `http.Post(url, "application/json", ...)`; non-2xx is an error (logged by caller, alert state keeps the event delivered=false? NO — v1: webhook is fire-and-forget per event, failures logged only; the console keymail path is the reliable channel). |
| + |
| +- [ ] **Step 1: Failing tests** (DirStore + fixtures, no OS): |
| + |
| +```go |
| +func TestSamplerTickAndFlush(t *testing.T) { |
| + ctx := context.Background() |
| + store := release.NewDirStore(t.TempDir()) |
| + at := time.Date(2026, 8, 2, 10, 59, 30, 0, time.UTC) |
| + c0, c1 := &metrics.CPUCounters{Busy: 100, Total: 1000}, &metrics.CPUCounters{Busy: 160, Total: 1100} |
| + mem, disk := 0.5, 0.3 |
| + usec0, usec1 := int64(0), int64(3_000_000) |
| + sm := &sampler{} |
| + probes := func(c *metrics.CPUCounters, usec int64, now time.Time) sampleProbes { |
| + return sampleProbes{ |
| + procStat: func() *metrics.CPUCounters { return c }, |
| + loadavg: func() (*float64, *float64, *float64) { return nil, nil, nil }, |
| + memFrac: func() *float64 { return &mem }, |
| + diskFrac: func() *float64 { return &disk }, |
| + units: func() map[string]unitRaw { |
| + return map[string]unitRaw{"carlos-app@a.example.com.service": {cpuUsec: &usec}} |
| + }, |
| + now: func() time.Time { return now }, |
| + } |
| + } |
| + sm.tick(probes(c0, usec0, at)) |
| + sm.tick(probes(c1, usec1, at.Add(time.Minute))) |
| + if len(sm.buf) != 2 { |
| + t.Fatalf("buf=%d, want 2", len(sm.buf)) |
| + } |
| + if sm.buf[0].Host.CPUFrac != nil { |
| + t.Fatal("first tick has no delta — CPU must be nil, not 0") |
| + } |
| + if f := sm.buf[1].Host.CPUFrac; f == nil || math.Abs(*f-0.6) > 1e-9 { |
| + t.Fatalf("second tick CPU=%v, want 0.6", f) |
| + } |
| + if f := sm.buf[1].Units["carlos-app@a.example.com.service"].CPUFrac; f == nil || math.Abs(*f-0.05) > 1e-9 { |
| + t.Fatalf("unit CPU=%v, want 0.05 (3s of CPU over 60s)", f) |
| + } |
| + |
| + // Flush within the hour: one raw chunk, no rollup yet. |
| + if err := metricsFlush(ctx, store, "box-a", sm, at.Add(time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, _, err := store.Get(ctx, "metrics/boxes/box-a/raw/2026-08-02-10.json"); err != nil { |
| + t.Fatalf("raw chunk missing: %v", err) |
| + } |
| + // Tick past the hour boundary, flush again: previous hour rolled up. |
| + sm.tick(probes(c1, usec1, at.Add(2*time.Minute))) // 11:01:30 |
| + if err := metricsFlush(ctx, store, "box-a", sm, at.Add(2*time.Minute)); err != nil { |
| + t.Fatal(err) |
| + } |
| + body, _, err := store.Get(ctx, "metrics/boxes/box-a/days/2026-08-02.json") |
| + if err != nil { |
| + t.Fatalf("day rollup missing: %v", err) |
| + } |
| + var day metrics.DayRollup |
| + if err := json.Unmarshal(body, &day); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, ok := day.Hours["10"]; !ok { |
| + t.Fatalf("hour 10 not rolled up: %v", day.Hours) |
| + } |
| +} |
| +``` |
| + |
| +Plus: `TestMetricsFlushPrunesOldChunks` (seed a 3-day-old raw key, flush at an hour boundary, key gone; fresh key survives); `TestAlertTickPersistsPublishesAndPosts` (latest sample disk=0.95 → state file written under `t.TempDir()`, `alerts/boxes/box-a.json` in store parses to `AlertState` with disk firing, injected `post` captured exactly one body containing `"rule":"disk"`, and a second identical tick emits NO new post); `TestAlertTickWebhookFailureDoesNotBlock` (post returns error → alertTick returns nil, state still persisted/published). |
| + |
| +- [ ] **Step 2: FAIL. Step 3: Implement `metrics_loop.go`** and add `go metricsLoop()` in `RunAgent` (`agent.go:147` block). **Step 4: PASS, full `go test ./... -count=1`. Step 5: Commit** `host: metrics sampler loop — 60s ticks, 5min flush, hourly rollup+prune, on-box alerts` |
| + |
| +--- |
| + |
| +### Task 8: Privacy contract test — needles never reach `status/` |
| + |
| +**Files:** |
| +- Modify: `internal/host/host_test.go` |
| + |
| +**Interfaces:** Consumes `statusTick` (`internal/host/agent.go:553`), `release.NewDirStore`, `metrics.WriteRaw`. |
| + |
| +- [ ] **Step 1: Write the test** (it should PASS immediately — it's a tripwire for future regressions, and MUST be committed with a comment saying exactly that): |
| + |
| +```go |
| +// TestStatusPublishNeverReadsMetricsPrefixes is the privacy tripwire the |
| +// monitoring spec commits to: operator-grade needles live under metrics/ and |
| +// alerts/, and nothing the public status publisher writes may ever be |
| +// derived from them. Sentinel values planted in metrics objects must not |
| +// appear anywhere under status/ after a full publish tick. |
| +func TestStatusPublishNeverReadsMetricsPrefixes(t *testing.T) { |
| + ctx := context.Background() |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + store := release.NewDirStore(t.TempDir()) |
| + sentinel := 0.987654321 |
| + chunk := metrics.RawChunk{V: 1, Box: "box-a", Hour: "2026-08-02-10", Samples: []metrics.Sample{ |
| + {At: time.Now(), Host: metrics.HostSample{CPUFrac: &sentinel}}, |
| + }} |
| + if err := metrics.WriteRaw(ctx, store, chunk); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := statusTick(ctx, store, reg, status.Probes{MachineID: "m"}, true); err != nil { |
| + t.Fatal(err) |
| + } |
| + keys, err := store.List(ctx, "status/") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, k := range keys { |
| + body, _, err := store.Get(ctx, k) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if strings.Contains(string(body), "987654321") { |
| + t.Errorf("%s contains the metrics sentinel — the status publisher read a metrics object", k) |
| + } |
| + } |
| +} |
| +``` |
| + |
| +- [ ] **Step 2: Run — PASS. Step 3: Commit** `host: contract test — status publish never derives from metrics/ or alerts/` |
| + |
| +--- |
| + |
| +### Task 9: Console — operator list, alert watcher, keymail delivery |
| + |
| +**Files:** |
| +- Modify: `internal/console/store.go` (new query + migration), `internal/console/server.go` (watcher fields), `cmd/console/main.go:208-223` (start watcher in `serve()`) |
| +- Create: `internal/console/alertwatch.go` |
| +- Test: `internal/console/alertwatch_test.go` |
| + |
| +**Interfaces:** |
| +- Consumes: `Store.IsOperator` pattern (`store.go:1055`), `Store.GrantToken(address)` (`store.go:1081`), `Keymail.SendApp` (`keymail.go:238`), `s.cfg.Releases release.Store`, `metrics.AlertState`/`AlertKey`. |
| +- Produces: |
| + |
| +```go |
| +// store.go |
| +func (s *Store) Operators() ([]string, error) // SELECT address FROM users WHERE operator != 0 ORDER BY address |
| +// migration appended to the migrations list, same style as store.go:172: |
| +// CREATE TABLE IF NOT EXISTS alert_seen (box TEXT NOT NULL, rule TEXT NOT NULL, |
| +// firing INTEGER NOT NULL, PRIMARY KEY (box, rule)); |
| +func (s *Store) AlertSeen(box, rule string) (firing bool, known bool, err error) |
| +func (s *Store) SetAlertSeen(box, rule string, firing bool) error |
| + |
| +// alertwatch.go |
| +// alertPass reads every alerts/boxes/*.json, diffs each rule's firing state |
| +// against alert_seen, and for each TRANSITION sends one keymail per operator |
| +// (via send, using that operator's own grant token) and records the new state. |
| +// A send failure for ANY operator leaves alert_seen unrecorded for that rule, |
| +// so the next pass retries the whole transition — matching the spec's |
| +// "retried on the next tick". Operators with no grant token are skipped |
| +// (logged once per pass, not fatal). Missing alerts/ prefix = nothing to do. |
| +func (s *Server) alertPass(ctx context.Context) error |
| +// StartAlertWatcher runs alertPass every minute until ctx is done. Called |
| +// from cmd/console serve(), NOT NewServer — tests construct servers without |
| +// background goroutines. |
| +func (s *Server) StartAlertWatcher(ctx context.Context) |
| +``` |
| + |
| +`Server` gains one field: `sendApp func(ctx context.Context, token, to, subject, body string) (string, error)` — `NewServer` defaults it to `s.km.SendApp`; tests overwrite it. Message copy: subject `"[carlos] <box>: <rule> alert"` / `"[carlos] <box>: <rule> cleared"`, body includes rule, box, value, and UTC timestamp — plain sentences, no needles beyond the triggering value. |
| + |
| +- [ ] **Step 1: Failing tests:** seed a console `Store` (see `store_test.go` for the open-in-tempdir pattern) with two operator users, one holding a grant token; seed a DirStore with `alerts/boxes/box-a.json` marshalled from `metrics.AlertState{V:1, Box:"box-a", Rules: map[string]metrics.RuleState{"disk": {Firing: true, Value: 0.95}}}`; a captured `sendApp` func. Assert: first `alertPass` sends exactly 1 keymail (the token-holding operator) and records seen; second pass sends 0; flipping the stored state to cleared sends a "cleared" keymail; a failing `sendApp` leaves the transition unrecorded (next pass retries). |
| +- [ ] **Step 2: FAIL. Step 3: Implement** store methods + migration + `alertwatch.go` + `serve()` wiring (`go srv.StartAlertWatcher(context.Background())` after `NewServer`). **Step 4: PASS. Step 5: Commit** `console: alert watcher — keymail on fire/clear via operators' own grants` |
| + |
| +--- |
| + |
| +### Task 10: Console — metrics view assembly (bucket reads + cache) |
| + |
| +**Files:** |
| +- Create: `internal/console/metricsview.go`, `internal/console/metricsview_test.go` |
| + |
| +**Interfaces:** |
| +- Consumes: `release.Store`, `metrics` types, `Assemble`'s route data for host→app mapping (Task 12 uses that part). |
| +- Produces (consumed by Tasks 11-12): |
| + |
| +```go |
| +type UnitRow struct { |
| + Unit string // full unit name |
| + CPUFrac *float64 |
| + MemBytes *int64 |
| + ReadBytes, WriteBytes *int64 |
| +} |
| +type BoxMetricsView struct { |
| + Box string |
| + At time.Time // latest sample time |
| + CPU, Mem, Disk *float64 |
| + SparkCPU, SparkMem, SparkDisk string // SVG polyline points, "" when no data |
| + Units []UnitRow // sorted by CPUFrac desc, nil-CPU rows last |
| +} |
| +type AlertRow struct { |
| + Box, Rule string |
| + Value float64 |
| + FiredAt time.Time |
| +} |
| +// LoadMetrics reads every box under metrics/boxes/, the raw chunks of the |
| +// last 48 hours (missing hours skipped silently), and alerts/boxes/*.json. |
| +// Results are cached in-process for 60s (all callers share one bucket sweep). |
| +func (s *Server) LoadMetrics(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) |
| +// sparkPoints maps up to 48h of minute samples into an SVG polyline points |
| +// string on a 240x40 viewBox, nil samples breaking the line into gaps is NOT |
| +// attempted in v1 — nil values are simply skipped (documented in the comment). |
| +func sparkPoints(vals []*float64, w, h int) string |
| +``` |
| + |
| +Cache: unexported `metricsCache struct{ mu sync.Mutex; at time.Time; boxes []BoxMetricsView; alerts []AlertRow }` field on `Server`; TTL 60s; a failed sweep is never cached. Box discovery: `List(ctx, "metrics/boxes/")` and take the path segment after `boxes/`. Per box read `RawKey(box, HourOf(h))` for each of the last 48 UTC hours; `os.ErrNotExist` → skip. |
| + |
| +- [ ] **Step 1: Failing tests:** DirStore seeded with 2 hour-chunks for one box (via `metrics.WriteRaw`) + one alert object; construct `Server` the way existing console tests do (see `server_test.go` for the minimal `Config`); assert `LoadMetrics` returns 1 box with latest values from the newest sample, non-empty `SparkCPU`, units sorted CPU-desc, 1 `AlertRow`; assert a second call within TTL does not re-read (poison the store after first call — e.g. delete the objects — and assert same result returned). `sparkPoints([]*float64{p(0.1), nil, p(0.9)}, 240, 40)` → two coordinate pairs, y inverted (higher value = smaller y). |
| +- [ ] **Step 2: FAIL. Step 3: Implement. Step 4: PASS. Step 5: Commit** `console: metrics view assembly — 48h sweep, 60s cache, server-side sparklines` |
| + |
| +--- |
| + |
| +### Task 11: Console — operator metrics dashboard page |
| + |
| +**Files:** |
| +- Modify: `internal/console/server.go` (route + handler + `page` struct fields), `internal/console/templates/layout.html` only if a nav link for operators exists for /fleet (mirror however /fleet is linked; if it isn't, don't add one) |
| +- Create: `internal/console/templates/metrics.html` |
| +- Test: `internal/console/server_test.go` additions |
| + |
| +**Interfaces:** |
| +- Consumes: `LoadMetrics` (Task 10), operator gate pattern from `handleFleet` (`server.go:888-912`), `s.render` (`server.go:258`). |
| +- Produces: `GET /fleet/metrics` → `handleFleetMetrics`; `page` gains `Metrics []BoxMetricsView` and `MetricAlerts []AlertRow`. |
| + |
| +Handler (copy `handleFleet`'s gate EXACTLY — every failure path is plain `http.NotFound`, no styled 404): |
| + |
| +```go |
| +func (s *Server) handleFleetMetrics(w http.ResponseWriter, r *http.Request) { |
| + addr, raw, ok := s.session(r) |
| + if !ok { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + isOp, err := s.store.IsOperator(addr) |
| + if err != nil { |
| + log.Printf("console: fleet metrics: is-operator: %v", err) |
| + http.NotFound(w, r) |
| + return |
| + } |
| + if !isOp { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + boxes, alerts, err := s.LoadMetrics(r.Context()) |
| + if err != nil { |
| + log.Printf("console: fleet metrics: %v", err) |
| + s.renderMessage(w, r, http.StatusInternalServerError, "Unavailable", "Metrics could not be read.") |
| + return |
| + } |
| + s.render(w, http.StatusOK, "metrics", page{Title: "Fleet metrics", Address: addr, CSRF: raw, Metrics: boxes, MetricAlerts: alerts}) |
| +} |
| +``` |
| + |
| +Template `metrics.html` (`{{define "metrics"}}{{template "top" .}} ... {{template "bottom" .}}{{end}}`): alerts section first (red, only when non-empty), then one card per box — headline numbers (`{{printf "%.0f%%" ...}}` via a small `pct` template func added beside `strOrDash` in `server.go:33-77`, nil → "—"), three labelled inline sparklines (`<svg viewBox="0 0 240 40"><polyline fill="none" stroke="currentColor" points="{{.SparkCPU}}"/></svg>`), then the per-unit table (Unit / CPU / Memory / IO), nil cells rendered "—" never "0". |
| + |
| +- [ ] **Step 1: Failing tests:** non-operator session GETs `/fleet/metrics` → 404 byte-identical to an unrouted URL (compare against a GET of `/definitely-not-a-route`); operator session → 200 containing a seeded box label and a seeded unit name; signed-out → 404. |
| +- [ ] **Step 2: FAIL. Step 3: Implement. Step 4: PASS. Step 5: Commit** `console: operator metrics dashboard — needles, sparklines, noisy-neighbour table` |
| + |
| +--- |
| + |
| +### Task 12: Console — owner-facing app metrics on the app page |
| + |
| +**Files:** |
| +- Modify: `internal/console/server.go` (`handleApp` at :1010-1041, `page` struct), `internal/console/metricsview.go`, `internal/console/templates/app.html` |
| +- Test: `internal/console/server_test.go` additions |
| + |
| +**Interfaces:** |
| +- Consumes: `handleApp`'s existing `accountApp` gate + `Assemble` output (`AppView.Routes[].Host` — `view.go:171`), `activator.UnitName(host)` (`internal/activator/backend_unit.go:65`), the Task-10 cache. |
| +- Produces: |
| + |
| +```go |
| +type InstanceMetrics struct { |
| + Host string // the route host — how owners name their instances |
| + Box string |
| + CPUFrac *float64 |
| + MemBytes *int64 |
| + SparkCPU, SparkMem string |
| +} |
| +// OwnerAppMetrics filters the cached 48h sweep down to EXACTLY the units |
| +// carlos-app@<host>.service for the given hosts. It returns unit-level rows |
| +// only — no host totals, no other units, no alert data. The allowlist is the |
| +// hosts slice the caller derived from the app's OWN routes; nothing else can |
| +// enter the view. |
| +func (s *Server) OwnerAppMetrics(ctx context.Context, hosts []string) ([]InstanceMetrics, error) |
| +``` |
| + |
| +`page` gains `AppMetrics []InstanceMetrics`. In `handleApp`, after the existing `Assemble` call, collect `hosts` from the app's own `views[0].Routes` and populate `AppMetrics` (errors logged, page renders without metrics — metrics must never break the app page). `app.html` gains a "Resources" section: per instance-host, CPU/memory numbers + two sparklines; absent (hibernating) instances render "asleep — no recent readings", never zeros. |
| + |
| +- [ ] **Step 1: Failing tests:** |
| + - Seed metrics for units `carlos-app@owned.example.com.service` AND `carlos-app@other-tenant.example.com.service` on one box. An account member's GET of their app page (owning only `owned.example.com`) → 200 containing `owned.example.com` metrics markup and NOT containing the string `other-tenant` anywhere in the body. |
| + - The app page contains no host-level needle: assert the body does not contain the box's host CPU sentinel value seeded distinctively (e.g. host CPU 0.777 → "77.7" absent; pick sentinel values that cannot collide with unit numbers). |
| + - `OwnerAppMetrics` with empty hosts → empty, no error. |
| +- [ ] **Step 2: FAIL. Step 3: Implement. Step 4: PASS. Step 5: Commit** `console: owner app metrics — own units only, strict boundary` |
| + |
| +--- |
| + |
| +### Task 13: Docs, full verification, review, ship |
| + |
| +**Files:** |
| +- Modify: `README.md` (if it documents env vars/agent loops — add `CARLOS_ALERT_*` and the metrics/alert bucket prefixes where the status page's surface is documented), `docs/superpowers/specs/2026-08-02-host-process-monitoring-design.md` (only if implementation deviated — record deviations, don't rewrite history) |
| + |
| +- [ ] **Step 1:** `gofmt -l` on all touched files → empty; `go build ./... && go vet ./... && go test ./... -count=1` → all clean (expect 15 packages now with `internal/metrics`). |
| +- [ ] **Step 2:** Dispatch an **Opus** reviewer subagent (CLAUDE.md delegation rule — this touches production paths and a security-sensitive console surface) over the full branch diff, with the spec and this plan as context. Fix everything it finds at Critical/Important; use judgment on nits. |
| +- [ ] **Step 3:** Re-run the full gate after fixes. |
| +- [ ] **Step 4:** Push branch, open draft PR titled `monitoring: host & process metrics, alerts, operator + owner dashboards` with a body summarizing spec, storage shapes, and the two delivery paths, ending with the standard attribution footer. |
| +- [ ] **Step 5:** Deploy: nothing is in production yet and deploys are pre-approved. If the pinfra checkout / ship scripts are reachable from this environment, ship + promote per `scripts/ship-app.sh` conventions; otherwise document exact deploy steps (bootstrap unchanged — same units; the new loops arm via env vars: `CARLOS_ALERT_WEBHOOK` optional, thresholds optional, metrics loop arms automatically wherever the status loop is already armed) in the PR body. |