feat(new source): aws_s3 source

[jszwedko]

See RFC: https://github.com/timberio/vector/blob/master/rfcs/2020-09-29-4155-aws-s3-source.md
Fixes #1017

This is the initial implementation of an aws_s3 source that relies an AWS SQS for bucket notifications to inform of new S3 objects to consume. See RFC for discussion of other approaches and why this one was chosen. The source does have an enum strategy configuration to allow for additional approaches (like SNS or long polling) to be supported.

The basic flow is:

User setup:

• bucket is created
• queue is created
• bucket is configured to notify the queue for ObjectCreated events
• vector is configured with the aws_s3 source using the queue configuration whereupon it will process the ObjectCreated events to read each S3 object.

Example configuration:

[sources.s3]
  type = "aws_s3"

  region = "us-east-1"
  strategy = "sqs"
  compression = "gzip"

  sqs.queue_name = "jesse-test-flow-logs"
  sqs.delete_message = false

[sinks.console]
  type = "console"
  inputs = ["s3"]
  encoding.codec = "json"

The commits can be viewed in-order, but the resulting diff probably isn't to bad either (it's mostly just sources/aws_s3). It may be worth looking at the added cue documentation first.

The source also behaves very much like the file source where it emits one event per-line, but also supports the same multiline configuration that the file source supports. Note there is a rough edge here where the multiline config supports a timeout_ms option that isn't really applicable here but is applied just the same.

Future work:

1. Additional codec support (like application/ndjson). For now, this acts very much like the file source. This could be looped into the general work around codecs Decouple metrics parsers as codecs #4278
2. Additional compression formats (Snappy, LZ4, Zip). This was requested in the original issue. I started with just supporting the formats that were supported out-of-the-box by the async_compression crate we are using.
3. (potential) multi-region support. Currently we only support reading from a queue and a bucket in the same region. I expect this will cover most cases since AWS requires the bucket to publish notifications to a queue in the same region. One could forward messages from a queue in one region to another, but this seems unlikely. I'd prefer to wait and see if anyone asks for multi-region support; especially given that fluentbit and filebeat have the same restriction.
4. Concurrent processing. Right now one message is processed at a time which leads to predictable behavior, but we may observe some performance improvements by processing multiple objects at once. The benefit should be vetted though, the process may be limited by incoming network bandwidth anyway.
5. Refresh the message visibility timeout. Right now, the visibility timeout is set once, when the message is retrieved, but we could refresh this timeout if we are still processing a message when it gets close to the end of the timeout to avoid another vector instance picking it up. This would let users have the best of both worlds: short visibility timeouts to quickly reprocess messages when a vector instance falls over, but also avoiding concurrent processing of messages for large objects where the processing time exceeds the visibility timeout.

I'll create issues for 2 and 5. I think the others can be left until we observe their necessity.

[jszwedko]

I think this could probably be better refactored by having the SQS bits implement an interface to drive a top-level object reader by having it simply return S3 keys to read, but I figured that was better left until we actually add another "strategy" which could then refactor this a bit to share the logic doing the object reads and event creation.

[jszwedko]

Ideas here also welcome 😄

[JeanMertz]

I'm not really sure what the issue is here?

[jszwedko]

It just feels a little messy to me to record intermediate state rather than just streaming. We ended up having a conversation about it in discord though: https://discordapp.com/channels/742820443487993987/751107681217019944/771486734110162964

My conclusion from that conversation is that it'd be nice to use more TryStreams and have send_all stop at the first error and return it so that I could flow all errors and then handle them in one place.

[jszwedko]

I abstained from using use super::* here to see how it felt. I think I like the explicitness better.

[JeanMertz]

I would personally use use super::* in tests, especially since basically all the use statements below are redundant to what's already imported in the parent module.

But, it's a minor nit, so I'm fine leaving this experiment in.

[jszwedko]

Yeah, I'm not 100% on it. Some of the imports are duplicated, but there is also a much smaller set of imports in the integration test and I found it easier to see where certain types were coming from rather than having to check first the integration_tests imports and then super's imports.

[jszwedko]

For the S3 event record version validation, I ended up pulling in the semver crate to handle the deserialization and comparison. It's a small crate and we were already indirectly depending on it:

$ cargo tree -i  semver      
semver v0.9.0
├── cargo-lock v4.0.1
│   └── built v0.4.3
│       [build-dependencies]
│       └── vector v0.11.0 (/home/jesse/workspace/vector)
└── rustc_version v0.2.3
    [build-dependencies]
    ├── cast v0.2.3
    │   ├── criterion v0.3.3
    │   │   [dev-dependencies]
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   └── criterion-plot v0.4.3
    │       └── criterion v0.3.3 (*)
    ├── raw-cpuid v8.1.1
    │   └── heim-virt v0.1.0-beta.1
    │       └── heim v0.1.0-beta.3
    │           └── vector v0.11.0 (/home/jesse/workspace/vector)
    ├── rusoto_core v0.45.0
    │   ├── rusoto_cloudwatch v0.45.0
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   ├── rusoto_firehose v0.45.0
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   ├── rusoto_kinesis v0.45.0
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   ├── rusoto_logs v0.45.0
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   ├── rusoto_s3 v0.45.0
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   ├── rusoto_sts v0.45.0
    │   │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    │   └── vector v0.11.0 (/home/jesse/workspace/vector)
    └── rusoto_signature v0.45.0
        ├── rusoto_core v0.45.0 (*)
        └── vector v0.11.0 (/home/jesse/workspace/vector)

[jszwedko]

Noting for myself that I need to update this to not allow timeout_secs to apply for the multiline aggregation.

[JeanMertz]

Sweet!

Left some comments, nothing major, except for maybe Windows newlines.

[JeanMertz]

Can't we still use enums, but have a catch all that at least logs if something is received that's unexpected (or just ignore "other")?

Still, this seems fine by me as well 👍

[jszwedko]

👍 I'll investigate this.

[jszwedko]

I came back around the other way on this one 😄 I'd like to, at least, recognize any new ObjectCreated events without needing to add new enum kinds. I'll just leave this as-is until we find a need.

[jszwedko]

@bruceg do you mind giving this one another look?

[bruceg]

Just one suggestion about a long function, nothing blocking though.

[bruceg]

This function is already pretty long. I would tend to suggest moving this all either into handle_sqs_message or into a new wrapper function, and then moving the call to it into the Some(ref handle) above so you can drop the continue.

[jszwedko]

Note for myself that I broke the deseralization of SQS messages in a recent commit due to relying on the semvar's serde support which requires a fully formed version number (e.g. 2.1.0 rather than 2.1). I opened dtolnay/semver#219 but I'll probably back this out to do custom parsing for now until upstream support is added (if it is at all).

[jszwedko]

Windows failure is the memory issue that also exists on master.