[Filebeat] Refactor AWS S3 input with workers

[andrewkroh]

What does this PR do?

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 end of processing the batch (as seen in the monitoring metrics
below). 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 | 36380.253532518276 | 1.2 MB           |        1.243366816 |   12 |
|                 2 | 61727.549671738896 | 2.1 MB           | 1.0951187980000001 |   12 |
|                 4 |  86218.70431874547 | 3.0 MB           |        1.208577661 |   12 |
|                 8 | 131900.69854257331 | 4.5 MB           |        1.179751144 |   12 |
|                16 |   151824.404438336 | 5.2 MB           |        1.083857372 |   12 |
|                32 | 155548.56015625654 | 5.3 MB           |        1.170502638 |   12 |
|                64 | 166188.27838709904 | 5.7 MB           |         1.17403587 |   12 |
|               128 | 185429.33410590928 | 6.4 MB           |        3.380193339 |   12 |
|               256 | 186181.15705271234 | 6.4 MB           |         1.66313823 |   12 |
|               512 |  197793.9906993712 | 7.3 MB           |        1.230735065 |   12 |
|              1024 | 211492.91373843007 | 7.4 MB           |        1.246637513 |   12 |
+-------------------+--------------------+------------------+--------------------+------+

Relates #25750

Why is it important?

This enabled easier vertical scaling of Filebeat aws-s3 input and helps it better utilize the CPU that's available.

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.

Author's Checklist

• Update documentation
• Make max receive message retries configurable.
• Streaming JSON parser code needs cleaned up
• Check for any recent changes since May that were not merged into refactoring
• Changelog

Related issues

• Relates [Filebeat] aws-s3 input visibility timeout extensions lead to deadlocked ACKHandlers #25750

[elasticmachine]

Pinging @elastic/integrations (Team:Integrations)

[elasticmachine]

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

[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: 2021-08-11T20:10:48.769+0000

• Duration: 218 min 16 sec

• Commit: d246c04

Test stats 🧪

Test	Results
Failed	0
Passed	53000
Skipped	5318
Total	58318

Trends 🧪

💚 Flaky test report

Tests succeeded.

Expand to view the summary

Test stats 🧪

Test	Results
Failed	0
Passed	53000
Skipped	5318
Total	58318

[aspacca]

LGTM

[exekias]

@kaiyan-sheng would this work with our existing AWS ci integration? that would be great.

[kaiyan-sheng]

I think we have the aws tests run weekly right now in the CI and this would work. @v1v Do you know if this is easy to add?

[v1v]

The AWS tests run on a weekly basis, as you said, and it runs the commands defined in:

• beats/x-pack/metricbeat/Jenkinsfile.yml

  Lines 35 to 47 in d898533

  	cloud:
  	cloud: "mage build test"
  	withModule: true ## run the ITs only if the changeset affects a specific module.
  	dirs: ## run the cloud tests for the given modules.
  	- "x-pack/metricbeat/module/aws"
  	when: ## Override the top-level when.
  	parameters:
  	- "awsCloudTests"
  	comments:
  	- "/test x-pack/metricbeat for aws cloud"
  	labels:
  	- "aws"
  	stage: extended

mage build test runs in the folder x-pack/metricbeat so, if the above-mentioned command is part of the mage then it should work.

[aspacca]

the test requires applying a terraform file (and destroying it at the end):
https://github.com/elastic/beats/blob/master/x-pack/filebeat/input/awss3/_meta/terraform/README.md#usage

will this happen?

[aspacca]

btw, shouldn't be the file https://github.com/elastic/beats/blob/master/x-pack/filebeat/Jenkinsfile.yml?

[aspacca]

actually one doubt here: do we really want to wait for all the events to be acked for the worker to shutdown?

[andrewkroh]

Yes. S3 event notifications for ObjectCreated sometimes contain information for more than object. This function returns after all the events generated from each of those objects (usually it's just 1 object) has been ACKed. Then the caller of this function can stop updating the SQS visibility timeout for this one SQS message and decide based on whether an error occurred while processing the S3 object if it should delete the SQS message or try to process it again.

[aspacca]

Yes, I understand the reason for waiting in the PR, but I was wondering if we should wait for ACK for actually handling the SQS message.
The current implementation doesn't have this: in theory ACK could be blocked for a certain amount of time (network issue reaching elasticsearch, for example), filebeat queue should be able to later catch up. While here we will block starting new workers if I got it right.

[andrewkroh]

The reason for waiting until all ACKs are received before deleting the SQS message (and freeing the worker) is that the queued events are non-persistent by default. If we delete the SQS message before receiving all ACKs then Filebeat is vulnerable to data loss if the process is killed before the queue is emptied (events are ACKed).

By waiting until all ACKs are received we ensure that another instance of Filebeat will process the same message again if this process is killed before all events are ACKed.

[aspacca]

I agree we should not delete the SQS message before receiving all ACKs, but by making the worker wait for the ACKs we prevent it from filling the Filebeat queue in cases of backpressure. If the ACK wait / keepalive / delete message handling was done somewhere else (in a goroutine) that would free the worker for keep processing other SQS messages and filling the queue.

[andrewkroh]

We could write it that way.

The current implementation is derived from having a single tuning knob of max_number_of_messages that controls the number of inflight messages. If we change the model then we'd add another option that controls the number of workers independently of the max_number_of_messages.

I'm hesitant to change it now without doing some testing to see if that is actually a problem. I think that if you size the max_number_of_messages parameter appropriately based on the files you're processing and your queue.mem.events size then you should be able to keep the internal memory queue full in this model without much of a problem.

[andrewkroh]

@aspacca I updated the S3 interface to account for your need to call ListObjects. I included a helper to mock the S3 pagination calls for your tests.

[leehinman]

LGTM. Just a question about the acker wait function.

[aspacca]

/test

[aspacca]

LGTM

please, @andrewkroh , could you wait for #27126 to be merged and adapt any needed changes in your PR?

E2E tests failure should be fixed if you merge latest master

[andrewkroh]

Sure, I'll wait for that and then fix any merge conflicts.

[kaiyan-sheng]

Thank you @andrewkroh for this enhancement. Should filebeat.reference.yml also get updated in the aws-s3 input section? Other than that, everything looks good.

[andrewkroh]

I'll add the two new settings to filebeat.reference.yml when I do the merge for #27126.

sqs.max_receive_count

sqs.wait_time