[9.5](backport #51077) Add Filestream harvester ingestion progress metrics

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 {