Resolve Mergify backport conflicts on 9.3

The automated cherry-pick of #51863 left conflict markers in fswatch.go and
fswatch_test.go and pulled in main-only APIs (harvester progress metrics,
the GetFiles options/metrics signature, completedFingerprints) that do not
exist on 9.3.

Reapply only the two allocation optimizations relevant to this branch,
adapted to its filestream APIs:
  - fileWatcher.watch resolves the file identity (srcID) lazily via an
    ensureSrcID closure; the closed-harvester reconciliation is split into
    hasClosedHarvesters/reconcileClosedHarvester so an idle scan never
    resolves an identity.
  - fileScanner.GetFiles pre-sizes its per-scan maps from the previous
    scan's unique-file count via a new lastCount field.

Adapt the new BenchmarkWatchIdle to the branch's watch/newFileWatcher
signatures.

Author: orestisfl

filebeat/input/filestream/fswatch.go

@@ -172,19 +172,7 @@ func (w *fileWatcher) processNotification(evt loginp.HarvesterStatus) {
 func (w *fileWatcher) watch(ctx unison.Canceler) {
 	w.log.Debug("Start next scan")
 
-<<<<<<< HEAD
 	paths := w.scanner.GetFiles()
-=======
-	// file identity is updated in GetFiles
-	now := time.Now()
-	scanOpts := loginp.FileScanOptions{
-		CurrentTime:         now,
-		IgnoreOlder:         ignoreOlder,
-		IgnoreInactiveSince: ignoreInactiveSince,
-	}
-	paths, scanMetrics := w.scanner.GetFiles(scanOpts)
-	metrics.UpdateFileScanMetrics(scanMetrics)
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 
 	// for debugging purposes
 	writtenCount := 0
@@ -207,8 +195,9 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 			continue
 		}
 
-		// srcID is the file identity (harvester ID/registry key), resolved lazily via ensureSrcID:
-		// an unchanged, untracked file (gzip, empty, ignore_older) never needs one, saving allocs.
+		// srcID is the file identity (harvester ID/registry key), resolved lazily via
+		// ensureSrcID: an unchanged file that produces no event never needs one, saving a
+		// per-file identity allocation on every idle scan.
 		var srcID string
 		ensureSrcID := func() string {
 			if srcID == "" { // getFileIdentity never returns ""
@@ -217,7 +206,8 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 			return srcID
 		}
 
-		// closedHarvesters is empty in the steady state; this reconciliation is usually skipped.
+		// closedHarvesters is empty in the steady state, so this reconciliation is usually
+		// skipped and the file identity is never resolved.
 		if w.hasClosedHarvesters() {
 			w.reconcileClosedHarvester(&prevDesc, ensureSrcID())
 		}
@@ -260,14 +250,6 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 			}
 		}
 
-<<<<<<< HEAD
-=======
-		// Record progress metrics for trackable, non-truncated files (tracksHarvesterProgress).
-		if e.Op != loginp.OpTruncate && tracksHarvesterProgress(&fd, scanOpts) {
-			harvesterFiles = append(harvesterFiles, loginp.HarvesterFile{ID: ensureSrcID(), Size: fd.Info.Size()})
-		}
-
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 		// delete from previous state to mark that we've seen the existing file again
 		delete(w.prev, path)
 	}
@@ -307,14 +289,6 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 		case w.events <- createEvent(path, *fd, w.getFileIdentity(*fd)):
 			createdCount++
 		}
-<<<<<<< HEAD
-=======
-
-		// New files skip the main loop via early continue, so collect their metrics here.
-		if tracksHarvesterProgress(fd, scanOpts) {
-			harvesterFiles = append(harvesterFiles, loginp.HarvesterFile{ID: srcID, Size: fd.Info.Size()})
-		}
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 	}
 
 	w.log.Debugw("File scan complete",
@@ -329,18 +303,27 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 	w.prev = paths
 }
 
-<<<<<<< HEAD
-=======
 // hasClosedHarvesters reports whether any harvester-close notification awaits reconciliation.
+// It is a cheap, lock-guarded length check so the watch loop can skip resolving a file identity
+// for every scanned file when there is nothing to reconcile (the common steady-state case).
 func (w *fileWatcher) hasClosedHarvesters() bool {
 	w.closedHarvestersMutex.Lock()
 	defer w.closedHarvestersMutex.Unlock()
 	return len(w.closedHarvesters) > 0
 }
 
 // reconcileClosedHarvester folds a recently-closed harvester's ingested offset (from
-// closedHarvesters) into prevDesc so a restarted harvester resumes from the right position.
-// It guards a close-during-backoff race that would otherwise withhold writes and lose lines.
+// closedHarvesters) into prevDesc so a restarted harvester resumes from the right position,
+// then drops the entry.
+//
+// This prevents a race condition: when the reader/harvester reaches EOF it blocks on a backoff.
+// If during this time [logFile.shouldBeClosed] marks the file inactive and closes the reader
+// context, once the backoff expires the reader and harvester are closed without ingesting any
+// more data. If the fileWatcher sends a write event while the harvester was blocked, no new
+// harvester is started because one is already running, yet the fileWatcher updates its internal
+// state and won't send further write events until more data is added, so some lines can be
+// missed. By applying the offset reported when the harvester closed we realign our state and
+// start a new harvester if needed.
 func (w *fileWatcher) reconcileClosedHarvester(prevDesc *loginp.FileDescriptor, id string) {
 	w.closedHarvestersMutex.Lock()
 	defer w.closedHarvestersMutex.Unlock()
@@ -353,30 +336,6 @@ func (w *fileWatcher) reconcileClosedHarvester(prevDesc *loginp.FileDescriptor,
 	delete(w.closedHarvesters, id)
 }
 
-// tracksHarvesterProgress reports whether a file contributes to the harvester progress metrics.
-func tracksHarvesterProgress(fd *loginp.FileDescriptor, opts loginp.FileScanOptions) bool {
-	return !fd.GZIP && fd.Info.Size() > 0 && !isFileIgnored(*fd, opts)
-}
-
-// isFileIgnored returns true when a file is ignored, no matter the reason.
-func isFileIgnored(
-	fd loginp.FileDescriptor,
-	opts loginp.FileScanOptions,
-) bool {
-	modTime := fd.Info.ModTime()
-
-	if opts.IgnoreOlder > 0 && opts.CurrentTime.Sub(modTime) > opts.IgnoreOlder {
-		return true
-	}
-
-	if !opts.IgnoreInactiveSince.IsZero() && modTime.Sub(opts.IgnoreInactiveSince) <= 0 {
-		return true
-	}
-
-	return false
-}
-
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 // getFileIdentity mimics the same algorithm used by the harvester to generate
 // the file identity to any given file.
 // See 'startHarvester' on internal/input-logfile/harvester.go.
@@ -413,18 +372,8 @@ func (w *fileWatcher) Event() loginp.FSEvent {
 	return <-w.events
 }
 
-<<<<<<< HEAD
 func (w *fileWatcher) GetFiles() map[string]loginp.FileDescriptor {
 	return w.scanner.GetFiles()
-=======
-// GetFiles runs a one-off enumeration scan for the prospector's Init and
-// TakeOver phases. Unlike the watch loop it does not advance the scanner's
-// completedFingerprints set, so these pre-watch scans cannot suppress the
-// bridging raw header a still-growing entry needs to migrate its registry key
-// after a restart.
-func (w *fileWatcher) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.FileDescriptor, loginp.FileScanMetrics) {
-	return w.scanner.GetFiles(opts)
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 }
 
 type fingerprintConfig struct {
@@ -463,21 +412,11 @@ type fileScanner struct {
 	hasher           hash.Hash
 	readBuffer       []byte
 	compression      string
-<<<<<<< HEAD
-=======
-	// 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 map[string]struct{}
 
 	// lastCount is the number of unique files the previous scan produced.
+	// It pre-sizes the per-scan maps to avoid repeated grow/rehash allocations
+	// on a stable file set.
 	lastCount int
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 }
 
 func newFileScanner(logger *logp.Logger, paths []string, config fileScannerConfig, compression string) (*fileScanner, error) {
@@ -549,27 +488,13 @@ func (s *fileScanner) normalizeGlobPatterns() error {
 
 // GetFiles returns a map of file descriptors by filenames that
 // match the configured paths.
-<<<<<<< HEAD
 func (s *fileScanner) GetFiles() map[string]loginp.FileDescriptor {
-	fdByName := map[string]loginp.FileDescriptor{}
-=======
-func (s *fileScanner) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.FileDescriptor, loginp.FileScanMetrics) {
-	if opts.CurrentTime.IsZero() {
-		opts.CurrentTime = time.Now()
-	}
-
-	// Pre-size the per-scan maps from the previous scan's count.
+	// Pre-size the per-scan maps from the previous scan's unique-file count.
 	fdByName := make(map[string]loginp.FileDescriptor, s.lastCount)
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 	// used to determine if a symlink resolves in a already known target
 	uniqueIDs := make(map[string]string, s.lastCount)
 	// used to filter out duplicate matches
-<<<<<<< HEAD
-	uniqueFiles := map[string]struct{}{}
-=======
 	uniqueFiles := make(map[string]struct{}, s.lastCount)
-	scanMetrics := loginp.FileScanMetrics{}
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 
 	for _, path := range s.paths {
 		matches, err := filepath.Glob(path)
@@ -621,13 +546,8 @@ func (s *fileScanner) GetFiles(opts loginp.FileScanOptions) (map[string]loginp.F
 		}
 	}
 
-<<<<<<< HEAD
-	return fdByName
-=======
-	scanMetrics.FilesUnique = int64(len(fdByName))
 	s.lastCount = len(fdByName)
-	return fdByName, scanMetrics
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
+	return fdByName
 }
 
 type ingestTarget struct {

filebeat/input/filestream/fswatch_test.go

@@ -1139,64 +1139,11 @@ func BenchmarkGetFilesWithFingerprint(b *testing.B) {
 	}
 }
 
-<<<<<<< HEAD
-=======
-// BenchmarkGetFilesWithFingerprintGrowing measures repeated scans of stable,
-// above-threshold files. Each GetFiles call is one scan, so b.N iterations model
-// files that have grown past the fingerprint threshold and then sit unchanged
-// across many scans — the scenario the sub-threshold growing benchmarks in
-// BenchmarkFilestream do not exercise.
-//
-// The "static" and "growing" sub-benchmarks fingerprint the identical
-// above-threshold files; their only difference is growing mode. The delta is the
-// per-scan raw-header re-encode that growing mode performs on every completed
-// file to bridge a possible threshold crossing. It calls GetFiles directly
-// without advancing the completedFingerprints set, so it models the UNSUPPRESSED
-// 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.
-func BenchmarkGetFilesWithFingerprintGrowing(b *testing.B) {
-	cases := []struct {
-		name    string
-		growing bool
-	}{
-		{"static", false},
-		{"growing", true},
-	}
-	for _, tc := range cases {
-		b.Run(tc.name, func(b *testing.B) {
-			paths := writeBenchmarkFiles(b, b.TempDir(), benchmarkFileCount)
-			cfg := fileScannerConfig{
-				Fingerprint: fingerprintConfig{
-					Enabled: true,
-					Offset:  0,
-					Length:  1024,
-					Growing: tc.growing,
-				},
-			}
-			s, err := newFileScanner(logp.NewNopLogger(), paths, cfg, CompressionNone)
-			require.NoError(b, err)
-
-			b.ResetTimer()
-			for i := 0; i < b.N; i++ {
-				files, _ := s.GetFiles(loginp.FileScanOptions{})
-				require.Len(b, files, benchmarkFileCount)
-			}
-		})
-	}
-}
-
-// BenchmarkWatchIdle measures fileWatcher.watch over an unchanged file set: the mostly-idle steady
-// state where allocs/op is the cost of observing nothing changed (identity cache + pre-sized maps).
+// BenchmarkWatchIdle measures fileWatcher.watch over an unchanged file set: the mostly-idle
+// steady state where allocs/op is the cost of observing that nothing changed. With the lazy
+// file-identity resolution and pre-sized scan maps, an idle rescan should not allocate a file
+// identity per file.
 func BenchmarkWatchIdle(b *testing.B) {
-	states := []struct {
-		name        string
-		ignoreOlder time.Duration
-		aged        bool
-	}{
-		{"active", 0, false},
-		{"ignored", time.Hour, true},
-	}
 	identities := []struct {
 		name        string
 		fingerprint bool
@@ -1209,66 +1156,51 @@ func BenchmarkWatchIdle(b *testing.B) {
 		}},
 	}
 
-	for _, st := range states {
-		for _, id := range identities {
-			for _, fileCount := range []int{100, 1000, 10000} {
-				b.Run(fmt.Sprintf("%s/%s/%d_files", st.name, id.name, fileCount), func(b *testing.B) {
-					paths := writeBenchmarkFiles(b, b.TempDir(), fileCount)
-
-					if st.aged {
-						// Age every file well past ignoreOlder so it is excluded.
-						old := time.Now().Add(-48 * time.Hour)
-						matches, err := filepath.Glob(paths[0])
-						require.NoError(b, err)
-						for _, m := range matches {
-							require.NoError(b, os.Chtimes(m, old, old))
+	for _, id := range identities {
+		for _, fileCount := range []int{100, 1000, 10000} {
+			b.Run(fmt.Sprintf("%s/%d_files", id.name, fileCount), func(b *testing.B) {
+				paths := writeBenchmarkFiles(b, b.TempDir(), fileCount)
+
+				cfg := defaultFileWatcherConfig()
+				cfg.Scanner.Fingerprint.Enabled = id.fingerprint
+
+				fw, err := newFileWatcher(
+					logp.NewNopLogger(),
+					paths,
+					cfg,
+					CompressionNone,
+					false,
+					id.newID(),
+					mustSourceIdentifier("bench-id"),
+				)
+				require.NoError(b, err)
+
+				// A create event is emitted for every file on the first scan; idle rescans emit
+				// nothing. Drain in the background so watch's sends never block.
+				ctx := b.Context()
+				go func() {
+					for {
+						select {
+						case <-ctx.Done():
+							return
+						case <-fw.events:
 						}
 					}
+				}()
 
-					cfg := defaultFileWatcherConfig()
-					cfg.Scanner.Fingerprint.Enabled = id.fingerprint
-					cfg.Scanner.Fingerprint.Growing = id.fingerprint
-
-					fw, err := newFileWatcher(
-						logp.NewNopLogger(),
-						paths,
-						cfg,
-						CompressionNone,
-						false,
-						id.newID(),
-						mustSourceIdentifier("bench-id"),
-					)
-					require.NoError(b, err)
-
-					// A create event is emitted for every file on the first scan; idle rescans emit
-					// nothing. Drain in the background so watch's sends never block.
-					ctx := b.Context()
-					go func() {
-						for {
-							select {
-							case <-ctx.Done():
-								return
-							case <-fw.events:
-							}
-						}
-					}()
-
-					metrics := newTestMetrics()
-					// Prime w.prev so the benchmarked scans are pure steady state.
-					fw.watch(ctx, metrics, st.ignoreOlder, time.Time{})
+				// Prime w.prev so the benchmarked scans are pure steady state.
+				fw.watch(ctx)
 
-					b.ReportAllocs()
-					b.ResetTimer()
-					for b.Loop() {
-						fw.watch(ctx, metrics, st.ignoreOlder, time.Time{})
-					}
-				})
-			}
+				b.ReportAllocs()
+				b.ResetTimer()
+				for b.Loop() {
+					fw.watch(ctx)
+				}
+			})
 		}
 	}
 }
 
->>>>>>> c05890c43 (filestream: reduce per-scan allocations (#51863))
 func createWatcherWithConfig(t *testing.T, logger *logp.Logger, paths []string, cfgStr string) *fileWatcher {
 	tmpCfg := struct {
 		Scaner fileWatcherConfig `config:"scanner"`