Add support for dynamic bucket and default bucket in S3 sink #4402

graytaylor0 · 2024-04-09T20:10:20Z

Description

This change adds support for passing a dynamic bucket expression to route data to different buckets based on Events. It also supports a default_bucket

Check List

New functionality includes testing.
New functionality has a documentation issue. Please link to it in this PR. ([DOC] Data Prepper S3 sink new features documentation-website#6925)
- 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.

dlvenable · 2024-04-11T15:38:05Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/accumulator/BufferUtilities.java

+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import software.amazon.awssdk.services.s3.model.S3Exception;
+
+public class BufferUtilities {


Please make this package private.

dlvenable · 2024-04-11T15:39:52Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/Event.java

@@ -143,11 +143,12 @@ public interface Event extends Serializable {
     * of a Data Prepper expression
     * @param format input format
     * @param expressionEvaluator - The expression evaluator that will support formatting from Data Prepper expressions
+     * @param replacementForFailures - The String to use as a replacement for when keys in Events can't be found


Should we just call this defaultValue?

Yeah that does make more sense I think

dlvenable · 2024-04-11T15:41:15Z

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/Event.java

     * @return returns a string with no formatted parts, returns null if no value is found
     * @throws RuntimeException if the input string is not properly formatted
     * @since 2.1
     */
-    String formatString(String format, ExpressionEvaluator expressionEvaluator);
+    String formatString(final String format, final ExpressionEvaluator expressionEvaluator, final String replacementForFailures);


You can override this to avoid the changes that have the explicit null.

String formatString(final String format, final ExpressionEvaluator expressionEvaluator, final String replacementForFailures); default String formatString(final String format, final ExpressionEvaluator expressionEvaluator) { return formatString(format, expressionEvaluator, null); }

dlvenable · 2024-04-11T15:42:06Z

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/JacksonEventTest.java

+    @Test
+    public void testBuild_withFormatStringWithValueNotFound_and_replacement_failure() {
+
+        final String replacementForMissingKeys = "REPLACED";


Use a random string here.

dlvenable · 2024-04-11T15:42:40Z

...rc/main/java/org/opensearch/dataprepper/plugins/processor/mutateevent/AddEntryProcessor.java

@@ -54,7 +54,7 @@ public Collection<Record<Event>> doExecute(final Collection<Record<Event>> recor
                    }

                    try {
-                        final String key = (entry.getKey() == null) ? null : recordEvent.formatString(entry.getKey(), expressionEvaluator);
+                        final String key = (entry.getKey() == null) ? null : recordEvent.formatString(entry.getKey(), expressionEvaluator, null);


This line, and many others, can be removed by making the override I suggested above.

dlvenable · 2024-04-11T15:44:56Z

...s/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/codec/parquet/S3OutputStream.java

@@ -42,6 +46,8 @@ public class S3OutputStream extends PositionOutputStream {
     */
    private final String key;

+    private String targetBucket;


I don't think we need this field. Just use the existing bucket field and make it non-final.

dlvenable · 2024-04-11T15:45:05Z

...s/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/codec/parquet/S3OutputStream.java

+    /**
+     * The default bucket to send to when upload fails with dynamic bucket
+     */
+    private String defaultBucket;


Make this value final.

dinujoh · 2024-04-11T15:48:26Z

...s/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/codec/parquet/S3OutputStream.java

+                createMultipartUpload();
+            } catch (final S3Exception e) {
+                if (defaultBucket != null && (e instanceof NoSuchBucketException || e.getMessage().contains(ACCESS_DENIED))) {
+                    targetBucket = defaultBucket;


The createMultipartUpload method can be updated to take bucket as parameter.

This doesn't really help because we are going to use a non-final bucket name since we need the bucket in other locations in the code

dinujoh · 2024-04-11T15:50:12Z

...ugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3Group.java

@@ -46,6 +46,6 @@ public void releaseEventHandles(final boolean result) {

    @Override
    public int compareTo(final S3Group o) {
-        return Long.compare(o.getBuffer().getSize(), buffer.getSize());
+        return Long.compare(buffer.getSize(), o.getBuffer().getSize());


was this a bug ?

Not a bug but we were sorting in order, now we sort in reverse order

Thanks for remembering this!

@dinujoh , For reference: #4385 (comment)

dinujoh · 2024-04-11T15:50:55Z

...3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3GroupManager.java

@@ -56,7 +57,7 @@ public Collection<S3Group> getS3GroupEntries() {
    }

    public Collection<S3Group> getS3GroupsSortedBySize() {
-        return allGroups.values().stream().sorted().collect(Collectors.toList());
+        return allGroups.values().stream().sorted(Collections.reverseOrder()).collect(Collectors.toList());


what the reason for reverse order ?

Because we swapped the comparator

@dinujoh , Also see: #4385 (comment)

Signed-off-by: Taylor Gray <[email protected]>

dlvenable · 2024-04-11T19:08:52Z

...rc/test/java/org/opensearch/dataprepper/plugins/sink/s3/accumulator/BufferUtilitiesTest.java

+    }
+
+    @Test
+    void putObjectOrSendToDefaultBucket_with_no_such_bucket_exception_and_null_default_bucket_throws_exception() {


You can combine this test and the one below into a @ParameterizedTest.

dlvenable · 2024-04-11T19:10:06Z

...rc/test/java/org/opensearch/dataprepper/plugins/sink/s3/accumulator/BufferUtilitiesTest.java

+    }
+
+    @Test
+    void putObjectOrSendToDefaultBucket_with_NoSuchBucketException_sends_to_default_bucket() {


These two tests can be combined using @ParameterizedTest

dlvenable · 2024-04-11T19:11:20Z

...rc/test/java/org/opensearch/dataprepper/plugins/sink/s3/accumulator/BufferUtilitiesTest.java

+    }
+
+    @Test
+    void putObjectOrSendToDefaultBucket_with_S3Exception_that_is_not_access_denied_or_no_such_bucket_throws_exception() {


I think you also want a test case that has a defaultBucket, but an exception that is not Access Denied or NoSuchBucket. That should throw and not call putObject.

That is what this test was supposed to be but looks like I didn't set the default bucket. I will turn it into a parameterized test

dlvenable · 2024-04-11T19:11:36Z

...rc/test/java/org/opensearch/dataprepper/plugins/sink/s3/accumulator/BufferUtilitiesTest.java

+    void putObjectOrSendToDefaultBucket_with_no_such_bucket_exception_and_null_default_bucket_throws_exception() {
+        when(s3Client.putObject(any(PutObjectRequest.class), eq(requestBody))).thenThrow(NoSuchBucketException.class);
+
+        assertThrows(NoSuchBucketException.class, () -> BufferUtilities.putObjectOrSendToDefaultBucket(s3Client, requestBody, objectKey, targetBucket, null));


Please also put a verification on the number of calls to putObject. There should be only one.

dlvenable · 2024-04-11T19:12:41Z

...ugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3Group.java

@@ -46,6 +46,6 @@ public void releaseEventHandles(final boolean result) {

    @Override
    public int compareTo(final S3Group o) {
-        return Long.compare(o.getBuffer().getSize(), buffer.getSize());
+        return Long.compare(buffer.getSize(), o.getBuffer().getSize());


Thanks for remembering this!

@dinujoh , For reference: #4385 (comment)

dlvenable · 2024-04-11T19:13:14Z

...3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3GroupManager.java

@@ -56,7 +57,7 @@ public Collection<S3Group> getS3GroupEntries() {
    }

    public Collection<S3Group> getS3GroupsSortedBySize() {
-        return allGroups.values().stream().sorted().collect(Collectors.toList());
+        return allGroups.values().stream().sorted(Collections.reverseOrder()).collect(Collectors.toList());


@dinujoh , Also see: #4385 (comment)

Signed-off-by: Taylor Gray <[email protected]>

dlvenable

Thank you @graytaylor0 !

graytaylor0 requested review from chenqi0805, engechas, dinujoh, kkondaka, asifsmohammed, KarstenSchnitter, dlvenable and oeyh as code owners April 9, 2024 20:10

graytaylor0 force-pushed the DynamicBucketSupport branch 2 times, most recently from ed5956a to 5f70f0d Compare April 9, 2024 21:13

dlvenable requested changes Apr 11, 2024

View reviewed changes

dinujoh reviewed Apr 11, 2024

View reviewed changes

graytaylor0 added 2 commits April 11, 2024 11:44

Add support for dynamic bucket in S3 sink

2848c15

Signed-off-by: Taylor Gray <[email protected]>

Address PR comments

57c4e14

Signed-off-by: Taylor Gray <[email protected]>

graytaylor0 force-pushed the DynamicBucketSupport branch from 5f70f0d to 57c4e14 Compare April 11, 2024 18:22

graytaylor0 requested review from dlvenable and dinujoh April 11, 2024 19:02

dlvenable requested changes Apr 11, 2024

View reviewed changes

Add / modify unit tests

5d63680

Signed-off-by: Taylor Gray <[email protected]>

graytaylor0 force-pushed the DynamicBucketSupport branch from 72254d5 to 5d63680 Compare April 11, 2024 19:39

graytaylor0 requested a review from dlvenable April 11, 2024 20:46

dlvenable approved these changes Apr 11, 2024

View reviewed changes

dinujoh approved these changes Apr 11, 2024

View reviewed changes

graytaylor0 merged commit 02aef54 into opensearch-project:main Apr 11, 2024
46 of 47 checks passed

graytaylor0 deleted the DynamicBucketSupport branch April 11, 2024 21:51

graytaylor0 mentioned this pull request Apr 11, 2024

Fix bug in s3 sink dynamic bucket and catch invalid bucket message #4413

Merged

4 tasks

kkondaka added this to the v2.8 milestone May 14, 2024

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for dynamic bucket and default bucket in S3 sink #4402

Add support for dynamic bucket and default bucket in S3 sink #4402

graytaylor0 commented Apr 9, 2024 •

edited

Loading

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

graytaylor0 Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dinujoh Apr 11, 2024

graytaylor0 Apr 11, 2024

dinujoh Apr 11, 2024

graytaylor0 Apr 11, 2024

dlvenable Apr 11, 2024

dinujoh Apr 11, 2024

graytaylor0 Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

graytaylor0 Apr 11, 2024 •

edited

Loading

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable Apr 11, 2024

dlvenable left a comment

Add support for dynamic bucket and default bucket in S3 sink #4402

Add support for dynamic bucket and default bucket in S3 sink #4402

Conversation

graytaylor0 commented Apr 9, 2024 • edited Loading

Description

Check List

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 Apr 11, 2024 • 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

dlvenable left a comment

Choose a reason for hiding this comment

graytaylor0 commented Apr 9, 2024 •

edited

Loading

graytaylor0 Apr 11, 2024 •

edited

Loading