[9.1](backport #47905) [Filebeat] Correctly report V2 inputs as failed to the Manager

[mergify[bot]]

Proposed commit message

Correctly report V2 inputs as failed to the manager. If a V2 input Run
method returns an error, it is now correctly reported to the input
manager. The Filestream input reports the input as degraded if the
pipeline fails to connect.

Checklist

• [x] My code follows the style guidelines of this project
• [x] I have commented my code, particularly in hard-to-understand areas
• [x] I have made corresponding changes to the documentation
• [x] I have made corresponding change to the default configuration files
• [x] 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.
• [x] I have added an entry in ./changelog/fragments using the changelog tool.

~~## Disruptive User Impact~~ ~~## Author's Checklist~~

How to test this PR locally

Run the tests

cd x-pack/filebeat
mage buildSystemTestBinary
go test -tags=integration -v -count=1 -run=TestPipelineConnectionError ./tests/integration

Manual test

1. Download Elastic Agent snapthost

Make sure to download the latest snapshot in the correct version for your OS/Architecture

wget https://artifacts-snapshot.elastic.co/elastic-agent-package/9.3.0-9fb1a0c3/downloads/beats/elastic-agent/elastic-agent-9.3.0-SNAPSHOT-linux-x86_64.tar.gz

2. Build Agentbeat

cd x-pack/agentbeat
DEV=true mage -v build

3. Extract Elastic Agent and copy the Agentbeat binary

# Extract Elastic Agent using your favorite tool
cd elastic-agent-9.3.0-SNAPSHOT-linux-arm64/data/elastic-agent-24d365/components

4. Copy the Agentbeat binary you built, replacing the one from the archive

cp ~/devel/beats/x-pack/agentbeat/agentbeat .
cd ../../../

5.Create a log file for Filestream to ingest

Using flog

docker run -it --rm mingrammer/flog -n 50 > /tmp/flog.log

6.Create the elastic-agent.yml

The connection credentials to Elasticsearch are irrelevant, no data will be shipped, so they can point to an inexistent Elasticsearch host.

elastic-agent.yml

outputs:
  default:
    type: elasticsearch
    hosts: [http://localhost:9200]

inputs:
  - type: filestream
    id: filestream-unit-id
    type: filestream
    streams:
      - type: filestream
        id: filestream-input-id
        paths:
          - /tmp/flog.log
        processors:
          - add_fields:
              INVALID_CONFIG_KEY: true
              fields:
                labels:
                  foo: bar
  - type: cel
    id: cel-unit-id
    type: cel
    streams:
      - type: cel
        id: cel-input-id
        interval: 1m
        resource.url: https://api.ipify.org/?format=text
        program: |
          {"events": [{"ip": string(get(state.url).Body)}]}
        processors:
          - add_fields:
              INVALID_CONFIG_KEY: true
              fields:
                labels:
                  foo: bar

agent.monitoring:
  enabled: false
  logs: false
  http:
    enabled: false

agent.grpc:
  address: localhost
  port: 4242

agent.logging.level: info
agent.logging.to_stderr: true

7.Look for the logs showing Filestream and Cel input failing

Filestream:

{
  "@timestamp": "2025-12-08T09:56:57.403-0500",
  "component": {
    "id": "filestream-default",
    "state": "HEALTHY"
  },
  "ecs.version": "1.6.0",
  "log": {
    "source": "elastic-agent"
  },
  "log.level": "warn",
  "log.origin": {
    "file.line": 1040,
    "file.name": "coordinator/coordinator.go",
    "function": "github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator.logComponentStateChange"
  },
  "message": "Unit state changed filestream-default-filestream-unit-id (HEALTHY->DEGRADED): Harvester for Filestream input \"filestream-input-id\" failed: error while connecting to output with pipeline: unexpected INVALID_CONFIG_KEY option in processors.6.add_fields",
  "unit": {
    "id": "filestream-default-filestream-unit-id",
    "old_state": "HEALTHY",
    "state": "DEGRADED",
    "type": "input"
  }
}

Cel:

{
  "@timestamp": "2025-12-08T09:56:57.761-0500",
  "component": {
    "id": "cel-default",
    "state": "HEALTHY"
  },
  "ecs.version": "1.6.0",
  "log": {
    "source": "elastic-agent"
  },
  "log.level": "error",
  "log.origin": {
    "file.line": 1040,
    "file.name": "coordinator/coordinator.go",
    "function": "github.com/elastic/elastic-agent/internal/pkg/agent/application/coordinator.logComponentStateChange"
  },
  "message": "Unit state changed cel-default-cel-unit-id (HEALTHY->FAILED): Input 'cel' failed with: input cel-input-id failed: unexpected INVALID_CONFIG_KEY option in processors.6.add_fields",
  "unit": {
    "id": "cel-default-cel-unit-id",
    "old_state": "HEALTHY",
    "state": "FAILED",
    "type": "input"
  }
}

8.Look at the Elastic Agent status

The Filestream input must be in a degraded state and Cel in a failed state. Both inputs must have errored with

unexpected INVALID_CONFIG_KEY option in processors...

% ./elastic-agent status
┌─ fleet
│  └─ status: (STOPPED) Not enrolled into Fleet
└─ elastic-agent
   ├─ status: (DEGRADED) 1 or more components/units in a failed state
   ├─ cel-default
   │  ├─ status: (HEALTHY) Healthy: communicating with pid '53315'
   │  └─ cel-default-cel-unit-id
   │     └─ status: (FAILED) Input 'cel' failed with: input cel-input-id failed: unexpected INVALID_CONFIG_KEY option in processors.6.add_fields
   └─ filestream-default
      ├─ status: (HEALTHY) Healthy: communicating with pid '53289'
      └─ filestream-default-filestream-unit-id
         └─ status: (DEGRADED) Harvester for Filestream input "filestream-input-id" failed: error while connecting to output with pipeline: unexpected INVALID_CONFIG_KEY option in processors.6.add_fields

Related issues

• Closes https://github.com/elastic/beats/issues/45649

~~## Use cases~~ ~~## Screenshots~~ ~~## Logs~~

---

This is an automatic backport of pull request #47905 done by Mergify.

[github-actions[bot]]

:robot: 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]

Blocked until the backport for https://github.com/elastic/beats/pull/48075 gets merged

[mergify[bot]]

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

[belimawr]

I'm waiting until https://github.com/elastic/beats/pull/48089 is merged and backported to fully fix panics caused by the nil status reporter.

[mergify[bot]]

This pull request is now in conflicts. Could you fix it? 🙏 To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b mergify/bp/9.1/pr-47905 upstream/mergify/bp/9.1/pr-47905
git merge upstream/9.1
git push upstream mergify/bp/9.1/pr-47905