Skip to content

Commit 4aa16d7

Browse files
fdelbrayellefdelbrayelle
committed
Revert "feat(lambda): fetch CloudWatch logs after Lambda invocation (#677)"
This reverts commit a10e610.
1 parent 1715657 commit 4aa16d7

File tree

4 files changed

+53
-194
lines changed

4 files changed

+53
-194
lines changed

‎gradle.properties‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,2 +1,2 @@
11
version=1.4.2-SNAPSHOT
2-
kestraVersion=1.2.0
2+
kestraVersion=1.1.0

‎src/main/java/io/kestra/plugin/aws/cloudwatch/CloudWatchLogs.java‎

Lines changed: 0 additions & 34 deletions
This file was deleted.

‎src/main/java/io/kestra/plugin/aws/lambda/Invoke.java‎

Lines changed: 29 additions & 85 deletions
Original file line numberDiff line numberDiff line change
@@ -1,22 +1,8 @@
11
package io.kestra.plugin.aws.lambda;
22

3-
import java.io.File;
4-
import java.io.IOException;
5-
import java.io.UncheckedIOException;
6-
import java.net.URI;
7-
import java.nio.file.Files;
8-
import java.time.Duration;
9-
import java.time.Instant;
10-
import java.util.Map;
11-
import java.util.Optional;
12-
13-
import org.apache.http.HttpHeaders;
14-
import org.apache.http.entity.ContentType;
15-
163
import com.fasterxml.jackson.core.JsonProcessingException;
174
import com.fasterxml.jackson.databind.ObjectMapper;
185
import com.google.common.annotations.VisibleForTesting;
19-
206
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
217
import io.kestra.core.models.annotations.Example;
228
import io.kestra.core.models.annotations.Metric;
@@ -25,13 +11,10 @@
2511
import io.kestra.core.models.executions.metrics.Timer;
2612
import io.kestra.core.models.property.Property;
2713
import io.kestra.core.models.tasks.RunnableTask;
28-
import io.kestra.core.models.tasks.retrys.AbstractRetry;
2914
import io.kestra.core.runners.RunContext;
3015
import io.kestra.core.serializers.JacksonMapper;
31-
import io.kestra.core.utils.RetryUtils;
3216
import io.kestra.plugin.aws.AbstractConnection;
3317
import io.kestra.plugin.aws.ConnectionUtils;
34-
import io.kestra.plugin.aws.cloudwatch.CloudWatchLogs;
3518
import io.kestra.plugin.aws.lambda.Invoke.Output;
3619
import io.kestra.plugin.aws.s3.ObjectOutput;
3720
import io.swagger.v3.oas.annotations.media.Schema;
@@ -42,17 +25,22 @@
4225
import lombok.ToString;
4326
import lombok.experimental.SuperBuilder;
4427
import lombok.extern.slf4j.Slf4j;
28+
import org.apache.http.HttpHeaders;
29+
import org.apache.http.entity.ContentType;
4530
import software.amazon.awssdk.core.SdkBytes;
46-
import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
47-
import software.amazon.awssdk.services.cloudwatchlogs.model.FilterLogEventsRequest;
48-
import software.amazon.awssdk.services.cloudwatchlogs.model.FilteredLogEvent;
4931
import software.amazon.awssdk.services.lambda.LambdaClient;
5032
import software.amazon.awssdk.services.lambda.model.InvokeRequest;
5133
import software.amazon.awssdk.services.lambda.model.InvokeResponse;
5234
import software.amazon.awssdk.services.lambda.model.LambdaException;
53-
import io.kestra.core.models.tasks.retrys.Exponential;
5435

55-
import java.util.List;
36+
import java.io.File;
37+
import java.io.IOException;
38+
import java.io.UncheckedIOException;
39+
import java.net.URI;
40+
import java.nio.file.Files;
41+
import java.time.Duration;
42+
import java.util.Map;
43+
import java.util.Optional;
5644

5745
@SuperBuilder
5846
@ToString
@@ -106,7 +94,6 @@
10694
@Metric(name = "duration", type = Timer.TYPE)
10795
}
10896
)
109-
11097
@Slf4j
11198
public class Invoke extends AbstractConnection implements RunnableTask<Output> {
11299

@@ -116,28 +103,32 @@ public class Invoke extends AbstractConnection implements RunnableTask<Output> {
116103
@NotNull
117104
private Property<String> functionArn;
118105

119-
@Schema(title = "Function request payload.", description = "Request payload. It's a map of string -> object.")
106+
@Schema(
107+
title = "Function request payload.",
108+
description = "Request payload. It's a map of string -> object."
109+
)
120110
private Property<Map<String, Object>> functionPayload;
121111

122112
@Override
123113
public Output run(RunContext runContext) throws Exception {
124114
final long start = System.nanoTime();
125-
final Instant invocationStart = Instant.now().minusSeconds(5);
126115
var functionArn = runContext.render(this.functionArn).as(String.class).orElseThrow();
127-
var requestPayload = runContext.render(this.functionPayload).asMap(String.class, Object.class).isEmpty() ? null
128-
: runContext.render(this.functionPayload).asMap(String.class, Object.class);
116+
var requestPayload = runContext.render(this.functionPayload).asMap(String.class, Object.class).isEmpty() ?
117+
null :
118+
runContext.render(this.functionPayload).asMap(String.class, Object.class);
129119
var logger = runContext.logger();
130120

131121
try (var lambda = client(runContext)) {
132122
var builder = InvokeRequest.builder().functionName(functionArn);
133123
if (requestPayload != null && requestPayload.size() > 0) {
134-
var payload = SdkBytes.fromUtf8String(OBJECT_MAPPER.writeValueAsString(requestPayload));
124+
var payload = SdkBytes.fromUtf8String(OBJECT_MAPPER.writeValueAsString(requestPayload)) ;
135125
builder.payload(payload);
136126
}
137127
InvokeRequest request = builder.build();
138128
// TODO take care about long-running functions: your client might disconnect during
139129
// synchronous invocation while it waits for a response. Configure your HTTP client,
140130
// SDK, firewall, proxy, or operating system to allow for long connections with timeout
131+
// or keep-alive settings.
141132
InvokeResponse res = lambda.invoke(request);
142133
Optional<String> contentTypeHeader = res.sdkHttpResponse().firstMatchingHeader(HttpHeaders.CONTENT_TYPE);
143134
ContentType contentType = parseContentType(contentTypeHeader);
@@ -150,19 +141,13 @@ public Output run(RunContext runContext) throws Exception {
150141
logger.debug("Lambda {} invoked successfully", functionArn);
151142
}
152143
Output out = handleContent(runContext, functionArn, contentType, res.payload());
153-
fetchAndLogLambdaLogs(runContext, functionArn, invocationStart);
154144
runContext.metric(Timer.of("duration", Duration.ofNanos(System.nanoTime() - start)));
155145
return out;
156146
} catch (LambdaException e) {
157147
throw new LambdaInvokeException("Lambda Invoke task execution failed for function: " + functionArn, e);
158148
}
159149
}
160150

161-
@VisibleForTesting
162-
CloudWatchLogsClient getCloudWatchLogsClient(RunContext runContext) throws IllegalVariableEvaluationException {
163-
return new CloudWatchLogs().logsClient(runContext);
164-
}
165-
166151
@VisibleForTesting
167152
LambdaClient client(final RunContext runContext) throws IllegalVariableEvaluationException {
168153
final AwsClientConfig clientConfig = awsClientConfig(runContext);
@@ -180,7 +165,7 @@ ContentType parseContentType(Optional<String> contentType) {
180165
// Apply charset only if it was provided originally
181166
return ContentType.create(known.getMimeType(), parsed.getCharset());
182167
}
183-
} catch (Exception cte) {
168+
} catch(Exception cte) {
184169
log.warn("Unable to parse Lambda response content type {}: {}", contentType.get(), cte.getMessage());
185170
}
186171
}
@@ -210,53 +195,6 @@ Optional<String> readError(String payload) {
210195
return Optional.empty();
211196
}
212197

213-
@VisibleForTesting
214-
void fetchAndLogLambdaLogs(RunContext runContext, String functionArn, Instant startTime) {
215-
var logger = runContext.logger();
216-
String functionName = extractFunctionName(functionArn);
217-
String logGroupName = "/aws/lambda/" + functionName;
218-
219-
// Explicit retry policy: 5 attempts, 3s interval, maxInterval 3s
220-
AbstractRetry retryPolicy = Exponential.builder()
221-
.interval(Duration.ofSeconds(3))
222-
.maxAttempts(5)
223-
.maxInterval(Duration.ofSeconds(10))
224-
.build();
225-
226-
try (CloudWatchLogsClient logsClient = getCloudWatchLogsClient(runContext)) {
227-
// Explicitly specify generic type for RetryUtils
228-
List<FilteredLogEvent> events = RetryUtils.<List<FilteredLogEvent>, Exception>of(retryPolicy, logger)
229-
.run(
230-
result -> result == null || result.isEmpty(),
231-
() -> {
232-
FilterLogEventsRequest request = FilterLogEventsRequest.builder()
233-
.logGroupName(logGroupName)
234-
.startTime(startTime.toEpochMilli())
235-
.build();
236-
237-
var response = logsClient.filterLogEvents(request);
238-
return response.events();
239-
});
240-
241-
if (events != null) {
242-
events.forEach(event -> logger.info("[lambda] {}", event.message().trim()));
243-
}
244-
245-
} catch (Exception e) {
246-
logger.warn("Failed to fetch CloudWatch logs for Lambda {}: {}", functionArn, e.getMessage());
247-
}
248-
}
249-
250-
@VisibleForTesting
251-
private String extractFunctionName(String functionArnOrName) {
252-
if (functionArnOrName.contains(":function:")) {
253-
// Handle Full ARN
254-
return functionArnOrName.split(":function:")[1].split(":")[0];
255-
}
256-
// Handle just the name
257-
return functionArnOrName;
258-
}
259-
260198
@VisibleForTesting
261199
void handleError(String functionArn, ContentType contentType, SdkBytes payload) {
262200
String errorPayload;
@@ -313,14 +251,20 @@ Output handleContent(RunContext runContext, String functionArn, ContentType cont
313251
@Getter
314252
public static class Output extends ObjectOutput implements io.kestra.core.models.tasks.Output {
315253

316-
@Schema(title = "Response file URI.")
254+
@Schema(
255+
title = "Response file URI."
256+
)
317257
private final URI uri;
318258

319-
@Schema(title = "Size of the response content in bytes.")
259+
@Schema(
260+
title = "Size of the response content in bytes."
261+
)
320262

321263
private final Long contentLength;
322264

323-
@Schema(title = "A standard MIME type describing the format of the content.")
265+
@Schema(
266+
title = "A standard MIME type describing the format of the content."
267+
)
324268
private final String contentType;
325269
}
326270
}

‎src/test/java/io/kestra/plugin/aws/lambda/InvokeUnitTest.java‎

Lines changed: 23 additions & 74 deletions
Original file line numberDiff line numberDiff line change
@@ -1,60 +1,45 @@
11
package io.kestra.plugin.aws.lambda;
22

3-
import java.io.ByteArrayInputStream;
4-
import java.io.File;
5-
import java.io.IOException;
6-
import java.net.URI;
7-
import java.nio.file.Files;
8-
import java.time.Instant;
9-
import java.util.Collections;
10-
import java.util.List;
11-
import java.util.Map;
12-
import java.util.Optional;
3+
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
4+
import io.kestra.core.models.property.Property;
5+
import io.kestra.core.runners.RunContext;
6+
import io.kestra.core.runners.RunContextProperty;
7+
import io.kestra.core.runners.WorkingDir;
8+
import io.kestra.core.storages.Storage;
9+
import io.kestra.plugin.aws.lambda.Invoke.Output;
1310
import org.apache.http.entity.ContentType;
1411
import org.junit.jupiter.api.AfterEach;
15-
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
16-
import static org.junit.jupiter.api.Assertions.assertEquals;
17-
import static org.junit.jupiter.api.Assertions.assertNotNull;
18-
import static org.junit.jupiter.api.Assertions.assertThrows;
19-
import static org.junit.jupiter.api.Assertions.assertTrue;
2012
import org.junit.jupiter.api.BeforeEach;
2113
import org.junit.jupiter.api.Test;
2214
import org.junit.jupiter.api.extension.ExtendWith;
23-
import static org.mockito.ArgumentMatchers.any;
24-
import static org.mockito.ArgumentMatchers.eq;
25-
import static org.mockito.BDDMockito.given;
2615
import org.mockito.Mock;
2716
import org.mockito.Mock.Strictness;
28-
29-
import static org.mockito.Mockito.atLeastOnce;
30-
import static org.mockito.Mockito.doReturn;
31-
import static org.mockito.Mockito.mock;
32-
import static org.mockito.Mockito.spy;
33-
import static org.mockito.Mockito.times;
34-
import static org.mockito.Mockito.verify;
3517
import org.mockito.invocation.InvocationOnMock;
3618
import org.mockito.junit.jupiter.MockitoExtension;
3719
import org.mockito.stubbing.Answer;
3820
import org.slf4j.Logger;
39-
40-
import io.kestra.core.exceptions.IllegalVariableEvaluationException;
41-
import io.kestra.core.models.property.Property;
42-
import io.kestra.core.runners.RunContext;
43-
import io.kestra.core.runners.RunContextProperty;
44-
import io.kestra.core.runners.WorkingDir;
45-
import io.kestra.core.storages.Storage;
46-
import io.kestra.plugin.aws.lambda.Invoke.Output;
4721
import software.amazon.awssdk.core.SdkBytes;
4822
import software.amazon.awssdk.http.SdkHttpResponse;
49-
import software.amazon.awssdk.services.cloudwatchlogs.CloudWatchLogsClient;
50-
import software.amazon.awssdk.services.cloudwatchlogs.model.FilterLogEventsRequest;
51-
import software.amazon.awssdk.services.cloudwatchlogs.model.FilterLogEventsResponse;
52-
import software.amazon.awssdk.services.cloudwatchlogs.model.FilteredLogEvent;
53-
import software.amazon.awssdk.services.cloudwatchlogs.paginators.FilterLogEventsIterable;
5423
import software.amazon.awssdk.services.lambda.LambdaClient;
5524
import software.amazon.awssdk.services.lambda.model.InvokeRequest;
5625
import software.amazon.awssdk.services.lambda.model.InvokeResponse;
5726

27+
import java.io.ByteArrayInputStream;
28+
import java.io.File;
29+
import java.io.IOException;
30+
import java.net.URI;
31+
import java.nio.file.Files;
32+
import java.util.Collections;
33+
import java.util.Map;
34+
import java.util.Optional;
35+
36+
import static org.junit.jupiter.api.Assertions.*;
37+
import static org.mockito.ArgumentMatchers.any;
38+
import static org.mockito.ArgumentMatchers.eq;
39+
import static org.mockito.BDDMockito.given;
40+
import static org.mockito.Mockito.doReturn;
41+
import static org.mockito.Mockito.spy;
42+
5843
@ExtendWith(MockitoExtension.class)
5944
public class InvokeUnitTest {
6045

@@ -63,9 +48,6 @@ public class InvokeUnitTest {
6348
@Mock(strictness = Strictness.LENIENT)
6449
private RunContext context;
6550

66-
@Mock
67-
private CloudWatchLogsClient logsClient;
68-
6951
@Mock(strictness = Strictness.LENIENT)
7052
private RunContextProperty runContextProperty;
7153

@@ -221,37 +203,4 @@ void givenFunctionArnNoParams_whenInvokeLambda_thenOutputWithFile(
221203
// Then
222204
checkOutput(data, res);
223205
}
224-
225-
@Test
226-
void givenLambdaInvocation_whenLogsAreFetched_thenLogsAreLoggedCorrectly() throws Throwable {
227-
// Arrange
228-
Instant startTime = Instant.parse("2026-01-15T10:00:00Z");
229-
230-
FilteredLogEvent logEvent = FilteredLogEvent.builder()
231-
.message("Hello from CloudWatch Logs")
232-
.timestamp(startTime.plusSeconds(1).toEpochMilli())
233-
.build();
234-
235-
FilterLogEventsResponse response = FilterLogEventsResponse.builder()
236-
.events(List.of(logEvent))
237-
.build();
238-
239-
given(logsClient.filterLogEvents(any(FilterLogEventsRequest.class)))
240-
.willReturn(response);
241-
242-
given(context.logger()).willReturn(logger);
243-
244-
Invoke spyInvoke = spy(invoke);
245-
doReturn(logsClient).when(spyInvoke).getCloudWatchLogsClient(any());
246-
247-
// Act
248-
spyInvoke.fetchAndLogLambdaLogs(
249-
context,
250-
"arn:aws:lambda:ap-south-1:123456789012:function:test-function",
251-
startTime);
252-
253-
// Assert (THIS is what really matters)
254-
verify(logger).info("[lambda] {}", "Hello from CloudWatch Logs");
255-
}
256206
}
257-

0 commit comments

Comments
 (0)