New serverless pattern - Kinesis DS with ESM filtering (CDK + Python)

kinesis-data-stream-lambda-esm-cdk-python/README.md

@@ -0,0 +1,124 @@
+# Amazon Kinesis Data Streams to AWS Lambda with event filtering
+
+This pattern demonstrates the ability configure Amazon Kinesis as an event source for AWS Lambda to use event filtering to control which records are sent to your function for processing. The pattern deploys a Kinesis data stream and Lambda functions that are subscribed to the stream with different event filter configurations.
+
+Review [Filter rule syntax](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html#filtering-syntax) for more details on the event filtering configuration.
+
+Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/kinesis-data-stream-lambda-esm-cdk-python/
+
+Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. You are responsible for any AWS costs incurred. No warranty is implied in this example.
+
+## Requirements
+
+* [Create an AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) if you do not already have one and log in. The IAM user that you use must have sufficient permissions to make necessary AWS service calls and manage AWS resources.
+* [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html) installed and configured
+* [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git)
+* [AWS Cloud Development Kit](https://docs.aws.amazon.com/cdk/latest/guide/cli.html) (AWS CDK) installed
+
+
+## Deployment Instructions
+
+1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository:
+    ``` 
+    git clone https://github.com/aws-samples/serverless-patterns
+    ```
+1. Change directory to the pattern directory:
+    ```
+    cd kinesis-data-stream-lambda-esm-cdk-python/cdk
+    ```
+1. Create a Python virtual environment
+    ```
+    python -m venv .venv
+    ```
+1. Activate the virtualenv
+    ```
+    source .venv/bin/activate
+    ```
+
+    If you are using a Windows platform, you would activate the virtualenv like this:
+    ```
+    .venv\Scripts\activate.bat
+    ```
+2. After the virtualenv is activated, you can install the required dependencies.
+    ```
+    pip install -r requirements.txt
+    ```
+3. Bootstrap your AWS account and Region (if you have not already done so)
+    ```
+    cdk bootstrap
+    ```
+4. Deploy the stack to your AWS account and region.
+    ```
+    cdk deploy
+    ```
+
+## How it works
+
+Multiple Lambda functions and a Kinesis data stream are created with Kinesis configured as the event source. Event source mappings are created with different event filter settings to demonstrate how filtering settings affect which events are sent to the Lambda functions for processing.
+
+
+
+## Testing
+
+You can execute a test Python script to write sample records to the stream.
+
+```bash
+python scripts/producer.py
+```
+
+### Example Records
+
+
+```json
+{
+    'EVENT_TIME': '2023-12-21T16:43:09.730234',
+    'SENSOR_ID': '4d894af2-aea5-4a38-bcc0-336b8741f476',
+    'VALUE': 65.9,
+    'STATUS': 'WARN'
+}
+```
+
+```json
+{
+    'EVENT_TIME': '2023-12-21T16:43:09.889185',
+    'SENSOR_ID': '8be06d7d-9278-4ba0-93d2-567bebbde784',
+    'VALUE': 49.62,
+    'STATUS': 'OK'
+}
+```
+
+```json
+{
+    'EVENT_TIME': '2023-12-21T16:43:10.005793',
+    'SENSOR_ID': 'eb560fc8-bb0b-4032-8229-69d864d2e7d5',
+    'VALUE': 31.81,
+    'STATUS': 'FAIL'
+}
+```
+
+### Viewing test results
+
+Navigate to the CloudWatch console and inspect messages logged to the log groups named similar to those listed below:
+
+| Log Group | Event filter pattern(s) | Comment |
+| --- | --- | --- |
+| /aws/lambda/KinesisLambdaStack-LambdaConsumerNoFilter | N/A | logs all records |
+| /aws/lambda/KinesisLambdaStack-LambdaConsumerFailStatus | `{"data":{"STATUS":["FAIL"]}}` |  logs records where STATUS equals FAIL |
+| /aws/lambda/KinesisLambdaStack-LambdaConsumerNotOkStatus | `{"data":{"STATUS":[{"anything-but":["OK"]}]}}`| logs records where STATUS is not "OK" |
+| /aws/lambda/KinesisLambdaStack-LambdaConsumerWarnValue | `{"data":{"STATUS":["WARN"], "VALUE":[{"numeric":[">",0,"<=",80]}]}}`| logs records where STATUS is "WARN" **and** VALUE is between 0 and 80 (inclusive) |
+| /aws/lambda/KinesisLambdaStack-LambdaConsumerWarnLessValue | `{"data":{"STATUS":["WARN"]}}` and `{"data":{"VALUE":[{"numeric":["<",80]}]}}` | logs records where STATUS is "WARN" **or** VALUE is greater than 80 |
+
+
+## Cleanup
+
+1. Run the following command to delete the resources
+
+```bash
+cdk destroy
+```
+
+
+----
+Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+SPDX-License-Identifier: MIT-0

kinesis-data-stream-lambda-esm-cdk-python/cdk/.gitignore

@@ -0,0 +1,10 @@
+*.swp
+package-lock.json
+__pycache__
+.pytest_cache
+.venv
+*.egg-info
+
+# CDK asset staging directory
+.cdk.staging
+cdk.out

kinesis-data-stream-lambda-esm-cdk-python/cdk/app.py

@@ -0,0 +1,15 @@
+#!/usr/bin/env python3
+import os
+
+import aws_cdk as cdk
+
+from data_stream_processor.kinesis_lambda import KinesisLambdaStack
+
+
+app = cdk.App()
+KinesisLambdaStack(
+    app,
+    "KinesisLambdaStack"
+)
+
+app.synth()

kinesis-data-stream-lambda-esm-cdk-python/cdk/cdk.json

@@ -0,0 +1,61 @@
+{
+  "app": "python3 app.py",
+  "watch": {
+    "include": [
+      "**"
+    ],
+    "exclude": [
+      "README.md",
+      "cdk*.json",
+      "requirements*.txt",
+      "source.bat",
+      "**/__init__.py",
+      "**/__pycache__",
+      "tests"
+    ]
+  },
+  "context": {
+    "@aws-cdk/aws-lambda:recognizeLayerVersion": true,
+    "@aws-cdk/core:checkSecretUsage": true,
+    "@aws-cdk/core:target-partitions": [
+      "aws",
+      "aws-cn"
+    ],
+    "@aws-cdk-containers/ecs-service-extensions:enableDefaultLogDriver": true,
+    "@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true,
+    "@aws-cdk/aws-ecs:arnFormatIncludesClusterName": true,
+    "@aws-cdk/aws-iam:minimizePolicies": true,
+    "@aws-cdk/core:validateSnapshotRemovalPolicy": true,
+    "@aws-cdk/aws-codepipeline:crossAccountKeyAliasStackSafeResourceName": true,
+    "@aws-cdk/aws-s3:createDefaultLoggingPolicy": true,
+    "@aws-cdk/aws-sns-subscriptions:restrictSqsDescryption": true,
+    "@aws-cdk/aws-apigateway:disableCloudWatchRole": true,
+    "@aws-cdk/core:enablePartitionLiterals": true,
+    "@aws-cdk/aws-events:eventsTargetQueueSameAccount": true,
+    "@aws-cdk/aws-iam:standardizedServicePrincipals": true,
+    "@aws-cdk/aws-ecs:disableExplicitDeploymentControllerForCircuitBreaker": true,
+    "@aws-cdk/aws-iam:importedRoleStackSafeDefaultPolicyName": true,
+    "@aws-cdk/aws-s3:serverAccessLogsUseBucketPolicy": true,
+    "@aws-cdk/aws-route53-patters:useCertificate": true,
+    "@aws-cdk/customresources:installLatestAwsSdkDefault": false,
+    "@aws-cdk/aws-rds:databaseProxyUniqueResourceName": true,
+    "@aws-cdk/aws-codedeploy:removeAlarmsFromDeploymentGroup": true,
+    "@aws-cdk/aws-apigateway:authorizerChangeDeploymentLogicalId": true,
+    "@aws-cdk/aws-ec2:launchTemplateDefaultUserData": true,
+    "@aws-cdk/aws-secretsmanager:useAttachedSecretResourcePolicyForSecretTargetAttachments": true,
+    "@aws-cdk/aws-redshift:columnId": true,
+    "@aws-cdk/aws-stepfunctions-tasks:enableEmrServicePolicyV2": true,
+    "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": true,
+    "@aws-cdk/aws-apigateway:requestValidatorUniqueId": true,
+    "@aws-cdk/aws-kms:aliasNameRef": true,
+    "@aws-cdk/aws-autoscaling:generateLaunchTemplateInsteadOfLaunchConfig": true,
+    "@aws-cdk/core:includePrefixInUniqueNameGeneration": true,
+    "@aws-cdk/aws-efs:denyAnonymousAccess": true,
+    "@aws-cdk/aws-opensearchservice:enableOpensearchMultiAzWithStandby": true,
+    "@aws-cdk/aws-lambda-nodejs:useLatestRuntimeVersion": true,
+    "@aws-cdk/aws-efs:mountTargetOrderInsensitiveLogicalId": true,
+    "@aws-cdk/aws-rds:auroraClusterChangeScopeOfInstanceParameterGroupWithEachParameters": true,
+    "@aws-cdk/aws-appsync:useArnForSourceApiAssociationIdentifier": true,
+    "@aws-cdk/aws-rds:preventRenderingDeprecatedCredentials": true
+  }
+}

kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/consumer/lambda_function.py

@@ -0,0 +1,11 @@
+import base64
+
+
+def handler(event, context):
+    print("Event Received: ")
+    print(event)
+
+    for record in event['Records']:
+        #Kinesis data is base64 encoded so decode here
+        payload=base64.b64decode(record["kinesis"]["data"])
+        print("Decoded payload: " + str(payload))

kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/kinesis_lambda.py

@@ -0,0 +1,139 @@
+from aws_cdk import (
+    Duration,
+    Stack,
+    aws_lambda as lambda_,
+    aws_kinesis as kinesis,
+    aws_lambda_event_sources as event_sources,
+)
+from constructs import Construct
+
+class KinesisLambdaStack(Stack):
+
+    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
+        super().__init__(scope, construct_id, **kwargs)
+
+        kinesis_stream = kinesis.Stream(self, "stream-lambda-esm-filter", stream_name="stream-lambda-esm-filter")
+
+
+        consumer_func_no_filter = lambda_.Function(
+            self, 'LambdaConsumerNoFilter',
+            handler='lambda_function.handler',
+            code=lambda_.Code.from_asset('data_stream_processor/consumer'),
+            runtime=lambda_.Runtime.PYTHON_3_11,
+            timeout=Duration.seconds(30)
+        )
+        kinesis_stream.grant_read(consumer_func_no_filter)
+
+        # Event Filter: None; receive all records from event source
+        consumer_func_no_filter.add_event_source(
+            event_sources.KinesisEventSource(
+                stream=kinesis_stream,
+                starting_position=lambda_.StartingPosition.LATEST,
+                batch_size=1
+            )
+        )
+
+        consumer_func_fail = lambda_.Function(
+            self, 'LambdaConsumerFailStatus',
+            handler='lambda_function.handler',
+            code=lambda_.Code.from_asset('data_stream_processor/consumer'),
+            runtime=lambda_.Runtime.PYTHON_3_11,
+            timeout=Duration.seconds(30)
+        )
+        kinesis_stream.grant_read(consumer_func_fail)
+
+        # Event Filter: records where "STATUS" attribute is "FAIL" only
+        # Equals comparison
+        consumer_func_fail.add_event_source(
+            event_sources.KinesisEventSource(
+                stream=kinesis_stream,
+                starting_position=lambda_.StartingPosition.LATEST,
+                batch_size=1,
+                filters=[
+                    lambda_.FilterCriteria.filter({"data": {
+                        "STATUS": lambda_.FilterRule.is_equal("FAIL")
+                    }
+                    })
+                ]
+            )
+        )
+
+        consumer_func_not_ok = lambda_.Function(
+            self, 'LambdaConsumerNotOkStatus',
+            handler='lambda_function.handler',
+            code=lambda_.Code.from_asset('data_stream_processor/consumer'),
+            runtime=lambda_.Runtime.PYTHON_3_11,
+            timeout=Duration.seconds(30)
+        )
+        kinesis_stream.grant_read(consumer_func_not_ok)
+
+        # Event Filter: records where "STATUS" attribute is not "OK"
+        # anything-but comparison
+        consumer_func_not_ok.add_event_source(
+            event_sources.KinesisEventSource(
+                stream=kinesis_stream,
+                starting_position=lambda_.StartingPosition.LATEST,
+                batch_size=1,
+                filters=[
+                    lambda_.FilterCriteria.filter({"data": {
+                        "STATUS": lambda_.FilterRule.not_equals("OK")
+                    }
+                    })
+                ]
+            )
+        )
+
+        consumer_func_warn_value = lambda_.Function(
+            self, 'LambdaConsumerWarnValue',
+            handler='lambda_function.handler',
+            code=lambda_.Code.from_asset('data_stream_processor/consumer'),
+            runtime=lambda_.Runtime.PYTHON_3_11,
+            timeout=Duration.seconds(30)
+        )
+        kinesis_stream.grant_read(consumer_func_warn_value)
+
+        # Event Filter: records where "STATUS" attribute is "WARN" and "VALUE" is between 0 and 80 (inclusive)
+        # AND comparison
+        consumer_func_warn_value.add_event_source(
+            event_sources.KinesisEventSource(
+                stream=kinesis_stream,
+                starting_position=lambda_.StartingPosition.LATEST,
+                batch_size=1,
+                filters=[
+                    lambda_.FilterCriteria.filter(
+                        {"data": 
+                            {
+                            "STATUS": lambda_.FilterRule.is_equal("WARN"),
+                            "VALUE": lambda_.FilterRule.between(0, 80)
+                            }
+                        }
+                    )
+                ]
+            )
+        )
+
+        consumer_func_warn_less_than_value = lambda_.Function(
+            self, 'LambdaConsumerWarnLessValue',
+            handler='lambda_function.handler',
+            code=lambda_.Code.from_asset('data_stream_processor/consumer'),
+            runtime=lambda_.Runtime.PYTHON_3_11,
+            timeout=Duration.seconds(30)
+        )
+        kinesis_stream.grant_read(consumer_func_warn_less_than_value)
+
+        # Event Filter: records where "STATUS" attribute is "WARN" or "VALUE" less than 80
+        # Defining filter rule without CDK FilterRule library
+        # multiple fields, Or comparison
+        consumer_func_warn_less_than_value.add_event_source(
+            event_sources.KinesisEventSource(
+                stream=kinesis_stream,
+                starting_position=lambda_.StartingPosition.LATEST,
+                batch_size=1,
+                filters=[
+                    lambda_.FilterCriteria.filter({"data": {"STATUS":["WARN"]}}),
+                    lambda_.FilterCriteria.filter(
+                        {"data": {"VALUE": [{"numeric": ["<", 80]}]}}
+                    )
+                ]
+            )
+        )

kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements-dev.txt

@@ -0,0 +1 @@
+pytest==6.2.5

kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements.txt

@@ -0,0 +1,4 @@
+aws-cdk-lib==2.102.0
+constructs>=10.0.0,<11.0.0
+boto3>=1.28.72
+botocore>=1.31.72