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

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

@@ -16,7 +16,9 @@
 import org.opensearch.dataprepper.model.sink.AbstractSink;
 import org.opensearch.dataprepper.model.sink.Sink;
 import org.opensearch.dataprepper.plugins.sink.accumulator.BufferFactory;
+import org.opensearch.dataprepper.plugins.sink.accumulator.BufferTypeOptions;
 import org.opensearch.dataprepper.plugins.sink.accumulator.InMemoryBufferFactory;
+import org.opensearch.dataprepper.plugins.sink.accumulator.LocalFileBufferFactory;
 import org.opensearch.dataprepper.plugins.sink.codec.Codec;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -52,7 +54,11 @@ public S3Sink(final PluginSetting pluginSetting, final S3SinkConfig s3SinkConfig
         codec = pluginFactory.loadPlugin(Codec.class, codecPluginSettings);
         sinkInitialized = Boolean.FALSE;
 
-        bufferFactory = new InMemoryBufferFactory();
+        if (s3SinkConfig.getBufferType().equals(BufferTypeOptions.LOCALFILE)) {
+            bufferFactory = new LocalFileBufferFactory();
+        } else {
+            bufferFactory = new InMemoryBufferFactory();
+        }
     }
 
     @Override

data-prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferTypeOptions.java

@@ -15,7 +15,8 @@
  */
 public enum BufferTypeOptions {
 
-    INMEMORY("in_memory", new InMemoryBuffer());
+    INMEMORY("in_memory", new InMemoryBuffer()),
+    LOCALFILE("local_file", new LocalFileBuffer());
 
     private final String option;
     private final Buffer bufferType;

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

@@ -0,0 +1,112 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.accumulator;
+
+import org.apache.commons.lang3.time.StopWatch;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A buffer can hold local file data and flushing it to S3.
+ */
+public class LocalFileBuffer implements Buffer {
+
+    private static final Logger LOG = LoggerFactory.getLogger(LocalFileBuffer.class);
+    private BufferedOutputStream bufferedOutputStream;
+    private int eventCount;
+    private final StopWatch watch;
+    private File localFile;
+
+    LocalFileBuffer() {
+        try {
+            localFile = new File(String.valueOf(UUID.randomUUID()));
+            bufferedOutputStream = new BufferedOutputStream(new FileOutputStream(localFile));
+        } catch (IOException e) {
+            LOG.error("Unable to create local file ", e);
+        }
+
+        eventCount = 0;
+
+        watch = new StopWatch();
+        watch.start();
+    }
+
+    @Override
+    public long getSize() {
+        try {
+            bufferedOutputStream.flush();
+        } catch (IOException e) {
+            LOG.error("An exception occurred while flushing data to buffered output stream :", e);
+        }
+        return localFile.length();
+    }
+
+    @Override
+    public int getEventCount() {
+        return eventCount;
+    }
+
+    @Override
+    public long getDuration(){
+        return watch.getTime(TimeUnit.SECONDS);
+    }
+
+    /**
+     * Upload accumulated data to amazon s3.
+     * @param s3Client s3 client object.
+     * @param bucket bucket name.
+     * @param key s3 object key path.
+     */
+    @Override
+    public void flushToS3(S3Client s3Client, String bucket, String key) {
+        try {
+            bufferedOutputStream.flush();
+            bufferedOutputStream.close();
+        } catch (IOException e) {
+            LOG.error("An exception occurred while flushing data to buffered output stream :", e);
+        }
+        s3Client.putObject(
+                PutObjectRequest.builder().bucket(bucket).key(key).build(),
+                RequestBody.fromFile(localFile));
+        removeTemporaryFile();
+    }
+
+    /**
+     * write byte array to output stream.
+     * @param bytes byte array.
+     * @throws IOException while writing to output stream fails.
+     */
+    @Override
+    public void writeEvent(byte[] bytes) throws IOException {
+        bufferedOutputStream.write(bytes);
+        bufferedOutputStream.write(System.lineSeparator().getBytes());
+        eventCount++;
+    }
+
+    /**
+     * Remove the local temp file after flushing data to s3.
+     */
+    private void removeTemporaryFile() {
+        if (localFile != null) {
+            try {
+                Files.deleteIfExists(Paths.get(localFile.toString()));
+            } catch (IOException e) {
+                LOG.error("Unable to delete Local file {}", localFile, e);
+            }
+        }
+    }
+}

data-prepper-plugins/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactory.java

@@ -0,0 +1,13 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.accumulator;
+
+public class LocalFileBufferFactory implements BufferFactory {
+    @Override
+    public Buffer getBuffer() {
+        return new LocalFileBuffer();
+    }
+}

data-prepper-plugins/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferFactoryTest.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.sink.accumulator;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class LocalFileBufferFactoryTest {
+    @Test
+    void test_localFileBufferFactory_notNull() {
+        LocalFileBufferFactory localFileBufferFactory = new LocalFileBufferFactory();
+        Assertions.assertNotNull(localFileBufferFactory);
+    }
+
+    @Test
+    void test_buffer_notNull() {
+        LocalFileBufferFactory localFileBufferFactory = new LocalFileBufferFactory();
+        Assertions.assertNotNull(localFileBufferFactory);
+        Buffer buffer = localFileBufferFactory.getBuffer();
+        Assertions.assertNotNull(buffer);
+        assertThat(buffer, instanceOf(Buffer.class));
+    }
+}

data-prepper-plugins/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/accumulator/LocalFileBufferTest.java

@@ -0,0 +1,79 @@
+package org.opensearch.dataprepper.plugins.sink.accumulator;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import software.amazon.awssdk.services.s3.S3Client;
+import java.io.IOException;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+
+@ExtendWith(MockitoExtension.class)
+class LocalFileBufferTest {
+
+    public static final String BUCKET_NAME = UUID.randomUUID().toString();
+    @Mock
+    private S3Client s3Client;
+    private LocalFileBuffer localFileBuffer;
+
+    @Test
+    void test_with_write_event_into_buffer() throws IOException {
+        localFileBuffer = new LocalFileBuffer();
+
+        while (localFileBuffer.getEventCount() < 55) {
+            localFileBuffer.writeEvent(generateByteArray());
+        }
+        assertThat(localFileBuffer.getSize(), greaterThan(1l));
+        assertThat(localFileBuffer.getEventCount(), equalTo(55));
+        assertThat(localFileBuffer.getDuration(), greaterThanOrEqualTo(0L));
+    }
+
+    @Test
+    void test_without_write_event_into_buffer() {
+        localFileBuffer = new LocalFileBuffer();
+        assertThat(localFileBuffer.getSize(), equalTo(0L));
+        assertThat(localFileBuffer.getEventCount(), equalTo(0));
+        assertThat(localFileBuffer.getDuration(), lessThanOrEqualTo(0L));
+
+    }
+
+    @Test
+    void test_with_write_event_into_buffer_and_flush_toS3() throws IOException {
+        localFileBuffer = new LocalFileBuffer();
+
+        while (localFileBuffer.getEventCount() < 55) {
+            localFileBuffer.writeEvent(generateByteArray());
+        }
+        assertThat(localFileBuffer.getSize(), greaterThan(1l));
+        assertThat(localFileBuffer.getEventCount(), equalTo(55));
+        assertThat(localFileBuffer.getDuration(), greaterThanOrEqualTo(0L));
+        assertDoesNotThrow(() -> {
+            localFileBuffer.flushToS3(s3Client, BUCKET_NAME, "log.txt");
+        });
+    }
+
+    @Test
+    void test_uploadedToS3_success() {
+        localFileBuffer = new LocalFileBuffer();
+        Assertions.assertNotNull(localFileBuffer);
+        assertDoesNotThrow(() -> {
+            localFileBuffer.flushToS3(s3Client, BUCKET_NAME, "log.txt");
+        });
+    }
+
+    private byte[] generateByteArray() {
+        byte[] bytes = new byte[1000];
+        for (int i = 0; i < 1000; i++) {
+            bytes[i] = (byte) i;
+        }
+        return bytes;
+    }
+}