app/webui: fix component-status.json multi-process clobber and unbounded growth
The long-lived bee-web process (writing PSU/kmsg watchdog records ~every
60s) and each short-lived "bee bee-worker" SAT-task subprocess each held
an independent in-memory copy of component-status.json. Whichever saved
last won outright, silently erasing whatever the other had just written —
e.g. a GPU SAT task's pcie:gpu:nvidia result vanishing the next time the
PSU watchdog ticked. ComponentStatusDB.Record now reloads on-disk state
(keyed by newer LastCheckedAt) before merging its own update.
Also stops re-logging identical repeat observations to History: the
ingest contract defines status_history as a transition log ("История
переходов статусов"), not a per-poll journal, but Record appended one
entry per call regardless — a continuously-polled PSU grew an unbounded
run of identical "still OK" entries. Now only appends when a source's
last recorded status for a key actually changes.
The PSU watchdog itself now backs off (60s -> doubling, capped at 30min)
while steady and resets to 60s the moment any PSU's status changes, cutting
ipmitool shellouts and file writes for a fleet that's been stable for a
while.
Also adds the missing "raid" case to the component-detail API (was
returning 404 for any RAID card's detail click on /topo).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Sonnet 5
parent
1d5c02ebaa
commit
cc3997f7b1
@@ -72,6 +72,8 @@ func (db *ComponentStatusDB) Record(key, source, status, detail string) {
|
|||||||
db.mu.Lock()
|
db.mu.Lock()
|
||||||
defer db.mu.Unlock()
|
defer db.mu.Unlock()
|
||||||
|
|
||||||
|
db.reloadLocked()
|
||||||
|
|
||||||
now := time.Now().UTC()
|
now := time.Now().UTC()
|
||||||
rec, exists := db.records[key]
|
rec, exists := db.records[key]
|
||||||
if !exists {
|
if !exists {
|
||||||
@@ -80,8 +82,18 @@ func (db *ComponentStatusDB) Record(key, source, status, detail string) {
|
|||||||
}
|
}
|
||||||
rec.LastCheckedAt = now
|
rec.LastCheckedAt = now
|
||||||
|
|
||||||
entry := ComponentStatusEntry{At: now, Status: status, Source: source, Detail: detail}
|
// History records status *transitions*, per the ingest contract
|
||||||
rec.History = append(rec.History, entry)
|
// (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.
|
// Status merge: OK never downgrades Warning/Critical.
|
||||||
newSev := componentSeverity(status)
|
newSev := componentSeverity(status)
|
||||||
@@ -98,6 +110,17 @@ func (db *ComponentStatusDB) Record(key, source, status, detail string) {
|
|||||||
_ = db.saveLocked()
|
_ = 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.
|
// Get returns the current record for a component key.
|
||||||
func (db *ComponentStatusDB) Get(key string) (ComponentStatusRecord, bool) {
|
func (db *ComponentStatusDB) Get(key string) (ComponentStatusRecord, bool) {
|
||||||
if db == nil {
|
if db == nil {
|
||||||
@@ -126,6 +149,38 @@ func (db *ComponentStatusDB) All() []ComponentStatusRecord {
|
|||||||
return out
|
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 {
|
func (db *ComponentStatusDB) saveLocked() error {
|
||||||
records := make([]ComponentStatusRecord, 0, len(db.records))
|
records := make([]ComponentStatusRecord, 0, len(db.records))
|
||||||
for _, r := range db.records {
|
for _, r := range db.records {
|
||||||
|
|||||||
@@ -97,3 +97,83 @@ func TestApplyComponentStatusDBMatchesGPUByVendor(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func strPtr(s string) *string { return &s }
|
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")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -1851,6 +1851,9 @@ func (h *handler) handleAPIComponentDetail(w http.ResponseWriter, r *http.Reques
|
|||||||
case "psu":
|
case "psu":
|
||||||
title = "PSU"
|
title = "PSU"
|
||||||
prefixes = []string{"psu:"}
|
prefixes = []string{"psu:"}
|
||||||
|
case "raid":
|
||||||
|
title = "RAID"
|
||||||
|
prefixes = []string{"pcie:raid:"}
|
||||||
default:
|
default:
|
||||||
http.NotFound(w, r)
|
http.NotFound(w, r)
|
||||||
return
|
return
|
||||||
|
|||||||
@@ -11,17 +11,32 @@ import (
|
|||||||
"bee/audit/internal/collector"
|
"bee/audit/internal/collector"
|
||||||
)
|
)
|
||||||
|
|
||||||
const healthPollInterval = 60 * time.Second
|
const (
|
||||||
const psuIPMITimeout = 15 * time.Second
|
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
|
// healthPoller runs periodic health checks for hardware components that do not
|
||||||
// emit kernel log events (e.g. PSU). Results are written to ComponentStatusDB.
|
// 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 {
|
type healthPoller struct {
|
||||||
statusDB *app.ComponentStatusDB
|
statusDB *app.ComponentStatusDB
|
||||||
|
interval time.Duration
|
||||||
}
|
}
|
||||||
|
|
||||||
func newHealthPoller(statusDB *app.ComponentStatusDB) *healthPoller {
|
func newHealthPoller(statusDB *app.ComponentStatusDB) *healthPoller {
|
||||||
return &healthPoller{statusDB: statusDB}
|
return &healthPoller{statusDB: statusDB, interval: healthPollIntervalMin}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *healthPoller) start() {
|
func (p *healthPoller) start() {
|
||||||
@@ -29,16 +44,34 @@ func (p *healthPoller) start() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (p *healthPoller) run() {
|
func (p *healthPoller) run() {
|
||||||
ticker := time.NewTicker(healthPollInterval)
|
timer := time.NewTimer(p.interval)
|
||||||
defer ticker.Stop()
|
defer timer.Stop()
|
||||||
for range ticker.C {
|
for range timer.C {
|
||||||
p.pollPSU()
|
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 {
|
if p.statusDB == nil {
|
||||||
return
|
return false
|
||||||
}
|
}
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), psuIPMITimeout)
|
ctx, cancel := context.WithTimeout(context.Background(), psuIPMITimeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
@@ -49,21 +82,25 @@ func (p *healthPoller) pollPSU() {
|
|||||||
if err := cmd.Run(); err != nil {
|
if err := cmd.Run(); err != nil {
|
||||||
// IPMI not available or not a server — skip silently.
|
// IPMI not available or not a server — skip silently.
|
||||||
slog.Debug("health poller: ipmitool sdr unavailable", "err", err)
|
slog.Debug("health poller: ipmitool sdr unavailable", "err", err)
|
||||||
return
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
slots := collector.PSUSlotsFromSDR(out.String())
|
slots := collector.PSUSlotsFromSDR(out.String())
|
||||||
if len(slots) == 0 {
|
if len(slots) == 0 {
|
||||||
return
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
const source = "watchdog:psu"
|
const source = "watchdog:psu"
|
||||||
|
changed := false
|
||||||
for slot, psu := range slots {
|
for slot, psu := range slots {
|
||||||
key := "psu:" + slot
|
key := "psu:" + slot
|
||||||
status := psu.Status
|
status := psu.Status
|
||||||
if status == "" {
|
if status == "" {
|
||||||
status = "Unknown"
|
status = "Unknown"
|
||||||
}
|
}
|
||||||
|
if prev, ok := p.statusDB.Get(key); !ok || prev.Status != status {
|
||||||
|
changed = true
|
||||||
|
}
|
||||||
detail := ""
|
detail := ""
|
||||||
switch status {
|
switch status {
|
||||||
case "Critical":
|
case "Critical":
|
||||||
@@ -73,4 +110,5 @@ func (p *healthPoller) pollPSU() {
|
|||||||
}
|
}
|
||||||
p.statusDB.Record(key, source, status, detail)
|
p.statusDB.Record(key, source, status, detail)
|
||||||
}
|
}
|
||||||
|
return changed
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user