fix zombie leak, add quiet-hours toggle, improve query reply, configurable router threshold, JS dashboard

2026-07-03 00:42:35 +02:00
parent 612583d59a
commit e00cb07658
26 changed files with 3652 additions and 15 deletions
@@ -0,0 +1,341 @@
+// mavpoll — the env poller module.
+//
+// maven doesn't collect metrics; netdata and uptime-kuma already do, tuned to
+// the box. This is a thin adapter: it reads their alarms/status and writes
+// `facts (kind=env, source=poll:*)` through core's IPC socket. Key-free,
+// restart-free, fail-independent — a crashing poller can't touch the store key
+// (it never had it), worst case a stale env fact until the next tick.
+//
+// Two sources, each its own provenance (the loop's rules trust source):
+//   - netdata  → poll:netdata      resource alarms (disk/mem/cert/temp)
+//   - kuma     → poll:uptimekuma   service up/down (the source of truth for it)
+//
+// Netdata needs no auth over the wg-fronted net. Kuma's /metrics needs an API
+// key (basic-auth); without -kuma the whole kuma path is skipped (netdata-only
+// still lights up an end-to-end nudge).
+//
+// Append-only discipline: a fact is written only when its value CHANGED vs the
+// latest for that key+source. A poller that wrote every 60s would churn the
+// facts table for nothing; the store is the audit trail, not a metrics sink.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"os/exec"
+	"os/signal"
+	"regexp"
+	"strconv"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/kami/maven/internal/ipc"
+)
+
+func main() {
+	if err := run(os.Args[1:]); err != nil {
+		fmt.Fprintln(os.Stderr, "mavpoll:", err)
+		os.Exit(1)
+	}
+}
+
+func run(args []string) error {
+	fs := flag.NewFlagSet("mavpoll", flag.ContinueOnError)
+	socket := fs.String("socket", "", "core IPC socket path (required)")
+	netdataURL := fs.String("netdata", "http://127.0.0.1:19999", "netdata base URL ('' to disable)")
+	kumaURL := fs.String("kuma", "", "uptime-kuma metrics URL, e.g. http://127.0.0.1:3001/metrics ('' to disable)")
+	kumaKey := fs.String("kuma-key", "", "uptime-kuma API key (basic-auth username)")
+	wgIface := fs.String("wg", "", "wireguard interface for the presence signal, e.g. wg0 or 'all' ('' to disable)")
+	wgCmd := fs.String("wg-cmd", "wg", "wg binary (use e.g. 'sudo wg' if the poller lacks CAP_NET_ADMIN)")
+	interval := fs.Duration("interval", 60*time.Second, "poll cadence")
+	timeout := fs.Duration("timeout", 8*time.Second, "per-request HTTP timeout")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	if *socket == "" {
+		return fmt.Errorf("-socket is required")
+	}
+	if *netdataURL == "" && *kumaURL == "" && *wgIface == "" {
+		return fmt.Errorf("nothing to poll: set -netdata, -kuma and/or -wg")
+	}
+
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	core, err := ipc.Dial(*socket)
+	if err != nil {
+		return err
+	}
+	defer core.Close()
+
+	p := &poller{
+		core:       core,
+		http:       &http.Client{Timeout: *timeout},
+		netdataURL: strings.TrimRight(*netdataURL, "/"),
+		kumaURL:    *kumaURL,
+		kumaKey:    *kumaKey,
+		wgIface:    *wgIface,
+		wgCmd:      *wgCmd,
+	}
+
+	log.Printf("mavpoll: polling every %s (netdata=%q kuma=%q wg=%q)", *interval, *netdataURL, *kumaURL, *wgIface)
+	p.pollOnce(ctx) // fire immediately; don't idle a full interval on start
+	t := time.NewTicker(*interval)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			log.Printf("mavpoll: bye")
+			return nil
+		case <-t.C:
+			p.pollOnce(ctx)
+		}
+	}
+}
+
+type poller struct {
+	core       ipc.CoreAPI
+	http       *http.Client
+	netdataURL string
+	kumaURL    string
+	kumaKey    string
+	wgIface    string
+	wgCmd      string
+}
+
+// pollOnce — one sweep of both sources. A failure in one source logs and does
+// NOT abort the other: netdata being down shouldn't blind kuma and vice versa.
+func (p *poller) pollOnce(ctx context.Context) {
+	now := time.Now()
+	if p.netdataURL != "" {
+		if err := p.pollNetdata(ctx, now); err != nil {
+			log.Printf("mavpoll: netdata: %v", err)
+		}
+	}
+	if p.kumaURL != "" {
+		if err := p.pollKuma(ctx, now); err != nil {
+			log.Printf("mavpoll: kuma: %v", err)
+		}
+	}
+	if p.wgIface != "" {
+		if err := p.pollWg(ctx); err != nil {
+			log.Printf("mavpoll: wg: %v", err)
+		}
+	}
+}
+
+// ---- wireguard: latest handshake → presence signal -------------------------
+
+// pollWg reads `wg show <iface> latest-handshakes` and writes a wg_handshake
+// fact (source=infer:wg) stamped with the MOST RECENT peer handshake time — not
+// now(). Presence decays from the real handshake instant, so the fact's ts must
+// be that instant. We write only when the handshake ADVANCES vs the last fact,
+// so a quiet tunnel produces no churn (and presence just decays out, τ=20min).
+//
+// `wg show` needs CAP_NET_ADMIN; run mavpoll with the cap or set -wg-cmd "sudo wg".
+func (p *poller) pollWg(ctx context.Context) error {
+	fields := strings.Fields(p.wgCmd)
+	args := append(fields[1:], "show", p.wgIface, "latest-handshakes")
+	out, err := exec.CommandContext(ctx, fields[0], args...).Output()
+	if err != nil {
+		return fmt.Errorf("run %s: %w", p.wgCmd, err)
+	}
+	maxTs := parseMaxHandshake(string(out))
+	if maxTs == 0 {
+		return nil // no peer has ever handshaked → drop out of presence
+	}
+	hs := time.Unix(maxTs, 0)
+	prev, err := p.core.LatestFactBySource(ctx, "wg_handshake", "infer:wg")
+	if err == nil && !hs.After(prev.Ts) {
+		return nil // not newer → no churn
+	}
+	if err != nil && err != ipc.ErrNoFact && !isNoFact(err) {
+		return fmt.Errorf("read wg_handshake: %w", err)
+	}
+	if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
+		Ts: hs, Kind: "env", Key: "wg_handshake", Value: `"up"`,
+		Source: "infer:wg", Confidence: 1.0,
+	}); err != nil {
+		return fmt.Errorf("write wg_handshake: %w", err)
+	}
+	log.Printf("mavpoll: wg_handshake @ %s (infer:wg)", hs.Format(time.RFC3339))
+	return nil
+}
+
+// parseMaxHandshake — max last-field unix ts across `wg show latest-handshakes`
+// lines. Handles both the per-iface form (`<pubkey>\t<ts>`) and the `all` form
+// (`<iface>\t<pubkey>\t<ts>`); the timestamp is always the last field. 0 = none.
+func parseMaxHandshake(out string) int64 {
+	var max int64
+	for _, line := range strings.Split(out, "\n") {
+		f := strings.Fields(line)
+		if len(f) == 0 {
+			continue
+		}
+		ts, err := strconv.ParseInt(f[len(f)-1], 10, 64)
+		if err == nil && ts > max {
+			max = ts
+		}
+	}
+	return max
+}
+
+// ---- netdata: active alarms → aggregate severity ---------------------------
+
+// netdata /api/v1/alarms?active=true returns {"alarms": {"<chart.name>": {...,
+// "status": "WARNING"|"CRITICAL"|"CLEAR"|...}}}. We only need the max active
+// severity; a rule fires on "critical". The per-alarm detail lives in netdata's
+// own UI — we don't re-store it (YAGNI; add a per-alarm fact when a rule needs
+// one specific alarm by name).
+type netdataAlarms struct {
+	Alarms map[string]struct {
+		Status string `json:"status"`
+	} `json:"alarms"`
+}
+
+func (p *poller) pollNetdata(ctx context.Context, now time.Time) error {
+	body, err := p.get(ctx, p.netdataURL+"/api/v1/alarms?active=true", "")
+	if err != nil {
+		return err
+	}
+	var a netdataAlarms
+	if err := json.Unmarshal(body, &a); err != nil {
+		return fmt.Errorf("decode alarms: %w", err)
+	}
+	return p.writeIfChanged(ctx, "netdata_alarm", "poll:netdata", maxSeverity(a), now)
+}
+
+// maxSeverity reduces active alarms to the aggregate the rule consumes.
+func maxSeverity(a netdataAlarms) string {
+	sev := "clear"
+	for _, al := range a.Alarms {
+		switch strings.ToUpper(al.Status) {
+		case "CRITICAL":
+			return "critical" // highest — short-circuit
+		case "WARNING":
+			sev = "warning"
+		}
+	}
+	return sev
+}
+
+// ---- kuma: monitor_status gauge → aggregate service_down -------------------
+
+// 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+-]+)`)
+
+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 {
+		return fmt.Errorf("no monitor_status metrics (auth/endpoint wrong?)")
+	}
+	val := "up"
+	if down {
+		val = "down"
+	}
+	return p.writeIfChanged(ctx, "service_down", "poll:uptimekuma", val, now)
+}
+
+// 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) {
+	for _, line := range strings.Split(string(body), "\n") {
+		m := kumaLine.FindStringSubmatch(strings.TrimSpace(line))
+		if m == nil {
+			continue
+		}
+		seen = true
+		v, err := strconv.ParseFloat(m[2], 64)
+		if err != nil {
+			continue
+		}
+		if v == 0 {
+			down = true
+		}
+	}
+	return down, seen
+}
+
+// ---- helpers ---------------------------------------------------------------
+
+// writeIfChanged writes a `facts(kind=env)` row only when val differs from the
+// latest fact for (key, source). Values are stored JSON-encoded (the store's
+// convention: `"down"`, `"critical"`), matching how rules compare f.Value.
+func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error {
+	jv, _ := json.Marshal(val) // string never fails to marshal
+	prev, err := p.core.LatestFactBySource(ctx, key, source)
+	switch {
+	case err == nil && prev.Value == string(jv):
+		return nil // unchanged → no churn
+	case err != nil && err != ipc.ErrNoFact && !isNoFact(err):
+		return fmt.Errorf("read %s: %w", key, err)
+	}
+	_, err = p.core.WriteFact(ctx, ipc.WriteFactReq{
+		Ts:         now,
+		Kind:       "env",
+		Key:        key,
+		Value:      string(jv),
+		Source:     source,
+		Confidence: 1.0, // a direct reading, not an inference
+	})
+	if err != nil {
+		return fmt.Errorf("write %s: %w", key, err)
+	}
+	log.Printf("mavpoll: %s=%s (%s)", key, val, source)
+	return nil
+}
+
+// isNoFact — ErrNoFact rehydrated over the wire is wrapped (fmt.Errorf %w), so
+// errors.Is is the right check; keep a helper so the switch above reads clean.
+func isNoFact(err error) bool {
+	for e := err; e != nil; {
+		if e == ipc.ErrNoFact {
+			return true
+		}
+		u, ok := e.(interface{ Unwrap() error })
+		if !ok {
+			return false
+		}
+		e = u.Unwrap()
+	}
+	return false
+}
+
+func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
+	if err != nil {
+		return nil, err
+	}
+	if basicUser != "" {
+		req.SetBasicAuth(basicUser, "") // kuma: API key as username, empty password
+	}
+	resp, err := p.http.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+	body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("GET %s: %s", url, resp.Status)
+	}
+	return body, nil
+}
@@ -0,0 +1,64 @@
+package main
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+func TestMaxSeverity(t *testing.T) {
+	parse := func(s string) netdataAlarms {
+		var a netdataAlarms
+		if err := json.Unmarshal([]byte(s), &a); err != nil {
+			t.Fatal(err)
+		}
+		return a
+	}
+	cases := []struct{ body, want string }{
+		{`{"alarms":{}}`, "clear"},
+		{`{"alarms":{"a":{"status":"WARNING"}}}`, "warning"},
+		{`{"alarms":{"a":{"status":"WARNING"},"b":{"status":"CRITICAL"}}}`, "critical"},
+		{`{"alarms":{"a":{"status":"CLEAR"}}}`, "clear"},
+	}
+	for _, c := range cases {
+		if got := maxSeverity(parse(c.body)); got != c.want {
+			t.Errorf("maxSeverity(%s) = %q, want %q", c.body, got, c.want)
+		}
+	}
+}
+
+func TestKumaAnyDown(t *testing.T) {
+	cases := []struct {
+		body       string
+		down, seen bool
+	}{
+		{"", 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},
+	}
+	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)
+		}
+	}
+}
+
+func TestParseMaxHandshake(t *testing.T) {
+	cases := []struct {
+		out  string
+		want int64
+	}{
+		{"", 0},
+		{"pubkeyAAA\t0\n", 0}, // never handshaked
+		{"pubkeyAAA\t1700000000\n", 1700000000},
+		{"pubkeyAAA\t1700000000\npubkeyBBB\t1700000500\n", 1700000500}, // max wins
+		{"wg0\tpubkeyAAA\t1700000000\nwg0\tpubkeyBBB\t0\n", 1700000000}, // 'all' 3-field form
+	}
+	for _, c := range cases {
+		if got := parseMaxHandshake(c.out); got != c.want {
+			t.Errorf("parseMaxHandshake(%q) = %d, want %d", c.out, got, c.want)
+		}
+	}
+}