Add creation and aggregation of dynamic S3 groups based on events #4346

graytaylor0 · 2024-03-27T19:28:43Z

Description

This change adds a concept of S3Groups to the S3 Sink. Groups will be created dynamically based on the Events that come into the sink, and will be sent to S3 as a single object when the thresholds are reached for that group.

This PR only adds support for a dynamic path_prefix. The support for exposing object_name as a dynamic expression will be added in a future PR

Future Planned PRs

Refactor the buffer tracking of thresholds to the S3Group tracking this to reduce deduplication
Support aggregate_threshold for flushing the largest groups when the threshold is reached
Support injection of getEpochNanos() and getRandomUUID() into the dynamic expressions

Issues Resolved

Related to #4345

Ran the s3 sink integration tests successfully

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.

Signed-off-by: Taylor Gray <tylgry@amazon.com>

dinujoh · 2024-03-28T16:28:04Z

...prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/S3Sink.java

+        final S3GroupManager s3GroupManager = new S3GroupManager(s3SinkConfig, s3GroupIdentifierFactory, bufferFactory, s3Client);
+
+
+        s3SinkService = new S3SinkService(s3SinkConfig, codec, s3OutputCodecContext, s3Client, keyGenerator, Duration.ofSeconds(5), pluginMetrics, s3GroupManager);


nit: what is the duration ? Can you move that to constant ?

This was already here. Looks like the variable is called retrySleepTime. I can make it a constant

dinujoh · 2024-03-28T16:40:57Z

...-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/S3SinkService.java

+                }
+            }
+
+            for (final Map.Entry<S3GroupIdentifier, S3Group> s3GroupEntry : s3GroupManager.getS3GroupEntries()) {


should this be flushed when global threshold is reached ?

The global threshold will be handled in a follow-up PR, and will be configured separately as aggregate_threshold

I believe that @graytaylor0 will submit a follow-on to support the aggregate_threshold to do just that.

dlvenable · 2024-03-28T16:37:35Z

...src/test/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3GroupIdentifierTest.java

+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+public class S3GroupIdentifierTest {


We should test hashCode() as well.

dlvenable · 2024-03-28T16:46:00Z

...ns/s3-sink/src/integrationTest/java/org/opensearch/dataprepper/plugins/sink/s3/S3SinkIT.java

@@ -45,6 +45,7 @@
 import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;


We should add an integration test that verifies the key path is dynamic.

Is there a way for me to create an actual GenericExpressionEvaluator for the integration tests? Or can I verify this while mocking ExpressionEvaluator?

dlvenable · 2024-03-28T16:49:29Z

...-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/S3SinkService.java

                final Event event = record.getData();
+                final Map.Entry<S3GroupIdentifier, S3Group> s3GroupEntry = s3GroupManager.getOrCreateGroupForEvent(event);


You can add S3GroupIdentifier to S3Group as a private field. With this, you can simplify this method to return just a S3Group rather than a Map<Entry>. You would need to update removeGroup() to take in a S3Group. But, this will work because it would have the S3GroupIdentifier.

dlvenable · 2024-03-28T16:50:28Z

...ink/src/main/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3GroupIdentifier.java

+
+import java.util.Objects;
+
+public class S3GroupIdentifier {


If you take my comments above about having S3Group hold the S3GroupIdentifier, then you can make this class package protected. Let the other classes just deal with S3Group and not have to work about these internals.

dlvenable · 2024-03-28T16:55:57Z

.../main/java/org/opensearch/dataprepper/plugins/sink/s3/grouping/S3GroupIdentifierFactory.java

+
+    public S3GroupIdentifier getS3GroupIdentifierForEvent(final Event event) {
+
+        final String groupIdentificationHash = keyGenerator.generateKeyForEvent(event, false);


This logic couples the generateKeyForEvent code greatly with the hashing. A change in that method would break this logic. Not only that, but it depends on ObjectKey which thus broadens the places where a behavioral change could break this.

I recommend that you make this a Map<String, Object> instead of a String. The values might look like:

/my_key/path -> myStringValue1

You would need to add some code that extracts key paths from the expression. This might be something to add to ExpressionEvaluator, though I'm not sure if that is the best place.

I think this would have to be in ExpressionEvaluator since it would also need to run the expressions to create the values of the Map.

Or are you thinking that expression evaluator would support a function to extract all of the ${} sections out of a format expression, and then each of these individual expressions would need to be run separately to get the values? Or can we just have expression evaluator return the full Map result?

Or are you thinking that expression evaluator would support a function to extract all of the ${} sections out of a format expression, and then each of these individual expressions would need to be run separately to get the values?

This is what I was thinking. And they should come out in a normalized manner. That is, all have the leading /. This way you always get a consistent value.

Or can we just have expression evaluator return the full Map result?

This could be an option as well. But, it seems a little more specific to this use-case. Getting the list of keys may be more extensible.

This is what I was thinking. And they should come out in a normalized manner. That is, all have the leading /. This way you always get a consistent value.

So in this case, the S3 sink would have to explicitly check for keys and other expressions, and would have to check the Event itself for the values and run the expression evaluation functions individually as well

Signed-off-by: Taylor Gray <tylgry@amazon.com>

dinujoh · 2024-04-01T20:50:36Z

...pression/src/main/java/org/opensearch/dataprepper/expression/GenericExpressionEvaluator.java

+        int position = 0;
+        while ((position = format.indexOf("${", fromIndex)) != -1) {
+            int endPosition = format.indexOf("}", position + 1);
+            if (endPosition == -1) {


This might be a user error in dynamic expression. Looks like we are ignoring any errors ?

This method is not responsible for catching invalid expressions, it will just parse out the valid ones. If you think it should throw, then it can here but it's job is not to validate, just to parse and return

Will add some unit tests for validating the expression on startup of the S3 sink in the next PR

We can add a test case to cover validation on startup for s3 sink constructor as discussed offline.

dinujoh · 2024-04-01T20:57:17Z

...sion/src/test/java/org/opensearch/dataprepper/expression/GenericExpressionEvaluatorTest.java

+
+    @ParameterizedTest
+    @ArgumentsSource(FormatExpressionsToExtractedDynamicExpressionsArgumentProvider.class)
+    void extractDynamicExpressionsFromFormatExpression_returns_expected_result(final String formatExpression, final List<String> expectedDynamicExpressions) {


can we add a negative test where invalid expression is passed ?

There is a negative case where invalid expression syntax is found, it will just not be added to the list. This method is just supposed to parse out valid existing expressions

Add creation and aggregation of dynamic S3 groups based on events

33b2e85

Signed-off-by: Taylor Gray <tylgry@amazon.com>

graytaylor0 requested review from chenqi0805, engechas, dinujoh, kkondaka, asifsmohammed, dlvenable and oeyh as code owners March 27, 2024 19:28

dinujoh reviewed Mar 28, 2024

View reviewed changes

dlvenable requested changes Mar 28, 2024

View reviewed changes

Address PR comments

3a6ad06

Signed-off-by: Taylor Gray <tylgry@amazon.com>

graytaylor0 force-pushed the S3Groups branch from 9b506d9 to 8daaf43 Compare April 1, 2024 19:00

Track event handles in S3Group, add integration test

c175522

Signed-off-by: Taylor Gray <tylgry@amazon.com>

graytaylor0 force-pushed the S3Groups branch from 8daaf43 to c175522 Compare April 1, 2024 19:05

dlvenable previously approved these changes Apr 1, 2024

View reviewed changes

Add more test coverage

27fb6a9

Signed-off-by: Taylor Gray <tylgry@amazon.com>

graytaylor0 dismissed dlvenable’s stale review via 27fb6a9 April 1, 2024 19:36

dlvenable approved these changes Apr 1, 2024

View reviewed changes

dinujoh reviewed Apr 1, 2024

View reviewed changes

dinujoh approved these changes Apr 1, 2024

View reviewed changes

graytaylor0 merged commit 1a7e099 into opensearch-project:main Apr 1, 2024
46 of 47 checks passed

graytaylor0 deleted the S3Groups branch April 1, 2024 22:43

graytaylor0 mentioned this pull request Apr 8, 2024

[DOC] Data Prepper S3 sink new features opensearch-project/documentation-website#6925

Closed

4 tasks

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

dlvenable mentioned this pull request May 16, 2024

Allow using event fields in s3 sink object_key #3310

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add creation and aggregation of dynamic S3 groups based on events #4346

Add creation and aggregation of dynamic S3 groups based on events #4346

graytaylor0 commented Mar 27, 2024 •

edited

Loading

dinujoh Mar 28, 2024

graytaylor0 Mar 28, 2024

dinujoh Mar 28, 2024

graytaylor0 Mar 28, 2024

dlvenable Mar 28, 2024

dlvenable Mar 28, 2024

dlvenable Mar 28, 2024

graytaylor0 Mar 28, 2024

dlvenable Mar 28, 2024

dlvenable Mar 28, 2024

dlvenable Mar 28, 2024

graytaylor0 Mar 28, 2024

graytaylor0 Mar 28, 2024

dlvenable Mar 28, 2024

graytaylor0 Mar 28, 2024

dinujoh Apr 1, 2024

graytaylor0 Apr 1, 2024

graytaylor0 Apr 1, 2024

dinujoh Apr 1, 2024

dinujoh Apr 1, 2024

graytaylor0 Apr 1, 2024

		final S3GroupManager s3GroupManager = new S3GroupManager(s3SinkConfig, s3GroupIdentifierFactory, bufferFactory, s3Client);


		s3SinkService = new S3SinkService(s3SinkConfig, codec, s3OutputCodecContext, s3Client, keyGenerator, Duration.ofSeconds(5), pluginMetrics, s3GroupManager);

		@@ -45,6 +45,7 @@
		import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;

		final Event event = record.getData();
		final Map.Entry<S3GroupIdentifier, S3Group> s3GroupEntry = s3GroupManager.getOrCreateGroupForEvent(event);


		public S3GroupIdentifier getS3GroupIdentifierForEvent(final Event event) {

		final String groupIdentificationHash = keyGenerator.generateKeyForEvent(event, false);

Add creation and aggregation of dynamic S3 groups based on events #4346

Add creation and aggregation of dynamic S3 groups based on events #4346

Conversation

graytaylor0 commented Mar 27, 2024 • edited Loading

Description

Issues Resolved

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

graytaylor0 commented Mar 27, 2024 •

edited

Loading