| new file mode 100644 |
| index 000000000..6b99e7164 |
| --- /dev/null |
| +++ b/internal/host/metrics_loop.go |
| @@ -0,0 +1,428 @@ |
| +// The metrics sampler loop — the box-side half of host-process monitoring. |
| +// Every 60s it samples the whole host plus every monitored systemd unit |
| +// (sampler.tick), evaluates on-box alert rules against the freshest sample |
| +// (alertTick), and every fifth tick flushes the buffered samples out to the |
| +// deployment bucket as raw chunks, rolling up and pruning across hour |
| +// boundaries (metricsFlush). The sampler/probes split mirrors statusLoop's |
| +// own status.Probes seam: sampleProbes is the injectable fixture bundle so |
| +// host_test.go never touches /proc, a cgroup file, or a real clock. |
| +package host |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "net/http" |
| + "os" |
| + "path/filepath" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/metrics" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/status" |
| +) |
| + |
| +// sampleProbes is the injectable fixture bundle sampler.tick reads from each |
| +// tick, like status.Probes for statusTick. now is its own seam (rather than |
| +// time.Now called directly) so tests control the elapsed time between ticks |
| +// that unit CPU deltas depend on. |
| +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: the previous /proc/stat |
| +// reading (for host CPU), the previous per-unit cgroup usage_usec reading |
| +// (for unit CPU), the wall-clock time of the previous tick (unit CPU needs |
| +// elapsed wall time, not just the counter delta), and the buffered samples |
| +// awaiting their next flush. |
| +type sampler struct { |
| + prevCPU *metrics.CPUCounters |
| + prevUnit map[string]*int64 // unit -> cpu usec at previous tick |
| + prevAt time.Time |
| + buf []metrics.Sample |
| +} |
| + |
| +// tick reads p once and appends one metrics.Sample to sm.buf. The very |
| +// first tick on a fresh sampler has no previous reading to diff against, so |
| +// its host and unit CPU fractions are nil — not a guessed zero — exactly |
| +// like metrics.CPUFrac/UnitCPUFrac's own nil-on-no-prior-reading contract. |
| +func (sm *sampler) tick(p sampleProbes) { |
| + now := p.now() |
| + curCPU := p.procStat() |
| + l1, l5, l15 := p.loadavg() |
| + mem := p.memFrac() |
| + disk := p.diskFrac() |
| + units := p.units() |
| + |
| + var elapsed time.Duration |
| + if !sm.prevAt.IsZero() { |
| + elapsed = now.Sub(sm.prevAt) |
| + } |
| + |
| + hostCPU := metrics.CPUFrac(sm.prevCPU, curCPU) |
| + |
| + var unitSamples map[string]metrics.UnitSample |
| + var newPrevUnit map[string]*int64 |
| + if units != nil { |
| + unitSamples = make(map[string]metrics.UnitSample, len(units)) |
| + newPrevUnit = make(map[string]*int64, len(units)) |
| + for name, u := range units { |
| + var prevUsec *int64 |
| + if sm.prevUnit != nil { |
| + prevUsec = sm.prevUnit[name] |
| + } |
| + unitSamples[name] = metrics.UnitSample{ |
| + CPUFrac: metrics.UnitCPUFrac(prevUsec, u.cpuUsec, elapsed), |
| + MemBytes: u.memBytes, |
| + MemPeakBytes: u.memPeakBytes, |
| + ReadBytes: u.readBytes, |
| + WriteBytes: u.writeBytes, |
| + } |
| + newPrevUnit[name] = u.cpuUsec |
| + } |
| + } |
| + |
| + sm.buf = append(sm.buf, metrics.Sample{ |
| + At: now, |
| + Host: metrics.HostSample{ |
| + CPUFrac: hostCPU, |
| + Load1: l1, |
| + Load5: l5, |
| + Load15: l15, |
| + MemFrac: mem, |
| + DiskFrac: disk, |
| + }, |
| + Units: unitSamples, |
| + }) |
| + |
| + sm.prevCPU = curCPU |
| + sm.prevUnit = newPrevUnit |
| + sm.prevAt = now |
| +} |
| + |
| +// metricsRawPrefix is the bucket prefix under which a box's raw chunks live |
| +// — metrics.RawKey's own prefix, duplicated here only for List's sake (List |
| +// takes a bare prefix, not a templated key). |
| +func metricsRawPrefix(box string) string { |
| + return "metrics/boxes/" + box + "/raw/" |
| +} |
| + |
| +// metricsFlush writes sm.buf out as raw chunks grouped by each sample's own |
| +// hour — never mixing two hours into one chunk, even when a single flush |
| +// call spans a boundary. Samples in the current hour (relative to now) are |
| +// written as that hour's chunk and kept in buf (the current hour's chunk |
| +// gets rewritten wholesale on every flush as more samples accumulate, |
| +// mirroring WriteRaw's own "replace whatever is there" contract). Samples |
| +// from any older hour are written to their own chunk, rolled up via |
| +// RollupHour+UpsertRollup, and then dropped from buf — that hour is done, |
| +// buf must not grow unbounded by carrying it forever. Pruning of raw chunks |
| +// older than 48h runs only when an hour actually rolled over (there is |
| +// nothing new to prune otherwise). |
| +func metricsFlush(ctx context.Context, s release.Store, box string, sm *sampler, now time.Time) error { |
| + if len(sm.buf) == 0 { |
| + return nil |
| + } |
| + curHour := metrics.HourOf(now) |
| + |
| + curSamples := make([]metrics.Sample, 0, len(sm.buf)) |
| + oldByHour := map[string][]metrics.Sample{} |
| + var oldHours []string |
| + for _, samp := range sm.buf { |
| + h := metrics.HourOf(samp.At) |
| + if h == curHour { |
| + curSamples = append(curSamples, samp) |
| + continue |
| + } |
| + if _, ok := oldByHour[h]; !ok { |
| + oldHours = append(oldHours, h) |
| + } |
| + oldByHour[h] = append(oldByHour[h], samp) |
| + } |
| + |
| + for _, hour := range oldHours { |
| + samples := oldByHour[hour] |
| + if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: hour, Samples: samples}); err != nil { |
| + return err |
| + } |
| + agg := metrics.RollupHour(samples) |
| + if err := metrics.UpsertRollup(ctx, s, box, samples[0].At, agg); err != nil { |
| + return err |
| + } |
| + } |
| + |
| + if len(oldHours) > 0 { |
| + if err := pruneStaleRaw(ctx, s, box, now); err != nil { |
| + return err |
| + } |
| + } |
| + |
| + if len(curSamples) > 0 { |
| + if err := metrics.WriteRaw(ctx, s, metrics.RawChunk{V: 1, Box: box, Hour: curHour, Samples: curSamples}); err != nil { |
| + return err |
| + } |
| + } |
| + |
| + sm.buf = curSamples |
| + return nil |
| +} |
| + |
| +// pruneStaleRaw deletes this box's raw chunks older than 48h — List + |
| +// metrics.StaleRawKeys (which never returns a key it cannot parse — see its |
| +// own doc comment) + Delete. |
| +func pruneStaleRaw(ctx context.Context, s release.Store, box string, now time.Time) error { |
| + keys, err := s.List(ctx, metricsRawPrefix(box)) |
| + if err != nil { |
| + return err |
| + } |
| + for _, key := range metrics.StaleRawKeys(keys, now, 48*time.Hour) { |
| + if err := s.Delete(ctx, key); err != nil { |
| + return err |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// alertTick evaluates rules against latest, persists the resulting |
| +// AlertState to statePath (create parent dir; write via temp+rename — never |
| +// leaving a half-written state file a restart could load), publishes the |
| +// new state to metrics.AlertKey(box) only when a rule's firing state |
| +// actually changed (Evaluate emits Events exactly on those transitions — |
| +// see its own doc comment — so "any events" is exactly the "state changed" |
| +// test), and posts each event to webhookURL ("" skips the webhook |
| +// entirely). A webhook post failure is logged and otherwise ignored: |
| +// alertTick's job is to persist and publish reliably, and it must not let an |
| +// unreachable third party stop that — the console watcher polling |
| +// AlertKey(box) is the reliable channel, the webhook is best-effort. |
| +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 { |
| + |
| + obs := metrics.Observation{ |
| + At: latest.At, |
| + CPUFrac: latest.Host.CPUFrac, |
| + MemFrac: latest.Host.MemFrac, |
| + DiskFrac: latest.Host.DiskFrac, |
| + LagSecs: nil, // v1: no lag reading feeds the agent loop yet |
| + } |
| + newState, events := metrics.Evaluate(rules, *st, obs) |
| + |
| + if err := persistAlertState(statePath, newState); err != nil { |
| + return err |
| + } |
| + |
| + if len(events) > 0 { |
| + body, err := json.MarshalIndent(newState, "", " ") |
| + if err != nil { |
| + return err |
| + } |
| + if err := putAlertState(ctx, s, metrics.AlertKey(box), body); err != nil { |
| + return err |
| + } |
| + } |
| + |
| + for _, ev := range events { |
| + if webhookURL == "" { |
| + continue |
| + } |
| + evBody, err := json.Marshal(ev) |
| + if err != nil { |
| + log.Printf("host: alert event %s: marshal: %v", ev.Rule, err) |
| + continue |
| + } |
| + if err := post(webhookURL, evBody); err != nil { |
| + log.Printf("host: alert webhook post (rule %s): %v", ev.Rule, err) |
| + } |
| + } |
| + |
| + *st = newState |
| + return nil |
| +} |
| + |
| +// persistAlertState writes st to path as indented JSON via the repo's |
| +// standard temp+rename idiom (internal/appconfig/sync.go's renderEnv, |
| +// internal/release/store.go's DirStore.Put): a crashed write must never |
| +// leave a half-written state file behind for the next boot to load and |
| +// choke on. The parent directory (<dataDir>/monitor/) is created if |
| +// missing — this is the first thing that ever writes into it. |
| +func persistAlertState(path string, st metrics.AlertState) error { |
| + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { |
| + return err |
| + } |
| + body, err := json.MarshalIndent(st, "", " ") |
| + if err != nil { |
| + return err |
| + } |
| + tmp := path + ".tmp" |
| + if err := os.WriteFile(tmp, body, 0o644); err != nil { |
| + return err |
| + } |
| + if err := os.Rename(tmp, path); err != nil { |
| + os.Remove(tmp) |
| + return err |
| + } |
| + return nil |
| +} |
| + |
| +// putAlertState upserts key with a replace-or-create Put, retrying once on |
| +// an ErrCASConflict — the same read-etag/Put/retry-once dance WriteRaw and |
| +// UpsertRollup use, since this box is the only writer that should ever be |
| +// targeting its own alert key. |
| +func putAlertState(ctx context.Context, s release.Store, key string, body []byte) error { |
| + etag, err := currentAlertEtag(ctx, s, key) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := s.Put(ctx, key, body, etag); err != nil { |
| + if !errors.Is(err, release.ErrCASConflict) { |
| + return err |
| + } |
| + etag, err = currentAlertEtag(ctx, s, key) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := s.Put(ctx, key, body, etag); err != nil { |
| + return err |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// currentAlertEtag reads key's current etag for a replace-only Put, folding |
| +// "doesn't exist yet" into "" (create-only) — the same convention |
| +// metrics.WriteRaw's own currentRawEtag uses. |
| +func currentAlertEtag(ctx context.Context, s release.Store, key string) (string, error) { |
| + _, etag, err := s.Get(ctx, key) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + return "", nil |
| + } |
| + return "", err |
| + } |
| + return etag, nil |
| +} |
| + |
| +// httpPostWebhook is the production webhook post func: a 5s-timeout POST |
| +// with a JSON content type. Any non-2xx status is an error — the caller |
| +// (alertTick) only logs it, but it still needs to know delivery failed. |
| +func httpPostWebhook(url string, body []byte) error { |
| + client := &http.Client{Timeout: 5 * time.Second} |
| + resp, err := client.Post(url, "application/json", bytes.NewReader(body)) |
| + if err != nil { |
| + return err |
| + } |
| + defer resp.Body.Close() |
| + if resp.StatusCode < 200 || resp.StatusCode >= 300 { |
| + return fmt.Errorf("host: webhook: non-2xx status %s", resp.Status) |
| + } |
| + return nil |
| +} |
| + |
| +// 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 { |
| + return sampleProbes{ |
| + procStat: readProcStat, |
| + loadavg: readLoadAvg, |
| + memFrac: readMemFrac, |
| + diskFrac: readDiskFrac, |
| + units: realUnitReadings, |
| + now: time.Now, |
| + } |
| +} |
| + |
| +// realUnitReadings lists the box's currently active CARLOS units and reads |
| +// each one's cgroup usage — the production units func sampleProbes.units |
| +// wraps. |
| +func realUnitReadings() map[string]unitRaw { |
| + names := listCarlosUnits() |
| + if names == nil { |
| + return nil |
| + } |
| + out := make(map[string]unitRaw, len(names)) |
| + for _, name := range names { |
| + out[name] = readUnitRaw(name) |
| + } |
| + return out |
| +} |
| + |
| +// loadAlertState reads a persisted AlertState from path, falling back to a |
| +// fresh state for box on anything that isn't a clean read — a missing file |
| +// (first boot) or a corrupt one (a previous crash mid-write, or a hand |
| +// edit) must never crash the agent loop; monitoring starting from a blank |
| +// slate is always a safe, honest state to resume from. |
| +func loadAlertState(path, box string) metrics.AlertState { |
| + fresh := metrics.AlertState{V: 1, Box: box, Rules: map[string]metrics.RuleState{}} |
| + body, err := os.ReadFile(path) |
| + if err != nil { |
| + return fresh |
| + } |
| + var st metrics.AlertState |
| + if err := json.Unmarshal(body, &st); err != nil { |
| + return fresh |
| + } |
| + if st.Rules == nil { |
| + st.Rules = map[string]metrics.RuleState{} |
| + } |
| + return st |
| +} |
| + |
| +// metricsLoop runs the metrics sampler's 60s tick forever, once armed — thin |
| +// and untested by the same convention as statusLoop/adoptLoop: everything |
| +// with behavior lives in sampler.tick, metricsFlush, and alertTick, all |
| +// exercised above with fixtures. Missing status/deployment bucket config |
| +// disables the loop with one log line, the same "a box with no store is |
| +// 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() { |
| + ctx := context.Background() |
| + bucket := resolveStatusBucket(os.Getenv) |
| + if bucket == "" { |
| + log.Printf("agent: metrics loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set") |
| + return |
| + } |
| + s, err := release.NewS3Store(ctx, bucket) |
| + if err != nil { |
| + log.Printf("agent: metrics loop disabled: %v", err) |
| + return |
| + } |
| + |
| + box := status.Label(readMachineID()) |
| + statePath := filepath.Join(dataDir(), "monitor", "alerts.json") |
| + st := loadAlertState(statePath, box) |
| + rules := metrics.RulesFromEnv(os.Getenv) |
| + webhookURL := os.Getenv("CARLOS_ALERT_WEBHOOK") |
| + |
| + log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to bucket %s", bucket) |
| + |
| + sm := &sampler{} |
| + probes := realSampleProbes() |
| + tick := 0 |
| + for { |
| + time.Sleep(60 * time.Second) |
| + tick++ |
| + |
| + sm.tick(probes) |
| + latest := sm.buf[len(sm.buf)-1] |
| + if err := alertTick(ctx, s, box, rules, &st, latest, statePath, webhookURL, httpPostWebhook); err != nil { |
| + log.Printf("agent: alert tick: %v", err) |
| + } |
| + |
| + if tick%5 == 0 { |
| + if err := metricsFlush(ctx, s, box, sm, time.Now()); err != nil { |
| + log.Printf("agent: metrics flush: %v", err) |
| + } |
| + } |
| + } |
| +} |
amadan