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

[mergify]

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

• 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
## 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 Agent state is Healthy when input fails to start #45649

## Use cases
## Screenshots
## Logs

---

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

[mergify]

Cherry-pick of 462751c has failed:

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

You are currently cherry-picking commit 462751c83.
  (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/1765204121-Correctly-report-V2-inputs-as-failed-to-the-Manager.yaml
	modified:   filebeat/input/filestream/internal/input-logfile/harvester.go
	modified:   filebeat/input/filestream/internal/input-logfile/harvester_test.go
	modified:   filebeat/input/filestream/internal/input-logfile/input.go
	modified:   filebeat/input/v2/compat/compat.go
	modified:   filebeat/input/v2/input-cursor/input.go

Unmerged paths:
  (use "git add <file>..." to mark resolution)
	both modified:   x-pack/filebeat/tests/integration/managerV2_test.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]

Blocked until the backport for #48075 gets merged

[mergify]

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