Address Scale Items for lambda plugin by srikanthjg · Pull Request #5032 · opensearch-project/data-prepper

srikanthjg · 2024-10-08T06:10:19Z

Description

Following are the changes made:

Address Scale issues:

Added support for lambda async client in lambda processor and sink
1.1 We will have a asynchronous call to lambda at a batch level, ie, we could send multiple batches to lambda at the
same time. We will wait for the futures only after the entire set of records that was received by the processors
are done.
1.2 Handle metrics and buffer per batch based on futures processing.
Make sdk timeout a user configurable parameter.
Add Codec for request and response from lambda. NOTE: Json codec as input and output is the current default. And lambda response codec always assumes json array as the response.
Removed payload_model and will no more support SINGLE_EVENT, will only support BATCH based calls by default

Acknowledgements:
5. Address Acknowledgements for processor and sink.
For Processors:
5.1. When a batch of N events that is configured by the user (N could be <=pipeline batch) ie, request to lambda contains N events in a batch , the lambda could return back N responses or M responses(N!=M). When N responses are sent as a json array, we resuse the original records and clear the old event data and populate it with the response from lambda, that way the acknowledgement set need not be changed.
5.2 When M responses(N!=M), we create new events and populate them to the original acknowledgement set. The older events are also retained in the ack set but will be released by core later.

Handling Failure:
6. Address failures at process the events, the events in the processor will be tagged and forwarded. This processor will NOT drop events on failure.
7. Lambda sink will send to DLQ on failure and will acknowledge as true. If a dlq is not setup, we will send a negative acknowledgement.

Refactor:
8. Refactor aws lambda plugin to have a class for common methods between processor and sink.

Issues Resolved

Resolves #5031

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.

graytaylor0 · 2024-10-08T19:52:43Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+        }
+        codec.writeEvent(event, currentBuffer.getOutputStream());
+        int count = currentBuffer.getEventCount() + 1;
+        LOG.info("CurrentBuffer event count: {}", count);


Will this get logged for every event? If so it should be debug

Thanks for noticing that, i will make the noisy logs debug level.

graytaylor0 · 2024-10-08T19:53:35Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+
+    public void flushToLambdaIfNeeded(List<Record<Event>> resultRecords, boolean forceFlush) {
+
+        LOG.info("currentBufferEventCount:{}, maxEvents:{}, maxBytes:{}, maxCollectionDuration:{}, isBatch:{}, forceFlush:{} ", currentBuffer.getEventCount(),maxEvents,maxBytes,maxCollectionDuration,isBatchEnabled, forceFlush);


This seems like a helpful log, but if it may be noisy as INFO if this is called for every Event

graytaylor0 · 2024-10-08T19:53:52Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+                // Handle future
+                CompletableFuture<Void> processingFuture = future.thenAccept(response -> {
+                    handleLambdaResponse(response);
+                    LOG.info("Successfully flushed {} events", eventCount);


Same here on this being a noisy log

graytaylor0 · 2024-10-08T19:54:47Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+
+    public void resetBuffer() {
+        try {
+            LOG.info("Resetting buffer");


Is this helpful to have as info or can it be debug?

graytaylor0 · 2024-10-08T19:57:03Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

            isBatchEnabled = true;
            LOG.info("maxEvents:" + maxEvents + " maxbytes:" + maxBytes + " maxDuration:" + maxCollectionDuration);
        } else if(payloadModel.equals(SINGLE_EVENT)) {
+            LOG.info("Single events");


This should be debug

graytaylor0 · 2024-10-08T19:58:56Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

+                codec,
+                codecContext,
+                isBatchEnabled,
+                whenCondition,
+                maxEvents,
+                maxBytes,
+                maxCollectionDuration,
+                isSink,
+                dlqPushHandler,
+                pluginSetting);


Can we put this all into a single config class to consolidate?

graytaylor0 · 2024-10-08T19:59:11Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

            return records;
        }

+        LOG.info("Received " + records.size() + "records to lambda Processor" );


This should also be debug

graytaylor0 · 2024-10-08T19:59:24Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

+            }
+        }
+
+        LOG.info("Force Flushing the remaining {} events in the buffer", lambdaCommonHandler.getCurrentBuffer().getEventCount());


Same here, should be debug

graytaylor0 · 2024-10-08T20:01:19Z

...src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessorConfig.java

    @JsonProperty("payload_model")
    private String payloadModel = BATCH_EVENT;

+    @JsonProperty("sdk_timeout")


You should add the @JsonPropertyDescription annotation to all of these config parameters.

Why do you call this sdk_timeout? This seems more related to the implementation than what is actually timing out. We should probably look at using socket_timeout or connect_timeout.

graytaylor0 · 2024-10-08T20:02:24Z

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

+                    lambdaCommonHandler.resetBuffer();
+                }
+            }
+            LOG.info("Force Flushing the remaining {} events in the buffer", lambdaCommonHandler.getCurrentBuffer().getEventCount());


kkondaka · 2024-10-08T22:22:39Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+
+                }).exceptionally(throwable -> {
+                    LOG.error(NOISY, "Exception occurred while invoking Lambda. Function: {} | Exception: ", functionName, throwable);
+                    handleFailure(throwable);


Need a metric to keep track of failures.

kkondaka · 2024-10-08T22:23:02Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+    public void handleLambdaResponse(InvokeResponse response) {
+        int statusCode = response.statusCode();
+        if (statusCode < 200 || statusCode >= 300) {
+            LOG.warn("Lambda invocation returned with non-success status code: {}", statusCode);


Add a metric here.

kkondaka · 2024-10-22T16:53:51Z

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

-        Thread.sleep(Duration.ofSeconds(10).toMillis());
-    }
-}
+///*


Why is this file commented out?

kkondaka · 2024-10-22T16:54:58Z

...mbda/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/LambdaCommonHandler.java

+    /*
+     * Release events per batch
+     */
+    public void releaseEventHandles(final boolean result, List<EventHandle> bufferedEventHandles) {


I think this should not be in common code. Processor doesn't need to explicitly release event handles.

dlvenable · 2024-10-23T22:03:17Z

...src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessorConfig.java

-    private static final int DEFAULT_CONNECTION_RETRIES = 3;
+public class LambdaProcessorConfig {
+    //Ensures 1:1 mapping of events input to lambda and response from lambda
+    public static final String STRICT = "strict";


Please use enums for these. This is important for correct support for validations and schemas.

Here is an example:

https://github.com/opensearch-project/data-prepper/blob/27337ea36ee7a8a4b3496b32a46c541abd9aa609/data-prepper-plugins/decompress-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/decompress/DecompressionType.java

dlvenable · 2024-10-23T22:03:54Z

...src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessorConfig.java

    @JsonProperty("payload_model")
    private String payloadModel = BATCH_EVENT;

+    @JsonProperty("sdk_timeout")


Why do you call this sdk_timeout? This seems more related to the implementation than what is actually timing out. We should probably look at using socket_timeout or connect_timeout.

...src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessorConfig.java

dlvenable · 2024-10-23T22:13:11Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

-            }
-            //Reset Buffer
+        LOG.debug("currentBufferPerBatchEventCount:{}, maxEvents:{}, maxBytes:{}, maxCollectionDuration:{}, forceFlush:{} ", currentBufferPerBatch.getEventCount(),maxEvents,maxBytes,maxCollectionDuration, forceFlush);
+        if (forceFlush || ThresholdCheck.checkThresholdExceed(currentBufferPerBatch, maxEvents, maxBytes, maxCollectionDuration)) {


We should be able to handle this in common with the sink with some slight refactoring.

flush method looks very similar for both sink and processor and i tried combining that in the earlier PR, but we need to handle release of events differently for sink. For that i introduced isSink flag. But Comment from Krishna was that common code should not have flag like that and i think agree with his point; and now reverted it back to both processor and sink having their own flush methods.

I agree that a flag is not ideal. But, you can have an injectable strategy. We can resolve this later though.

dlvenable · 2024-10-23T22:18:29Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

+            LOG.debug("Parsed Event Size:{}, FlushedBuffer eventCount:{}, FlushedBuffer size:{}",parsedEvents.size(),flushedBuffer.getEventCount(),flushedBuffer.getSize());
+            // Check if the response is a JSON array and the codec is JSON
+            if (isJsonCodec) {
+                if (parsedEvents.size() == flushedBuffer.getEventCount()) {


We should use the strategy pattern to inject code for handling the parsed events based on aggregate or strict. Then you don't need this large conditional.

Something like this:

responseStrategy.handleEvents(parsedEvents, originalRecords);

Also, this will grow with some upcoming changes to support aggregations in end-to-end acknowledgements.

dlvenable · 2024-10-23T22:19:19Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

+
+            LOG.debug("Parsed Event Size:{}, FlushedBuffer eventCount:{}, FlushedBuffer size:{}",parsedEvents.size(),flushedBuffer.getEventCount(),flushedBuffer.getSize());
+            // Check if the response is a JSON array and the codec is JSON
+            if (isJsonCodec) {


This condition does not belong here. We should verify that the codec is one we support in the constructor. This allows the processor validation to fail fast and give the user quicker feedback.

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

dlvenable · 2024-10-25T21:32:10Z

...src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessorConfig.java

-    private Duration sdkTimeout = DEFAULT_SDK_TIMEOUT;
+    @JsonPropertyDescription("Defines the way Data Prepper treats the response from Lambda")
+    @JsonProperty("response_cardinality")
+    private String responseCardinality;


Suggested change

private String responseCardinality;

private ResponseCardinality responseCardinality;

You will need to make other changes.

dlvenable · 2024-10-25T21:32:36Z

...a/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/ResponseCardinality.java

+        }
+    }
+
+    // Default value is STRICT


Suggested change

// Default value is STRICT

@JsonCreator

This will work better with schema validation.

dlvenable · 2024-10-25T21:33:24Z

...a/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/ResponseCardinality.java

+        this.value = value;
+    }
+
+    public String getValue() {


Suggested change

public String getValue() {

@JsonCreator

public String getValue() {

This will allow this to work well with the schema generation as well.

dlvenable · 2024-10-25T21:34:43Z

...da/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/config/InvocationType.java

+        this.awsLambdaValue = awsLambdaValue;
+    }
+
+    public String getUserInputValue() {


Suggested change

public String getUserInputValue() {

@JsonValue

public String getUserInputValue() {

This is needed to work with schema generation.

dlvenable · 2024-10-25T21:35:08Z

...da/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/config/InvocationType.java

+        }
+    }
+
+    public static InvocationType fromString(String value) {


Suggested change

public static InvocationType fromString(String value) {

@JsonCreator

public static InvocationType fromString(String value) {

This is important for pipeline validation.

dlvenable · 2024-10-25T21:36:01Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

-            }
-            //Reset Buffer
+        LOG.debug("currentBufferPerBatchEventCount:{}, maxEvents:{}, maxBytes:{}, maxCollectionDuration:{}, forceFlush:{} ", currentBufferPerBatch.getEventCount(),maxEvents,maxBytes,maxCollectionDuration, forceFlush);
+        if (forceFlush || ThresholdCheck.checkThresholdExceed(currentBufferPerBatch, maxEvents, maxBytes, maxCollectionDuration)) {


I agree that a flag is not ideal. But, you can have an injectable strategy. We can resolve this later though.

dlvenable · 2024-10-25T21:37:21Z

...ambda/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/LambdaProcessor.java

+            LOG.debug("Parsed Event Size:{}, FlushedBuffer eventCount:{}, FlushedBuffer size:{}", parsedEvents.size(), flushedBuffer.getEventCount(), flushedBuffer.getSize());
+
+            responseStrategy.handleEvents(parsedEvents, originalRecords, resultRecords, flushedBuffer);
+//            if (parsedEvents.size() == flushedBuffer.getEventCount()) {


Let's remove these comments.

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com> Refactor aws lambda plugin to have a class for common methods between processor and sink Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com> Add support for lambda async client in lambda sink Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

…onse codec Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

…nality enums; Change reponse_processing_mode option to response_cardinality Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

dlvenable · 2024-10-28T17:05:05Z

...da/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/config/InvocationType.java

+
+    @JsonCreator
+    public static InvocationType fromString(String value) {
+        return INVOCATION_TYPE_MAP.get(value.toLowerCase());


Please remove the toLowerCase(). We want to be sure that the values provided by the user are the expected values.

dlvenable · 2024-10-28T17:05:20Z

...da/src/main/java/org/opensearch/dataprepper/plugins/lambda/common/config/InvocationType.java

+
+    static {
+        for (InvocationType type : InvocationType.values()) {
+            INVOCATION_TYPE_MAP.put(type.getUserInputValue().toLowerCase(), type);


Please remove the toLowerCase() here.

dlvenable · 2024-10-28T17:06:00Z

...a/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/ResponseCardinality.java

+
+    static {
+        for (ResponseCardinality type : ResponseCardinality.values()) {
+            RESPONSE_CARDINALITY_MAP.put(type.getValue().toLowerCase(), type);


Please remove toLowerCase().

dlvenable · 2024-10-28T17:06:12Z

...a/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/ResponseCardinality.java

+        if (value == null) {
+            return STRICT;
+        }
+        return RESPONSE_CARDINALITY_MAP.getOrDefault(value.toLowerCase(), STRICT);


Please remove .toLowerCase().

Also, do not get the default. This will result in this being valid:

response_cardinality: oops!

dlvenable · 2024-10-28T17:07:23Z

...a/src/main/java/org/opensearch/dataprepper/plugins/lambda/processor/ResponseCardinality.java

+
+    @JsonCreator
+    public static ResponseCardinality fromString(String value) {
+        if (value == null) {


Do not return a default value for null here. This can result in odd behavior such as the following being allowed.

response_cardinality: null

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

srikanthjg requested review from KarstenSchnitter, chenqi0805, dinujoh, dlvenable, engechas, graytaylor0, kkondaka and oeyh as code owners October 8, 2024 06:10

srikanthjg changed the title ~~Add support for lambda async client in lambda processor~~ Address Scale Iterms for lambda plugin Oct 8, 2024

srikanthjg force-pushed the lambda-async-client branch from 6bd7c07 to 3840423 Compare October 8, 2024 16:29

srikanthjg changed the title ~~Address Scale Iterms for lambda plugin~~ Address Scale Items for lambda plugin Oct 8, 2024

graytaylor0 reviewed Oct 8, 2024

View reviewed changes

srikanthjg force-pushed the lambda-async-client branch from abc2c61 to 547dcfd Compare October 18, 2024 05:38

kkondaka reviewed Oct 22, 2024

View reviewed changes

srikanthjg requested a review from sb2k16 as a code owner October 23, 2024 06:42

dlvenable requested changes Oct 23, 2024

View reviewed changes

dlvenable reviewed Oct 25, 2024

View reviewed changes

dlvenable requested changes Oct 25, 2024

View reviewed changes

srikanthjg added 8 commits October 27, 2024 21:55

Changes to Lambda Plugin Integration Test

9b4120d

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Add JsonPropertyDescription to all Config, add debug logs and add ITs

08dc3b3

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Address Acknowledgements for processor and sink; Add request and resp…

7912114

…onse codec Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Address comments

7077c69

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Add response processing mode to processor configuration

8d75227

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Add Response Handling Strategy; Make InvocationType and ResponseCardi…

39df8c0

…nality enums; Change reponse_processing_mode option to response_cardinality Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

Address Enum

81a30c9

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

srikanthjg force-pushed the lambda-async-client branch from 7559f42 to 81a30c9 Compare October 28, 2024 07:13

dlvenable requested changes Oct 28, 2024

View reviewed changes

Address Enum2

16008df

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

dlvenable previously approved these changes Oct 28, 2024

View reviewed changes

kkondaka previously approved these changes Oct 28, 2024

View reviewed changes

Fix checkstyle

f86f679

Signed-off-by: Srikanth Govindarajan <srigovs@amazon.com>

srikanthjg dismissed stale reviews from kkondaka and dlvenable via f86f679 October 28, 2024 22:26

kkondaka approved these changes Oct 29, 2024

View reviewed changes

kkondaka merged commit e311f0d into opensearch-project:main Oct 29, 2024

dlvenable mentioned this pull request Jan 16, 2025

Welcoming Srikanth Govindarajan (srikanthjg) to the Data Prepper maintainers #5337

Merged

4 tasks


		public void flushToLambdaIfNeeded(List<Record<Event>> resultRecords, boolean forceFlush) {

		LOG.info("currentBufferEventCount:{}, maxEvents:{}, maxBytes:{}, maxCollectionDuration:{}, isBatch:{}, forceFlush:{} ", currentBuffer.getEventCount(),maxEvents,maxBytes,maxCollectionDuration,isBatchEnabled, forceFlush);

	private String responseCardinality;
	private ResponseCardinality responseCardinality;

	public String getValue() {
	@JsonCreator
	public String getValue() {

	public String getUserInputValue() {
	@JsonValue
	public String getUserInputValue() {

	public static InvocationType fromString(String value) {
	@JsonCreator
	public static InvocationType fromString(String value) {

Conversation

srikanthjg commented Oct 8, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Issues Resolved

Check List

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srikanthjg Oct 23, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

srikanthjg commented Oct 8, 2024 •

edited

Loading

srikanthjg Oct 23, 2024 •

edited

Loading