Maven/cmd/mavpoll/main.go

// 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
}