feat(lambda): fetch CloudWatch logs after Lambda invocation

[shivamwayal37]

closes kestra-io/kestra#8978

What changes are being made and why?

This PR adds support for fetching CloudWatch Logs for AWS Lambda invocations and streaming them directly into the task logger.

Key changes:

• Introduces a reusable CloudWatchLogs abstraction to query logs via the FilterLogEvents paginator.
• Modifies the Invoke task to fetch and log Lambda invocation logs when enabled.
• Adds unit tests covering log retrieval and logging behavior to ensure reliability.

Why:
This improves observability for Lambda tasks in Kestra by allowing developers to see the Lambda execution logs directly in the task output, making debugging and monitoring easier.

---

How the changes have been QAed?

Added unit tests in InvokeUnitTest that:

• Mock the CloudWatch Logs paginator
• Verify that log messages are correctly fetched and logged
• Verified locally using ./gradlew clean test

id: lambda_with_logs
namespace: dev

tasks:
  - id: invoke_lambda
    type: io.kestra.plugin.aws.lambda.Invoke
    functionArn: "arn:aws:lambda:eu-central-1:123456789012:function:test"
    accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
    region: "eu-central-1"

Terminal Output:

Setup Instructions

No additional setup is required beyond standard AWS credentials with permissions to:

• Invoke the Lambda function
• Read CloudWatch Logs (logs:FilterLogEvents)

---

Contributor Checklist ✅

• PR Title and commits follows conventional commits
• Add a closes #ISSUE_ID or fixes #ISSUE_ID in the description if the PR relates to an opened issue.
• Documentation updated (plugin docs from @Schema for properties and outputs, @Plugin with examples, README.md file with basic knowledge and specifics).
• Setup instructions included if needed (API keys, accounts, etc.).
• Prefix all rendered properties by r not rendered (eg: rHost).
• Use runContext.logger() to log enough important infos where it's needed and with the best level (DEBUG, INFO, WARN or ERROR).

⚙️ Properties

• Properties are declared with Property<T> carrier type, do not use @PluginProperty.
• Mandatory properties must be annotated with @NotNull and checked during the rendering.
• You can model a JSON thanks to a simple Property<Map<String, Object>>.

🌐 HTTP

• Must use Kestra’s internal HTTP client from io.kestra.core.http.client

📦 JSON

• If you are serializing response from an external API, you may have to add a @JsonIgnoreProperties(ignoreUnknown = true) at the mapped class level. So that we will avoid to crash the plugin if the provider add a new field suddenly.
• Must use Jackson mappers provided by core (io.kestra.core.serializers)

✨ New plugins / subplugins

• Make sure your new plugin is configured like mentioned here.
• Add a package-info.java under each sub package respecting this format and choosing the right category.
• Every time you use runContext.metric(...) you have to add a @Metric (see this doc)
• Docs don't support to have both tasks/triggers in the root package (e.g. io.kestra.plugin.kubernetes) and in a sub package (e.g. io.kestra.plugin.kubernetes.kubectl), whether it's: all tasks/triggers in the root package OR only tasks/triggers in sub packages.
• Icons added in src/main/resources/icons in SVG format and not in thumbnail (keep it big):
  • plugin-icon.svg
  • One icon per package, e.g. io.kestra.plugin.aws.svg
  • For subpackages, e.g. io.kestra.plugin.aws.s3, add io.kestra.plugin.aws.s3.svg
    See example here.
• Use "{{ secret('YOUR_SECRET') }}" in the examples for sensible infos such as an API KEY.
• If you are fetching data (one, many or too many), you must add a Property<FetchType> fetchType to be able to use FETCH_ONE, FETCH and even STORE to store big amount of data in the internal storage.
• Align the """ to close examples blocks with the flow id.
• Update the existing index.yaml for the main plugin, and for each new subpackage add a metadata file named exactly after the subpackage (e.g. s3.yaml for io.kestra.plugin.aws.s3) under src/main/resources/metadata/, following the same schema.

🧪 Tests

• Unit Tests added or updated to cover the change (using the RunContext to actually run tasks).
• Add sanity checks if possible with a YAML flow inside src/test/resources/flows.
• Avoid disabling tests for CI. Instead, configure a local environment whenever it's possible with .github/setup-unit.sh (to be set executable with chmod +x setup-unit.sh) (which can be executed locally and in the CI) all along with a new docker-compose-ci.yml file (do not edit the existing docker-compose.yml). If needed, create an executable (chmod +x cleanup-unit.sh) cleanup-unit.sh to remove the potential costly resources (tables, datasets, etc).
• Provide screenshots from your QA / tests locally in the PR description. The goal here is to use the JAR of the plugin and directly test it locally in Kestra UI to ensure it integrates well.

📤 Outputs

• Do not send back as outputs the same infos you already have in your properties.
• If you do not have any output use VoidOutput.
• Do not output twice the same infos (eg: a status code, an error code saying the same thing...).

[fdelbrayelle]

Hello @shivamwayal37 👋
Could you provide screenshots where you used this plugin fix with an actual AWS actual with Lambdas and show the logs?
Thanks!

[shivamwayal37]

Hello @fdelbrayelle,

Thanks for the review and approval 🙂
I’ll run this against a real AWS Lambda and add screenshots from the Kestra UI showing the CloudWatch logs streamed into the task logs. I’ll follow up shortly with the screenshots.

Thanks again!

[shivamwayal37]

Kestra Execution Logs: CloudWatch Stream with [lambda] Prefix:

AWS Lambda Configuration: Test Function and Runtime Environment:

Source Comparison: Original CloudWatch Log Events in AWS Console:

[fdelbrayelle]

Thank you @shivamwayal37, it seems great but CI fails on a test, could you check please? 🙏

[shivamwayal37]

Hello @fdelbrayelle,

Thanks for pointing that out! I've investigated the CI failure and can confirm it's unrelated to my changes.

The failure occurred in io.kestra.plugin.aws.kinesis.ConsumeTest (java.lang.AssertionError: Expected: is <1> but: was <10>). This appears to be a flaky test in the Kinesis package.

On my side, I’ve verified that all tests in io.kestra.plugin.aws.lambda.* (including the updated unit test for the new polling logic) are passing locally and in the CI logs.

I have also cleaned up the code (reverted the version and removed debug logs) as requested. I will try to rerun the CI job now to see if it clears up!

[shivamwayal37]

Hey @fdelbrayelle,

Technical Improvements:

• Reliable Polling: Implemented a retry loop (5 attempts, 3s interval) to handle CloudWatch ingestion latency, ensuring logs are captured even if not immediately available post-invocation.
• Dynamic Group Discovery: Added extractFunctionName to automatically resolve the log group (/aws/lambda/) from the provided ARN.
• Enhanced Visibility: Prefixed streamed logs with [lambda] for clear distinction in the Kestra console.
• Stability: Added InterruptedException handling within the polling loop to ensure worker thread stability.

[fdelbrayelle]

Hi @shivamwayal37 👋 Indeed there's another PR to fix the flaky tests, so LGTM!

[shivamwayal37]

Thanks for the heads-up.

The CI failures don’t seem related to this PR’s changes. The failing tests are integration-style tests (AWS CLI, CloudWatch, triggers) and are erroring due to external dependencies not being available in CI (e.g. LocalStack/CloudWatch returning 500 or connection refused on 127.0.0.1:4566).

The Lambda-related changes introduced in this PR are covered by InvokeUnitTest and InvokeTest, which are passing.

I’ll rebase on the latest main and re-run the checks to see if this is a transient CI issue.

[shivamwayal37]

I noticed the latest CI failures are in integration-style tests (AwsCLITest, CloudWatch Push/Query/Trigger) that depend on external services.

The errors look like LocalStack / CloudWatch not being available in CI (127.0.0.1:4566 connection refused / 500 responses).

All Lambda-related tests touched by this PR (InvokeUnitTest, InvokeTest) are passing consistently.

Happy to help investigate or adjust the CI setup for these tests if that would be useful.

[fdelbrayelle]

OK let's merge it seems related to the PR coming from a fork!

[shivamwayal37]

Thanks for the review!

[fdelbrayelle]

It seems to be still failing on main @shivamwayal37 and since you updated Invoke and it's failing since a given commits from your PR, could you have a check please? 🙏

[shivamwayal37]

Thanks for the heads-up 🙏
I’ll take a look at the failures on main and see if they’re related to the Invoke changes or still coming from the integration tests / CI setup.
I’ll report back as soon as I have more clarity.

[shivamwayal37]

I checked main locally and can reproduce the failures.

All failing tests are Testcontainers / LocalStack integration tests and fail early with Could not find a valid Docker environment (/var/run/docker.sock not found).

InvokeUnitTest and other unit/mocked tests are passing consistently.

This looks like a Docker availability issue in CI rather than a regression from the Invoke changes.

Happy to help adjust the CI setup or skip these tests when Docker isn’t available if that helps.

[fdelbrayelle]

Hmm but it was passing before your PR so it must be related to one of your changes... Could you investigate more please?

[shivamwayal37]

Thanks for the clarification — that’s fair.
I’ll dig a bit deeper to see if any of the Invoke changes could indirectly affect the integration tests (timing, client config, startup behavior, etc.), and I’ll also compare against the last green CI run before the merge.
I’ll report back with concrete findings shortly.