From 11eb339540b389ec4aa0065ffd4a62009091251c Mon Sep 17 00:00:00 2001 From: naiba Date: Mon, 20 Jul 2026 22:10:29 +0000 Subject: [PATCH] test(agentcompat): exercise real FD lifecycle diagnostics Co-authored-by: naiba/CloudCode --- ...fd_diagnostic_process_helper_fault_test.go | 38 +++ ..._fd_diagnostic_process_integration_test.go | 228 ++++++++++++++++++ 2 files changed, 266 insertions(+) create mode 100644 integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_helper_fault_test.go create mode 100644 integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_integration_test.go diff --git a/integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_helper_fault_test.go b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_helper_fault_test.go new file mode 100644 index 00000000..04ff7ec0 --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_helper_fault_test.go @@ -0,0 +1,38 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestFDDiagnosticShell_NextLineReturnsDeadlineWhenOutputStalls(t *testing.T) { + // Given + ctx, cancel := context.WithTimeout(t.Context(), 100*time.Millisecond) + defer cancel() + child := startFDDiagnosticShell(t, t.Context(), "") + defer func() { require.NoError(t, child.close()) }() + + // When + _, err := child.nextLine(ctx) + + // Then + require.ErrorIs(t, err, context.DeadlineExceeded) +} + +func TestFDDiagnosticShell_CloseReapsChildWhenExitWriteFails(t *testing.T) { + // Given + child := startFDDiagnosticShell(t, t.Context(), "") + require.NoError(t, child.input.Close()) + + // When + err := child.close() + + // Then + require.Error(t, err) + require.NotNil(t, child.command.ProcessState) +} diff --git a/integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_integration_test.go b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_integration_test.go new file mode 100644 index 00000000..d657d8ed --- /dev/null +++ b/integration/agentcompat/internal/scenario/stress_fd_diagnostic_process_integration_test.go @@ -0,0 +1,228 @@ +//go:build linux && agentcompat + +package scenario + +import ( + "bufio" + "context" + "errors" + "fmt" + "io" + "os" + "os/exec" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/nezhahq/nezha/integration/agentcompat/internal/agent" + processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process" +) + +const fdDiagnosticShellTargetEnv = "NEZHA_AGENTCOMPAT_FD_DIAGNOSTIC_TARGET" + +func TestFDDiagnosticCollector_TracksChildDescriptorLifecycle(t *testing.T) { + // Given + ctx, cancel := context.WithTimeout(t.Context(), time.Second) + defer cancel() + target := filepath.Join(t.TempDir(), "fd-diagnostic-target") + require.NoError(t, os.WriteFile(target, []byte("target"), 0o600)) + child := startFDDiagnosticShell(t, ctx, target) + defer func() { require.NoError(t, child.close()) }() + identity := agent.ProcessIdentity{Generation: 1, PID: child.command.Process.Pid} + ordinal, err := NewStressAgentOrdinal(1) + require.NoError(t, err) + process, err := NewStressAgentProcess(ordinal, identity.PID) + require.NoError(t, err) + tail := fdDiagnosticChildTailControl{secondStarted: make(chan struct{}), releaseSecond: make(chan struct{})} + collector := newFDDiagnosticCollector(fdDiagnosticCollectorSpec{Enabled: true, TailResultCapacity: 8, TailInterval: 0, Sample: tail.sampler()}) + collector.RecordBaseline(fdDiagnosticChildWindow(t, ctx, process, identity)) + require.NoError(t, child.send("exec 3<\"$"+fdDiagnosticShellTargetEnv+"\"; echo opened")) + opened, err := child.nextLine(ctx) + require.NoError(t, err) + require.Equal(t, "opened", opened) + end := fdDiagnosticChildWindow(t, ctx, process, identity) + + // When + collector.RecordEnd(ctx, end) + tail.waitSecondStart(t, ctx) + require.NoError(t, child.send("exec 3<&-; echo closed")) + closed, err := child.nextLine(ctx) + require.NoError(t, err) + require.Equal(t, "closed", closed) + close(tail.releaseSecond) + records := collector.WaitRecords() + + // Then + require.Len(t, records, 1) + record := records[0] + added := fdDiagnosticObservationTarget(record.AddedFinal, target) + require.NotNil(t, added) + require.Equal(t, "cleared_during_tail", fdDiagnosticLifecycleStatus(record.Lifecycle, *added)) + require.NotNil(t, fdDiagnosticObservationTarget(record.Tail[0].FDObservations, target)) + for _, sample := range record.Tail[1:] { + require.Nil(t, fdDiagnosticObservationTarget(sample.FDObservations, target)) + } + for _, window := range []*fdDiagnosticWindow{record.Baseline, record.End} { + for _, sample := range window.Samples { + require.False(t, sample.ObservedAt.IsZero()) + } + } + for _, sample := range record.Tail { + require.False(t, sample.ObservedAt.IsZero()) + } +} + +type fdDiagnosticShell struct { + command *exec.Cmd + input io.WriteCloser + cancel context.CancelFunc + lines <-chan fdDiagnosticShellLine + waitOnce sync.Once + waitErr error +} + +type fdDiagnosticShellLine struct { + text string + err error +} + +func startFDDiagnosticShell(t *testing.T, ctx context.Context, target string) *fdDiagnosticShell { + t.Helper() + processCtx, cancel := context.WithCancel(ctx) + command := exec.CommandContext(processCtx, "/bin/sh") + command.Env = append(os.Environ(), fdDiagnosticShellTargetEnv+"="+target) + input, err := command.StdinPipe() + require.NoError(t, err) + output, err := command.StdoutPipe() + require.NoError(t, err) + require.NoError(t, command.Start()) + lines := make(chan fdDiagnosticShellLine) + go collectFDDiagnosticShellLines(processCtx, output, lines) + return &fdDiagnosticShell{command: command, input: input, cancel: cancel, lines: lines} +} + +func collectFDDiagnosticShellLines(ctx context.Context, output io.Reader, lines chan<- fdDiagnosticShellLine) { + defer close(lines) + scanner := bufio.NewScanner(output) + for scanner.Scan() { + select { + case lines <- fdDiagnosticShellLine{text: scanner.Text()}: + case <-ctx.Done(): + return + } + } + if err := scanner.Err(); err != nil { + select { + case lines <- fdDiagnosticShellLine{err: err}: + case <-ctx.Done(): + } + } +} + +func (child *fdDiagnosticShell) close() error { + sendErr := child.send("exit") + var killErr error + if sendErr != nil { + killErr = child.command.Process.Kill() + } + waitErr := child.wait() + if sendErr == nil { + return waitErr + } + if errors.Is(killErr, os.ErrProcessDone) { + killErr = nil + } + if killErr == nil { + return sendErr + } + return errors.Join(sendErr, killErr, waitErr) +} + +func (child *fdDiagnosticShell) wait() error { + child.waitOnce.Do(func() { + child.waitErr = child.command.Wait() + child.cancel() + }) + return child.waitErr +} + +func (child *fdDiagnosticShell) send(command string) error { + _, err := fmt.Fprintln(child.input, command) + return err +} + +func (child *fdDiagnosticShell) nextLine(ctx context.Context) (string, error) { + if err := ctx.Err(); err != nil { + return "", err + } + select { + case line, open := <-child.lines: + if err := ctx.Err(); err != nil { + return "", err + } + if !open { + return "", io.EOF + } + return line.text, line.err + case <-ctx.Done(): + return "", ctx.Err() + } +} + +type fdDiagnosticChildTailControl struct { + secondStarted chan struct{} + releaseSecond chan struct{} +} + +func (control fdDiagnosticChildTailControl) sampler() fdDiagnosticSampleFunc { + calls := 0 + return func(ctx context.Context, pid int) (processharness.Sample, error) { + calls++ + if calls == 2 { + close(control.secondStarted) + select { + case <-control.releaseSecond: + case <-ctx.Done(): + return processharness.Sample{}, ctx.Err() + } + } + return processharness.SampleProcessWithFDObservations(pid) + } +} + +func (control fdDiagnosticChildTailControl) waitSecondStart(t *testing.T, ctx context.Context) { + t.Helper() + select { + case <-control.secondStarted: + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } +} + +func fdDiagnosticChildWindow(t *testing.T, ctx context.Context, process StressProcessIdentity, identity agent.ProcessIdentity) fdDiagnosticAgentWindow { + t.Helper() + window, err := processharness.SampleWindow(ctx, processharness.WindowSpec{PID: identity.PID, Interval: time.Nanosecond, CaptureFDObservations: true}) + require.NoError(t, err) + return fdDiagnosticAgentWindow{Process: process, Identity: identity, Window: window} +} + +func fdDiagnosticObservationTarget(observations []processharness.FDObservation, target string) *processharness.FDObservation { + for _, observation := range observations { + if observation.Target == target { + return &observation + } + } + return nil +} + +func fdDiagnosticLifecycleStatus(lifecycle []fdDiagnosticLifecycle, observation processharness.FDObservation) string { + for _, entry := range lifecycle { + if entry.Observation == observation { + return entry.Status + } + } + return "" +}