[8.19](backport #50324) filebeat: add read_until_eof to filestream

[mergify]

Proposed commit message

filebeat: add `read_until_eof` to filestream

Add a new `read_until_eof` option that defers a filestream input's
shutdown until the file currently being harvested reaches EOF (or a
configurable timeout fires). When the input is cancelled — for example
by a Kubernetes autodiscover provider removing the input on pod
termination — the harvester previously exited immediately and any
unread bytes were lost. With `read_until_eof.enabled: true`, the
harvester finishes draining the file before exiting.

Approach:

* `logFile` gets a one-shot `startReadUntilEOF` method that stops the
  file-monitoring goroutines (close.on_state_change.* /
  close.reader.after_interval), swaps the reader's cancellation
  context for one scoped to `read_until_eof.timeout`, and sets
  `closeOnEOF=true`. `sync.Once` makes the swap idempotent and
  guarantees that, by the time it returns, the monitoring goroutines
  cannot race with the swapped-in context.
* `readFromSource` now has a normal read loop and, when the feature
  is enabled, a `read_until_eof` continuation. When the normal loop
  exits because `ctx.Cancelation` was cancelled, `handleReadError`
  recognises the resulting `ErrClosed` and falls through to the
  continuation: `startReadUntilEOF` is invoked with a fresh
  read-until-EOF-scoped context and the loop resumes until EOF or
  timeout.
* The harvester's publisher canceler is `nil` while the feature is
  enabled, so a `Publish` that returns after backpressure releases
  does not surface `context.Canceled` and short-circuit the
  continuation.
* The harvester task-group's stop deadline is extended by
  `read_until_eof.timeout + 100ms` so the drain is not killed prematurely.


Assisted-By: Claude Opus 4.7

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. Where relevant, I have used the stresstest.sh script to run them under stress conditions and race detector to verify their stability.
• I have added an entry in ./changelog/fragments using the changelog tool.

Disruptive User Impact

None, last lines on a file will not be missed and the input might take a bit longer to close.

The new option is enabled by default.

When the option is set, the only observable behaviour change is the input taking up to read_until_eof.timeout longer to exit on cancellation. Output / pipeline shutdown semantics are unchanged.

How to test this PR locally

The realistic scenario the feature targets is reproduced by TestFilestreamReadUntilEOFOnInputStop in filebeat/tests/integration/filestream_read_until_eof_test.go. To run it:

cd filebeat
mage buildSystemTestBinary
go test -v -race -tags=integration -run TestFilestreamReadUntilEOFOnInputStop ./tests/integration/...
go test -v -race -tags=integration -run TestFilestreamReadUntilEOFWithoutCloseOnEOF ./tests/integration/...

The faster in-process integration suite covers the timeout fallback, the no-close-on-EOF (tail) path, gzip files, file-deletion-during-drain, and a pure happy path:

go test -v -race -tags=integration -run "TestWaitUntilEOF" ./input/filestream/...

The unit suite locks down startReadUntilEOF semantics, handleReadError's ErrClosed bridge, and the config:

go test -v -race -run "TestConfigValidate/read_until_eof|TestLogFile_startReadUntilEOF|TestNewFileReader_startReadUntilEOFClosure|TestLogFile_readUntilEOFAfterReaderCtxCancel|TestNewFileReader_backoffWakesOnCanceler|TestFilestream_handleReadError" ./input/filestream/...

Manual end-to-end exercise (autodiscover-style):

1. Start filebeat with filebeat.config.inputs: { path: inputs.d/*.yml, reload.enabled: true } and an output that can be made to apply backpressure.
2. Drop a filestream input config in inputs.d/ with a large file, read_until_eof.enabled: true, read_until_eof.timeout: 1m.
3. Apply backpressure on the output (or use mock-es with percentTooMany=100).
4. Rename the input config (mv inputs.d/x.yml inputs.d/x.yml.disabled) so the reload removes the input.
5. Release backpressure. Filebeat should publish the rest of the file.
6. Without the feature, only events published before step 4 reach the output; with the feature, the entire file is delivered.

Related issues

• Closes [Filestream] Filestream add flag to only terminate the input after reaching EOF on all files #40447

Use cases

Kubernetes autodiscover with short-lived pods. A pod terminates while filebeat is still reading its log file. Without read_until_eof, the input is removed mid-read and the tail of the file is lost. With read_until_eof.enabled: true (the default), the harvester finishes the file before the input exits, and the Kubernetes-bound metadata is still attached because the input itself has not yet been torn down.

Reload-driven input removal in any deployment. Any time the input set is reduced (config reload, autodiscover, central management), the removed inputs now have a bounded grace window to finish their current file rather than dropping bytes.

## Screenshots

Logs

The drain emits three messages, with read_until_eof:

DEBUG: input closing, read_until_eof enabled, waiting EOF or 1m0s timeout, whichever happens first
DEBUG: read_until_eof enabled, EOF reached. closing input
INFO: read_until_eof enabled, 1m0s timeout reached. closing input

---

This is an automatic backport of pull request #50324 done by [Mergify](https://mergify.com).

[mergify]

Cherry-pick of 14ddacb has failed:

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

You are currently cherry-picking commit 14ddacbbc.
  (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/1778068402-filestream-read-until-eof.yaml
	modified:   filebeat/input/filestream/environment_test.go
	modified:   filebeat/input/filestream/filestream.go
	modified:   filebeat/input/filestream/filestream_test_non_windows.go
	modified:   filebeat/input/filestream/input_test.go
	new file:   filebeat/input/filestream/input_wait_until_eof_integration_test.go
	modified:   filebeat/input/filestream/internal/input-logfile/harvester.go
	new file:   filebeat/tests/integration/filestream_read_until_eof_test.go

Unmerged paths:
  (use "git add/rm <file>..." as appropriate to mark resolution)
	deleted by us:   docs/reference/filebeat/filebeat-input-filestream.md
	both modified:   filebeat/input/filestream/config.go
	both modified:   filebeat/input/filestream/config_test.go
	both modified:   filebeat/input/filestream/filestream_test.go
	both modified:   filebeat/input/filestream/input.go
	both modified:   filebeat/input/filestream/internal/input-logfile/input.go
	both modified:   filebeat/input/filestream/internal/input-logfile/manager.go
	both modified:   filebeat/testing/integration/log_generator.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!)
• /test : Run the Buildkite pipeline.

[infra-vault-gh-plugin-prod]

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

[infra-vault-gh-plugin-prod]

Pinging @elastic/obs-ds-hosted-services (Team:obs-ds-hosted-services)

[github-actions]

TL;DR

The Buildkite failure is caused by unresolved Git merge-conflict markers committed into this backport branch, so pre-commit fails immediately on check-merge-conflict. Resolve the conflicted files and push the conflict-resolved backport commit.

Remediation

• Resolve all conflict markers (<<<<<<<, =======, >>>>>>>) in the filestream files flagged by the hook, then commit the resolved content.
• Re-run pre-commit run --all-files (or /test) to confirm the check-merge-conflict hook passes.

Investigation details

Root Cause

This is a configuration/backport conflict-resolution issue (not a runtime code regression): the PR branch contains raw merge conflict markers.

Confirmed by CI log output from agentbeat: Run pre-commit and by reading PR head content:

• filebeat/input/filestream/config.go at PR head (refs/pull/51206/head) includes conflict markers around defaultConfig().
• Build log reports conflict markers across multiple files, for example:
  • filebeat/input/filestream/config_test.go:21
  • filebeat/input/filestream/config.go:102
  • filebeat/input/filestream/internal/input-logfile/input.go:35
  • filebeat/input/filestream/input.go:63
  • filebeat/input/filestream/filestream_test.go:31
  • filebeat/testing/integration/log_generator.go:162

Evidence

• Build: https://buildkite.com/elastic/beats/builds/47613
• Job/step: agentbeat: Run pre-commit → pre-commit run --all-files
• Key log excerpt:

check for merge conflicts................................................Failed
- hook id: check-merge-conflict
filebeat/input/filestream/config.go:102: Merge conflict string '<<<<<<<' found
filebeat/input/filestream/config.go:111: Merge conflict string '=======' found
filebeat/input/filestream/config.go:124: Merge conflict string '>>>>>>>' found

Verification

• Not run locally; diagnosis is conclusive from the failed hook output plus PR-head file content showing unresolved markers.

Follow-up

After conflict resolution, ensure the backported docs/file deletions referenced by Mergify’s conflict summary are reconciled consistently with 8.19 branch expectations before rerunning CI.

---

What is this? | From workflow: PR Buildkite Detective

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.

[mergify]

This pull request has not been merged yet. Could you please review and merge it @AndersonQ? 🙏

[mergify]

This pull request has not been merged yet. Could you please review and merge it @AndersonQ? 🙏

[vishaangelova]

The docs changes need to be ported to the asciidoc file: filebeat/docs/inputs/input-filestream.asciidoc.

[mergify]

This pull request has not been merged yet. Could you please review and merge it @AndersonQ? 🙏

[github-actions]

TL;DR

golangci-lint still fails on all three OS jobs, now with 10 remaining lint findings in the filestream backport. Finish the conflict-resolution cleanup in filebeat/input/filestream/..., then rerun the golangci-lint workflow.

Remediation

• Fix the 10 reported findings: remove the unused previousSrcIdentifiers field and unused logContains helper, check or intentionally discard the ch.Send(...) error in newInputACKHandler, replace deprecated io/ioutil test usage with os.CreateTemp, change assert.Nil(t, err) to assert.NoError(t, err), use require.Len, make the assert.ErrorIs message a string, lowercase the Can not start registry cleanup process error string, and rewrite the ErrClosed/read_until_eof branch so it no longer returns nil while handling a non-nil error.
• Re-run golangci-lint for PR #51206, or locally run the repository changed-file lint command against mergify/bp/8.19/pr-50324, to confirm the remaining lint set is clear.

Investigation details

Root Cause

This backport has progressed past the earlier typecheck and broad lint failures, but conflict-resolution cleanup is still incomplete. The current golangci-lint run reaches normal analysis and fails on 10 concrete findings across filestream production and test files.

Evidence

• Workflow: https://github.com/elastic/beats/actions/runs/28532173844
• PR: #51206, head elastic/beats:mergify/bp/8.19/pr-50324, base elastic/beats:8.19, author mergify[bot]
• Jobs/step: lint (ubuntu-latest), lint (windows-latest), and lint (macos-latest) -> golangci-lint
• Key log excerpt:

filebeat/input/filestream/internal/input-logfile/input.go:136:9: Error return value is not checked (errcheck)
filebeat/input/filestream/filestream_test.go:119:12: inline: Call of ioutil.TempFile should be inlined (govet)
filebeat/input/filestream/input.go:486:4: error is not nil (line 485) but it returns nil (nilerr)
filebeat/input/filestream/filestream_test.go:23:2: SA1019: "io/ioutil" has been deprecated since Go 1.19 (staticcheck)
filebeat/input/filestream/internal/input-logfile/manager.go:141:10: ST1005: error strings should not be capitalized (staticcheck)
filebeat/input/filestream/environment_test.go:405:3: len: use require.Len (testifylint)
filebeat/input/filestream/filestream_test.go:106:2: error-nil: use assert.NoError (testifylint)
filebeat/input/filestream/filestream_test.go:358:2: formatter: using msgAndArgs with non-string first element (msg) causes panic (testifylint)
filebeat/input/filestream/environment_test.go:550:35: func (*inputTestingEnvironment).logContains is unused (unused)
filebeat/input/filestream/internal/input-logfile/input.go:39:2: field previousSrcIdentifiers is unused (unused)
10 issues:
* errcheck: 1
* govet: 1
* nilerr: 1
* staticcheck: 2
* testifylint: 3
* unused: 2

Validation

• Not run locally; diagnosis is from the failed workflow logs plus PR-head source checks for the implicated code paths.

Follow-up

• This is not a duplicate of the latest prior PR Actions Detective comment, which covered a separate check-docs/make check struct-field mismatch. Compared with the earlier golangci-lint detective report, the lint set shrank from 15 to 10 findings, so this report reflects the current remaining blockers.

---

What is this? | From workflow: PR Actions Detective

Give us feedback! React with 🚀 if perfect, 👍 if helpful, 👎 if not.

[mergify]

Tick the box to add this pull request to the merge queue (same as @mergifyio queue).

• Queue this pull request