hamlab/internal/api/alerts.go

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"`
}

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 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 <op> 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.host}, 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.host {
				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
				}
			}
		}
	}
	for key := range e.state { // a sensor that is gone (a removed container, a deleted rule) takes its state with it
		if !seen[key] {
			delete(e.state, key)
		}
	}
	e.last = now
	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 {
		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
			}
			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.host, Sensor: "hamlab.test", Value: "this is only a test", Severity: "info"})
	w.WriteHeader(204)
}