feat(lambda-event-sources): add filters to SQS, DynamoDB, and Kinesis event sources

[marciocadev]

Description

Adds the ability to create filters for SQS, DynamoDB and Kinesis, enabling filter criteria settings for event sources

Use Cases

With this PR will be possible, for example, to filter events from a DynamoDB Stream allowing only INSERT events to be transmitted as shown in the example below

    const fn = new NodejsFunction(this, 'Fn');
    const table = new dynamodb.Table(this, 'T', {
      partitionKey: {
        name: 'id',
        type: dynamodb.AttributeType.STRING,
      },
      stream: dynamodb.StreamViewType.NEW_IMAGE,
    });

    fn.addEventSource(new sources.DynamoEventSource(table, {
      startingPosition: lambda.StartingPosition.LATEST,
      filters: [
         lambda.FilterCriteria.filter({
            eventName: FilterRule.isEqual('INSERT'),
         }),
      ],
    }));

Closes #17874

All Submissions:

• Have you followed the guidelines in our Contributing guide?

Adding new Unconventional Dependencies:

• This PR adds new unconventional dependencies following the process described here

New Features

• Have you added the new feature to an integration test?
  • Did you use yarn integ to deploy the infrastructure and generate the snapshot (i.e. yarn integ without --dry-run)?

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache-2.0 license

[gitpod-io]

[TheRealAmazonKendra]

@Mergifyio update

[TheRealAmazonKendra]

Thank you for your contribution! I didn't dive in too deep to this because I think the user experience needs to be tweaked a little bit. I'm also seeing a bunch of new integ expected.json files that were produced, which shouldn't be the case without tests being altered as well. This may be a result of old style tests still being used here? Do me a favor and delete those expected files, and then re-run the tests.

[TheRealAmazonKendra]

I don't think this is quite the right user experience here. We like to use props rather than functions in cases like this. I would suggest renaming filterCriteria to filters and add the filters directly as a list. Then we can handle transforming that

[marciocadev]

Thank you for you reply

Something like this would be better right?

  filters: [
    { eventName: lambda.FilterRule.textEquals('INSERT') },
  ]

I will also make a change in the comparison because I didn't like textEquals and numericEquals

An isEqual call to string or number will be better

[TheRealAmazonKendra]

Yep, that's precisely what I was thinking.

Pull request has been modified.

[TheRealAmazonKendra]

Looks like two files got deleted that shouldn't have been. They're the two under cloudformation-include. That's what's causing the build failure.

Also, I forgot to comment this before but please make sure that your PR title confirms to the conventional commit standard (fix, feat, chore) and that it is written in a style that will reflect correctly in the change log (See Contributing Guide, Pull Requests)

Pull request has been modified.

[TheRealAmazonKendra]

So, I misread your intention here. I'm going to take the input to one of your integration tests here to clarify what I thought you meant here:

Instead of this:

    new EventSourceMapping(stack, 'test', {
      target: fn,
      eventSourceArn: eventSourceArn,
      kafkaTopic: topicNameParam.valueAsString,
      filters: [{
        orFilter: FilterRule.or('one', 'two'),
        stringEquals: FilterRule.isEqual('test'),
        numericEquals: FilterRule.isEqual(1),
      }],
    });

I thought you meant it would look like this:

    new EventSourceMapping(stack, 'test', {
      target: fn,
      eventSourceArn: eventSourceArn,
      kafkaTopic: topicNameParam.valueAsString,
      filters: {
        orFilter: FilterRule.or('one', 'two'),
        stringEquals: FilterRule.isEqual('test'),
        numericEquals: FilterRule.isEqual(1),
      },
    });

who's contract would look like { [key: string]: any }. In other words, I read too quickly and gave you a bad answer. I apologize for that. However, the other piece of this that I'd like to suggest an improvement upon is tightening the contract and not using any in here. See my comments in FilterRule.

[marciocadev]

The problem is that the event source allows up to 5 filters (can be expanded if requested) and these filters can have several rules, so I put them in an array where each element would be a filter (with several rules)

The FilterRule needs to match the exact format of the message that transits the stream, as in this example where the message will be sent if it is an INSERT and has the gameScore field

Note that to navigate the message I had to follow the path to the property that must be validated
dynamodb -> NewImage -> gameScore -> S -> RULE

fn.addEventSource(new DynamoEventSource(table, {
  batchSize: 5,
  startingPosition: lambda.StartingPosition.LATEST,
  filters: [{
    eventName: lambda.FilterRule.isEqual('INSERT'),
    dynamodb: {
      NewImage: {
        gameScore: {
          S: lambda.FilterRule.exists(),
        },
      },
    },
  }],
}));

[TheRealAmazonKendra]

Ohhhhh. I see. Thanks for the extra explanation. No need to change this, then.

[marciocadev]

Thinking about readability of this feature, what do you think about going back to my earlier code concept and create a class representation of a single filter ?

like this:

fn.addEventSource(new DynamoEventSource(table, {
  batchSize: 5,
  startingPosition: StartingPosition.LATEST,
  filters: [
    FilterCriteria.filter({
      eventName: FilterRule.isEqual('INSERT'),
      dynamodb: {
        NewImage: {
          userName: {
            S: FilterRule.exists(),
          },
        },
      },
    }),
    FilterCriteria.filter({ 
      eventName: FilterRule.isEqual('MODIFY'),
      dynamodb: {
        NewImage: {
          gameScore: {
            N: FilterRule.exists(),
          },
        },
      },
    }),
  ],
}));

[TheRealAmazonKendra]

I like this.

[marciocadev]

Done

[TheRealAmazonKendra]

This is an excellent start here. I'd like to see the return be a class on its own so that the contract for filters can be { [key: string]: FilterRule }. Schedule is an excellent example of what I'd like to see here.

[marciocadev]

This is good, I believe I can use this approach

[TheRealAmazonKendra]

If you feel like this can be further refined, I'm all for it, but given your note above, I'm not entirely sure it's feasible. Let me know if you want to take a crack at it still or if you just want to re-review with what's already done. Let me know what you think.

[marciocadev]

You're right, FilterRule's returns are so diverse that it's difficult to create a single contract

[Lewenhaupt]

@marciocadev quick question as I was a bit confused at first when trying to use this myself. I read the docs on https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_lambda.FilterRule.html but that does not mention the wrapping with lambda.FilterCriteria.filter. Without this it seems the filter isn't actually applied, but with it everything works as expected.
Is this simply an omission in the docs or should it work without wrapping it?

[Spjak]

I am in the same situation as @Lewenhaupt - also spent a while in the docs before I found this PR and realized I had to use the FilterCriteria to make it work.
The incorrect syntax is also mentioned here:
https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.aws_lambda_event_sources.DynamoEventSource.html

[FaizanRaza-InterWeave]

@Spjak I just spent an hour crying that it wasnt working :)

Thankyou!

Pull request has been modified.

[mergify]

Thank you for contributing! Your pull request will be updated from main and then merged automatically (do not update manually, and be sure to allow changes to be pushed to your fork).

[aws-cdk-automation]

AWS CodeBuild CI Report

• CodeBuild project: AutoBuildv2Project1C6BFA3F-wQm2hXv2jqQv
• Commit ID: 7d6423a
• Result: SUCCEEDED
• Build Logs (available for 30 days)

Powered by github-codebuild-logs, available on the AWS Serverless Application Repository

[mergify]

Thank you for contributing! Your pull request will be updated from main and then merged automatically (do not update manually, and be sure to allow changes to be pushed to your fork).