kinesis.yaml

AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31
Description: Example project demoing Kinesis Data Streams processing using the Batch Processing Utility in Powertools for AWS Lambda (.NET)

Globals:
  Function:
    Timeout: 20
    Runtime: dotnet8
    MemorySize: 1024
    Environment:
      Variables:
        POWERTOOLS_SERVICE_NAME: powertools-dotnet-sample-batch-kinesis
        POWERTOOLS_LOG_LEVEL: Debug
        POWERTOOLS_LOGGER_CASE: PascalCase
        POWERTOOLS_BATCH_ERROR_HANDLING_POLICY: DeriveFromEvent
        POWERTOOLS_BATCH_MAX_DEGREE_OF_PARALLELISM: 1
        POWERTOOLS_BATCH_PARALLEL_ENABLED : false
        POWERTOOLS_BATCH_THROW_ON_FULL_BATCH_FAILURE: true

Resources:

  # --------------
  # KMS key for encrypted messages / records
  CustomerKey:
    Type: AWS::KMS::Key
    Properties:
      Description: KMS key for encrypted queues
      Enabled: true
      KeyPolicy:
        Version: "2012-10-17"
        Statement:
          - Sid: Enable IAM User Permissions
            Effect: Allow
            Principal:
              AWS: !Sub "arn:aws:iam::${AWS::AccountId}:root"
            Action: "kms:*"
            Resource: "*"
          - Sid: Allow AWS Lambda to use the key
            Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action:
              - kms:Decrypt
              - kms:GenerateDataKey
            Resource: "*"

  CustomerKeyAlias:
    Type: AWS::KMS::Alias
    Properties:
      AliasName: !Sub alias/${AWS::StackName}-kms-key
      TargetKeyId: !Ref CustomerKey

  # --------------
  # Batch Processing for Kinesis Data Stream
  KinesisStreamDeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      KmsMasterKeyId: !Ref CustomerKey

  KinesisStream:
    Type: AWS::Kinesis::Stream
    Properties:
      ShardCount: 1
      StreamEncryption:
        EncryptionType: KMS
        KeyId: !Ref CustomerKey

  KinesisStreamConsumer:
    Type: AWS::Kinesis::StreamConsumer
    Properties:
      ConsumerName: powertools-dotnet-sample-batch-kds-consumer
      StreamARN: !GetAtt KinesisStream.Arn

  KinesisBatchProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      Policies:
        - Statement:
            - Sid: KinesisStreamConsumerPermissions
              Effect: Allow
              Action:
                - kinesis:DescribeStreamConsumer
              Resource:
                - !GetAtt KinesisStreamConsumer.ConsumerARN
            - Sid: DlqPermissions
              Effect: Allow
              Action:
                - sqs:SendMessage
                - sqs:SendMessageBatch
              Resource: !GetAtt KinesisStreamDeadLetterQueue.Arn
            - Sid: KmsKeyPermissions
              Effect: Allow
              Action:
                - kms:Decrypt
                - kms:GenerateDataKey
              Resource: !GetAtt CustomerKey.Arn
      CodeUri: ./src/HelloWorld/
      Handler: HelloWorld::HelloWorld.Function::KinesisEventHandlerUsingAttribute
      Events:
        Kinesis:
          Type: Kinesis
          Properties:
            BatchSize: 5
            BisectBatchOnFunctionError: true
            DestinationConfig:
              OnFailure:
                Destination: !GetAtt KinesisStreamDeadLetterQueue.Arn
            Enabled: true
            FunctionResponseTypes:
              - ReportBatchItemFailures
            MaximumRetryAttempts: 2
            ParallelizationFactor: 1
            StartingPosition: LATEST
            Stream: !GetAtt KinesisStreamConsumer.ConsumerARN

  KinesisBatchProcessorFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub "/aws/lambda/${KinesisBatchProcessorFunction}"
      RetentionInDays: 7

Outputs:
  KinesisStreamArn:
    Description: "Kinesis Stream ARN"
    Value: !GetAtt KinesisStream.Arn