diff --git a/cmd/mavend/tick_test.go b/cmd/mavend/tick_test.go index 7290ece..c2c4ac9 100644 --- a/cmd/mavend/tick_test.go +++ b/cmd/mavend/tick_test.go @@ -584,7 +584,7 @@ func TestDigestSev4BypassesQueue(t *testing.T) { ctx := context.Background() now := refNow() markPresent(t, st, ctx, now) - if _, err := st.SetValue(ctx, store.KindSelf, "service_down", "poll:uptimekuma", "down", now); err != nil { + if _, err := st.SetValue(ctx, store.KindSelf, "service_down:db", "poll:uptimekuma", "down", now); err != nil { t.Fatalf("seed service_down: %v", err) } sink := &fakeSink{} diff --git a/cmd/mavpoll/main.go b/cmd/mavpoll/main.go index 69c9eb4..f3f0702 100644 --- a/cmd/mavpoll/main.go +++ b/cmd/mavpoll/main.go @@ -140,6 +140,10 @@ type poller struct { wgIface string wgCmd string + // kumaSeen — monitor name → state as of the last poll, so a monitor that + // disappears from the gauge can be marked unknown instead of staying down. + kumaSeen map[string]string + // zen is nil unless a token file was configured — money tracking is a // capability, off by default like weather and telegram. zen *zenmoney.Client @@ -333,50 +337,96 @@ func maxSeverity(a netdataAlarms) string { return sev } -// ---- kuma: monitor_status gauge → aggregate service_down ------------------- +// ---- kuma: monitor_status gauge → one fact per monitor --------------------- // Kuma exposes Prometheus text: `monitor_status{...,monitor_name="X"} V` where -// V is 1=up 0=down 2=pending 3=maintenance. We reduce to one aggregate the -// existing ServiceDownRule consumes: "down" if ANY monitor reads 0, else "up". -// Per-service granularity is a later add (a fact per monitor) — the MVP nudge -// only needs "something is down". -var kumaLine = regexp.MustCompile(`^monitor_status\{([^}]*)\}\s+([0-9.eE+-]+)`) +// V is 1=up 0=down 2=pending 3=maintenance. We write one fact per monitor, +// keyed `service_down:`, because the nudge has to say WHICH +// service is down. The aggregate this used to write could not, which is why +// the rule shipped disabled. +var ( + kumaLine = regexp.MustCompile(`^monitor_status\{([^}]*)\}\s+([0-9.eE+-]+)`) + kumaName = regexp.MustCompile(`monitor_name="([^"]*)"`) +) func (p *poller) pollKuma(ctx context.Context, now time.Time) error { body, err := p.get(ctx, p.kumaURL, p.kumaKey) if err != nil { return err } - down, seen := kumaAnyDown(body) - if !seen { + states := kumaMonitors(body) + if len(states) == 0 { return fmt.Errorf("no monitor_status metrics (auth/endpoint wrong?)") } - val := "up" - if down { - val = "down" + var firstErr error + for name, val := range states { + if err := p.writeIfChanged(ctx, kumaFactKey(name), kumaSource, val, now); err != nil && firstErr == nil { + firstErr = err // one bad monitor must not blind the rest + } } - return p.writeIfChanged(ctx, "service_down", "poll:uptimekuma", val, now) + // A monitor deleted in kuma stops appearing in the gauge, and its last fact + // would otherwise read "down" forever. Mark it unknown, which no rule fires + // on. The seen-set is in memory, so a restart forgets it — harmless, since + // the next poll that still lacks the monitor says nothing new either. + for name := range p.kumaSeen { + if _, still := states[name]; !still { + if err := p.writeIfChanged(ctx, kumaFactKey(name), kumaSource, "unknown", now); err != nil && firstErr == nil { + firstErr = err + } + } + } + p.kumaSeen = states + return firstErr } -// kumaAnyDown parses kuma's Prometheus text: down=true if any monitor reads 0 -// (pending=2/maintenance=3 are not "down"). seen=false ⇒ no monitor_status -// lines matched at all (wrong endpoint or auth rejected before the body). -func kumaAnyDown(body []byte) (down, seen bool) { +// kumaSource — the provenance the loop rule requires. Written here, checked in +// loop.ServiceDownRule; a poller under any other source cannot fire it. +const kumaSource = "poll:uptimekuma" + +// kumaFactKey — the fact key for one monitor. The suffix is the name he hears, +// so it stays as kuma spells it rather than being slugged into something else. +func kumaFactKey(name string) string { return "service_down:" + name } + +// kumaMonitors parses kuma's Prometheus text into monitor name → state +// ("up"/"down"/"pending"/"maintenance"). An empty map means no monitor_status +// line matched at all (wrong endpoint, or auth rejected before the body). +// A line with no monitor_name label is skipped: a fact nobody can name is +// exactly the thing this replaced. +func kumaMonitors(body []byte) map[string]string { + out := make(map[string]string) for _, line := range strings.Split(string(body), "\n") { m := kumaLine.FindStringSubmatch(strings.TrimSpace(line)) if m == nil { continue } - seen = true + nm := kumaName.FindStringSubmatch(m[1]) + if nm == nil || strings.TrimSpace(nm[1]) == "" { + continue + } v, err := strconv.ParseFloat(m[2], 64) if err != nil { continue } - if v == 0 { - down = true - } + out[strings.TrimSpace(nm[1])] = kumaState(v) + } + return out +} + +// kumaState — the gauge's four values. pending and maintenance are not "down": +// a monitor paused in kuma should silence that monitor, not page him. +func kumaState(v float64) string { + switch v { + case 0: + return "down" + case 1: + return "up" + case 2: + return "pending" + case 3: + return "maintenance" + default: + return "unknown" } - return down, seen } // ---- helpers --------------------------------------------------------------- diff --git a/cmd/mavpoll/main_test.go b/cmd/mavpoll/main_test.go index 7004518..229dd19 100644 --- a/cmd/mavpoll/main_test.go +++ b/cmd/mavpoll/main_test.go @@ -35,22 +35,46 @@ func TestMaxSeverity(t *testing.T) { } } -func TestKumaAnyDown(t *testing.T) { +func TestKumaMonitorsNamesEveryOne(t *testing.T) { cases := []struct { - body string - down, seen bool + name string + body string + want map[string]string }{ - {"", false, false}, - {`monitor_status{monitor_name="web"} 1`, false, true}, - {`monitor_status{monitor_name="web"} 1` + "\n" + `monitor_status{monitor_name="db"} 0`, true, true}, - {`monitor_status{monitor_name="mnt"} 3`, false, true}, // maintenance ≠ down - {`# HELP monitor_status ...`, false, false}, + {"empty body", "", map[string]string{}}, + {"help line only", `# HELP monitor_status ...`, map[string]string{}}, + { + "one up one down", + `monitor_status{monitor_name="web"} 1` + "\n" + `monitor_status{monitor_name="db"} 0`, + map[string]string{"web": "up", "db": "down"}, + }, + {"maintenance is not down", `monitor_status{monitor_name="mnt"} 3`, map[string]string{"mnt": "maintenance"}}, + {"pending is not down", `monitor_status{monitor_name="p"} 2`, map[string]string{"p": "pending"}}, + { + "other labels do not hide the name", + `monitor_status{monitor_type="http",monitor_name="ci",monitor_url="x"} 0`, + map[string]string{"ci": "down"}, + }, + {"a nameless line is skipped", `monitor_status{monitor_type="http"} 0`, map[string]string{}}, } for _, c := range cases { - down, seen := kumaAnyDown([]byte(c.body)) - if down != c.down || seen != c.seen { - t.Errorf("kumaAnyDown(%q) = (%v,%v), want (%v,%v)", c.body, down, seen, c.down, c.seen) - } + t.Run(c.name, func(t *testing.T) { + got := kumaMonitors([]byte(c.body)) + if len(got) != len(c.want) { + t.Fatalf("kumaMonitors = %v, want %v", got, c.want) + } + for k, v := range c.want { + if got[k] != v { + t.Errorf("monitor %q = %q, want %q", k, got[k], v) + } + } + }) + } +} + +func TestKumaFactKeyCarriesTheName(t *testing.T) { + if got := kumaFactKey("nexus db"); got != "service_down:nexus db" { + t.Errorf("kumaFactKey = %q", got) } } diff --git a/deploy/mavend.json b/deploy/mavend.json index bb67b34..e1d1a9e 100644 --- a/deploy/mavend.json +++ b/deploy/mavend.json @@ -8,12 +8,12 @@ "//disabled_rules": [ "Nudge rules that are not wired at all. Names come from loop.DefaultRules:", "water, meal, break, service_down, netdata_critical.", - "service_down is off because it cannot say WHICH service — mavpoll folds the", - "whole kuma gauge into one boolean, so the nudge is always the generic 'a", - "service on homesrv is down'. Nothing to act on, every fifteen minutes.", - "Turn it back on once Vikunja #444 lands a fact per monitor." + "service_down is back on: mavpoll now writes one fact per kuma monitor", + "(service_down:), so the nudge names the service and pausing a monitor", + "in kuma silences that monitor. It is also edge-triggered, so a service that", + "stays down is one nudge, not one every fifteen minutes." ], - "disabled_rules": ["service_down"], + "disabled_rules": [], "phraser": { "model_path": "/opt/maven/models/llm/qwen3/Qwen3-1.7B-UD-Q4_K_XL.gguf", diff --git a/docs/design.md b/docs/design.md index 858feea..484ad33 100644 --- a/docs/design.md +++ b/docs/design.md @@ -424,6 +424,29 @@ lives in `source`; rules trust provenance. `source=poll:healthcheck`. A compromised poller must not be able to forge a trigger. +#### A fact per monitor, not an aggregate + +mavpoll writes one fact per kuma monitor, keyed `service_down:`. +It used to fold the whole gauge into a single boolean, and the nudge could then +only say that something on homesrv was down. That is not something he can act +on, so the rule shipped disabled. + +Three things follow from the split: + +- The key set is no longer known at wiring time. A rule declares + `WantPrefixes` and the gatherer resolves the family per tick, which is the + only prefix read in the loop. +- Pausing a monitor in kuma silences that monitor. Under the aggregate it + silenced nothing, because some other monitor kept the boolean at "down". +- A monitor deleted in kuma would keep its last fact reading "down" forever, so + mavpoll marks a vanished monitor "unknown". No rule fires on "unknown". + +The rule is also edge-triggered: it fires on a transition it has not already +nudged about (`State.NudgedSince`). A polled fact is written only when the +value changes, but the predicate reads the current value, so without the edge +check a service that stays down qualifies on every tick and cooldown is the +only brake. + ### Presence — concrete scoring **Combiner — noisy-OR, not weighted sum.** These are independent-ish positive diff --git a/internal/config/config.go b/internal/config/config.go index 990ae1e..51297bb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -149,10 +149,11 @@ type Config struct { // // Rules are code, not config (see loop.DefaultRules), and that stays true: // this only subtracts. It exists because a rule can be right in principle - // and useless in practice — kuma's service_down cannot name the service it - // is nudging about (Vikunja #444), so being told "a service on homesrv is - // down" every fifteen minutes is noise with no action attached. Turning it - // off beats learning to ignore her. + // and useless in practice. service_down was the case that forced it: it + // could not name the service it was nudging about, so being told "a service + // on homesrv is down" every fifteen minutes was noise with no action + // attached. That is fixed — one fact per kuma monitor — and the rule ships + // enabled again. The escape hatch stays. // // A disabled rule is never gathered for, never evaluated, and never // delivered on any channel. Unknown names are ignored, so removing a rule diff --git a/internal/loop/explain_test.go b/internal/loop/explain_test.go index 0acbf30..1110106 100644 --- a/internal/loop/explain_test.go +++ b/internal/loop/explain_test.go @@ -103,7 +103,7 @@ func TestExplainGate_PresenceAway(t *testing.T) { func TestExplainGate_PresenceAwayOpsBypass(t *testing.T) { now := refTime() s := State{Now: now, Presence: store.Away, - Facts: map[string]store.Fact{"service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute))}, + Facts: map[string]store.Fact{"service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute))}, } r := ServiceDownRule() // Sev4 ops passed, blocked, d := ExplainGate(s, r) @@ -259,8 +259,8 @@ func TestExplainTick_WinnerRecorded(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), - "service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), + "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), + "service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), }, } cand, trace := ExplainTick(s, DefaultRules()) diff --git a/internal/loop/gate_test.go b/internal/loop/gate_test.go index eb613d4..4ad6e90 100644 --- a/internal/loop/gate_test.go +++ b/internal/loop/gate_test.go @@ -198,12 +198,12 @@ func TestTickNeverDogpilesAndPicksLoudest(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "water": ago("water", "tap:water", `"250ml"`, 5*time.Hour), - "meal": ago("meal", "voice", `"lunch"`, 8*time.Hour), - "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), - "break": ago("break", "voice", `"walk"`, 3*time.Hour), - "service_down": ago("service_down", "poll:uptimekuma", `"down"`, time.Minute), - "netdata_alarm": ago("netdata_alarm", "poll:netdata", `"critical"`, time.Minute), + "water": ago("water", "tap:water", `"250ml"`, 5*time.Hour), + "meal": ago("meal", "voice", `"lunch"`, 8*time.Hour), + "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), + "break": ago("break", "voice", `"walk"`, 3*time.Hour), + "service_down:db": ago("service_down:db", "poll:uptimekuma", `"down"`, time.Minute), + "netdata_alarm": ago("netdata_alarm", "poll:netdata", `"critical"`, time.Minute), }, } // sanity: every rule really does want to fire, so the pick is a real choice. diff --git a/internal/loop/gather.go b/internal/loop/gather.go index c55919f..73e164f 100644 --- a/internal/loop/gather.go +++ b/internal/loop/gather.go @@ -91,6 +91,20 @@ func (g *Gatherer) GatherState(ctx context.Context, now time.Time) (State, []sto return State{}, nil, err } + // prefix families — the keys a rule cannot name at wiring time (one fact + // per kuma monitor). Loaded into the same map; State.FactsUnder reads them. + for _, r := range g.rules { + for _, p := range r.WantPrefixes { + fam, err := g.store.LatestFactsByPrefix(ctx, p) + if err != nil { + return State{}, nil, err + } + for _, f := range fam { + facts[f.Key] = f + } + } + } + // last nudge per rule + cooldown-until derived from the active cooldown. // "active" = the feedback tuner's persisted base if one exists, else the // rule's static Base. LatestFactBySource is the trust-by-provenance read diff --git a/internal/loop/gather_test.go b/internal/loop/gather_test.go new file mode 100644 index 0000000..c854eba --- /dev/null +++ b/internal/loop/gather_test.go @@ -0,0 +1,47 @@ +package loop + +import ( + "context" + "path/filepath" + "testing" + "time" + + "github.com/kami/maven/internal/store" +) + +// The gatherer is the only impure piece, and a rule over a prefix has no keys +// to declare at wiring time. This is the end of that path: mavpoll's per-monitor +// facts reach the snapshot, and the rule fires on the one that is down. +func TestGatherStateLoadsPrefixFamilies(t *testing.T) { + ctx := context.Background() + s, err := store.Open(ctx, filepath.Join(t.TempDir(), "loop_test.db")) + if err != nil { + t.Fatalf("Open: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + + now := time.Now().UTC().Truncate(time.Millisecond) + for key, val := range map[string]string{ + "service_down:db": "down", + "service_down:web": "up", + } { + if _, err := s.SetValue(ctx, store.KindEnv, key, ServiceDownSource, val, now.Add(-time.Minute)); err != nil { + t.Fatalf("SetValue %s: %v", key, err) + } + } + + rules := []Rule{ServiceDownRule()} + st, _, err := NewGatherer(s, rules).GatherState(ctx, now) + if err != nil { + t.Fatalf("GatherState: %v", err) + } + if _, ok := st.Facts["service_down:db"]; !ok { + t.Fatalf("prefix family not gathered: %v", st.Facts) + } + if got := DownServices(st); len(got) != 1 || got[0] != "db" { + t.Fatalf("DownServices = %v, want [db]", got) + } + if !rules[0].Predicate(st) { + t.Fatal("the rule must fire on a gathered per-monitor fact") + } +} diff --git a/internal/loop/loop_test.go b/internal/loop/loop_test.go index d6837c8..0f5cf1a 100644 --- a/internal/loop/loop_test.go +++ b/internal/loop/loop_test.go @@ -82,7 +82,7 @@ func TestTickOpsHardSurvivesAwayAndQuiet(t *testing.T) { Presence: store.Away, QuietHours: true, Facts: map[string]store.Fact{ - "service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), + "service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), }, } got := Tick(s, DefaultRules()) @@ -99,7 +99,7 @@ func TestTickServiceSourceTrustRefusesForgedTrigger(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "service_down": factAt("service_down", "ambient", `"down"`, now.Add(-1*time.Minute)), + "service_down:db": factAt("service_down:db", "ambient", `"down"`, now.Add(-1*time.Minute)), }, } if got := Tick(s, DefaultRules()); got != nil { @@ -115,8 +115,8 @@ func TestTickOneNudgePerTickMaxSeverityWins(t *testing.T) { Now: now, Presence: store.Present, Facts: map[string]store.Fact{ - "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), - "service_down": factAt("service_down", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), + "water": factAt("water", "tap:water", `"250ml"`, now.Add(-4*time.Hour)), + "service_down:db": factAt("service_down:db", "poll:uptimekuma", `"down"`, now.Add(-1*time.Minute)), }, } got := Tick(s, DefaultRules()) diff --git a/internal/loop/rules.go b/internal/loop/rules.go index f144e2d..1fc5ff6 100644 --- a/internal/loop/rules.go +++ b/internal/loop/rules.go @@ -28,6 +28,13 @@ type Rule struct { // that check itself, leave this empty. Otherwise set to the key(s) the rule // needs and the gate will skip the rule when any are missing. InertWhenNoData []string + + // WantPrefixes — key prefixes whose whole family the gatherer must load. + // InertWhenNoData names keys that exist at wiring time; a rule over a key + // set that is only known at read time (one fact per kuma monitor) declares + // the prefix here instead. Prefixes never make a rule inert: an empty + // family is the predicate's own "no data" case. + WantPrefixes []string } // Cooldown — tunable bounded by the envelope so a weird week (auto-tuned) can't @@ -98,23 +105,58 @@ func BreakRule() Rule { } } -// ServiceDownRule — sev4 ops hard: the `service_down` aggregate fact reads -// "down". Source must be poll:uptimekuma — kuma is the source of truth for -// service up/down (mavpoll writes this key). The predicate is provenance-scoped: -// a compromised poller writing under a different source can't forge the trigger. +// ServiceDownPrefix — mavpoll writes one fact per kuma monitor under this +// prefix, `service_down:`. The suffix is the name he hears. +const ServiceDownPrefix = "service_down:" + +// ServiceDownSource — kuma is the source of truth for service up/down. The +// rule is provenance-scoped: a poller writing under a different source cannot +// forge the trigger. +const ServiceDownSource = "poll:uptimekuma" + +// DownServices — the monitors currently reading "down", by name, in key order. +// +// Pure, and the rule and the phraser both call it, so the message can never +// name a service the predicate did not fire on. +func DownServices(s State) []string { + var out []string + for _, f := range s.FactsUnder(ServiceDownPrefix) { + if f.Source == ServiceDownSource && f.Value == `"down"` { + out = append(out, strings.TrimPrefix(f.Key, ServiceDownPrefix)) + } + } + return out +} + +// ServiceDownRule — sev4 ops hard: at least one kuma monitor reads "down". +// +// It used to read one aggregate `service_down` fact, which is why it was +// disabled in deploy: the nudge could say that something on homesrv was down +// but never which thing. Per-monitor facts fix that, and pausing a monitor in +// kuma now silences that monitor rather than nothing. +// +// Edge-triggered — see State.NudgedSince. Without it a service that stays down +// for a day qualifies on every tick and cooldown alone is the only brake. func ServiceDownRule() Rule { return Rule{ - Name: "service_down", - Severity: Sev4, - Cooldown: Cooldown{Base: 15 * time.Minute, Min: 5 * time.Minute, Max: 1 * time.Hour}, - InertWhenNoData: []string{"service_down"}, + Name: "service_down", + Severity: Sev4, + Cooldown: Cooldown{Base: 15 * time.Minute, Min: 5 * time.Minute, Max: 1 * time.Hour}, + WantPrefixes: []string{ServiceDownPrefix}, Predicate: func(s State) bool { - f, ok := s.Fact("service_down") - if !ok || f.Ts.IsZero() { - return false + var newest time.Time + for _, f := range s.FactsUnder(ServiceDownPrefix) { + if f.Source != ServiceDownSource || f.Value != `"down"` { + continue + } + if f.Ts.After(newest) { + newest = f.Ts + } } - // value is json `"down"`; trivial check keyed off source provenance. - return f.Source == "poll:uptimekuma" && f.Value == `"down"` + if newest.IsZero() { + return false // nothing down, or no data at all → shut up + } + return !s.NudgedSince("service_down", newest) }, } } diff --git a/internal/loop/rules_test.go b/internal/loop/rules_test.go index 3bec077..70cc11c 100644 --- a/internal/loop/rules_test.go +++ b/internal/loop/rules_test.go @@ -193,13 +193,13 @@ func TestOpsRulePredicates(t *testing.T) { { name: "service_down fires on a kuma down fact", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma", `"down"`, time.Minute)}, want: true, }, { name: "service_down quiet when kuma says up", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma", `"up"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma", `"up"`, time.Minute)}, want: false, }, { @@ -211,38 +211,38 @@ func TestOpsRulePredicates(t *testing.T) { { name: "service_down quiet on a zero-timestamp fact", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": {Key: "service_down", Source: "poll:uptimekuma", Value: `"down"`}}, + facts: map[string]store.Fact{"service_down:db": {Key: "service_down:db", Source: "poll:uptimekuma", Value: `"down"`}}, want: false, }, // forgery attempts — right value, wrong writer. { name: "service_down refuses a forgery from the netdata poller", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:netdata", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:netdata", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses a forgery from ambient audio", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "ambient:other", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "ambient:other", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses a forgery from the user's own voice", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "voice", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "voice", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses a source that only looks like kuma", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma-staging", `"down"`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma-staging", `"down"`, time.Minute)}, want: false, }, { name: "service_down refuses an unquoted down value", rule: ServiceDownRule(), - facts: map[string]store.Fact{"service_down": ago("service_down", "poll:uptimekuma", `down`, time.Minute)}, + facts: map[string]store.Fact{"service_down:db": ago("service_down:db", "poll:uptimekuma", `down`, time.Minute)}, want: false, }, @@ -312,9 +312,11 @@ func TestOpsRulePredicates(t *testing.T) { // a second no-data backstop, so a rule that forgets it loses the safety net // even if its predicate happens to check. func TestDefaultRulesDeclareInertKeys(t *testing.T) { + // A rule over a key set that only exists at read time declares a prefix + // instead — the gatherer still needs to be told what to load. for _, r := range DefaultRules() { - if len(r.InertWhenNoData) == 0 { - t.Errorf("rule %q declares no InertWhenNoData keys", r.Name) + if len(r.InertWhenNoData) == 0 && len(r.WantPrefixes) == 0 { + t.Errorf("rule %q declares neither InertWhenNoData keys nor WantPrefixes", r.Name) } } } @@ -377,11 +379,11 @@ func TestDefaultRuleCooldownsAreBounded(t *testing.T) { // hidden state, no clock reads. func TestPredicatesArePure(t *testing.T) { s := stateWith(map[string]store.Fact{ - "water": ago("water", "tap:water", `"250ml"`, 4*time.Hour), - "meal": ago("meal", "voice", `"lunch"`, 7*time.Hour), - "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), - "break": ago("break", "voice", `"walk"`, 2*time.Hour), - "service_down": ago("service_down", "poll:uptimekuma", `"down"`, time.Minute), + "water": ago("water", "tap:water", `"250ml"`, 4*time.Hour), + "meal": ago("meal", "voice", `"lunch"`, 7*time.Hour), + "desk_active": ago("desk_active", "infer:hyprland", "1", 30*time.Second), + "break": ago("break", "voice", `"walk"`, 2*time.Hour), + "service_down:db": ago("service_down:db", "poll:uptimekuma", `"down"`, time.Minute), }) for _, r := range DefaultRules() { first := r.Predicate(s) @@ -434,3 +436,66 @@ func TestRulesExceptEmptyKeepsEverything(t *testing.T) { t.Errorf("rules = %v, dropped = %v", ruleNames(rules), dropped) } } + +// ---------------------------- per-monitor service_down ----------------------- + +// The rule must name what fired on it, and the phraser reads the same helper, +// so a service that is up can never be spoken as down. +func TestDownServicesNamesOnlyTheDownOnes(t *testing.T) { + s := State{ + Now: refTime(), + Facts: map[string]store.Fact{ + "service_down:web": ago("service_down:web", ServiceDownSource, `"up"`, time.Minute), + "service_down:db": ago("service_down:db", ServiceDownSource, `"down"`, time.Minute), + "service_down:vault": ago("service_down:vault", ServiceDownSource, `"down"`, time.Minute), + "service_down:paused": ago("service_down:paused", ServiceDownSource, `"maintenance"`, time.Minute), + "service_down:forged": ago("service_down:forged", "voice", `"down"`, time.Minute), + "service_down:missing": ago("service_down:missing", ServiceDownSource, `"unknown"`, time.Minute), + }, + } + got := DownServices(s) + want := []string{"db", "vault"} // key order, so speech is stable + if len(got) != len(want) { + t.Fatalf("DownServices = %v, want %v", got, want) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("DownServices = %v, want %v", got, want) + } + } +} + +// A monitor paused in kuma must silence that monitor. Before per-monitor facts +// the aggregate stayed "down" and pausing achieved nothing. +func TestPausedMonitorSilencesOnlyItself(t *testing.T) { + base := map[string]store.Fact{ + "service_down:db": ago("service_down:db", ServiceDownSource, `"maintenance"`, time.Minute), + "service_down:web": ago("service_down:web", ServiceDownSource, `"down"`, time.Minute), + } + if !ServiceDownRule().Predicate(State{Now: refTime(), Facts: base}) { + t.Fatal("web is still down, the rule must fire") + } + delete(base, "service_down:web") + if ServiceDownRule().Predicate(State{Now: refTime(), Facts: base}) { + t.Fatal("only a paused monitor is left, the rule must be quiet") + } +} + +// Edge-triggered: he is told once per transition. A service that stays down +// for a day used to qualify on every tick, with cooldown as the only brake. +func TestServiceDownFiresOncePerTransition(t *testing.T) { + down := ago("service_down:db", ServiceDownSource, `"down"`, time.Hour) + s := State{Now: refTime(), Facts: map[string]store.Fact{"service_down:db": down}} + if !ServiceDownRule().Predicate(s) { + t.Fatal("first sight of the transition must fire") + } + s.LastNudge = map[string]store.Nudge{"service_down": {Ts: down.Ts.Add(time.Minute)}} + if ServiceDownRule().Predicate(s) { + t.Fatal("already told about this transition, must be quiet") + } + // A second service goes down after that nudge — a new edge, so it fires. + s.Facts["service_down:web"] = ago("service_down:web", ServiceDownSource, `"down"`, time.Minute) + if !ServiceDownRule().Predicate(s) { + t.Fatal("a later transition must fire again") + } +} diff --git a/internal/loop/state.go b/internal/loop/state.go index f17b074..1c42f6e 100644 --- a/internal/loop/state.go +++ b/internal/loop/state.go @@ -23,6 +23,8 @@ package loop import ( + "sort" + "strings" "time" "github.com/kami/maven/internal/store" @@ -95,6 +97,32 @@ func (s State) Fact(key string) (store.Fact, bool) { return f, true } +// FactsUnder returns every gathered fact whose key starts with prefix, ordered +// by key so a caller that names them speaks them in a stable order. Facts with +// a zero Ts are skipped, the same "no data" rule Fact applies. +func (s State) FactsUnder(prefix string) []store.Fact { + var out []store.Fact + for k, f := range s.Facts { + if strings.HasPrefix(k, prefix) && !f.Ts.IsZero() { + out = append(out, f) + } + } + sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key }) + return out +} + +// NudgedSince reports whether rule already sent a nudge at or after ts. +// +// It is what makes a rule edge-triggered. A polled fact is written only when +// the value changes, so its Ts is the moment the service went down — but the +// predicate reads the current value, so a service that stays down keeps +// qualifying forever and cooldown alone only slows the repetition. Asking +// whether he was already told about THIS transition stops it. +func (s State) NudgedSince(rule string, ts time.Time) bool { + n, ok := s.LastNudge[rule] + return ok && !n.Ts.Before(ts) +} + // Since returns the duration since the latest fact for key, or (0,false). // "false" ⇒ no data ⇒ shuts up when uncertain. func (s State) Since(key string) (time.Duration, bool) { diff --git a/internal/phraser/llmphraser.go b/internal/phraser/llmphraser.go index 1fcfd4f..1296868 100644 --- a/internal/phraser/llmphraser.go +++ b/internal/phraser/llmphraser.go @@ -994,6 +994,9 @@ var fallbackNudges = map[string]string{ } func fallbackNudge(c loop.Candidate) string { + if down := loop.DownServices(c.State); len(down) > 0 { + return "Не отвечает: " + strings.Join(down, ", ") + "." + } if s, ok := fallbackNudges[c.Rule.Name]; ok { return s } @@ -1009,6 +1012,11 @@ func buildNudgePrompt(c loop.Candidate) string { if f, ok := c.State.Facts[c.Rule.Name]; ok && f.Key != "" && f.Key != c.Rule.Name { ctxParts = append(ctxParts, "Что именно: "+f.Key) } + if down := loop.DownServices(c.State); len(down) > 0 { + // The names come from the same helper the rule fired on, so the model + // is never handed a service that is actually up. + ctxParts = append(ctxParts, "Какие сервисы лежат: "+strings.Join(down, ", ")) + } if d, ok := c.State.Since(c.Rule.Name); ok { ctxParts = append(ctxParts, "Прошло: "+ruDur(d)) } diff --git a/internal/phraser/phraser.go b/internal/phraser/phraser.go index 79488d2..cb3fa9d 100644 --- a/internal/phraser/phraser.go +++ b/internal/phraser/phraser.go @@ -142,16 +142,21 @@ func phraseNudge(c loop.Candidate) (body, summary string) { } return body, "take a break" case "service_down": - // the fact value is json `"down"`; the key carries the service name. - body = "a service on homesrv is down — check journalctl." - summary = "service down on homesrv" - if f, ok := c.State.Fact("service_down"); ok { - if f.Key != "" && f.Key != "service_down" { - body = fmt.Sprintf("%s on homesrv is down — check journalctl.", f.Key) - summary = fmt.Sprintf("%s down on homesrv", f.Key) - } + // One fact per kuma monitor, so the nudge names the service. The rule + // and this share loop.DownServices, so the message cannot name a + // service the predicate did not fire on. + down := loop.DownServices(c.State) + switch len(down) { + case 0: + return "a service on homesrv is down — check journalctl.", "service down on homesrv" + case 1: + return fmt.Sprintf("%s on homesrv is down — check journalctl.", down[0]), + fmt.Sprintf("%s down on homesrv", down[0]) + default: + list := strings.Join(down, ", ") + return fmt.Sprintf("%s on homesrv are down — check journalctl.", list), + fmt.Sprintf("%d services down on homesrv", len(down)) } - return body, summary default: // generic: name the rule + severity; the LLM impl replaces this with // a prompted phrase. the Stub never editorializes beyond the rule name. diff --git a/internal/phraser/phraser_test.go b/internal/phraser/phraser_test.go index 23cddfc..69bc2ea 100644 --- a/internal/phraser/phraser_test.go +++ b/internal/phraser/phraser_test.go @@ -78,12 +78,14 @@ func TestPhraseNudgeBreakDeskDuration(t *testing.T) { } func TestPhraseNudgeServiceDownNamedService(t *testing.T) { - // a service_down fact whose Key is the specific service name → the phrase - // names the service, not just "service down". + // one fact per kuma monitor → the phrase names the monitor that is down. now := time.Now().UTC() st := loop.State{ - Now: now, - Facts: map[string]store.Fact{"service_down": {Key: "nginx", Ts: now, Source: "poll:healthcheck", Value: `"down"`}}, + Now: now, + Facts: map[string]store.Fact{ + "service_down:nginx": {Key: "service_down:nginx", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}, + "service_down:db": {Key: "service_down:db", Ts: now, Source: loop.ServiceDownSource, Value: `"up"`}, + }, } c := loop.Candidate{Rule: loop.ServiceDownRule(), Severity: loop.Sev4, State: st} pn, _ := NewStub().PhraseNudge(context.Background(), c) @@ -96,12 +98,12 @@ func TestPhraseNudgeServiceDownNamedService(t *testing.T) { } func TestPhraseNudgeServiceDownGenericKey(t *testing.T) { - // the rule key itself ("service_down") rather than a specific service → - // the generic phrase, not a phantom "service_down down on homesrv". + // the old aggregate key, still in the store from before the per-monitor + // facts landed → the generic phrase, never a phantom "service_down down". now := time.Now().UTC() st := loop.State{ Now: now, - Facts: map[string]store.Fact{"service_down": {Key: "service_down", Ts: now, Source: "poll:healthcheck", Value: `"down"`}}, + Facts: map[string]store.Fact{"service_down": {Key: "service_down", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}}, } c := loop.Candidate{Rule: loop.ServiceDownRule(), Severity: loop.Sev4, State: st} pn, _ := NewStub().PhraseNudge(context.Background(), c) @@ -110,6 +112,26 @@ func TestPhraseNudgeServiceDownGenericKey(t *testing.T) { } } +// Two monitors down at once must both be named — he needs to know the blast +// radius, and "a service is down" was the whole defect being fixed here. +func TestPhraseNudgeServiceDownNamesEveryDownMonitor(t *testing.T) { + now := time.Now().UTC() + st := loop.State{ + Now: now, + Facts: map[string]store.Fact{ + "service_down:nginx": {Key: "service_down:nginx", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}, + "service_down:db": {Key: "service_down:db", Ts: now, Source: loop.ServiceDownSource, Value: `"down"`}, + }, + } + c := loop.Candidate{Rule: loop.ServiceDownRule(), Severity: loop.Sev4, State: st} + pn, _ := NewStub().PhraseNudge(context.Background(), c) + for _, want := range []string{"nginx", "db"} { + if !strings.Contains(pn.Body, want) { + t.Fatalf("body should name %q, got %q", want, pn.Body) + } + } +} + func TestPhraseNudgeUnknownRuleFallsBack(t *testing.T) { // a rule without a dedicated template — generic fallback names the rule + // severity gist. never empty. diff --git a/internal/store/facts.go b/internal/store/facts.go index 6e0a2c6..0644fda 100644 --- a/internal/store/facts.go +++ b/internal/store/facts.go @@ -254,6 +254,41 @@ func (s *Store) LatestFactBySource(ctx context.Context, key, source string) (Fac return scanFact(row) } +// LatestFactsByPrefix — the latest non-voided fact for every key that starts +// with prefix, newest-per-key, ordered by key. +// +// The loop's gatherer loads the keys its rules declare, which works while the +// key set is static. Kuma's monitors are not: one fact per monitor means the +// keys are only known once the gauge is read, so the rule declares the prefix +// and this read resolves it per tick. `_` and `%` are escaped — a monitor name +// is user text and must not act as a LIKE wildcard. +func (s *Store) LatestFactsByPrefix(ctx context.Context, prefix string) ([]Fact, error) { + esc := strings.NewReplacer(`\`, `\\`, `%`, `\%`, `_`, `\_`).Replace(prefix) + rows, err := s.db.QueryContext(ctx, ` + SELECT id, ts, kind, key, value, source, confidence, voids_id + FROM facts f + WHERE key LIKE ? ESCAPE '\' + AND id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL) + AND id = (SELECT id FROM facts g + WHERE g.key = f.key + AND g.id NOT IN (SELECT voids_id FROM facts WHERE voids_id IS NOT NULL) + ORDER BY g.ts DESC, g.id DESC LIMIT 1) + ORDER BY key`, esc+"%") + if err != nil { + return nil, fmt.Errorf("facts by prefix %q: %w", prefix, err) + } + defer rows.Close() + var out []Fact + for rows.Next() { + f, err := scanFact(rows) + if err != nil { + return nil, err + } + out = append(out, f) + } + return out, rows.Err() +} + // Since returns how long ago the latest non-voided fact for key landed, or // (0, ErrNoFact). Implements the `since(key)==null → don't fire` guard from // the spec — silence on no-data is "shuts up when uncertain". diff --git a/internal/store/facts_test.go b/internal/store/facts_test.go new file mode 100644 index 0000000..47e666e --- /dev/null +++ b/internal/store/facts_test.go @@ -0,0 +1,62 @@ +package store + +import ( + "context" + "testing" + "time" +) + +// One fact per kuma monitor means the loop cannot name its keys at wiring time, +// so it asks for the family by prefix. The read must return the newest row per +// key and stop at the prefix boundary. +func TestLatestFactsByPrefix(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Millisecond) + write := func(key, val string, at time.Time) int64 { + id, err := s.SetValue(ctx, KindEnv, key, "poll:uptimekuma", val, at) + if err != nil { + t.Fatalf("SetValue %s: %v", key, err) + } + return id + } + write("service_down:db", "up", now.Add(-2*time.Hour)) + write("service_down:db", "down", now.Add(-time.Hour)) // newer wins + write("service_down:web", "up", now.Add(-time.Hour)) + write("service_downtime", "irrelevant", now) // no colon, not in the family + write("water", "250ml", now) + + got, err := s.LatestFactsByPrefix(ctx, "service_down:") + if err != nil { + t.Fatalf("LatestFactsByPrefix: %v", err) + } + if len(got) != 2 { + t.Fatalf("got %d facts, want 2: %+v", len(got), got) + } + if got[0].Key != "service_down:db" || got[0].Value != `"down"` { + t.Errorf("first = %s=%s, want the newest db row", got[0].Key, got[0].Value) + } + if got[1].Key != "service_down:web" { + t.Errorf("second = %s, want service_down:web", got[1].Key) + } +} + +// A monitor name is user text. An underscore in it must match itself, not act +// as a LIKE wildcard and drag in every other monitor. +func TestLatestFactsByPrefixEscapesWildcards(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + now := time.Now().UTC().Truncate(time.Millisecond) + for _, k := range []string{"a_b:one", "axb:two"} { + if _, err := s.SetValue(ctx, KindEnv, k, "poll:uptimekuma", "down", now); err != nil { + t.Fatalf("SetValue %s: %v", k, err) + } + } + got, err := s.LatestFactsByPrefix(ctx, "a_b:") + if err != nil { + t.Fatalf("LatestFactsByPrefix: %v", err) + } + if len(got) != 1 || got[0].Key != "a_b:one" { + t.Fatalf("got %+v, want only a_b:one", got) + } +}