POC: filestream growing fingerprint identity

Add a proof-of-concept for a new "growing_fingerprint" file identity mode
that addresses the limitation where files smaller than the fingerprint size
(default 1024 bytes) cannot be tracked.

### Key changes:

- Add growingFingerprintIdentifier that stores raw bytes (hex-encoded) instead
  of a hash, allowing the fingerprint to grow as the file grows
- Files can be tracked immediately regardless of size (no minimum threshold)
- Implement prefix matching: when a file grows, match the old (shorter)
  fingerprint as a prefix of the new (longer) fingerprint
- Add IterateOnPrefix() and UpdateKey() to the store for registry key migration
- Support in-place key updates without interrupting running harvesters
- Default max_length is 1000 bytes (matching OTEL's filelog receiver)

Configuration:
```
prospector.scanner:
  fingerprint.growing: true
  fingerprint.max_length: 1000  # optional, default 1000
  file_identity.growing_fingerprint: ~
```

This enables tracking small files that share initial content (e.g., common
headers) by allowing their fingerprints to diverge as they grow with
unique content.

### Includes integration tests covering:
- Small files tracked immediately
- Files with identical initial content differentiated as they grow
- Fingerprint migration on file growth
- Restart scenarios

### Open question/issue

Using the raw bytes (hex-encoded) as the fingerprint makes it easier to compare
prefix matches, however increases the memory consumption, up to 1000 bytes per
file per in-memory instance of the fingerprint. Also, it increases the storage
used by the registry on disk.

Author: AndersonQ

filebeat/input/filestream/copytruncate_prospector.go

@@ -264,7 +264,6 @@ func (p *copyTruncateFileProspector) onFSEvent(
 		// check if the event belongs to a rotated file
 		if p.isRotated(event) {
 			log.Debugf("File %s is rotated", event.NewPath)
-
 			p.onRotatedFile(log, ctx, event, src, group)
 
 		} else {

filebeat/input/filestream/fswatch.go

@@ -26,6 +26,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 
@@ -39,10 +40,11 @@ import (
 )
 
 const (
-	RecursiveGlobDepth           = 8
-	DefaultFingerprintSize int64 = 1024 // 1KB
-	scannerDebugKey              = "scanner"
-	watcherDebugKey              = "file_watcher"
+	RecursiveGlobDepth                  = 8
+	DefaultFingerprintSize        int64 = 1024 // 1KB
+	DefaultGrowingFingerprintSize int64 = 1000 // Same as OTEL's filelog receiver
+	scannerDebugKey                     = "scanner"
+	watcherDebugKey                     = "file_watcher"
 )
 
 var (
@@ -167,9 +169,20 @@ func (w *fileWatcher) processNotification(evt loginp.HarvesterStatus) {
 	w.closedHarvestersMutex.Unlock()
 }
 
+// AndersonQ: fsEvents created here.
+// here when f1 grwos -> same file / written
+//
+//	f2 same "old" f1 fingerprint -> new file
 func (w *fileWatcher) watch(ctx unison.Canceler) {
 	w.log.Debug("Start next scan")
 
+	prevKeys := make([]string, 0, len(w.prev))
+	for k := range w.prev {
+		prevKeys = append(prevKeys, k)
+	}
+	w.log.Debugf("Start next scan: prevs: %s", strings.Join(prevKeys, ","))
+
+	// file identity is updated in GetFiles
 	paths := w.scanner.GetFiles()
 
 	// for debugging purposes
@@ -189,9 +202,17 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 
 		// if the scanner found a new path or an existing path
 		// with a different file, it is a new file
+		// fileDescriptor key is the path
 		prevDesc, ok := w.prev[path]
 		sfd := fd // to avoid memory aliasing
-		if !ok || !loginp.SameFile(&prevDesc, &sfd) {
+		// AndersonQ: new, longer fingerprint makes it looks like a new file. It
+		// has to remain the same file
+		// If growing fingerprint is enabled, just the ID match isn't enough.
+		if !ok || !loginp.SameFile(w.log, &prevDesc, &sfd) {
+			if ok {
+				w.log.Infof("file %q has been replaced by a new file. Old ID %q, new ID %q",
+					path, prevDesc.FileID(), fd.FileID())
+			}
 			newFilesByName[path] = &sfd
 			newFilesByID[fd.FileID()] = &sfd
 			continue
@@ -244,6 +265,8 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 		// If a harvester for this file was closed recently,
 		// we use its state instead of the one we have cached.
 		case prevDesc.SizeOrBytesIngested() < fd.Info.Size():
+			// AndersonQ: here, file with new/longer fingerprint has ti be
+			// treated as the same file.
 			e = writeEvent(path, fd, srcID)
 			writtenCount++
 
@@ -300,6 +323,8 @@ func (w *fileWatcher) watch(ctx unison.Canceler) {
 	}
 
 	// remaining files in newFiles are newly created files
+	// AndersonQ: when a file grows, it isn't a new file. Even though it has a
+	// new ID
 	for path, fd := range newFilesByName {
 		// no need to react on empty new files
 		if fd.Info.Size() == 0 {
@@ -371,6 +396,12 @@ type fingerprintConfig struct {
 	Enabled bool  `config:"enabled"`
 	Offset  int64 `config:"offset"`
 	Length  int64 `config:"length"`
+	// Growing enables the growing fingerprint mode where the fingerprint
+	// is the raw bytes (not a hash) and can grow as the file grows.
+	Growing bool `config:"growing"`
+	// MaxLength is the maximum number of bytes to use for the growing fingerprint.
+	// Default is 1000 bytes (same as OTEL's filelog receiver).
+	MaxLength int64 `config:"max_length"`
 }
 
 type fileScannerConfig struct {
@@ -386,22 +417,25 @@ func defaultFileScannerConfig() fileScannerConfig {
 		Symlinks:      false,
 		RecursiveGlob: true,
 		Fingerprint: fingerprintConfig{
-			Enabled: true,
-			Offset:  0,
-			Length:  DefaultFingerprintSize,
+			Enabled:   true,
+			Offset:    0,
+			Length:    DefaultFingerprintSize,
+			Growing:   false,
+			MaxLength: DefaultGrowingFingerprintSize,
 		},
 	}
 }
 
 // fileScanner looks for files which match the patterns in paths.
 // It is able to exclude files and symlinks.
 type fileScanner struct {
-	paths       []string
-	cfg         fileScannerConfig
-	log         *logp.Logger
-	hasher      hash.Hash
-	readBuffer  []byte
-	gzipAllowed bool
+	paths         []string
+	cfg           fileScannerConfig
+	log           *logp.Logger
+	hasher        hash.Hash
+	readBuffer    []byte
+	growingBuffer []byte // buffer for growing fingerprint mode
+	gzipAllowed   bool
 }
 
 func newFileScanner(logger *logp.Logger, paths []string, config fileScannerConfig, gzipAllowed bool) (*fileScanner, error) {
@@ -413,7 +447,11 @@ func newFileScanner(logger *logp.Logger, paths []string, config fileScannerConfi
 		gzipAllowed: gzipAllowed,
 	}
 
-	if s.cfg.Fingerprint.Enabled {
+	if s.cfg.Fingerprint.Growing {
+		// Growing fingerprint mode: use raw bytes, no minimum size requirement
+		s.log.Debugf("growing fingerprint mode enabled: max_length %d", s.cfg.Fingerprint.MaxLength)
+		s.growingBuffer = make([]byte, s.cfg.Fingerprint.MaxLength)
+	} else if s.cfg.Fingerprint.Enabled { // TODO(AndersonQ): it's confusing having cfg.Fingerprint.Enabled and cfg.Fingerprint.Growing meaning different things
 		if s.cfg.Fingerprint.Length < sha256.BlockSize {
 			err := fmt.Errorf("fingerprint size %d bytes cannot be smaller than %d bytes", config.Fingerprint.Length, sha256.BlockSize)
 			return nil, fmt.Errorf("error while reading configuration of fingerprint: %w", err)
@@ -473,6 +511,7 @@ func (s *fileScanner) normalizeGlobPatterns() error {
 
 // GetFiles returns a map of file descriptors by filenames that
 // match the configured paths.
+// AndersonQ: files are found here, duplicates are filtered out here
 func (s *fileScanner) GetFiles() map[string]loginp.FileDescriptor {
 	fdByName := map[string]loginp.FileDescriptor{}
 	// used to determine if a symlink resolves in a already known target
@@ -513,6 +552,8 @@ func (s *fileScanner) GetFiles() map[string]loginp.FileDescriptor {
 			}
 
 			fileID := fd.FileID()
+			// AndersonQ: was it a file that grew? It should not be relevant here
+			// we're just getting the files we need to ingest.
 			if knownFilename, exists := uniqueIDs[fileID]; exists {
 				s.log.Warnf("%q points to an already known ingest target %q [%s==%s]. Skipping", fd.Filename, knownFilename, fileID, fileID)
 				continue
@@ -609,6 +650,11 @@ func (s *fileScanner) toFileDescriptor(it *ingestTarget) (fd loginp.FileDescript
 	var osFile *os.File
 	var file File
 
+	// Growing fingerprint mode: compute raw bytes fingerprint
+	if s.cfg.Fingerprint.Growing {
+		return s.computeGrowingFingerprint(it, fd)
+	}
+
 	if !s.cfg.Fingerprint.Enabled {
 		return fd, nil
 	}
@@ -684,6 +730,64 @@ func (s *fileScanner) toFileDescriptor(it *ingestTarget) (fd loginp.FileDescript
 	return fd, nil
 }
 
+// computeGrowingFingerprint computes a raw bytes fingerprint for the growing
+// fingerprint file identity. Unlike the hash-based fingerprint, this stores
+// the actual file content (hex-encoded) and can grow as the file grows.
+// Files of any size are supported - if the file is smaller than max_length,
+// the entire content is used as the fingerprint.
+func (s *fileScanner) computeGrowingFingerprint(it *ingestTarget, fd loginp.FileDescriptor) (loginp.FileDescriptor, error) {
+	// Empty files have no fingerprint yet - empty string fingerprint
+	if it.info.Size() == 0 {
+		s.log.Info("fileScanner: computeGrowingFingerprint: size 0, nothing to compute", fd.Filename)
+		return fd, nil
+	}
+
+	osFile, err := os.Open(it.originalFilename)
+	if err != nil {
+		return fd, fmt.Errorf("failed to open %q for growing fingerprint: %w", it.originalFilename, err)
+	}
+	defer osFile.Close()
+
+	// Check for GZIP if allowed
+	if s.gzipAllowed {
+		fd.GZIP, err = IsGZIP(osFile)
+		if err != nil {
+			return fd, fmt.Errorf("failed to check if %q is gzip: %w", it.originalFilename, err)
+		}
+	}
+
+	// Determine how many bytes to read
+	maxLength := s.cfg.Fingerprint.MaxLength
+	fileSize := it.info.Size()
+	// TODO(AndersonQ): If the file is GZIP-compressed, we cannot use the compressed size
+	//   it needs to read until maxLength or EOF
+	toRead := fileSize
+	if toRead > maxLength {
+		toRead = maxLength
+	}
+	s.log.Infof("fileScanner: computeGrowingFingerprint: for file %s: %d/%d bytes",
+		fd.Filename, toRead, maxLength)
+
+	var r io.Reader = osFile
+	if fd.GZIP {
+		// fingerprint is computed on decompressed data
+		gzReader, err := newGzipSeekerReader(osFile, int(maxLength))
+		if err != nil {
+			return fd, fmt.Errorf("failed to create gzip reader for %q: %w", it.originalFilename, err)
+		}
+		defer gzReader.Close()
+		r = gzReader
+	}
+
+	n, err := io.ReadFull(r, s.growingBuffer[:toRead])
+	if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
+		return fd, fmt.Errorf("failed to read %q for growing fingerprint: %w", it.originalFilename, err)
+	}
+	fd.Fingerprint = hex.EncodeToString(s.growingBuffer[:n])
+
+	return fd, nil
+}
+
 func (s *fileScanner) isFileExcluded(file string) bool {
 	return len(s.cfg.ExcludedFiles) > 0 && s.matchAny(s.cfg.ExcludedFiles, file)
 }

filebeat/input/filestream/identifier.go

@@ -32,20 +32,22 @@ const (
 	// IDs if a source is renamed.
 	trackRename identifierFeature = iota
 
-	nativeName      = "native"
-	pathName        = "path"
-	inodeMarkerName = "inode_marker"
-	fingerprintName = "fingerprint"
+	nativeName             = "native"
+	pathName               = "path"
+	inodeMarkerName        = "inode_marker"
+	fingerprintName        = "fingerprint"
+	growingFingerprintName = "growing_fingerprint"
 
 	DefaultIdentifierName = nativeName
 	identitySep           = "::"
 )
 
 var identifierFactories = map[string]identifierFactory{
-	nativeName:      newINodeDeviceIdentifier,
-	pathName:        newPathIdentifier,
-	inodeMarkerName: newINodeMarkerIdentifier,
-	fingerprintName: newFingerprintIdentifier,
+	nativeName:             newINodeDeviceIdentifier,
+	pathName:               newPathIdentifier,
+	inodeMarkerName:        newINodeMarkerIdentifier,
+	fingerprintName:        newFingerprintIdentifier,
+	growingFingerprintName: newGrowingFingerprintIdentifier,
 }
 
 type identifierFactory func(*conf.C, *logp.Logger) (fileIdentifier, error)

filebeat/input/filestream/identifier_growing_fingerprint.go

@@ -0,0 +1,62 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package filestream
+
+import (
+	loginp "github.com/elastic/beats/v7/filebeat/input/filestream/internal/input-logfile"
+	conf "github.com/elastic/elastic-agent-libs/config"
+	"github.com/elastic/elastic-agent-libs/logp"
+)
+
+// growingFingerprintIdentifier identifies files by their raw content fingerprint.
+// Unlike the hash-based fingerprint identifier, this stores the actual bytes
+// (hex-encoded) of the file header and the fingerprint can grow as the file grows.
+// This allows tracking files of any size immediately, without waiting for them
+// to reach a minimum size threshold.
+type growingFingerprintIdentifier struct{}
+
+func newGrowingFingerprintIdentifier(_ *conf.C, _ *logp.Logger) (fileIdentifier, error) {
+	return &growingFingerprintIdentifier{}, nil
+}
+
+func (i *growingFingerprintIdentifier) GetSource(e loginp.FSEvent) fileSource {
+	return fileSource{
+		desc:                e.Descriptor,
+		newPath:             e.NewPath,
+		oldPath:             e.OldPath,
+		truncated:           e.Op == loginp.OpTruncate,
+		archived:            e.Op == loginp.OpArchived,
+		fileID:              growingFingerprintName + identitySep + e.Descriptor.Fingerprint,
+		identifierGenerator: growingFingerprintName,
+	}
+}
+
+func (i *growingFingerprintIdentifier) Name() string {
+	return growingFingerprintName
+}
+
+func (i *growingFingerprintIdentifier) Supports(f identifierFeature) bool {
+	switch f {
+	case trackRename:
+		return true
+	default:
+		return false
+	}
+}
+
+

filebeat/input/filestream/internal/input-logfile/fswatch.go

@@ -18,6 +18,9 @@
 package input_logfile
 
 import (
+	"strings"
+
+	"github.com/elastic/elastic-agent-libs/logp"
 	"github.com/elastic/go-concert/unison"
 
 	"github.com/elastic/beats/v7/libbeat/common/file"
@@ -64,7 +67,9 @@ type FileDescriptor struct {
 	Filename string
 	// Info is the result of file stat
 	Info file.ExtendedFileInfo
-	// Fingerprint is a computed hash of the file header
+	// Fingerprint is used for file identity. For the "fingerprint" identity,
+	// this is a hash of the file header. For "growing_fingerprint", this is
+	// the hex-encoded raw bytes of the file header (which can grow).
 	Fingerprint string
 	// GZIP indicates if the file is compressed with GZIP.
 	GZIP bool
@@ -101,11 +106,34 @@ func (fd FileDescriptor) FileID() string {
 }
 
 // SameFile returns true if descriptors point to the same file.
-func SameFile(a, b *FileDescriptor) bool {
-	return a.FileID() == b.FileID()
+// For growing fingerprint identity, this handles the case where a file's
+// fingerprint has grown - checks if one fingerprint is a prefix of the other
+// and verifies it's the same physical file via OS state (inode/device).
+func SameFile(log *logp.Logger, prev, current *FileDescriptor) bool {
+	// return prev.FileID() == current.FileID()
+
+	// Fast path: exact match
+	if prev.FileID() == current.FileID() {
+		return true
+	}
+
+	// For growing fingerprint: check if one fingerprint is a prefix of the other
+	// This happens when a file grows and its fingerprint expands
+	if prev.Fingerprint != "" && current.Fingerprint != "" {
+
+		// If shorter is a prefix of longer, verify it's the same physical file
+		same := strings.HasPrefix(current.Fingerprint, prev.Fingerprint) &&
+			prev.Filename == current.Filename
+
+		log.Infof("SameFile: %t: prev=%s, current=%s. prevPath: %s, currPath: %s",
+			same, prev.Fingerprint, current.Fingerprint, prev.Filename, current.Filename)
+		return same
+	}
+
+	return false
 }
 
-// FSEvent returns inforamation about file system changes.
+// FSEvent returns information about file system changes.
 type FSEvent struct {
 	// NewPath is the new path of the file.
 	NewPath string