[libbeat/processing] add a new shared processor wrapper

[VihasMakwana]

Proposed commit message

Right now, per input processors are instantiated for every new harvester created.

beats/filebeat/channel/runner.go

Lines 136 to 150 in c37e794

	return func(clientCfg beat.ClientConfig) (beat.ClientConfig, error) {
	var indexProcessor beat.Processor
	if !config.Index.IsEmpty() {
	staticFields := fmtstr.FieldsForBeat(beatInfo.Beat, beatInfo.Version)
	timestampFormat, err := fmtstr.NewTimestampFormatString(&config.Index, staticFields)
	if err != nil {
	return clientCfg, err
	}
	indexProcessor = add_formatted_index.New(timestampFormat)
	}
	userProcessors, err := processors.New(config.Processors, beatInfo.Logger)
	if err != nil {
	return clientCfg, err
	}

In environments with a large number of log files, this will lead to excessive memory usage and can cause memory exhaustion/OOM issues.

Fix this by adding a shared processor wrapper, that will create one processor instance for a given config, no matter how many times it was instantiated.

Why is it important

Processors interested in using a "shared" state can opt-in by wrapping their constructors:

Before:

processors.RegisterPlugin("add_kubernetes_metadata", New)

After:

processors.RegisterPlugin("add_kubernetes_metadata", shared.New(New))

This will result in much less memory allocation over long term.

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.

How to test this PR locally

You can test this with following config:

filebeat.inputs:

- type: filestream
  id: my-filestream-id1
  enabled: true
  paths:
    - /path/to/logs/*.log
  processors:
    - add_kubernetes_metadata

output.console:
  enabled: true
  # path: "/tmp/filebeat"
logging.level: debug

With new files being created, the memory increase will still be there, but slow and steady. Without my changes, the usage is drastic.
It is more drastic if you're testing this in k8s environment with add_kubernetes_metadata processor.

Related issues

• Related [kubernetes/container_logs]: Allow creating single input to monitor all log files integrations#17250

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

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @VihasMakwana? 🙏.
For such, you'll need to label your PR with:

• The upcoming major version of the Elastic Stack
• The upcoming minor version of the Elastic Stack (if you're not pushing a breaking change)

To fixup this pull request, you need to add the backport labels for the needed
branches, such as:

• backport-8./d is the label to automatically backport to the 8./d branch. /d is the digit
• backport-active-all is the label that automatically backports to all active branches.
• backport-active-8 is the label that automatically backports to all active minor branches for the 8 major.
• backport-active-9 is the label that automatically backports to all active minor branches for the 9 major.

[github-actions]

TL;DR

Libbeat: Run check/update failed because the new file libbeat/processors/shared/shared.go is missing the required Apache license header, so go-licenser stops make -C libbeat check update.

Remediation

• Add the standard Elasticsearch/Apache-2.0 header block at the top of libbeat/processors/shared/shared.go (before package shared).
• Re-run make -C libbeat check update and make check-no-changes to confirm header checks pass.

Investigation details

Root Cause

The header check stage (check-headers) failed on the new shared processor file. In the current PR commit (61b4346e79f1a7bf4cb65b3dd4e869963f6a2229), libbeat/processors/shared/shared.go starts directly with package shared and contains no license header, which violates the repository header policy.

Evidence

• Build: https://buildkite.com/elastic/beats/builds/46148
• Job/step: Libbeat: Run check/update (beats-libbeat build 29678)
• Key log excerpt:
  • processors/shared/shared.go: is missing the license header
  • Error: running "go-licenser -d -license ASL2" failed with exit code 1
  • make: *** [scripts/Makefile:146: check-headers] Error 1

Verification

• Not run locally in this workflow environment.

Follow-up

• After adding the header, re-trigger CI; this failure should clear unless another new file is also missing headers.

Note

🔒 Integrity filter blocked 3 items

The following items were blocked because they don't meet the GitHub integrity level.

• [libbeat/processing] add a new shared processor wrapper #50713 pull_request_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• #50713 pull_request_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".
• [libbeat/processing] add a new shared processor wrapper #50713 issue_read: has lower integrity than agent requires. The agent cannot read data with integrity below "approved".

To allow these resources, lower min-integrity in your GitHub frontmatter:

tools:
  github:
    min-integrity: approved  # merged | approved | unapproved | none

---

What is this? | From workflow: PR Buildkite Detective

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

[orestisfl]

Approach LGTM

[belimawr]

So far it looks great!

[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: 383acd5b-14bc-4617-9fa7-d40b21d49693

📥 Commits

Reviewing files that changed from the base of the PR and between b56b327 and 1ba5101.

📒 Files selected for processing (2)

• libbeat/processors/shared/shared.go
• libbeat/processors/shared/shared_test.go

🚧 Files skipped from review as they are similar to previous changes (1)

• libbeat/processors/shared/shared_test.go

---

📝 Walkthrough

Walkthrough

This PR adds a shared processor wrapper that caches processor instances by a configuration hash with reference counting and lifecycle-managed Close. The kubernetes metadata processor is registered using shared.New(New) and a test helper signature was fixed. New concurrency tests for both the shared wrapper and the kubernetes annotator validate caching, reference-counted Close, and event isolation under concurrent Run/Close and concurrent cache mutations.

🚥 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.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ 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 and usage tips.}

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@libbeat/processors/shared/shared.go`:
- Around line 59-61: The sharedProcessors map currently hands out the same
wrapper instance so duplicate Close() calls from different borrowers can
decrement p.refCount multiple times; fix by introducing a per-borrower handle
type (e.g., sharedHandle) that holds a pointer to the shared entry (e.g.,
sharedEntry with proc and refCount) plus its own closed flag and mutex, have
Borrow/lookup (the code around sharedProcessors[hash] and p.refCount) increment
only sharedEntry.refCount and return a new sharedHandle; implement
sharedHandle.Close() to be idempotent by checking its closed flag, setting it
once, then decrementing sharedEntry.refCount and when that hits zero close the
underlying processor and remove the sharedEntry from sharedProcessors (also
protect shared map and refCount with the existing lock used in the code paths
around lines 59 and 79-90).

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Enterprise

Run ID: 9b242bef-2171-4fa8-825f-a107ce0ae187

📥 Commits

Reviewing files that changed from the base of the PR and between 01f436f and 9cb291f.

📒 Files selected for processing (5)

• libbeat/processors/add_kubernetes_metadata/kubernetes.go
• libbeat/processors/add_kubernetes_metadata/kubernetes_concurrent_test.go
• libbeat/processors/add_kubernetes_metadata/kubernetes_test.go
• libbeat/processors/shared/shared.go
• libbeat/processors/shared/shared_test.go

[VihasMakwana]

I'm also backporting this to 9.3.