Support AWS Kinesis Data Streams as a Source

[dlvenable]

Is your feature request related to a problem? Please describe.

Some pipeline authors want to retrieve events from Amazon Kinesis Data Streams.

Describe the solution you'd like

Create a kinesis_data_streams source plugin. The Kinesis Client Library (KCL) can manage much of the client needs. So I propose that the Data Prepper source use KCL for reading from Kinesis.

KCL uses DynamoDB to coordinate consumers. Because KCL uses DynamoDB and Kinesis presumes an AWS account anyway, I propose that Data Prepper uses DynamoDB for consumer coordination.

Data Prepper should support configuring the AWS resources and access to the AWS resources that KCL needs. And also configuring the Kinesis stream name.

Example configuration:

source:
  kinesis_data_streams:
    stream_name: MyStream
    coordination_table_name: MyDynamoDbTable

Additional context

https://javadoc.io/doc/software.amazon.kinesis/amazon-kinesis-client/latest/index.html

[daixba]

I would like to propose below design for KDS as source

pipeline:
  source:
    kinesis:
      streams:
        - stream_name: "test-stream1"
          initial_position: "TRIM_HORIZON"
          consumer_strategy: "Polling"
        - stream_name: "test-stream2"
          initial_position: "LATEST"
          consumer_strategy: "Enhanced-Fan-Out"
      application_name: "my-app"
      aws:
        region: "us-west-2"
        sts_role_arn: "xxx"   

Here are the configuration options:

• stream_name - (required) - The Kinesis Data Stream name to be consumed. A single pipeline can consume data from multiple streams.
• initial_position - (optional) - Supported value: LATEST, TRIM_HORIZON. Default to LATEST.
• consumer_strategy - (optional) - Supported value: Polling and Enhanced-Fan-Out, default to Polling.
• application_name - (required) - Consumer client application name. This name will also be used as KCL coordination table name.
• aws - (required) - AWS Auth to read from multiple streams. All streams must be in the same AWS region.

[cameronattard]

This would be great for processing logs from Cloudwatch logs.

[dlvenable]

It would be nice to support multi-tenancy in the KCL library when awslabs/amazon-kinesis-client#1368 is available.

source:
  kinesis:
    tenancy: multi

The default value for tenancy could be single as this works for most use-cases.

Also, it would be nice to support the KCL configurations for DDB in data-prepper-config.yaml. This would allow multiple sources to share a single KCL library and thus the same DDB table.

[sb2k16]

I would like to work on this issue. Could you please assign this to me?