felhom-agent/internal/guestnet/watchdog.go

package guestnet

import (
	"context"
	"log/slog"
	"sort"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-agent/internal/proxmox"
)

// GuestSource yields the guests this agent OWNS. Production passes the pool-verified source
// (ListLXC ∩ pool membership, audit A1) — never a bare ListLXC, which under a broad token would let
// the watchdog run dhclient inside a co-tenant's container.
type GuestSource interface {
	Guests(ctx context.Context) ([]proxmox.Guest, error)
}

// Defaults. Every one of these is a damper: this watchdog runs a privileged command inside a
// customer's container, so it is designed to under-act.
const (
	DefaultInterval = 60 * time.Second
	// DefaultMinHealInterval is the per-guest cool-off between heals.
	DefaultMinHealInterval = 10 * time.Minute
	// DefaultMaxHealsPerHour caps a guest's heals; beyond it the watchdog only reports, because a
	// guest needing a fourth heal in an hour has a problem dhclient cannot fix.
	DefaultMaxHealsPerHour = 3
	// DefaultSettle is the boot-race guard, applied to BOTH the guest's uptime and the agent's own.
	// A guest that booted 40 s ago legitimately has no lease yet.
	DefaultSettle = 3 * time.Minute
	// requiredBadProbes: two CONSECUTIVE bad cycles before any heal. One blip is not a diagnosis.
	requiredBadProbes = 2
)

// Watchdog probes each owned, running guest's network every interval and heals a DHCP guest whose
// client has died. It never returns an error: a guest-network fault is a reported fact, not an agent
// failure.
type Watchdog struct {
	runner Runner
	guests GuestSource
	logger *slog.Logger

	interval        time.Duration
	minHealInterval time.Duration
	maxHealsPerHour int
	settle          time.Duration

	now       func() time.Time
	startedAt time.Time

	mu    sync.Mutex
	state map[int]*guestState
}

type guestState struct {
	badProbes  int
	heals      []time.Time // heal timestamps, pruned to the last hour
	lastHealAt time.Time
	lastState  State
	report     GuestReport
}

// New builds a Watchdog with the shipped dampers. interval <= 0 uses DefaultInterval.
func New(runner Runner, guests GuestSource, interval time.Duration, logger *slog.Logger) *Watchdog {
	if interval <= 0 {
		interval = DefaultInterval
	}
	if logger == nil {
		logger = slog.Default()
	}
	w := &Watchdog{
		runner:          runner,
		guests:          guests,
		logger:          logger,
		interval:        interval,
		minHealInterval: DefaultMinHealInterval,
		maxHealsPerHour: DefaultMaxHealsPerHour,
		settle:          DefaultSettle,
		now:             time.Now,
		state:           map[int]*guestState{},
	}
	w.startedAt = w.now()
	return w
}

// SetDampers overrides the three rate limits from config. Non-positive values keep the default,
// the same "0 = package default" convention the storage watchdog and wg loop use.
func (w *Watchdog) SetDampers(minHealInterval time.Duration, maxHealsPerHour int, settle time.Duration) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if minHealInterval > 0 {
		w.minHealInterval = minHealInterval
	}
	if maxHealsPerHour > 0 {
		w.maxHealsPerHour = maxHealsPerHour
	}
	if settle > 0 {
		w.settle = settle
	}
}

// Watch runs until ctx is cancelled. Started with `go wd.Watch(ctx)` — deliberately not part of the
// errc fan-out, because a guest-network watchdog must never be able to bring the agent down.
func (w *Watchdog) Watch(ctx context.Context) {
	w.logger.Info("guestnet: watchdog starting",
		"interval", w.interval, "min_heal_interval", w.minHealInterval,
		"max_heals_per_hour", w.maxHealsPerHour, "settle", w.settle)
	t := time.NewTicker(w.interval)
	defer t.Stop()
	w.Tick(ctx) // immediate baseline
	for {
		select {
		case <-ctx.Done():
			w.logger.Info("guestnet: watchdog shutting down", "reason", ctx.Err())
			return
		case <-t.C:
			w.Tick(ctx)
		}
	}
}

// Tick performs one full sweep. Exported so the wiring test and the live STOP leg can drive exactly
// one cycle instead of waiting on a ticker.
func (w *Watchdog) Tick(ctx context.Context) {
	guests, err := w.guests.Guests(ctx)
	if err != nil {
		// Unknown ownership ⇒ do nothing. Never fall back to an unfiltered guest list.
		w.logger.Warn("guestnet: guest list unavailable — skipping sweep (ownership unproven)", "err", err)
		return
	}
	for _, g := range guests {
		if ctx.Err() != nil {
			return
		}
		if g.Status != "running" {
			w.forget(g.VMID)
			continue
		}
		w.checkGuest(ctx, g)
	}
}

// forget drops state for a guest that is no longer running, so a stopped-and-restarted guest starts
// from a clean slate rather than inheriting a stale bad-probe count.
func (w *Watchdog) forget(vmid int) {
	w.mu.Lock()
	delete(w.state, vmid)
	w.mu.Unlock()
}

func (w *Watchdog) checkGuest(ctx context.Context, g proxmox.Guest) {
	now := w.now()
	p := w.probe(ctx, g.VMID)
	state, detail := classify(p)

	w.mu.Lock()
	st := w.state[g.VMID]
	if st == nil {
		st = &guestState{}
		w.state[g.VMID] = st
	}
	prev := st.lastState
	st.lastState = state
	st.pruneHeals(now)
	rep := GuestReport{
		VMID: g.VMID, Mode: string(p.Mode), IP: p.IP, HasRoute: p.HasRoute,
		DHClientAlive: p.DHCPAlive, State: string(state), Message: detail,
		HealsLastHour: len(st.heals), CheckedAt: now.UTC().Format(time.RFC3339),
	}
	if !st.lastHealAt.IsZero() {
		rep.LastHealAt = st.lastHealAt.UTC().Format(time.RFC3339)
	}

	switch state {
	case StateHealthy:
		st.badProbes = 0
		w.mu.Unlock()
		// The healthy path MUST be observable. v0.91.2's lesson, learned the hard way one day
		// earlier: if a healthy cycle logs nothing, "no alarms" and "never probed" are the same
		// line of evidence, and an inert watchdog is indistinguishable from a working one.
		w.logger.Debug("guestnet: guest network healthy", "vmid", g.VMID, "mode", string(p.Mode),
			"has_route", p.HasRoute, "dhclient_alive", p.DHCPAlive)
		if prev != "" && prev != StateHealthy {
			w.logger.Info("guestnet: guest network recovered", "vmid", g.VMID, "previous_state", string(prev))
		}
		w.record(g.VMID, rep)
		return

	case StateUnknown, StateStaticFault:
		st.badProbes = 0 // neither is a dhclient fault; don't accumulate toward a heal
		w.mu.Unlock()
		if prev != state { // loud once per transition, not once per minute
			w.logger.Warn("guestnet: guest network not actionable — reporting only",
				"vmid", g.VMID, "state", string(state), "mode", string(p.Mode),
				"has_ip", p.IP != "", "has_route", p.HasRoute, "detail", detail)
		}
		w.record(g.VMID, rep)
		return
	}

	// --- StateUnhealthy: a DHCP guest with something dhclient can fix ---------------------------

	st.badProbes++
	bad := st.badProbes
	lastHeal := st.lastHealAt
	healsInHour := len(st.heals)
	w.mu.Unlock()

	if reason, ok := w.observeOnly(g, now); !ok {
		rep.Damped = true
		rep.Message = detail + " — observing only: " + reason
		w.logger.Info("guestnet: guest network unhealthy but not acting", "vmid", g.VMID,
			"reason", reason, "detail", detail)
		w.record(g.VMID, rep)
		return
	}
	if bad < requiredBadProbes {
		rep.Message = detail + " — awaiting a second consecutive bad probe before healing"
		w.logger.Info("guestnet: guest network unhealthy (first bad probe — not acting yet)",
			"vmid", g.VMID, "detail", detail, "bad_probes", bad, "required", requiredBadProbes)
		w.record(g.VMID, rep)
		return
	}
	if damped, reason := w.damped(lastHeal, healsInHour, now); damped {
		rep.Damped = true
		rep.Message = detail + " — heal damped: " + reason
		w.logger.Warn("guestnet: guest network unhealthy but healing is DAMPED — reporting only",
			"vmid", g.VMID, "reason", reason, "heals_last_hour", healsInHour, "detail", detail)
		w.record(g.VMID, rep)
		return
	}

	// --- heal ----------------------------------------------------------------------------------
	w.logger.Warn("guestnet: guest network unhealthy — healing",
		"vmid", g.VMID, "detail", detail, "bad_probes", bad)
	healed, healErr := w.heal(ctx, g.VMID)

	w.mu.Lock()
	st = w.state[g.VMID]
	if st != nil {
		st.heals = append(st.heals, now)
		st.lastHealAt = now
		st.badProbes = 0 // the post-heal probe below is the new evidence
		healsInHour = len(st.heals)
	}
	w.mu.Unlock()

	after := w.probe(ctx, g.VMID)
	afterState, afterDetail := classify(after)

	rep = GuestReport{
		VMID: g.VMID, Mode: string(after.Mode), IP: after.IP, HasRoute: after.HasRoute,
		DHClientAlive: after.DHCPAlive, State: string(afterState), Message: afterDetail,
		Healed: true, HealSucceeded: afterState == StateHealthy,
		LastHealAt: now.UTC().Format(time.RFC3339), HealsLastHour: healsInHour,
		CheckedAt: w.now().UTC().Format(time.RFC3339),
	}
	if healErr != nil {
		rep.Message = "heal command failed: " + healErr.Error() + "; " + afterDetail
	}

	if afterState == StateHealthy {
		w.logger.Info("guestnet: guest network healed", "vmid", g.VMID, "ip", after.IP,
			"has_route", after.HasRoute, "dhclient_alive", after.DHCPAlive, "heals_last_hour", healsInHour)
	} else {
		w.logger.Error("guestnet: heal did not restore the guest network", "vmid", g.VMID,
			"state", string(afterState), "detail", afterDetail, "heal_ran", healed, "err", healErr)
	}

	w.mu.Lock()
	if st = w.state[g.VMID]; st != nil {
		st.lastState = afterState
	}
	w.mu.Unlock()
	w.record(g.VMID, rep)
}

// heal runs the incident's restored invocation, VERBATIM (INCIDENT-guest-dhclient-killed-2026-07-20
// §5) — the same argv that brought guest 9201 back at 10:04:3x UTC. Fixed shape, no shell, no guest
// data interpolated. Logged at INFO before it runs so the operator sees the exact command.
func (w *Watchdog) heal(ctx context.Context, vmid int) (bool, error) {
	args := []string{"exec", itoa(vmid), "--", "dhclient",
		"-pf", "/run/dhclient." + eth0 + ".pid",
		"-lf", "/var/lib/dhcp/dhclient." + eth0 + ".leases", eth0}
	w.logger.Info("guestnet: running heal command", "vmid", vmid, "cmd", "pct "+joinArgs(args))
	_, errOut, err := w.runner.Run(ctx, "pct", args...)
	if err != nil {
		w.logger.Error("guestnet: heal command failed", "vmid", vmid, "stderr", firstLine(string(errOut)), "err", err)
		return false, err
	}
	return true, nil
}

// observeOnly reports whether a boot race means this cycle must look and not touch.
func (w *Watchdog) observeOnly(g proxmox.Guest, now time.Time) (string, bool) {
	if now.Sub(w.startedAt) < w.settle {
		return "agent started less than " + w.settle.String() + " ago", false
	}
	if g.Uptime > 0 && time.Duration(g.Uptime)*time.Second < w.settle {
		return "guest has been up for less than " + w.settle.String(), false
	}
	return "", true
}

// damped applies the two rate limits.
func (w *Watchdog) damped(lastHeal time.Time, healsInHour int, now time.Time) (bool, string) {
	if !lastHeal.IsZero() && now.Sub(lastHeal) < w.minHealInterval {
		return true, "last heal was less than " + w.minHealInterval.String() + " ago"
	}
	if healsInHour >= w.maxHealsPerHour {
		return true, "heal budget for the hour is spent (a guest needing more than this has a fault dhclient cannot fix)"
	}
	return false, ""
}

func (s *guestState) pruneHeals(now time.Time) {
	kept := s.heals[:0]
	for _, t := range s.heals {
		if now.Sub(t) < time.Hour {
			kept = append(kept, t)
		}
	}
	s.heals = kept
}

func (w *Watchdog) record(vmid int, rep GuestReport) {
	w.mu.Lock()
	defer w.mu.Unlock()
	if st := w.state[vmid]; st != nil {
		st.report = rep
	}
}

func joinArgs(args []string) string {
	out := ""
	for i, a := range args {
		if i > 0 {
			out += " "
		}
		out += a
	}
	return out
}

// --- the hub report block ------------------------------------------------------------------------

// GuestReport is one guest's last observed network state, mirrored to the hub.
type GuestReport struct {
	VMID          int    `json:"vmid"`
	State         string `json:"state"` // healthy | unhealthy | static_fault | unknown
	Mode          string `json:"mode"`  // dhcp | static | unknown
	IP            string `json:"ip,omitempty"`
	HasRoute      bool   `json:"has_route"`
	DHClientAlive bool   `json:"dhclient_alive"`
	CheckedAt     string `json:"checked_at,omitempty"`
	Healed        bool   `json:"healed,omitempty"`         // a heal ran on THIS cycle
	HealSucceeded bool   `json:"heal_succeeded,omitempty"` // and the re-probe came back healthy
	LastHealAt    string `json:"last_heal_at,omitempty"`
	HealsLastHour int    `json:"heals_last_hour,omitempty"`
	Damped        bool   `json:"damped,omitempty"`
	Message       string `json:"message,omitempty"`
}

// Snapshot returns the per-guest blocks for the heartbeat, VMID-sorted for a stable wire shape.
func (w *Watchdog) Snapshot() []GuestReport {
	w.mu.Lock()
	defer w.mu.Unlock()
	out := make([]GuestReport, 0, len(w.state))
	for _, st := range w.state {
		if st.report.VMID != 0 {
			out = append(out, st.report)
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].VMID < out[j].VMID })
	return out
}