Add SafeProcessor wrapper

[rdner]

What does this PR do?

Each processor might end up in multiple processor groups. Every group has its own Close that calls Close on each processor of that group which leads to multiple Close calls on the same processor.

If the SafeProcessor wrapper was not used,
the processor would have to handle multiple Close calls with adding sync.Once in its Close function.

We make it easer for processor developers and take care of it in the processor registry instead.

Why is it important?

Our customers could see panics in logs which makes it noisy.

Also, there are reports of higher impact:

• add_kubernetes_metadata panics when a Beat shuts down #34219 (comment)
• K8s Integration does not report correct container.id when container restarts integrations#5348 (comment)

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
• I have added an entry in CHANGELOG.next.asciidoc or CHANGELOG-developer.next.asciidoc.

How to test this PR locally

Unfortunately, it's very difficult to reproduce the panic described in the issue, it seems to be a result of some kind of race condition on Beats shutting down.

The change is covered with unit tests and I made sure that a simple processor with the following config still works and there is no regression:

filebeat.inputs:
  - type: filestream
    id: my-filestream-id
    enabled: true
    paths:
      - "metadata-panic/logs/log*.ndjson"
    processors:
      - add_fields:
          target: project
          fields:
            name: myproject
            id: 'UNIQUE_VALUE'

path.data: "metadata-panic/data"
logging:
  level: debug
output.console:
  enabled: true

and I got this event published:

{
  "@timestamp": "2023-02-23T10:33:22.673Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "8.8.0"
  },
  "log": {
    "offset": 32170800,
    "file": {
      "path": "metadata-panic/logs/log1.ndjson"
    }
  },
  "message": "{\"data\": \"sample\"}",
  "input": {
    "type": "filestream"
  },
  "project": {
    "name": "myproject",
    "id": "UNIQUE_VALUE"
  },
  "agent": {
    "ephemeral_id": "926e973d-f35b-4f1e-95aa-6b9cf239e21c",
    "id": "2f6fa2ce-1a37-4259-91e8-1df581639d07",
    "name": "MacBook-Pro.localdomain",
    "type": "filebeat",
    "version": "8.8.0"
  },
  "ecs": {
    "version": "8.0.0"
  },
  "host": {
    "name": "MacBook-Pro.localdomain"
  }
}

Note

"project": {
    "name": "myproject",
    "id": "UNIQUE_VALUE"
  }

Related issues

• Closes add_kubernetes_metadata panics when a Beat shuts down #34219

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @rdner? 🙏.
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-v8./d.0 is the label to automatically backport to the 8./d branch. /d is the digit

[elasticmachine]

💚 Build Succeeded

the below badges are clickable and redirect to their specific view in the CI or DOCS

Expand to view the summary

Build stats

• Start Time: 2023-02-24T08:20:13.190+0000

• Duration: 66 min 5 sec

Test stats 🧪

Test	Results
Failed	0
Passed	25993
Skipped	1962
Total	27955

💚 Flaky test report

Tests succeeded.

🤖 GitHub comments

Expand to view the GitHub comments

To re-run your PR in the CI, just comment with:

• /test : Re-trigger the build.

• /package : Generate the packages and run the E2E tests.

• /beats-tester : Run the installation tests with beats-tester.

• run elasticsearch-ci/docs : Re-trigger the docs validation. (use unformatted text in the comment!)

[elasticmachine]

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

[rdner]

@cmacknz there will be no 8.6 releases anymore, I don't think we need that backport label.

[leehinman]

code LGTM.

One optional nit on the name SafeProcessor. The name doesn't tell me what it is safe from, so if you just encountered this in registry.go you wouldn't have many clues as to why the constructor is being wrapped. Names like CloseOnce or AtomicCloser might be better.

[rdner]

@leehinman the name comes from the fact that it's not just addressing the Close function call but also the Run – so, you can't run the processor that was closed before and cause some panics by doing so.

I've put a lot of thought into the naming and I'd rather keep it this way, so we could perhaps extend the functionality of this wrapper in the future, e.g. blocking certain events from being processed by any processor, etc.