SQS source plugin implementation #5274

jmsusanto · 2024-12-18T23:03:37Z

Description

added the sqs source plugin

Issues Resolved

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.

Signed-off-by: Jeremy Michael <[email protected]>

…e implementation Signed-off-by: Jeremy Michael <[email protected]>

Signed-off-by: Jeremy Michael <[email protected]>

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java

san81 · 2024-12-27T17:49:27Z

...ce/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/AwsAuthenticationOptions.java

+
+     @JsonProperty("sts_role_arn")
+     @Size(min = 20, max = 2048, message = "awsStsRoleArn length should be between 1 and 2048 characters")
+     protected String awsStsRoleArn;


is there a reason to make these are protected? Will there be a possible child class or something?

great catch, there's no reason for awsStsRoleArn to be protected. I think IntelliJ originally changed it for a test. Changing this to private.

san81 · 2024-12-27T17:58:12Z

...qs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsEventProcessor.java

+
+    void addSqsObject(final Message message,
+                      final String url,
+                      final BufferAccumulator<Record<Event>> bufferAccumulator,


BufferAccumulator is not thread safe. I see that you are creating BufferAccumulator in each SqsWorker thread which is good. Just that I am not sure event processor (or the message handler) becomes another thread? if not, then we are good. If yes then you need make sure that it is not creating any synchronization issues here.

As you've said, I created each BufferAccumulator in each SqsWorker thread. Within each SqsWorker, no new threads are created, so concurrency shouldn't be an issue here. However, please let me know if my understanding is incorrect here.

Signed-off-by: Jeremy Michael <[email protected]>

kkondaka

Please write an integration Test like https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/s3-source/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/s3/SqsServiceIT.java and https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/s3-source/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorkerIT.java

dlvenable · 2024-12-31T15:55:58Z

...gins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java

+    private static final Boolean DEFAULT_VISIBILITY_DUPLICATE_PROTECTION = false;
+    private static final Duration DEFAULT_VISIBILITY_TIMEOUT_SECONDS = null;
+    private static final Duration DEFAULT_VISIBILITY_DUPLICATE_PROTECTION_TIMEOUT = Duration.ofHours(2);
+    private static final Duration DEFAULT_WAIT_TIME_SECONDS = Duration.ofSeconds(20);


This default should also be null so that we use the SQS queue default value.

The SQS queue default is 0 seconds which results in short-polling. Is this desirable? There are many benefits to long-polling (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-short-and-long-polling.html#sqs-long-polling)

I agree that long-polling is more desirable. So maybe this would make more sense to use a default of 20 seconds. The main reason I suggested using the queue default is because it would be consistent with the other behaviors.

But, maybe because long polling is a little different from the other configurations, this would be better to use a Data Prepper-defined default.

@kkondaka , Any thoughts on this behavior?

dlvenable · 2024-12-31T15:56:48Z

...gins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java

+public class QueueConfig {
+
+    private static final Integer DEFAULT_MAXIMUM_MESSAGES = null;
+    private static final Boolean DEFAULT_VISIBILITY_DUPLICATE_PROTECTION = false;


I think we should use this by default. It provides a lot of benefit and the main downside is that it requires an additional permission.

S3 source current defaults to not using this, and i think this is more significant in the S3 case since a single message can refer to a large s3 object and hence taking a lot of time.

On the contrary, I don't see a reason not to use this given the benefits

We'd like to make the S3 source using this as the default in Data Prepper 3.0. See #3679.

...gins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfig.java

dlvenable · 2024-12-31T16:00:02Z

...ce/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/AwsAuthenticationOptions.java

+         try {
+             return Arn.fromString(awsStsRoleArn);
+         } catch (final Exception e) {
+             throw new IllegalArgumentException(String.format("Invalid ARN format for awsStsRoleArn. Check the format of %s", awsStsRoleArn));


Suggested change

throw new IllegalArgumentException(String.format("Invalid ARN format for awsStsRoleArn. Check the format of %s", awsStsRoleArn));

throw new IllegalArgumentException(String.format("The value provided for sts_role_arn is not a valid AWS ARN. Provided value: %s", awsStsRoleArn));

we should make this change in s3 source's aws config as well for consistency

I agree. We should also consolidate this code.

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java

...ugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsService.java

dlvenable · 2024-12-31T16:08:53Z

...lugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsWorker.java

+import java.util.Optional;
+
+
+public class SqsWorker implements Runnable {


This code looks to duplicate much of the work from the S3 source. Having duplicate classes is going to be very difficult to maintain as we fix bugs. We should consolidate this into a library.

...lugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsWorker.java

dlvenable · 2024-12-31T16:16:12Z

...ce/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/AwsAuthenticationOptions.java

+     @Size(max = 5, message = "sts_header_overrides supports a maximum of 5 headers to override")
+     private Map<String, String> awsStsHeaderOverrides;
+
+     void validateStsRoleArn() {


Please use @AssertTrue here.

Something like:

@AssertTrue("The sts_role_arn must be a valid IAM Role ARN.")

similar to above, we should change this in S3 source too for consistency

another note: this method isn't used by either s3 source or sqs at the moment

dlvenable · 2024-12-31T16:17:21Z

Please write an integration Test like https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/s3-source/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/s3/SqsServiceIT.java and https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/s3-source/src/integrationTest/java/org/opensearch/dataprepper/plugins/source/s3/SqsWorkerIT.java

I completely agree, but I am also ok with this coming in a follow-on PR in order to reduce the PR scope.

Signed-off-by: Jeremy Michael <[email protected]>

dlvenable

Thank you @jmsusanto ! Just a few more smaller comments now.

dlvenable · 2025-01-07T15:41:55Z

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java

@@ -1,76 +1,51 @@
 package org.opensearch.dataprepper.plugins.source.sqs;


This file appears to be missing the copyright header.

Great catch. Thought I already looked over every file, my bad

dlvenable · 2025-01-07T15:42:55Z

...lugins/sqs-source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/SqsWorker.java

@@ -160,16 +162,18 @@ private void applyBackoff() {
    private ReceiveMessageRequest createReceiveMessageRequest() {
        ReceiveMessageRequest.Builder requestBuilder = ReceiveMessageRequest.builder()
                .queueUrl(queueConfig.getUrl())
-                .waitTimeSeconds((int) queueConfig.getWaitTime().getSeconds());
+                .attributeNamesWithStrings("All")


We should make this configurable because it probably affects performance. But, I think that would be appropriate for a follow-on PR. Using All as the default makes sense.

We can have the customer specify the metadata that they want to intake in the config. Definitely has performance and cost implications

dlvenable · 2025-01-07T15:45:09Z

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java

+                    .withData(Collections.singletonMap("message", message.body()))
+                    .build();
+            final EventMetadata eventMetadata = event.getMetadata();
+            eventMetadata.setAttribute("url", url);


Let's use metadata keys that match the terminology from SQS. So keep this as queueUrl as you had previously.

dlvenable · 2025-01-07T15:45:40Z

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java

+            final EventMetadata eventMetadata = event.getMetadata();
+            eventMetadata.setAttribute("url", url);
+            final String sentTimestamp = systemAttributes.get(MessageSystemAttributeName.SENT_TIMESTAMP);
+            eventMetadata.setAttribute("SentTimestamp", sentTimestamp);


For consistency with other Data Prepper metadata, we should probably use lowerCamelCase here - sentTimestamp.

missed that, thanks

dlvenable · 2025-01-07T15:48:26Z

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java

+            final String sentTimestamp = systemAttributes.get(MessageSystemAttributeName.SENT_TIMESTAMP);
+            eventMetadata.setAttribute("SentTimestamp", sentTimestamp);
+            for (Map.Entry<String, MessageAttributeValue> entry : customAttributes.entrySet()) {
+                eventMetadata.setAttribute(entry.getKey(), entry.getValue().stringValue());


Can you use this for SentTimestamp as well?

I think it may sense to lowercase the first letter of the key to have consistency with other Data Prepper conventions. So we'd have keys like senderId and messageDeduplicationId.

@kkondaka , Any thoughts on this?

@dlvenable I agree. Let's make it consistent with lowercase "s"

Do you mean the lower the first letter of all custom keys? l think this may confuse the customer since they specify their own custom keys when sending the sqs message. They'll probably expect the exact same key name when referencing it later in the pipeline.

Yes, that is what we mean. I think this may help with consistency.

got it, changed

dlvenable · 2025-01-07T15:49:16Z

.../sqs-source/src/test/java/org/opensearch/dataprepper/plugins/source/sqs/QueueConfigTest.java

@@ -0,0 +1,26 @@
+package org.opensearch.dataprepper.plugins.source.sqs;


Please add the copyright header to all the tests.

kkondaka · 2025-01-07T18:14:55Z

@jmsusanto following up from email chain with "cyberark", I think we should parse the sqs message to create multiple events if there are array of events. Can you check with CyberArk folks see how exactly their SQS message with multiple events would look and see if that can be addressed in the SQS source?

Signed-off-by: Jeremy Michael <[email protected]>

dlvenable

Thank you @jmsusanto for this contribution and the improvements toward it!

Jeremy Michael and others added 9 commits December 18, 2024 14:58

initial poc of new sqs-source plugin

e2fc86b

Signed-off-by: Jeremy Michael <[email protected]>

modified README

1448ada

Signed-off-by: Jeremy Michael <[email protected]>

Changed parser to format described in issue

f91d939

Signed-off-by: Jeremy Michael <[email protected]>

Renamed sqs-source-new to sqs-source and removed deprecated sqs-sourc…

b303f92

…e implementation Signed-off-by: Jeremy Michael <[email protected]>

Update README.md

05d1daa

Signed-off-by: Jeremy Michael <[email protected]>

Update README.md

51db627

Signed-off-by: Jeremy Michael <[email protected]>

Add queueUrl and sentTimestamp metadata to SQS events

5e5e84d

Signed-off-by: Jeremy Michael <[email protected]>

Assign individual BufferAccumulators to each SQS worker

b79f0d8

Signed-off-by: Jeremy Michael <[email protected]>

added unit tests and addressed comments in design doc

29701d0

Signed-off-by: Jeremy Michael <[email protected]>

jmsusanto requested review from sb2k16, chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, KarstenSchnitter, dlvenable and oeyh as code owners December 18, 2024 23:03

added additional unit tests

b103f7e

Signed-off-by: Jeremy Michael <[email protected]>

jmsusanto force-pushed the sqsSource branch from 8bbb8d9 to b103f7e Compare December 26, 2024 19:00

jmsusanto changed the title ~~SQS source: initial working implementation~~ SQS source plugin implementation Dec 26, 2024

Fix region loading check

114e227

Signed-off-by: Jeremy Michael <[email protected]>

san81 reviewed Dec 27, 2024

View reviewed changes

...source/src/main/java/org/opensearch/dataprepper/plugins/source/sqs/RawSqsMessageHandler.java Outdated Show resolved Hide resolved

san81 reviewed Dec 27, 2024

View reviewed changes

addressed PR comments

e5f2f79

Signed-off-by: Jeremy Michael <[email protected]>

kkondaka reviewed Dec 29, 2024

View reviewed changes

dlvenable requested changes Dec 31, 2024

View reviewed changes

addressed PR comments - modified Buffer and parser implementation

be297ca

Signed-off-by: Jeremy Michael <[email protected]>

dlvenable requested changes Jan 7, 2025

View reviewed changes

Jeremy Michael added 2 commits January 7, 2025 13:17

added copyright headers to tests and updated metadata naming

b95a1a5

Signed-off-by: Jeremy Michael <[email protected]>

some cleanup and converted all metadata to lowerCamelCase

02f42c3

Signed-off-by: Jeremy Michael <[email protected]>

dlvenable approved these changes Jan 9, 2025

View reviewed changes

kkondaka approved these changes Jan 9, 2025

View reviewed changes

kkondaka merged commit 2aa376e into opensearch-project:main Jan 9, 2025
72 of 79 checks passed

	throw new IllegalArgumentException(String.format("Invalid ARN format for awsStsRoleArn. Check the format of %s", awsStsRoleArn));
	throw new IllegalArgumentException(String.format("The value provided for sts_role_arn is not a valid AWS ARN. Provided value: %s", awsStsRoleArn));

		import java.util.Optional;


		public class SqsWorker implements Runnable {

		@@ -1,76 +1,51 @@
		package org.opensearch.dataprepper.plugins.source.sqs;

		@@ -0,0 +1,26 @@
		package org.opensearch.dataprepper.plugins.source.sqs;

SQS source plugin implementation #5274

SQS source plugin implementation #5274

Conversation

jmsusanto commented Dec 18, 2024 • edited Loading

Description

Issues Resolved

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

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

jmsusanto Jan 2, 2025 • 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

dlvenable commented Dec 31, 2024

dlvenable 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kkondaka commented Jan 7, 2025

dlvenable left a comment

Choose a reason for hiding this comment

jmsusanto commented Dec 18, 2024 •

edited

Loading

jmsusanto Jan 2, 2025 •

edited

Loading