amadan

carlos / platform Public

metrics: alert rules — sustain windows, hysteresis, nil never fires or clears

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell ff5360ad86ec06b3f63d189f0fe1a9a8d7b8c6c3 parent 231963a
2 files changed, +403 −0
  • internal/metrics/alert.go +243 −0
  • internal/metrics/alert_test.go +160 −0
diff --git a/internal/metrics/alert.go b/internal/metrics/alert.go
new file mode 100644
index 000000000..8f1f4924a
--- /dev/null
+++ b/internal/metrics/alert.go
@@ -0,0 +1,243 @@
+package metrics
+
+import (
+ "strconv"
+ "time"
+)
+
+// Rules is the thresholds and sustain windows Evaluate checks each tick
+// against. Frac fields are fractions (0..1), matching HostSample; LagSecs is
+// the one exception — keymail relay lag arrives in seconds, not a fraction.
+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
+}
+
+// DefaultRules are the thresholds shipped with the platform.
+func DefaultRules() Rules {
+ return Rules{
+ DiskFrac: 0.90,
+ MemFrac: 0.90,
+ MemSustain: 10 * time.Minute,
+ CPUFrac: 0.95,
+ CPUSustain: 15 * time.Minute,
+ LagSecs: 900,
+ ClearBy: 0.05,
+ }
+}
+
+// RulesFromEnv applies percent-integer overrides on top of DefaultRules:
+// CARLOS_ALERT_DISK_PCT, CARLOS_ALERT_MEM_PCT, CARLOS_ALERT_CPU_PCT. An
+// unset, unparsable, or out-of-range (0..100) value keeps the default —
+// deliberately pure (no logging); a caller that wants to log a rejected
+// override does so itself, since this package touches neither stderr nor
+// the OS.
+func RulesFromEnv(getenv func(string) string) Rules {
+ r := DefaultRules()
+ if v, ok := pctFrac(getenv("CARLOS_ALERT_DISK_PCT")); ok {
+ r.DiskFrac = v
+ }
+ if v, ok := pctFrac(getenv("CARLOS_ALERT_MEM_PCT")); ok {
+ r.MemFrac = v
+ }
+ if v, ok := pctFrac(getenv("CARLOS_ALERT_CPU_PCT")); ok {
+ r.CPUFrac = v
+ }
+ return r
+}
+
+func pctFrac(s string) (float64, bool) {
+ if s == "" {
+ return 0, false
+ }
+ n, err := strconv.Atoi(s)
+ if err != nil || n < 0 || n > 100 {
+ return 0, false
+ }
+ return float64(n) / 100.0, true
+}
+
+// Observation is one tick's readings for a box. A nil field means "no
+// data this tick" and is handled identically everywhere in Evaluate: the
+// corresponding rule's state is left untouched (a firing alert stays
+// firing, a quiet rule stays quiet, nothing is ever emitted for it).
+type Observation struct {
+ At time.Time
+ CPUFrac *float64
+ MemFrac *float64
+ DiskFrac *float64
+ LagSecs *int64
+}
+
+// RuleState is one rule's sustain/hysteresis bookkeeping. Since and FiredAt
+// use omitempty so a never-triggered rule round-trips through JSON as a
+// bare {"firing":false} rather than carrying zero-value time.Time noise.
+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"` // set on fire, cleared on clear
+ Value float64 `json:"value,omitempty"` // reading at last transition
+}
+
+// AlertState is the JSON persisted to disk by the agent loop and published
+// to alerts/boxes/<box>.json (see AlertKey), and read back by the console
+// watcher that diffs Rules[...].Firing across polls.
+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"
+}
+
+// Event is emitted on a firing or clearing transition — never on every
+// tick, and never for a nil reading.
+type Event struct {
+ Rule string `json:"rule"`
+ Firing bool `json:"firing"` // true=fired, false=cleared
+ Value float64 `json:"value"`
+ Box string `json:"box"`
+}
+
+// AlertKey is the bucket key an AlertState for box is stored/published
+// under.
+func AlertKey(box string) string {
+ return "alerts/boxes/" + box + ".json"
+}
+
+// Evaluate advances AlertState by one Observation and returns the new
+// state plus any fire/clear transitions as Events.
+//
+// Evaluate never mutates st: it builds a fresh Rules map (copying each
+// RuleState by value — RuleState holds no pointers or slices, so the copy
+// is a true deep copy) and returns that in the new AlertState. Callers
+// comparing old vs new state across the call — the console watcher this
+// feeds diffs Rules[...].Firing on exactly that assumption — see two
+// independent maps, never the same one mutated in place.
+//
+// Semantics, per rule ("disk", "mem", "cpu", "lag"):
+// - nil reading: that rule's state is left exactly as-is. No data never
+// fires and never clears — clearing needs evidence.
+// - Not firing, reading > threshold: Since is set if it was zero; the
+// rule fires (emitting an Event) once obs.At.Sub(Since) >= sustain
+// (immediate rules use sustain 0, so they fire the same tick Since is
+// set). A reading back at-or-below threshold before sustain elapses
+// resets Since to zero — the sustain window must be continuous.
+// - Firing, reading < threshold-ClearBy: the rule clears (emitting an
+// Event) and Since/FiredAt are zeroed.
+// - Firing, reading in [threshold-ClearBy, threshold] (or still above
+// threshold): stays firing — this is the hysteresis band, and it is
+// deliberately wider than just "below threshold" so a reading
+// oscillating right at the line doesn't flap.
+func Evaluate(r Rules, st AlertState, obs Observation) (AlertState, []Event) {
+ out := AlertState{
+ V: st.V,
+ Box: st.Box,
+ UpdatedAt: obs.At,
+ Rules: make(map[string]RuleState, len(st.Rules)),
+ }
+ for name, rs := range st.Rules {
+ out.Rules[name] = rs
+ }
+
+ var events []Event
+ emit := func(name string, firing bool, value float64) {
+ events = append(events, Event{Rule: name, Firing: firing, Value: value, Box: st.Box})
+ }
+
+ evalFrac := func(name string, reading *float64, threshold float64, sustain time.Duration) {
+ if reading == nil {
+ return
+ }
+ val := *reading
+ cur := out.Rules[name]
+ clearLine := threshold - r.ClearBy
+
+ if cur.Firing {
+ if val < clearLine {
+ cur.Firing = false
+ cur.Since = time.Time{}
+ cur.FiredAt = time.Time{}
+ cur.Value = val
+ out.Rules[name] = cur
+ emit(name, false, val)
+ }
+ return
+ }
+
+ if val > threshold {
+ if cur.Since.IsZero() {
+ cur.Since = obs.At
+ }
+ if obs.At.Sub(cur.Since) >= sustain {
+ cur.Firing = true
+ cur.FiredAt = obs.At
+ cur.Value = val
+ out.Rules[name] = cur
+ emit(name, true, val)
+ return
+ }
+ out.Rules[name] = cur
+ return
+ }
+
+ if !cur.Since.IsZero() {
+ cur.Since = time.Time{}
+ out.Rules[name] = cur
+ }
+ }
+
+ evalLagSecs := func(name string, reading *int64, threshold int64, clearBy int64, sustain time.Duration) {
+ if reading == nil {
+ return
+ }
+ val := float64(*reading)
+ cur := out.Rules[name]
+ clearLine := float64(threshold - clearBy)
+ thresholdF := float64(threshold)
+
+ if cur.Firing {
+ if val < clearLine {
+ cur.Firing = false
+ cur.Since = time.Time{}
+ cur.FiredAt = time.Time{}
+ cur.Value = val
+ out.Rules[name] = cur
+ emit(name, false, val)
+ }
+ return
+ }
+
+ if val > thresholdF {
+ if cur.Since.IsZero() {
+ cur.Since = obs.At
+ }
+ if obs.At.Sub(cur.Since) >= sustain {
+ cur.Firing = true
+ cur.FiredAt = obs.At
+ cur.Value = val
+ out.Rules[name] = cur
+ emit(name, true, val)
+ return
+ }
+ out.Rules[name] = cur
+ return
+ }
+
+ if !cur.Since.IsZero() {
+ cur.Since = time.Time{}
+ out.Rules[name] = cur
+ }
+ }
+
+ evalFrac("disk", obs.DiskFrac, r.DiskFrac, 0)
+ evalFrac("mem", obs.MemFrac, r.MemFrac, r.MemSustain)
+ evalFrac("cpu", obs.CPUFrac, r.CPUFrac, r.CPUSustain)
+ evalLagSecs("lag", obs.LagSecs, r.LagSecs, 300, 0)
+
+ return out, events
+}
diff --git a/internal/metrics/alert_test.go b/internal/metrics/alert_test.go
new file mode 100644
index 000000000..19c025959
--- /dev/null
+++ b/internal/metrics/alert_test.go
@@ -0,0 +1,160 @@
+package metrics
+
+import (
+ "testing"
+ "time"
+)
+
+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)
+ }
+}
+
+func TestEvaluateDiskFiresImmediately(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)
+ d := 0.95
+
+ _, ev := Evaluate(r, st, Observation{At: t0, DiskFrac: &d})
+ if len(ev) != 1 || !ev[0].Firing || ev[0].Rule != "disk" {
+ t.Fatalf("disk rule is immediate (sustain 0): want fire on first tick, got %v", ev)
+ }
+}
+
+func TestEvaluateSustainResetsOnDip(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, _ = Evaluate(r, st, obs(0.95, t0)) // sustain starts
+ st, _ = Evaluate(r, st, obs(0.80, t0.Add(5*time.Minute))) // dip below threshold — resets Since
+ if !st.Rules["mem"].Since.IsZero() {
+ t.Fatalf("dip below threshold must reset Since, got %v", st.Rules["mem"].Since)
+ }
+ // Climb back above threshold; Since restarts at t0+10m. If sustain had
+ // wrongly resumed counting from the original t0, this would already have
+ // fired (t0+10m is 10 minutes after t0) — it must not.
+ st, ev := Evaluate(r, st, obs(0.95, t0.Add(10*time.Minute)))
+ if len(ev) != 0 || st.Rules["mem"].Firing {
+ t.Fatalf("sustain must restart after a dip, not resume from the original Since: ev=%v", ev)
+ }
+ // A full 10 minutes after the restart (t0+20m) — now it fires.
+ _, ev = Evaluate(r, st, obs(0.95, t0.Add(20*time.Minute)))
+ if len(ev) != 1 || !ev[0].Firing {
+ t.Fatalf("want fire once the restarted sustain window elapses, got %v", ev)
+ }
+}
+
+func TestEvaluateNilReadingHoldsState(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)
+ d := 0.95
+
+ // Fire disk (immediate).
+ st, ev := Evaluate(r, st, Observation{At: t0, DiskFrac: &d})
+ if len(ev) != 1 || !st.Rules["disk"].Firing {
+ t.Fatalf("setup: disk should be firing, got %v %v", ev, st.Rules["disk"])
+ }
+ quietBefore := st.Rules["mem"]
+
+ // A tick with nil disk and nil mem must leave both rules exactly as they are:
+ // a firing alert stays firing, a quiet rule stays quiet — no event either way.
+ st, ev = Evaluate(r, st, Observation{At: t0.Add(time.Minute)})
+ if len(ev) != 0 {
+ t.Fatalf("nil readings must never emit events, got %v", ev)
+ }
+ if !st.Rules["disk"].Firing {
+ t.Fatal("nil reading must not clear a firing rule")
+ }
+ if st.Rules["mem"] != quietBefore {
+ t.Fatalf("nil reading must not change a quiet rule's state: before=%v after=%v", quietBefore, st.Rules["mem"])
+ }
+}
+
+func TestEvaluateLagNilForeverNoEvent(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)
+
+ for i := 0; i < 5; i++ {
+ var ev []Event
+ st, ev = Evaluate(r, st, Observation{At: t0.Add(time.Duration(i) * time.Minute)})
+ if len(ev) != 0 {
+ t.Fatalf("lag rule must never fire while readings stay nil, got %v", ev)
+ }
+ if st.Rules["lag"].Firing {
+ t.Fatal("lag rule must never fire while readings stay nil")
+ }
+ }
+}
+
+func TestEvaluateLagFiresAndClears(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)
+ lag := func(s int64) *int64 { return &s }
+
+ st, ev := Evaluate(r, st, Observation{At: t0, LagSecs: lag(1000)}) // > 900, immediate
+ if len(ev) != 1 || !ev[0].Firing || ev[0].Rule != "lag" {
+ t.Fatalf("want lag fire on first tick over threshold, got %v", ev)
+ }
+ // Clear line is LagSecs-300 = 600. 700 is inside the hysteresis band (600..900) — stays firing.
+ st, ev = Evaluate(r, st, Observation{At: t0.Add(time.Minute), LagSecs: lag(700)})
+ if len(ev) != 0 || !st.Rules["lag"].Firing {
+ t.Fatalf("lag must hold inside its hysteresis band, got ev=%v", ev)
+ }
+ // 500 < 600 — clears.
+ _, ev = Evaluate(r, st, Observation{At: t0.Add(2 * time.Minute), LagSecs: lag(500)})
+ if len(ev) != 1 || ev[0].Firing {
+ t.Fatalf("want lag clear below LagSecs-300, got %v", ev)
+ }
+}
+
+func TestRulesFromEnv(t *testing.T) {
+ env := func(m map[string]string) func(string) string {
+ return func(k string) string { return m[k] }
+ }
+
+ r := RulesFromEnv(env(map[string]string{"CARLOS_ALERT_DISK_PCT": "95"}))
+ if r.DiskFrac != 0.95 {
+ t.Fatalf("want DiskFrac 0.95, got %v", r.DiskFrac)
+ }
+
+ def := DefaultRules()
+ r = RulesFromEnv(env(map[string]string{"CARLOS_ALERT_DISK_PCT": "nonsense"}))
+ if r.DiskFrac != def.DiskFrac {
+ t.Fatalf("unparsable value must keep default %v, got %v", def.DiskFrac, r.DiskFrac)
+ }
+
+ r = RulesFromEnv(env(map[string]string{"CARLOS_ALERT_MEM_PCT": "150"}))
+ if r.MemFrac != def.MemFrac {
+ t.Fatalf("out-of-range value must keep default %v, got %v", def.MemFrac, r.MemFrac)
+ }
+
+ r = RulesFromEnv(env(map[string]string{"CARLOS_ALERT_CPU_PCT": "80"}))
+ if r.CPUFrac != 0.80 {
+ t.Fatalf("want CPUFrac 0.80, got %v", r.CPUFrac)
+ }
+}