-
Notifications
You must be signed in to change notification settings - Fork 4.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[Filebeat] Refactor AWS S3 input with workers (#27199)
* Refactor AWS S3 input with workers
This changes the AWS S3 input to allow it to process more SQS messages in parallel
by having workers that are fully utilized while there are SQS message to process.
The previous design processed SQS messages in batches ranging from 1 to 10 in size.
It waited until all messages were processed before requesting more. This left some
workers idle toward the the end of processing the batch. This also limited the maximum
number of messages processed in parallel to 10 because that is the largest request
size allowed by SQS.
The refactored input uses ephemeral goroutines as workers to process SQS messages. It
receives as many SQS messages as there are free workers. The total number of workers
is controlled by `max_number_of_messages` (same as before but without an upper limit).
Other changes
Prevent poison pill messages
When an S3 object processing error occurs the SQS message is returned to the after
the visibility timeout expires. This allows it to be reprocessed again or moved to
the SQS dead letter queue (if configured). But if no dead letter queue policy is
configured and the error is permanent (reprocessing won't fix it) then the message
would continuosly be reprocessed. On error the input will now check the
`ApproximateReceiveCount` attribute of the SQS message and delete it if it exceeds
the configured maximum retries.
Removal of api_timeout from S3 GetObject calls
The `api_timeout` has been removed from the S3 `GetObject` call. This limited the
maximum amount of time for processing the object since the response body is processed
as a stream while the request is open. Requests can still timeout in the server due
to inactivity.
Improved debug logs
The log messages have been enriched with more data about the related SQS message and
S3 object. For example the SQS `message_id`, `s3_bucket`, and `s3_object` are
included in some messages.
`DEBUG [aws-s3.sqs_s3_event] awss3/s3.go:127 End S3 object processing. {"id": "test_id", "queue_url": "https://sqs.us-east-1.amazonaws.com/144492464627/filebeat-s3-integtest-lxlmx6", "message_id": "a11de9f9-0a68-4c4e-a09d-979b87602958", "s3_bucket": "filebeat-s3-integtest-lxlmx6", "s3_object": "events-array.json", "elapsed_time_ns": 23262327}`
Increased test coverage
The refactored input has about 88% test coverage.
The specific AWS API methods used by the input were turned into interfaces to allow
for easier testing. The unit tests mock the AWS interfaces.
The parts of the input were separted into three components listed below. There's a defined
interface for each to allow for mock testing there too. To test the interactions between
these components go-mock is used to generate mocks and then assert the expectations.
1. The SQS receiver. (sqs.go)
2. The S3 Notification Event handler. (sqs_s3_event.go)
3. The S3 Object reader. (s3.go)
Terraform setup for integration test
Setup for executing the integration tests is now handled by Terraform.
See _meta/terraform/README.md for instructions.
Benchmark test
I added a benchmark that tests the input in isolation with mocked SQS and S3 responeses.
It uses a 7KB cloudtrail json.gz file containing about 60 messages for its input.
This removes any variability related to the network, but also means these do not reflect
real-work rates. They can be used to measure the effect of future changes.
+-------------------+--------------------+------------------+--------------------+------+
| MAX MSGS INFLIGHT | EVENTS PER SEC | S3 BYTES PER SEC | TIME (SEC) | CPUS |
+-------------------+--------------------+------------------+--------------------+------+
| 1 | 23019.782175720782 | 3.0 MB | 1.257266458 | 12 |
| 2 | 36237.53174269319 | 4.8 MB | 1.158798571 | 12 |
| 4 | 56456.84532752983 | 7.5 MB | 1.138285351 | 12 |
| 8 | 90485.15755430676 | 12 MB | 1.117244007 | 12 |
| 16 | 103853.8984324643 | 14 MB | 1.165541225 | 12 |
| 32 | 110380.28141417276 | 15 MB | 1.110814345 | 12 |
| 64 | 116074.13735061679 | 15 MB | 1.408100062 | 12 |
| 128 | 114854.80273666105 | 15 MB | 1.5331357140000001 | 12 |
| 256 | 118767.73924992209 | 16 MB | 2.041783413 | 12 |
| 512 | 122933.1033660647 | 16 MB | 1.255463303 | 12 |
| 1024 | 124222.51861746894 | 17 MB | 1.505765638 | 12 |
+-------------------+--------------------+------------------+--------------------+------+
Relates #25750
* Use InitializeAWSConfig
* Add s3Lister interface for mocking pagination of S3 ListObjects calls
* Add new config parameters to reference.yml
* Optimize uploading b/c it was slow in aws v2 sdk
(cherry picked from commit 7c76158)
# Conflicts:
# x-pack/filebeat/input/awss3/collector.go
# x-pack/filebeat/input/awss3/collector_test.go- Loading branch information
1 parent
7e43fd9
commit 6b41f01
Showing
42 changed files
with
3,583 additions
and
694 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file was deleted.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,3 @@ | ||
| terraform/ | ||
| outputs.yml | ||
| *.tfstate* |
57 changes: 57 additions & 0 deletions
57
x-pack/filebeat/input/awss3/_meta/terraform/.terraform.lock.hcl
Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| # Terraform setup for AWS S3 Input Integration Tests | ||
|
|
||
| This directory contains a Terrafrom module that creates the AWS resources needed | ||
| for executing the integration tests for the `aws-s3` Filebeat input. It creates | ||
| an S3 bucket and SQS queue and configures S3 `ObjectCreated:*` notifications to | ||
| be delivered to SQS. | ||
|
|
||
| It outputs configuration information that is consumed by the tests to | ||
| `outputs.yml`. The AWS resources are randomly named to prevent name collisions | ||
| between multiple users. | ||
|
|
||
| ### Usage | ||
|
|
||
| You must have the appropriate AWS environment variables for authentication set | ||
| before running Terraform or the integration tests. The AWS key must be | ||
| authorized to create and destroy S3 buckets and SQS queues. | ||
|
|
||
| 1. Execute terraform in this directory to create the resources. This will also | ||
| write the `outputs.yml`. | ||
|
|
||
| `terraform apply` | ||
|
|
||
| 2. (Optional) View the output configuration. | ||
|
|
||
| ```yaml | ||
| "aws_region": "us-east-1" | ||
| "bucket_name": "filebeat-s3-integtest-8iok1h" | ||
| "queue_url": "https://sqs.us-east-1.amazonaws.com/144492464627/filebeat-s3-integtest-8iok1h" | ||
| ``` | ||
| 2. Execute the integration test. | ||
| ``` | ||
| cd x-pack/filebeat/inputs/awss3 | ||
| go test -tags aws,integration -run TestInputRun -v . | ||
| ``` | ||
|
|
||
| 3. Cleanup AWS resources. Execute terraform to remove the SQS queue and delete | ||
| the S3 bucket and its contents. | ||
|
|
||
| `terraform destroy` | ||
|
|
||
|
|
Oops, something went wrong.