[filebeat][gcs] Fix for concurrency issues and context timeouts in the GCS input

[ShourieG]

Type of change

• Bug

What does this PR do?

1. It fixes a map concurrency issue which was revealed when testing the input at scale.
2. Fixes the issue were bucket_timeout was being applied to the bucket poll interval instead of individual bucket object read ops.
3. Fixes flakey test issue.

Why is it important?

This relates to an SDH from one of our customers and fixes those issues.
This also addresses any issues the input could have at scale.

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.

[elasticmachine]

Pinging @elastic/security-external-integrations (Team:Security-External Integrations)

[mergify]

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @ShourieG? 🙏.
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-06-01T08:13:47.883+0000

• Duration: 74 min 39 sec

Test stats 🧪

Test	Results
Failed	0
Passed	2959
Skipped	173
Total	3132

💚 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!)

[ShourieG]

@efd6 , I've addressed the changes and updated the PR

[ShourieG]

@efd6, Updated the changes and left comment regarding #35605 (comment)

[efd6]

I think I'm OK with the concurrency fix, mod the unlocked use of the *Checkpoint returned by *state.checkpoint(). I think that method needs documentation clearly stating that it's not safe for concurrent use and that the returned value can not be used while the receiver in operating concurrently. Similar for *state.setCheckpoint().

Overall, I'd like to see this broken into separate PRs.

Please wait for @andrewkroh

[ShourieG]

@efd6, I've removed the enhancements from this PR, I will put up another PR post merging of this PR which will contain those enhancements.

[efd6]

Thanks. Please still wait for Andrew.

[mergify]

This pull request is now in conflicts. Could you fix it? 🙏
To fixup this pull request, you can check out it locally. See documentation: https://help.github.com/articles/checking-out-pull-requests-locally/

git fetch upstream
git checkout -b gcs/enhancement upstream/gcs/enhancement
git merge upstream/main
git push upstream gcs/enhancement

[ShourieG]

@andrewkroh does the PR look good for merging at the moment ?

[andrewkroh]

This looks like a good improvement in addressing the concurrency issues. So if you wanted to go ahead as is that's good with me.

But I think there may be some correctness issues w.r.t. the comment I raised. This could be apparent during shutdown where events in the queue are discarded, but the state was already updated with the assumption that the events where written to ES.

[andrewkroh]

What are the assumptions that you are making about the published state of the event after Publish returns? I see this comment about it being "saved and published" and that's not correct. So if that is important to your implementation you'll need changes.

When Publish returns the event has been queued internally. It will have undergone processing via Beats processors then put onto the queue. If it needs to know when the event has been accepted by the destination output then you can register and EventListener to get a callback.

beats/libbeat/beat/pipeline.go

Lines 60 to 61 in 327b5fe

	// Callbacks for when events are added / acknowledged
	EventListener EventListener

As an example, the GCP pub/sub input uses an EventListener to defer ACK'ing the pubsub message until the event has been written to ES.

beats/x-pack/filebeat/input/gcppubsub/input.go

Lines 117 to 121 in 78dc664

	EventListener: acker.ConnectionOnly(
	acker.EventPrivateReporter(func(_ int, privates []interface{}) {
	for _, priv := range privates {
	if msg, ok := priv.(*pubsub.Message); ok {
	msg.Ack()

[ShourieG]

@andrewkroh thanks for the feedback, I will revisit this when I do some planned feature enhancements for the gcs input in the near future, will have to separate out the cursor publish from the event publish to properly implement this feature. For now I'm merging then PR since it's related to an sdh.