[9.1](backport #47247) [Filebeat/Filestream] Fix missing last few lines of a file

[mergify]

Proposed commit message

Filestream could miss ingesting the last few lines of a file when the
following happened:
- The Harvester reaches EOF and stops on its backoff
- The inactive check that runs on its own go routine marks the file as
  inactive and cancels the reader/harvester context.
- The file watcher, that runs on its own goroutine, detects a change
  in file size and tries to start a new harvester. The file watcher
  updates its internal state of the file as its current size.
- The harvester fails to start because there is one already running
  (the one blocked on backoff wait).
- The backoff expires and the Harvester resumes running and exits
  right away.
- The file watcher has a state (size) for the file that is different
  than what was actually ingested, so it does not try to start a new
  harvester until there is another change in the file. 

This makes Filebeat to miss the last few lines added to the file.

This commit fixes this problem by making the harvester notify the file
watcher when it stops and the amount of data is has read. During the
scan the file watcher can replace its internal state by the harvester,
allowing it to start a new harvester if necessary.

Checklist

• My code follows the style guidelines of this project
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation
• I have made corresponding change to the default configuration files
• I have added tests that prove my fix is effective or that my feature works
• I have added an entry in ./changelog/fragments using the changelog tool.

## Disruptive User Impact

Author's Checklist

• Remove Go mod replace once [logp/logptest] Add logptest.Logger elastic-agent-libs#362 is merged and released
• Remove Go mod replace once Log the 'log.logger' key elastic-agent-libs#365 is merged and released
• Run a Benchbuilder benchmark

How to test this PR locally

Run the tests

cd filebeat/input/filestream
go test -race -v -count=1 -tags=integration -run="TestFileWatcherNotifications|TestDataAddedAfterCloseInactive" ./input/filestream
# Test the fixed race condition
go test -race -v -count=1 -tags=integration -run=TestRotatingCloseInactiveLargerWriteRate ./input/filestream

Manual test

Testing this fix manually is possible, but requires you to monitor the
logs and add data to the file being ingested at a specific time.

At a very high level, the steps are:

1. Start Filebeat
2. Wait it to read the whole file
3. While the harvester is blocked:
   1. Add more data to the file
   2. Wait for the file scanner to run and fail to start a new
      harvester
4. Wait for the harvester to close
5. Wait for the file scanner to run again
6. Wait for a new harvester to start
7. Wait the harvester to reach EOF
8. Ensure the whole file has been ingested

If you ran this test without the fix from this PR, after #4
Filestream will not try to start any more harvesters for the file,
effectively missing the last few lines.

---

The best way to manually test this PR is to have two terminals open,
one running Filebeat and another ready to append data to the file
Filebeat is ingesting.

1. Create a file with at least 1kb of data and write down its size

   flog -n 20 > /tmp/flog.log
   wc -c /tmp/flog.log

2. Start Filebeat with following config:

   filebeat.yml

   filebeat.inputs:
     - type: filestream
       paths:
         - /tmp/flog.log
       prospector.scanner.check_interval: 4s
       close.on_state_change.inactive: 2s
       close.on_state_change.check_interval: 2s
       backoff:
         init: 15s
         max: 15s
   
   output.file:
     path: ${path.home}
     filename: output
   
   logging:
     level: debug
     to_stderr: true
     selectors:
       - input.file_watcher
       - input.filestream
       - input.scanner

   To make the logs easier to read, you can send the logs to stdout
   and pipe them through jq:

   go run . --path.home=$PWD 2>&1 | jq '{"ts": ."@timestamp", "lvl": ."log.level", "logger": ."log.logger", "m": .message}' -c
   

3. Wait for the log entry: '/tmp/flog.log' is inactive

4. Add data to the file flog -n 2 >> /tmp/flog.log

5. Wait for the log entry: File /tmp/flog.log has been updated

6. Wait for the log entry: Harvester already running

7. Wait for the log entry: File is inactive. Closing. Path='/tmp/flog.log'

8. Wait for the log entry: Stopped harvester for file

9. Wait for the log entry: Updating previous state because harvester was closed. '/tmp/flog.log': xxx, where xxx is the original file size.

10. Wait for the log entry: File /tmp/flog.log has been updated

11. Wait for the log entry: Starting harvester for file

12. Wait for the log entry: End of file reached: /tmp/flog.log; Backoff now.

13. Ensure all events have been read: wc -l output*.ndjson.

Related issues

• Closes [filestream] - Filestream stops reading last few lines #46923
• Supersedes: [Filebeat/Filestream] Prevent modified files close due to inactivity #47107

## Use cases
## Screenshots
## Logs

Benchmarks

Go Benchmark

This is likely not very relevant to the final form of this PR, but I ran some benchmarks comparing the different strategies to prevent the race condition when accessing the offset and lastTimeRead in the harvester, below are the results and the code

% go test -bench=BenchmarkOffset .   
goos: linux
goarch: amd64
pkg: github.com/elastic/beats/v7/filebeat/input/filestream
cpu: 11th Gen Intel(R) Core(TM) i9-11950H @ 2.60GHz
BenchmarkOffsetAndLastTimeRead/atomic-16                33326760                39.15 ns/op
BenchmarkOffsetAndLastTimeRead/mutex-16                 32199968                35.71 ns/op
BenchmarkOffsetAndLastTimeRead/race-16                  41037943                26.28 ns/op
PASS
ok      github.com/elastic/beats/v7/filebeat/input/filestream   13.290s

% go test -bench=BenchmarkOffset .
goos: linux
goarch: amd64
pkg: github.com/elastic/beats/v7/filebeat/input/filestream
cpu: 11th Gen Intel(R) Core(TM) i9-11950H @ 2.60GHz
BenchmarkOffsetAndLastTimeRead/atomic-16                34732719                37.35 ns/op
BenchmarkOffsetAndLastTimeRead/mutex-16                 33407095                35.86 ns/op
BenchmarkOffsetAndLastTimeRead/race-16                  45181587                25.54 ns/op
PASS
ok      github.com/elastic/beats/v7/filebeat/input/filestream   13.408s

filebeat/input/filestream/filestream_test.go

func BenchmarkOffsetAndLastTimeRead(b *testing.B) {
	a := benchAtomic{}
	b.Run("atomic", func(b *testing.B) {
		for b.Loop() {
			a.Inc()
		}
	})
	m := benchMutex{}
	b.Run("mutex", func(b *testing.B) {
		for b.Loop() {
			m.Inc()
		}
	})
	r := benchRace{}
	b.Run("race", func(b *testing.B) {
		for b.Loop() {
			r.Inc()
		}
	})
}

type benchAtomic struct {
	offset       atomic.Int64
	lastTimeRead atomic.Int64
}

func (b *benchAtomic) Inc() {
	b.offset.Add(42)
	b.lastTimeRead.Store(time.Now().UnixNano())
}

type benchMutex struct {
	mutex        sync.Mutex
	offset       int64
	lastTimeRead time.Time
}

func (b *benchMutex) Inc() {
	b.mutex.Lock()
	b.offset += 42
	b.lastTimeRead = time.Now()
	b.mutex.Unlock()
}

type benchRace struct {
	offset       int64
	lastTimeRead time.Time
}

func (b *benchRace) Inc() {
	b.offset += 42
	b.lastTimeRead = time.Now()
}

Benchbuilder

Latest release: v9.2.1

Suite Name	Benchmark Name	Version	Duration (sec)	EPS	Bytes/Event
logs-datastream	filestream-json-1024b-balanced	9.2.1	2m43.075351941s	12264.000000	175.3162
logs-datastream	filestream-json-1024b-throughput	9.2.1	48.46343038s	41269.000000	183.1183
logs-datastream	filestream-json-1024b-scale	9.2.1	2m47.897040994s	11912.000000	176.5148
logs-datastream	filestream-json-1024b-latency	9.2.1	4m51.107096736s	6870.000000	178.5985

PR version

Suite Name	Benchmark Name	Version	Duration (sec)	EPS	Bytes/Event
logs-datastream	filestream-json-1024b-balanced	9.3.0	2m41.103916351s	12414.000000	175.3734
logs-datastream	filestream-json-1024b-throughput	9.3.0	47.520195331s	42088.000000	182.5625
logs-datastream	filestream-json-1024b-scale	9.3.0	2m44.102216849s	12188.000000	175.8389
logs-datastream	filestream-json-1024b-latency	9.3.0	4m56.598482898s	6743.000000	179.3721

[mergify]

Cherry-pick of 3fa1a5e has failed:

On branch mergify/bp/9.1/pr-47247
Your branch is up to date with 'origin/9.1'.

You are currently cherry-picking commit 3fa1a5ef7.
  (fix conflicts and run "git cherry-pick --continue")
  (use "git cherry-pick --skip" to skip this patch)
  (use "git cherry-pick --abort" to cancel the cherry-pick operation)

Changes to be committed:
	new file:   changelog/fragments/1760488402-Prevent-modified-files-close-due-to-inactivity.yaml
	new file:   filebeat/input/filestream/fswatch_integration_test.go
	modified:   filebeat/input/filestream/input_integration_test.go
	modified:   filebeat/input/filestream/internal/input-logfile/input.go
	modified:   filebeat/input/filestream/internal/input-logfile/manager.go
	modified:   filebeat/input/filestream/internal/input-logfile/manager_test.go
	modified:   filebeat/input/filestream/internal/input-logfile/publish.go
	modified:   filebeat/input/filestream/internal/input-logfile/store.go
	modified:   filebeat/input/filestream/internal/input-logfile/store_test.go
	modified:   filebeat/input/filestream/prospector.go
	modified:   filebeat/input/filestream/prospector_test.go
	modified:   filebeat/tests/integration/filestream_test.go
	modified:   libbeat/tests/integration/framework.go

Unmerged paths:
  (use "git add/rm <file>..." as appropriate to mark resolution)
	both modified:   filebeat/input/filestream/environment_test.go
	both modified:   filebeat/input/filestream/filestream.go
	both modified:   filebeat/input/filestream/fswatch.go
	both modified:   filebeat/input/filestream/fswatch_test.go
	both modified:   filebeat/input/filestream/input.go
	deleted by us:   filebeat/input/filestream/input_delete_integration_test.go
	both modified:   filebeat/input/filestream/internal/input-logfile/fswatch.go
	both modified:   filebeat/input/filestream/internal/input-logfile/harvester.go
	both modified:   filebeat/input/filestream/internal/input-logfile/harvester_test.go
	both modified:   filebeat/input/filestream/prospector_creator.go
	both modified:   filebeat/input/filestream/prospector_creator_test.go
	both modified:   filebeat/tests/integration/filestream_truncation_test.go
	deleted by us:   libbeat/tests/integration/datagenerator.go

To fix up this pull request, you can check it out locally. See documentation: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/reviewing-changes-in-pull-requests/checking-out-pull-requests-locally

[github-actions]

🤖 GitHub comments

Just comment with:

• run docs-build : Re-trigger the docs validation. (use unformatted text in the comment!)

[elasticmachine]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

[belimawr]

Closing this for now as it is bringing features that are not part of 9.1