Add DynamoDB source plugin #3349

daixba · 2023-09-18T08:11:32Z

Description

Add support of DynamoDB as source.

Issues Resolved

Resolves #2932

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR.
- New functionality has javadoc added
Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

daixba · 2023-09-18T08:13:00Z

Submit the PR to receive early feedbacks. Still working on testing, will submit the UTs soon.

graytaylor0 · 2023-09-19T15:12:23Z

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

+    coordinator:
+      dynamodb:
+        table_name: "coordinator-demo"
+        region: "us-west-2"


Can you make an issue to get the configuration for this coordinator moved to the data-prepper-config.yaml to be completed in a future PR?

Sure. Will do once this code is merged. I will create one to refer to this.

It looks like we can either expose a getCoordinationStore from current LeaseBasedCoordinator or by default only create coordinationStore from the data-prepper-config.yaml only instead of the coordinator. Not sure what is the best way.

We will want to use the data-prepper-config only. We can use the same flow that we currently use to provide plugins with a source coordinaor

graytaylor0 · 2023-09-19T15:13:57Z

...source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/DynamoDBService.java

+        LOG.info("Gracefully shutdown DynamoDB schedulers");
+        executor.shutdownNow();


Nothing against calling shutdownNow, but I don't think it's categorized as graceful shutdown since it forces the thread to quit immediately

You are right, I think this is confusing, actually, the scheduler will be interrupped immediately (which is OK), and the scheduler will implement logic to shutdown the real jobs (such as stream consumer) gracefully. I will remove this.

graytaylor0 · 2023-09-19T15:19:02Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/DynamoDBSource.java

+    @Override
+    public void stop() {
+        LOG.info("Stop DynamoDB Source");
+        dynamoDBService.shutdown();


Should wrap this call in an if (Objects.nonNull(dynamoDBService) check. Not a big deal but NullPointer would happen if this gets run before line 66

Will update.

graytaylor0 · 2023-09-19T15:20:37Z

...c/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/coordination/Coordinator.java

+ *     <li>Support multiple leases at the same time</li>
+ * </ul>
+ */
+public interface Coordinator {


Could call this EnhancedSourceCoordinator

graytaylor0 · 2023-09-19T15:24:59Z

...c/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/DataFileScheduler.java

+                // Do nothing
+                // The consumer must already do one last checkpointing and then release the lease.
+                LOG.debug("Shard consumer completed with exception");
+                LOG.error(ex.toString());


Let's give more context to this error with a message saying where the exception has occurred

kkondaka · 2023-09-20T13:43:15Z

...main/java/org/opensearch/dataprepper/plugins/source/dynamodb/configuration/StreamConfig.java

+
+public class StreamConfig {
+
+    public static final String BEGINNING = "BEGINNING";


This should be "enum"

Update to use Enum.

I changed this back to String due to an exception that was raised in the new jackson lib (it was running OK previously) : Caused by: com.fasterxml.jackson.databind.JsonMappingException: Cannot construct instance of java.lang.Enum (no Creators, like default constructor, exist): abstract types either need to be mapped to concrete types, have custom deserializer, or contain additional type information.

Will try to fix this in later PRs.

.../main/java/org/opensearch/dataprepper/plugins/source/dynamodb/processor/RecordProcessor.java

dlvenable · 2023-09-20T19:05:06Z

...e/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/DynamoDBSourceConfig.java

+
+
+    @JsonProperty("coordinator")
+    private PluginModel coordinationStoreConfig;


I don't think we should be injecting coordination configurations here. This duplicates configurations and creates additional confusion for customers.

Perhaps the ideal is to require that customers configure the necessary coordinator in data-prepper-config.yaml?

This will be a separate PR (https://github.com/opensearch-project/data-prepper/pull/3349/files#r1330297254)

graytaylor0

Just realized the build is failing due to unit tests in the dynamo source. Once these are resolved I am good to approve. Thanks!

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

kkondaka

@daixba I do not see any code for e2e acknowledgement support? Will that be added as a separate PR?

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

dlvenable · 2023-10-02T16:01:50Z

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/DynamoDBSource.java

+
+    @Override
+    public void start(Buffer<Record<Event>> buffer) {
+        LOG.info("Start Processing");


Can you add something to this log line to help clarify what it starting. Perhaps "Start processing DynamoDB streams for {table} on {bucket}" or something similar.

Update to just 'Start DynamoDB service' for what it's actually doing, more log details will be in the service side.

...-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/DynamoDBSource.java

dlvenable · 2023-10-02T16:05:08Z

.../main/java/org/opensearch/dataprepper/plugins/source/dynamodb/converter/RecordConverter.java

+        return new Record<>(event);
+    }
+
+    public Record<Event> convertToEvent(Map<String, Object> data) {


It's generally better design to split concerns such as writing to the buffer and converting record types. Perhaps this can be a follow-on.

It was called RecordProcessor which purpose is to convert the raw records to Jackson event and then write to buffer. Now it's renamed to RecordConverter but I didn't change the scope of this class.

...c/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/DataFileScheduler.java

dlvenable · 2023-10-02T16:46:51Z

.../src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/DataFileLoader.java

+    /**
+     * Currently, this is to stop all consumers.
+     */
+    public static void stopAll() {


Since this is static, it will stop all consumers in all pipelines. Use an instance per pipeline instead.

Why this has impacts to other pipelines, if they are running in separate containers or different nodes?

Data Prepper can run multiple parallel pipelines. If you have two pipelines with the dynamodb source they will share the same static shouldStop variable. Thus, when one pipeline shuts down, the other pipeline's DataFileLoader is effectively stopped. This may not be what the user wants.

dlvenable · 2023-10-02T16:51:44Z

...c/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/DataFileScheduler.java

+
+    private final AtomicInteger numOfWorkers = new AtomicInteger(0);
+
+    private static final int MAX_JOB_COUNT = 2;


This should probably be tunable.

Will do more performance test later. The observation is that the file read is too fast but the ingestion can't catch up.

dlvenable · 2023-10-02T16:52:49Z

...c/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/export/DataFileScheduler.java

+    }
+
+    @Override
+    public void run() {


It seems we have this one thread which then creates two other threads. Can we eliminate this "scheduler" thread somehow? Perhaps just run two threads that each acquire a partition and then process it.

dlvenable · 2023-10-02T16:56:36Z

...e/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/stream/ShardConsumer.java

+                } else {
+                    records = response.records();
+                }
+                recordConverter.writeToBuffer(records);


Why do both the file reader thread path and stream scheduler thread path write to the buffer? What is the difference in these outputs?

They are converted into the same Jackson events and then sent to the same buffer and eventually same destination.

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

daixba · 2023-10-03T03:12:39Z

@daixba I do not see any code for e2e acknowledgement support? Will that be added as a separate PR?

It's not implemented yet (I don't know how it works). Will need a separate PR if needed.

graytaylor0 · 2023-10-03T15:30:37Z

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

+- `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


Looks like these metrics mainly cover export. Are there any metrics that we should add for streams? Things like number of shards processed?

dlvenable

There are still some unresolved comments from the PR. I have left the conversations in the PR as unresolved. Let's follow up with another PR to address them.

daixba requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, asifsmohammed, dlvenable and oeyh as code owners September 18, 2023 08:11

graytaylor0 reviewed Sep 19, 2023

View reviewed changes

kkondaka reviewed Sep 20, 2023

View reviewed changes

dlvenable requested changes Sep 21, 2023

View reviewed changes

daixba requested a review from dlvenable September 22, 2023 12:15

graytaylor0 approved these changes Sep 26, 2023

View reviewed changes

graytaylor0 requested changes Sep 26, 2023

View reviewed changes

Add DynamoDB source plugin

aefc57b

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

daixba force-pushed the ddb-source branch from 12b2474 to aefc57b Compare September 26, 2023 06:35

Fix UT error

af033b5

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

daixba force-pushed the ddb-source branch from 463216e to af033b5 Compare September 26, 2023 07:14

daixba added 5 commits September 26, 2023 15:48

Fix UT error

1e08325

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

Remove the source test

c8cc23a

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

Fix UT error

a3f5015

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

Fix typo

f02b5bd

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

Update stream config

7d513f8

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

graytaylor0 previously approved these changes Sep 26, 2023

View reviewed changes

kkondaka reviewed Sep 30, 2023

View reviewed changes

dlvenable requested changes Oct 2, 2023

View reviewed changes

Address PR comments

c12c051

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

daixba dismissed graytaylor0’s stale review via c12c051 October 3, 2023 03:11

graytaylor0 reviewed Oct 3, 2023

View reviewed changes

dlvenable approved these changes Oct 3, 2023

View reviewed changes

graytaylor0 approved these changes Oct 3, 2023

View reviewed changes

graytaylor0 merged commit b69f81b into opensearch-project:main Oct 3, 2023
46 of 47 checks passed

daixba deleted the ddb-source branch October 20, 2023 02:49

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add DynamoDB source plugin #3349

Add DynamoDB source plugin #3349

daixba commented Sep 18, 2023 •

edited

Loading

daixba commented Sep 18, 2023

graytaylor0 Sep 19, 2023

daixba Sep 20, 2023

graytaylor0 Sep 21, 2023

graytaylor0 Sep 19, 2023

daixba Sep 20, 2023

graytaylor0 Sep 19, 2023

daixba Sep 20, 2023 •

edited

Loading

graytaylor0 Sep 19, 2023

graytaylor0 Sep 19, 2023

kkondaka Sep 20, 2023

daixba Sep 22, 2023

daixba Sep 28, 2023

dlvenable Sep 20, 2023

graytaylor0 Sep 21, 2023

graytaylor0 left a comment

kkondaka left a comment

dlvenable Oct 2, 2023

daixba Oct 3, 2023

dlvenable Oct 2, 2023

daixba Oct 3, 2023

dlvenable Oct 2, 2023

daixba Oct 3, 2023

dlvenable Oct 3, 2023

dlvenable Oct 2, 2023

daixba Oct 3, 2023

dlvenable Oct 2, 2023

dlvenable Oct 2, 2023

daixba Oct 3, 2023

daixba commented Oct 3, 2023

graytaylor0 Oct 3, 2023 •

edited

Loading

dlvenable left a comment

		LOG.info("Gracefully shutdown DynamoDB schedulers");
		executor.shutdownNow();


		public class StreamConfig {

		public static final String BEGINNING = "BEGINNING";



		@JsonProperty("coordinator")
		private PluginModel coordinationStoreConfig;


		private final AtomicInteger numOfWorkers = new AtomicInteger(0);

		private static final int MAX_JOB_COUNT = 2;

Add DynamoDB source plugin #3349

Add DynamoDB source plugin #3349

Conversation

daixba commented Sep 18, 2023 • edited Loading

Description

Issues Resolved

Check List

daixba commented Sep 18, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

daixba Sep 20, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 left a comment

Choose a reason for hiding this comment

kkondaka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

daixba commented Oct 3, 2023

graytaylor0 Oct 3, 2023 • edited Loading

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

daixba commented Sep 18, 2023 •

edited

Loading

daixba Sep 20, 2023 •

edited

Loading

graytaylor0 Oct 3, 2023 •

edited

Loading