feat(monitors): metric rule evaluators and per-server state decisions

2026-09-17 08:40:06 +00:00
parent 1a1f6998a4
commit 12136d1c17
2 changed files with 315 additions and 0 deletions
@@ -0,0 +1,192 @@
+package services
+
+import (
+	"fmt"
+	"path"
+	"strings"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+)
+
+const (
+	MetricDiskPct            = "disk_pct"
+	MetricDiskFreeGB         = "disk_free_gb"
+	MetricMemPct             = "mem_pct"
+	MetricLoadPerCore        = "load_per_core"
+	MetricUnitFailed         = "unit_failed"
+	MetricContainerUnhealthy = "container_unhealthy"
+	MetricRebootPendingDays  = "reboot_pending_days"
+	MetricAgentOfflineMin    = "agent_offline_min"
+)
+
+// metricStaleAfter is how old an agent's metrics can be before a rule stops
+// judging them. A dead agent's last report must not hold an alert open or
+// clear one; agent_offline_min is the rule for a dead agent.
+const metricStaleAfter = 5 * time.Minute
+
+func metricNeedsWorkloads(kind string) bool {
+	return kind == MetricUnitFailed || kind == MetricContainerUnhealthy
+}
+
+func metricUsesThreshold(kind string) bool {
+	return !metricNeedsWorkloads(kind)
+}
+
+func validateMetric(m *models.Monitor) error {
+	t := &m.Target
+	switch t.Metric {
+	case MetricDiskPct, MetricDiskFreeGB, MetricMemPct, MetricLoadPerCore,
+		MetricUnitFailed, MetricContainerUnhealthy, MetricRebootPendingDays, MetricAgentOfflineMin:
+	default:
+		return fmt.Errorf("%w: unknown metric %q", ErrInvalidMonitor, t.Metric)
+	}
+	if metricUsesThreshold(t.Metric) && t.Threshold <= 0 {
+		return fmt.Errorf("%w: threshold must be greater than 0", ErrInvalidMonitor)
+	}
+	if (t.Metric == MetricDiskPct || t.Metric == MetricMemPct) && t.Threshold > 100 {
+		return fmt.Errorf("%w: threshold must be 100 or less", ErrInvalidMonitor)
+	}
+	if t.Mount != "" && !path.IsAbs(t.Mount) {
+		return fmt.Errorf("%w: mount must be an absolute path", ErrInvalidMonitor)
+	}
+	if m.ForSec < 0 {
+		m.ForSec = 0
+	}
+	return nil
+}
+
+// EvaluateMetric judges one server against one rule. ok=false means there is
+// nothing trustworthy to judge - stale metrics, a mount that does not exist on
+// this host - and the caller keeps the previous state rather than guessing.
+func EvaluateMetric(t models.MonitorTarget, srv models.Server, wls []models.Workload, now time.Time) (bool, float64, string, bool) {
+	if t.Metric == MetricAgentOfflineMin {
+		if srv.LastSeen == nil {
+			return false, 0, "", false
+		}
+		mins := now.Sub(*srv.LastSeen).Minutes()
+		return mins >= t.Threshold, mins, fmt.Sprintf("agent not seen for %.0f minutes", mins), true
+	}
+	inv := srv.Inventory
+	if inv == nil || inv.MetricsAt == nil || now.Sub(*inv.MetricsAt) > metricStaleAfter {
+		return false, 0, "", false
+	}
+
+	switch t.Metric {
+	case MetricDiskPct, MetricDiskFreeGB:
+		return evalDisk(t, inv.Partitions)
+	case MetricMemPct:
+		if inv.Memory.TotalBytes == 0 {
+			return false, 0, "", false
+		}
+		pct := float64(inv.Memory.UsedBytes) / float64(inv.Memory.TotalBytes) * 100
+		return pct >= t.Threshold, pct, fmt.Sprintf("memory %.1f%% used", pct), true
+	case MetricLoadPerCore:
+		if inv.CPU.Cores == 0 {
+			return false, 0, "", false
+		}
+		v := inv.CPU.Load1 / float64(inv.CPU.Cores)
+		return v >= t.Threshold, v, fmt.Sprintf("load %.2f per core", v), true
+	case MetricRebootPendingDays:
+		if !inv.RebootRequired || inv.RebootRequiredSince == nil {
+			return false, 0, "", true
+		}
+		days := now.Sub(*inv.RebootRequiredSince).Hours() / 24
+		return days >= t.Threshold, days, fmt.Sprintf("reboot pending for %.0f days", days), true
+	case MetricUnitFailed:
+		return evalWorkloads(wls, "unit", func(w models.Workload) bool { return w.State == "failed" }, "failed")
+	case MetricContainerUnhealthy:
+		return evalWorkloads(wls, "container", func(w models.Workload) bool { return w.Health == "unhealthy" }, "unhealthy")
+	}
+	return false, 0, "", false
+}
+
+// evalDisk reports the worst matching partition, so "any mount" names the one
+// that is actually full.
+func evalDisk(t models.MonitorTarget, parts []models.Partition) (bool, float64, string, bool) {
+	found := false
+	var worstBreach bool
+	var worstVal float64
+	var worstMsg string
+	for _, p := range parts {
+		if p.TotalBytes == 0 || (t.Mount != "" && p.Mountpoint != t.Mount) {
+			continue
+		}
+		var breach bool
+		var val float64
+		var msg string
+		if t.Metric == MetricDiskPct {
+			val = float64(p.UsedBytes) / float64(p.TotalBytes) * 100
+			breach = val >= t.Threshold
+			msg = fmt.Sprintf("%s %.1f%% used", p.Mountpoint, val)
+		} else {
+			val = float64(p.TotalBytes-p.UsedBytes) / 1e9
+			breach = val <= t.Threshold
+			msg = fmt.Sprintf("%s %.1f GB free", p.Mountpoint, val)
+		}
+		worse := !found ||
+			(t.Metric == MetricDiskPct && val > worstVal) ||
+			(t.Metric == MetricDiskFreeGB && val < worstVal)
+		if worse {
+			worstBreach, worstVal, worstMsg = breach, val, msg
+		}
+		found = true
+	}
+	if !found {
+		return false, 0, "", false
+	}
+	return worstBreach, worstVal, worstMsg, true
+}
+
+func evalWorkloads(wls []models.Workload, kind string, bad func(models.Workload) bool, word string) (bool, float64, string, bool) {
+	var names []string
+	for _, w := range wls {
+		if w.Kind == kind && bad(w) {
+			names = append(names, w.Name)
+		}
+	}
+	if len(names) == 0 {
+		return false, 0, "", true
+	}
+	return true, float64(len(names)), fmt.Sprintf("%s %s: %s", kind, word, strings.Join(names, ", ")), true
+}
+
+// nextServerState applies the "for N seconds" gate. BreachSince is the first
+// sweep that saw the condition, so the gate measures continuous breach and a
+// single clear sweep restarts it.
+func nextServerState(prev *models.MonitorServerState, breach bool, forSec int, now time.Time) (string, *time.Time) {
+	if !breach {
+		return models.StatusUp, nil
+	}
+	since := now
+	if prev != nil && prev.BreachSince != nil {
+		since = *prev.BreachSince
+	}
+	if now.Sub(since) >= time.Duration(forSec)*time.Second {
+		return models.StatusDown, &since
+	}
+	return models.StatusPending, &since
+}
+
+func rollupParent(states []models.MonitorServerState) (string, string) {
+	if len(states) == 0 {
+		return models.StatusUp, "no matching servers"
+	}
+	down, pending := 0, 0
+	for _, s := range states {
+		switch s.Status {
+		case models.StatusDown:
+			down++
+		case models.StatusPending:
+			pending++
+		}
+	}
+	msg := fmt.Sprintf("%d of %d servers breaching", down, len(states))
+	switch {
+	case down > 0:
+		return models.StatusDown, msg
+	case pending > 0:
+		return models.StatusPending, msg
+	}
+	return models.StatusUp, msg
+}
@@ -0,0 +1,123 @@
+package services
+
+import (
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.hostxtra.co.uk/mrhid6/vantage/server/internal/models"
+)
+
+var evalNow = time.Date(2026, 9, 17, 12, 0, 0, 0, time.UTC)
+
+func fresh() *time.Time { t := evalNow.Add(-time.Minute); return &t }
+
+func srvWith(inv models.Inventory) models.Server {
+	if inv.MetricsAt == nil {
+		inv.MetricsAt = fresh()
+	}
+	ls := evalNow.Add(-30 * time.Second)
+	return models.Server{ServerID: "s1", Hostname: "web-01", Inventory: &inv, LastSeen: &ls}
+}
+
+func TestEvaluateMetric(t *testing.T) {
+	disk := models.Inventory{Partitions: []models.Partition{
+		{Mountpoint: "/", TotalBytes: 100e9, UsedBytes: 50e9},
+		{Mountpoint: "/var", TotalBytes: 100e9, UsedBytes: 95e9},
+	}}
+	stale := evalNow.Add(-10 * time.Minute)
+	since := evalNow.Add(-8 * 24 * time.Hour)
+	oldSeen := evalNow.Add(-20 * time.Minute)
+
+	cases := []struct {
+		name       string
+		t          models.MonitorTarget
+		srv        models.Server
+		wls        []models.Workload
+		wantBreach bool
+		wantOK     bool
+		msgHas     string
+	}{
+		{"disk pct any mount breaches", models.MonitorTarget{Metric: MetricDiskPct, Threshold: 90}, srvWith(disk), nil, true, true, "/var"},
+		{"disk pct named mount clear", models.MonitorTarget{Metric: MetricDiskPct, Threshold: 90, Mount: "/"}, srvWith(disk), nil, false, true, "/"},
+		{"disk pct missing mount skips", models.MonitorTarget{Metric: MetricDiskPct, Threshold: 90, Mount: "/data"}, srvWith(disk), nil, false, false, ""},
+		{"disk free gb breaches", models.MonitorTarget{Metric: MetricDiskFreeGB, Threshold: 10}, srvWith(disk), nil, true, true, "GB free"},
+		{"zero total partition ignored", models.MonitorTarget{Metric: MetricDiskPct, Threshold: 1}, srvWith(models.Inventory{Partitions: []models.Partition{{Mountpoint: "/proc"}}}), nil, false, false, ""},
+		{"mem pct", models.MonitorTarget{Metric: MetricMemPct, Threshold: 80}, srvWith(models.Inventory{Memory: models.MemInfo{TotalBytes: 100, UsedBytes: 85}}), nil, true, true, "memory"},
+		{"load per core", models.MonitorTarget{Metric: MetricLoadPerCore, Threshold: 1.5}, srvWith(models.Inventory{CPU: models.CPUInfo{Cores: 4, Load1: 8}}), nil, true, true, "load"},
+		{"load no cores skips", models.MonitorTarget{Metric: MetricLoadPerCore, Threshold: 1.5}, srvWith(models.Inventory{CPU: models.CPUInfo{Load1: 8}}), nil, false, false, ""},
+		{"stale inventory skips", models.MonitorTarget{Metric: MetricMemPct, Threshold: 1}, srvWith(models.Inventory{MetricsAt: &stale, Memory: models.MemInfo{TotalBytes: 100, UsedBytes: 99}}), nil, false, false, ""},
+		{"unit failed", models.MonitorTarget{Metric: MetricUnitFailed}, srvWith(models.Inventory{}), []models.Workload{{Kind: "unit", Name: "backup.service", State: "failed"}}, true, true, "backup.service"},
+		{"container unhealthy clear", models.MonitorTarget{Metric: MetricContainerUnhealthy}, srvWith(models.Inventory{}), []models.Workload{{Kind: "container", Name: "db", Health: "healthy"}}, false, true, ""},
+		{"reboot pending days", models.MonitorTarget{Metric: MetricRebootPendingDays, Threshold: 7}, srvWith(models.Inventory{RebootRequired: true, RebootRequiredSince: &since}), nil, true, true, "reboot"},
+		{"reboot not required", models.MonitorTarget{Metric: MetricRebootPendingDays, Threshold: 7}, srvWith(models.Inventory{}), nil, false, true, ""},
+		{"agent offline ignores stale inventory", models.MonitorTarget{Metric: MetricAgentOfflineMin, Threshold: 10},
+			func() models.Server { s := srvWith(models.Inventory{MetricsAt: &stale}); s.LastSeen = &oldSeen; return s }(), nil, true, true, "not seen"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			breach, _, msg, ok := EvaluateMetric(c.t, c.srv, c.wls, evalNow)
+			if breach != c.wantBreach || ok != c.wantOK || !strings.Contains(msg, c.msgHas) {
+				t.Fatalf("got breach=%v ok=%v msg=%q", breach, ok, msg)
+			}
+		})
+	}
+}
+
+func TestNextServerState(t *testing.T) {
+	past := func(d time.Duration) *time.Time { v := evalNow.Add(-d); return &v }
+
+	st, since := nextServerState(nil, true, 300, evalNow)
+	if st != models.StatusPending || since == nil || !since.Equal(evalNow) {
+		t.Fatalf("new breach: %s %v", st, since)
+	}
+	st, _ = nextServerState(&models.MonitorServerState{Status: models.StatusPending, BreachSince: past(2 * time.Minute)}, true, 300, evalNow)
+	if st != models.StatusPending {
+		t.Fatalf("under for_sec should stay pending, got %s", st)
+	}
+	st, since = nextServerState(&models.MonitorServerState{Status: models.StatusPending, BreachSince: past(6 * time.Minute)}, true, 300, evalNow)
+	if st != models.StatusDown || !since.Equal(*past(6 * time.Minute)) {
+		t.Fatalf("over for_sec should be down keeping since, got %s %v", st, since)
+	}
+	st, _ = nextServerState(nil, true, 0, evalNow)
+	if st != models.StatusDown {
+		t.Fatalf("for_sec 0 is down immediately, got %s", st)
+	}
+	st, since = nextServerState(&models.MonitorServerState{Status: models.StatusDown, BreachSince: past(time.Hour)}, false, 300, evalNow)
+	if st != models.StatusUp || since != nil {
+		t.Fatalf("clear should be up with no since, got %s %v", st, since)
+	}
+}
+
+func TestRollupParent(t *testing.T) {
+	if s, m := rollupParent(nil); s != models.StatusUp || m != "no matching servers" {
+		t.Fatalf("empty: %s %q", s, m)
+	}
+	states := []models.MonitorServerState{{Status: models.StatusUp}, {Status: models.StatusPending}, {Status: models.StatusDown}}
+	if s, m := rollupParent(states); s != models.StatusDown || m != "1 of 3 servers breaching" {
+		t.Fatalf("got %s %q", s, m)
+	}
+	if s, _ := rollupParent(states[:2]); s != models.StatusPending {
+		t.Fatalf("pending wins over up, got %s", s)
+	}
+}
+
+func TestValidateMetric(t *testing.T) {
+	bad := []models.MonitorTarget{
+		{Metric: "cpu_magic", Threshold: 1},
+		{Metric: MetricDiskPct, Threshold: 0},
+		{Metric: MetricDiskPct, Threshold: 101},
+		{Metric: MetricDiskPct, Threshold: 90, Mount: "var"},
+	}
+	for _, tg := range bad {
+		m := models.Monitor{Type: models.MonitorMetric, Target: tg}
+		if err := validateMetric(&m); !errors.Is(err, ErrInvalidMonitor) {
+			t.Errorf("%+v: want ErrInvalidMonitor, got %v", tg, err)
+		}
+	}
+	ok := models.Monitor{Type: models.MonitorMetric, ForSec: -5, Target: models.MonitorTarget{Metric: MetricUnitFailed}}
+	if err := validateMetric(&ok); err != nil || ok.ForSec != 0 {
+		t.Fatalf("unit_failed needs no threshold and negative for_sec clamps to 0: %v %d", err, ok.ForSec)
+	}
+}