AWS Cloud Native Event Streaming with Amazon MSK

AWSTemplateFormatVersion: '2010-09-09'
Description: Provisions an MSK cluster with IAM role
Resources:
  MSKCluster:
    Type: AWS::MSK::Cluster
    Properties:
      ClusterName: MyMSKCluster
      KafkaVersion: '2.8.1'
      NumberOfBrokerNodes: 3
      BrokerNodeGroupInfo:
        InstanceType: kafka.t3.medium
        ClientSubnets:
          - subnet-12345678
          - subnet-87654321
          - subnet-abcdef12
        StorageInfo:
          EbsStorageInfo:
            VolumeSize: 1000
        SecurityGroups:
          - sg-12345678
      EncryptionInfo:
        EncryptionAtRest:
          KmsKeyId: alias/aws/kafka
      Tags:
        - Key: Environment
          Value: production
  MSKRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: MSKAccessRole
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: kafka.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonMSKFullAccess
Outputs:
  ClusterArn:
    Value: !GetAtt MSKCluster.Arn
  RoleArn:
    Value: !GetAtt MSKRole.Arn
                

from aws_cdk import (
    core,
    aws_msk as msk,
    aws_lambda as lambda_,
    aws_iam as iam,
    aws_ec2 as ec2
)

class MSKLambdaStack(core.Stack):
    def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
        super().__init__(scope, id, **kwargs)

        # Define VPC
        vpc = ec2.Vpc(
            self, "MSKVpc",
            max_azs=3,
            cidr="10.0.0.0/16",
            subnet_configuration=[
                ec2.SubnetConfiguration(
                    name="Public",
                    subnet_type=ec2.SubnetType.PUBLIC,
                    cidr_mask=24
                ),
                ec2.SubnetConfiguration(
                    name="Private",
                    subnet_type=ec2.SubnetType.PRIVATE_WITH_NAT,
                    cidr_mask=24
                )
            ]
        )

        # Define MSK cluster
        msk_cluster = msk.CfnCluster(
            self, "MyMSKCluster",
            cluster_name="MyMSKCluster",
            kafka_version="2.8.1",
            number_of_broker_nodes=3,
            broker_node_group_info=msk.CfnCluster.BrokerNodeGroupInfoProperty(
                instance_type="kafka.t3.medium",
                client_subnets=vpc.select_subnets(subnet_type=ec2.SubnetType.PRIVATE_WITH_NAT).subnet_ids,
                storage_info=msk.CfnCluster.StorageInfoProperty(
                    ebs_storage_info=msk.CfnCluster.EBSStorageInfoProperty(
                        volume_size=1000
                    )
                ),
                security_groups=[vpc.vpc_default_security_group]
            ),
            encryption_info=msk.CfnCluster.EncryptionInfoProperty(
                encryption_at_rest=msk.CfnCluster.EncryptionAtRestProperty(
                    kms_key_id="alias/aws/kafka"
                )
            ),
            tags={"Environment": "production"}
        )

        # Define Lambda function to consume MSK events
        msk_lambda = lambda_.Function(
            self, "MSKConsumerLambda",
            runtime=lambda_.Runtime.PYTHON_3_8,
            handler="lambda_function.lambda_handler",
            code=lambda_.Code.from_inline("""
import json

def lambda_handler(event, context):
    # Process MSK events
    for record in event['records']:
        print("Received MSK event:", json.loads(record['value']))
    return {"statusCode": 200}
"""),
            vpc=vpc,
            environment={
                "MSK_CLUSTER_ARN": msk_cluster.ref
            },
            timeout=core.Duration.seconds(30)
        )

        # Grant Lambda access to MSK
        msk_lambda.add_to_role_policy(
            iam.PolicyStatement(
                effect=iam.Effect.ALLOW,
                actions=[
                    "kafka:DescribeCluster",
                    "kafka:GetBootstrapBrokers",
                    "kafka:DescribeClusterV2"
                ],
                resources=[msk_cluster.ref]
            )
        )

app = core.App()
MSKLambdaStack(app, "MSKLambdaStack")
app.synth()
                

provider "aws" {
  region = "us-west-2"
}

resource "aws_vpc" "msk_vpc" {
  cidr_block = "10.0.0.0/16"
  tags = {
    Environment = "production"
  }
}

resource "aws_subnet" "msk_subnet_a" {
  vpc_id     = aws_vpc.msk_vpc.id
  cidr_block = "10.0.1.0/24"
  availability_zone = "us-west-2a"
  tags = {
    Environment = "production"
  }
}

resource "aws_subnet" "msk_subnet_b" {
  vpc_id     = aws_vpc.msk_vpc.id
  cidr_block = "10.0.2.0/24"
  availability_zone = "us-west-2b"
  tags = {
    Environment = "production"
  }
}

resource "aws_security_group" "msk_sg" {
  vpc_id = aws_vpc.msk_vpc.id
  ingress {
    from_port   = 9092
    to_port     = 9092
    protocol    = "tcp"
    cidr_blocks = ["10.0.0.0/16"]
  }
  tags = {
    Environment = "production"
  }
}

resource "aws_msk_cluster" "msk_cluster" {
  cluster_name = "my-msk-cluster"
  kafka_version = "2.8.1"
  number_of_broker_nodes = 3
  broker_node_group_info {
    instance_type   = "kafka.t3.medium"
    ebs_volume_size = 1000
    client_subnets  = [aws_subnet.msk_subnet_a.id, aws_subnet.msk_subnet_b.id]
    security_groups = [aws_security_group.msk_sg.id]
  }
  encryption_info {
    encryption_at_rest_kms_key_arn = "arn:aws:kms:us-west-2:aws:kafka"
  }
  tags = {
    Environment = "production"
  }
}

resource "aws_s3_bucket" "msk_sink" {
  bucket = "my-msk-sink-bucket-123"
  tags = {
    Environment = "production"
  }
}

resource "aws_iam_role" "msk_connect_role" {
  name = "msk-connect-role"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "kafkaconnect.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "msk_connect_policy" {
  name = "msk-connect-policy"
  role = aws_iam_role.msk_connect_role.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:PutObject",
          "s3:GetObject",
          "s3:ListBucket"
        ]
        Resource = [
          aws_s3_bucket.msk_sink.arn,
          "${aws_s3_bucket.msk_sink.arn}/*"
        ]
      },
      {
        Effect = "Allow"
        Action = [
          "kafka:DescribeCluster",
          "kafka:GetBootstrapBrokers"
        ]
        Resource = aws_msk_cluster.msk_cluster.arn
      }
    ]
  })
}