filestream: encode bridging raw header only for dedup winners

filebeat/input/filestream/fswatch.go

@@ -610,14 +610,9 @@ type fileScanner struct {
 	hasher           hash.Hash
 	readBuffer       []byte
 	compression      string
-	// completedFingerprints holds the paths whose fingerprint was already a
-	// final SHA-256 on the previous watch-loop scan (growing mode only). The
-	// bridging raw header is only useful on the scan a file crosses the
-	// threshold, so for paths in this set toFileDescriptor skips recomputing it.
-	// GetFiles itself is pure with respect to this set: only fileWatcher.watch
-	// advances it (after each scan), so the enumeration-only scans the
-	// prospector runs in Init/TakeOver cannot suppress the header a
-	// still-growing entry needs to migrate across a restart.
+	// completedFingerprints holds paths already complete on the previous watch scan
+	// (growing mode), so attachBridgingRaw can skip re-encoding their bridging header.
+	// Only fileWatcher.watch advances it, so prospector enumeration can't wrongly suppress it.
 	completedFingerprints map[string]struct{}
 
 	// lastCount is the number of unique files the previous scan produced.
@@ -772,6 +767,7 @@ func (s *fileScanner) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.F
 				continue
 			}
 			uniqueIDs[fileID] = fd.Filename
+			s.attachBridgingRaw(&fd)
 			fdByName[filename] = fd
 			if isFileIgnored(fd, opts) {
 				scanMetrics.FilesIgnored++
@@ -863,10 +859,8 @@ func (s *fileScanner) getIngestTarget(filename string) (it ingestTarget, err err
 //   - dataSize <= offset: file is too small to read anything from offset;
 //     return errFileTooSmall.
 //   - dataSize >= offset+length: read bytes[offset:offset+length] and hash
-//     with SHA-256 (FingerprintID.Sum, so Complete() is true). In growing mode
-//     the raw header bytes are also carried in FingerprintID.Raw so the one-time
-//     crossing to the SHA-256 identity can be prefix-matched against a still
-//     growing predecessor.
+//     with SHA-256 (FingerprintID.Sum, so Complete() is true). The growing-mode
+//     bridging raw header is added later by attachBridgingRaw, not here.
 //   - dataSize in (offset, offset+length) under growing mode: read
 //     bytes[offset:dataSize] and carry its hex as FingerprintID.Raw, leaving
 //     Sum empty so Complete() is false.
@@ -1016,23 +1010,20 @@ func (s *fileScanner) toFileDescriptor(it *ingestTarget) (fd loginp.FileDescript
 		Sum: hex.EncodeToString(s.hasher.Sum(nil)),
 	}
 
-	// In growing mode the raw header is carried alongside the SHA-256 so the
-	// one-time transition to the final identity can be prefix-matched against a
-	// still-growing predecessor (in-place growth or rename+grow). It is only
-	// needed on the scan a file crosses the threshold: a path the watch loop
-	// already saw complete on its previous scan has no growing predecessor left
-	// to bridge, so recomputing the ~2*length-byte hex header every scan would
-	// be wasted work. New and just-crossed paths are absent from
-	// completedFingerprints and get the bridging header.
-	if s.cfg.Fingerprint.Growing {
-		if _, done := s.completedFingerprints[it.filename]; !done {
-			fd.Fingerprint.Raw = hex.EncodeToString(s.readBuffer[:length])
-		}
-	}
-
 	return fd, nil
 }
 
+// attachBridgingRaw sets a complete descriptor's raw header.
+func (s *fileScanner) attachBridgingRaw(fd *loginp.FileDescriptor) {
+	if !s.cfg.Fingerprint.Growing || !fd.Fingerprint.Complete() {
+		return
+	}
+	if _, done := s.completedFingerprints[fd.Filename]; done {
+		return
+	}
+	fd.Fingerprint.Raw = hex.EncodeToString(s.readBuffer[:s.cfg.Fingerprint.Length])
+}
+
 func (s *fileScanner) isFileExcluded(file string) bool {
 	return len(s.cfg.ExcludedFiles) > 0 && s.matchAny(s.cfg.ExcludedFiles, file)
 }

filebeat/input/filestream/fswatch_test.go

@@ -1569,6 +1569,18 @@ func writeBenchmarkFiles(tb testing.TB, dir string, n int) []string {
 	return []string{filepath.Join(dir, "*.log")}
 }
 
+// writeBenchmarkDuplicateFiles writes n identical-content files, so every file
+// collides on one fingerprint and GetFiles dedups them to a single winner.
+func writeBenchmarkDuplicateFiles(tb testing.TB, dir string, n int) []string {
+	tb.Helper()
+	content := []byte(strings.Repeat("duplicate-content\n", 1024))
+	for i := range n {
+		filename := filepath.Join(dir, fmt.Sprintf("file-%d.log", i))
+		require.NoError(tb, os.WriteFile(filename, content, 0o644))
+	}
+	return []string{filepath.Join(dir, "*.log")}
+}
+
 func BenchmarkGetFiles(b *testing.B) {
 	paths := writeBenchmarkFiles(b, b.TempDir(), benchmarkFileCount)
 	cfg := fileScannerConfig{
@@ -1618,17 +1630,30 @@ func BenchmarkGetFilesWithFingerprint(b *testing.B) {
 // cost — exactly the work the watch loop elides on stable files by tracking
 // completed paths. It therefore quantifies the per-scan cost that suppression
 // removes after a file's first crossing scan.
+//
+// The "duplicates" case gives growing mode identical-content files: they dedup to
+// one winner, so only that winner encodes the bridging raw header.
 func BenchmarkGetFilesWithFingerprintGrowing(b *testing.B) {
 	cases := []struct {
-		name    string
-		growing bool
+		name       string
+		growing    bool
+		duplicates bool
 	}{
-		{"static", false},
-		{"growing", true},
+		{"static", false, false},
+		{"growing", true, false},
+		{"duplicates", true, true},
 	}
 	for _, tc := range cases {
 		b.Run(tc.name, func(b *testing.B) {
-			paths := writeBenchmarkFiles(b, b.TempDir(), benchmarkFileCount)
+			dir := b.TempDir()
+			var paths []string
+			wantLen := benchmarkFileCount
+			if tc.duplicates {
+				paths = writeBenchmarkDuplicateFiles(b, dir, benchmarkFileCount)
+				wantLen = 1
+			} else {
+				paths = writeBenchmarkFiles(b, dir, benchmarkFileCount)
+			}
 			cfg := fileScannerConfig{
 				Fingerprint: fingerprintConfig{
 					Enabled: true,
@@ -1643,7 +1668,7 @@ func BenchmarkGetFilesWithFingerprintGrowing(b *testing.B) {
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
 				files, _ := s.GetFiles(loginp.FileScanOptions{})
-				require.Len(b, files, benchmarkFileCount)
+				require.Len(b, files, wantLen)
 			}
 		})
 	}
@@ -1892,13 +1917,10 @@ func TestToFileDescriptor_TooSmallFile_NoFileOpen(t *testing.T) {
 //
 //  1. small file (below offset+length) under growing mode → raw-hex
 //     FingerprintID.Raw, Complete=false, no Sum.
-//  2. file grows past threshold → SHA-256 Sum matching the static fingerprint
-//     output for the same bytes, Complete=true, and the raw header carried in
-//     Raw (so the crossing can be prefix-matched against the predecessor).
-//  3. second scan still at threshold via toFileDescriptor → identical
-//     FingerprintID. The per-scan raw-header suppression lives in GetFiles
-//     (exercised by TestGetFiles_GrowingRawSuppression), so a direct
-//     toFileDescriptor call always recomputes Raw.
+//  2. file grows past threshold → SHA-256 Sum, Complete=true, Raw empty;
+//     attachBridgingRaw (mirrored here) then adds the bridging header.
+//  3. second scan at threshold → same Sum and empty Raw; attachBridgingRaw
+//     re-adds the bridging header identically.
 //  4. file truncated back below threshold → raw-hex Raw, Complete=false.
 func TestToFileDescriptor_GrowingLifecycle(t *testing.T) {
 	dir := t.TempDir()
@@ -1956,8 +1978,13 @@ func TestToFileDescriptor_GrowingLifecycle(t *testing.T) {
 	expectedSHA := sha256.Sum256([]byte(strings.Repeat("abcd", 256)[:length]))
 	assert.Equal(t, hex.EncodeToString(expectedSHA[:]), fd2.Fingerprint.Sum,
 		"step 2: Sum must be SHA-256 of bytes[0:length]")
+	assert.Empty(t, fd2.Fingerprint.Raw,
+		"step 2: toFileDescriptor alone leaves Raw empty (bridging header attached post-dedup)")
+
+	// Mirror GetFiles: attach the bridging header to the winner descriptor.
+	s.attachBridgingRaw(&fd2)
 	assert.Equal(t, expectedRawHex, fd2.Fingerprint.Raw,
-		"step 2: Raw must be raw hex of the same bytes for bridging the transition")
+		"step 2: attachBridgingRaw must carry raw hex of the same bytes for bridging the transition")
 
 	// Verify the threshold-transition prefix relationship: the previous raw-hex
 	// (200 bytes) is a prefix of the current completed Raw (1024 bytes). This is
@@ -1972,9 +1999,16 @@ func TestToFileDescriptor_GrowingLifecycle(t *testing.T) {
 	fd3, err := s.toFileDescriptor(&it)
 	require.NoError(t, err, "toFileDescriptor failed")
 
+	assert.True(t, fd3.Fingerprint.Complete(), "step 3: still Complete at threshold")
+	assert.Equal(t, fd2.Fingerprint.Sum, fd3.Fingerprint.Sum,
+		"step 3: Sum is stable across scans")
+	assert.Empty(t, fd3.Fingerprint.Raw,
+		"step 3: toFileDescriptor alone leaves Raw empty; the bridging header is attached post-dedup")
+
+	// After attaching the bridging header the descriptor matches the crossing scan.
+	s.attachBridgingRaw(&fd3)
 	assert.Equal(t, fd2.Fingerprint, fd3.Fingerprint,
-		"step 3: direct toFileDescriptor recomputes the same FingerprintID "+
-			"(GetFiles-level Raw suppression does not apply here)")
+		"step 3: attachBridgingRaw reproduces the same FingerprintID as the crossing scan")
 
 	// --- Step 4: file truncated back below threshold ---
 	writeFile(t, 100)
@@ -1992,12 +2026,24 @@ func TestToFileDescriptor_GrowingLifecycle(t *testing.T) {
 		"step 4: no SHA-256 Sum while below threshold")
 }
 
-// TestGetFiles_GrowingRawSuppression verifies the optimization that avoids
-// recomputing the bridging raw header for files that are already complete: the
-// header is emitted on the scan a file crosses the threshold (so the transition
-// can still be prefix-matched) and dropped on subsequent scans of the now-stable
-// file. If the file is truncated back below the threshold the suppression is
-// lifted, and a fresh crossing re-emits the header.
+// advanceCompletedFingerprints mirrors fileWatcher.watch: it seeds completedFingerprints
+// from a scan result so the next scan suppresses stable files' bridging headers.
+func advanceCompletedFingerprints(s *fileScanner, files map[string]loginp.FileDescriptor) {
+	completed := map[string]struct{}{}
+	for p, fd := range files {
+		if fd.Fingerprint.Complete() {
+			completed[p] = struct{}{}
+		}
+	}
+	s.completedFingerprints = completed
+}
+
+// TestGetFiles_GrowingRawSuppression verifies that the bridging raw header is not
+// recomputed for files that are already complete: the header is emitted on the scan
+// a file crosses the threshold (so the transition can still be prefix-matched) and
+// dropped on subsequent scans of the now-stable file. If the file is truncated back
+// below the threshold the suppression is lifted, and a fresh crossing re-emits the
+// header.
 //
 // GetFiles is pure with respect to the completedFingerprints set; the watch
 // loop is what advances it. The scan helper below reproduces that contract:
@@ -2029,15 +2075,7 @@ func TestGetFiles_GrowingRawSuppression(t *testing.T) {
 		files, _ := s.GetFiles(loginp.FileScanOptions{})
 		require.Contains(t, files, filename, "file must be scanned")
 		fp := files[filename].Fingerprint
-		// Mirror the watch loop: tell the scanner which paths are now complete
-		// so the next scan can skip recomputing their bridging raw header.
-		completed := map[string]struct{}{}
-		for p, fd := range files {
-			if fd.Fingerprint.Complete() {
-				completed[p] = struct{}{}
-			}
-		}
-		s.completedFingerprints = completed
+		advanceCompletedFingerprints(s, files)
 		return fp
 	}
 
@@ -2076,6 +2114,105 @@ func TestGetFiles_GrowingRawSuppression(t *testing.T) {
 	assert.NotEmpty(t, fp.Raw, "re-crossing must carry the bridging raw header again")
 }
 
+// TestGetFiles_DuplicateFingerprint checks that among files sharing a fingerprint, only
+// the dedup winner carries the bridging raw header, including after the winner is deleted.
+func TestGetFiles_DuplicateFingerprint(t *testing.T) {
+	dir := t.TempDir()
+	const length int64 = 1024
+	const n = 3
+
+	// Identical, above-threshold content: every file collides on the same Sum.
+	content := []byte(strings.Repeat("abcd", 512)) // 2048 bytes >= length
+	for i := range n {
+		fn := filepath.Join(dir, fmt.Sprintf("dup-%d.log", i))
+		require.NoError(t, os.WriteFile(fn, content, 0o644),
+			"could not write duplicate file")
+	}
+
+	cfg := fileScannerConfig{
+		Fingerprint: fingerprintConfig{
+			Enabled: true,
+			Offset:  0,
+			Length:  length,
+			Growing: true,
+		},
+	}
+	s, err := newFileScanner(
+		logp.NewNopLogger(), []string{filepath.Join(dir, "*.log")}, cfg, CompressionNone)
+	require.NoError(t, err, "could not create file scanner")
+
+	expectedRawHex := hex.EncodeToString(content[:length])
+
+	scan := func() map[string]loginp.FileDescriptor {
+		files, _ := s.GetFiles(loginp.FileScanOptions{})
+		advanceCompletedFingerprints(s, files)
+		return files
+	}
+
+	// First scan: duplicates dedup to one winner, which carries the bridging header.
+	files := scan()
+	require.Len(t, files, 1, "duplicate-fingerprint files must dedup to a single winner")
+	var winner string
+	for p, fd := range files {
+		winner = p
+		require.True(t, fd.Fingerprint.Complete(), "winner must be Complete")
+		assert.Equal(t, expectedRawHex, fd.Fingerprint.Raw,
+			"first scan: winner carries the bridging raw header")
+	}
+
+	// Second scan: winner now in completedFingerprints, so its header is suppressed.
+	files = scan()
+	require.Len(t, files, 1, "still a single winner")
+	require.Contains(t, files, winner, "same winner across stable scans")
+	assert.Empty(t, files[winner].Fingerprint.Raw,
+		"second scan: winner's bridging raw header is suppressed once complete")
+
+	// Delete the winner: a promoted loser (never in the completed set) carries the header.
+	require.NoError(t, os.Remove(winner), "could not remove the winner file")
+	files = scan()
+	require.Len(t, files, 1, "still a single winner after deleting the previous one")
+	require.NotContains(t, files, winner, "deleted winner must be gone")
+	for _, fd := range files {
+		require.True(t, fd.Fingerprint.Complete(), "promoted winner must be Complete")
+		assert.Equal(t, expectedRawHex, fd.Fingerprint.Raw,
+			"promoted loser must carry the bridging raw header (never in completed set)")
+	}
+}
+
+// TestGetFiles_DuplicateFingerprintAllocBudget asserts only dedup winners encode the bridging raw
+// header; a stray encode is invisible to a behavior test, so bound the growing-vs-static delta.
+func TestGetFiles_DuplicateFingerprintAllocBudget(t *testing.T) {
+	dir := t.TempDir()
+	glob := writeBenchmarkDuplicateFiles(t, dir, benchmarkFileCount)
+
+	measure := func(growing bool) float64 {
+		cfg := fileScannerConfig{Fingerprint: fingerprintConfig{
+			Enabled: true, Offset: 0, Length: 1024, Growing: growing,
+		}}
+		s, err := newFileScanner(logp.NewNopLogger(), glob, cfg, CompressionNone)
+		require.NoError(t, err, "could not create file scanner")
+
+		files, _ := s.GetFiles(loginp.FileScanOptions{})
+		require.Len(t, files, 1, "duplicates must dedup to a single winner")
+
+		// completedFingerprints is never advanced, so the winner re-encodes on every run.
+		return testing.AllocsPerRun(10, func() {
+			_, _ = s.GetFiles(loginp.FileScanOptions{})
+		})
+	}
+
+	growingAllocs := measure(true)
+	staticAllocs := measure(false)
+	delta := growingAllocs - staticAllocs
+
+	// Budget one alloc per file: far below the ~2*benchmarkFileCount per-duplicate cost, above noise.
+	const budget = float64(benchmarkFileCount)
+	assert.Less(t, delta, budget,
+		"growing mode must not re-encode the bridging raw header per dropped "+
+			"duplicate: growing=%.0f static=%.0f delta=%.0f allocs exceeds "+
+			"budget=%.0f", growingAllocs, staticAllocs, delta, budget)
+}
+
 func BenchmarkToFileDescriptor(b *testing.B) {
 	dir := b.TempDir()
 	basename := "created.log"