diff --git a/kinesis-data-stream-lambda-esm-cdk-python/README.md b/kinesis-data-stream-lambda-esm-cdk-python/README.md new file mode 100644 index 000000000..e7556649c --- /dev/null +++ b/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 \ No newline at end of file diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/.gitignore b/kinesis-data-stream-lambda-esm-cdk-python/cdk/.gitignore new file mode 100644 index 000000000..37833f8be --- /dev/null +++ b/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 diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/app.py b/kinesis-data-stream-lambda-esm-cdk-python/cdk/app.py new file mode 100644 index 000000000..c0a2a843b --- /dev/null +++ b/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() diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/cdk.json b/kinesis-data-stream-lambda-esm-cdk-python/cdk/cdk.json new file mode 100644 index 000000000..33ab988ba --- /dev/null +++ b/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 + } +} diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/__init__.py b/kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/consumer/lambda_function.py b/kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/consumer/lambda_function.py new file mode 100644 index 000000000..18f313056 --- /dev/null +++ b/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)) \ No newline at end of file diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/kinesis_lambda.py b/kinesis-data-stream-lambda-esm-cdk-python/cdk/data_stream_processor/kinesis_lambda.py new file mode 100644 index 000000000..0608edbcb --- /dev/null +++ b/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]}]}} + ) + ] + ) + ) \ No newline at end of file diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements-dev.txt b/kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements-dev.txt new file mode 100644 index 000000000..927094516 --- /dev/null +++ b/kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements-dev.txt @@ -0,0 +1 @@ +pytest==6.2.5 diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements.txt b/kinesis-data-stream-lambda-esm-cdk-python/cdk/requirements.txt new file mode 100644 index 000000000..b988e2de1 --- /dev/null +++ b/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 \ No newline at end of file diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/scripts/producer.py b/kinesis-data-stream-lambda-esm-cdk-python/cdk/scripts/producer.py new file mode 100644 index 000000000..f63d4e5de --- /dev/null +++ b/kinesis-data-stream-lambda-esm-cdk-python/cdk/scripts/producer.py @@ -0,0 +1,34 @@ +import datetime +import json +import random +import uuid +import boto3 + + +# Change the stream name per your testing requirements +STREAM_NAME = "stream-lambda-esm-filter" +MAX_RECORDS = 10 + +def get_data(): + + return { + "EVENT_TIME": datetime.datetime.now().isoformat(), + "SENSOR_ID": "{}".format(uuid.uuid4()), + "VALUE": round(random.random() * 100, 2), + "STATUS": random.choice(["OK", "FAIL", "WARN"]) + } + + +def generate(stream_name, kinesis_client): + for _ in range(0, MAX_RECORDS): + data = get_data() + print(data) + + kinesis_client.put_record( + StreamName=stream_name, Data=json.dumps(data), + PartitionKey="myPartitionKey" #random.choice(["pk1", "pk2", "pk3", "pk4"]) + ) + + +if __name__ == "__main__": + generate(STREAM_NAME, boto3.client("kinesis")) diff --git a/kinesis-data-stream-lambda-esm-cdk-python/cdk/source.bat b/kinesis-data-stream-lambda-esm-cdk-python/cdk/source.bat new file mode 100644 index 000000000..9e1a83442 --- /dev/null +++ b/kinesis-data-stream-lambda-esm-cdk-python/cdk/source.bat @@ -0,0 +1,13 @@ +@echo off + +rem The sole purpose of this script is to make the command +rem +rem source .venv/bin/activate +rem +rem (which activates a Python virtualenv on Linux or Mac OS X) work on Windows. +rem On Windows, this command just runs this batch file (the argument is ignored). +rem +rem Now we don't need to document a Windows command for activating a virtualenv. + +echo Executing .venv\Scripts\activate.bat for you +.venv\Scripts\activate.bat diff --git a/kinesis-data-stream-lambda-esm-cdk-python/example-pattern.json b/kinesis-data-stream-lambda-esm-cdk-python/example-pattern.json new file mode 100644 index 000000000..070bba13f --- /dev/null +++ b/kinesis-data-stream-lambda-esm-cdk-python/example-pattern.json @@ -0,0 +1,61 @@ +{ + "title": "Amazon Kinesis Data Streams to AWS Lambda with event filtering", + "description": "Process filtered events from a Kinesis data stream with AWS Lambda", + "language": "Python", + "level": "200", + "framework": "CDK", + "introBox": { + "headline": "How it works", + "text": [ + "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.", + "This pattern deploys a Kinesis data stream and Lambda functions that are subscribed to the stream with different event filter configurations." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/kinesis-data-stream-lambda-esm-cdk-python", + "templateURL": "serverless-patterns/kinesis-data-stream-lambda-esm-cdk-python", + "projectFolder": "kinesis-data-stream-lambda-esm-cdk-python", + "templateFile": "cdk/data_stream_processor/kinesis_lambda.py" + } + }, + "resources": { + "bullets": [ + { + "text": "Using AWS Lambda with Amazon Kinesis", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html" + }, + { + "text": "Lambda event source mappings", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html" + }, + { + "text": "Filtering Kinesis events", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html#filtering-kinesis" + } + ] + }, + "deploy": { + "text": [ + "cdk deploy" + ] + }, + "testing": { + "text": [ + "See the GitHub repo for detailed testing instructions." + ] + }, + "cleanup": { + "text": [ + "cdk destroy" + ] + }, + "authors": [ + { + "name": "Edward Schaefer", + "image": "https://d2siip5gg18ho0.cloudfront.net/images/schaeedw-photo-centered_250x250.jpg", + "bio": "Solutions Architect @ Amazon Web Services", + "linkedin": "ejschaefer" + } + ] +} diff --git a/kinesis-data-stream-lambda-esm-cdk-python/kinesis-data-stream-lambda-esm-cdk-python.json b/kinesis-data-stream-lambda-esm-cdk-python/kinesis-data-stream-lambda-esm-cdk-python.json new file mode 100644 index 000000000..6fa9b9c13 --- /dev/null +++ b/kinesis-data-stream-lambda-esm-cdk-python/kinesis-data-stream-lambda-esm-cdk-python.json @@ -0,0 +1,79 @@ +{ + "title": "Amazon Kinesis Data Streams to AWS Lambda with event filtering", + "description": "Process filtered events from a Kinesis data stream with AWS Lambda", + "language": "Python", + "level": "200", + "framework": "CDK", + "introBox": { + "headline": "How it works", + "text": [ + "This pattern shows how to configure Amazon Kinesis as an event source for AWS Lambda and use event filtering to control which records are sent to your function for processing.", + "This pattern deploys a Kinesis data stream and Lambda functions that are subscribed to the stream with different event filter configurations." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/kinesis-data-stream-lambda-esm-cdk-python", + "templateURL": "serverless-patterns/kinesis-data-stream-lambda-esm-cdk-python", + "projectFolder": "kinesis-data-stream-lambda-esm-cdk-python", + "templateFile": "cdk/data_stream_processor/kinesis_lambda.py" + } + }, + "resources": { + "bullets": [ + { + "text": "Using AWS Lambda with Amazon Kinesis", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html" + }, + { + "text": "AWS Lambda event source mappings", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html" + }, + { + "text": "Filtering Amazon Kinesis events", + "link": "https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html" + } + ] + }, + "deploy": { + "text": [ + "cdk deploy" + ] + }, + "testing": { + "text": [ + "See the GitHub repo for detailed testing instructions." + ] + }, + "cleanup": { + "text": [ + "Delete the stack: cdk destroy --all." + ] + }, + "authors": [ + { + "name": "Edward Schaefer", + "image": "https://d2siip5gg18ho0.cloudfront.net/images/schaeedw-photo-centered_250x250.jpg", + "bio": "Solutions Architect @ Amazon Web Services", + "linkedin": "ejschaefer" + } + ], + "patternArch": { + "icon1": { + "x": 20, + "y": 50, + "service": "kinesis-datastreams", + "label": "Amazon Kinesis Data Streams" + }, + "icon2": { + "x": 80, + "y": 50, + "service": "lambda", + "label": "AWS Lambda" + }, + "line1": { + "from": "icon1", + "to": "icon2" + } + } +}