processors: close when construction fails partway

[orestisfl]

Proposed commit message

libbeat/processors: close processors when construction fails partway

Processor construction had three error paths that dropped
already-constructed processors without calling Close, leaking any
resources they held for the lifetime of the process:

- processors.New: while building a list, a processor that failed to
  construct caused the function to return without closing the
  processors built earlier in the same list.
- NewConditional: every registered processor is wrapped as
  NewConditional(SafeWrap(constructor)). The inner constructor runs
  first, then the `when` condition is parsed. A malformed condition
  returned an error but left the already-constructed processor open.
- NewIfElseThenProcessor: if the `else` branch failed to build, the
  already-built `then` processors were dropped without Close.

The fix closes the already-constructed processors on each of these error
paths.

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. This only affects error paths during processor construction and does not change behavior for successfully built processors. The leaked resources are now released instead of being held for the process lifetime.

How to test this PR locally

Run the package tests, including the two new tests that assert the constructed processor is closed on each error path:

cd libbeat
go test -race -run 'TestConditionErrorClosesProcessor|TestIfElseThenErrorClosesThenProcessors' ./processors/...
go test -race ./processors/...

Reproduce the leak with a real Filebeat config

This needs no Docker. add_docker_metadata starts a background reconnect goroutine at construction that is stopped only by Close(); with wait_for_metadata_timeout: 0 it retries forever. The empty when condition fails to build after that processor has already been constructed, so before this change the processor was dropped without Close() and its reconnect goroutine was orphaned for the lifetime of the process. The leak is an orphaned goroutine, so it is observed through the goroutine profile rather than through the config or the logs.

mkdir -p /tmp/fb-repro/data
for i in $(seq 1 60); do printf 'log line %04d filler filler filler filler filler filler filler\n' "$i"; done > /tmp/fb-repro/log.txt

/tmp/fb-repro/filebeat.yml:

filebeat.inputs:
  - type: filestream
    id: leaking-input
    enabled: true
    paths:
      - /tmp/fb-repro/log.txt
    processors:
      - add_docker_metadata:
          host: "unix:///var/run/nonexistent-docker.sock"
          wait_for_metadata_timeout: 0
          when: {}

output.file:
  path: /tmp/fb-repro/out
  filename: filebeat-out

http.enabled: true
http.host: localhost
http.port: 5066
http.pprof.enabled: true

Run filebeat, then read the goroutine profile:

./filebeat -e -c /tmp/fb-repro/filebeat.yml \
  --path.home /tmp/fb-repro --path.data /tmp/fb-repro/data --strict.perms=false &
sleep 10
curl -s "http://localhost:5066/debug/pprof/goroutine?debug=1" | grep -c retryConnectToDocker

On main the grep -c prints 1: the reconnect goroutine is orphaned and remains in the profile, with a parentless stack:

1 @ ...
#	...add_docker_metadata.(*addDockerMetadata).retryConnectToDocker	add_docker_metadata.go:227
#	...startDockerConnectionRetry.func3				add_docker_metadata.go:187
#	sync.(*WaitGroup).Go.func1					waitgroup.go:258

With this change the constructed processor is closed on the error path and the count is 0.

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

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: 422e2e7b-49aa-4004-b079-bb7f4f9a6037

📥 Commits

Reviewing files that changed from the base of the PR and between 2f63f0e and bb3b354.

📒 Files selected for processing (1)

• libbeat/processors/conditionals_test.go

👮 Files not reviewed due to content moderation or server errors (1)

• libbeat/processors/conditionals_test.go

---

📝 Walkthrough

[!WARNING]

Walkthrough skipped

File diffs could not be summarized.

🚥 Pre-merge checks | ✅ 2

✅ Passed checks (2 passed)

Check name	Status	Explanation
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

• 🛠️ Update Documentation

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[Copilot]

Pull request overview

This PR prevents resource leaks when processor construction fails partway through building a processor chain in libbeat/processors, ensuring already-constructed processors are properly Close()d on error paths.

Changes:

• Close partially-constructed processor lists on processors.New failures.
• Close already-constructed processors when conditional parsing fails (NewConditional) and when if/then/else construction fails while building the else branch.
• Add unit tests covering two of the fixed error paths and include a changelog fragment.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

File	Description
libbeat/processors/processor.go	Adds an abort helper to close already-built processors when list construction errors.
libbeat/processors/conditionals.go	Ensures constructed processors are closed if condition parsing or else branch construction fails.
libbeat/processors/conditionals_test.go	Adds tests asserting processors are closed on the new conditional and if/then/else error paths.
changelog/fragments/1783070100-close-processors-on-construction-errors.yaml	Adds a changelog entry documenting the leak fix.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[mergify]

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

• Queue this pull request

[github-actions]

@Mergifyio backport 9.4 9.3 8.19

[mergify]

backport 9.4 9.3 8.19

✅ Backports have been created

Details

• #51725 [9.4](backport #51687) processors: close when construction fails partway has been created for branch 9.4
• #51726 [9.3](backport #51687) processors: close when construction fails partway has been created for branch 9.3
• #51727 [8.19](backport #51687) processors: close when construction fails partway has been created for branch 8.19