Add Filestream harvester ingestion progress metrics (#51077) (#51838)

Track active harvesters by ingestion percentage so users can see which
files are fully caught up, nearly caught up, or lagging. Use
scanner-observed file sizes and atomic harvester offsets to avoid
extra syscalls and keep the harvester read path free of shared mutex
contention.

GZIP files and files ignored for any reason are excluded because their
progress cannot be represented accurately by plain-file offset/size
comparisons.

I ran the Filestream benchmark (BenchmarkFilestream) against the latest main (69c5d2b), here are the results of 10 iterations:

goos: linux
goarch: amd64
pkg: github.com/elastic/beats/v7/filebeat/input/filestream
cpu: Intel(R) Core(TM) Ultra 9 285H
                                              │  before.txt  │              after.txt              │
                                              │    sec/op    │    sec/op     vs base               │
Filestream/1_file/inode-16                      18.98m ±  3%   19.32m ±  4%       ~ (p=0.684 n=10)
Filestream/1_file/fingerprint-16                18.31m ±  7%   18.12m ±  4%       ~ (p=0.853 n=10)
Filestream/100_files/inode-16                   79.06m ±  9%   82.39m ± 16%       ~ (p=0.190 n=10)
Filestream/100_files/fingerprint-16             79.14m ±  8%   77.90m ±  8%       ~ (p=0.796 n=10)
Filestream/1000_files/fingerprint-16            44.76m ± 12%   44.62m ±  8%       ~ (p=0.739 n=10)
Filestream/10000_files/fingerprint-16           459.6m ± 22%   448.5m ± 13%       ~ (p=0.796 n=10)
Filestream/line_filter/none-16                  15.20m ± 23%   16.52m ± 13%       ~ (p=0.436 n=10)
Filestream/line_filter/include-16               16.28m ±  5%   15.39m ±  9%       ~ (p=0.247 n=10)
Filestream/line_filter/exclude-16               17.03m ± 10%   15.93m ± 10%  -6.46% (p=0.043 n=10)
Filestream/line_filter/include_and_exclude-16   15.91m ±  9%   15.37m ± 12%       ~ (p=0.684 n=10)
Filestream/line_filter/drop_all-16              14.16m ± 16%   13.24m ±  8%       ~ (p=0.684 n=10)
geomean                                         32.48m         32.08m        -1.24%

GenAI-Assisted: Yes
Human-Reviewed: Yes
Tool: Cursor, Model: GPT-5.5 Agent Mode
Assisted-By: Cursor
(cherry picked from commit 768b58b)

Co-authored-by: Tiago Queiroz <tiago.queiroz@elastic.co>

Author: mergify[bot]

changelog/fragments/1783018351-add-filestream-harvester-metrics.yaml

@@ -0,0 +1,13 @@
+kind: enhancement
+summary: Add Filestream harvester metrics to monitoring logs
+description: |
+  The following Filestream metrics are added to the logs and the `/stats` HTTP endpoint:
+   - files_ingested_percent_100
+   - files_ingested_percent_95_99
+   - files_ingested_percent_lt_95
+  
+  These gauges are aggregated across all running Filestream inputs and
+  count active plain-file harvesters only.
+
+component: filebeat
+issue: https://github.com/elastic/beats/issues/36653

docs/reference/filebeat/filebeat-input-filestream.md

@@ -441,14 +441,14 @@ file_identity.fingerprint: ~
 
     `growing` {applies_to}`stack: ga 9.5.0+`
   :   When `true` (default), files smaller than the fingerprint size (`offset` + `length`) are tracked using the bytes available so far, instead of being skipped until they grow large enough. Once a file reaches the fingerprint size, it's automatically migrated to the regular SHA-256 fingerprint, with no data duplication. Refer to [Enhanced fingerprint](/reference/filebeat/file-identity.md#file-identity-fingerprint-growing) for details.
-      
+
       Set to `false` to restore the pre-9.5 behavior.
 
       ```yaml
       file_identity.fingerprint:
         growing: true
       ```
-    
+
 
 **`native`**
 :   Differentiates between files using their inodes and device IDs. This is the default file identity in Filebeat versions prior to 9.0.0.
@@ -1371,22 +1371,28 @@ Note: Each metric listed has a corresponding gzip_* counterpart (e.g.,
 the same data but exclusively for GZIP compressed files. The original metrics
 provide the total count, including both plain and GZIP files.
 
-### Harvester metrics [_harvester_metrics]
+### Scanner and harvester metrics [_harvester_metrics]
 
 ```{applies_to}
 stack: ga 9.5+
 ```
 
-The `filestream` input also exposes scanner metrics under
+The `filestream` input also exposes scanner and harvester progress metrics under
 `.monitoring.metrics.filebeat.filestream` in monitoring logs and under
 `filebeat.filestream` in the `/stats` HTTP endpoint output. These metrics are
 aggregate gauges across all running `filestream` inputs. They are updated after
-each scanner pass and reset when inputs stop.
+each scanner pass and reset when inputs stop. Harvester progress metrics measure
+how much data active plain-file harvesters have read from their files. They do
+not measure output publishing or acknowledgment progress. GZIP files and files
+ignored by `filestream` settings or state are excluded.
 
 | Metric | Description |
 | --- | --- |
 | `files_empty` | Number of matched files that are empty. |
 | `files_ignored` | Number of matched files ignored by `filestream` settings or state, such as `prospector.scanner.exclude_files`, `ignore_older`, or `ignore_inactive`. |
+| `files_ingested_percent_100` | Number of active plain-file harvesters whose read offset is at or beyond the scanner-observed file size. |
+| `files_ingested_percent_95_99` | Number of active plain-file harvesters whose read offset is at least 95% and less than 100% of the scanner-observed file size. |
+| `files_ingested_percent_lt_95` | Number of active plain-file harvesters whose read offset is less than 95% of the scanner-observed file size. |
 | `files_matched` | Number of filesystem path matches returned by the configured `paths` globs before duplicate, ignore, and ingestibility filtering. |
 | `files_no_ingest_target` | Number of matched non-empty files that did not produce an ingest target, such as duplicate matches, files that are too small to fingerprint or symlinks to already known files. |
 | `files_unique` | Number of unique files that produced ingest targets after scanner filtering and de-duplication. |

filebeat/input/filestream/fswatch.go

@@ -159,7 +159,7 @@ func (w *fileWatcher) Run(
 	ignoreInactiveSince time.Time,
 ) {
 	defer close(w.events)
-	defer metrics.CleanupFileScanMetrics()
+	defer metrics.Cleanup()
 
 	// run initial scan before starting regular
 	w.watch(ctx, metrics, ignoreOlder, ignoreInactiveSince)
@@ -222,6 +222,7 @@ func (w *fileWatcher) watch(
 
 	newFilesByName := make(map[string]*loginp.FileDescriptor)
 	newFilesByID := make(map[string]*loginp.FileDescriptor)
+	harvesterFiles := make([]loginp.HarvesterFile, 0, len(paths))
 
 	for path, fd := range paths {
 		// srcID is the file identity, it is the same value used to identify
@@ -305,6 +306,11 @@ func (w *fileWatcher) watch(
 			}
 		}
 
+		// We want progress metrics for all files except truncated ones
+		if e.Op != loginp.OpTruncate {
+			harvesterFiles = appendHarvesterFile(harvesterFiles, fd, srcID, now, ignoreOlder, ignoreInactiveSince)
+		}
+
 		// delete from previous state to mark that we've seen the existing file again
 		delete(w.prev, path)
 		// Delete used state from closedHarvesters
@@ -431,12 +437,18 @@ func (w *fileWatcher) watch(
 	// Unmatched-leftover creates: new files left over after both rename
 	// passes are genuinely new.
 	for path, fd := range newFilesByName {
+		srcID := w.getFileIdentity(*fd)
+
 		select {
 		case <-ctx.Done():
 			return
-		case w.events <- createEvent(path, *fd, w.getFileIdentity(*fd)):
+		case w.events <- createEvent(path, *fd, srcID):
 			createdCount++
 		}
+
+		// The previous for loop has an early continue for new files,
+		// so we need to collect their metrics here.
+		harvesterFiles = appendHarvesterFile(harvesterFiles, *fd, srcID, now, ignoreOlder, ignoreInactiveSince)
 	}
 
 	w.log.Debugw("File scan complete",
@@ -474,9 +486,34 @@ func (w *fileWatcher) watch(
 		}
 	}
 
+	metrics.UpdateHarvesterBuckets(harvesterFiles)
+
 	w.prev = paths
 }
 
+func appendHarvesterFile(
+	files []loginp.HarvesterFile,
+	fd loginp.FileDescriptor,
+	srcID string,
+	now time.Time,
+	ignoreOlder time.Duration,
+	ignoreInactiveSince time.Time,
+) []loginp.HarvesterFile {
+	opts := loginp.FileScanOptions{
+		CurrentTime:         now,
+		IgnoreOlder:         ignoreOlder,
+		IgnoreInactiveSince: ignoreInactiveSince,
+	}
+	if fd.GZIP || fd.Info.Size() <= 0 || isFileIgnored(fd, opts) {
+		return files
+	}
+
+	return append(files, loginp.HarvesterFile{
+		ID:   srcID,
+		Size: fd.Info.Size(),
+	})
+}
+
 // isFileIgnored returns true when a file is ignored, no matter the reason.
 func isFileIgnored(
 	fd loginp.FileDescriptor,

filebeat/input/filestream/fswatch_test.go

@@ -1398,6 +1398,147 @@ scanner:
 	assert.Equal(t, baseline.FilesEmpty, metrics.FilesEmpty.Get(), "files_empty")
 }
 
+func TestFileWatcherHarvesterMetrics(t *testing.T) {
+	identifier, err := newFingerprintIdentifier(nil, logp.NewNopLogger())
+	require.NoError(t, err, "failed to create fingerprint identifier")
+	fw := &fileWatcher{
+		fileIdentifier:   identifier,
+		sourceIdentifier: mustSourceIdentifier("foo-id"),
+		log:              logp.NewNopLogger(),
+		events:           make(chan loginp.FSEvent, 10),
+	}
+
+	now := time.Now()
+	oldModTime := now.Add(-2 * time.Hour)
+	descriptor := func(name string, size int64, modTime time.Time, gzip bool) loginp.FileDescriptor {
+		return loginp.FileDescriptor{
+			Filename:    name,
+			Fingerprint: loginp.FingerprintID{Sum: name},
+			GZIP:        gzip,
+			Info:        file.ExtendFileInfo(&testFileInfo{name: name, size: size, time: modTime}),
+		}
+	}
+	paths := map[string]loginp.FileDescriptor{
+		"complete":  descriptor("complete", 100, now, false),
+		"near":      descriptor("near", 100, now, false),
+		"lagging":   descriptor("lagging", 100, now, false),
+		"no-active": descriptor("no-active", 100, now, false),
+		"gzip":      descriptor("gzip", 100, now, true),
+		"ignored":   descriptor("ignored", 100, oldModTime, false),
+	}
+	fw.prev = map[string]loginp.FileDescriptor{
+		"complete":  descriptor("complete", 100, now, false),
+		"near":      descriptor("near", 100, now, false),
+		"lagging":   descriptor("lagging", 100, now, false),
+		"no-active": descriptor("no-active", 100, now, false),
+		"gzip":      descriptor("gzip", 100, now, true),
+		"ignored":   descriptor("ignored", 100, oldModTime, false),
+	}
+	fw.scanner = &testFileScanner{files: paths}
+
+	metrics := loginp.NewMetrics(monitoring.NewRegistry(), logp.NewNopLogger())
+	// Register some files/offsets, like a harvester would do.
+	completeOffset, cleanupCompleteOffset := metrics.RegisterHarvesterOffset(fw.getFileIdentity(paths["complete"]), 10)
+	nearOffset, _ := metrics.RegisterHarvesterOffset(fw.getFileIdentity(paths["near"]), 5)
+	laggingOffset, _ := metrics.RegisterHarvesterOffset(fw.getFileIdentity(paths["lagging"]), 4)
+	gzipOffset, _ := metrics.RegisterHarvesterOffset(fw.getFileIdentity(paths["gzip"]), 10)
+	ignoredOffset, _ := metrics.RegisterHarvesterOffset(fw.getFileIdentity(paths["ignored"]), 10)
+
+	// Make sure the test uses the same atomic update path as harvesters.
+	// Update to the actually expected values, like a harvester would do.
+	completeOffset.Store(100)
+	nearOffset.Store(95)
+	laggingOffset.Store(94)
+	gzipOffset.Store(100)
+	ignoredOffset.Store(100)
+
+	fw.watch(t.Context(), metrics, time.Hour, time.Time{})
+
+	assert.EqualValues(t, 1, metrics.FilesIngestedPercent100.Get(), "files_ingested_percent_100")
+	assert.EqualValues(t, 1, metrics.FilesIngestedPercent95To99.Get(), "files_ingested_percent_95_99")
+	assert.EqualValues(t, 1, metrics.FilesIngestedPercentLt95.Get(), "files_ingested_percent_lt_95")
+
+	// Copy paths and 'truncate' one file
+	truncatedPaths := map[string]loginp.FileDescriptor{}
+	for path, fd := range paths {
+		truncatedPaths[path] = fd
+	}
+	truncatedPaths["complete"] = descriptor("complete", 50, now, false)
+	fw.scanner = &testFileScanner{files: truncatedPaths}
+	fw.watch(t.Context(), metrics, time.Hour, time.Time{})
+
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent100.Get(), "files_ingested_percent_100 after truncation")
+	assert.EqualValues(t, 1, metrics.FilesIngestedPercent95To99.Get(), "files_ingested_percent_95_99 after truncation")
+	assert.EqualValues(t, 1, metrics.FilesIngestedPercentLt95.Get(), "files_ingested_percent_lt_95 after truncation")
+
+	// Simulate the harvester restart caused by truncation.
+	cleanupCompleteOffset()
+	_, _ = metrics.RegisterHarvesterOffset(fw.getFileIdentity(truncatedPaths["complete"]), 0)
+
+	// Copy truncatedPaths and make one file older
+	ignoredPaths := map[string]loginp.FileDescriptor{}
+	for path, fd := range truncatedPaths {
+		ignoredPaths[path] = fd
+	}
+	ignoredPaths["near"] = descriptor("near", 100, oldModTime, false)
+	fw.scanner = &testFileScanner{files: ignoredPaths}
+	fw.watch(t.Context(), metrics, time.Hour, time.Time{})
+
+	// The truncated file from the previous step is now a 'normal file at 50%'
+	// The 'near' file (95% ingested) is ignored because of ignore_older
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent100.Get(), "files_ingested_percent_100 after ignored")
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent95To99.Get(), "files_ingested_percent_95_99 after ignored")
+	assert.EqualValues(t, 2, metrics.FilesIngestedPercentLt95.Get(), "files_ingested_percent_lt_95 after ignored")
+
+	// An update with no paths slice effectively removes all files from the
+	// last update from the metrics
+	fw.scanner = &testFileScanner{}
+	fw.watch(t.Context(), metrics, time.Hour, time.Time{})
+
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent100.Get(), "files_ingested_percent_100 after reset")
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent95To99.Get(), "files_ingested_percent_95_99 after reset")
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercentLt95.Get(), "files_ingested_percent_lt_95 after reset")
+}
+
+func TestFileWatcherRunCleansHarvesterMetricsOnShutdown(t *testing.T) {
+	identifier, err := newFingerprintIdentifier(nil, logp.NewNopLogger())
+	require.NoError(t, err, "failed to create fingerprint identifier")
+
+	now := time.Now()
+	fd := loginp.FileDescriptor{
+		Filename:    "complete",
+		Fingerprint: loginp.FingerprintID{Sum: "complete"},
+		Info:        file.ExtendFileInfo(&testFileInfo{name: "complete", size: 100, time: now}),
+	}
+	paths := map[string]loginp.FileDescriptor{
+		"complete": fd,
+	}
+
+	fw := &fileWatcher{
+		cfg:              fileWatcherConfig{Interval: time.Hour},
+		prev:             map[string]loginp.FileDescriptor{"complete": fd},
+		scanner:          &testFileScanner{files: paths},
+		log:              logp.NewNopLogger(),
+		events:           make(chan loginp.FSEvent, 1),
+		notifyChan:       make(chan loginp.HarvesterStatus, 1),
+		closedHarvesters: map[string]int64{},
+		fileIdentifier:   identifier,
+		sourceIdentifier: mustSourceIdentifier("foo-id"),
+	}
+
+	metrics := loginp.NewMetrics(monitoring.NewRegistry(), logp.NewNopLogger())
+	sourceID := fw.getFileIdentity(fd)
+	_, _ = metrics.RegisterHarvesterOffset(sourceID, 100)
+
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+	fw.Run(ctx, metrics, time.Hour, time.Time{})
+
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent100.Get(), "files_ingested_percent_100 after watcher shutdown")
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercent95To99.Get(), "files_ingested_percent_95_99 after watcher shutdown")
+	assert.EqualValues(t, 0, metrics.FilesIngestedPercentLt95.Get(), "files_ingested_percent_lt_95 after watcher shutdown")
+}
+
 func mustSourceIdentifier(inputID string) *loginp.SourceIdentifier {
 	si, err := loginp.NewSourceIdentifier("filestream", inputID)
 	if err != nil {
@@ -1408,6 +1549,15 @@ func mustSourceIdentifier(inputID string) *loginp.SourceIdentifier {
 	return si
 }
 
+type testFileScanner struct {
+	files map[string]loginp.FileDescriptor
+}
+
+// GetFiles returns s.files and empty metrics.
+func (s *testFileScanner) GetFiles(loginp.FileScanOptions) (map[string]loginp.FileDescriptor, loginp.FileScanMetrics) {
+	return s.files, loginp.FileScanMetrics{}
+}
+
 const benchmarkFileCount = 1000
 
 func BenchmarkGetFiles(b *testing.B) {

filebeat/input/filestream/input.go

@@ -24,6 +24,7 @@ import (
 	"io"
 	"os"
 	"slices"
+	"sync/atomic"
 	"time"
 
 	"go.uber.org/zap"
@@ -235,6 +236,7 @@ func (inp *filestream) Test(src loginp.Source, ctx input.TestContext) error {
 func (inp *filestream) Run(
 	ctx input.Context,
 	src loginp.Source,
+	sourceID string,
 	cursor loginp.Cursor,
 	publisher loginp.Publisher,
 	metrics *loginp.Metrics) error {
@@ -266,6 +268,13 @@ func (inp *filestream) Run(
 		state.Offset = 0
 	}
 
+	var metricsOffset *atomic.Int64
+	if !fs.desc.GZIP {
+		var cleanupActiveOffset func()
+		metricsOffset, cleanupActiveOffset = metrics.RegisterHarvesterOffset(sourceID, state.Offset)
+		defer cleanupActiveOffset()
+	}
+
 	metrics.FilesActive.Inc()
 	metrics.HarvesterRunning.Inc()
 	defer metrics.FilesActive.Dec()
@@ -287,7 +296,7 @@ func (inp *filestream) Run(
 	// The caller of Run already reports the error and filters out errors that
 	// must not be reported, like 'context cancelled'.
 	err = inp.readFromSource(
-		ctx, log, r, fs.newPath, state, publisher, fs.desc.GZIP, metrics,
+		ctx, log, r, fs.newPath, state, publisher, fs.desc.GZIP, metricsOffset, metrics,
 		startReadUntilEOF)
 	if err != nil {
 		// First handle actual errors
@@ -729,6 +738,7 @@ func (inp *filestream) readFromSource(
 	s state,
 	p loginp.Publisher,
 	isGZIP bool,
+	metricsOffset *atomic.Int64,
 	metrics *loginp.Metrics,
 	startReadUntilEOF func(ctxtool.CancelContext)) error {
 
@@ -750,7 +760,7 @@ func (inp *filestream) readFromSource(
 
 	var err error
 	for ctx.Cancelation.Err() == nil {
-		err = inp.readLineFromSource(r, log, metrics, isGZIP, &s, p)
+		err = inp.readLineFromSource(r, log, metrics, isGZIP, &s, metricsOffset, p)
 		err, shouldContinue := inp.handleReadError(ctx, err, log, path, metrics, isGZIP)
 		if !shouldContinue {
 			return err
@@ -770,7 +780,7 @@ func (inp *filestream) readFromSource(
 			inp.readUntilEOF.Timeout)
 	LOOP:
 		for eofCancelCtx.Err() == nil {
-			err = inp.readLineFromSource(r, log, metrics, isGZIP, &s, p)
+			err = inp.readLineFromSource(r, log, metrics, isGZIP, &s, metricsOffset, p)
 			err, shouldContinue := inp.handleReadError(ctx, err, log, path, metrics, isGZIP)
 			if errors.Is(err, io.EOF) {
 				log.Debug("read_until_eof enabled, EOF reached. closing input")
@@ -788,7 +798,7 @@ func (inp *filestream) readFromSource(
 	return nil
 }
 
-func (inp *filestream) readLineFromSource(r reader.Reader, log *logp.Logger, metrics *loginp.Metrics, isGZIP bool, s *state, p loginp.Publisher) error {
+func (inp *filestream) readLineFromSource(r reader.Reader, log *logp.Logger, metrics *loginp.Metrics, isGZIP bool, s *state, metricsOffset *atomic.Int64, p loginp.Publisher) error {
 	message, err := r.Next()
 	if err != nil {
 		return err
@@ -797,6 +807,9 @@ func (inp *filestream) readLineFromSource(r reader.Reader, log *logp.Logger, met
 	// state offset increase. Mutated through *s so subsequent reads in
 	// readFromSource see the accumulated offset
 	s.Offset += int64(message.Bytes) + int64(message.Offset)
+	if metricsOffset != nil {
+		metricsOffset.Store(s.Offset)
+	}
 
 	flags, err := message.Fields.GetValue("log.flags")
 	if err == nil {