Amazon MSK Serverless + Lambda + EventBridge PipesでサーバーレスなKafka環境を構築してみた

AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31
Description: "SAM Application with Serverless Kafka"

Resources:
  # VPC
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.10.0.0/16
      EnableDnsHostnames: true
      EnableDnsSupport: true
      Tags:
        - Key: Name
          Value: uehara-msk-vpc

  # パブリックサブネット (AZ: ap-northeast-1a)
  PublicSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.10.0.128/26
      AvailabilityZone: ap-northeast-1a
      MapPublicIpOnLaunch: true
      Tags:
        - Key: Name
          Value: msk-public-subnet-1a

  # プライベートサブネット1 (AZ: ap-northeast-1a)
  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.10.0.0/26
      AvailabilityZone: ap-northeast-1a
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: msk-private-subnet-1a

  # プライベートサブネット2 (AZ: ap-northeast-1c)
  PrivateSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref VPC
      CidrBlock: 10.10.0.64/26
      AvailabilityZone: ap-northeast-1c
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: msk-private-subnet-1c

  # インターネットゲートウェイ
  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
        - Key: Name
          Value: msk-internet-gateway

  # インターネットゲートウェイをVPCにアタッチ
  AttachGateway:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId: !Ref VPC
      InternetGatewayId: !Ref InternetGateway

  # NAT Gateway用のElastic IP
  NATGatewayEIP:
    Type: AWS::EC2::EIP
    DependsOn: AttachGateway
    Properties:
      Domain: vpc
      Tags:
        - Key: Name
          Value: msk-nat-gateway-eip

  # NAT Gateway
  NATGateway:
    Type: AWS::EC2::NatGateway
    Properties:
      AllocationId: !GetAtt NATGatewayEIP.AllocationId
      SubnetId: !Ref PublicSubnet1
      Tags:
        - Key: Name
          Value: msk-nat-gateway

  # パブリックルートテーブル
  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: msk-public-route-table

  # パブリックルート（インターネットゲートウェイへ）
  PublicRoute:
    Type: AWS::EC2::Route
    DependsOn: AttachGateway
    Properties:
      RouteTableId: !Ref PublicRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway

  # パブリックサブネットとルートテーブルの関連付け
  PublicSubnetRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PublicSubnet1
      RouteTableId: !Ref PublicRouteTable

  # プライベートルートテーブル
  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref VPC
      Tags:
        - Key: Name
          Value: msk-private-route-table

  # プライベートルート（NAT Gatewayへ）
  PrivateRoute:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId: !Ref NATGateway

  # プライベートサブネット1とルートテーブルの関連付け
  PrivateSubnet1RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PrivateSubnet1
      RouteTableId: !Ref PrivateRouteTable

  # プライベートサブネット2とルートテーブルの関連付け
  PrivateSubnet2RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PrivateSubnet2
      RouteTableId: !Ref PrivateRouteTable

  # セキュリティグループ
  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Security group for MSK Serverless cluster
      VpcId: !Ref VPC
      SecurityGroupIngress:
        - IpProtocol: tcp
          FromPort: 9098
          ToPort: 9098
          CidrIp: 10.10.0.0/16
          Description: MSK Serverless bootstrap servers (VPC access)
        - IpProtocol: tcp
          FromPort: 9092
          ToPort: 9092
          CidrIp: 10.10.0.0/16
          Description: MSK Kafka protocol (VPC access)
      Tags:
        - Key: Name
          Value: msk-serverless-sg

  # MSK Serverlessクラスター
  MSKServerlessCluster:
    Type: AWS::MSK::ServerlessCluster
    Properties:
      ClusterName: uehara-msk-serverless-cluster
      VpcConfigs:
        - SubnetIds:
            - !Ref PrivateSubnet1
            - !Ref PrivateSubnet2
          SecurityGroups:
            - !Ref MSKSecurityGroup
      ClientAuthentication:
        Sasl:
          Iam:
            Enabled: true

  # Lambda実行ロール
  MSKProcessorExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: msk-processor-execution-role
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
        - arn:aws:iam::aws:policy/AmazonSQSFullAccess

  # MSKイベントソースLambda関数
  MSKProcessorFunction:
    Type: AWS::Serverless::Function
    Properties:
      FunctionName: msk-processor-function
      CodeUri: function/
      Handler: msk_processor.lambda_handler
      Runtime: python3.11
      Architectures:
        - x86_64
      Timeout: 300
      MemorySize: 512
      Role: !GetAtt MSKProcessorExecutionRole.Arn
      Layers:
        - "arn:aws:lambda:ap-northeast-1:336392948345:layer:AWSSDKPandas-Python311:22"
      Environment:
        Variables:
          LOG_LEVEL: INFO
      EventInvokeConfig:
        MaximumRetryAttempts: 2
        MaximumEventAgeInSeconds: 300
      DeadLetterQueue:
        Type: SQS
        TargetArn: !GetAtt DeadLetterQueue.Arn

  # DLQ
  DeadLetterQueue:
    Type: AWS::SQS::Queue
    Properties:
      QueueName: msk-dlq
      MessageRetentionPeriod: 1209600  # 14日

  # Pipe実行ロール
  PipeExecutionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Statement:
          - Effect: Allow
            Principal:
              Service: pipes.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: MSKAccessPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - kafka:DescribeCluster
                  - kafka:DescribeClusterV2
                  - kafka:GetBootstrapBrokers
                  - ec2:CreateNetworkInterface
                  - ec2:DescribeNetworkInterfaces
                  - ec2:DescribeVpcs
                  - ec2:DeleteNetworkInterface
                  - ec2:DescribeSubnets
                  - ec2:DescribeSecurityGroups
                Resource: "*"
              - Effect: Allow
                Action:
                  - kafka-cluster:Connect
                  - kafka-cluster:AlterCluster
                  - kafka-cluster:DescribeCluster
                Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:cluster/uehara-msk-serverless-cluster/*"
              - Effect: Allow
                Action:
                  - kafka-cluster:*Topic*
                  - kafka-cluster:WriteData
                  - kafka-cluster:ReadData
                Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:topic/uehara-msk-serverless-cluster/*"
              - Effect: Allow
                Action:
                  - kafka-cluster:AlterGroup
                  - kafka-cluster:DescribeGroup
                Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:group/uehara-msk-serverless-cluster/*"
        - PolicyName: LambdaInvoke
          PolicyDocument:
            Statement:
              - Effect: Allow
                Action: lambda:InvokeFunction
                Resource: !GetAtt MSKProcessorFunction.Arn

  # EventBridge Pipe
  MSKPipe:
    Type: AWS::Pipes::Pipe
    Properties:
      Name: msk-to-lambda-pipe
      Source: !GetAtt MSKServerlessCluster.Arn
      Target: !GetAtt MSKProcessorFunction.Arn
      RoleArn: !GetAtt PipeExecutionRole.Arn

      SourceParameters:
        ManagedStreamingKafkaParameters:
          TopicName: test-topic
          StartingPosition: LATEST
          BatchSize: 10

      TargetParameters:
        LambdaFunctionParameters:
          InvocationType: FIRE_AND_FORGET

Outputs:
  VpcId:
    Description: ID of the created VPC
    Value: !Ref VPC
    Export:
      Name: !Sub "${AWS::StackName}-VpcId"

  MSKClusterArn:
    Description: ARN of the MSK Serverless cluster
    Value: !GetAtt MSKServerlessCluster.Arn
    Export:
      Name: !Sub "${AWS::StackName}-MSKClusterArn"

  PrivateSubnet1Id:
    Description: ID of Private Subnet 1
    Value: !Ref PrivateSubnet1
    Export:
      Name: !Sub "${AWS::StackName}-PrivateSubnet1"

  PrivateSubnet2Id:
    Description: ID of Private Subnet 2
    Value: !Ref PrivateSubnet2
    Export:
      Name: !Sub "${AWS::StackName}-PrivateSubnet2"

  SecurityGroupId:
    Description: ID of MSK Security Group
    Value: !Ref MSKSecurityGroup
    Export:
      Name: !Sub "${AWS::StackName}-MSKSecurityGroup"

  MSKProcessorFunctionArn:
    Description: ARN of the MSK Processor Lambda function
    Value: !GetAtt MSKProcessorFunction.Arn
    Export:
      Name: !Sub "${AWS::StackName}-MSKProcessorFunctionArn"

  MSKProcessorExecutionRoleArn:
    Description: ARN of the MSK Processor Lambda execution role
    Value: !GetAtt MSKProcessorExecutionRole.Arn
    Export:
      Name: !Sub "${AWS::StackName}-MSKProcessorExecutionRoleArn"

  PublicSubnet1Id:
    Description: ID of Public Subnet 1
    Value: !Ref PublicSubnet1
    Export:
      Name: !Sub "${AWS::StackName}-PublicSubnet1"

  InternetGatewayId:
    Description: ID of Internet Gateway
    Value: !Ref InternetGateway
    Export:
      Name: !Sub "${AWS::StackName}-InternetGateway"

  NATGatewayId:
    Description: ID of NAT Gateway
    Value: !Ref NATGateway
    Export:
      Name: !Sub "${AWS::StackName}-NATGateway"

  NATGatewayEIPId:
    Description: ID of NAT Gateway Elastic IP
    Value: !Ref NATGatewayEIP
    Export:
      Name: !Sub "${AWS::StackName}-NATGatewayEIP"

  PublicRouteTableId:
    Description: ID of Public Route Table
    Value: !Ref PublicRouteTable
    Export:
      Name: !Sub "${AWS::StackName}-PublicRouteTable"

  PrivateRouteTableId:
    Description: ID of Private Route Table
    Value: !Ref PrivateRouteTable
    Export:
      Name: !Sub "${AWS::StackName}-PrivateRouteTable"