// Package channelhealth periodically proves the controller→agent local-API channel and surfaces a // classified operator alert + dashboard entry on a state change. It closes the gap the R1 // pin-mismatch incident exposed: the channel was only probed once at startup (probeLocalAPI) and only // logged, so a mid-life agent re-key went unnoticed until a user hit the dead disk UI. // // Design is spike-driven (SPIKE-controller-agent-channel-health-2026-06-29, GO): // - Probe via the PRODUCTION memoized agent client (reused, not a fresh one): it self-heals after an // agent restart, reflects exactly what the UI sees, and avoids the per-call transport leak the // singleton fixed. The construction error (bad fingerprint → agentClient() fails, latching via // sync.Once) is surfaced distinctly. // - Classify the failure by the error substring (the spike Q1 map). // - Debounce transient reasons (connection-refused / timeout): a clean agent restart shows a ~1s // socket gap (one failed probe) that must NOT page — require N>=2 consecutive before alerting. // Pin-mismatch / 401 / construction / DNS alert on first observation (they don't self-clear). // - First scheduler observation SEEDS state without notifying (mirrors host_staleness/host_capability). package channelhealth import ( "context" "log" "strings" "sync" "time" ) // debounceThreshold is the number of consecutive down-probes a TRANSIENT reason needs before it is // treated as a real outage (suppresses the ~1s agent-restart blip from the spike's Q2). const debounceThreshold = 2 // probeTimeout mirrors probeLocalAPI's 15s budget for the GET /storage call. const probeTimeout = 15 * time.Second // Reason is the classified channel-down cause. type Reason string const ( ReasonPinMismatch Reason = "pin_mismatch" ReasonUnauthorized Reason = "unauthorized" ReasonUnreachable Reason = "unreachable" ReasonTimeout Reason = "timeout" ReasonMisconfigured Reason = "misconfigured" // endpoint unresolvable (DNS) ReasonConstruction Reason = "construction_error" // agentClient() build error — LATCHES until restart ReasonUnknown Reason = "unknown" ) // Probe runs one channel check via the production memoized client. It returns whether the failure was // a CONSTRUCTION error (agentClient() couldn't even build — a latching config fault) and the error // (nil = channel up). Production wires this to Server.ProbeAgentChannel; tests inject a fake. type Probe func(ctx context.Context) (constructionErr bool, err error) // Sink receives the checker's outputs. SetDashboard reflects the CURRENT state every check (idempotent // — a born-down channel shows immediately, with no notification). NotifyDown/NotifyRecovered fire ONLY // on a transition (the operator relay, with the hub-side cooldown). Implemented in main.go over the // notify.Notifier + web.AlertManager (kept out of this package to avoid an import cycle). type Sink interface { SetDashboard(down bool, reason Reason, hungarianMsg string) NotifyDown(reason Reason, eventType, severity, englishMsg string) NotifyRecovered() } type classification struct { reason Reason eventType string severity string // "error" (critical) | "warning" (transient) english string // operator alert message (relayed to the hub) hungarian string // short dashboard line debounce bool // transient → require N>=2 consecutive } // classify maps a probe result to a classification. constructionErr is authoritative for the latching // config fault; otherwise the error substring decides (spike Q1 map). Match substrings, not exact // strings (the wrapped Get "..." prefix varies). func classify(constructionErr bool, err error) classification { if constructionErr { return classification{ReasonConstruction, "agent_channel_construction_error", "error", "Controller→agent channel down: agent client misconfigured — local_api.fingerprint is not a valid SHA-256, the controller cannot build the client (config fix + controller restart needed).", "A tárolókezelő ügynök kapcsolata hibásan beállítva.", false} } s := err.Error() switch { case strings.Contains(s, "TLS pin mismatch"): return classification{ReasonPinMismatch, "agent_channel_pin_mismatch", "error", "Controller→agent channel down: TLS pin mismatch — the agent's leaf cert no longer matches the controller's bootstrap fingerprint (re-pin / re-bootstrap).", "A tárolókezelő ügynök tanúsítványa megváltozott.", false} case strings.Contains(s, "HTTP 401"): return classification{ReasonUnauthorized, "agent_channel_unauthorized", "error", "Controller→agent channel down: agent rejected the controller token (HTTP 401) — token stale/rotated (re-bootstrap).", "A tárolókezelő ügynök elutasította a hozzáférést.", false} case strings.Contains(s, "no such host") || strings.Contains(s, "lookup "): return classification{ReasonMisconfigured, "agent_channel_misconfigured", "error", "Controller→agent channel down: agent endpoint unresolvable — local_api.endpoint misconfigured.", "A tárolókezelő ügynök címe nem feloldható.", false} case strings.Contains(s, "connection refused"): return classification{ReasonUnreachable, "agent_channel_unreachable", "warning", "Controller→agent channel down: agent unreachable (connection refused) — felhom-agent down or :8443 closed.", "A tárolókezelő ügynök nem elérhető.", true} case strings.Contains(s, "context deadline exceeded") || strings.Contains(s, "i/o timeout") || strings.Contains(s, "Client.Timeout"): return classification{ReasonTimeout, "agent_channel_timeout", "warning", "Controller→agent channel down: timeout — host unreachable / packets dropped.", "A tárolókezelő ügynök nem válaszol.", true} default: return classification{ReasonUnknown, "agent_channel_unknown", "warning", "Controller→agent channel down: " + s, "A tárolókezelő ügynök nem elérhető.", true} } } // Checker holds the in-memory channel state. No persistence — the state is re-derived each run // (mirrors the AlertManager's state-based model). Safe for the single scheduler caller; the mutex // guards against an overlapping run. type Checker struct { probe Probe sink Sink logger *log.Logger mu sync.Mutex state string // "" (unseeded) | "up" | "down:" consecutiveDown int alerted bool // have we emitted a down alert for the CURRENT down-spell? (F2: drives // alerting instead of `prev==""`, so a BORN-down — broken at startup/reseed — alerts too, not // just a live up→down transition; re-armed on recovery / reason-change.) } // New builds a checker over the probe + sink seams. func New(probe Probe, sink Sink, logger *log.Logger) *Checker { return &Checker{probe: probe, sink: sink, logger: logger} } // Check runs one probe cycle: classify, debounce, reflect on the dashboard, and notify on a real // transition. Best-effort + idempotent — safe to call on a timer. Never returns an error to the // scheduler (a probe failure IS the signal, not a job failure). func (c *Checker) Check(ctx context.Context) error { pctx, cancel := context.WithTimeout(ctx, probeTimeout) constructionErr, perr := c.probe(pctx) cancel() c.mu.Lock() defer c.mu.Unlock() // Channel UP. if perr == nil { c.consecutiveDown = 0 c.sink.SetDashboard(false, "", "") prev := c.state c.state = "up" c.alerted = false // re-arm for the next down-spell if prev != "" && prev != "up" { c.logger.Printf("[INFO] [channel] agent channel recovered (was %s)", prev) c.sink.NotifyRecovered() } return nil // healthy first-obs / steady-up → no notify } // Channel DOWN — classify + debounce transient reasons. cls := classify(constructionErr, perr) c.consecutiveDown++ if cls.debounce && c.consecutiveDown < debounceThreshold { // A transient blip (e.g. the ~1s agent-restart socket gap, or the agent not yet up on a cold // boot). Hold the previous state — do NOT flip the dashboard or notify. Unseeded → assume up // until confirmed (so a transient born-down still needs N>=2 before it alerts). if c.state == "" { c.state = "up" } c.logger.Printf("[DEBUG] [channel] transient down (%s, %d/%d) — suppressed pending confirmation: %v", cls.reason, c.consecutiveDown, debounceThreshold, perr) return nil } // Confirmed down. F2: a NEW down-spell — coming from up/unseeded OR a reason change — re-arms the // alert, so a BORN-down (broken at startup/reseed) alerts on cycle 1 for non-transient reasons, // not only a live up->down transition. A steady down that already alerted does not re-fire. newState := "down:" + string(cls.reason) c.sink.SetDashboard(true, cls.reason, cls.hungarian) // dashboard reflects current state always prev := c.state if prev == "" || prev == "up" || prev != newState { c.alerted = false } c.state = newState if c.alerted { return nil // steady down, same reason, already alerted → no duplicate (dashboard stays set) } c.logger.Printf("[WARN] [channel] agent channel DOWN (%s->%s): %v", orUnseeded(prev), newState, perr) c.sink.NotifyDown(cls.reason, cls.eventType, cls.severity, cls.english) c.alerted = true return nil } func orUnseeded(s string) string { if s == "" { return "unseeded" } return s } // State returns the current channel state (for tests/diagnostics). func (c *Checker) State() string { c.mu.Lock() defer c.mu.Unlock() if c.state == "" { return "unseeded" } return c.state }