Add DynamoDB source plugin (#3349)

Signed-off-by: Aiden Dai <[email protected]>

data-prepper-plugins/dynamodb-source/README.md

@@ -0,0 +1,83 @@
+# DynamoDB Source
+
+This is a source plugin that supports retrieve data from DynamoDB tables. Basic use case of this source plugin is to
+sync the data from DynamoDB tables to OpenSearch indexes. With this CDC support, customer can run the end to end data
+sync pipeline and capture changed data in near real-time without writing any codes and without any downtime of business.
+Such pipeline can run on multiple nodes in parallel to support data capture of large scale tables.
+
+This plugin can support below three different modes:
+
+1. Full load only:  One time full data export and load
+2. CDC Only:  DynamoDB Stream
+3. Full load + CDC:  One time full export and load + DynamoDB Stream.
+
+## Usages
+
+To get started with this DynamoDB source, create the following source configuration:
+
+```yaml
+source:
+  dynamodb:
+    tables:
+      - table_arn: "arn:aws:dynamodb:us-west-2:123456789012:table/my-table"
+        stream:
+          start_position:
+        export:
+          s3_bucket: "my-bucket"
+          s3_prefix: "export/"
+    aws:
+      region: "us-west-2"
+      sts_role_arn: "arn:aws:iam::123456789012:role/DataPrepperRole"
+
+    coordinator:
+      dynamodb:
+        table_name: "coordinator-demo"
+        region: "us-west-2"
+
+
+```
+
+## Configurations
+
+### Shared Configurations:
+
+* coordinator (Required):  Coordination store setting. This design create a custom coordinator based on existing
+  coordination store implementation. Only DynamoDB is tested so far.
+* aws (Required):  High level AWS Auth. Note Data Prepper will use the same AWS auth to access all tables, check
+  Security for more details.
+    * region
+    * sts_role_arn
+
+### Export Configurations:
+
+* s3_bucket (Required):  The destination bucket to store the exported data files
+* s3_prefix (Optional):  Custom prefix.
+
+### Stream Configurations
+
+* start_position (Optional):  start position of the stream, can be either BEGINNING or LATEST. If export is required,
+  this value will be ignored and set to LATEST by default. This is useful if customer don’t want to run initial export,
+  so they can
+  choose either from the beginning of the stream (up to 24 hours) or from latest (from the time point when pipeline is
+  started)
+
+## Metrics
+
+### Counter
+
+- `exportJobsSuccess`: measures total number of export jobs run with status completed.
+- `exportJobsErrors`: measures total number of export jobs cannot be submitted or run with status failed.
+- `exportFilesTotal`: measures total number of export files generated.
+- `exportFilesSuccess`: measures total number of export files read (till the last line) successfully.
+- `exportRecordsTotal`: measures total number of export records generated
+- `exportRecordsSuccess`: measures total number of export records processed successfully .
+- `exportRecordsErrors`: measures total number of export records processed failed
+- `changeEventsSucceeded`: measures total number of changed events in total processed successfully
+- `changeEventsFailed`:  measures total number of changed events in total processed failed
+
+## Developer Guide
+
+This plugin is compatible with Java 17. See
+
+- [CONTRIBUTING](https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md)
+- [monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)

data-prepper-plugins/dynamodb-source/build.gradle

@@ -0,0 +1,40 @@
+plugins {
+    id 'java'
+}
+
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    implementation project(path: ':data-prepper-api')
+    implementation project(path: ':data-prepper-core')
+    implementation project(path: ':data-prepper-plugins:dynamodb-source-coordination-store')
+
+    implementation libs.armeria.core
+    implementation 'io.micrometer:micrometer-core'
+
+    implementation 'software.amazon.awssdk:sts'
+    implementation 'software.amazon.awssdk:arns'
+    implementation 'software.amazon.awssdk:dynamodb'
+    implementation 'software.amazon.awssdk:dynamodb-enhanced'
+    implementation 'software.amazon.awssdk:s3'
+    implementation 'software.amazon.awssdk:netty-nio-client'
+
+    implementation 'com.fasterxml.jackson.core:jackson-core'
+    implementation 'com.fasterxml.jackson.core:jackson-databind'
+    // https://mvnrepository.com/artifact/com.fasterxml.jackson.dataformat/jackson-dataformat-ion
+    implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-ion'
+
+    implementation project(path: ':data-prepper-plugins:aws-plugin-api')
+
+
+    testImplementation platform('org.junit:junit-bom:5.9.1')
+    testImplementation 'org.junit.jupiter:junit-jupiter'
+    testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
+}
+
+test {
+    useJUnitPlatform()
+}

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/ClientFactory.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.source.dynamodb;
+
+import org.opensearch.dataprepper.aws.api.AwsCredentialsOptions;
+import org.opensearch.dataprepper.aws.api.AwsCredentialsSupplier;
+import org.opensearch.dataprepper.plugins.source.dynamodb.configuration.AwsAuthenticationConfig;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
+import software.amazon.awssdk.services.dynamodb.streams.DynamoDbStreamsClient;
+import software.amazon.awssdk.services.s3.S3Client;
+
+public class ClientFactory {
+
+    private final AwsCredentialsProvider awsCredentialsProvider;
+
+    public ClientFactory(AwsCredentialsSupplier awsCredentialsSupplier, AwsAuthenticationConfig awsAuthenticationConfig) {
+        awsCredentialsProvider = awsCredentialsSupplier.getProvider(AwsCredentialsOptions.builder()
+                .withRegion(awsAuthenticationConfig.getAwsRegion())
+                .withStsRoleArn(awsAuthenticationConfig.getAwsStsRoleArn())
+                .withStsExternalId(awsAuthenticationConfig.getAwsStsExternalId())
+                .withStsHeaderOverrides(awsAuthenticationConfig.getAwsStsHeaderOverrides())
+                .build());
+    }
+
+
+    public DynamoDbStreamsClient buildDynamoDbStreamClient() {
+        return DynamoDbStreamsClient.builder()
+                .credentialsProvider(awsCredentialsProvider)
+                .build();
+    }
+
+
+    public DynamoDbClient buildDynamoDBClient() {
+        return DynamoDbClient.builder()
+                .credentialsProvider(awsCredentialsProvider)
+                .build();
+    }
+
+
+    public S3Client buildS3Client() {
+        return S3Client.builder()
+                .credentialsProvider(awsCredentialsProvider)
+                .build();
+    }
+
+}