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

deepaksahu562 · 2023-05-05T13:53:54Z

Description

Implementation of s3-sink local-file 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.

kkondaka · 2023-05-09T05:52:05Z

...ink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferTypeOptions.java

 /**
 * Defines all the buffer types enumerations.
 */
 public enum BufferTypeOptions {

-    INMEMORY,
-    LOCALFILE
+    LOCALFILE("local_file", new LocalFileBuffer());


I thought the plan is to support both INMEMORY and LOCALFILE. Isn't it?

We are supporting both INMEMORY and LOCALFILE buffer type. The INMEMORY buffer functionality was raised as separate PR: #2623

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

codecov · 2023-05-15T11:23:01Z

Codecov Report

Merging #2645 (c2394dc) into main (191ede2) will increase coverage by 0.04%.
The diff coverage is n/a.

@@             Coverage Diff              @@
##               main    #2645      +/-   ##
============================================
+ Coverage     93.56%   93.61%   +0.04%     
  Complexity     2254     2254              
============================================
  Files           262      262              
  Lines          6308     6308              
  Branches        521      521              
============================================
+ Hits           5902     5905       +3     
+ Misses          268      266       -2     
+ Partials        138      137       -1

see 2 files with indirect coverage changes

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

deepaksahu562 · 2023-05-15T18:46:47Z

@ashoktelukuntla \ @dlvenable
Rebase the branch and all checks have been passed.
Would you please review and merge it.

dlvenable · 2023-05-15T18:37:53Z

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

+
+    LocalFileBuffer() {
+        try {
+            localFile = new File(String.valueOf(UUID.randomUUID()));


We should use a temp file here. It could be in the system default temp, or we can allow a user to configure this directory.

Java provides two createTempFile functions that can help here.

Addressed. Implemented using Java provided functionality.

dlvenable · 2023-05-15T18:38:48Z

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

+public class LocalFileBuffer implements Buffer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(LocalFileBuffer.class);
+    private BufferedOutputStream bufferedOutputStream;


I'd recommend making this just an OutputStream so that the code doesn't have to change if you change this stream class.

private OutputStream outputStream;

dlvenable · 2023-05-15T18:41:07Z

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

+     */
+    @Override
+    public void writeEvent(byte[] bytes) throws IOException {
+        bufferedOutputStream.write(bytes);


We need some synchronization for these writes. Or the events may end up out-of-order.

We can add it here (and to the InMemoryBuffer). Or perhaps we can synchronize in the S3SinkService itself.

Yes, we have synchronized in the S3SinkService itself.

dlvenable · 2023-05-15T18:43:17Z

...est/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactoryTest.java

+        Assertions.assertNotNull(localFileBufferFactory);
+        Buffer buffer = localFileBufferFactory.getBuffer();
+        Assertions.assertNotNull(buffer);
+        assertThat(buffer, instanceOf(Buffer.class));


You should assert that this is an instance of LocalFileBuffer as well.

assertThat(buffer, instanceOf(LocalFileBuffer.class))

@deepaksahu562 , Did you push the change here? I don't see a new assertion for this.

@dlvenable, Earlier, I didn't understand. Now I can understand and modify as requested.

It should be as simple as changing this line to:

assertThat(buffer, instanceOf(LocalFileBuffer.class))

Here it is with a little more context.

Buffer buffer = localFileBufferFactory.getBuffer(); Assertions.assertNotNull(buffer); assertThat(buffer, instanceOf(LocalFileBuffer.class))

We already know this is a Buffer. That is handled statically by Java. But, we want to know that this is returning the LocalFileBuffer and not an InMemoryBuffer (or some other buffer sub-class).

Please change this line per my latest comment. Then we should be good.

Thanks for your clarification, Modified as you suggested.

dlvenable · 2023-05-15T18:44:40Z

...k/src/test/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferTest.java

+        }
+        assertThat(localFileBuffer.getSize(), greaterThan(1l));
+        assertThat(localFileBuffer.getEventCount(), equalTo(55));
+        assertThat(localFileBuffer.getDuration(), greaterThanOrEqualTo(0L));


You should be able to assert the exact duration here.

assertThat(localFileBuffer.getDuration(), equalTo(55 * 1000L));

dlvenable · 2023-05-15T18:48:20Z

...k/src/test/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferTest.java

+        assertDoesNotThrow(() -> {
+            localFileBuffer.flushToS3(s3Client, BUCKET_NAME, "log.txt");
+        });
+    }


You need to verify that the file was flushed to S3. Since this is not an integration test, you will do this via mocking.

It should look something like the following:

ArgumentCaptor<PutObjectRequest> putObjectRequestArgumentCaptor = ArgumentCaptor.forClass(PutObjectRequest.class); verify(s3Client).putObject(putObjectRequestArgumentCaptor.capture(), any(RequestBody.class)); PutObjectRequest actualRequest = putObjectRequestArgumentCaptor.getValue(); // add assertions for the bucket and key values.

Also, this should validate that the file was actually deleted. You will do this with real files (not mocks).

Modified and addressed as per suggestions.

dlvenable · 2023-05-15T18:49:44Z

...rc/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactory.java

+public class LocalFileBufferFactory implements BufferFactory {
+    @Override
+    public Buffer getBuffer() {
+        return new LocalFileBuffer();


You can probably improve your testing by choosing the File here and passing it in as a parameter.

File = File.createTempDir(...); return new LocalFileBuffer(file);

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

dlvenable · 2023-05-16T14:28:55Z

...ink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferTypeOptions.java

@@ -15,7 +18,16 @@
 */
 public enum BufferTypeOptions {

-    INMEMORY("in_memory", new InMemoryBuffer());
+    INMEMORY("in_memory", new InMemoryBuffer()),
+    LOCALFILE("local_file", new Object() {


Why did you change to make this anonymous class? I think you can continue to make this work with new LocalFileBufferFactory. This helps with unit testing.

Modified with new LocalFileBufferFactory.

dlvenable · 2023-05-16T14:31:02Z

...est/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactoryTest.java

+        Assertions.assertNotNull(localFileBufferFactory);
+        Buffer buffer = localFileBufferFactory.getBuffer();
+        Assertions.assertNotNull(buffer);
+        assertThat(buffer, instanceOf(Buffer.class));


@deepaksahu562 , Did you push the change here? I don't see a new assertion for this.

dlvenable · 2023-05-16T14:32:22Z

...rc/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactory.java

+        File tempFile = null;
+        Buffer localfileBuffer = null;
+        try {
+            tempFile = File.createTempFile(PREFIX, null);


Maybe the suffix should be .log?

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

dlvenable · 2023-05-16T15:36:38Z

...ink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferTypeOptions.java

        this.option = option.toLowerCase();
        this.bufferType = bufferType;
    }

-    public Buffer getBufferType() {
+    public BufferFactory getBufferType() {


Yes, this is a good change. Thanks!

dlvenable · 2023-05-16T15:36:55Z

...est/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactoryTest.java

+        Assertions.assertNotNull(localFileBufferFactory);
+        Buffer buffer = localFileBufferFactory.getBuffer();
+        Assertions.assertNotNull(buffer);
+        assertThat(buffer, instanceOf(Buffer.class));


Please change this line per my latest comment. Then we should be good.

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

kkondaka

Looks good.

deepaksahu562 requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, cmanning09, asifsmohammed, dlvenable and oeyh as code owners May 5, 2023 13:53

deepaksahu562 mentioned this pull request May 5, 2023

Github-issue#1048 : s3-sink with local-file buffer implementation. deepaksahu562/data-prepper#2

Closed

4 tasks

kkondaka reviewed May 9, 2023

View reviewed changes

deepaksahu562 closed this May 13, 2023

deepaksahu562 force-pushed the s3-sink-temp-local-file branch from 4bfe6b1 to 2512d1c Compare May 13, 2023 19:36

GitHub-issue#1048 : Rebase the code from DP main branch.

ab1862c

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

deepaksahu562 reopened this May 15, 2023

deepaksahu562 added 2 commits May 15, 2023 18:45

GitHub-issue#1048

00e2e5f

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

GitHub-issue#1048

dbe2811

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

dlvenable requested changes May 15, 2023

View reviewed changes

deepaksahu562 added 2 commits May 16, 2023 15:25

Merge branch 'opensearch-project:main' into s3-sink-temp-local-file

1cff631

GitHub-issue#1048 : Incorporated review comments.

45e1242

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

deepaksahu562 requested a review from dlvenable May 16, 2023 13:58

dlvenable requested changes May 16, 2023

View reviewed changes

GitHub-issue#1048 : Incorporated review comments.

00f59cf

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

deepaksahu562 requested a review from dlvenable May 16, 2023 15:17

dlvenable requested changes May 16, 2023

View reviewed changes

GitHub-issue#1048 : Incorporated review comments.

c2394dc

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

deepaksahu562 requested a review from dlvenable May 16, 2023 15:54

dlvenable approved these changes May 16, 2023

View reviewed changes

kkondaka approved these changes May 16, 2023

View reviewed changes

kkondaka merged commit 9c7b3ec into opensearch-project:main May 16, 2023
26 checks passed

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 local-file buffer implementation. #2645

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

deepaksahu562 commented May 5, 2023 •

edited

kkondaka May 9, 2023

venkataraopasyavula May 12, 2023

codecov bot commented May 15, 2023 •

edited

deepaksahu562 commented May 15, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 16, 2023

deepaksahu562 May 16, 2023

dlvenable May 16, 2023

dlvenable May 16, 2023

deepaksahu562 May 16, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 15, 2023

deepaksahu562 May 16, 2023

dlvenable May 16, 2023

deepaksahu562 May 16, 2023

dlvenable May 16, 2023

dlvenable May 16, 2023

deepaksahu562 May 16, 2023

dlvenable May 16, 2023

dlvenable May 16, 2023

kkondaka left a comment

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

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

Conversation

deepaksahu562 commented May 5, 2023 • edited

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented May 15, 2023 • edited

Codecov Report

deepaksahu562 commented May 15, 2023

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

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 left a comment

Choose a reason for hiding this comment

deepaksahu562 commented May 5, 2023 •

edited

codecov bot commented May 15, 2023 •

edited