hub: HostCapabilityChecker — operator alert on agent capability-degraded (v0.19.0)
Companion to felhom-agent v0.44.0. New monitor.HostCapabilityChecker (sibling of HostStalenessChecker) reads the capabilities snapshot from the latest host report and emits agent_capability_degraded/recovered (operator-only, 1h cooldown) on ok<->degraded transitions for any Critical capability. store.GetHostCapabilities (MAX(id), no migration). Goldens mirror the new capabilities field. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01EPZ4GJ8L5Jqf8UiPwbn1kt
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
package monitor
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"log"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-hub/internal/store"
|
||||
)
|
||||
|
||||
// HostCapabilityChecker is the operator-facing alert for the agent's privileged-capability
|
||||
// self-check (felhom-agent v0.44.0). It is a deliberate SIBLING of HostStalenessChecker: same
|
||||
// state-transition shape (per-host ok/degraded map, seed-without-event, emit only on change), same
|
||||
// dispatcher/event plumbing — but it keys on the capability snapshot the agent rides in its host
|
||||
// report (whether each required `sudo -n` grant is usable) rather than report recency.
|
||||
//
|
||||
// A host is "degraded" iff its latest report has any CRITICAL capability with status "degraded"
|
||||
// (a missing user-facing grant — the multi-drive-flapping class). Non-critical degradations ride
|
||||
// the report + the agent's own logs but do NOT alert the operator (avoid noise). The event is
|
||||
// attributed to the host's customer, so the existing operator notification UX picks it up
|
||||
// unchanged. Customer is NEVER notified — internal capability health is operator-only.
|
||||
type HostCapabilityChecker struct {
|
||||
store *store.Store
|
||||
logger *log.Logger
|
||||
onEvent EventNotifyFunc
|
||||
|
||||
mu sync.Mutex
|
||||
states map[string]string // hostID → "ok" | "degraded"
|
||||
customerOf map[string]string // hostID → customerID
|
||||
}
|
||||
|
||||
// NewHostCapabilityChecker creates the checker and seeds state from the current snapshots. No
|
||||
// events are generated during initialization (mirrors the staleness checker).
|
||||
func NewHostCapabilityChecker(s *store.Store, onEvent EventNotifyFunc, logger *log.Logger) *HostCapabilityChecker {
|
||||
cc := &HostCapabilityChecker{
|
||||
store: s,
|
||||
logger: logger,
|
||||
onEvent: onEvent,
|
||||
states: make(map[string]string),
|
||||
customerOf: make(map[string]string),
|
||||
}
|
||||
rows, err := s.GetHostCapabilities()
|
||||
if err != nil {
|
||||
logger.Printf("[WARN] Host capability checker: failed to seed states: %v", err)
|
||||
return cc
|
||||
}
|
||||
var okCount, degCount int
|
||||
for _, row := range rows {
|
||||
if s.IsCustomerBlocked(row.CustomerID) {
|
||||
continue
|
||||
}
|
||||
cc.customerOf[row.HostID] = row.CustomerID
|
||||
st, _, _ := capabilityState(row.Capabilities)
|
||||
cc.states[row.HostID] = st
|
||||
if st == "degraded" {
|
||||
degCount++
|
||||
} else {
|
||||
okCount++
|
||||
}
|
||||
}
|
||||
logger.Printf("[INFO] Host capability checker initialized: %d ok, %d degraded", okCount, degCount)
|
||||
return cc
|
||||
}
|
||||
|
||||
// Check evaluates all hosts and emits an operator event on each ok↔degraded transition. Call on the
|
||||
// same sweep as the staleness checker (every 60s).
|
||||
func (cc *HostCapabilityChecker) Check() {
|
||||
rows, err := cc.store.GetHostCapabilities()
|
||||
if err != nil {
|
||||
cc.logger.Printf("[WARN] Host capability check failed: %v", err)
|
||||
return
|
||||
}
|
||||
cc.mu.Lock()
|
||||
defer cc.mu.Unlock()
|
||||
|
||||
seen := make(map[string]bool, len(rows))
|
||||
for _, row := range rows {
|
||||
seen[row.HostID] = true
|
||||
if cc.store.IsCustomerBlocked(row.CustomerID) {
|
||||
delete(cc.states, row.HostID)
|
||||
continue
|
||||
}
|
||||
cc.customerOf[row.HostID] = row.CustomerID
|
||||
|
||||
newState, names, features := capabilityState(row.Capabilities)
|
||||
oldState := cc.states[row.HostID]
|
||||
if oldState == "" {
|
||||
cc.states[row.HostID] = newState // first observation — no event
|
||||
continue
|
||||
}
|
||||
if oldState == newState {
|
||||
continue
|
||||
}
|
||||
cc.states[row.HostID] = newState
|
||||
cc.emitTransition(row.HostID, row.CustomerID, oldState, newState, names, features)
|
||||
}
|
||||
|
||||
for id := range cc.states {
|
||||
if !seen[id] {
|
||||
delete(cc.states, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// GetState returns the current capability state for a host ("unknown" if unseen).
|
||||
func (cc *HostCapabilityChecker) GetState(hostID string) string {
|
||||
cc.mu.Lock()
|
||||
defer cc.mu.Unlock()
|
||||
s := cc.states[hostID]
|
||||
if s == "" {
|
||||
return "unknown"
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// capabilityState reduces a snapshot to "ok"/"degraded" plus the degraded CRITICAL capability names
|
||||
// + features (for the event detail). An empty/absent snapshot (pre-v0.44.0 agent) is "ok" — an old
|
||||
// agent can't trip a false alert.
|
||||
func capabilityState(caps []store.CapabilityStatus) (state string, names, features []string) {
|
||||
for _, c := range caps {
|
||||
if c.Critical && c.Status == "degraded" {
|
||||
names = append(names, c.Name)
|
||||
features = append(features, c.Feature)
|
||||
}
|
||||
}
|
||||
sort.Strings(names)
|
||||
if len(names) > 0 {
|
||||
return "degraded", names, features
|
||||
}
|
||||
return "ok", nil, nil
|
||||
}
|
||||
|
||||
func (cc *HostCapabilityChecker) emitTransition(hostID, customerID, oldState, newState string, names, features []string) {
|
||||
var eventType, severity, message string
|
||||
switch {
|
||||
case newState == "degraded":
|
||||
eventType = "agent_capability_degraded"
|
||||
severity = "warning"
|
||||
message = "Host " + hostID + ": agent privileged capability degraded — " +
|
||||
strings.Join(names, ", ") + " (impairs: " + strings.Join(dedup(features), "; ") + ")"
|
||||
case newState == "ok" && oldState == "degraded":
|
||||
eventType = "agent_capability_recovered"
|
||||
severity = "info"
|
||||
message = "Host " + hostID + ": agent privileged capabilities recovered (all required grants restored)"
|
||||
default:
|
||||
return
|
||||
}
|
||||
|
||||
details, _ := json.Marshal(map[string]any{
|
||||
"host_id": hostID,
|
||||
"degraded_capabilities": names,
|
||||
"gated_features": dedup(features),
|
||||
})
|
||||
|
||||
cc.logger.Printf("[INFO] Host capability: %s %s → %s (%s)", hostID, oldState, newState, eventType)
|
||||
|
||||
if _, err := cc.store.SaveEvent(customerID, eventType, severity, message, string(details), "hub"); err != nil {
|
||||
cc.logger.Printf("[WARN] Failed to save host capability event for %s: %v", hostID, err)
|
||||
return
|
||||
}
|
||||
if cc.onEvent != nil {
|
||||
cc.onEvent(customerID, eventType, severity, message, string(details), "hub")
|
||||
}
|
||||
}
|
||||
|
||||
func dedup(in []string) []string {
|
||||
seen := make(map[string]bool, len(in))
|
||||
var out []string
|
||||
for _, s := range in {
|
||||
if !seen[s] {
|
||||
seen[s] = true
|
||||
out = append(out, s)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,107 @@
|
||||
package monitor
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"gitea.dooplex.hu/admin/felhom-hub/internal/store"
|
||||
_ "modernc.org/sqlite"
|
||||
)
|
||||
|
||||
// reportWith builds a host-report body carrying the given capability snapshot (the only field the
|
||||
// capability checker reads). One critical-degraded entry makes the host "degraded".
|
||||
func reportWith(caps string) []byte {
|
||||
return []byte(`{"host_id":"h1","capabilities":` + caps + `}`)
|
||||
}
|
||||
|
||||
const capAllOK = `[{"name":"guest-init-pid","feature":"drive-gate","critical":true,"status":"ok"}]`
|
||||
const capCritDegraded = `[{"name":"guest-init-pid","feature":"drive-gate guest-sees","critical":true,"status":"degraded","reason":"sudo policy denied"}]`
|
||||
const capNonCritDegraded = `[{"name":"disk-smart","feature":"SMART","critical":false,"status":"degraded","reason":"binary not found"}]`
|
||||
|
||||
func newCapStore(t *testing.T) *store.Store {
|
||||
t.Helper()
|
||||
st, err := store.New(filepath.Join(t.TempDir(), "test.db"), log.New(io.Discard, "", 0))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { st.Close() })
|
||||
st.SaveCustomerConfig(&store.CustomerConfig{CustomerID: "c1", APIKey: "ck", RetrievalPassword: "p"})
|
||||
st.UpsertHost(&store.Host{HostID: "h1", CustomerID: "c1", APIKey: "k1"})
|
||||
return st
|
||||
}
|
||||
|
||||
// TestHostCapabilityChecker covers the full transition contract (§10): seed (no event), ok→degraded
|
||||
// (one event), steady degraded (none), degraded→ok (recovered), and that a non-critical degradation
|
||||
// never trips the host state.
|
||||
func TestHostCapabilityChecker(t *testing.T) {
|
||||
st := newCapStore(t)
|
||||
st.SaveHostReport("h1", "c1", reportWith(capAllOK), store.HostReportDenorm{})
|
||||
|
||||
var events []string
|
||||
onEvent := func(_, eventType, _, _, _, _ string) { events = append(events, eventType) }
|
||||
|
||||
// Seed ok → no event on init.
|
||||
cc := NewHostCapabilityChecker(st, onEvent, log.New(io.Discard, "", 0))
|
||||
if cc.GetState("h1") != "ok" {
|
||||
t.Fatalf("seed state = %s, want ok", cc.GetState("h1"))
|
||||
}
|
||||
if len(events) != 0 {
|
||||
t.Fatalf("seed must not emit, got %v", events)
|
||||
}
|
||||
|
||||
// ok → degraded: exactly ONE operator event.
|
||||
st.SaveHostReport("h1", "c1", reportWith(capCritDegraded), store.HostReportDenorm{})
|
||||
cc.Check()
|
||||
if cc.GetState("h1") != "degraded" {
|
||||
t.Fatalf("state = %s, want degraded", cc.GetState("h1"))
|
||||
}
|
||||
if len(events) != 1 || events[0] != "agent_capability_degraded" {
|
||||
t.Fatalf("want one agent_capability_degraded, got %v", events)
|
||||
}
|
||||
|
||||
// Steady degraded: NO duplicate event.
|
||||
st.SaveHostReport("h1", "c1", reportWith(capCritDegraded), store.HostReportDenorm{})
|
||||
cc.Check()
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("steady degraded must not re-emit, got %v", events)
|
||||
}
|
||||
|
||||
// degraded → ok: recovered event.
|
||||
st.SaveHostReport("h1", "c1", reportWith(capAllOK), store.HostReportDenorm{})
|
||||
cc.Check()
|
||||
if cc.GetState("h1") != "ok" {
|
||||
t.Fatalf("state = %s, want ok", cc.GetState("h1"))
|
||||
}
|
||||
if len(events) != 2 || events[1] != "agent_capability_recovered" {
|
||||
t.Fatalf("want recovered event, got %v", events)
|
||||
}
|
||||
}
|
||||
|
||||
// A NON-critical degradation must NOT flip the host to degraded (no operator noise).
|
||||
func TestHostCapabilityChecker_NonCriticalIgnored(t *testing.T) {
|
||||
st := newCapStore(t)
|
||||
st.SaveHostReport("h1", "c1", reportWith(capAllOK), store.HostReportDenorm{})
|
||||
var events []string
|
||||
cc := NewHostCapabilityChecker(st, func(_, et, _, _, _, _ string) { events = append(events, et) }, log.New(io.Discard, "", 0))
|
||||
|
||||
st.SaveHostReport("h1", "c1", reportWith(capNonCritDegraded), store.HostReportDenorm{})
|
||||
cc.Check()
|
||||
if cc.GetState("h1") != "ok" {
|
||||
t.Fatalf("non-critical degradation flipped state to %s, want ok", cc.GetState("h1"))
|
||||
}
|
||||
if len(events) != 0 {
|
||||
t.Fatalf("non-critical degradation must not alert, got %v", events)
|
||||
}
|
||||
}
|
||||
|
||||
// A pre-v0.44.0 agent (no capabilities array) is "ok" — an old agent can't trip a false alert.
|
||||
func TestHostCapabilityChecker_OldAgentNoCaps(t *testing.T) {
|
||||
st := newCapStore(t)
|
||||
st.SaveHostReport("h1", "c1", []byte(`{"host_id":"h1"}`), store.HostReportDenorm{})
|
||||
cc := NewHostCapabilityChecker(st, func(_, _, _, _, _, _ string) {}, log.New(io.Discard, "", 0))
|
||||
if cc.GetState("h1") != "ok" {
|
||||
t.Fatalf("old agent (no caps) state = %s, want ok", cc.GetState("h1"))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user