256 lines
9.6 KiB
Go
256 lines
9.6 KiB
Go
package monitor
|
||
|
||
import (
|
||
"log"
|
||
"sync"
|
||
"time"
|
||
|
||
"gitea.dooplex.hu/admin/felhom-hub/internal/store"
|
||
)
|
||
|
||
// HostStalenessChecker is the host-domain dead-man's-switch (v0.7.0, slice 3). It
|
||
// is a deliberate SIBLING of StalenessChecker, not a rename: during slices 3–9 the
|
||
// controller report stream (reports) and the agent host-report stream
|
||
// (host_reports) are both live, so both checkers run. It keys on host↔host_reports
|
||
// and emits host_stale / host_down / host_recovered. Merging is a slice-10 job.
|
||
//
|
||
// Events are attributed to the host's CUSTOMER (SaveEvent + onEvent take the
|
||
// customer_id) so the existing per-customer notification/event UX picks them up
|
||
// unchanged.
|
||
type HostStalenessChecker struct {
|
||
store *store.Store
|
||
threshold time.Duration // "stale" after this (default 30m — same as the controller checker)
|
||
downAfter time.Duration // "down" after this (2x threshold)
|
||
logger *log.Logger
|
||
onEvent EventNotifyFunc
|
||
|
||
mu sync.Mutex
|
||
states map[string]string // hostID → "ok" | "stale" | "down"
|
||
customerOf map[string]string // hostID → customerID (for event attribution)
|
||
downtimeStart map[string]time.Time // hostID → when it first became unreachable
|
||
}
|
||
|
||
// NewHostStalenessChecker creates the checker and seeds state from current
|
||
// host-report recency. No events are generated during initialization.
|
||
func NewHostStalenessChecker(s *store.Store, threshold time.Duration, onEvent EventNotifyFunc, logger *log.Logger) *HostStalenessChecker {
|
||
sc := &HostStalenessChecker{
|
||
store: s,
|
||
threshold: threshold,
|
||
downAfter: 2 * threshold,
|
||
logger: logger,
|
||
onEvent: onEvent,
|
||
states: make(map[string]string),
|
||
customerOf: make(map[string]string),
|
||
downtimeStart: make(map[string]time.Time),
|
||
}
|
||
|
||
rows, err := s.GetHostStaleness()
|
||
if err != nil {
|
||
logger.Printf("[WARN] Host staleness checker: failed to seed states: %v", err)
|
||
return sc
|
||
}
|
||
var okCount, staleCount, downCount, knownCount int
|
||
for _, row := range rows {
|
||
if s.IsCustomerBlocked(row.CustomerID) {
|
||
continue
|
||
}
|
||
sc.customerOf[row.HostID] = row.CustomerID
|
||
age := time.Since(row.LastReportAt)
|
||
// F2: a host already stale/down at (re)start is left UNSEEDED so the first Check() observes oldState==""
|
||
// and emits once (a box that went silent while the hub was down would otherwise stay silent).
|
||
//
|
||
// R-1001 (hub 0.162.0): UNLESS the alarm was already raised for this silence — the newest host_stale /
|
||
// host_down event for the host is newer than its last report. Then the host is seeded with the state that
|
||
// event recorded, silently. The old comment said „the dispatcher's 1h cooldown dedups the re-emit"; it did
|
||
// not (host liveness events bypass that cooldown, `dispatcher.go` nodeLivenessEvents), so every hub restart
|
||
// re-mailed the operator about Tester 2, silent for days (measured 2026-10-13). A host seeded „stale" still
|
||
// emits host_down when it crosses downAfter — a new transition. A failed lookup leaves it unseeded (the
|
||
// alarm, never the silence). Pinned by TestHostStaleness_R1001_RestartDoesNotReMail.
|
||
if age > sc.threshold {
|
||
if known := sc.alreadyAlarmed(row); known != "" {
|
||
sc.states[row.HostID] = known
|
||
sc.downtimeStart[row.HostID] = row.LastReportAt
|
||
knownCount++
|
||
continue
|
||
}
|
||
}
|
||
switch {
|
||
case age > sc.downAfter:
|
||
downCount++
|
||
case age > sc.threshold:
|
||
staleCount++
|
||
default:
|
||
sc.states[row.HostID] = "ok"
|
||
okCount++
|
||
}
|
||
}
|
||
logger.Printf("[INFO] Host staleness checker initialized: %d ok, %d already alarmed (seeded silently, R-1001), %d stale, %d down (left unseeded → first Check emits; host_stale after %s, host_down after %s)", okCount, knownCount, staleCount, downCount, threshold, 2*threshold)
|
||
return sc
|
||
}
|
||
|
||
// alreadyAlarmed returns "stale" or "down" when the newest liveness event for the host is that alarm and is not older
|
||
// than the host's last report (the operator was told about THIS silence), else "".
|
||
func (sc *HostStalenessChecker) alreadyAlarmed(row store.HostStaleRow) string {
|
||
ev, err := sc.store.LatestHostLivenessEvent(row.CustomerID, row.HostID)
|
||
if err != nil {
|
||
sc.logger.Printf("[WARN] Host staleness: could not read the last alarm for %s (%v) — the first Check alarms (R-1001)", row.HostID, err)
|
||
return ""
|
||
}
|
||
if ev == nil || ev.CreatedAt.Before(row.LastReportAt) {
|
||
return ""
|
||
}
|
||
switch ev.EventType {
|
||
case "host_down":
|
||
return "down"
|
||
case "host_stale":
|
||
return "stale"
|
||
}
|
||
return ""
|
||
}
|
||
|
||
// Check evaluates all hosts and emits events on state transitions. Call every 60s.
|
||
func (sc *HostStalenessChecker) Check() {
|
||
rows, err := sc.store.GetHostStaleness()
|
||
if err != nil {
|
||
sc.logger.Printf("[WARN] Host staleness check failed: %v", err)
|
||
return
|
||
}
|
||
|
||
sc.mu.Lock()
|
||
defer sc.mu.Unlock()
|
||
|
||
seen := make(map[string]bool, len(rows))
|
||
for _, row := range rows {
|
||
seen[row.HostID] = true
|
||
if sc.store.IsCustomerBlocked(row.CustomerID) {
|
||
delete(sc.states, row.HostID)
|
||
continue
|
||
}
|
||
sc.customerOf[row.HostID] = row.CustomerID
|
||
|
||
age := time.Since(row.LastReportAt)
|
||
var newState string
|
||
switch {
|
||
case age > sc.threshold && store.DeliberateOffHeld(row.DeliberateOffAt, time.Now()):
|
||
// R-998: the household switched the box off on purpose — silent, for the 14-day hold.
|
||
newState = "off"
|
||
case age > sc.downAfter:
|
||
newState = "down"
|
||
case age > sc.threshold:
|
||
newState = "stale"
|
||
default:
|
||
newState = "ok"
|
||
}
|
||
|
||
oldState := sc.states[row.HostID]
|
||
if newState == "off" || oldState == "off" {
|
||
// Into the hold, or out of it back to ok: no event (the report ingest records the info events).
|
||
// Out of it into stale/down (the hold ran out with the box still silent): the usual alarm, with a
|
||
// message that says why it comes now.
|
||
sc.states[row.HostID] = newState
|
||
if newState == "stale" || newState == "down" {
|
||
sc.downtimeStart[row.HostID] = time.Now()
|
||
sc.emitHoldEnded(row, newState, age)
|
||
} else if newState == "off" && oldState != "off" {
|
||
sc.logger.Printf("[INFO] Host staleness: %s deliberately off since %s — alarms held until %s",
|
||
row.HostID, row.DeliberateOffAt.Format(time.RFC3339), row.DeliberateOffAt.Add(store.DeliberateOffHold).Format(time.RFC3339))
|
||
}
|
||
continue
|
||
}
|
||
if oldState == "" {
|
||
// F2: first observation (incl. a hub restart). Healthy seeds silently; a host already
|
||
// stale/down at first sight emits once (else a box that went silent while the hub was
|
||
// down would never alert). A host whose alarm was already raised was seeded at start (R-1001).
|
||
sc.states[row.HostID] = newState
|
||
if newState != "ok" {
|
||
sc.emitTransition(row.HostID, row.CustomerID, "unknown", newState, age)
|
||
}
|
||
continue
|
||
}
|
||
if oldState == newState {
|
||
continue
|
||
}
|
||
|
||
sc.states[row.HostID] = newState
|
||
if newState == "stale" && oldState == "ok" {
|
||
sc.downtimeStart[row.HostID] = time.Now()
|
||
}
|
||
downtimeDur := age
|
||
if newState == "ok" {
|
||
if t, ok := sc.downtimeStart[row.HostID]; ok {
|
||
downtimeDur = time.Since(t)
|
||
}
|
||
delete(sc.downtimeStart, row.HostID)
|
||
}
|
||
sc.emitTransition(row.HostID, row.CustomerID, oldState, newState, downtimeDur)
|
||
}
|
||
|
||
for id := range sc.states {
|
||
if !seen[id] {
|
||
delete(sc.states, id)
|
||
delete(sc.downtimeStart, id)
|
||
}
|
||
}
|
||
}
|
||
|
||
// GetState returns the current staleness state for a host.
|
||
func (sc *HostStalenessChecker) GetState(hostID string) string {
|
||
sc.mu.Lock()
|
||
defer sc.mu.Unlock()
|
||
s := sc.states[hostID]
|
||
if s == "" {
|
||
return "unknown"
|
||
}
|
||
return s
|
||
}
|
||
|
||
func (sc *HostStalenessChecker) emitTransition(hostID, customerID, oldState, newState string, age time.Duration) {
|
||
var eventType, severity, message string
|
||
switch {
|
||
case newState == "stale":
|
||
eventType = "host_stale"
|
||
severity = "warning"
|
||
message = "Host " + hostID + ": no report for " + formatDuration(age)
|
||
case newState == "down":
|
||
eventType = "host_down"
|
||
severity = "error"
|
||
message = "Host " + hostID + ": no report for " + formatDuration(age)
|
||
case newState == "ok" && (oldState == "stale" || oldState == "down"):
|
||
eventType = "host_recovered"
|
||
severity = "info"
|
||
message = "Host " + hostID + ": reports resumed (was " + oldState + " for " + formatDuration(age) + ")"
|
||
default:
|
||
return
|
||
}
|
||
|
||
sc.logger.Printf("[INFO] Host staleness: %s %s → %s (%s)", hostID, oldState, newState, eventType)
|
||
|
||
if _, err := sc.store.SaveEvent(customerID, eventType, severity, message, "{}", "hub"); err != nil {
|
||
sc.logger.Printf("[WARN] Failed to save host staleness event for %s: %v", hostID, err)
|
||
return
|
||
}
|
||
if sc.onEvent != nil {
|
||
sc.onEvent(customerID, eventType, severity, message, "{}", "hub")
|
||
}
|
||
}
|
||
|
||
// emitHoldEnded raises the alarm a deliberate-off hold was holding (R-998): the box was switched off by the household
|
||
// and is still silent when the hold runs out. The text names the power-off — „no report for 14d" alone reads as an
|
||
// outage nobody saw coming.
|
||
func (sc *HostStalenessChecker) emitHoldEnded(row store.HostStaleRow, newState string, age time.Duration) {
|
||
eventType, severity := "host_stale", "warning"
|
||
if newState == "down" {
|
||
eventType, severity = "host_down", "error"
|
||
}
|
||
message := "Host " + row.HostID + ": switched off by the household on " + row.DeliberateOffAt.UTC().Format("2006-01-02 15:04 UTC") +
|
||
" and still silent after the " + formatDuration(store.DeliberateOffHold) + " hold (no report for " + formatDuration(age) + ")"
|
||
sc.logger.Printf("[INFO] Host staleness: %s off → %s (%s, the deliberate-off hold ended)", row.HostID, newState, eventType)
|
||
if _, err := sc.store.SaveEvent(row.CustomerID, eventType, severity, message, "{}", "hub"); err != nil {
|
||
sc.logger.Printf("[WARN] Failed to save host staleness event for %s: %v", row.HostID, err)
|
||
return
|
||
}
|
||
if sc.onEvent != nil {
|
||
sc.onEvent(row.CustomerID, eventType, severity, message, "{}", "hub")
|
||
}
|
||
}
|