[libbeat/processing] Deduplicate processors based on config

[orestisfl]

Proposed commit message

Deduplicate processor instances in SafeWrap using (name, hash).
Constructor calls with an identical configuration return handles to one
shared, reference-counted instance.

Beat-level global processors were already shared before this PR so most
processors should already work with that level of concurrency.

Design points:

- Exceptions:
  1. PathSetter implementations
  2. The rate_limit processor: a shared token bucket would divide the
     configured per-input limit by the number of owners.
- Each constructor call returns a handle; Close is idempotent per
  handle.
- Concurrent constructors sharing the same key wait for a single
  instance to initialize. Construction is done outside the global lock
  to avoid one processor blocking all constructors.

This reverts #50713, which shared only add_kubernetes_metadata through a
dedicated wrapper package, cherry-picks and hardens the approach from
#50586.

Processors that need to opt-out of being shared can implement the
Unshareable interface.

Closes #50376

Co-authored-by: Vihas Makwana <vihas.makwana@elastic.co>

Note

I kept four commits for cleaner reviewing:

1. Contract tests for processors
2. Revert [libbeat/processing] add a new shared processor wrapper #50713
3. Re-introduce [libbeat/processing] create a unique processor for a given config #50586
4. Harden

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 directly but indirectly, inputs with identical processor configurations now share one instance.

How to test this PR locally

e2e

Filebeat config:

filebeat.inputs:
  - type: filestream
    id: metadata-repro
    paths: ["/tmp/filebeat-processors-repro/logs/*.log"]
    processors:
      - add_docker_metadata:
          host: unix:///tmp/filebeat-repro-no-docker.sock
          wait_for_metadata_timeout: 0
          wait_for_metadata_retry_period: 1h

output.discard.enabled: true
http.enabled: true
http.host: 127.0.0.1
http.port: 5066
logging.level: error
logging.to_files: false

Create files:

ROOT=/tmp/filebeat-processors-repro
mkdir -p "$ROOT/logs"
for i in $(seq 1 120); do
  printf 'container-%03d %01024d\n' "$i" 0 > "$ROOT/logs/$i.log"
done

Run each binary separately:

mkdir -p "$ROOT"/{home,data,logs}

systemd-run --user --wait --pipe \
  --unit="filebeat-repro" \
  --property=MemoryAccounting=yes \
  "$ROOT/filebeat" -e --strict.perms=false -c "$ROOT/filebeat.yml" \
  --path.home="$ROOT/home" --path.data="$ROOT/data" \
  --path.logs="$ROOT/logs" >"$ROOT/run.txt" 2>&1 &
runner=$!

until stats=$(curl --fail --silent http://127.0.0.1:5066/stats) &&
  [ "$(jq -r '.filebeat.harvester.running' <<<"$stats")" = 120 ]; do
  sleep 0.1
done
jq '{harvesters: .filebeat.harvester.running,
     goroutines: .beat.runtime.goroutines}' <<<"$stats"
systemctl --user stop "filebeat-repro.service"
wait "$runner" || true
grep 'Memory peak' "$ROOT/run.txt"

At 120 active harvesters	main	This PR
Goroutines	383	264
Cgroup memory peak	39.4–40.7 MB	31.9 MB

Unit tests

go test -race ./libbeat/processors ./filebeat/channel
timeout 120 ./script/stresstest.sh --race ./libbeat/processors '^TestContract|^TestShared|^TestSafe' -p 8
timeout 120 ./script/stresstest.sh --race ./filebeat/channel '^Test' -p 8

The contract tests (first commit) were written black-box against main before the implementation existed and pass unmodified on both.

Related issues

• Closes add_kubernetes_metadata: Memory and harvester unexpected increase when using filestream #50376
• Closes [beatreceiver] - Wrap add_fields with sharedprocessor #51616
• Replaces [libbeat/processing] create a unique processor for a given config #50586, (Closes filebeat: build per-input processors once, not per harvester #50611) and [libbeat/processing] add a new shared processor wrapper #50713 (reverted in this PR)

[infra-vault-gh-plugin-prod]

Pinging @elastic/elastic-agent-data-plane (Team:Elastic-Agent-Data-Plane)

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

[cmacknz]

I think the approach makes sense, still digesting the concurrency interactions but found only some potential ref counting bugs on the first pass.

[orestisfl]

@cmacknz your comments are correct in the sense that the evicted core will not have an accurate refCount but an evicted core should never be re-used. I decided to not change the code to update the refCount because we would need to re-lock and a reader might question why we want to keep an accurate refCount for evicted cores.

I refactored some code around the construct-or-wait logic and commented the paths you highlighted.

[cmacknz]

LGTM, will let @khushijain21 do the approval since she had some comments to address as well.

[mergify]

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

• Queue this pull request