AWSTemplateFormatVersion: 2010-09-09
Description: Create a MSK VPC with primary MSK Cluster (Express broker)
Parameters:
  LatestAmiId:
    Type:  AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
    Default: /aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-gp2

Resources:
  MSKVPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 172.16.0.0/16
      EnableDnsHostnames: true
      EnableDnsSupport: true
      Tags:
        - Key: Name
          Value: MSKVPC

  MSKVPCPublicSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref MSKVPC
      AvailabilityZone:
        Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: "AWS::Region"}
      CidrBlock: 172.16.0.0/24
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Join ["-", ["MSKVPC","public-subnet"]]
  
  MSKVPCPrivateSubnet1:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref MSKVPC
      AvailabilityZone:
        Fn::Select:
          - 0
          - Fn::GetAZs: {Ref: "AWS::Region"}
      CidrBlock: 172.16.1.0/24
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Join ["-", ["MSKVPC","private-subnet-a"]]
  
  MSKVPCPrivateSubnet2:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref MSKVPC
      AvailabilityZone:
        Fn::Select:
          - 1
          - Fn::GetAZs: {Ref: "AWS::Region"}
      CidrBlock: 172.16.2.0/24
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Join ["-", ["MSKVPC","private-subnet-b"]]
  
  MSKVPCPrivateSubnet3:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref MSKVPC
      AvailabilityZone:
        Fn::Select:
          - 2
          - Fn::GetAZs: {Ref: "AWS::Region"}
      CidrBlock: 172.16.3.0/24
      MapPublicIpOnLaunch: false
      Tags:
        - Key: Name
          Value: !Join ["-", ["MSKVPC","private-subnet-c"]]
  
  MSKVPCInternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags:
        - Key: Name
          Value: !Join ["-", ["MSKVPC","gateway"]]
  
  MSKVPCGatewayToInternet:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId: !Ref MSKVPC
      InternetGatewayId: !Ref MSKVPCInternetGateway
  
  MSKVPCPublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref MSKVPC
  
  MSKVPCPublicRoute:
    Type: AWS::EC2::Route
    DependsOn: MSKVPCGatewayToInternet
    Properties:
      RouteTableId: !Ref MSKVPCPublicRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref MSKVPCInternetGateway
  
  MSKVPCPublicSubnet1RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref MSKVPCPublicSubnet1
      RouteTableId: !Ref MSKVPCPublicRouteTable
  
  MSKVPCNatGateway:
    Type: AWS::EC2::NatGateway
    DependsOn: MSKVPCNatPublicIP
    Properties:
      AllocationId: !GetAtt MSKVPCNatPublicIP.AllocationId
      SubnetId: !Ref MSKVPCPublicSubnet1
  
  MSKVPCNatPublicIP:
    Type: AWS::EC2::EIP
    DependsOn: MSKVPC
    Properties:
      Domain: vpc
 
  MSKVPCPrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref MSKVPC

  MSKVPCPrivateRoute:
    Type: AWS::EC2::Route
    Properties:
      RouteTableId: !Ref MSKVPCPrivateRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId: !Ref MSKVPCNatGateway
  
  MSKVPCPrivateSubnet1RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref MSKVPCPrivateSubnet1
      RouteTableId: !Ref MSKVPCPrivateRouteTable
  
  MSKVPCPrivateSubnet2RouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref MSKVPCPrivateSubnet2
      RouteTableId: !Ref MSKVPCPrivateRouteTable
 
  KafkaClientInstanceSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: EC2 Security Group
      VpcId: !Ref MSKVPC
      SecurityGroupEgress:
      - CidrIp: 0.0.0.0/0
        IpProtocol: -1
        Description: "Allow Outbound To Any IP"

  MSKReplicatorIngress:
    Type: AWS::EC2::SecurityGroupIngress 
    Properties:
      GroupId: !GetAtt MSKSecurityGroup.GroupId
      IpProtocol: tcp
      FromPort: 9098
      ToPort: 9098
      SourceSecurityGroupId: !GetAtt MSKSecurityGroup.GroupId
      Description: "Allow Replicator to talk to MSK"

  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK Security Group
      VpcId: !Ref MSKVPC
      SecurityGroupIngress:
      - IpProtocol: tcp
        Description: "Zookeeper access from Kafka Client Instance"
        FromPort: 2181
        ToPort: 2181
        SourceSecurityGroupId: !Ref KafkaClientInstanceSecurityGroup
      - IpProtocol: tcp
        Description: "MSK access from Kafka Client Instance while using IAM"
        FromPort: 9098
        ToPort: 9098
        SourceSecurityGroupId: !Ref KafkaClientInstanceSecurityGroup
      SecurityGroupEgress:
      - CidrIp: 172.16.0.0/16
        IpProtocol: -1
        Description: "Allow Outbound To all VPC address"

  MSKPrimaryCluster:
    Type: AWS::MSK::Cluster
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets:
          - !Ref MSKVPCPrivateSubnet1
          - !Ref MSKVPCPrivateSubnet2
          - !Ref MSKVPCPrivateSubnet3
        InstanceType: express.m7g.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
      ClusterName: !Join
                    - "-"
                    - - !Ref "AWS::StackName"
                      - "msk-cluster"
      EncryptionInfo:
        EncryptionInTransit:
          ClientBroker: TLS
          InCluster: true
      ClientAuthentication:
        Sasl:
          Iam:
            Enabled: true
      EnhancedMonitoring: DEFAULT
      KafkaVersion: 3.6.0
      NumberOfBrokerNodes: 3

  KafkaClientEC2Instance:
    Type: AWS::EC2::Instance
    Properties:
      InstanceType: t3.large
      EbsOptimized: true
      Monitoring: true
      IamInstanceProfile: !Ref EC2InstanceProfile
      AvailabilityZone: 
        Fn::Select:
        - 0
        - Fn::GetAZs: {Ref: "AWS::Region"}
      SubnetId: !Ref MSKVPCPrivateSubnet1
      SecurityGroupIds: [!GetAtt KafkaClientInstanceSecurityGroup.GroupId]
      ImageId: !Ref LatestAmiId
      Tags:
        - Key: "Name"
          Value: !Sub ${AWS::StackName}-KafkaClientInstance1
      UserData: 
        Fn::Base64: 
          !Sub |
            #!/bin/bash
            yum update -y
            yum install java-11-amazon-corretto -y
            yum install jq -y

            cd  /home/ec2-user
            
            # Create dirs, get Apache Kafka 3.6.0
            su -c "mkdir -p kafka360" -s /bin/sh ec2-user
            ln -s /home/ec2-user/kafka360 /home/ec2-user/kafka
            cd kafka
            su -c "wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.12-3.6.0.tgz" -s /bin/sh ec2-user
            su -c "tar -xzf kafka_2.12-3.6.0.tgz --strip 1" -s /bin/sh ec2-user

            # Add CloudFormation Outputs Script
            cat <<'EOF' > /home/ec2-user/cloudformation_outputs.sh
            #Create CloudFormation Outputs
            stack_details=$(aws cloudformation --region ${AWS::Region} describe-stacks --stack-name ${AWS::StackName} --output json)
            read -r -a output_keys <<< $(echo "$stack_details" | jq ".Stacks[].Outputs[].OutputKey")
            read -r -a output_vals <<< $(echo "$stack_details" | jq ".Stacks[].Outputs[].OutputValue")
            for ((i=0;i<${!#output_keys[@]};++i)); do
              key=$(echo "${!output_keys[i]}" | sed -e 's/^"//'  -e 's/"$//')
              val=$(echo "${!output_vals[i]}" | sed -e 's/^"//'  -e 's/"$//')
              export "$key"="$val"
            done
            EOF
            chown ec2-user:ec2-user /home/ec2-user/cloudformation_outputs.sh
            chmod 770 /home/ec2-user/cloudformation_outputs.sh

            # Download Kafka IAM library
            cd /home/ec2-user/kafka/libs
            su -c "wget https://github.com/aws/aws-msk-iam-auth/releases/download/v1.1.1/aws-msk-iam-auth-1.1.1-all.jar" -s /bin/sh ec2-user

  KafkaClientEC2InstancePolicy:
    Type: AWS::IAM::Policy
    Properties:
      PolicyName: KafkaClientEC2InstancePolicy
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action:
              - kafka-cluster:Connect
              - kafka-cluster:AlterCluster
              - kafka-cluster:DescribeCluster
              - kafka-cluster:DescribeClusterDynamicConfiguration
              - kafka:DescribeCluster
            Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:cluster/*"
          - Effect: Allow
            Action:
              - kafka-cluster:*Topic*
              - kafka-cluster:WriteData
              - kafka-cluster:ReadData
            Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:topic/*"
          - Effect: Allow
            Action:
              - kafka-cluster:AlterGroup
              - kafka-cluster:DescribeGroup
            Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:group/*"
          - Effect: Allow
            Action:
              - kafka:GetBootstrapBrokers
            Resource: !Sub "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:cluster/*"
          - Effect: Allow
            Action:
              - cloudformation:DeleteStack
            Resource: !Sub "arn:aws:cloudformation:${AWS::Region}:${AWS::AccountId}:stack/${AWS::StackName}/*"
      Roles:
        - !Ref EC2Role

  EC2InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties: 
      InstanceProfileName: !Join
                            - "-"
                            - - "EC2MMMSKCFProfile"
                              - !Ref "AWS::StackName"
      Roles:
        - !Ref EC2Role

  EC2Role: 
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument: 
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: "sts:AssumeRole"
      Path: "/"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore

Outputs:
  MSKVPCId:
    Description: VPC in which Amazon MSK will be deployed
    Value: !Ref MSKVPC
  MSKPrimaryClusterArn:
    Description: The Arn for the Source MSK cluster
    Value: !Ref MSKPrimaryCluster
  KafkaClientEC2InstanceSecurityGroupId:
    Description: The security group id for the EC2 instance
    Value: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
  MSKClusterSecurityGroupId:
    Description: The security group id for the Amazon MSK cluster
    Value: !GetAtt MSKSecurityGroup.GroupId
  KafkaClientEC2InstanceId:
    Description: Kafka client EC2 instance ID
    Value: !Ref KafkaClientEC2Instance
  REGION:
    Description: AWS Region resources are deployed in
    Value: !Ref "AWS::Region"
  MSKVPCPrivateSubnet1:
    Description: Private subnet 1 Id
    Value: !Ref MSKVPCPrivateSubnet1
  MSKVPCPrivateSubnet2:
    Description: Private subnet 2 Id
    Value: !Ref MSKVPCPrivateSubnet2
  MSKVPCPrivateSubnet3:
    Description: Private subnet 3 Id
    Value: !Ref MSKVPCPrivateSubnet3    
  StackName:
    Description: Name of CloudFormation Stack
    Value: !Ref AWS::StackName