Github-issue#1048 : s3-sink with in_memory buffer implementation. #2623

deepaksahu562 · 2023-05-02T17:02:24Z

Description

Implementation of s3-sink in-memory buffer functionality.

Issues Resolved

GitHub-issue #1048

Check List

New functionality includes testing.
New functionality has been documented.
- 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: Deepak Sahu <deepak.sahu562@gmail.com>

asifsmohammed · 2023-05-02T17:20:14Z

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

+        numEvents = s3SinkConfig.getThresholdOptions().getEventCount();
+        byteCapacity = s3SinkConfig.getThresholdOptions().getMaximumSize();
+        duration = s3SinkConfig.getThresholdOptions().getEventCollectTimeOut().getSeconds();


Please update these variable names for better readability, maybe maxEvents, maxBytes, maxCollectionDuration.

asifsmohammed · 2023-05-02T17:22:25Z

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

+                    if (isUploadedToS3) {
+                        LOG.info("Snapshot uploaded successfully");
+                    } else {
+                        LOG.info("Snapshot upload failed");
+                    }


Are we dropping the records which are failed to upload to S3?
I think metrics are more helpful here instead of logs.

We need to have retry or send the data to DLQ.

Yes records will be dropped if upload fails even after reaching max retries.
As suggested added metrics.

We need to have retry or send the data to DLQ.

Added retry functionality.

asifsmohammed · 2023-05-02T17:37:25Z

...3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/InMemoryBuffer.java

+    private final StopWatch watch;
+
+    InMemoryBuffer() {
+        byteArrayOutputStream = new ByteArrayOutputStream();


Can we reset the stream instead of creating a new stream object every time?

Modified the code to reset the stream instead of creating a new stream object every time.

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

kkondaka · 2023-05-03T15:18:49Z

...plugins/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/S3SinkServiceTest.java

+        when(codec.parse(any())).thenReturn("{\"message\":\"31824252-adba-4c47-a2ac-05d16c5b8140\"}");
+        S3SinkService s3SinkService = new S3SinkService(s3SinkConfig, bufferFactory, codec);
+        assertNotNull(s3SinkService);
+        s3SinkService.output(generateRandomStringEventRecord());


You have to assert some metrics here.

Added metrics.

kkondaka · 2023-05-03T15:19:37Z

...plugins/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/S3SinkServiceTest.java

+        S3SinkService s3SinkService = new S3SinkService(s3SinkConfig, bufferFactory, codec);
+        assertNotNull(s3SinkService);
+        s3SinkService.output(generateRandomStringEventRecord());
+        assertThat(s3SinkService, instanceOf(S3SinkService.class));


Please move this one line above. No point to asserting this after doing output().

kkondaka · 2023-05-03T15:26:17Z

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

+        reentrantLock.lock();
+        final String bucket = s3SinkConfig.getBucketOptions().getBucketName();
+        if (currentBuffer == null) {
+            currentBuffer = bufferFactory.getBuffer();


Why initialize the currentBuffer here?

Approaches suggested by David.

kkondaka · 2023-05-03T15:28:04Z

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

+                            " Event_count = {} Records & Event_collection_duration = {} Sec",
+                            maxBytes.getBytes(), currentBuffer.getEventCount(), currentBuffer.getDuration());
+                    boolean isUploadedToS3 = currentBuffer.flushToS3(s3Client, bucket, generateKey());
+                    if (isUploadedToS3) {


We should retry few times before failing. And log the failure reason clearly.

kkondaka · 2023-05-03T15:30:25Z

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

+                final String encodedEvent;
+                encodedEvent = codec.parse(event);
+                final byte[] encodedBytes = encodedEvent.getBytes();
+                if (willExceedThreshold()) {


What if this condition never becomes true? It will never be uploaded to S3?

event_count & event_collect_timeout are mandatory attributes so the condition will toggle to false/true.

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

…data-prepper into s3-sink-in_memory

deepaksahu562 · 2023-05-08T13:18:15Z

@dlvenable, Thanks for your review suggestions.
Addressed Change request, Please review one again.

deepaksahu562 · 2023-05-08T13:19:36Z

The build is failing with the following:

Error: eckstyle] [ERROR] /home/runner/work/data-prepper/data-prepper/data-prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/S3Sink.java:18:59: Using the '.*' form of import should be avoided - org.opensearch.dataprepper.plugins.sink.accumulator.*. [AvoidStarImport]

You can check your build locally to see any other errors:

./gradlew -p data-prepper-plugins/s3-sink clean check

Resolved.

kkondaka · 2023-05-08T15:51:42Z

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

+                encodedEvent = codec.parse(event);
+                final byte[] encodedBytes = encodedEvent.getBytes();
+
+                if (ThresholdCheck.checkThresholdExceed(currentBuffer, maxEvents, maxBytes, maxCollectionDuration)) {


If the maxEvents is 10 and only one event is received, no other event is sent for one hour, I don't see how that one event is flushed to S3.

This is handled in event collection duration i.e. event_collect_timeout . When the duration is met it will be flushed to S3

As mentioned in the email, this is not handled today. We need a way to handle this in all sinks. We will tackle this in a separate PR (either sink specific or at data prepper core level)

ashoktelukuntla

Looks good

kkondaka

Looks good.

dlvenable

Overall this is good. But, the build is currently failing:

> Task :data-prepper-plugins:s3-sink:checkstyleTest
FAILURE: Build failed with an exception.
> Task :data-prepper-plugins:s3-sink:spotlessMarkdownCheck FAILED

* What went wrong:
Execution failed for task ':data-prepper-plugins:s3-sink:spotlessMarkdownCheck'.
> The following files had format violations:
      README.md
          @@ -63,4 +63,4 @@
           This·plugin·is·compatible·with·Java·8.·See
           
           -·[CONTRIBUTING](https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md)
          --·[monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
          +-·[monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
  Run './gradlew :data-prepper-plugins:s3-sink:spotlessApply' to fix these violations.

You can run the checkstyle locally to verify:

./gradlew -p data-prepper-plugins/s3-sink check

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

deepaksahu562 · 2023-05-11T13:34:25Z

Overall this is good. But, the build is currently failing:

> Task :data-prepper-plugins:s3-sink:checkstyleTest
FAILURE: Build failed with an exception.
> Task :data-prepper-plugins:s3-sink:spotlessMarkdownCheck FAILED

* What went wrong:
Execution failed for task ':data-prepper-plugins:s3-sink:spotlessMarkdownCheck'.
> The following files had format violations:
      README.md
          @@ -63,4 +63,4 @@
           This·plugin·is·compatible·with·Java·8.·See
           
           -·[CONTRIBUTING](https://github.com/opensearch-project/data-prepper/blob/main/CONTRIBUTING.md)
          --·[monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
          +-·[monitoring](https://github.com/opensearch-project/data-prepper/blob/main/docs/monitoring.md)
  Run './gradlew :data-prepper-plugins:s3-sink:spotlessApply' to fix these violations.

You can run the checkstyle locally to verify:

./gradlew -p data-prepper-plugins/s3-sink check

Addressed.

dlvenable · 2023-05-11T14:15:54Z

@deepaksahu562 , I see the same error still:

https://github.com/opensearch-project/data-prepper/actions/runs/4948572707/jobs/8849475477?pr=2623

dlvenable · 2023-05-11T15:35:22Z

I checked out your work, and the suggestion seemed to get me past the error.

./gradlew :data-prepper-plugins:s3-sink:spotlessApply

However, I do see failing unit tests.

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

codecov · 2023-05-12T19:26:46Z

Codecov Report

Merging #2623 (55816be) into main (961d492) will increase coverage by 0.01%.
The diff coverage is 100.00%.

@@             Coverage Diff              @@
##               main    #2623      +/-   ##
============================================
+ Coverage     93.52%   93.54%   +0.01%     
- Complexity     2238     2250      +12     
============================================
  Files           261      262       +1     
  Lines          6275     6291      +16     
  Branches        519      520       +1     
============================================
+ Hits           5869     5885      +16     
  Misses          268      268              
  Partials        138      138

Impacted Files	Coverage Δ
.../dataprepper/model/event/DefaultEventMetadata.java	`91.30% <100.00%> (+0.82%)`	⬆️
...aprepper/expression/HasTagsExpressionFunction.java	`100.00% <100.00%> (ø)`

deepaksahu562 · 2023-05-12T19:49:07Z

I checked out your work, and the suggestion seemed to get me past the error.
./gradlew :data-prepper-plugins:s3-sink:spotlessApply
However, I do see failing unit tests.

@dlvenable \ @ashoktelukuntla Thanks for your suggestions. Build issue resolved and all checks have passed. Could you please review

kkondaka

Looks good.

…ensearch-project#2623) * Github-issue#1048 : s3-sink with in-memory buffer implementation. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink with in-memory buffer implementation. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink with in-memory buffer implementation. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink - added JUnit test classes. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink - incorporated review comment. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink - incorporated review comment. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink - local-file buffer implementation. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : s3-sink - in-memory buffer implementation. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : resolved - checkstyle error. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : incorporated review comment. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * Github-issue#1048 : incorporated review comment. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * GitHub-issue#1048 : Incorporated review comments. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * GitHub-issue#1048 : Incorporated review comments. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * GitHub-issue#1048 : Incorporated review comments. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> * GitHub-issue#1048 : Resolved javadoc issues. Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> --------- Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com> Signed-off-by: rajeshLovesToCode <rajesh.dharamdasani3021@gmail.com>

…ion. (opensearch-project#2623)" This reverts commit e71f866.

deepaksahu562 added 3 commits May 2, 2023 22:15

Github-issue#1048 : s3-sink with in-memory buffer implementation.

6949b23

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

Merge branch 'opensearch-project:main' into s3-sink-in_memory

36c8473

Github-issue#1048 : s3-sink with in-memory buffer implementation.

9fb9077

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

deepaksahu562 requested review from chenqi0805, engechas, graytaylor0, dinujoh, cmanning09, asifsmohammed, dlvenable and oeyh as code owners May 2, 2023 17:02

Github-issue#1048 : s3-sink with in-memory buffer implementation.

f43d521

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

asifsmohammed reviewed May 2, 2023

View reviewed changes

deepaksahu562 changed the title ~~S3 sink in memory~~ Github-issue#1048 : s3-sink with in memory buffer implementation. May 2, 2023

deepaksahu562 changed the title ~~Github-issue#1048 : s3-sink with in memory buffer implementation.~~ Github-issue#1048 : s3-sink with in_memory buffer implementation. May 3, 2023

Merge branch 'opensearch-project:main' into s3-sink-in_memory

156c938

deepaksahu562 requested a review from kkondaka as a code owner May 3, 2023 08:28

deepaksahu562 added 2 commits May 3, 2023 16:47

Github-issue#1048 : s3-sink - added JUnit test classes.

e33b943

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

Github-issue#1048 : s3-sink - incorporated review comment.

8d9c427

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

kkondaka reviewed May 3, 2023

View reviewed changes

deepaksahu562 added 6 commits May 4, 2023 19:27

Merge branch 'opensearch-project:main' into s3-sink-in_memory

fc1b0ce

Github-issue#1048 : s3-sink - incorporated review comment.

deed3d2

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

Merge branch 'opensearch-project:main' into s3-sink-in_memory

9d064f4

Github-issue#1048 : s3-sink - local-file buffer implementation.

3b737f8

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

Github-issue#1048 : s3-sink - in-memory buffer implementation.

2a52abf

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

Merge branch 's3-sink-in_memory' of https://github.com/deepaksahu562/…

e02195b

…data-prepper into s3-sink-in_memory

kkondaka reviewed May 8, 2023

View reviewed changes

ashoktelukuntla previously approved these changes May 8, 2023

View reviewed changes

kkondaka previously approved these changes May 9, 2023

View reviewed changes

dlvenable approved these changes May 11, 2023

View reviewed changes

dlvenable requested changes May 11, 2023

View reviewed changes

deepaksahu562 added 2 commits May 11, 2023 18:58

Merge branch 'opensearch-project:main' into s3-sink-in_memory

093bde5

GitHub-issue#1048 : Incorporated review comments.

2e722ab

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

deepaksahu562 requested a review from dlvenable May 11, 2023 13:36

GitHub-issue#1048 : Incorporated review comments.

159663c

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

deepaksahu562 dismissed stale reviews from ashoktelukuntla and kkondaka via 159663c May 11, 2023 15:58

venkataraopasyavula mentioned this pull request May 12, 2023

Github-issue#1048 : s3-sink with local-file buffer implementation. #2645

Merged

4 tasks

deepaksahu562 added 4 commits May 12, 2023 18:01

Merge branch 'opensearch-project:main' into s3-sink-in_memory

610697e

GitHub-issue#1048 : Incorporated review comments.

837f4c2

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

Merge branch 'opensearch-project:main' into s3-sink-in_memory

837f146

GitHub-issue#1048 : Resolved javadoc issues.

55816be

Signed-off-by: Deepak Sahu <deepak.sahu562@gmail.com>

dlvenable approved these changes May 12, 2023

View reviewed changes

kkondaka approved these changes May 12, 2023

View reviewed changes

kkondaka merged commit 95c319d into opensearch-project:main May 12, 2023
48 checks passed

rajeshLovesToCode added a commit to rajeshLovesToCode/data-prepper that referenced this pull request May 18, 2023

Revert "Github-issue#1048 : s3-sink with in_memory buffer implementat…

3e4549f

…ion. (opensearch-project#2623)" This reverts commit e71f866.

rajeshLovesToCode added a commit to rajeshLovesToCode/data-prepper that referenced this pull request May 18, 2023

Revert "Github-issue#1048 : s3-sink with in_memory buffer implementat…

32ee71b

…ion. (opensearch-project#2623)" This reverts commit e71f866.

dlvenable mentioned this pull request Jun 5, 2023

Support S3 as a Sink #1048

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Github-issue#1048 : s3-sink with in_memory buffer implementation. #2623

Github-issue#1048 : s3-sink with in_memory buffer implementation. #2623

deepaksahu562 commented May 2, 2023 •

edited

asifsmohammed May 2, 2023

deepaksahu562 May 4, 2023

asifsmohammed May 2, 2023

ashoktelukuntla May 4, 2023

deepaksahu562 May 4, 2023

deepaksahu562 May 8, 2023

asifsmohammed May 2, 2023

deepaksahu562 May 4, 2023

kkondaka May 3, 2023

deepaksahu562 May 4, 2023

kkondaka May 3, 2023

deepaksahu562 May 4, 2023

kkondaka May 3, 2023

deepaksahu562 May 8, 2023

kkondaka May 3, 2023

deepaksahu562 May 4, 2023

kkondaka May 3, 2023

deepaksahu562 May 4, 2023

deepaksahu562 commented May 8, 2023

deepaksahu562 commented May 8, 2023

kkondaka May 8, 2023

ashoktelukuntla May 8, 2023

kkondaka May 9, 2023

ashoktelukuntla left a comment

kkondaka left a comment

dlvenable left a comment

deepaksahu562 commented May 11, 2023

dlvenable commented May 11, 2023

dlvenable commented May 11, 2023

codecov bot commented May 12, 2023 •

edited

deepaksahu562 commented May 12, 2023

kkondaka left a comment

Github-issue#1048 : s3-sink with in_memory buffer implementation. #2623

Github-issue#1048 : s3-sink with in_memory buffer implementation. #2623

Conversation

deepaksahu562 commented May 2, 2023 • edited

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

deepaksahu562 commented May 8, 2023

deepaksahu562 commented May 8, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ashoktelukuntla left a comment

Choose a reason for hiding this comment

kkondaka left a comment

Choose a reason for hiding this comment

dlvenable left a comment

Choose a reason for hiding this comment

deepaksahu562 commented May 11, 2023

dlvenable commented May 11, 2023

dlvenable commented May 11, 2023

codecov bot commented May 12, 2023 • edited

Codecov Report

deepaksahu562 commented May 12, 2023

kkondaka left a comment

Choose a reason for hiding this comment

deepaksahu562 commented May 2, 2023 •

edited

codecov bot commented May 12, 2023 •

edited