x-pack/filebeat/input/awss3: allow a grace time on shutdown #43369

efd6 · 2025-03-20T02:05:21Z

Proposed commit message

x-pack/filebeat/input/awss3: allow a grace time on shutdown

We may have SQS message requests in flight that are being serviced when the
input's context is cancelled. This allows the user to specify a time to wait
after the context is cancelled before the processing and publication work is
terminated.

We finish as soon as the set of pending requests had been received and
published, cut short after the grace time.

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.

Disruptive User Impact

Author's Checklist

[ ]

How to test this PR locally

Related issues

Closes [Crowdstrike FDR] Handle deduplication integrations#13062

Use cases

Screenshots

Logs

elasticmachine · 2025-03-20T04:02:46Z

Pinging @elastic/obs-ds-hosted-services (Team:obs-ds-hosted-services)

elasticmachine · 2025-03-20T04:02:47Z

Pinging @elastic/security-service-integrations (Team:Security-Service Integrations)

docs/reference/filebeat/filebeat-input-aws-s3.md

kcreddy

LGTM.
Suggest to merge after someone more familiar with input approves.

pierrehilbert · 2025-03-21T07:21:01Z

@bturquet could we have someone from your time to review this one please?
@faec will only be back on Monday.

bturquet · 2025-03-21T14:16:54Z

@Kavindu-Dodan could you help with the review please ?

x-pack/filebeat/input/awss3/sqs_input.go

efd6 · 2025-03-24T01:17:48Z

x-pack/filebeat/input/awss3/sqs_input.go

@@ -226,19 +270,23 @@ func (w *sqsWorker) processMessage(ctx context.Context, msg types.Message) {
 		w.client.Publish(e)
 		publishCount++


Really, this var should not be needed since the implementation already returns the sum of to-publish events in result.eventCount, but it seems that this is not implemented in testing due to the approach taken in mock testing.

AndersonQ · 2025-03-24T12:42:39Z

x-pack/filebeat/input/awss3/sqs_input.go

+func cancelWithGrace(parent context.Context, timeout time.Duration) (context.Context, context.CancelFunc) {
+	ctx, cancel := context.WithCancel(context.WithoutCancel(parent))
+	stop := context.AfterFunc(parent, func() {
+		time.AfterFunc(timeout, cancel)
+	})
+	return ctx, func() {
+		stop()
+		cancel()
+	}
+}


I think it'd be good to have a unit test for this function.

x-pack/filebeat/input/awss3/sqs_input.go

Kavindu-Dodan

Looks good for me :)

AndersonQ · 2025-03-26T11:09:23Z

x-pack/filebeat/input/awss3/sqs_test.go

+	case <-time.After(tooLong):
+		t.Fatal("parent context failed to cancel within timeout")
+	case <-parentCtx.Done():
+		parentCancelled = time.Now()


[NIT]
parent is already cancelled, you don't need the wait here. If <-parentCtx.Done() does not get selected, you could fail the test immediately.

Yeah, it's here for symmetry.

mergify · 2025-03-26T11:15:33Z

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 i13062-awss3 upstream/i13062-awss3
git merge upstream/main
git push upstream i13062-awss3

We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. The approach is a little blunt; it merely waits for the grace period after the requester has been stopped to allow for incoming messages to be processed and published. Ideally, we would finish as soon as the set of pending requests had been received and published, cut short after the grace time. However, the current structure of data flow does not lend itself sharing that information between parts of the input, so a minimal change is made.

This fixes handling of the case where graceCtx has not yet been cancelled, but ctx has. This would leave the first select in a blocked state when and so starve the second select, resulting in failure to work through the pending messages. This also adds an opportunity to exit the loop early when all the pending messages have been handled.

We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c) # Conflicts: # docs/reference/filebeat/filebeat-input-aws-s3.md

We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c)

…e on shutdown (#43529) * x-pack/filebeat/input/awss3: allow a grace time on shutdown (#43369) We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c) # Conflicts: # docs/reference/filebeat/filebeat-input-aws-s3.md * resolve conflicts * remove irrelevant changelog entry * apply doc changes to asciidoc and remove markdown addition --------- Co-authored-by: Dan Kortschak <[email protected]>

…me on shutdown (#43530) * x-pack/filebeat/input/awss3: allow a grace time on shutdown (#43369) We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c) # Conflicts: # docs/reference/filebeat/filebeat-input-aws-s3.md * resolve conflicts * apply doc changes to asciidoc and remove markdown addition --------- Co-authored-by: Dan Kortschak <[email protected]>

…me on shutdown (#43531) * x-pack/filebeat/input/awss3: allow a grace time on shutdown (#43369) We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c) # Conflicts: # docs/reference/filebeat/filebeat-input-aws-s3.md * resolve conflicts * remove irrelevant changelog entry * apply doc changes to asciidoc and remove markdown addition --------- Co-authored-by: Dan Kortschak <[email protected]>

We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c)

…e on shutdown (#43532) * x-pack/filebeat/input/awss3: allow a grace time on shutdown (#43369) We may have SQS message requests in flight that are being serviced when the input's context is cancelled. This allows the user to specify a time to wait after the context is cancelled before the processing and publication work is terminated. We finish as soon as the set of pending requests had been received and published, cut short after the grace time. (cherry picked from commit c0d439c) * remove irrelevant changelog entries --------- Co-authored-by: Dan Kortschak <[email protected]>

efd6 self-assigned this Mar 20, 2025

botelastic bot added needs_team Indicates that the issue/PR needs a Team:* label and removed needs_team Indicates that the issue/PR needs a Team:* label labels Mar 20, 2025

efd6 force-pushed the i13062-awss3 branch 2 times, most recently from 944fa21 to 6e73b10 Compare March 20, 2025 02:09

efd6 added the backport-8.x Automated backport to the 8.x branch with mergify label Mar 20, 2025

efd6 marked this pull request as ready for review March 20, 2025 04:02

efd6 requested review from a team as code owners March 20, 2025 04:02

efd6 requested review from AndersonQ and khushijain21 March 20, 2025 04:02

efd6 requested review from andrewkroh and kcreddy March 20, 2025 04:02

kcreddy reviewed Mar 21, 2025

View reviewed changes

docs/reference/filebeat/filebeat-input-aws-s3.md Show resolved Hide resolved

kcreddy approved these changes Mar 21, 2025

View reviewed changes

Kavindu-Dodan reviewed Mar 21, 2025

View reviewed changes

x-pack/filebeat/input/awss3/sqs_input.go Outdated Show resolved Hide resolved

efd6 commented Mar 24, 2025

View reviewed changes

AndersonQ reviewed Mar 24, 2025

View reviewed changes

Kavindu-Dodan reviewed Mar 24, 2025

View reviewed changes

x-pack/filebeat/input/awss3/sqs_input.go Show resolved Hide resolved

efd6 force-pushed the i13062-awss3 branch from 07617d9 to 938fbdb Compare March 24, 2025 21:06

Kavindu-Dodan approved these changes Mar 24, 2025

View reviewed changes

efd6 requested a review from AndersonQ March 26, 2025 10:58

AndersonQ approved these changes Mar 26, 2025

View reviewed changes

andrewkroh approved these changes Mar 27, 2025

View reviewed changes

efd6 added 6 commits March 27, 2025 11:55

add test for cancelWithGrace

9314f61

fix pending parity

d2f5506

fix waitgroup bug tickled by change

21d535d

fix handling of grace period none-pending state

84da67f

efd6 force-pushed the i13062-awss3 branch from 2ad6ac0 to 84da67f Compare March 27, 2025 01:26

efd6 merged commit c0d439c into elastic:main Mar 27, 2025
28 checks passed

mergify bot mentioned this pull request Mar 27, 2025

[8.x](backport #43369) x-pack/filebeat/input/awss3: allow a grace time on shutdown #43529

Merged

6 tasks

mergify bot mentioned this pull request Mar 27, 2025

[8.17](backport #43369) x-pack/filebeat/input/awss3: allow a grace time on shutdown #43530

Merged

6 tasks

mergify bot mentioned this pull request Mar 27, 2025

[8.18](backport #43369) x-pack/filebeat/input/awss3: allow a grace time on shutdown #43531

Merged

6 tasks

mergify bot mentioned this pull request Mar 27, 2025

[9.0](backport #43369) x-pack/filebeat/input/awss3: allow a grace time on shutdown #43532

Merged

6 tasks

mauri870 mentioned this pull request Apr 3, 2025

[Flaky Test] x-pack/filebeat/input/awss3 TestCancelWithGrace #43668

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

x-pack/filebeat/input/awss3: allow a grace time on shutdown #43369

x-pack/filebeat/input/awss3: allow a grace time on shutdown #43369

efd6 commented Mar 20, 2025 •

edited

Loading

elasticmachine commented Mar 20, 2025

elasticmachine commented Mar 20, 2025

kcreddy left a comment

pierrehilbert commented Mar 21, 2025

bturquet commented Mar 21, 2025

efd6 Mar 24, 2025

AndersonQ Mar 24, 2025

efd6 Mar 24, 2025

Kavindu-Dodan left a comment

AndersonQ Mar 26, 2025 •

edited

Loading

efd6 Mar 27, 2025

mergify bot commented Mar 26, 2025

		@@ -226,19 +270,23 @@ func (w *sqsWorker) processMessage(ctx context.Context, msg types.Message) {
		w.client.Publish(e)
		publishCount++

x-pack/filebeat/input/awss3: allow a grace time on shutdown #43369

x-pack/filebeat/input/awss3: allow a grace time on shutdown #43369

Conversation

efd6 commented Mar 20, 2025 • edited Loading

Proposed commit message

Checklist

Disruptive User Impact

Author's Checklist

How to test this PR locally

Related issues

Use cases

Screenshots

Logs

elasticmachine commented Mar 20, 2025

elasticmachine commented Mar 20, 2025

kcreddy left a comment

Choose a reason for hiding this comment

pierrehilbert commented Mar 21, 2025

bturquet commented Mar 21, 2025

efd6 Mar 24, 2025

Choose a reason for hiding this comment

AndersonQ Mar 24, 2025

Choose a reason for hiding this comment

efd6 Mar 24, 2025

Choose a reason for hiding this comment

Kavindu-Dodan left a comment

Choose a reason for hiding this comment

AndersonQ Mar 26, 2025 • edited Loading

Choose a reason for hiding this comment

efd6 Mar 27, 2025

Choose a reason for hiding this comment

mergify bot commented Mar 26, 2025

efd6 commented Mar 20, 2025 •

edited

Loading

AndersonQ Mar 26, 2025 •

edited

Loading