diff --git a/integration/agentcompat/internal/process/proc.go b/integration/agentcompat/internal/process/proc.go index 356c57fc..6083d057 100644 --- a/integration/agentcompat/internal/process/proc.go +++ b/integration/agentcompat/internal/process/proc.go @@ -103,14 +103,23 @@ func readParentPID(pid int) (int, error) { return parentPID, nil } -func processFDs(pid int) (int, map[uint64]struct{}, error) { +type FDObservation struct { + Number int + Target string +} + +func processFDs(pid int, captureObservations bool) (int, map[uint64]struct{}, []FDObservation, error) { directory := filepath.Join("/proc", strconv.Itoa(pid), "fd") entries, err := os.ReadDir(directory) if err != nil { - return 0, nil, err + return 0, nil, nil, err } count := 0 sockets := make(map[uint64]struct{}) + var observations []FDObservation + if captureObservations { + observations = make([]FDObservation, 0, len(entries)) + } for _, entry := range entries { descriptor, err := strconv.Atoi(entry.Name()) if err != nil || descriptor < 3 { @@ -121,14 +130,22 @@ func processFDs(pid int) (int, map[uint64]struct{}, error) { if os.IsNotExist(err) { continue } - return 0, nil, err + return 0, nil, nil, err } count++ + if captureObservations { + observations = append(observations, FDObservation{Number: descriptor, Target: target}) + } if inode, exists := parseSocketInode(target); exists { sockets[inode] = struct{}{} } } - return count, sockets, nil + if captureObservations { + sort.Slice(observations, func(left, right int) bool { + return observations[left].Number < observations[right].Number || (observations[left].Number == observations[right].Number && observations[left].Target < observations[right].Target) + }) + } + return count, sockets, observations, nil } func parseSocketInode(target string) (uint64, bool) { diff --git a/integration/agentcompat/internal/process/sampler.go b/integration/agentcompat/internal/process/sampler.go index 1c24de43..e644d2ca 100644 --- a/integration/agentcompat/internal/process/sampler.go +++ b/integration/agentcompat/internal/process/sampler.go @@ -13,13 +13,15 @@ import ( ) type Sample struct { - PID int `json:"pid"` - RSSBytes uint64 `json:"rss_bytes"` - DescendantPIDs []int `json:"descendant_pids"` - DescendantCount int `json:"descendant_count"` - NonStdioFDCount int `json:"non_stdio_fd_count"` - TCPListenerCount int `json:"tcp_listener_count"` - TCP6ListenerCount int `json:"tcp6_listener_count"` + PID int `json:"pid"` + RSSBytes uint64 `json:"rss_bytes"` + DescendantPIDs []int `json:"descendant_pids"` + DescendantCount int `json:"descendant_count"` + NonStdioFDCount int `json:"non_stdio_fd_count"` + TCPListenerCount int `json:"tcp_listener_count"` + TCP6ListenerCount int `json:"tcp6_listener_count"` + FDObservations []FDObservation `json:"-"` + SampledAt time.Time `json:"-"` } type Window struct { @@ -28,13 +30,22 @@ type Window struct { } type WindowSpec struct { - PID int - Interval time.Duration - AllowTerminated bool - ObserveSample func(context.Context, Sample) error + PID int + Interval time.Duration + AllowTerminated bool + CaptureFDObservations bool + ObserveSample func(context.Context, Sample) error } func SampleProcess(pid int) (Sample, error) { + return sampleProcess(pid, false) +} + +func SampleProcessWithFDObservations(pid int) (Sample, error) { + return sampleProcess(pid, true) +} + +func sampleProcess(pid int, captureFDObservations bool) (Sample, error) { rssBytes, err := readRSSBytes(pid) if err != nil { return Sample{}, err @@ -43,7 +54,7 @@ func SampleProcess(pid int) (Sample, error) { if err != nil { return Sample{}, err } - fdCount, socketInodes, err := processFDs(pid) + fdCount, socketInodes, fdObservations, err := processFDs(pid, captureFDObservations) if err != nil { return Sample{}, err } @@ -63,6 +74,8 @@ func SampleProcess(pid int) (Sample, error) { NonStdioFDCount: fdCount, TCPListenerCount: intersectionCount(socketInodes, tcpListeners), TCP6ListenerCount: intersectionCount(socketInodes, tcp6Listeners), + FDObservations: fdObservations, + SampledAt: time.Now(), }, nil } @@ -81,7 +94,7 @@ func SampleWindow(ctx context.Context, spec WindowSpec) (Window, error) { return Window{}, ctx.Err() } } - sample, err := SampleProcess(spec.PID) + sample, err := sampleProcess(spec.PID, spec.CaptureFDObservations) if err != nil { if spec.AllowTerminated && os.IsNotExist(err) { return window, nil diff --git a/integration/agentcompat/internal/process/sampler_fd_observations_test.go b/integration/agentcompat/internal/process/sampler_fd_observations_test.go new file mode 100644 index 00000000..071bb9c7 --- /dev/null +++ b/integration/agentcompat/internal/process/sampler_fd_observations_test.go @@ -0,0 +1,45 @@ +//go:build linux + +package process + +import ( + "context" + "encoding/json" + "os" + "testing" + "time" +) + +func TestSampleProcessWithFDObservations_CapturesEverySuccessfulNonStdioReadlink(t *testing.T) { + // Given / When + sample, err := SampleProcessWithFDObservations(os.Getpid()) + + // Then + requireNoError(t, err) + if len(sample.FDObservations) != sample.NonStdioFDCount { + t.Fatalf("FD observations = %d, non-stdio FD count = %d", len(sample.FDObservations), sample.NonStdioFDCount) + } + if sample.SampledAt.IsZero() { + t.Fatal("sampled at is zero") + } + encoded, err := json.Marshal(sample) + requireNoError(t, err) + var evidence map[string]json.RawMessage + requireNoError(t, json.Unmarshal(encoded, &evidence)) + if _, exists := evidence["sampled_at"]; exists { + t.Fatalf("sampled_at leaked into evidence JSON: %s", encoded) + } +} + +func TestSampleWindowWithFDObservations_RecordsCompletionTimeForEverySample(t *testing.T) { + // Given / When + window, err := SampleWindow(context.Background(), WindowSpec{PID: os.Getpid(), Interval: time.Nanosecond, CaptureFDObservations: true}) + + // Then + requireNoError(t, err) + for index, sample := range window.Samples { + if sample.SampledAt.IsZero() { + t.Fatalf("sample %d sampled at is zero", index+1) + } + } +}