package app import ( "encoding/json" "fmt" "os" "path/filepath" "sort" "strings" "sync" "time" "bee/audit/internal/collector" ) // ComponentStatusDB is a persistent, append-only store of hardware component health records. // Records are keyed by component identity strings (e.g. "pcie:0000:c8:00.0", "storage:nvme0n1"). // Once a component is marked Warning or Critical, subsequent OK entries do not downgrade it — // the component stays at the highest observed severity until explicitly reset. type ComponentStatusDB struct { path string mu sync.Mutex records map[string]*ComponentStatusRecord } // ComponentStatusRecord holds the current and historical health of one hardware component. type ComponentStatusRecord struct { ComponentKey string `json:"component_key"` Status string `json:"status"` // "OK", "Warning", "Critical", "Unknown" LastCheckedAt time.Time `json:"last_checked_at"` LastChangedAt time.Time `json:"last_changed_at"` ErrorSummary string `json:"error_summary,omitempty"` History []ComponentStatusEntry `json:"history"` } // ComponentStatusEntry is one observation written to a component's history. type ComponentStatusEntry struct { At time.Time `json:"at"` Status string `json:"status"` Source string `json:"source"` // e.g. "sat:nvidia", "sat:memory", "watchdog:kmsg" Detail string `json:"detail,omitempty"` } // OpenComponentStatusDB opens (or creates) the JSON status DB at path. func OpenComponentStatusDB(path string) (*ComponentStatusDB, error) { db := &ComponentStatusDB{ path: path, records: make(map[string]*ComponentStatusRecord), } if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { return nil, err } data, err := readFileLimited(path, 10<<20) if err != nil && !os.IsNotExist(err) { return nil, err } if len(data) > 0 { var records []ComponentStatusRecord if err := json.Unmarshal(data, &records); err == nil { for i := range records { db.records[records[i].ComponentKey] = &records[i] } } } return db, nil } // Record writes one observation for the given component key. // source is a short label like "sat:nvidia" or "watchdog:kmsg". // status is "OK", "Warning", "Critical", or "Unknown". // OK never downgrades an existing Warning or Critical status. func (db *ComponentStatusDB) Record(key, source, status, detail string) { if db == nil || strings.TrimSpace(key) == "" { return } db.mu.Lock() defer db.mu.Unlock() db.reloadLocked() now := time.Now().UTC() rec, exists := db.records[key] if !exists { rec = &ComponentStatusRecord{ComponentKey: key} db.records[key] = rec } rec.LastCheckedAt = now // 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) curSev := componentSeverity(rec.Status) if newSev > curSev { rec.Status = status rec.LastChangedAt = now rec.ErrorSummary = detail } else if rec.Status == "" { rec.Status = status rec.LastChangedAt = now } _ = 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 { return ComponentStatusRecord{}, false } db.mu.Lock() defer db.mu.Unlock() r, ok := db.records[key] if !ok { return ComponentStatusRecord{}, false } return *r, true } // All returns a snapshot of all records. func (db *ComponentStatusDB) All() []ComponentStatusRecord { if db == nil { return nil } db.mu.Lock() defer db.mu.Unlock() out := make([]ComponentStatusRecord, 0, len(db.records)) for _, r := range db.records { out = append(out, *r) } 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 { records = append(records, *r) } data, err := json.MarshalIndent(records, "", " ") if err != nil { return err } return os.WriteFile(db.path, data, 0644) } // componentSeverity returns a numeric severity so higher values win. func componentSeverity(status string) int { switch strings.TrimSpace(status) { case "Critical": return 3 case "Warning": return 2 case "OK": return 1 default: return 0 } } // ApplySATResultToDB reads a SAT summary.txt from the run directory next to archivePath // and writes component status records to db for the given SAT target. // archivePath may be either a bare .tar.gz path or "Archive written to /path/foo.tar.gz". func ApplySATResultToDB(db *ComponentStatusDB, target, archivePath string) { if db == nil || strings.TrimSpace(archivePath) == "" { return } archivePath = extractArchivePath(archivePath) if archivePath == "" { return } runDir := strings.TrimSuffix(archivePath, ".tar.gz") data, err := os.ReadFile(filepath.Join(runDir, "summary.txt")) if err != nil { return } kv := parseSATKV(string(data)) overall := strings.ToUpper(strings.TrimSpace(kv["overall_status"])) if overall == "" { return } source := "sat:" + target dbStatus := satStatusToDBStatus(overall) detail := target + " SAT: " + overall if overall != "OK" { if reason := prependHardwareFaultBanner(runDir, satFailureDetailFromKV(kv)); reason != "" { detail += " — " + reason } } // Map SAT target to component keys. GPU targets are keyed by vendor, not // by the raw target string: "nvidia" (Check tier) and "nvidia-stress" / // "nvidia-targeted-stress" (Load/Burn tier) all exercise the same // physical GPUs, so they must share one severity-tracked record — a // severity-1 Check run after a severity-3 Load failure must not lose // that failure just because it ran more recently. Recording each target // under its own key (the previous behavior) also silently broke // applyComponentStatusDB below, which expects "pcie:gpu:" and // otherwise fails to match any real BDF. switch target { case "nvidia", "nvidia-targeted-stress", "nvidia-compute", "nvidia-targeted-power", "nvidia-pulse", "nvidia-interconnect", "nvidia-bandwidth", "nvidia-stress", "nvidia-config": db.Record("pcie:gpu:nvidia", source, dbStatus, detail) case "amd", "amd-stress", "amd-mem", "amd-bandwidth": db.Record("pcie:gpu:amd", source, dbStatus, detail) case "memory", "memory-stress", "sat-stress": db.Record("memory:all", source, dbStatus, detail) case "cpu", "platform-stress": db.Record("cpu:all", source, dbStatus, detail) case "storage": // Try to record per-device if available in summary. recordedAny := false for key, val := range kv { if !strings.HasSuffix(key, "_status") || key == "overall_status" { continue } base := strings.TrimSuffix(key, "_status") idx := strings.Index(base, "_") if idx <= 0 { continue } devName := base[:idx] devStatus := satStatusToDBStatus(strings.ToUpper(strings.TrimSpace(val))) devDetail := "storage SAT: " + val if strings.ToUpper(strings.TrimSpace(val)) != "OK" { if rc, ok := kv[base+"_rc"]; ok { devDetail = fmt.Sprintf("storage SAT job %q: %s (rc=%s)", base, val, rc) } else { devDetail = fmt.Sprintf("storage SAT job %q: %s", base, val) } } db.Record("storage:"+devName, source, devStatus, devDetail) recordedAny = true } if !recordedAny { db.Record("storage:all", source, dbStatus, detail) } } } func satStatusToDBStatus(overall string) string { switch overall { case "OK": return "OK" case "FAILED": return "Warning" case "PARTIAL", "UNSUPPORTED": return "Unknown" default: return "Unknown" } } // ExtractArchivePath extracts a bare path from a string that may be // "Archive written to /path/to/run-dir" or already a bare path. func ExtractArchivePath(s string) string { return extractArchivePath(s) } // ReadSATOverallStatus reads the overall_status value from the summary.txt // file located in the run directory alongside archivePath. // Returns "" if the file cannot be read. func ReadSATOverallStatus(archivePath string) string { if strings.TrimSpace(archivePath) == "" { return "" } runDir := strings.TrimSuffix(archivePath, ".tar.gz") data, err := os.ReadFile(filepath.Join(runDir, "summary.txt")) if err != nil { return "" } kv := parseSATKV(string(data)) return strings.ToUpper(strings.TrimSpace(kv["overall_status"])) } // SATFailureDetail explains *why* a SAT run's overall_status isn't OK, read // from the summary.txt next to archivePath. "SAT overall_status=FAILED (see // summary.txt)" tells an engineer nothing without opening the run directory // themselves; this pulls the specific reason out so it can be surfaced // directly in the task's error message and in the component status DB. // Returns "" if summary.txt is unreadable or carries no identifiable reason. func SATFailureDetail(archivePath string) string { if strings.TrimSpace(archivePath) == "" { return "" } runDir := strings.TrimSuffix(extractArchivePath(archivePath), ".tar.gz") data, err := os.ReadFile(filepath.Join(runDir, "summary.txt")) if err != nil { return "" } reason := satFailureDetailFromKV(parseSATKV(string(data))) return prependHardwareFaultBanner(runDir, reason) } // prependHardwareFaultBanner checks the SAT run directory's captured logs // for a known GPU hardware fault (e.g. Xid 79 "fallen off the bus") and, if // found, puts a plain-English banner in front of reason — so the task's // error message reads "GPU fell off the bus, reboot required" directly // instead of just "failed sub-job(s): ...", which tells an engineer nothing // without opening the run directory and cross-referencing Xid codes by hand. func prependHardwareFaultBanner(runDir, reason string) string { banner := gpuHardwareFaultBanner(runDir) switch { case banner == "": return reason case reason == "": return banner default: return banner + " (" + reason + ")" } } // gpuHardwareFaultBanner scans a SAT run directory's captured *.log files // for NVIDIA Xid codes that mean the GPU cannot recover without a physical // reboot, returning a de-duplicated, human-readable summary. func gpuHardwareFaultBanner(runDir string) string { entries, err := os.ReadDir(runDir) if err != nil { return "" } seen := map[string]bool{} var messages []string for _, e := range entries { if e.IsDir() || !strings.HasSuffix(e.Name(), ".log") { continue } data, err := readFileLimited(filepath.Join(runDir, e.Name()), 2<<20) if err != nil { continue } for _, line := range strings.Split(string(data), "\n") { msg, ok := collector.XidHardwareFaultMessage(line) if !ok || seen[msg] { continue } seen[msg] = true messages = append(messages, msg) } } if len(messages) == 0 { return "" } return strings.Join(messages, "; ") } // satFailureDetailFromKV inspects an already-parsed summary.txt for the // reason behind a non-OK overall_status. // // - Checks that write structured findings (nvidia-config's GPU config / // NVLink topology check) put the human-readable reason straight into a // "warnings" field — return that verbatim. // - Generic SAT acceptance packs (nvidia, amd, memory, cpu, storage, ...) // instead record one "_status"/"_rc" pair per sub-job; walk // those and report whichever job(s) didn't come back OK/UNSUPPORTED. func satFailureDetailFromKV(kv map[string]string) string { if w := strings.TrimSpace(kv["warnings"]); w != "" { return w } keys := make([]string, 0, len(kv)) for k := range kv { keys = append(keys, k) } sort.Strings(keys) var failed []string for _, k := range keys { if k == "overall_status" || !strings.HasSuffix(k, "_status") { continue } v := strings.ToUpper(strings.TrimSpace(kv[k])) if v == "" || v == "OK" || v == "UNSUPPORTED" { continue } job := strings.TrimSuffix(k, "_status") if rc, ok := kv[job+"_rc"]; ok && strings.TrimSpace(rc) != "" { failed = append(failed, fmt.Sprintf("%s=%s (rc=%s)", job, v, rc)) } else { failed = append(failed, fmt.Sprintf("%s=%s", job, v)) } } if len(failed) == 0 { return "" } return "failed sub-job(s): " + strings.Join(failed, ", ") } func extractArchivePath(s string) string { s = strings.TrimSpace(s) if rest, ok := strings.CutPrefix(s, "Archive written to "); ok { return strings.TrimSpace(rest) } return s } func parseSATKV(raw string) map[string]string { kv := make(map[string]string) for _, line := range strings.Split(raw, "\n") { k, v, ok := strings.Cut(strings.TrimSpace(line), "=") if ok { kv[strings.TrimSpace(k)] = strings.TrimSpace(v) } } return kv }