Add support for lambda sink #4292

srikanthjg · 2024-03-16T00:26:21Z

Description

Add support for lambda sink

Issues Resolved

Resolves #4170

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.

chenqi0805 · 2024-03-19T17:57:00Z

...integrationTest/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkServiceIT.java

+    private final ObjectMapper objectMapper = new ObjectMapper(new YAMLFactory().enable(YAMLGenerator.Feature.USE_PLATFORM_LINE_BREAKS));
+    private String stsRoleArn;
+
+    final String LAMBDA_SINK_CONFIG_YAML =


Can we make this runtime input instead of hardcoded value?

chenqi0805 · 2024-03-25T22:46:14Z

...integrationTest/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkServiceIT.java

+import static org.mockito.Mockito.verify;
+
+@ExtendWith(MockitoExtension.class)
+class LambdaSinkServiceIT {


How to run this test should be included in README but can come in a separate PR.

chenqi0805 · 2024-03-25T22:53:45Z

...mbda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkConfig.java

+    private int maxConnectionRetries = DEFAULT_CONNECTION_RETRIES;
+
+    @JsonProperty("sync")
+    private Boolean sync = DEFAULT_INVOCATION;


QUES: is there reason we want async invocation as default?

async provides a better way to communicate to external services

chenqi0805 · 2024-03-25T23:02:35Z

.../src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/accumlator/InMemoryBuffer.java

+import java.util.concurrent.TimeUnit;
+
+/**
+ * A buffer can hold in memory data and flushing it to S3.


flush to Lambda?

chenqi0805 · 2024-03-25T23:08:28Z

.../src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/accumlator/InMemoryBuffer.java

+     * Upload accumulated data to s3 bucket.
+     */
+    @Override
+    public void flushToLambda() {


e2e acknowledgment seems missing in this PR.

e2e ack is handled. refer releaseEventHandles in the pr.

chenqi0805 · 2024-03-25T23:08:38Z

.../src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/accumlator/InMemoryBuffer.java

+    }
+
+    /**
+     * Upload accumulated data to s3 bucket.


s3 -> lambda

kkondaka · 2024-03-31T16:03:38Z

...bda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkService.java

+                    //reset buffer after flush
+                    currentBuffer = bufferFactory.getBuffer(lambdaClient,functionName,invocationType);
+                } catch (final IOException e) {
+                    LOG.error("Exception while completing codec", e);


You should do releaseEventHandles(false) in case of exceptionl

kkondaka · 2024-03-31T16:05:25Z

...bda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkService.java

+    private int maxRetries = 0;
+    private final Counter numberOfRecordsSuccessCounter;
+    private final Counter numberOfRecordsFailedCounter;
+    private final String SYNC_INVOCATION_TYPE = "RequestResponse";


I think we discussed that sync doesn't make sense in case of lambda as sink. So, invocation type is not needed.

kkondaka · 2024-03-31T16:06:32Z

...bda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkService.java

+                        dlqPushHandler.perform(pluginSetting,new LambdaSinkFailedDlqData(payload,errorMsgObj.get(),0));
+                    }
+                    //release even if failed
+                    releaseEventHandles(true);


Should be released with false if failed.

I think it is considered successful if it is uploaded to dlq?

But what if DLQ is not configured?

kkondaka · 2024-03-31T16:07:15Z

...bda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/accumlator/Buffer.java

+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+


Maybe we should move this to common directory so that other sinks can use this future?

can refactor be part of another PR?

kkondaka · 2024-03-31T16:07:54Z

...ambda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/codec/JsonCodec.java

+import java.util.Map;
+import java.util.Objects;
+
+public class JsonCodec implements OutputCodec {


As discussed, let's rename it to be more specific to lambda

kkondaka · 2024-05-10T00:49:25Z

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

+        aws:
+            region: "us-east-1"
+            sts_role_arn: "<arn>"
+        function_name: "uploadToS3Lambda"


Did we not discuss about passing arguments to lambda function? Do you not see any need for that option?

kkondaka · 2024-05-10T01:01:54Z

...bda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkService.java

+        reentrantLock.lock();
+        try {
+            for (Record<Event> record : records) {
+                final Event event = record.getData();


We probably need to support sending event metadata as well?

yes we could, we can add this incrementally?

kkondaka · 2024-05-10T01:03:09Z

...bda-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/LambdaSinkService.java

+                        dlqPushHandler.perform(pluginSetting,new LambdaSinkFailedDlqData(payload,errorMsgObj.get(),0));
+                    }
+                    //release even if failed
+                    releaseEventHandles(true);


But what if DLQ is not configured?

kkondaka · 2024-05-10T01:04:19Z

.../src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/accumlator/InMemoryBuffer.java

+ * A buffer can hold in memory data and flushing it.
+ */
+public class InMemoryBuffer implements Buffer {
+


Is it not possible to use existing BufferAccumulator ?

Interface definitions are different. Will not be able to use the existing one

kkondaka · 2024-05-10T01:05:53Z

...sink/src/main/java/org/opensearch/dataprepper/plugins/sink/lambda/codec/LambdaJsonCodec.java

+import java.util.Objects;
+
+public class LambdaJsonCodec implements OutputCodec {
+    private final ObjectMapper objectMapper = new ObjectMapper();


How is this different from Json Output Codec we already have? Can we re-use it somehow?

i tried to use existing one but it is not possible without making a whole bunch of changes specific to lambda, so i chose to have a new one instead.

srikanthjg · 2024-06-03T20:24:24Z

I think there was an issue with merge, will resubmit the PR. 2 commits(5031b38 and 2f6b832) that are not part of my changes sneeked in.

Signed-off-by: srigovs <[email protected]>

Signed-off-by: Srikanth Govindarajan <[email protected]>

dls314 · 2024-11-07T16:51:37Z

Hi @srikanthjg, I know that the chance of catching you on this merged PR are slim, but I wanted to ask if you had a pointer to any documentation of the lambda event type and return type that's used for a lambda sink.

https://opensearch.org/docs/latest/data-prepper/pipelines/configuration/processors/aws-lambda/#aws-lambda-processor-configuration doesn't really convey the shape of the lambda event or response unless I'm missing something.

If there's something that can be pointed at, I'd be happy to ask for it to be included in https://github.com/DefinitelyTyped/DefinitelyTyped/tree/master/types/aws-lambda/trigger to help those of us using TS for these.

Thanks much for any help, Dave

srikanthjg requested review from chenqi0805, engechas, graytaylor0, dinujoh, kkondaka, asifsmohammed, dlvenable and oeyh as code owners March 16, 2024 00:26

chenqi0805 reviewed Mar 25, 2024

View reviewed changes

kkondaka reviewed Mar 31, 2024

View reviewed changes

srikanthjg force-pushed the lambda-sink branch from 97c5aff to f0fcd11 Compare May 9, 2024 01:51

srikanthjg requested a review from KarstenSchnitter as a code owner May 9, 2024 01:51

srikanthjg force-pushed the lambda-sink branch 3 times, most recently from c63dd00 to dddb205 Compare May 10, 2024 04:54

kkondaka reviewed May 10, 2024

View reviewed changes

kkondaka previously approved these changes Jun 3, 2024

View reviewed changes

srikanthjg force-pushed the lambda-sink branch from fc15b70 to 1391d66 Compare June 3, 2024 21:05

srikanthjg and others added 2 commits June 4, 2024 22:08

Add support for lambda sink

8bc572a

Signed-off-by: srigovs <[email protected]>

Address event handle comment

48a17fb

Signed-off-by: Srikanth Govindarajan <[email protected]>

srikanthjg dismissed kkondaka’s stale review via 4e4aeb8 June 5, 2024 06:41

srikanthjg force-pushed the lambda-sink branch 2 times, most recently from 4e4aeb8 to 48a17fb Compare June 5, 2024 17:36

kkondaka approved these changes Jun 5, 2024

View reviewed changes

chenqi0805 approved these changes Jun 5, 2024

View reviewed changes

kkondaka merged commit 4cf86e9 into opensearch-project:main Jun 6, 2024
68 of 74 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for lambda sink #4292

Add support for lambda sink #4292

srikanthjg commented Mar 16, 2024 •

edited

Loading

chenqi0805 Mar 19, 2024

chenqi0805 Mar 25, 2024

chenqi0805 Mar 25, 2024

srikanthjg Apr 30, 2024

chenqi0805 Mar 25, 2024

chenqi0805 Mar 25, 2024

srikanthjg Apr 30, 2024

chenqi0805 Mar 25, 2024

kkondaka Mar 31, 2024

kkondaka Mar 31, 2024

kkondaka Mar 31, 2024

srikanthjg May 1, 2024

kkondaka May 10, 2024

kkondaka Mar 31, 2024

srikanthjg May 1, 2024

kkondaka Mar 31, 2024

kkondaka May 10, 2024

kkondaka May 10, 2024

srikanthjg May 10, 2024

kkondaka May 10, 2024

kkondaka May 10, 2024

srikanthjg May 10, 2024

kkondaka May 10, 2024

srikanthjg May 10, 2024

srikanthjg commented Jun 3, 2024

dls314 commented Nov 7, 2024

Add support for lambda sink #4292

Add support for lambda sink #4292

Conversation

srikanthjg commented Mar 16, 2024 • 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

srikanthjg commented Jun 3, 2024

dls314 commented Nov 7, 2024

srikanthjg commented Mar 16, 2024 •

edited

Loading