Created "s3-sink" plugin. github-issue#1048 #2362

deepaksahu562 · 2023-03-03T16:29:07Z

Description

Created "s3-sink" plugin.
Configurations for the bucket name, key path and key pattern.
The key pattern support timestamps such as -logs-${YYYY.mm}-${uniqueId}. Collection of objects from Buffer and store it in RAM/Local file.

Issues Resolved

Github issue: #1048

Check List

New functionality "s3-sink" plugin.
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-03-07T20:45:16Z

data-prepper-plugins/s3-sink/README.md

@@ -0,0 +1,9 @@
+Attached simple-sample-pipeline: opensearch-data-prepper\data-prepper\data-prepper-plugins\s3-sink\src\main\resources\pipelines.yaml


Please look at other README.md files for the format of this file. And modify this file to be similar to those files.

Thanks, @kkondaka for your review. I have modified README.md file as per other existing file format.

kkondaka · 2023-03-07T20:46:45Z

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

+    /**
+     *   Validate the index with the regular expression pattern. Throws exception if validation fails
+     */
+    public static DateTimeFormatter getDatePatternFormatter(final String indexAlias) {


Looks like this is similar to the code in the OpenSearchSink code. Could we move this to some common library and use it in both places?

Thanks, @kkondaka for your review. I have moved this code into common library.

kkondaka · 2023-03-07T20:50:52Z

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

+                .region(s3SinkConfig.getAwsAuthenticationOptions().getAwsRegion())
+                .credentialsProvider(s3SinkConfig.getAwsAuthenticationOptions().authenticateAwsConfiguration())
+                .overrideConfiguration(ClientOverrideConfiguration.builder()
+                        .retryPolicy(RetryPolicy.builder().numRetries(5).build())


number of retries should be configurable

Configured in yaml.

kkondaka · 2023-03-07T20:54:55Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ThresholdOptions.java

+    /**
+     *  Read the event count configuration
+     */
+    public int getEventCount() {


If you are expecting the event_count to be a multiple of 100, add the check for that here like if (eventCount % 100 != 0) throw new RuntimeException("...."

This code has been refactored, we are not using multiple of 100 for now. however check applied.

kkondaka · 2023-03-07T20:58:50Z

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

+/**
+ * {@link SinkAccumulator} Accumulate buffer records
+ */
+public interface SinkAccumulator {


Where is this being used? the inMemoryAccumulator and localFileAccumulator should implement this interface

Removed this interface, not being used.

kkondaka · 2023-03-27T17:29:12Z

...lugins/common/src/main/java/org/opensearch/dataprepper/plugins/s3keyindex/S3ObjectIndex.java

+    // For a string like "data-prepper-${yyyy-MM-dd}", "${yyyy-MM-dd}" is matched.
+    private static final String TIME_PATTERN_REGULAR_EXPRESSION = "\\$\\{.*?\\}";
+
+    // For matching a string enclosed by "%{" and "}".


The comment says "%{" but the code has "${". If we want to be consistent with Opensearch Sink, "%{" is more accurate. Not sure what was decided but make sure at the minimum comment and the code are consistent. I think it would be better to "%{" just like OpenSearch Sink for consistency across different sink configutations. But if there is a good reason to use "${", I am OK.

Modified the comment as "%{".

kkondaka · 2023-03-27T17:32:00Z

...lugins/common/src/main/java/org/opensearch/dataprepper/plugins/s3keyindex/S3ObjectIndex.java

+     * Validate the index with the regular expression pattern. Throws exception if
+     * validation fails
+     */
+    public static DateTimeFormatter getDatePatternFormatter(final String indexAlias) {


I have also previously commented that it might be good idea to move these to a common api directory so that same code can be used by OpenSearchSink code. Changes to OpenSearchSink will not be your responsibility.

As per David's suggestion, we have moved to a common package inside data-prepper-plugins instead of the common api directory.

kkondaka · 2023-03-27T17:35:32Z

data-prepper-plugins/s3-sink/README.md

+
+- `max_connection_retries` (Optional) : An integer value indicates the maximum number of times that a single request should be retried in-order to connect s3 client.
+
+- `max_upload_retries` (Optional) : An integer value indicates the maximum number of times that single request should be retired in-order to ingest data to amazon s3.


If there is a default value it should be specified here.

Added the default values.

kkondaka · 2023-03-27T17:35:45Z

data-prepper-plugins/s3-sink/README.md

+
+- `aws_sts_header_overrides`  (Optional) : An optional map of header overrides to make when assuming the IAM role for the sink plugin.
+
+- `max_connection_retries` (Optional) : An integer value indicates the maximum number of times that a single request should be retried in-order to connect s3 client.


If there is a default value, it should be specified here.

Added the default values.

kkondaka · 2023-03-27T17:36:23Z

data-prepper-plugins/s3-sink/README.md

+        temporary_storage: local_file
+```
+
+## Configuration


For all the optional fields, if there is a default value, it should be mentioned.

Added the default values.

kkondaka · 2023-03-27T17:49:11Z

...ain/java/org/opensearch/dataprepper/plugins/sink/configuration/AwsAuthenticationOptions.java

@@ -0,0 +1,74 @@
+/*


The directory path for this file has dataprepper/plugins/sink/configuration, Is this code not valid for source S3 configurations? Except possibly for "read" vs "write" permissions, I am assuming most of this code would be common for reading from S3 or writing S3. Correct me if that's not the case.

Yes, this code is common for s3-source and s3-sink. Please let us know whether we can move to a common package under data-prepper-plugins.

I'm not sure we need to move this presently. I am working on an alternative solution to this duplication.

kkondaka · 2023-03-27T17:50:04Z

...-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ObjectOptions.java

+import jakarta.validation.constraints.NotNull;
+
+/**
+ * An implementation class of Threshold configuration Options


Is this comment accurate? It looks like this file has file patten configuration.

Modified the comment based on the respective functionality.

kkondaka · 2023-03-27T17:51:56Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ThresholdOptions.java

+    private int eventCount;
+
+    @JsonProperty("maximum_size")
+    private String maximumSize = DEFAULT_BYTE_CAPACITY;


Does this have min/max values to be checked against?

There is no annotation to check the byte-size min and max size. As per David's suggestion, we have used the existing ByteCount, and validations are taken care of in the ByteCount.parse method.

kkondaka · 2023-03-27T17:53:42Z

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

+    }
+
+    @Test
+    void test_s3_Sink_plugin_isReady_ture() {


nit. typo. test_s3_Sink_plugin_isReady_ture -> test_s3_Sink_plugin_isReady_true

Rectified as test_s3_sink_plugin_isReady_true

kkondaka · 2023-03-27T17:57:25Z

...src/test/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferAccumulatorTest.java

+        S3SinkWorker worker = new S3SinkWorker(s3Client, s3SinkConfig, codec, codecFileExtension);
+        BufferAccumulator bufferAccumulator = worker.bufferAccumulator(generateEventQueue());
+        bufferAccumulator.doAccumulate();
+        assertNotNull(bufferAccumulator);


you should verify the contents of the buffer accumulator.

doAccumulate() does not have any return type rather than calling localfileProcessor() and inMemoryProcessor().

dlvenable

Thank you for this contribution! I've made a few suggestions to the configurations and to help with some class design.

I hope to take a further look at the S3 sink logic later.

dlvenable · 2023-03-28T14:24:38Z

...ain/java/org/opensearch/dataprepper/plugins/sink/configuration/AwsAuthenticationOptions.java

@@ -0,0 +1,74 @@
+/*


I'm not sure we need to move this presently. I am working on an alternative solution to this duplication.

dlvenable · 2023-03-28T14:26:13Z

...-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ObjectOptions.java

+ * An implementation class of file pattern configuration Options
+ */
+public class ObjectOptions {
+    private static final String DEFAULT_KEY_PATTERN = "logs-${yyyy-MM-dd hh:mm:ss}";


Please use a default value without spaces. So add a T instead of a space. This is inline with the ISO-8601 standard.

Also, I think we might not want logs in the name. Maybe "events" instead.

events-${yyyy-MM-ddThh:mm:ss}

Updaded name_pattern as per ISO-8601 standard as below.
name_pattern: event-%{yyyy-MM-dd'Thh-mm-ss'}

dlvenable · 2023-03-28T14:27:00Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ThresholdOptions.java

+    private String maximumSize = DEFAULT_BYTE_CAPACITY;
+
+    @JsonProperty("event_collection_duration")
+    @DurationMin(seconds = 1)


Nice use of annotations!

dlvenable · 2023-03-28T14:29:48Z

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

+    private PluginModel codec;
+
+    @JsonProperty("temporary_storage")
+    private String temporaryStorage = DEFAULT_STORAGE;


This should be an enum. See the CompressionOption enum for an example.

https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/configuration/CompressionOption.java

And here is an example of it being used:

data-prepper/data-prepper-plugins/s3-source/src/main/java/org/opensearch/dataprepper/plugins/source/S3SourceConfig.java

Lines 29 to 30 in 7bd967a

@JsonProperty("compression")

private CompressionOption compression = CompressionOption.NONE;

dlvenable · 2023-03-28T14:30:51Z

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

+    @NotNull
+    private PluginModel codec;
+
+    @JsonProperty("temporary_storage")


I wonder if there is a better name here. Perhaps buffer_type? Or event_buffer?

Updated the code with buffer_type

dlvenable · 2023-03-28T14:32:21Z

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

+        assertNull(s3SinkConfig.getCodec());
+    }
+
+    private void reflectivelySetField(final S3SinkConfig s3SinkConfig, final String fieldName, final Object value)


Data Prepper has a utility for this. Please use this instead.

https://github.com/opensearch-project/data-prepper/blob/main/data-prepper-test-common/src/main/java/org/opensearch/dataprepper/test/helper/ReflectivelySetField.java

As per updated code, the above ReflectivelySetField is not required.

dlvenable · 2023-03-28T14:35:37Z

...ugins/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/codec/JsonCodecTest.java

+        eventData.put("key2", "value");
+        final JacksonEvent event = JacksonLog.builder().withData(eventData).withEventType("LOG").build();
+        String output = createObjectUnderTest().parse(event);
+        assertNotNull(output);


Please also verify the actual object.

I think the best way in this scenario is to have an ObjectMapper in the test. Then you can deserialize the output and verify it has the same data.

Map deserializedData = objectMapper.readValue(output, Map.class); assertThat(deserializedData.get("key1"), equalTo(value1)); assertThat(deserializedData..get("key2"), equalTo(value2)); // maybe some more assertions?

Resolved and added more assertions.

dlvenable · 2023-03-28T14:36:20Z

...ugins/s3-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/codec/JsonCodecTest.java

+    void parse_with_events_outputstream_json_codec(final int numberOfObjects) throws IOException {
+
+        final Map<String, String> eventData = new HashMap<>();
+        eventData.put("key1", "value");


Please create random and distinct values for both of these keys.

String value1 = UUID.randomUUID().toString(); eventData.put("key1", value1); String value2 = UUID.randomUUID().toString(); eventData.put("key2", value2);

Updated the code as per the above commet.

dlvenable · 2023-03-28T14:39:17Z

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

+    private String bucketName;
+
+    @JsonProperty("key_path_prefix")
+    private String keyPathPrefix;


Let's move this property into the ObjectOptions configuration.

Moved the path_prefix property to the ObjectKeyOptions under BucketOptions as per latest YAML.

dlvenable · 2023-03-28T14:41:29Z

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

+    /**
+     * Store event data into local file
+     */
+    private void localFileProcessor() {


Using the strategy pattern to decide how to handle local file versus in-memory can help make the code cleaner. You can have independent tests for each strategy.

kkondaka · 2023-04-04T23:42:42Z

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

+
+    private static final Logger LOG = LoggerFactory.getLogger(S3SinkService.class);
+    private final S3SinkConfig s3SinkConfig;
+    private static final int EVENT_QUEUE_SIZE = 100000;


I am not sure about this. @dlvenable could you clarify?

kkondaka · 2023-04-04T23:54:25Z

...ns/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferType.java

+
+            if (retryCount == 0) {
+                LOG.warn("Maximum retry count reached, Unable to store {} into amazon S3 bucket", s3ObjectFileName);
+                break;


You should do return false here otherwise the "unable to upload" message would be printed twice

kkondaka · 2023-04-05T00:02:16Z

...-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/BucketOptions.java

+     * S3 {@link ObjectKeyOptions} configuration Options.
+     */
+    public ObjectKeyOptions getObjectKeyOptions() {
+        if (objectKeyOptions == null)


This looks wrong. I think you meant

if (objectKeyOptions == null) { objectKeyOptions = new ObjectKeyOptions(); } return objectKeyOptions;

kkondaka · 2023-04-05T00:18:09Z

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

+        s3SinkService = mock(S3SinkService.class);
+        Collection<Record<Event>> records = generateRandomStringEventRecord();
+        s3SinkService.accumulateBufferEvents(worker);
+        verify(s3SinkService, atLeastOnce()).accumulateBufferEvents(worker);


I am not sure if this is right check. accumualteBufferEvents is called above and verify that it is called at least once here seems unnecessary. You have to verify what logic took place inside accumualteBufferEvents and if that happened correctly. Please validate the functionality of the functions involved.

Verifid the logic of accumualteBufferEvents.

kkondaka · 2023-04-05T00:19:09Z

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

+        s3Sink = new S3Sink(pluginSetting, s3SinkConfig, pluginFactory);
+        s3Sink.doInitialize();
+        s3Sink.doOutput(generateRandomStringEventRecord());
+        assertNotNull(s3Sink);


Please validate if doOutput() worked as expected.

Yes, this is very important.

You should be able to perform the following:

InOrder inOrder = inOrder(s3SinkService); inOrder.verify(s3SinkService).processRecords(records); inOrder.verify(s3SinkService).accumulateBufferEvents(sinkWorker);

Updated test-cases using inOrder and verified.

Have you uploaded new diffs?

dlvenable · 2023-04-04T16:50:54Z

...lugins/common/src/main/java/org/opensearch/dataprepper/plugins/s3keyindex/S3ObjectIndex.java

+/**
+ * Class responsible for creation of s3 key pattern based on date time stamp
+ */
+public class S3ObjectIndex {


It seems this code was moved along with items that are probably beyond the scope of what it should include.

The OpenSearch sink deals with indexes. The S3 sink will deal with objects. Indexes and objects have different constraints and there is no relationship between the two.

The common code is the ability to parse the date-time information. This is the part which can be made into a common utility.

Here is some pseudo-code for how this might look.

OpenSearch:

String unparsedIndexName = getIndexName(); // This might look like logs-${mm.dd} String parsedIndexName = DestinationExpressionEvaluator.parseExpression(unparsedIndexName); // Now this looks like logs-04.04 validateIndexName(parsedIndexName); createIndex(parsedIndexName);

S3 sink

String unparsedObjectName = getObject(); // This might look like logs-${mm}-${dd}.json String parsedObjectName = DestinationExpressionEvaluator.parseExpression(unparsedObjectName); // Now this looks like logs-04-04.json validateS3ObjectName(parsedObjectName); createIndex(parsedObjectName);

Modified the functions name based on the date time stamp of the object.

dlvenable · 2023-04-04T16:54:21Z

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

+        boolean isFileUploadedToS3 = Boolean.FALSE;
+        String s3ObjectPath = buildingPathPrefix(s3SinkConfig);
+        String s3ObjectFileName = S3ObjectIndex.getIndexAliasWithDate(
+                s3SinkConfig.getBucketOptions().getObjectKeyOptions().getNamePattern()) + "." + codecFileExtension;


I think the name should be WYSIWYG.

So the pipeline author needs to specify the .${extension}. You can then expand that variable - ${extension} into the correct value.

Modified the code as below:
S3Sink object name extension will be based on the user provided extension in "name_pattern".
The default file extension is json if user does not provide any extension.

dlvenable · 2023-04-05T02:01:47Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ObjectKeyOptions.java

+ * An implementation class of path prefix and file pattern configuration Options
+ */
+public class ObjectKeyOptions {
+    private static final String DEFAULT_FILE_PATTERN = "events-%{yyyy-MM-dd'T'hh-mm-ss}";


Should this be events-%{yyyy-MM-dd'T'hh-mm-ss}-%{epochSeconds}-%{random}.%{extension}.

Either that, or we should rename the property to be clearer that this is only part of the name.

Modifid the property name from DEFAULT_FILE_PATTERN to DEFAULT_NAME_PATTERN.

dlvenable · 2023-04-05T02:05:25Z

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

+        s3Sink = new S3Sink(pluginSetting, s3SinkConfig, pluginFactory);
+        s3Sink.doInitialize();
+        s3Sink.doOutput(generateRandomStringEventRecord());
+        assertNotNull(s3Sink);


Yes, this is very important.

You should be able to perform the following:

InOrder inOrder = inOrder(s3SinkService); inOrder.verify(s3SinkService).processRecords(records); inOrder.verify(s3SinkService).accumulateBufferEvents(sinkWorker);

dlvenable · 2023-04-05T02:07:19Z

...ns/s3-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/accumulator/BufferType.java

+/**
+ * Interface for building buffer types.
+ */
+public interface BufferType {


This class should be unit tested.

Added Junit test-cases for BufferType interface.

dlvenable · 2023-04-05T02:08:08Z

...nk/src/test/java/org/opensearch/dataprepper/plugins/sink/accumulator/InMemoryBufferTest.java

+    @Test
+    void test_in_memoryAccumulate_with_s3Upoload_success() throws InterruptedException {
+        S3SinkService s3SinkService = new S3SinkService(s3SinkConfig);
+        S3Client s3Client = s3SinkService.createS3Client();


You should use a mock object here. This will let you verify the interactions with the S3Client.

Mocked the s3SinkService and verified the interactions with the S3Client.

dlvenable · 2023-04-05T02:08:27Z

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

+    @Test
+    void test_local_file_accumulate_with_s3Upoload_success() throws InterruptedException {
+        S3SinkService s3SinkService = new S3SinkService(s3SinkConfig);
+        S3Client s3Client = s3SinkService.createS3Client();


Again, this should be a mocked object. You can then verify the interactions.

Mocked the s3SinkService and verified the interactions.

dlvenable · 2023-04-05T02:10:29Z

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

+     * @throws InterruptedException
+     */
+    public boolean inMemoryAccumulate(final NavigableSet<String> bufferedEventSet) throws InterruptedException {
+        String s3ObjectPath = buildingPathPrefix(s3SinkConfig);


Some of this code is duplicated with the LocalFileBuffer. This indicates there should be some refactoring.

Additionally, the decision for building the path is completely independent of the buffering logic. It should probably be moved into its own class which can get the whole object path. Such a small, focused class would be quite easy to unit test as well.

Added new class ObjectKey for the object path constuction functionality.

kkondaka · 2023-04-12T04:07:11Z

...ns/common/src/test/java/org/opensearch/dataprepper/plugins/s3keyindex/S3ObjectIndexTest.java

+
+        String expectedIndex = S3ObjectIndex.getObjectNameWithDateTimeId("events-%{yyyy-MM-dd}");
+        String actualIndex = S3ObjectIndex.getObjectNameWithDateTimeId("events-%{yyyy-MM-dd}");
+        assertNotEquals(actualIndex.toString(), expectedIndex.toString());


Why is this assertNotEquals()?

As per requirement, EPOCH SECONDS and UniqueId are appended for the name_pattern.
assertNotEquals are added since both the responses are not equal.

You mean it would be expected String + EPOCH seconds + unique id? In that case, can we compare that expected String is in the actualIndex.toString()?

kkondaka · 2023-04-12T04:12:47Z

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

+    void test_call_createS3Client() {
+        S3SinkService s3SinkService = new S3SinkService(s3SinkConfig);
+        S3Client s3Client = s3SinkService.createS3Client();
+        assertNotNull(s3Client);


You need to assert() more here to make sures3SinkService is created properly and s3Client is properly created with the expected config.

These changes will be addressed as incremental changes, and they will be available in a separate PR.

kkondaka · 2023-04-12T04:14:45Z

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

+        s3SinkService = mock(S3SinkService.class);
+        Collection<Record<Event>> records = generateRandomStringEventRecord();
+        s3SinkService.processRecords(records);
+        verify(s3SinkService, atLeastOnce()).processRecords(records);


You need to assert that processRecords worked as expected. In this case may be you can check to see that event queue has expected number of records in it. You can add an protected API to S3SinkService to return the size of eventQueue to do this.

These changes will be addressed as incremental changes, and they will be available in a separate PR.

kkondaka · 2023-04-12T04:20:08Z

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

+        S3SinkWorker worker = mock(S3SinkWorker.class);
+        service.processRecords(records);
+        service.accumulateBufferEvents(worker);
+        verify(worker, only()).bufferAccumulator(any(BlockingQueue.class));


Same here. You can check to the eventQueue is empty after doing accumulateBufferEvents

These changes will be addressed as incremental changes, and they will be available in a separate PR.

kkondaka · 2023-04-12T04:21:43Z

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

+        s3Sink = new S3Sink(pluginSetting, s3SinkConfig, pluginFactory);
+        s3Sink.doInitialize();
+        s3Sink.doOutput(new ArrayList<>());
+        assertNotNull(s3Sink);


Please do not do this here. s3Sink is created in 109 and even if it were null, it would have crashed in line 110. You should verify the metrics that s3Sink has actually did output of 0 records here.

kkondaka · 2023-04-12T04:21:58Z

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

+        s3Sink = new S3Sink(pluginSetting, s3SinkConfig, pluginFactory);
+        s3Sink.doInitialize();
+        s3Sink.doOutput(generateRandomStringEventRecord());
+        assertNotNull(s3Sink);


kkondaka · 2023-04-12T04:22:43Z

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

+        s3Sink = new S3Sink(pluginSetting, s3SinkConfig, pluginFactory);
+        s3Sink.doInitialize();
+        s3Sink.doOutput(generateRandomStringEventRecord());
+        assertNotNull(s3Sink);


Have you uploaded new diffs?

dlvenable · 2023-04-12T15:59:10Z

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

+
+    private static final Logger LOG = LoggerFactory.getLogger(S3SinkService.class);
+    private final S3SinkConfig s3SinkConfig;
+    private static final int EVENT_QUEUE_SIZE = 100000;


It appears that you are putting a hard-coded restriction on the number of events to 100,000 here. This will put an upper limit on the ArrayBlockingQueue.

I believe this will conflict with some user configuration values.

See ThresholdOptions line 24, where we allow a maximum event count of 10,000,000. So if the user sets this value, won't the ArrayBlockingQueue fill up?

Also, what the user is using bytes instead and it exceeds this value of 100,000?

dlvenable · 2023-04-12T16:08:51Z

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

+    private static final Logger LOG = LoggerFactory.getLogger(S3SinkService.class);
+    private final S3SinkConfig s3SinkConfig;
+    private static final int EVENT_QUEUE_SIZE = 100000;
+    private BlockingQueue<Event> eventQueue;


Why are use using a BlockingQueue here?

These changes will be addressed as incremental changes, and they will be available in a separate PR.

dlvenable · 2023-04-12T16:14:34Z

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

+                    "Snapshot info : Byte_capacity = {} Bytes, Event_count = {} Records & Event_collection_duration = {} Sec",
+                    byteCount, eventCount, eventCollectionDuration);
+
+            if (s3SinkConfig.getBufferType().equals(BufferTypeOptions.LOCALFILE)) {


Every time the Data Prepper pipeline sends events to the sink, it calls doOutput on the S3Sink class. This in turn, adds events to a BlockingQueue and then calls this method.

This method is then taking events from the BlockingQueue, adds them to a new NavigableSet. After that it, it creates the buffer (local-file or in-memory). Then it calls the "accumulate" methods.

These methods then immediately send to S3. So there is no buffer accumulation. The idea behind the accumulation is to keep data from multiple calls to doOutput. Otherwise, the user may get very small S3 objects.

These changes will be addressed as incremental changes, and they will be available in a separate PR.

dlvenable · 2023-04-12T16:16:25Z

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

+
+        StringBuilder eventBuilder = new StringBuilder();
+        for (String event : bufferedEventSet) {
+            eventBuilder.append(event);


Please provide event.toJsonString() here.

We are already applying toJsonString() jsonCodec parse method before control comes to this file.

dlvenable · 2023-04-12T16:16:41Z

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

+        File file = new File(s3ObjectFileName);
+        try (BufferedWriter eventWriter = new BufferedWriter(new FileWriter(s3ObjectFileName))) {
+            for (String event : bufferedEventSet) {
+                eventWriter.write(event);


Please provide event.toJsonString() here.

We are already applying toJsonString() jsonCodec parse method before control comes to this file.

dlvenable · 2023-04-12T16:25:56Z

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

+    public void bufferAccumulator(BlockingQueue<Event> eventQueue) {
+        boolean isFileUploadedToS3 = Boolean.FALSE;
+        DB eventDb = null;
+        NavigableSet<String> bufferedEventSet = null;


Why is this a NavigableSet? I'm not even sure you need a collection.

I suggested elsewhere that you pass in a List<Event> to this method. Just provide that List to the accumulators. Those accumulators can then iterate over the lists and add each Event into their respective buffers.

These changes will be addressed as incremental changes, and they will be available in a separate PR.

dlvenable · 2023-04-12T16:26:14Z

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

+        try {
+            StopWatch watch = new StopWatch();
+            watch.start();
+            eventDb = DBMaker.memoryDB().make();


Why do we have an eventDb here?

These changes will be addressed as incremental changes, and they will be available in a separate PR.

dlvenable · 2023-04-12T16:27:13Z

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

+ */
+public enum BufferTypeOptions {
+
+    INMEMORY("in_memory", new InMemoryBuffer()),


These should be IN_MEMORY and LOCAL_FILE. You can use your IDE to rename these easily.

Followed the same notation (lower-case) similar to s3-source CompressionOption.

dlvenable · 2023-04-12T16:49:56Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ObjectKeyOptions.java

+ * An implementation class of path prefix and file pattern configuration Options
+ */
+public class ObjectKeyOptions {
+    private static final String DEFAULT_OBJECT_NAME_PATTERN = "events-%{yyyy-MM-dd'T'hh-mm-ss}";


This value should be:

events-%{yyyy-MM-dd'T'hh-mm-ss}-${epochSeconds}-${uniqueId}.${extension}

I think there may be some broad issues here about format strings as well.

So, my suggestion is that what we do for now is remove the name_pattern option from customers. We can consider making this configurable later. For this PR, please always create the pattern as the following.

events-%{yyyy-MM-dd'T'hh-mm-ss}-${epochSeconds}-${uniqueId}.${extension}

The epochSeconds will the the same time as the string in yyyy-MM-dd'T'hh-mm-ss, but represented as seconds since the 1970 epoch. The unique Id is generated once per Data Prepper instance. And the extension comes from the codec.

dlvenable · 2023-04-12T16:50:52Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ObjectKeyOptions.java

+    private String pathPrefix;
+
+    @JsonProperty("name_pattern")
+    private String namePattern = DEFAULT_OBJECT_NAME_PATTERN;


As noted above, let's not make this configurable for now.

dlvenable · 2023-04-18T20:26:15Z

...nk/src/main/java/org/opensearch/dataprepper/plugins/sink/configuration/ObjectKeyOptions.java

+    @JsonProperty("path_prefix")
+    private String pathPrefix;
+
+    @JsonProperty("name_pattern")


Let's not expose this as a feature for pipeline authors yet.

Remove these two lines.

@JsonProperty("name_pattern") private String namePattern = DEFAULT_OBJECT_NAME_PATTERN;

Change getNamePattern to this:

public String getNamePattern() { return DEFAULT_OBJECT_NAME_PATTERN; }

These changes should remove this as a user configuration. And you won't have to change the code that uses this.

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

deepaksahu562 requested a review from a team as a code owner March 3, 2023 16:29

kkondaka requested changes Mar 7, 2023

View reviewed changes

kkondaka requested changes Mar 27, 2023

View reviewed changes

dlvenable requested changes Mar 28, 2023

View reviewed changes

deepaksahu562 requested a review from dlvenable April 3, 2023 15:56

kkondaka requested changes Apr 5, 2023

View reviewed changes

dlvenable requested changes Apr 5, 2023

View reviewed changes

dlvenable self-assigned this Apr 12, 2023

kkondaka requested changes Apr 12, 2023

View reviewed changes

dlvenable requested changes Apr 12, 2023

View reviewed changes

deepaksahu562 force-pushed the master branch from 8e79289 to b8c0b81 Compare April 17, 2023 13:00

dlvenable requested changes Apr 18, 2023

View reviewed changes

deepaksahu562 closed this Apr 18, 2023

deepaksahu562 force-pushed the master branch from 181d123 to 232f0de Compare April 18, 2023 22:21

deepaksahu562 reopened this Apr 18, 2023

Adding the S3 sink plugin. Contributes to opensearch-project#1048

be5a4f5

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

deepaksahu562 force-pushed the master branch from 987b5b5 to be5a4f5 Compare April 19, 2023 14:35

Github-issue#1048

da9d3fe

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

deepaksahu562 closed this May 15, 2023

		@@ -0,0 +1,9 @@
		Attached simple-sample-pipeline: opensearch-data-prepper\data-prepper\data-prepper-plugins\s3-sink\src\main\resources\pipelines.yaml


		- `max_connection_retries` (Optional) : An integer value indicates the maximum number of times that a single request should be retried in-order to connect s3 client.

		- `max_upload_retries` (Optional) : An integer value indicates the maximum number of times that single request should be retired in-order to ingest data to amazon s3.


		- `aws_sts_header_overrides` (Optional) : An optional map of header overrides to make when assuming the IAM role for the sink plugin.

		- `max_connection_retries` (Optional) : An integer value indicates the maximum number of times that a single request should be retried in-order to connect s3 client.

	@JsonProperty("compression")
	private CompressionOption compression = CompressionOption.NONE;

Created "s3-sink" plugin. github-issue#1048 #2362

Created "s3-sink" plugin. github-issue#1048 #2362

Conversation

deepaksahu562 commented Mar 3, 2023 • 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

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

dlvenable 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

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

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 Mar 3, 2023 •

edited

Loading