diff --git a/audit/internal/app/app.go b/audit/internal/app/app.go index 387dbca..6a5d359 100644 --- a/audit/internal/app/app.go +++ b/audit/internal/app/app.go @@ -161,19 +161,25 @@ type runtimeChecker interface { CaptureTechnicalDump(baseDir string) error } -func New(platform *platform.System) *App { +func New(sys *platform.System) *App { a := &App{ - network: platform, - services: platform, - exports: platform, - tools: platform, - sat: platform, - runtime: platform, - installer: platform, + network: sys, + services: sys, + exports: sys, + tools: sys, + sat: sys, + runtime: sys, + installer: sys, } if db, err := OpenComponentStatusDB(DefaultExportDir + "/component-status.json"); err == nil { a.StatusDB = db } + // Let any running blackbox worker know promptly when a SAT job finishes, + // instead of waiting out its own adaptive flush period — see + // requestBlackboxSync/blackboxWorker.run in blackbox.go. + platform.SetJobBoundaryHook(func(string) { + requestBlackboxSync(DefaultExportDir) + }) return a } diff --git a/audit/internal/app/blackbox.go b/audit/internal/app/blackbox.go index 200820b..5ca6971 100644 --- a/audit/internal/app/blackbox.go +++ b/audit/internal/app/blackbox.go @@ -26,8 +26,43 @@ const ( blackboxMinFlushPeriod = 1 * time.Second blackboxMaxFlushPeriod = 30 * time.Second blackboxRecoveryFastCount = 5 + + // blackboxKickFileName is a marker under exportDir that SAT job + // execution touches after each job finishes (see platform.SetJobBoundaryHook + // wiring in New()). Workers poll its mtime so a just-finished job's + // output reaches removable media promptly instead of waiting out + // whatever the current adaptive flush period happens to be. + blackboxKickFileName = ".blackbox-kick" ) +// blackboxKickPollInterval is how often an idle worker checks the kick file +// for a pending out-of-band sync request. A package var (not a const) so +// tests can shrink it instead of waiting out the real interval. +var blackboxKickPollInterval = 250 * time.Millisecond + +// requestBlackboxSync touches the kick file under exportDir, signalling any +// running blackbox worker to sync on its next poll instead of waiting out +// its current flush period. Best-effort: a failure here just means the next +// scheduled sync picks up the data instead of an early one. +func requestBlackboxSync(exportDir string) { + path := filepath.Join(exportDir, blackboxKickFileName) + now := blackboxNow() + if err := os.Chtimes(path, now, now); err == nil { + return + } + _ = os.WriteFile(path, []byte(now.Format(time.RFC3339Nano)+"\n"), 0644) +} + +// blackboxKickModTime returns the kick file's mtime, or the zero Time if it +// doesn't exist yet (nothing has requested a sync since boot). +func blackboxKickModTime(exportDir string) time.Time { + info, err := os.Stat(filepath.Join(exportDir, blackboxKickFileName)) + if err != nil { + return time.Time{} + } + return info.ModTime() +} + var DefaultBlackboxStatePath = DefaultExportDir + "/blackbox-state.json" var ( @@ -96,8 +131,14 @@ type blackboxWorker struct { flushPeriod time.Duration lastError string fastCycles int + lastKickSeen time.Time stopCh chan struct{} stoppedCh chan struct{} + + // syncCycleFunc defaults to w.syncCycle; overridable in tests so run()'s + // wait/wake-on-kick logic can be exercised without a real removable-media + // mount and copy. + syncCycleFunc func() error } func RunBlackbox(ctx context.Context, exportDir, statePath string, system *platform.System) error { @@ -309,7 +350,7 @@ func (rt *blackboxRuntime) discoverMarkedTargets() ([]discoveredBlackboxTarget, } func newBlackboxWorker(rt *blackboxRuntime, found discoveredBlackboxTarget) *blackboxWorker { - return &blackboxWorker{ + w := &blackboxWorker{ runtime: rt, enrollmentID: found.marker.EnrollmentID, target: found.target, @@ -319,24 +360,39 @@ func newBlackboxWorker(rt *blackboxRuntime, found discoveredBlackboxTarget) *bla stopCh: make(chan struct{}), stoppedCh: make(chan struct{}), } + w.syncCycleFunc = w.syncCycle + return w } func (w *blackboxWorker) run() { defer close(w.stoppedCh) + kickPoll := time.NewTicker(blackboxKickPollInterval) + defer kickPoll.Stop() + for { start := time.Now() - err := w.syncCycle() + err := w.syncCycleFunc() duration := time.Since(start) w.finishCycle(duration, err) + w.lastKickSeen = blackboxKickModTime(w.runtime.exportDir) wait := w.currentFlushPeriod() timer := time.NewTimer(wait) - select { - case <-w.stopCh: - timer.Stop() - w.cleanup() - return - case <-timer.C: + waitLoop: + for { + select { + case <-w.stopCh: + timer.Stop() + w.cleanup() + return + case <-timer.C: + break waitLoop + case <-kickPoll.C: + if mtime := blackboxKickModTime(w.runtime.exportDir); mtime.After(w.lastKickSeen) { + timer.Stop() + break waitLoop + } + } } } } diff --git a/audit/internal/app/blackbox_kick_test.go b/audit/internal/app/blackbox_kick_test.go new file mode 100644 index 0000000..e76172b --- /dev/null +++ b/audit/internal/app/blackbox_kick_test.go @@ -0,0 +1,136 @@ +package app + +import ( + "path/filepath" + "sync" + "sync/atomic" + "testing" + "time" +) + +func TestRequestBlackboxSyncUpdatesKickModTime(t *testing.T) { + dir := t.TempDir() + + before := blackboxKickModTime(dir) + if !before.IsZero() { + t.Fatalf("blackboxKickModTime before any request = %v, want zero", before) + } + + requestBlackboxSync(dir) + first := blackboxKickModTime(dir) + if first.IsZero() { + t.Fatalf("blackboxKickModTime after request is still zero") + } + + time.Sleep(5 * time.Millisecond) + requestBlackboxSync(dir) + second := blackboxKickModTime(dir) + if !second.After(first) { + t.Fatalf("second kick mtime %v did not advance past first %v", second, first) + } +} + +func newTestBlackboxWorker(t *testing.T, exportDir string) (*blackboxWorker, *int32) { + t.Helper() + rt := &blackboxRuntime{ + exportDir: exportDir, + statePath: filepath.Join(t.TempDir(), "blackbox-state.json"), + bootFolder: "boot-folder", + workers: make(map[string]*blackboxWorker), + } + w := &blackboxWorker{ + runtime: rt, + flushPeriod: blackboxMinFlushPeriod, + status: "running", + stopCh: make(chan struct{}), + stoppedCh: make(chan struct{}), + } + var calls int32 + w.syncCycleFunc = func() error { + atomic.AddInt32(&calls, 1) + return nil + } + return w, &calls +} + +func TestBlackboxWorkerWakesEarlyOnKick(t *testing.T) { + exportDir := t.TempDir() + old := blackboxKickPollInterval + blackboxKickPollInterval = 5 * time.Millisecond + t.Cleanup(func() { blackboxKickPollInterval = old }) + + w, calls := newTestBlackboxWorker(t, exportDir) + // A long flush period: without the kick mechanism, a second sync + // wouldn't happen within the test's timeout. + w.flushPeriod = 10 * time.Second + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + w.run() + }() + + // Let the first (immediate) cycle happen. + deadline := time.After(2 * time.Second) + for atomic.LoadInt32(calls) < 1 { + select { + case <-deadline: + t.Fatal("timed out waiting for first sync cycle") + case <-time.After(time.Millisecond): + } + } + + // Keep requesting a sync until a second cycle happens. A single kick can + // legitimately land in the brief window where the worker is still + // recording its post-cycle lastKickSeen baseline and get folded into it + // harmlessly (the point being tested — that *a* kick wakes the worker + // well before the 10s flush period — still holds as long as one of these + // repeated kicks lands after that baseline is recorded). + deadline = time.After(2 * time.Second) + for atomic.LoadInt32(calls) < 2 { + requestBlackboxSync(exportDir) + select { + case <-deadline: + t.Fatalf("timed out waiting for kick-triggered sync cycle; calls=%d", atomic.LoadInt32(calls)) + case <-time.After(5 * time.Millisecond): + } + } + + w.stop() + wg.Wait() +} + +func TestBlackboxWorkerStopsCleanly(t *testing.T) { + exportDir := t.TempDir() + w, calls := newTestBlackboxWorker(t, exportDir) + w.flushPeriod = time.Hour + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + w.run() + }() + + deadline := time.After(2 * time.Second) + for atomic.LoadInt32(calls) < 1 { + select { + case <-deadline: + t.Fatal("timed out waiting for first sync cycle") + case <-time.After(time.Millisecond): + } + } + + done := make(chan struct{}) + go func() { + w.stop() + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("worker.stop() did not return in time") + } + wg.Wait() +} diff --git a/audit/internal/platform/sat.go b/audit/internal/platform/sat.go index d36a46e..9609d2a 100644 --- a/audit/internal/platform/sat.go +++ b/audit/internal/platform/sat.go @@ -70,6 +70,19 @@ const ( SATEstimatedNvidiaBandwidthSec = 2700 ) +// satJobBoundaryHook, if set, is called with a SAT job's log file name right +// after that job's output has been written to disk — a natural point for an +// external blackbox sync to pick up newly-completed data promptly instead of +// waiting out its own adaptive schedule. Nil by default (no-op); set once +// via SetJobBoundaryHook by whichever process wires up blackbox. +var satJobBoundaryHook func(jobName string) + +// SetJobBoundaryHook installs the callback invoked after each SAT job +// finishes and its log file has been written. Pass nil to clear it. +func SetJobBoundaryHook(hook func(jobName string)) { + satJobBoundaryHook = hook +} + var ( satExecCommand = exec.Command satLookPath = exec.LookPath @@ -98,21 +111,39 @@ var ( ) // streamExecOutput runs cmd and streams each output line to logFunc (if non-nil). +// If livePath is non-empty, each line is also appended to that file as it +// arrives — so the command's own output already exists on disk (and is thus +// pickable up by a concurrent blackbox sync) while it's still running, +// instead of only appearing once the whole pack finishes writing the final +// job file. Best-effort: a failure to open/write livePath never fails the job. // Returns combined stdout+stderr as a byte slice. -func streamExecOutput(cmd *exec.Cmd, logFunc func(string)) ([]byte, error) { +func streamExecOutput(cmd *exec.Cmd, logFunc func(string), livePath string) ([]byte, error) { pr, pw := io.Pipe() cmd.Stdout = pw cmd.Stderr = pw + var liveFile *os.File + if livePath != "" { + if f, err := os.OpenFile(livePath, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0644); err == nil { + liveFile = f + } + } + var buf bytes.Buffer var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() + if liveFile != nil { + defer liveFile.Close() + } scanner := bufio.NewScanner(pr) for scanner.Scan() { line := scanner.Text() buf.WriteString(line + "\n") + if liveFile != nil { + _, _ = liveFile.WriteString(line + "\n") + } if logFunc != nil { logFunc(line) } @@ -787,11 +818,15 @@ func (s *System) RunStorageAcceptancePack(ctx context.Context, baseDir string, e break } name := fmt.Sprintf("%s-%02d-%s.log", prefix, cmdIndex+1, job.name) - out, err := runSATCommandCtx(ctx, verboseLog, job.name, job.cmd, nil, logFunc) + livePath := filepath.Join(runDir, name) + out, err := runSATCommandCtx(ctx, verboseLog, job.name, job.cmd, nil, logFunc, livePath) deviceOutputs[job.name] = out - if writeErr := os.WriteFile(filepath.Join(runDir, name), out, 0644); writeErr != nil { + if writeErr := os.WriteFile(livePath, out, 0644); writeErr != nil { return "", writeErr } + if satJobBoundaryHook != nil { + satJobBoundaryHook(name) + } status, rc := classifySATResult(job.name, out, err) stats.Add(status) key := filepath.Base(devPath) + "_" + strings.ReplaceAll(job.name, "-", "_") @@ -985,7 +1020,7 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa if job.collectGPU { out, err = runSATCommandWithMetrics(ctx, verboseLog, job.name, cmd, job.env, job.gpuIndices, runDir, logFunc) } else { - out, err = runSATCommandCtx(ctx, verboseLog, job.name, cmd, job.env, logFunc) + out, err = runSATCommandCtx(ctx, verboseLog, job.name, cmd, job.env, logFunc, filepath.Join(runDir, job.name)) } if err == nil || attempt >= job.retries || ctx.Err() != nil { break @@ -1015,6 +1050,9 @@ func runAcceptancePackCtx(ctx context.Context, baseDir, prefix string, jobs []sa if writeErr := os.WriteFile(filepath.Join(runDir, job.name), out, 0644); writeErr != nil { return "", writeErr } + if satJobBoundaryHook != nil { + satJobBoundaryHook(job.name) + } if ctx.Err() != nil { return "", ctx.Err() } @@ -1237,7 +1275,11 @@ func parseNvidiaGPUHealth(raw string) []nvidiaGPUHealth { return gpus } -func runSATCommandCtx(ctx context.Context, verboseLog, name string, cmd []string, env []string, logFunc func(string)) ([]byte, error) { +// 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 +// mid-command leaves whatever had printed so far instead of nothing at all. +func runSATCommandCtx(ctx context.Context, verboseLog, name string, cmd []string, env []string, logFunc func(string), livePath ...string) ([]byte, error) { start := time.Now().UTC() resolvedCmd, err := resolveSATCommand(cmd) appendSATVerboseLog(verboseLog, @@ -1268,7 +1310,11 @@ func runSATCommandCtx(ctx context.Context, verboseLog, name string, cmd []string if len(env) > 0 { c.Env = append(os.Environ(), env...) } - out, err := streamExecOutput(c, logFunc) + var live string + if len(livePath) > 0 { + live = livePath[0] + } + out, err := streamExecOutput(c, logFunc, live) rc := 0 if err != nil { @@ -1437,7 +1483,7 @@ func runSATCommand(verboseLog, name string, cmd []string, logFunc func(string)) return []byte(err.Error() + "\n"), err } - out, err := streamExecOutput(satExecCommand(resolvedCmd[0], resolvedCmd[1:]...), logFunc) + out, err := streamExecOutput(satExecCommand(resolvedCmd[0], resolvedCmd[1:]...), logFunc, "") rc := 0 if err != nil { @@ -1607,7 +1653,7 @@ func runSATCommandWithMetrics(ctx context.Context, verboseLog, name string, cmd } }() - out, err := runSATCommandCtx(ctx, verboseLog, name, cmd, env, logFunc) + out, err := runSATCommandCtx(ctx, verboseLog, name, cmd, env, logFunc, filepath.Join(runDir, name)) close(stopCh) <-doneCh diff --git a/audit/internal/platform/sat_live_output_test.go b/audit/internal/platform/sat_live_output_test.go new file mode 100644 index 0000000..3ad91ea --- /dev/null +++ b/audit/internal/platform/sat_live_output_test.go @@ -0,0 +1,97 @@ +package platform + +import ( + "context" + "os" + "os/exec" + "path/filepath" + "testing" +) + +func TestStreamExecOutputWritesLiveFileIncrementally(t *testing.T) { + dir := t.TempDir() + livePath := filepath.Join(dir, "job.log") + + cmd := satExecCommand("printf", "line1\nline2\n") + out, err := streamExecOutput(cmd, nil, livePath) + if err != nil { + t.Fatalf("streamExecOutput error: %v", err) + } + if string(out) != "line1\nline2\n" { + t.Fatalf("out=%q want %q", out, "line1\nline2\n") + } + got, err := os.ReadFile(livePath) + if err != nil { + t.Fatalf("ReadFile(livePath): %v", err) + } + if string(got) != "line1\nline2\n" { + t.Fatalf("livePath content=%q want %q", got, "line1\nline2\n") + } +} + +func TestStreamExecOutputSkipsLiveFileWhenPathEmpty(t *testing.T) { + cmd := satExecCommand("printf", "line1\n") + out, err := streamExecOutput(cmd, nil, "") + if err != nil { + t.Fatalf("streamExecOutput error: %v", err) + } + if string(out) != "line1\n" { + t.Fatalf("out=%q want %q", out, "line1\n") + } +} + +func TestRunSATCommandCtxWritesLivePath(t *testing.T) { + dir := t.TempDir() + verboseLog := filepath.Join(dir, "verbose.log") + livePath := filepath.Join(dir, "job.log") + + oldExecCommand := satExecCommand + satExecCommand = func(name string, args ...string) *exec.Cmd { + return exec.Command(name, args...) + } + t.Cleanup(func() { satExecCommand = oldExecCommand }) + + out, err := runSATCommandCtx(context.Background(), verboseLog, "job", []string{"printf", "hello\n"}, nil, nil, livePath) + if err != nil { + t.Fatalf("runSATCommandCtx error: %v", err) + } + if string(out) != "hello\n" { + t.Fatalf("out=%q want %q", out, "hello\n") + } + got, err := os.ReadFile(livePath) + if err != nil { + t.Fatalf("ReadFile(livePath): %v", err) + } + if string(got) != "hello\n" { + t.Fatalf("livePath content=%q want %q", got, "hello\n") + } +} + +func TestRunAcceptancePackCtxInvokesJobBoundaryHook(t *testing.T) { + old := satJobBoundaryHook + t.Cleanup(func() { satJobBoundaryHook = old }) + + var seen []string + SetJobBoundaryHook(func(jobName string) { + seen = append(seen, jobName) + }) + + oldExecCommand := satExecCommand + satExecCommand = func(name string, args ...string) *exec.Cmd { + return exec.Command(name, args...) + } + t.Cleanup(func() { satExecCommand = oldExecCommand }) + + dir := t.TempDir() + _, err := runAcceptancePackCtx(context.Background(), dir, "test-pack", []satJob{ + {name: "01-a.log", cmd: []string{"printf", "a\n"}}, + {name: "02-b.log", cmd: []string{"printf", "b\n"}}, + }, nil) + if err != nil { + t.Fatalf("runAcceptancePackCtx error: %v", err) + } + + if len(seen) != 2 || seen[0] != "01-a.log" || seen[1] != "02-b.log" { + t.Fatalf("seen=%v want [01-a.log 02-b.log]", seen) + } +}