[Filebeat] aws-s3 - create beat.Client for each SQS worker #33658

andrewkroh · 2022-11-13T20:04:17Z

What does this PR do?

To address mutex contention in the single beat.Client used for publishing S3 events, create a unique beat.Client for each worker goroutine that is processing an SQS message. The beat.Client is used for all S3 objects contained within the SQS message (they are processed serially). After all events are ACKed the beat.Client is closed.

Why is it important?

A mutex profile of the aws-s3 input showed that there was a lot of contention for the "Publish()" lock.

beats/libbeat/publisher/pipeline/client.go

Lines 72 to 77 in 3e928b7

    
           func (c *client) Publish(e beat.Event) { 
        
           	c.mutex.Lock() 
        
           	defer c.mutex.Unlock() 
        
           	c.publish(e) 
        
           }

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.

Pprof Mutex

I ran Filebeat for 90s with max_number_of_messages: 5.

Before

(pprof) top10
Showing nodes accounting for 48.18s, 100% of 48.18s total
      flat  flat%   sum%        cum   cum%
    48.18s   100%   100%     48.18s   100%  sync.(*Mutex).Unlock
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/filebeat/beater.(*countingClient).Publish
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/libbeat/publisher/pipeline.(*client).Publish
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*s3ObjectProcessor).ProcessS3Object
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*s3ObjectProcessor).publish
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*s3ObjectProcessor).readFile
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*sqsReader).Receive.func1
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*sqsS3EventProcessor).ProcessSQS
         0     0%   100%     48.18s   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*sqsS3EventProcessor).processS3Events

After

(pprof) top10
Showing nodes accounting for 46.26ms, 100% of 46.26ms total
Showing top 10 nodes out of 14
      flat  flat%   sum%        cum   cum%
   46.26ms   100%   100%    46.26ms   100%  sync.(*Mutex).Unlock (inline)
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/filebeat/beater.(*countingClient).Publish
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/libbeat/common/acker.(*eventDataACKer).AddEvent
         0     0%   100%     1.67ms  3.61%  github.com/elastic/beats/v7/libbeat/common/acker.(*trackingACKer).AddEvent
         0     0%   100%     1.67ms  3.61%  github.com/elastic/beats/v7/libbeat/common/acker.(*trackingACKer).addPublishedEvent
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/libbeat/common/acker.ackerList.AddEvent
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/libbeat/publisher/pipeline.(*client).Publish
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/libbeat/publisher/pipeline.(*client).publish
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*s3ObjectProcessor).ProcessS3Object
         0     0%   100%    46.26ms   100%  github.com/elastic/beats/v7/x-pack/filebeat/input/awss3.(*s3ObjectProcessor).publish

mergify · 2022-11-13T20:04:52Z

This pull request does not have a backport label.
If this is a bug or security fix, could you label this PR @andrewkroh? 🙏.
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

To address mutex connection in the single beat.Client used for publishing S3 events, create a unique beat.Client for each worker goroutine that is processing an SQS message. The beat.Client is used for all S3 objects contained within the SQS message (they are processed serially). After all events are ACKed the beat.Client is closed.

elasticmachine · 2022-11-13T20:25:33Z

💚 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: 2022-11-15T11:39:39.179+0000
Duration: 127 min 38 sec

Test stats 🧪

Test	Results
Failed	0
Passed	4924
Skipped	340
Total	5264

💚 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 · 2022-11-13T20:31:07Z

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

andrewkroh · 2022-11-13T22:11:20Z

/test

efd6 · 2022-11-13T23:15:16Z

I'm wondering if

beats/libbeat/publisher/pipeline/client.go

Lines 72 to 77 in 3e928b7

    
           func (c *client) Publish(e beat.Event) { 
        
           	c.mutex.Lock() 
        
           	defer c.mutex.Unlock() 
        
           	c.publish(e) 
        
           }

could be further improved to reduce contention.

If it's guaranteed that X can't panic, then it can be rewritted as

func (c *client) Publish(e beat.Event) {
	c.mutex.Lock()
	c.publish(e)
	c.mutex.Unlock()
}

which will reduce the time during the locked state.

If that can't be guaranteed, then this makes some (less) reduction without the potential for leaving the mutex in the locked state, by retaining the deferred unlock, but not locking until the defer is constructed.

func (c *client) Publish(e beat.Event) {
	defer c.mutex.Unlock()
	c.mutex.Lock()

	c.publish(e)
}

andrewkroh · 2022-11-14T20:14:50Z

I remember back in Go 1.14 the release notes came out saying that defer incurred almost zero overhead due to https://github.com/golang/proposal/blob/master/design/34481-opencoded-defers.md. I've never tried to measure the speed of defer vs no defer.

It is possible that a processor could panic so I would be wary of removing the defer.

defer c.mutex.Unlock()
c.mutex.Lock()

Interesting, I've never seen code that uses this ordering. I realize that it has the same outcome. What makes this any different?

…-reader-beat-client

efd6 · 2022-11-15T08:31:47Z

Interesting, I've never seen code that uses this ordering. I realize that it has the same outcome. What makes this any different?

The time taken is the same, but less of that time is in the locked state.

https://godbolt.org/z/4Ka9GTWsd

…-reader-beat-client

To address mutex connection in the single beat.Client used for publishing S3 events, create a unique beat.Client for each worker goroutine that is processing an SQS message. The beat.Client is used for all S3 objects contained within the SQS message (they are processed serially). After all events are ACKed the beat.Client is closed.

andrewkroh added enhancement Filebeat Filebeat labels Nov 13, 2022

botelastic bot added the needs_team Indicates that the issue/PR needs a Team:* label label Nov 13, 2022

andrewkroh force-pushed the feature/fb/awss3-per-reader-beat-client branch from 53cb8c3 to f231b1e Compare November 13, 2022 20:04

mergify bot assigned andrewkroh Nov 13, 2022

andrewkroh force-pushed the feature/fb/awss3-per-reader-beat-client branch from f231b1e to ae78fa0 Compare November 13, 2022 20:19

andrewkroh added Team:Security-External Integrations Team:Cloud-Monitoring Label for the Cloud Monitoring team labels Nov 13, 2022

botelastic bot removed the needs_team Indicates that the issue/PR needs a Team:* label label Nov 13, 2022

andrewkroh marked this pull request as ready for review November 13, 2022 20:31

andrewkroh requested a review from a team as a code owner November 13, 2022 20:31

This comment was marked as outdated.

Sign in to view

Merge remote-tracking branch 'elastic/main' into feature/fb/awss3-per…

c7ab679

…-reader-beat-client

andrewkroh requested a review from aspacca November 15, 2022 02:15

aspacca approved these changes Nov 15, 2022

View reviewed changes

aspacca mentioned this pull request Nov 15, 2022

sqs consumers #33659

Closed

6 tasks

Merge remote-tracking branch 'elastic/main' into feature/fb/awss3-per…

863994c

…-reader-beat-client

andrewkroh merged commit a461c17 into elastic:main Nov 15, 2022

andrewkroh mentioned this pull request Nov 21, 2022

Remove max_number_of_messages cap in Filebeat s3 input #32278

Closed

andrewkroh mentioned this pull request Nov 30, 2022

[Filebeat] aws-s3 input - fix integration test with AWS #33893

Merged

6 tasks

andrewkroh mentioned this pull request Jan 4, 2023

[aws.vpcflow] Default max_number_of_messages to 1 elastic/integrations#4599

Closed

4 tasks

cmacknz mentioned this pull request Apr 3, 2023

[Elastic Agent] Default processors created per input can result in high agent CPU usage #35000

Closed

cmacknz mentioned this pull request Apr 19, 2023

[Elastic Agent] Prototype a shipper gRPC input for Filebeat #35135

Closed

1 task

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Filebeat] aws-s3 - create beat.Client for each SQS worker #33658

[Filebeat] aws-s3 - create beat.Client for each SQS worker #33658

andrewkroh commented Nov 13, 2022 •

edited

Loading

mergify bot commented Nov 13, 2022

elasticmachine commented Nov 13, 2022 •

edited by jenkins-beats-ci bot

Loading

Build stats

Test stats 🧪

elasticmachine commented Nov 13, 2022

andrewkroh commented Nov 13, 2022

efd6 commented Nov 13, 2022

andrewkroh commented Nov 14, 2022

This comment was marked as outdated.

efd6 commented Nov 15, 2022 •

edited

Loading

	func (c *client) Publish(e beat.Event) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	c.publish(e)
	}

[Filebeat] aws-s3 - create beat.Client for each SQS worker #33658

[Filebeat] aws-s3 - create beat.Client for each SQS worker #33658

Conversation

andrewkroh commented Nov 13, 2022 • edited Loading

What does this PR do?

Why is it important?

Checklist

Related

Pprof Mutex

Before

After

mergify bot commented Nov 13, 2022

elasticmachine commented Nov 13, 2022 • edited by jenkins-beats-ci bot Loading

💚 Build Succeeded

Build stats

Test stats 🧪

💚 Flaky test report

🤖 GitHub comments

elasticmachine commented Nov 13, 2022

andrewkroh commented Nov 13, 2022

efd6 commented Nov 13, 2022

andrewkroh commented Nov 14, 2022

This comment was marked as outdated.

efd6 commented Nov 15, 2022 • edited Loading

andrewkroh commented Nov 13, 2022 •

edited

Loading

elasticmachine commented Nov 13, 2022 •

edited by jenkins-beats-ci bot

Loading

efd6 commented Nov 15, 2022 •

edited

Loading