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

[mergify]

Proposed commit message

Add Filestream harvester ingestion progress metrics

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 (69c5d2b37992300ada506ec95b35e1859dfaaf3c), 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

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

How to test this PR locally

Run the tests

Run the focused unit tests that cover the new Filestream harvester progress metrics:

cd filebeat
go test -v -count=1 ./input/filestream/internal/input-logfile/... ./input/filestream/... \
  -run 'TestHarvesterMetricsUpdate|TestHarvesterMetricsAddFile|TestHarvesterOffsetRegistration|TestMetricsCleanup|TestAggregateGaugeLifecycle|TestFileWatcherHarvesterMetrics|TestFileWatcherRunCleansHarvesterMetricsOnShutdown|TestReadFromSourceUpdatesActiveOffset'

Run the broader touched package set:

cd filebeat
go test -v -count=1 ./input/filestream/internal/input-logfile/... ./input/filestream/...

Manual test

1. Create test files:

export TEST_DIR="$(mktemp -d)"
mkdir -p "${TEST_DIR}/input-files"
flog -n 20 > "${TEST_DIR}/input-files/complete.log"
flog -n 20 > "${TEST_DIR}/input-files/excluded.log"
flog -n 20 > "${TEST_DIR}/input-files/old.log"
flog -n 2 > "${TEST_DIR}/input-files/lagging.log"
touch -d '2 hours ago' "${TEST_DIR}/input-files/old.log"

# Write down your temp dir:
echo "${TEST_DIR}/input-files"

2. Write the Filebeat configuration:

filebeat.yml

cat > "${TEST_DIR}/filebeat.yml" <<EOF
filebeat.inputs:
  - type: filestream
    id: "test-filestream-harvester-metrics"
    paths:
      - "${TEST_DIR}/input-files/*.log"
    prospector.scanner.exclude_files: ['excluded\.log$']
    ignore_older: 1h
    prospector.scanner.check_interval: 1s
    close.on_state_change.inactive: 1m
    processors:
      - script:
          lang: javascript
          tag: slow_filestream_processing
          timeout: 2s
          params:
            delay_ms: 200
          source: >
            var params = { delay_ms: 200 };
            function register(scriptParams) {
              params = scriptParams;
            }
            function process(event) {
              var until = Date.now() + params.delay_ms;
              while (Date.now() < until) {}
              return event;
            }

path.home: "${TEST_DIR}"

queue.mem:
  flush.timeout: 0s

output.file:
  path: "${TEST_DIR}"
  filename: "output-file"

http:
  enabled: true
  port: 5066

logging:
  level: debug
  metrics:
    enabled: true
    period: 1s
EOF

The JavaScript processor intentionally busy-waits for delay_ms per event. This
keeps ingestion behind file growth long enough to observe the bucket gauges. Keep
delay_ms below the processor timeout.

3. Build and run Filebeat:

cd filebeat
go build .
./filebeat --strict.perms=false -c "${TEST_DIR}/filebeat.yml"

4. Start monitoring the metrics:

In another terminal run the following command and start watching the metrics

while true; do curl -s http://localhost:5066/stats | jq -c -S '.filebeat.filestream | { files_ingested_percent_100, files_ingested_percent_95_99, files_ingested_percent_lt_95, files_ignored} '; sleep 1; done

You should see the beginning of the metrics:

{"files_ignored":null,"files_ingested_percent_100":null,"files_ingested_percent_95_99":null,"files_ingested_percent_lt_95":null}
{"files_ignored":2,"files_ingested_percent_100":0,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":0}
{"files_ignored":2,"files_ingested_percent_100":0,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}
{"files_ignored":2,"files_ingested_percent_100":0,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}
{"files_ignored":2,"files_ingested_percent_100":0,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}
{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":0}
{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":0}

5. In another terminal start adding data to a file:

Substitute /tmp/tmp.ZUzKF6JUG3/input-files/ by the temp dir from step #1

flog -d 0.1 -s 0.1 |tee -a /tmp/tmp.ZUzKF6JUG3/input-files/lagging.log

You will see the metrics changing as the file is ingested:

{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}
{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}

Let it run for a less than a minute and stop flog, you'll see that one file progressing in ingestion until fully ingested:

{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}
{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":1}
{"files_ignored":2,"files_ingested_percent_100":1,"files_ingested_percent_95_99":1,"files_ingested_percent_lt_95":0}
{"files_ignored":2,"files_ingested_percent_100":2,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":0}
{"files_ignored":2,"files_ingested_percent_100":2,"files_ingested_percent_95_99":0,"files_ingested_percent_lt_95":0}

6. You can also look at the monitoring logs:

cat "${TEST_DIR}"/logs/*.ndjson \
  | grep monitoring \
  | jq -c -S '.monitoring.metrics.filebeat.filestream
      | {
          files_ingested_percent_100,
          files_ingested_percent_95_99,
          files_ingested_percent_lt_95,
          files_ignored
        }'

Expected meaning:

• complete.log will be counted as files_ingested_percent_100 once it has been ingested.
• excluded.log is ignored, so it is not counted by any files_ingested_percent_* bucket.
• The ignored file still contribute to files_ignored.
• lagging.log is excluded from all files_ingested_percent_* until it is large enough to start ingestion.

As you add data to lagging.log it will start contributing to the metrics and will move through them as the data is ingested.

Related issues

• Part of Add more metrics for file ingestion #36653
• Blocked by Add Filestream scanner metrics #50963

## Use cases
## Screenshots
## Logs

---

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

[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)

[github-actions]

🔍 Preview links for changed docs

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

[github-actions]

✅ Elastic Docs Style Checker (Vale)

No issues found on modified lines!

---

The Vale linter checks documentation changes against the Elastic Docs style guide. To use Vale locally or report issues, refer to Elastic style guide for Vale.