reanimator/bee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

refactor: harden diagnostics and consolidate runtime code

Browse Source
This commit is contained in:
Mikhail Chusavitin
2026-09-01 13:01:28 +03:00
parent ac4bc0b2b7
commit 0a6ca8ba0f
49 changed files with 1441 additions and 837 deletions
Download Patch File Download Diff File Expand all files Collapse all files
audit/internal/platform/sat.go
+55 -115
View File
@@ -191,13 +191,19 @@ func streamExecOutput(cmd *exec.Cmd, logFunc func(string), livePath string) ([]b
return buf.Bytes(), waitErr
}
// NvidiaGPU holds basic GPU info from nvidia-smi.
// satJob describes one command and the checks needed to turn its process and
// output results into a SAT verdict.
type satJob struct {
name string
cmd []string
env []string // extra env vars (appended to os.Environ)
collectGPU bool // collect GPU metrics via nvidia-smi while this job runs
gpuIndices []int // GPU indices to collect metrics for (empty = all)
name string
cmd []string
env []string // extra env vars (appended to os.Environ)
// validate checks successful command output against the tool's documented
// result format. It may inspect artifacts from earlier jobs in runDir.
// FAILED means the tool proved a test failure; UNSUPPORTED means the pinned
// output contract could not be recognized safely.
validate func(runDir string, out []byte) (status, detail string)
collectGPU bool // collect GPU metrics via nvidia-smi while this job runs
gpuIndices []int // GPU indices to collect metrics for (empty = all)
// informational marks a preflight/metadata job (e.g. dcgmi discovery) whose
// failure shouldn't flip the pack's overall status — the diagnostic jobs
// that follow it are the actual test of GPU health.
@@ -314,7 +320,7 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
var summary strings.Builder
stats := satStats{}
nvidiaPack := strings.HasPrefix(prefix, "gpu-nvidia")
nvidiaPack := isNvidiaAcceptancePack(prefix)
perGPU := map[int]*nvidiaGPUStatusFile{}
selectedGPUIndices := map[int]struct{}{}
fmt.Fprintf(&summary, "run_at_utc=%s\n", time.Now().UTC().Format(time.RFC3339))
@@ -338,6 +344,7 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
var out []byte
var err error
jobDetail := ""
if nvidiaPack && nvidiaJobNeedsHealthCheck(job) {
if msg, healthErr := checkNvidiaJobHealth(job.gpuIndices); healthErr != nil {
@@ -346,6 +353,7 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
}
out = []byte(msg + "\n")
err = healthErr
jobDetail = msg
}
}
@@ -379,6 +387,23 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
if err == nil {
err = healthErr
}
jobDetail = msg
}
}
status, rc := classifySATResult(job.name, out, err)
validationDetail := singleLineSATDetail(jobDetail)
validated := false
if status == "OK" && job.validate != nil {
status, validationDetail = validateSATJobOutput(job, runDir, out)
validated = true
}
if validated {
if validationDetail != "" {
if len(out) > 0 && !bytes.HasSuffix(out, []byte("\n")) {
out = append(out, '\n')
}
out = append(out, []byte(fmt.Sprintf("[bee-validator] %s: %s\n", status, validationDetail))...)
}
}
@@ -392,7 +417,6 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
if ctx.Err() != nil {
return "", ctx.Err()
}
status, rc := classifySATResult(job.name, out, err)
if job.informational && status != "OK" {
stats.Informational++
} else {
@@ -406,6 +430,9 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
key := strings.TrimSuffix(strings.TrimPrefix(job.name, "0"), ".log")
fmt.Fprintf(&summary, "%s_rc=%d\n", key, rc)
fmt.Fprintf(&summary, "%s_status=%s\n", key, status)
if validationDetail != "" {
fmt.Fprintf(&summary, "%s_detail=%s\n", key, singleLineSATDetail(validationDetail))
}
}
writeSATStats(&summary, stats)
if err := os.WriteFile(filepath.Join(runDir, "summary.txt"), []byte(summary.String()), 0644); err != nil {
@@ -420,6 +447,25 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa
return runDir, nil
}
func isNvidiaAcceptancePack(prefix string) bool {
return strings.HasPrefix(prefix, "gpu-nvidia") || prefix == "nccl-tests"
}
func validateSATJobOutput(job satJob, runDir string, out []byte) (string, string) {
status, detail := job.validate(runDir, out)
status = strings.ToUpper(strings.TrimSpace(status))
switch status {
case "OK", "FAILED", "UNSUPPORTED", "PARTIAL":
return status, strings.TrimSpace(detail)
default:
return "UNSUPPORTED", fmt.Sprintf("validator returned invalid status %q", status)
}
}
func singleLineSATDetail(detail string) string {
return strings.Join(strings.Fields(detail), " ")
}
func updateNvidiaGPUStatus(perGPU map[int]*nvidiaGPUStatusFile, idx int, status, jobName, detail string) {
entry := perGPU[idx]
if entry == nil {
@@ -505,112 +551,6 @@ func writeNvidiaGPUStatusFiles(runDir, overall string, perGPU map[int]*nvidiaGPU
return nil
}
func nvidiaSATStatusSeverity(status string) int {
switch strings.ToUpper(strings.TrimSpace(status)) {
case "FAILED":
return 3
case "PARTIAL", "UNSUPPORTED":
return 2
case "OK":
return 1
default:
return 0
}
}
func firstLine(s string) string {
s = strings.TrimSpace(s)
if s == "" {
return ""
}
if idx := strings.IndexByte(s, '\n'); idx >= 0 {
return strings.TrimSpace(s[:idx])
}
return s
}
func nvidiaJobNeedsHealthCheck(job satJob) bool {
if job.collectGPU {
return true
}
name := strings.ToLower(strings.TrimSpace(job.name))
return strings.Contains(name, "dcgmi") ||
strings.Contains(name, "gpu-burn") ||
strings.Contains(name, "gpu-stress") ||
strings.Contains(name, "dcgmproftester")
}
func checkNvidiaJobHealth(selected []int) (string, error) {
health, err := readNvidiaGPUHealth()
if err != nil {
return "", nil
}
var bad []nvidiaGPUHealth
selectedSet := make(map[int]struct{}, len(selected))
for _, idx := range selected {
selectedSet[idx] = struct{}{}
}
for _, gpu := range health {
if len(selectedSet) > 0 {
if _, ok := selectedSet[gpu.Index]; !ok {
continue
}
}
if gpu.NeedsReset {
bad = append(bad, gpu)
}
}
if len(bad) == 0 {
return "", nil
}
lines := make([]string, 0, len(bad)+1)
lines = append(lines, "NVIDIA GPU health check failed:")
for _, gpu := range bad {
lines = append(lines, fmt.Sprintf("gpu %d (%s) requires reset: %s", gpu.Index, gpu.Name, gpu.RawLine))
}
return strings.Join(lines, "\n"), errors.New("nvidia gpu requires reset")
}
func readNvidiaGPUHealth() ([]nvidiaGPUHealth, error) {
out, err := satExecCommand(
"nvidia-smi",
"--query-gpu=index,name,temperature.gpu,power.draw,utilization.gpu,memory.used,memory.total",
"--format=csv,noheader,nounits",
).Output()
if err != nil {
return nil, fmt.Errorf("nvidia-smi: %w", err)
}
return parseNvidiaGPUHealth(string(out)), nil
}
func parseNvidiaGPUHealth(raw string) []nvidiaGPUHealth {
var gpus []nvidiaGPUHealth
for _, line := range strings.Split(strings.TrimSpace(raw), "\n") {
line = strings.TrimSpace(line)
if line == "" {
continue
}
parts := strings.Split(line, ",")
if len(parts) < 2 {
gpus = append(gpus, nvidiaGPUHealth{RawLine: line, ParseFailure: true})
continue
}
idx, err := strconv.Atoi(strings.TrimSpace(parts[0]))
if err != nil {
gpus = append(gpus, nvidiaGPUHealth{RawLine: line, ParseFailure: true})
continue
}
upper := strings.ToUpper(line)
gpus = append(gpus, nvidiaGPUHealth{
Index: idx,
Name: strings.TrimSpace(parts[1]),
NeedsReset: strings.Contains(upper, "GPU REQUIRES RESET"),
RawLine: line,
})
}
return gpus
}
// runSATCommandCtx runs cmd and returns its combined output. livePath is
// variadic purely so existing callers are unaffected: pass a path (job's
// output file) to also stream output to disk live as it runs, so a crash
@@ -736,7 +676,7 @@ func (s *satStats) Add(status string) {
switch status {
case "OK":
s.OK++
case "UNSUPPORTED":
case "UNSUPPORTED", "PARTIAL":
s.Unsupported++
default:
s.Failed++