diff --git a/audit/internal/app/component_status_db.go b/audit/internal/app/component_status_db.go index 638916f..4b68b9a 100644 --- a/audit/internal/app/component_status_db.go +++ b/audit/internal/app/component_status_db.go @@ -72,6 +72,8 @@ func (db *ComponentStatusDB) Record(key, source, status, detail string) { db.mu.Lock() defer db.mu.Unlock() + db.reloadLocked() + now := time.Now().UTC() rec, exists := db.records[key] if !exists { @@ -80,8 +82,18 @@ func (db *ComponentStatusDB) Record(key, source, status, detail string) { } rec.LastCheckedAt = now - entry := ComponentStatusEntry{At: now, Status: status, Source: source, Detail: detail} - rec.History = append(rec.History, entry) + // History records status *transitions*, per the ingest contract + // (bible-local/docs/hardware-ingest-contract.md: "История переходов + // статусов" — status_history is a transition log, not a per-poll + // journal). Skip the append when this source's last recorded status for + // this key is unchanged, or a continuously-polled component (e.g. the + // PSU watchdog, every 60s indefinitely) grows an unbounded run of + // identical "still OK" entries and the file never stops growing. + // LastCheckedAt above already carries "we still saw this as of now" for + // a steady-state component, so nothing is lost by not repeating it here. + if last := lastEntryFromSource(rec.History, source); last == nil || last.Status != status { + rec.History = append(rec.History, ComponentStatusEntry{At: now, Status: status, Source: source, Detail: detail}) + } // Status merge: OK never downgrades Warning/Critical. newSev := componentSeverity(status) @@ -98,6 +110,17 @@ func (db *ComponentStatusDB) Record(key, source, status, detail string) { _ = db.saveLocked() } +// lastEntryFromSource returns the most recent history entry recorded by the +// given source, or nil if that source has never reported for this key. +func lastEntryFromSource(history []ComponentStatusEntry, source string) *ComponentStatusEntry { + for i := len(history) - 1; i >= 0; i-- { + if history[i].Source == source { + return &history[i] + } + } + return nil +} + // Get returns the current record for a component key. func (db *ComponentStatusDB) Get(key string) (ComponentStatusRecord, bool) { if db == nil { @@ -126,6 +149,38 @@ func (db *ComponentStatusDB) All() []ComponentStatusRecord { return out } +// reloadLocked merges on-disk state into memory before this process applies +// its own update. component-status.json is shared by the long-lived bee-web +// process (writing PSU/kmsg watchdog records roughly every 60s) and each +// short-lived "bee bee-worker" subprocess spawned per SAT task (writing GPU/ +// CPU/memory/storage records once on completion) — each holds its own +// in-memory copy backed by the same file. Without this reload, saveLocked +// below would dump the caller's stale in-memory map over the file and erase +// whatever the other process wrote in between: e.g. a GPU SAT task's +// pcie:gpu:nvidia record, written by a worker subprocess, silently vanishing +// the next time the main process's health poller ticks and saves its own +// GPU-blind snapshot. Only keys with a newer LastCheckedAt on disk are +// pulled in, so this process's own pending (not-yet-saved) update for key is +// never discarded by its own reload. +func (db *ComponentStatusDB) reloadLocked() { + data, err := readFileLimited(db.path, 10<<20) + if err != nil || len(data) == 0 { + return + } + var onDisk []ComponentStatusRecord + if err := json.Unmarshal(data, &onDisk); err != nil { + return + } + for i := range onDisk { + key := onDisk[i].ComponentKey + if existing, ok := db.records[key]; ok && !onDisk[i].LastCheckedAt.After(existing.LastCheckedAt) { + continue + } + rec := onDisk[i] + db.records[key] = &rec + } +} + func (db *ComponentStatusDB) saveLocked() error { records := make([]ComponentStatusRecord, 0, len(db.records)) for _, r := range db.records { diff --git a/audit/internal/app/component_status_db_test.go b/audit/internal/app/component_status_db_test.go index ab1f89f..e2a263c 100644 --- a/audit/internal/app/component_status_db_test.go +++ b/audit/internal/app/component_status_db_test.go @@ -97,3 +97,83 @@ func TestApplyComponentStatusDBMatchesGPUByVendor(t *testing.T) { } func strPtr(s string) *string { return &s } + +// TestRecordDeduplicatesRepeatedIdenticalStatusFromSameSource guards the +// hardware-ingest-contract.md rule that status_history is a transition log +// ("История переходов статусов"), not a per-poll journal. A component +// checked continuously (the PSU watchdog, every 60s indefinitely) must not +// grow one History entry per poll while its status stays unchanged — that +// is exactly what made component-status.json grow without bound in +// practice. +func TestRecordDeduplicatesRepeatedIdenticalStatusFromSameSource(t *testing.T) { + db, err := OpenComponentStatusDB(filepath.Join(t.TempDir(), "component-status.json")) + if err != nil { + t.Fatal(err) + } + for i := 0; i < 50; i++ { + db.Record("psu:0", "watchdog:psu", "OK", "") + } + rec, ok := db.Get("psu:0") + if !ok { + t.Fatalf("expected psu:0 record") + } + if len(rec.History) != 1 { + t.Fatalf("history len=%d want 1 after 50 identical polls (dedup by transition)", len(rec.History)) + } + + // A real transition must still be recorded, and dedup resumes at the + // new status. + db.Record("psu:0", "watchdog:psu", "Critical", "PSU sensor reported non-OK state") + db.Record("psu:0", "watchdog:psu", "Critical", "PSU sensor reported non-OK state") + rec, _ = db.Get("psu:0") + if len(rec.History) != 2 { + t.Fatalf("history len=%d want 2 (OK, then Critical, second Critical deduped)", len(rec.History)) + } + if rec.Status != "Critical" { + t.Fatalf("status=%q want Critical", rec.Status) + } +} + +// TestRecordDoesNotClobberConcurrentWriterFromAnotherProcess reproduces the +// bug behind a real support bundle where many GPU/CPU/memory SAT tasks had +// completed successfully but component-status.json only ever held PSU +// records: bee-web keeps one ComponentStatusDB open for its whole process +// lifetime (writing PSU/kmsg watchdog records ~every 60s via a long-running +// health poller), while each SAT task runs as a short-lived "bee bee-worker" +// subprocess that opens its own separate ComponentStatusDB instance backed +// by the same file. Without a reload-before-write, the long-lived process's +// stale in-memory snapshot (which never learned about the subprocess's +// write) overwrites the whole file on its next save and erases it. +func TestRecordDoesNotClobberConcurrentWriterFromAnotherProcess(t *testing.T) { + path := filepath.Join(t.TempDir(), "component-status.json") + + // bee-web's long-lived DB instance, opened once at process start. + webDB, err := OpenComponentStatusDB(path) + if err != nil { + t.Fatal(err) + } + + // A "bee bee-worker" subprocess for a completed GPU SAT task opens its + // own instance backed by the same file and records the result. + workerDB, err := OpenComponentStatusDB(path) + if err != nil { + t.Fatal(err) + } + workerDB.Record("pcie:gpu:nvidia", "sat:nvidia", "OK", "nvidia SAT: OK") + + // bee-web's health poller ticks next, using its own (older) in-memory + // view, and records a PSU status it polled independently. + webDB.Record("psu:0", "watchdog:psu", "OK", "") + + // The GPU record written by the worker subprocess must survive. + onDisk, err := OpenComponentStatusDB(path) + if err != nil { + t.Fatal(err) + } + if rec, ok := onDisk.Get("pcie:gpu:nvidia"); !ok || rec.Status != "OK" { + t.Fatalf("pcie:gpu:nvidia record lost after concurrent PSU write: ok=%v rec=%+v", ok, rec) + } + if _, ok := onDisk.Get("psu:0"); !ok { + t.Fatalf("psu:0 record missing") + } +} diff --git a/audit/internal/webui/api.go b/audit/internal/webui/api.go index 146f62a..4ca831d 100644 --- a/audit/internal/webui/api.go +++ b/audit/internal/webui/api.go @@ -1851,6 +1851,9 @@ func (h *handler) handleAPIComponentDetail(w http.ResponseWriter, r *http.Reques case "psu": title = "PSU" prefixes = []string{"psu:"} + case "raid": + title = "RAID" + prefixes = []string{"pcie:raid:"} default: http.NotFound(w, r) return diff --git a/audit/internal/webui/health_poller.go b/audit/internal/webui/health_poller.go index 1cdce0e..ccaad5d 100644 --- a/audit/internal/webui/health_poller.go +++ b/audit/internal/webui/health_poller.go @@ -11,17 +11,32 @@ import ( "bee/audit/internal/collector" ) -const healthPollInterval = 60 * time.Second -const psuIPMITimeout = 15 * time.Second +const ( + healthPollIntervalMin = 60 * time.Second + // healthPollIntervalMax caps the backoff below: a PSU that has been + // steady for a while is polled at most this rarely, so a real failure + // is still caught within a bounded window even after a long quiet + // stretch. + healthPollIntervalMax = 30 * time.Minute + psuIPMITimeout = 15 * time.Second +) // healthPoller runs periodic health checks for hardware components that do not // emit kernel log events (e.g. PSU). Results are written to ComponentStatusDB. +// +// The poll interval backs off (doubling, capped at healthPollIntervalMax) +// each tick where no PSU's status changed, and resets to +// healthPollIntervalMin the moment any PSU does change — a steady-state PSU +// bank doesn't need re-checking every 60s forever (each poll is a +// component-status.json write and an ipmitool shellout), while a PSU that +// just started flapping gets caught quickly again. type healthPoller struct { statusDB *app.ComponentStatusDB + interval time.Duration } func newHealthPoller(statusDB *app.ComponentStatusDB) *healthPoller { - return &healthPoller{statusDB: statusDB} + return &healthPoller{statusDB: statusDB, interval: healthPollIntervalMin} } func (p *healthPoller) start() { @@ -29,16 +44,34 @@ func (p *healthPoller) start() { } func (p *healthPoller) run() { - ticker := time.NewTicker(healthPollInterval) - defer ticker.Stop() - for range ticker.C { - p.pollPSU() + timer := time.NewTimer(p.interval) + defer timer.Stop() + for range timer.C { + p.interval = nextHealthPollInterval(p.interval, p.pollPSU()) + timer.Reset(p.interval) } } -func (p *healthPoller) pollPSU() { +// nextHealthPollInterval computes the next poll interval given whether the +// last poll observed any status change: reset to the fast floor on change, +// otherwise double (capped) toward the slow ceiling. +func nextHealthPollInterval(current time.Duration, changed bool) time.Duration { + if changed { + return healthPollIntervalMin + } + next := current * 2 + if next > healthPollIntervalMax { + next = healthPollIntervalMax + } + return next +} + +// pollPSU polls PSU status via ipmitool and records it to statusDB. Returns +// true if any PSU's status differs from what statusDB currently has for it, +// which run uses to reset the backoff. +func (p *healthPoller) pollPSU() bool { if p.statusDB == nil { - return + return false } ctx, cancel := context.WithTimeout(context.Background(), psuIPMITimeout) defer cancel() @@ -49,21 +82,25 @@ func (p *healthPoller) pollPSU() { if err := cmd.Run(); err != nil { // IPMI not available or not a server — skip silently. slog.Debug("health poller: ipmitool sdr unavailable", "err", err) - return + return false } slots := collector.PSUSlotsFromSDR(out.String()) if len(slots) == 0 { - return + return false } const source = "watchdog:psu" + changed := false for slot, psu := range slots { key := "psu:" + slot status := psu.Status if status == "" { status = "Unknown" } + if prev, ok := p.statusDB.Get(key); !ok || prev.Status != status { + changed = true + } detail := "" switch status { case "Critical": @@ -73,4 +110,5 @@ func (p *healthPoller) pollPSU() { } p.statusDB.Record(key, source, status, detail) } + return changed } diff --git a/audit/internal/webui/health_poller_test.go b/audit/internal/webui/health_poller_test.go new file mode 100644 index 0000000..9c7f961 --- /dev/null +++ b/audit/internal/webui/health_poller_test.go @@ -0,0 +1,54 @@ +package webui + +import ( + "path/filepath" + "testing" + "time" + + "bee/audit/internal/app" +) + +func TestNextHealthPollIntervalBacksOffAndResetsOnChange(t *testing.T) { + interval := healthPollIntervalMin + for i := 0; i < 10; i++ { + interval = nextHealthPollInterval(interval, false) + if interval > healthPollIntervalMax { + t.Fatalf("interval=%s exceeded cap %s after %d steady ticks", interval, healthPollIntervalMax, i+1) + } + } + if interval != healthPollIntervalMax { + t.Fatalf("interval=%s want cap %s after repeated steady ticks", interval, healthPollIntervalMax) + } + + // A change resets straight back to the fast floor, regardless of how + // far the backoff had climbed. + if got := nextHealthPollInterval(interval, true); got != healthPollIntervalMin { + t.Fatalf("interval after change=%s want floor %s", got, healthPollIntervalMin) + } +} + +func TestHealthPollerPollPSUUnavailableReturnsFalseWithoutPanic(t *testing.T) { + // ipmitool is not present in this test environment, so pollPSU must + // degrade to a no-op (false, no crash) rather than erroring out the + // poller loop. + db, err := app.OpenComponentStatusDB(filepath.Join(t.TempDir(), "component-status.json")) + if err != nil { + t.Fatal(err) + } + p := newHealthPoller(db) + if p.interval != healthPollIntervalMin { + t.Fatalf("initial interval=%s want floor %s", p.interval, healthPollIntervalMin) + } + if got := p.pollPSU(); got { + t.Fatalf("pollPSU()=%v want false when ipmitool is unavailable", got) + } +} + +func TestHealthPollIntervalBoundsAreSane(t *testing.T) { + if healthPollIntervalMin >= healthPollIntervalMax { + t.Fatalf("min %s must be less than max %s", healthPollIntervalMin, healthPollIntervalMax) + } + if healthPollIntervalMin != 60*time.Second { + t.Fatalf("min=%s want 60s (unchanged default poll rate)", healthPollIntervalMin) + } +}