package api // Alerts: rules over sensors, watched by the hub for itself and every machine it manages. // // A rule is a selector, a comparison and a patience: `container.*.running == false for // 120 s`. Every alertEvery the hub asks each machine one shaped query for all the rules' // selectors, compares, and keeps a little state per (rule, machine, sensor): ok, pending // since, or firing since. Going to firing, and coming back, is an event — kept, shown on // the Alerts tab, and sent to whatever channels are set (ntfy, or a webhook: Home // Assistant, Slack, anything that takes a POST). `hamlab.reachable` is a sensor the hub // supplies itself, so "that machine stopped answering" is a rule like any other. import ( "bytes" "context" "encoding/json" "fmt" "io" "log" "net/http" "sort" "strconv" "strings" "sync" "time" "hamlab/internal/sensors" "hamlab/internal/store" ) const ( alertEvery = 15 * time.Second maxEvents = 200 ) type AlertRule struct { ID string `json:"id"` Name string `json:"name"` Sensor string `json:"sensor"` // an ID or a pattern Op string `json:"op"` // > >= < <= == != Value any `json:"value"` // number, bool or string, to match the sensor ForSec int `json:"for_sec"` Severity string `json:"severity"` // info | warn | crit Enabled bool `json:"enabled"` Machines string `json:"machines"` // "*" or comma-separated names; "" is "*" } type AlertChannel struct { Type string `json:"type"` // ntfy | webhook URL string `json:"url"` } type AlertEvent struct { T int64 `json:"t"` Kind string `json:"kind"` // firing | resolved Rule string `json:"rule"` RuleID string `json:"rule_id"` Machine string `json:"machine"` Sensor string `json:"sensor"` Value any `json:"value"` Severity string `json:"severity"` } type alertFiring struct { AlertEvent Since int64 `json:"since"` } type alertSaved struct { Rules []AlertRule `json:"rules"` Channels []AlertChannel `json:"channels"` Events []AlertEvent `json:"events"` Seeded bool `json:"seeded"` // Firing is what was firing when the hub last looked, so a restart (every deploy is one) // carries on where it was and doesn't announce the same alerts all over again. Firing map[string]alertKept `json:"firing,omitempty"` } type alertKept struct { Since int64 `json:"since"` Value any `json:"value"` } const maxRules = 200 type alertState struct { pending time.Time firing bool since time.Time value any } type alertEngine struct { s *Server mu sync.Mutex saved alertSaved state map[string]*alertState // ruleID|machine|sensor last time.Time } var defaultRules = []AlertRule{ {ID: "machine-offline", Name: "Machine not answering", Sensor: "hamlab.reachable", Op: "==", Value: false, ForSec: 120, Severity: "crit", Enabled: true}, {ID: "container-down", Name: "Container down", Sensor: "container.*.down", Op: "==", Value: true, ForSec: 120, Severity: "crit", Enabled: true}, {ID: "container-unhealthy", Name: "Container unhealthy", Sensor: "container.*.health", Op: "==", Value: "unhealthy", ForSec: 180, Severity: "warn", Enabled: true}, {ID: "service-failed", Name: "A service has failed", Sensor: "services.failed", Op: ">", Value: 0.0, ForSec: 300, Severity: "warn", Enabled: true}, {ID: "disk-full", Name: "Filesystem nearly full", Sensor: "fs.*.used_pct", Op: ">=", Value: 90.0, ForSec: 300, Severity: "warn", Enabled: true}, {ID: "pool-unhealthy", Name: "ZFS pool not healthy", Sensor: "zfs.*.healthy", Op: "==", Value: false, ForSec: 60, Severity: "crit", Enabled: true}, {ID: "drive-failing", Name: "Drive failing SMART", Sensor: "smart.*.passed", Op: "==", Value: false, ForSec: 0, Severity: "crit", Enabled: true}, {ID: "on-battery", Name: "UPS on battery", Sensor: "ups.on_battery", Op: "==", Value: true, ForSec: 30, Severity: "crit", Enabled: true}, {ID: "hot", Name: "CPU running hot", Sensor: "cpu.temp", Op: ">=", Value: 90.0, ForSec: 300, Severity: "warn", Enabled: true}, {ID: "memory", Name: "Memory nearly exhausted", Sensor: "mem.used_pct", Op: ">=", Value: 95.0, ForSec: 300, Severity: "warn", Enabled: true}, } func newAlertEngine(s *Server) *alertEngine { e := &alertEngine{s: s, state: map[string]*alertState{}} var raw json.RawMessage s.store.View(func(st *store.State) { raw = st.Alerts }) _ = json.Unmarshal(raw, &e.saved) if !e.saved.Seeded { e.saved.Rules, e.saved.Seeded = append([]AlertRule{}, defaultRules...), true } for key, k := range e.saved.Firing { e.state[key] = &alertState{pending: time.UnixMilli(k.Since), firing: true, since: time.UnixMilli(k.Since), value: k.Value} } for i, r := range e.saved.Rules { // the first day's default watched "not running", which a container stopped on purpose also is if r.ID == "container-down" && r.Sensor == "container.*.running" { e.saved.Rules[i].Name, e.saved.Rules[i].Sensor, e.saved.Rules[i].Value = "Container down", "container.*.down", true } } go e.run() return e } func (e *alertEngine) persist() { b, _ := json.Marshal(e.saved) _ = e.s.store.Update(func(st *store.State) { st.Alerts = b }) } func (e *alertEngine) run() { time.Sleep(20 * time.Second) // let the status sensors take their first readings for { e.evaluate() time.Sleep(alertEvery) } } func wants(r AlertRule, machine string) bool { if r.Machines == "" || r.Machines == "*" { return true } for _, m := range strings.Split(r.Machines, ",") { if strings.TrimSpace(m) == machine { return true } } return false } // holds reports whether `got want`, comparing like with like. func holds(got any, op string, want any) bool { switch g := got.(type) { case float64: w, ok := want.(float64) if !ok { if s, isStr := want.(string); isStr { f, err := strconv.ParseFloat(s, 64) if err != nil { return false } w = f } else { return false } } switch op { case ">": return g > w case ">=": return g >= w case "<": return g < w case "<=": return g <= w case "==": return g == w case "!=": return g != w } case bool: w, ok := want.(bool) if !ok { w = fmt.Sprint(want) == "true" } return (op == "==" && g == w) || (op == "!=" && g != w) case string: w := fmt.Sprint(want) return (op == "==" && g == w) || (op == "!=" && g != w) } return false } func (e *alertEngine) evaluate() { e.mu.Lock() rules := append([]AlertRule{}, e.saved.Rules...) e.mu.Unlock() query := map[string]string{} for _, r := range rules { if r.Enabled && r.Sensor != "hamlab.reachable" { query[r.ID] = r.Sensor } } qb, _ := json.Marshal(query) machines := append([]string{e.s.hostName()}, e.s.nodes.Names()...) results := map[string]map[string]any{} // machine → ruleID → value | {id: value} reach := map[string]bool{} var wg sync.WaitGroup var mu sync.Mutex for _, m := range machines { wg.Add(1) go func(m string) { defer wg.Done() var res map[string]any ok := false if m == e.s.hostName() { if out, err := sensors.Shape(e.s.sensors.All(), qb, false); err == nil { res, ok = out.(map[string]any) } } else { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) code, b, err := e.s.nodes.Do(ctx, m, "POST", "/api/sensors/query", qb, http.Header{"Content-Type": {"application/json"}}) cancel() ok = err == nil && code == 200 && json.Unmarshal(b, &res) == nil } mu.Lock() results[m], reach[m] = res, ok mu.Unlock() }(m) } wg.Wait() now := time.Now() seen := map[string]bool{} var events []AlertEvent e.mu.Lock() for _, r := range rules { if !r.Enabled { continue } for _, m := range machines { if !wants(r, m) { continue } readings := map[string]any{} if r.Sensor == "hamlab.reachable" { readings["hamlab.reachable"] = reach[m] } else if !reach[m] { continue // no reading is not a reading: an unreachable machine is the offline rule's business } else if v, ok := results[m][r.ID]; ok { if many, isMap := v.(map[string]any); isMap { readings = many } else if v != nil { readings[r.Sensor] = v } } for id, val := range readings { key := r.ID + "|" + m + "|" + id seen[key] = true st := e.state[key] if st == nil { st = &alertState{} e.state[key] = st } st.value = val if holds(val, r.Op, r.Value) { if st.pending.IsZero() { st.pending = now } if !st.firing && now.Sub(st.pending) >= time.Duration(r.ForSec)*time.Second { st.firing, st.since = true, now events = append(events, AlertEvent{T: now.UnixMilli(), Kind: "firing", Rule: r.Name, RuleID: r.ID, Machine: m, Sensor: id, Value: val, Severity: r.Severity}) } } else { if st.firing { events = append(events, AlertEvent{T: now.UnixMilli(), Kind: "resolved", Rule: r.Name, RuleID: r.ID, Machine: m, Sensor: id, Value: val, Severity: r.Severity}) } st.pending, st.firing = time.Time{}, false } } } } byID := map[string]AlertRule{} for _, r := range rules { byID[r.ID] = r } for key, st := range e.state { // no longer watched: the rule was paused, deleted or narrowed, or the sensor is gone if seen[key] { continue } if st.firing { // it was on the Alerts tab and perhaps on a phone: say that it is over, and why part := strings.SplitN(key, "|", 3) r, known := byID[part[0]] name, why := r.Name, "no longer watched" if !known { name, why = part[0], "rule deleted" } else if !r.Enabled { why = "rule paused" } if name == "" { name = part[0] } events = append(events, AlertEvent{T: now.UnixMilli(), Kind: "resolved", Rule: name, RuleID: part[0], Machine: part[1], Sensor: part[2], Value: why, Severity: r.Severity}) } delete(e.state, key) } e.last = now kept := map[string]alertKept{} for key, st := range e.state { if st.firing { kept[key] = alertKept{Since: st.since.UnixMilli(), Value: st.value} } } changed := len(kept) != len(e.saved.Firing) for k := range kept { if _, ok := e.saved.Firing[k]; !ok { changed = true } } e.saved.Firing = kept if changed && len(events) == 0 { e.persist() } if len(events) > 0 { e.saved.Events = append(e.saved.Events, events...) if len(e.saved.Events) > maxEvents { e.saved.Events = e.saved.Events[len(e.saved.Events)-maxEvents:] } e.persist() } channels := append([]AlertChannel{}, e.saved.Channels...) e.mu.Unlock() for _, ev := range events { e.notify(channels, ev) } } func (e *alertEngine) firing() []alertFiring { out := []alertFiring{} names := map[string]AlertRule{} for _, r := range e.saved.Rules { names[r.ID] = r } for key, st := range e.state { if !st.firing { continue } p := strings.SplitN(key, "|", 3) r := names[p[0]] out = append(out, alertFiring{AlertEvent: AlertEvent{T: st.since.UnixMilli(), Kind: "firing", Rule: r.Name, RuleID: r.ID, Machine: p[1], Sensor: p[2], Value: st.value, Severity: r.Severity}, Since: st.since.UnixMilli()}) } sort.Slice(out, func(i, j int) bool { return out[i].Since > out[j].Since }) return out } func describe(ev AlertEvent) (title, body string) { mark := map[string]string{"firing": "▲", "resolved": "✓"}[ev.Kind] title = fmt.Sprintf("%s %s — %s", mark, ev.Rule, ev.Machine) body = fmt.Sprintf("%s = %v", ev.Sensor, ev.Value) if ev.Kind == "resolved" { body = "back to normal: " + body } return } func (e *alertEngine) notify(channels []AlertChannel, ev AlertEvent) { title, body := describe(ev) for _, c := range channels { var req *http.Request switch c.Type { case "ntfy": req, _ = http.NewRequest("POST", c.URL, strings.NewReader(body)) req.Header.Set("Title", title) req.Header.Set("Priority", map[string]string{"crit": "urgent", "warn": "high"}[ev.Severity]) req.Header.Set("Tags", map[string]string{"firing": "warning", "resolved": "white_check_mark"}[ev.Kind]) if ev.Kind == "resolved" { req.Header.Set("Priority", "default") } case "webhook": b, _ := json.Marshal(map[string]any{"title": title, "message": body, "event": ev}) req, _ = http.NewRequest("POST", c.URL, bytes.NewReader(b)) req.Header.Set("Content-Type", "application/json") } if req == nil { continue } go func(req *http.Request, url string) { ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() resp, err := http.DefaultClient.Do(req.WithContext(ctx)) if err != nil { log.Printf("alert: notifying %s: %v", url, err) return } io.Copy(io.Discard, resp.Body) resp.Body.Close() if resp.StatusCode >= 300 { log.Printf("alert: %s answered %d", url, resp.StatusCode) } }(req, c.URL) } } // ---- API ---- func (s *Server) alertsGet(w http.ResponseWriter, _ *http.Request) { e := s.alerts e.mu.Lock() defer e.mu.Unlock() ev := append([]AlertEvent{}, e.saved.Events...) for i, j := 0, len(ev)-1; i < j; i, j = i+1, j-1 { ev[i], ev[j] = ev[j], ev[i] } rules, channels := e.saved.Rules, e.saved.Channels if rules == nil { // an empty list is a list: never hand the page a null to take the length of rules = []AlertRule{} } if channels == nil { channels = []AlertChannel{} } jsonOK(w, 200, map[string]any{"rules": rules, "channels": channels, "events": ev, "firing": e.firing(), "checked": e.last.UnixMilli(), "every_sec": int(alertEvery.Seconds())}) } func (s *Server) alertsPut(w http.ResponseWriter, r *http.Request) { var body struct { Rules *[]AlertRule `json:"rules"` Channels *[]AlertChannel `json:"channels"` } if err := json.NewDecoder(io.LimitReader(r.Body, 256<<10)).Decode(&body); err != nil { jsonErr(w, 400, "bad request") return } if body.Rules != nil && len(*body.Rules) > maxRules { jsonErr(w, 400, fmt.Sprintf("at most %d rules: every one is checked on every machine every %d s", maxRules, int(alertEvery.Seconds()))) return } if body.Rules != nil { ids := map[string]bool{} for i := range *body.Rules { ru := &(*body.Rules)[i] ru.Name, ru.Sensor = strings.TrimSpace(ru.Name), strings.TrimSpace(ru.Sensor) if ru.ID == "" || ids[ru.ID] || ru.Name == "" || len(ru.Name) > maxNameLen || ru.ForSec < 0 || ru.ForSec > 86400 { jsonErr(w, 400, "every rule needs a unique id, a name (64 characters at most) and a patience between 0 and 86400 s") return } ids[ru.ID] = true if _, err := sensors.Matcher(ru.Sensor); err != nil || ru.Sensor == "" { jsonErr(w, 400, fmt.Sprintf("rule %q: %q is not a sensor ID or pattern", ru.Name, ru.Sensor)) return } if !strings.Contains(" > >= < <= == != ", " "+ru.Op+" ") { jsonErr(w, 400, fmt.Sprintf("rule %q: the comparison must be one of > >= < <= == !=", ru.Name)) return } switch val := ru.Value.(type) { case float64, bool: case string: if len(val) > 200 { jsonErr(w, 400, fmt.Sprintf("rule %q: the value is too long", ru.Name)) return } if ru.Op != "==" && ru.Op != "!=" { // "cpu.usage > ninety" can never be true, and would never say so jsonErr(w, 400, fmt.Sprintf("rule %q: %s compares numbers, and %q is not one; text can only be == or !=", ru.Name, ru.Op, val)) return } default: jsonErr(w, 400, fmt.Sprintf("rule %q: the value must be a number, true/false or text", ru.Name)) return } if _, isBool := ru.Value.(bool); isBool && ru.Op != "==" && ru.Op != "!=" { jsonErr(w, 400, fmt.Sprintf("rule %q: true/false can only be == or !=", ru.Name)) return } if ru.Severity != "info" && ru.Severity != "crit" { ru.Severity = "warn" } } } if body.Channels != nil { for _, c := range *body.Channels { if (c.Type != "ntfy" && c.Type != "webhook") || !(strings.HasPrefix(c.URL, "https://") || strings.HasPrefix(c.URL, "http://")) { jsonErr(w, 400, "a channel is {type: ntfy|webhook, url: http(s)://…}") return } } } e := s.alerts e.mu.Lock() if body.Rules != nil { e.saved.Rules = *body.Rules } if body.Channels != nil { e.saved.Channels = *body.Channels } e.persist() e.mu.Unlock() w.WriteHeader(204) } // alertsTest sends one clearly-marked test notification through every channel. func (s *Server) alertsTest(w http.ResponseWriter, _ *http.Request) { e := s.alerts e.mu.Lock() channels := append([]AlertChannel{}, e.saved.Channels...) e.mu.Unlock() if len(channels) == 0 { jsonErr(w, 400, "no channels to test") return } e.notify(channels, AlertEvent{T: time.Now().UnixMilli(), Kind: "firing", Rule: "Test alert", Machine: s.hostName(), Sensor: "hamlab.test", Value: "this is only a test", Severity: "info"}) w.WriteHeader(204) }