test(agentcompat): capture FD observations in process samples

Co-authored-by: naiba/CloudCode <hi+cloudcode@nai.ba>

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) {

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

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)
+		}
+	}
+}