Merge pull request #121 from carlosframework/worktree-exec-instance-metrics
agent: exec-backed instances get per-instance metrics (#118)
11 files changed,
+609
−15
docs/superpowers/specs/2026-08-08-exec-instance-metrics-design.md+152 −0internal/activator/activator.go+29 −0internal/activator/activator_test.go+74 −1internal/activator/backend_exec.go+8 −0internal/console/metricsview.go+9 −3internal/host/agent.go+1 −1internal/host/metrics_loop.go+19 −6internal/host/metrics_probes.go+75 −4internal/host/metrics_probes_test.go+58 −0internal/metrics/pidparse.go+101 −0internal/metrics/pidparse_test.go+83 −0
diff --git a/docs/superpowers/specs/2026-08-08-exec-instance-metrics-design.md b/docs/superpowers/specs/2026-08-08-exec-instance-metrics-design.md| new file mode 100644 |
| index 000000000..d52cbbc25 |
| --- /dev/null |
| +++ b/docs/superpowers/specs/2026-08-08-exec-instance-metrics-design.md |
| @@ -0,0 +1,152 @@ |
| +# Exec-backed instance metrics (issue #118) |
| + |
| +**Date:** 2026-08-08 |
| +**Issue:** carlosframework/platform#118 — exec-backed instances report no |
| +CPU/memory: metrics sample unit cgroups only. |
| + |
| +## Problem |
| + |
| +The 2026-08-02 host-process-monitoring design defined "process" as |
| +**systemd unit**: the sampler walks `/sys/fs/cgroup/system.slice` for |
| +`carlos-*.service` cgroups. An exec-backed instance is a direct child of |
| +`carlos agent` inside the agent's own cgroup — no unit cgroup exists, so |
| +it is invisible to per-instance attribution. Its app page reads |
| +"asleep — no recent readings" even during long awake stretches, which is |
| +how the gap was found (owners can't tell "asleep" from "broken"). The |
| +monitoring design's owner-facing promise — a Resources section scoped to |
| +the app's own instances — can never be met for the backing most |
| +provisioned instances get by default. |
| + |
| +## Options considered |
| + |
| +1. **Pid-based sampling of activator children** (chosen). The activator |
| + already tracks every awake exec child (`actEntry.proc` holds the |
| + `*exec.Cmd`). Expose the awake host→pid map; the sampler reads |
| + `/proc/<pid>/{stat,status,io}` and publishes the readings under the |
| + same `carlos-app@<host>.service` key the console already matches. |
| +2. **Per-child cgroup at spawn** (`carlos.slice/instance-<host>` scope). |
| + Rejected for v1: needs `Delegate=yes` on the agent's unit fleet-wide |
| + (pinfra change + restart on every box), a cgroup-v2 |
| + no-internal-process workaround (the agent's own pid can't share a |
| + cgroup with controller-enabled children), and walker changes anyway — |
| + delegated children live under the agent's own unit cgroup, a level |
| + `listCarlosUnits` deliberately does not scan. Same data, far larger |
| + blast radius. |
| +3. **UI honesty only** ("not measured (exec-backed)"). Superseded: with |
| + option 1 the readings exist, so there is nothing to apologize for. |
| + Not implemented — old-agent boxes render the existing state-neutral |
| + strings until they take the new binary, a transitional state not |
| + worth a console change. |
| + |
| +## Design |
| + |
| +### Key decision: exec children publish under the unit-name key |
| + |
| +An exec-backed host has no `carlos-app@<host>.service` unit *by |
| +construction* (units are for the edge; instances are the agent's |
| +children), so the key can never collide with a real cgroup reading on |
| +the same box. Publishing under `activator.UnitName(host)` means: |
| + |
| +- **Wire shape unchanged** — `metrics.Sample.Units` is a string-keyed |
| + map; new entries are additive. Everything already in production |
| + buckets still parses, and old consoles render new-agent chunks as |
| + more unit rows. |
| +- **Console unchanged** — `OwnerAppMetrics` allowlists exactly |
| + `activator.UnitName(h)` per route host; exec readings match as-is. |
| + The operator dashboard's per-unit table picks them up the same way. |
| +- If a collision ever did occur, the cgroup reading wins (it accounts |
| + the whole subtree; the pid reading is one process). |
| + |
| +### Activator: exposing awake children |
| + |
| +- `childProc` gains `Pid() int` (the spawned `cmd.Process.Pid`). |
| +- `Service.AwakeExecPids() map[string]int` returns host→pid for entries |
| + that are awake, whose proc has not exited, and whose Proc implements |
| + `Pid() int`. Taken under `a.mu`, like every entry read. |
| +- `unitProc` (unit-backed tenants) implements no `Pid` and is naturally |
| + excluded — those instances already report via their unit cgroup. |
| +- Waking and stopping entries are excluded: no stable process to read. |
| + |
| +### Sampler: /proc readers |
| + |
| +New pure parsers in `internal/metrics` (fixture-tested, like the cgroup |
| +parsers), with thin `os.ReadFile` glue in `internal/host` (untested by |
| +the same convention as the other readers): |
| + |
| +- `/proc/<pid>/stat` → CPU microseconds: `(utime + stime)` clock ticks |
| + × 1e6 / USER_HZ. USER_HZ is a named constant 100 — the kernel fixes |
| + the /proc tick unit at 100 on every architecture CARLOS runs on, |
| + independent of CONFIG_HZ. Parsed after the last `)` (comm may contain |
| + spaces and parens). Feeds the existing `UnitCPUFrac` delta machinery |
| + unchanged; a respawned child's counter reset normally yields a nil |
| + fraction for that tick (`curUsec < prevUsec` guard, already present). |
| + The guard is not absolute: a same-window respawn whose new process |
| + has already accumulated more CPU than the dead one's last reading |
| + slips past it and blends one tick's fraction across the two |
| + processes — the identical, self-correcting edge the cgroup path has |
| + when a unit restart recreates its cgroup (prev-tick state is keyed |
| + by unit name, not by pid or cgroup inode, on both paths). One |
| + transient tick in a 60s-sampled view; accepted. |
| +- `/proc/<pid>/status` → `VmRSS` (memory), `VmHWM` (peak), kB → bytes. |
| + RSS, not `memory.current`: no page-cache attribution, which for "what |
| + your app uses" is the more honest number a single-process reading can |
| + give. |
| +- `/proc/<pid>/io` → `read_bytes` / `write_bytes` — cumulative since |
| + process start, same semantics as the cgroup `io.stat` counters. |
| + |
| +Every field fails toward nil independently, per the probe convention: a |
| +pid that vanished mid-read (raced its own exit) contributes nils, never |
| +zeros. |
| + |
| +### Wiring |
| + |
| +- `metricsLoop(children func() map[string]int)` — `agent.go` passes |
| + `svc.AwakeExecPids` (nil `svc` → nil func → cgroup-only, exactly |
| + today's behavior; `carlos edge` without hibernation is unaffected). |
| +- The units probe becomes a merge: cgroup readings (as today) plus one |
| + `unitRaw` per awake exec child keyed by `activator.UnitName(host)`. |
| + The merge is a pure function, unit-tested; cgroup wins collisions. |
| + |
| +### Known limits (documented, accepted) |
| + |
| +- **Grandchildren are not counted.** A pid reading covers one process; |
| + a cgroup covers a subtree. Carlos apps are single-process Go |
| + binaries; an app that forks workers under-reports. Option 2 remains |
| + the upgrade path if that ever matters. |
| +- **Sidecars are not counted** toward their host's reading (v1 scope: |
| + the instance child only). |
| +- **Pid-reuse race**: between `AwakeExecPids` and the /proc read, an |
| + exiting child's pid could in principle be recycled. The window is |
| + sub-millisecond within a 60s tick, the activator excludes |
| + already-exited procs, and the failure mode is one sample's nils or |
| + one stray reading — accepted. |
| +- **The operator dashboard shows the unit-name key for a unit systemd |
| + does not know.** An exec-backed instance's row reads |
| + `carlos-app@<host>.service` — its identity, and what the console |
| + matches — but `systemctl status` with that name on the box comes up |
| + empty (the process is a `carlos agent` child; its logs are in the |
| + agent's own journal). Accepted as the cost of the additive keying; |
| + `UnitRow`'s doc comment says so, and if it bites in practice the |
| + upgrade is an additive per-unit source marker in `UnitSample`, not a |
| + key change. |
| + |
| +## Testing |
| + |
| +- `internal/metrics`: fixture tests for the three parsers (mangled |
| + input → nil; comm-with-spaces stat lines; kB scaling). |
| +- `internal/activator`: `AwakeExecPids` over the fake-backend state |
| + machine — awake pid-bearing procs listed; waking/stopping/exited/ |
| + pid-less excluded. |
| +- `internal/host`: keying test (exec readings publish under |
| + `carlos-app@<host>.service` — get this wrong and readings match |
| + nothing) and merge test (cgroup+exec compose, collision precedence, |
| + both-empty → nil). The tick→`Sample.Units` flow is already pinned by |
| + the existing fixture tick tests; the composed production probe itself |
| + follows the package's untested-seam convention. |
| + |
| +## Deploy |
| + |
| +Ship the agent normally (promote → adopt → restart per deploy rules). |
| +No unit files, no pinfra, no console deploy required; readings appear |
| +for awake exec instances within one flush (≤5 min) of the restarted |
| +agent's first tick. |
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 9014bfe37..5500cee25 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -710,6 +710,35 @@ func (a *Service) beginStopLocked(e *actEntry) { |
| go a.stop(e) |
| } |
| +// AwakeExecPids reports the pid of every awake exec-backed child, keyed by |
| +// route host — the metrics sampler's seam (issue #118): an exec child runs |
| +// inside the agent's own cgroup, invisible to the per-unit cgroup walk, so |
| +// its per-instance attribution has to read /proc/<pid> instead, and only |
| +// the activator knows which pids those are. Included entries are awake, |
| +// not yet exited, and expose a pid: waking has no stable process yet, |
| +// stopping is on its way out, and a unit-backed tenant's proc (systemd- |
| +// driven, no Pid method) already reports through its own unit cgroup. Nil |
| +// when nothing qualifies — no readings, not an empty set of them. |
| +func (a *Service) AwakeExecPids() map[string]int { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + var out map[string]int |
| + for _, e := range a.entries { |
| + if e.state != actAwake || e.proc.Exited() { |
| + continue |
| + } |
| + p, ok := e.proc.(interface{ Pid() int }) |
| + if !ok { |
| + continue |
| + } |
| + if out == nil { |
| + out = make(map[string]int) |
| + } |
| + out[e.host] = p.Pid() |
| + } |
| + return out |
| +} |
| + |
| // Owns reports whether host has a live entry in ANY state — waking, |
| // awake, or stopping. This is the external-socket advisory's predicate |
| // (issue #17): while an entry exists, whatever is serving the host's |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index e861d928a..6afbbfa45 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -160,6 +160,11 @@ type fakeBackend struct { |
| // target). |
| execVerFn func() string |
| + // pidByHost, when a host is present, makes Start() return a pid-BEARING |
| + // proc (pidProc) for it — modeling ExecBackend's childProc. Hosts absent |
| + // from the map get the plain pid-less fakeProc, modeling unitProc. |
| + pidByHost map[string]int |
| + |
| // Sidecar co-spawn. sidecarStaged is HasSidecar's answer (false by |
| // default, so every pre-co-spawn test doubles as the "host without a |
| // sidecar spawns none" case); sidecarErr injects spawn failures. |
| @@ -215,7 +220,7 @@ func (b *fakeBackend) Seed(_ context.Context, rt registry.Route) error { |
| return b.seedErr |
| } |
| -func (b *fakeBackend) Start(context.Context, registry.Route) (Proc, string, error) { |
| +func (b *fakeBackend) Start(_ context.Context, rt registry.Route) (Proc, string, error) { |
| b.mu.Lock() |
| b.starts++ |
| b.order = append(b.order, "start:instance") |
| @@ -224,6 +229,7 @@ func (b *fakeBackend) Start(context.Context, registry.Route) (Proc, string, erro |
| startErr := b.startErr |
| proc := &fakeProc{label: "instance", rec: b} |
| b.procs = append(b.procs, proc) |
| + pid, hasPid := b.pidByHost[rt.Host] |
| b.mu.Unlock() |
| if startErr != nil { |
| return nil, "", startErr |
| @@ -235,9 +241,22 @@ func (b *fakeBackend) Start(context.Context, registry.Route) (Proc, string, erro |
| if fn != nil { |
| ver = fn() |
| } |
| + if hasPid { |
| + return &pidProc{fakeProc: proc, pid: pid}, ver, nil |
| + } |
| return proc, ver, nil |
| } |
| +// pidProc is fakeProc plus a pid — the shape ExecBackend's childProc has, |
| +// versus unitProc's pid-less one. AwakeExecPids distinguishes the two by |
| +// the optional Pid() interface, so the fakes must too. |
| +type pidProc struct { |
| + *fakeProc |
| + pid int |
| +} |
| + |
| +func (p *pidProc) Pid() int { return p.pid } |
| + |
| func (b *fakeBackend) NextDue(context.Context, registry.Route) (int64, error) { |
| b.mu.Lock() |
| defer b.mu.Unlock() |
| @@ -3248,3 +3267,57 @@ func TestExecBackendDoesNotNormalizeSquatterSocket(t *testing.T) { |
| t.Fatalf("squatter socket mode after Start = %o, want its own 600 — a socket our child did not bind is never ours to widen", fi.Mode().Perm()) |
| } |
| } |
| + |
| +// AwakeExecPids is the metrics sampler's seam (issue #118): the host→pid map |
| +// of awake exec children. Pid-less procs (unit-backed tenants — they report |
| +// via their own unit cgroup), dead children, and stopping entries must all |
| +// be excluded — a pid the sampler reads must belong to a process the |
| +// activator still vouches for. |
| +func TestAwakeExecPids(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + b.pidByHost = map[string]int{ |
| + "one.example.com": 4242, |
| + "three.example.com": 4343, |
| + } |
| + |
| + if got := a.AwakeExecPids(); got != nil { |
| + t.Fatalf("nothing awake: got %v, want nil", got) |
| + } |
| + |
| + for _, host := range []string{"one.example.com", "two.example.com", "three.example.com"} { |
| + release, err := a.beginRequest(context.Background(), testRoute(host)) |
| + if err != nil { |
| + t.Fatalf("beginRequest %s: %v", host, err) |
| + } |
| + release() |
| + } |
| + |
| + // two (pid-less, modeling a unit tenant) is excluded; one and three report. |
| + got := a.AwakeExecPids() |
| + if len(got) != 2 || got["one.example.com"] != 4242 || got["three.example.com"] != 4343 { |
| + t.Fatalf("got %v, want one=4242 three=4343 only", got) |
| + } |
| + |
| + // A stopping entry is excluded the moment it flips, not once the stop |
| + // finishes: hold three's stop open on its gate and look while it drains. |
| + pThree := b.procs[2] |
| + pThree.mu.Lock() |
| + pThree.stopGate = make(chan struct{}) |
| + pThree.mu.Unlock() |
| + if !a.Bounce("three.example.com") { |
| + t.Fatal("Bounce three: not awake?") |
| + } |
| + if got := a.AwakeExecPids(); len(got) != 1 || got["one.example.com"] != 4242 { |
| + t.Fatalf("with three stopping: got %v, want one=4242 only", got) |
| + } |
| + close(pThree.stopGate) |
| + |
| + // A dead-but-unreaped child is excluded even before a sweep notices it. |
| + pOne := b.procs[0] |
| + pOne.mu.Lock() |
| + pOne.dead = true |
| + pOne.mu.Unlock() |
| + if got := a.AwakeExecPids(); got != nil { |
| + t.Fatalf("with one dead: got %v, want nil", got) |
| + } |
| +} |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index c8514652e..e8c2a7e66 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -730,6 +730,14 @@ type childProc struct { |
| purge *edge.PurgeListener // nil if b.Cache was nil or ListenPurge failed |
| } |
| +// Pid reports the spawned child's OS pid — the seam AwakeExecPids reads so |
| +// the metrics sampler can find this child under /proc (issue #118: an exec |
| +// child lives in the agent's own cgroup, so per-pid reads are its only |
| +// per-instance attribution). cmd.Process is set once Start succeeded, which |
| +// is the only way a childProc is ever constructed, and stays valid for the |
| +// struct's life. |
| +func (p *childProc) Pid() int { return p.cmd.Process.Pid } |
| + |
| func (p *childProc) Exited() bool { |
| select { |
| case <-p.done: |
diff --git a/internal/console/metricsview.go b/internal/console/metricsview.go| index 7ee92eb31..4dd763e2a 100644 |
| --- a/internal/console/metricsview.go |
| +++ b/internal/console/metricsview.go |
| @@ -24,9 +24,15 @@ import ( |
| "github.com/carlosframework/platform/internal/metrics" |
| ) |
| -// UnitRow is one systemd unit's latest reading, ready to render, plus its |
| -// own 48h sparklines (the owner app-metrics view renders a unit on its own, |
| -// with no host row to borrow a shape from). |
| +// UnitRow is one per-instance reading, ready to render, plus its own 48h |
| +// sparklines (the owner app-metrics view renders a unit on its own, with no |
| +// host row to borrow a shape from). Keyed by unit name, but the name is the |
| +// reading's IDENTITY, not a promise that systemd knows it: an exec-backed |
| +// instance has no unit at all — it is a child of `carlos agent`, sampled by |
| +// pid — and publishes under the carlos-app@<host>.service name it WOULD |
| +// have (issue #118), so `systemctl status` on a box can come up empty for a |
| +// row that is live here. The route host inside the name is still the thing |
| +// to grep the agent's journal for. |
| type UnitRow struct { |
| Unit string // full unit name |
| CPUFrac *float64 |
diff --git a/internal/host/agent.go b/internal/host/agent.go| index a6612ebfc..41cecde9d 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -162,7 +162,7 @@ func RunAgent(args []string) { |
| go statusLoop(reg) |
| go domainsLoop(reg, peekAcmeCache(args)) |
| go instancesLoop(reg) |
| - go metricsLoop() |
| + go metricsLoop(svc) |
| go configSyncLoop(reg, svc, dd) |
| go configGenLoop(reg, svc, dd) |
| go socketAdvisoryLoop(reg, svc) |
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go| index a0de7bd66..4d0b3753d 100644 |
| --- a/internal/host/metrics_loop.go |
| +++ b/internal/host/metrics_loop.go |
| @@ -20,6 +20,7 @@ import ( |
| "path/filepath" |
| "time" |
| + "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/metrics" |
| "github.com/carlosframework/platform/internal/release" |
| "github.com/carlosframework/platform/internal/status" |
| @@ -617,14 +618,24 @@ func httpPostWebhook(url string, body []byte) error { |
| // realSampleProbes wires sampleProbes to the actual OS/cgroup readers in |
| // metrics_probes.go and probes.go — the production seam metricsLoop uses, |
| // never called from a test (host_test.go's convention: tests inject |
| -// fixtures, never touch /proc or a syscall). |
| -func realSampleProbes() sampleProbes { |
| +// fixtures, never touch /proc or a syscall). svc feeds the units probe the |
| +// activator's awake exec children (AwakeExecPids); nil — no activator |
| +// armed, e.g. plain `carlos edge` — keeps the units probe cgroup-only, |
| +// exactly the pre-#118 behavior, the same nilable-svc convention every |
| +// other activator-fed loop in this package uses. |
| +func realSampleProbes(svc *activator.Service) sampleProbes { |
| + units := realUnitReadings |
| + if svc != nil { |
| + units = func() map[string]unitRaw { |
| + return mergeUnitReadings(realUnitReadings(), execInstanceReadings(svc.AwakeExecPids(), readExecInstanceRaw)) |
| + } |
| + } |
| return sampleProbes{ |
| procStat: readProcStat, |
| loadavg: readLoadAvg, |
| memFrac: readMemFrac, |
| diskFrac: readDiskFrac, |
| - units: realUnitReadings, |
| + units: units, |
| now: time.Now, |
| } |
| } |
| @@ -698,8 +709,10 @@ func loadAlertState(path, box string) metrics.AlertState { |
| // still a valid deployment" shape every other optional loop in this package |
| // uses. Flush runs every 5th tick (5 minutes at the 60s tick cadence); |
| // alertTick runs every tick, since a threshold breach deserves faster |
| -// detection than a metrics flush does. |
| -func metricsLoop() { |
| +// detection than a metrics flush does. svc feeds the units probe the |
| +// activator's awake exec children (nil when no activator is armed) — see |
| +// realSampleProbes. |
| +func metricsLoop(svc *activator.Service) { |
| // This loop is always on and shares a process with the agent that serves |
| // this box's traffic, so a panic in it — an unreadable cgroup shape, a |
| // nil no one expected — must not take the box down with it. Monitoring |
| @@ -732,7 +745,7 @@ func metricsLoop() { |
| log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to bucket %s", bucket) |
| sm := &sampler{} |
| - probes := realSampleProbes() |
| + probes := realSampleProbes(svc) |
| tick := 0 |
| for { |
| time.Sleep(60 * time.Second) |
diff --git a/internal/host/metrics_probes.go b/internal/host/metrics_probes.go| index 899247438..ef0ffc744 100644 |
| --- a/internal/host/metrics_probes.go |
| +++ b/internal/host/metrics_probes.go |
| @@ -4,17 +4,21 @@ |
| // wrapping the pure parsers in internal/metrics — and the readers are |
| // deliberately untested by the same convention as probes.go readers; |
| // internal/metrics tests exercise the parsers with fixtures instead. The one |
| -// exception is listCarlosUnits: it is a directory WALK, not a single read, and |
| -// which directories it walks is a decision that has already been wrong once |
| -// (see its doc comment), so it takes its root as an argument and is pinned by |
| -// metrics_probes_test.go against a temp-dir fixture. |
| +// exceptions are the functions with decisions in them: listCarlosUnits is a |
| +// directory WALK, not a single read, and which directories it walks has |
| +// already been wrong once (see its doc comment), so it takes its root as an |
| +// argument and is pinned by metrics_probes_test.go against a temp-dir |
| +// fixture; execInstanceReadings/mergeUnitReadings are pure (keying and |
| +// precedence for exec-child readings, issue #118) and are pinned there too. |
| package host |
| import ( |
| "os" |
| "path/filepath" |
| + "strconv" |
| "strings" |
| + "github.com/carlosframework/platform/internal/activator" |
| "github.com/carlosframework/platform/internal/metrics" |
| ) |
| @@ -133,6 +137,73 @@ type unitRaw struct { |
| writeBytes *int64 |
| } |
| +// readExecInstanceRaw reads one awake exec child's resource usage from |
| +// /proc/<pid>/{stat,status,io} — the per-pid counterpart of readUnitRaw for |
| +// instances that have no unit cgroup to read (issue #118: an exec-backed |
| +// instance is a direct child of `carlos agent`, inside the agent's own |
| +// cgroup, so the unit walk above can never see it). Same contract as |
| +// readUnitRaw: each field reads and fails independently, toward nil — a pid |
| +// that vanished mid-read (raced its own exit) contributes nils, never zeros. |
| +// Known limit, documented in the design: a pid reading covers ONE process, |
| +// not a subtree — an app that forks workers under-reports here where a |
| +// cgroup would not. |
| +func readExecInstanceRaw(pid int) unitRaw { |
| + dir := filepath.Join("/proc", strconv.Itoa(pid)) |
| + raw := unitRaw{} |
| + if b, err := os.ReadFile(filepath.Join(dir, "stat")); err == nil { |
| + raw.cpuUsec = metrics.ParsePidStatCPUUsec(string(b)) |
| + } |
| + if b, err := os.ReadFile(filepath.Join(dir, "status")); err == nil { |
| + raw.memBytes, raw.memPeakBytes = metrics.ParsePidStatusMem(string(b)) |
| + } |
| + if b, err := os.ReadFile(filepath.Join(dir, "io")); err == nil { |
| + raw.readBytes, raw.writeBytes = metrics.ParsePidIO(string(b)) |
| + } |
| + return raw |
| +} |
| + |
| +// execInstanceReadings turns the activator's awake host→pid map into unit |
| +// readings keyed by the SAME carlos-app@<host>.service name the console's |
| +// owner view matches (activator.UnitName) — which is what makes the whole |
| +// fix additive: the wire shape (Sample.Units, a string-keyed map) and the |
| +// console's allowlist both take the new entries as-is. An exec-backed host |
| +// has no such unit by construction (units are for the edge; instances are |
| +// the agent's children), so the key cannot collide with a real cgroup |
| +// reading on this box; mergeUnitReadings still defines the winner in case |
| +// that ever changes. read is a parameter (readExecInstanceRaw in |
| +// production) so the keying is testable without /proc. |
| +func execInstanceReadings(pids map[string]int, read func(pid int) unitRaw) map[string]unitRaw { |
| + if len(pids) == 0 { |
| + return nil |
| + } |
| + out := make(map[string]unitRaw, len(pids)) |
| + for host, pid := range pids { |
| + out[activator.UnitName(host)] = read(pid) |
| + } |
| + return out |
| +} |
| + |
| +// mergeUnitReadings composes the units probe's two sources: the cgroup walk |
| +// (every systemd-backed CARLOS unit) plus the per-pid readings for awake |
| +// exec children. On a key collision the cgroup reading wins — it accounts a |
| +// whole subtree where the pid reading covers one process, so it can only be |
| +// more complete. Both-empty stays nil (no readings, not an empty set), and |
| +// a nil cgroup map with live exec children still reports the children — a |
| +// box whose cgroup walk failed has not stopped running instances. |
| +func mergeUnitReadings(cgroup, exec map[string]unitRaw) map[string]unitRaw { |
| + if len(exec) == 0 { |
| + return cgroup |
| + } |
| + out := make(map[string]unitRaw, len(cgroup)+len(exec)) |
| + for name, u := range exec { |
| + out[name] = u |
| + } |
| + for name, u := range cgroup { |
| + out[name] = u |
| + } |
| + return out |
| +} |
| + |
| // readUnitRaw reads a unit's resource usage from the cgroup v2 files in |
| // unitPath — the directory listCarlosUnits reported for that unit, which for |
| // an instantiated unit is NOT <cgroupSliceDir>/<name>. Each file read failure |
diff --git a/internal/host/metrics_probes_test.go b/internal/host/metrics_probes_test.go| index 8ea66188d..bf9d7e443 100644 |
| --- a/internal/host/metrics_probes_test.go |
| +++ b/internal/host/metrics_probes_test.go |
| @@ -88,3 +88,61 @@ func TestReadUnitRawReadsFromUnitDir(t *testing.T) { |
| t.Errorf("absent files must stay nil, got peak=%v r=%v w=%v", raw.memPeakBytes, raw.readBytes, raw.writeBytes) |
| } |
| } |
| + |
| +// Exec children publish under the SAME carlos-app@<host>.service key the |
| +// console's owner allowlist matches (activator.UnitName) — that keying is |
| +// the whole fix for issue #118, so it's pinned here: get it wrong and the |
| +// readings publish fine but match nothing, silently reopening the gap. |
| +func TestExecInstanceReadingsKeysByUnitName(t *testing.T) { |
| + cpu := int64(5_000_000) |
| + got := execInstanceReadings(map[string]int{"lpay.bde.oncarlos.com": 4242}, func(pid int) unitRaw { |
| + if pid != 4242 { |
| + t.Fatalf("read called with pid %d, want 4242", pid) |
| + } |
| + return unitRaw{cpuUsec: &cpu} |
| + }) |
| + raw, ok := got["carlos-app@lpay.bde.oncarlos.com.service"] |
| + if !ok || len(got) != 1 { |
| + t.Fatalf("got keys %v, want exactly carlos-app@lpay.bde.oncarlos.com.service", got) |
| + } |
| + if raw.cpuUsec == nil || *raw.cpuUsec != cpu { |
| + t.Fatalf("reading not carried through: %v", raw.cpuUsec) |
| + } |
| + |
| + if execInstanceReadings(nil, nil) != nil { |
| + t.Fatal("no pids must be nil, not an empty map") |
| + } |
| +} |
| + |
| +func TestMergeUnitReadings(t *testing.T) { |
| + v := func(n int64) unitRaw { return unitRaw{cpuUsec: &n} } |
| + deref := func(u unitRaw) int64 { return *u.cpuUsec } |
| + |
| + // Both sources present: union, cgroup wins the collision (it accounts a |
| + // whole subtree; the pid reading is one process). |
| + got := mergeUnitReadings( |
| + map[string]unitRaw{"carlos-edge.service": v(1), "carlos-app@both.example.com.service": v(2)}, |
| + map[string]unitRaw{"carlos-app@x.example.com.service": v(3), "carlos-app@both.example.com.service": v(4)}, |
| + ) |
| + if len(got) != 3 { |
| + t.Fatalf("got %d keys, want 3", len(got)) |
| + } |
| + if deref(got["carlos-app@both.example.com.service"]) != 2 { |
| + t.Fatal("collision must keep the cgroup reading") |
| + } |
| + if deref(got["carlos-edge.service"]) != 1 || deref(got["carlos-app@x.example.com.service"]) != 3 { |
| + t.Fatal("non-colliding readings must survive from both sides") |
| + } |
| + |
| + // A failed cgroup walk (nil) with live exec children still reports them. |
| + got = mergeUnitReadings(nil, map[string]unitRaw{"carlos-app@x.example.com.service": v(3)}) |
| + if len(got) != 1 { |
| + t.Fatalf("nil cgroup side: got %v, want the exec reading alone", got) |
| + } |
| + |
| + // No exec children: the cgroup map passes through untouched — including |
| + // nil staying nil ("no readings", not an empty set of them). |
| + if mergeUnitReadings(nil, nil) != nil { |
| + t.Fatal("both-nil must stay nil") |
| + } |
| +} |
diff --git a/internal/metrics/pidparse.go b/internal/metrics/pidparse.go| new file mode 100644 |
| index 000000000..1bb70e736 |
| --- /dev/null |
| +++ b/internal/metrics/pidparse.go |
| @@ -0,0 +1,101 @@ |
| +// /proc/<pid> parsers — the per-process counterpart of the cgroup parsers in |
| +// parse.go, for instances that have no unit cgroup to read: an exec-backed |
| +// instance is a direct child of `carlos agent` inside the agent's own cgroup |
| +// (issue #118), so its readings come from /proc/<pid>/{stat,status,io} |
| +// instead. Same package contract as everything else here: parse text the |
| +// caller read, fail toward nil, never touch the OS. |
| +package metrics |
| + |
| +import ( |
| + "strconv" |
| + "strings" |
| +) |
| + |
| +// userHZ is the tick unit of /proc/<pid>/stat's utime/stime fields. The |
| +// kernel fixes the userspace-visible USER_HZ at 100 on every architecture |
| +// CARLOS runs on (it is independent of the kernel's internal CONFIG_HZ), so |
| +// a constant is the honest value — there is no portable syscall to ask for |
| +// it without cgo. |
| +const userHZ = 100 |
| + |
| +// ParsePidStatCPUUsec parses /proc/<pid>/stat and returns utime+stime as CPU |
| +// microseconds — the same unit as cpu.stat's usage_usec, so the reading |
| +// feeds UnitCPUFrac's delta machinery unchanged. The comm field (2) is the |
| +// process name in parentheses and may itself contain spaces and parens, so |
| +// fields are located from the LAST ')' — counting from the front would |
| +// misparse any binary with a space in its name. After that anchor the line |
| +// reads: state ppid pgrp session tty tpgid flags minflt cminflt majflt |
| +// cmajflt utime stime — utime and stime at offsets 11 and 12. Too few |
| +// fields or a non-numeric tick count -> nil. |
| +func ParsePidStatCPUUsec(text string) *int64 { |
| + i := strings.LastIndexByte(text, ')') |
| + if i < 0 { |
| + return nil |
| + } |
| + fields := strings.Fields(text[i+1:]) |
| + if len(fields) < 13 { |
| + return nil |
| + } |
| + utime, err1 := strconv.ParseInt(fields[11], 10, 64) |
| + stime, err2 := strconv.ParseInt(fields[12], 10, 64) |
| + if err1 != nil || err2 != nil { |
| + return nil |
| + } |
| + usec := (utime + stime) * (1e6 / userHZ) |
| + return &usec |
| +} |
| + |
| +// ParsePidStatusMem parses /proc/<pid>/status and returns VmRSS (resident |
| +// set) and VmHWM (its peak) in bytes. RSS rather than anything |
| +// cache-inclusive on purpose: a per-process reading cannot attribute page |
| +// cache the way memory.current does, and for "what your app uses" resident |
| +// memory is the honest number it CAN give. Each field fails toward nil |
| +// independently — the kernel prints these in kB always, so any other unit |
| +// suffix means the line is being misread and nils that field rather than |
| +// guessing a scale. |
| +func ParsePidStatusMem(text string) (rss, hwm *int64) { |
| + for _, line := range strings.Split(text, "\n") { |
| + fields := strings.Fields(line) |
| + if len(fields) != 3 || fields[2] != "kB" { |
| + continue |
| + } |
| + v, err := strconv.ParseInt(fields[1], 10, 64) |
| + if err != nil { |
| + continue |
| + } |
| + bytes := v * 1024 |
| + switch fields[0] { |
| + case "VmRSS:": |
| + rss = &bytes |
| + case "VmHWM:": |
| + hwm = &bytes |
| + } |
| + } |
| + return rss, hwm |
| +} |
| + |
| +// ParsePidIO parses /proc/<pid>/io and returns read_bytes / write_bytes — |
| +// actual block-device I/O, cumulative since process start, the same |
| +// semantics as the cgroup io.stat counters. rchar/wchar are deliberately |
| +// ignored: they count every read/write syscall including cache hits, a |
| +// different (and much larger) number. Each field fails toward nil |
| +// independently. |
| +func ParsePidIO(text string) (rbytes, wbytes *int64) { |
| + for _, line := range strings.Split(text, "\n") { |
| + key, val, ok := strings.Cut(line, ":") |
| + if !ok { |
| + continue |
| + } |
| + v, err := strconv.ParseInt(strings.TrimSpace(val), 10, 64) |
| + if err != nil { |
| + continue |
| + } |
| + switch key { |
| + case "read_bytes": |
| + rbytes = &v |
| + case "write_bytes": |
| + wbytes = &v |
| + } |
| + } |
| + return rbytes, wbytes |
| +} |
diff --git a/internal/metrics/pidparse_test.go b/internal/metrics/pidparse_test.go| new file mode 100644 |
| index 000000000..85d8dfdd0 |
| --- /dev/null |
| +++ b/internal/metrics/pidparse_test.go |
| @@ -0,0 +1,83 @@ |
| +package metrics |
| + |
| +import "testing" |
| + |
| +func TestParsePidStatCPUUsec(t *testing.T) { |
| + // pid (comm) state ppid pgrp session tty tpgid flags minflt cminflt |
| + // majflt cmajflt utime stime ... — utime=250 stime=150 ticks at USER_HZ |
| + // 100 is 4.0 CPU-seconds = 4_000_000 usec. |
| + line := "1234 (lpay) S 1 1234 1234 0 -1 4194560 900 0 0 0 250 150 0 0 20 0 8 0 12345 1000000 500 18446744073709551615 1 1 0 0 0 0 0 0 0 0 0 0 17 3 0 0 0 0 0" |
| + v := ParsePidStatCPUUsec(line) |
| + if v == nil || *v != 4_000_000 { |
| + t.Fatalf("got %v, want 4000000", v) |
| + } |
| + |
| + // comm may contain spaces AND parens; everything before the LAST ')' is |
| + // name, so the field offsets must be taken from there. |
| + tricky := "1234 (my app) (v2)) R 1 1234 1234 0 -1 4194560 900 0 0 0 100 100 0 0 20 0 8 0 12345 1000000 500 0 1 1 0 0 0 0 0 0 0 0 0 0 17 3 0 0 0 0 0" |
| + v = ParsePidStatCPUUsec(tricky) |
| + if v == nil || *v != 2_000_000 { |
| + t.Fatalf("paren-comm: got %v, want 2000000", v) |
| + } |
| + |
| + for _, bad := range []string{ |
| + "", |
| + "no parens at all", |
| + "1234 (lpay) S 1 2", // too few fields after comm |
| + "1234 (lpay) S 1 2 3 4 5 6 7 8 9 10 x 150 0", // non-numeric utime |
| + } { |
| + if ParsePidStatCPUUsec(bad) != nil { |
| + t.Fatalf("garbage %q must be nil", bad) |
| + } |
| + } |
| +} |
| + |
| +func TestParsePidStatusMem(t *testing.T) { |
| + text := "Name:\tlpay\nVmPeak:\t 20000 kB\nVmHWM:\t 9000 kB\nVmRSS:\t 8192 kB\nThreads:\t8\n" |
| + rss, hwm := ParsePidStatusMem(text) |
| + if rss == nil || *rss != 8192*1024 { |
| + t.Fatalf("rss: got %v, want %d", rss, 8192*1024) |
| + } |
| + if hwm == nil || *hwm != 9000*1024 { |
| + t.Fatalf("hwm: got %v, want %d", hwm, 9000*1024) |
| + } |
| + |
| + // Each field fails toward nil independently: a kernel thread has no VmRSS |
| + // line at all, and a garbled VmHWM must not take VmRSS down with it. |
| + rss, hwm = ParsePidStatusMem("Name:\tx\nVmRSS:\t 100 kB\nVmHWM:\tgarbage kB\n") |
| + if rss == nil || *rss != 100*1024 || hwm != nil { |
| + t.Fatalf("independent fields: got rss=%v hwm=%v", rss, hwm) |
| + } |
| + |
| + rss, hwm = ParsePidStatusMem("") |
| + if rss != nil || hwm != nil { |
| + t.Fatal("empty must be both-nil") |
| + } |
| + |
| + // A VmRSS value not in kB (the kernel never emits another unit, so an |
| + // unexpected one means we're misreading) must be nil, not a wrong scale. |
| + rss, _ = ParsePidStatusMem("VmRSS:\t 100 mB\n") |
| + if rss != nil { |
| + t.Fatal("non-kB unit must be nil") |
| + } |
| +} |
| + |
| +func TestParsePidIO(t *testing.T) { |
| + text := "rchar: 999\nwchar: 888\nsyscr: 10\nsyscw: 20\nread_bytes: 4096\nwrite_bytes: 8192\ncancelled_write_bytes: 0\n" |
| + r, w := ParsePidIO(text) |
| + if r == nil || *r != 4096 || w == nil || *w != 8192 { |
| + t.Fatalf("got r=%v w=%v, want 4096/8192", r, w) |
| + } |
| + |
| + // Independent nils: rchar/wchar (which count ALL io including cached) |
| + // must never stand in for the missing block-io counters. |
| + r, w = ParsePidIO("rchar: 999\nread_bytes: 4096\n") |
| + if r == nil || *r != 4096 || w != nil { |
| + t.Fatalf("independent fields: got r=%v w=%v", r, w) |
| + } |
| + |
| + r, w = ParsePidIO("") |
| + if r != nil || w != nil { |
| + t.Fatal("empty must be both-nil") |
| + } |
| +} |