Add aggregate_threshold with maximum_size to s3 sink #4385

graytaylor0 · 2024-04-03T05:03:44Z

Description

This change adds an aggregate_threshold configuration to the s3 sink to prevent out of memory from holding too many groups in memory, including the following parameters

aggregate_threshold:
  // The size in bytes before force flushing begins
  maximum_size: "50mb" // default is 50 mb
  // Controls how aggressive the flush is by flushing the largest groups until this fraction of the maximum_size is remaining in the groups
  flush_capacity_ratio: 0.5 // default is 0.5

Issues Resolved

Related to #4345

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>

graytaylor0 · 2024-04-03T15:26:08Z

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

@@ -50,6 +55,10 @@ public Collection<S3Group> getS3GroupEntries() {
        return allGroups.values();
    }

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


I initially tried using a TreeSet for this, but the buffer size is changing so often, that groups need to be iterated over and re-added to the TreeSet whenever the function is called, which used more memory and is still O(n log(n)) just like sorting the list like this

graytaylor0 · 2024-04-03T15:31:39Z

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

@@ -46,6 +47,9 @@ public class S3SinkConfig {
    @NotNull
    private ThresholdOptions thresholdOptions;

+    @JsonProperty("aggregate_threshold")
+    private AggregateThresholdOptions aggregateThresholdOptions;


Should this be made to private AggregateThresholdOptions aggregateThresholdOptions = new AggregateThresholdOptions();, or should this remain null by default?

I like the idea of having a default. But, I think it would have to by dynamic. That is, we'd probably need to assign it like this:

max(DEFAULT_BYTE_CAPACITY, threshold.maximum_size)

As it is now, we may have a bug if you set this:

aggregate_threshold: flush_capacity_ratio: 0.25 threshold: maximum_size: 100mb

You would have an aggregate threshold of 50mb, but you are trying to get a 100mb threshold on each group.

This makes sense. It is a little strange to add defaults based on other config parameters though, as they aren't really connected in the code. Also, it may be better to not require the code that checks for aggregate thresholds to run every time by default, since in many cases (all existing cases) users only have a single group, making the aggregate threshold obsolete. For these reasons, I will just leave it with no default

Yea, let's not have any default - even the maximum_size. I think if the user is going to specify the aggregate_threshold at all, then he must also specify the maximum_size. This will require the user to think about it more and hopefully get it correct in relation to the other thresholds.

We could always change this later. But, if you remove the default 50mb for now, we won't have this possible confusion.

dlvenable · 2024-04-03T21:44:21Z

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

@@ -46,6 +47,9 @@ public class S3SinkConfig {
    @NotNull
    private ThresholdOptions thresholdOptions;

+    @JsonProperty("aggregate_threshold")
+    private AggregateThresholdOptions aggregateThresholdOptions;


I like the idea of having a default. But, I think it would have to by dynamic. That is, we'd probably need to assign it like this:

max(DEFAULT_BYTE_CAPACITY, threshold.maximum_size)

As it is now, we may have a bug if you set this:

aggregate_threshold: flush_capacity_ratio: 0.25 threshold: maximum_size: 100mb

You would have an aggregate threshold of 50mb, but you are trying to get a 100mb threshold on each group.

dlvenable · 2024-04-03T21:46:50Z

...java/org/opensearch/dataprepper/plugins/sink/s3/configuration/AggregateThresholdOptions.java

+    private static final String DEFAULT_BYTE_CAPACITY = "50mb";
+
+    @JsonProperty("maximum_size")
+    private String maximumSize = DEFAULT_BYTE_CAPACITY;


You can make this a ByteCount and data-prepper-core will parse it.

data-prepper/data-prepper-plugins/http-source/src/main/java/org/opensearch/dataprepper/plugins/source/loghttp/HTTPSourceConfig.java

Lines 96 to 97 in 41f2d64

@JsonProperty("max_request_length")

private ByteCount maxRequestLength;

dlvenable · 2024-04-03T21:48:37Z

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

-    Collection<EventHandle> getGroupEventHandles() {
-        return groupEventHandles;
+    @Override
+    public int compareTo(final S3Group o) {


Please add some unit tests for this.

dlvenable · 2024-04-03T21:52:14Z

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

+        final List<Buffer> expectedOrder = List.of(thirdBuffer, secondBuffer, buffer);
+
+        int index = 0;
+        for (final S3Group s3Group : sortedGroups) {


Can you do this?

assertThat(s3Group.size(), equalTo(3)); assertThat(sortedGroups, contains(thirdBuffer, secondBuffer, buffer));

dlvenable · 2024-04-03T21:52:47Z

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

+        assertThat(sortedGroupsAfterRemoval.size(), equalTo(2));
+        assertThat(objectUnderTest.getNumberOfGroups(), equalTo(2));
+
+        final List<Buffer> expectedOrderAfterRemoval = List.of(thirdBuffer, buffer);


Similarly:

assertThat(sortedGroupsAfterRemoval.size(), equalTo(2)); assertThat(sortedGroupsAfterRemoval, contains(thirdBuffer, buffer));

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

oeyh

Nice work identifying the potential memory issue and addressing it!

oeyh · 2024-04-04T17:36:55Z

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

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


I feel this is a little counter-intuitive that largerGroup.compareTo(smallerGroup) results in a negative value. But it's a minor thing.

Yeah it is a little odd for sure. But it means that sort() will give the reverse order, which is what we want.

@graytaylor0 , @oeyh , This is a good point. It might end up causing bugs later if we expand this.

How about making this return natural order and then use the following? That should work because it will provide a reverse comparison.

return allGroups.values().stream().sorted(Collections.reverseOrder()).collect(Collectors.toList());

Relevant Java code:

https://github.com/openjdk/jdk/blob/4a11db8b606f2b10f48f0b45335b661fe3095fc4/src/java.base/share/classes/java/util/Collections.java#L5548-L5550

and

https://github.com/openjdk/jdk/blob/4a11db8b606f2b10f48f0b45335b661fe3095fc4/src/java.base/share/classes/java/util/Collections.java#L5564-L5566

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

graytaylor0 requested review from chenqi0805, engechas, dinujoh, kkondaka, asifsmohammed, dlvenable and oeyh as code owners April 3, 2024 05:03

Add aggregate_threshold with maximum_size to s3 sink

7d052d9

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

graytaylor0 force-pushed the FinalAggregateThreshold branch from 1526dbd to 7d052d9 Compare April 3, 2024 05:04

graytaylor0 commented Apr 3, 2024

View reviewed changes

dlvenable requested changes Apr 3, 2024

View reviewed changes

Address PR comments

4709b66

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

graytaylor0 requested a review from dlvenable April 4, 2024 14:43

oeyh previously approved these changes Apr 4, 2024

View reviewed changes

graytaylor0 dismissed oeyh’s stale review via ebf9725 April 4, 2024 20:22

Address PR comment

ec98dc0

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

graytaylor0 force-pushed the FinalAggregateThreshold branch from ebf9725 to ec98dc0 Compare April 4, 2024 21:08

oeyh approved these changes Apr 4, 2024

View reviewed changes

dlvenable approved these changes Apr 4, 2024

View reviewed changes

graytaylor0 merged commit bcb1145 into opensearch-project:main Apr 4, 2024
47 checks passed

graytaylor0 deleted the FinalAggregateThreshold branch April 4, 2024 21:54

graytaylor0 mentioned this pull request Apr 8, 2024

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

Closed

4 tasks

dlvenable mentioned this pull request Apr 11, 2024

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

Merged

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 aggregate_threshold with maximum_size to s3 sink #4385

Add aggregate_threshold with maximum_size to s3 sink #4385

graytaylor0 commented Apr 3, 2024 •

edited

Loading

graytaylor0 Apr 3, 2024

graytaylor0 Apr 3, 2024

dlvenable Apr 3, 2024 •

edited

Loading

graytaylor0 Apr 4, 2024

dlvenable Apr 4, 2024

dlvenable Apr 3, 2024 •

edited

Loading

dlvenable Apr 3, 2024

dlvenable Apr 3, 2024

dlvenable Apr 3, 2024

dlvenable Apr 3, 2024

oeyh left a comment

oeyh Apr 4, 2024

graytaylor0 Apr 4, 2024

dlvenable Apr 5, 2024

	@JsonProperty("max_request_length")
	private ByteCount maxRequestLength;

Add aggregate_threshold with maximum_size to s3 sink #4385

Add aggregate_threshold with maximum_size to s3 sink #4385

Conversation

graytaylor0 commented Apr 3, 2024 • edited Loading

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable Apr 3, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dlvenable Apr 3, 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

Choose a reason for hiding this comment

oeyh 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

graytaylor0 commented Apr 3, 2024 •

edited

Loading

dlvenable Apr 3, 2024 •

edited

Loading

dlvenable Apr 3, 2024 •

edited

Loading