AWSTemplateFormatVersion: '2010-09-09'

Description: "MSK Plaintext Cluster.  Assumes `WorkshopMSKConfig` already created."

Parameters:
  LatestAmiId:
    Type: 'AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>'
    Default: '/aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-gp2'
  KeyName:
    Description: Name of an existing EC2 KeyPair to enable SSH access to the Kafka Client instance
    Type: AWS::EC2::KeyPair::KeyName
    ConstraintDescription: Can contain only ASCII characters.
  SSHLocation:
    Description: The IP address range that can be used to SSH to the EC2 Client instance.  This client instance
                  has Kafka downloaded and CLI scripts available to use with your MSK instance.
    Type: String
    MinLength: '9'
    MaxLength: '18'
    Default: 0.0.0.0/0
    AllowedPattern: "(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})\\.(\\d{1,3})/(\\d{1,2})"
    ConstraintDescription: Must be a valid IP CIDR range of the form x.x.x.x/x
  MSKConfiguration:
    Description: ARN of pre-existing MSK Configuration.  Example shown below.  Change to yours.
    Type: String
    Default: 'arn:aws:kafka:us-east-1:285287800200:configuration/WorkshopMSKConfig/16517401-e308-4c5a-a08c-38ab1f37b4bb-10'
  MSKKafkaVersion:
    Type: String
    Default: 3.6.0
    Description: The Apache Kafka version for the source Amazon MSK cluster.
    AllowedValues:
      - 3.6.0
  SaslScram:
    Description: Enable SASL/SCRAM for authentication
    Default: false
    Type: String
    AllowedValues: [true, false]

Conditions:
  EnableSaslScram: !Equals [!Ref SaslScram, true]

Mappings:
  SubnetConfig:
      VPC:
        CIDR: '10.0.0.0/16'
      PublicOne:
        CIDR: '10.0.0.0/24'
      PrivateSubnetMSKOne:
        CIDR: '10.0.1.0/24'
      PrivateSubnetMSKTwo:
        CIDR: '10.0.2.0/24'
      PrivateSubnetMSKThree:
        CIDR: '10.0.3.0/24'
  RegionAMI:
      us-east-1:
        HVM64: ami-00dc79254d0461090
      us-east-2:
        HVM64: ami-056b1936002ca8ede
      us-west-2:
        HVM64: ami-0a85857bfc5345c38
      eu-west-1:
        HVM64: ami-040ba9174949f6de4
      # us-east-1:
      #   #AMI with kafka, kafkacat, JDK, CLI pre-installed
      #   HVM64: ami-0dcd7e197722e6874
      # us-west-2:
      #   HVM64: ami-0cb72367e98845d43
      # eu-west-1:
      #   HVM64: ami-0aaef8347b162f148
Resources:
  VPC:
    Type: AWS::EC2::VPC
    Properties:
      EnableDnsSupport: true
      EnableDnsHostnames: true
      CidrBlock: !FindInMap ['SubnetConfig', 'VPC', 'CIDR']
      Tags:
        - Key: 'Name'
          Value: 'MSKVPCCF'

  PublicSubnetOne:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PublicOne', 'CIDR']
      MapPublicIpOnLaunch: true
      Tags:
        - Key: 'Name'
          Value: 'MMPublicSubnet'
  PrivateSubnetMSKOne:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PrivateSubnetMSKOne', 'CIDR']
      MapPublicIpOnLaunch: false
      Tags:
        - Key: 'Name'
          Value: 'PrivateSubnetMSKOne'
  PrivateSubnetMSKTwo:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 1
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PrivateSubnetMSKTwo', 'CIDR']
      MapPublicIpOnLaunch: false
      Tags:
        - Key: 'Name'
          Value: 'PrivateSubnetMSKTwo'
  PrivateSubnetMSKThree:
    Type: AWS::EC2::Subnet
    Properties:
      AvailabilityZone:
         Fn::Select:
         - 2
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      VpcId: !Ref 'VPC'
      CidrBlock: !FindInMap ['SubnetConfig', 'PrivateSubnetMSKThree', 'CIDR']
      MapPublicIpOnLaunch: false
      Tags:
        - Key: 'Name'
          Value: 'PrivateSubnetMSKThree'

  InternetGateway:
    Type: AWS::EC2::InternetGateway
  GatewayAttachement:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId: !Ref 'VPC'
      InternetGatewayId: !Ref 'InternetGateway'
  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref 'VPC'
  PublicRoute:
    Type: AWS::EC2::Route
    DependsOn: GatewayAttachement
    Properties:
      RouteTableId: !Ref 'PublicRouteTable'
      DestinationCidrBlock: '0.0.0.0/0'
      GatewayId: !Ref 'InternetGateway'
  PublicSubnetOneRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PublicSubnetOne
      RouteTableId: !Ref PublicRouteTable

  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref 'VPC'

  PrivateSubnetMSKOneRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnetMSKOne
  PrivateSubnetMSKTwoRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnetMSKTwo
  PrivateSubnetMSKThreeRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnetMSKThree

  KafkaClientInstanceSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Enable SSH access via port 22
      VpcId: !Ref 'VPC'
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 22
        ToPort: 22
        CidrIp: !Ref 'SSHLocation'

  MSKSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: MSK Security Group
      VpcId: !Ref 'VPC'
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 2181
        ToPort: 2181
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9094
        ToPort: 9094
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9092
        ToPort: 9092
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9096 #Enable access to port 9096 for SASL/SCRAM
        ToPort: 9096
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      - IpProtocol: tcp
        FromPort: 9098
        ToPort: 9098
        SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId

  KafkaClientEC2Instance:
    Type: AWS::EC2::Instance
    Properties:
      InstanceType: m5.large
      KeyName: !Ref 'KeyName'
      IamInstanceProfile: !Ref EC2InstanceProfile
      AvailabilityZone:
        Fn::Select:
         - 0
         - Fn::GetAZs: {Ref: 'AWS::Region'}
      SubnetId: !Ref 'PublicSubnetOne'
      SecurityGroupIds: [!GetAtt KafkaClientInstanceSecurityGroup.GroupId]
      ImageId: !Ref LatestAmiId
      Tags:
        - Key: 'Name'
          Value: 'KafkaClientInstance'
      BlockDeviceMappings:
        - DeviceName: /dev/xvda
          Ebs:
            VolumeSize: 20
            VolumeType: gp2
            DeleteOnTermination: true
      UserData:
        Fn::Base64:
          !Sub |
          #!/bin/bash
          yum update -y
          yum install python3.7 -y
          yum install java-1.8.0-openjdk-devel -y
          yum install nmap-ncat -y
          yum install git -y
          yum erase awscli -y
          yum install jq -y

          cd /home/ec2-user
          wget https://bootstrap.pypa.io/get-pip.py
          su -c "python3.7 get-pip.py --user" -s /bin/sh ec2-user
          su -c "/home/ec2-user/.local/bin/pip3 install boto3 --user" -s /bin/sh ec2-user
          su -c "/home/ec2-user/.local/bin/pip3 install awscli --user" -s /bin/sh ec2-user

          # install AWS CLI 2 - access with aws2
          curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
          unzip awscliv2.zip
          ./aws/install -b /usr/local/bin/aws2
          su -c "ln -s /usr/local/bin/aws2/aws ~/.local/bin/aws2" -s /bin/sh ec2-user

          cd /home/ec2-user
          su -c "mkdir -p kafka" -s /bin/sh ec2-user
          cd kafka
          su -c "wget https://archive.apache.org/dist/kafka/3.6.0/kafka_2.13-3.6.0.tgz" -s /bin/sh ec2-user
          su -c "tar -xzf kafka_2.13-3.6.0.tgz --strip 1" -s /bin/sh ec2-user

          # Create client.properties configuration file as well as download the corresponding msk iam auth jar library
          cd /home/ec2-user/kafka/libs
          wget https://github.com/aws/aws-msk-iam-auth/releases/download/v2.0.3/aws-msk-iam-auth-2.0.3-all.jar
          cd ..
          cd bin/
          echo 'security.protocol=SASL_SSL' >> client.properties
          echo 'sasl.mechanism=AWS_MSK_IAM' >> client.properties
          echo 'sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule required;' >> client.properties
          echo 'sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler' >> client.properties


  EC2Role:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: "/"
      Policies:
        - PolicyName: MSKClientPermissions
          PolicyDocument: !Sub '{
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Effect": "Allow",
                    "Action": [
                        "kafka-cluster:Connect",
                        "kafka-cluster:AlterCluster",
                        "kafka-cluster:DescribeCluster",
                        "kafka-cluster:DescribeClusterDynamicConfiguration"
                    ],
                    "Resource": [
                        "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:cluster/MSKCluster-${AWS::StackName}/*"
                    ]
                },
                {
                    "Effect": "Allow",
                    "Action": [
                        "kafka-cluster:*Topic*",
                        "kafka-cluster:WriteData",
                        "kafka-cluster:ReadData"
                    ],
                    "Resource": [
                        "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:topic/MSKCluster-${AWS::StackName}/*"
                    ]
                },
                {
                    "Effect": "Allow",
                    "Action": [
                        "kafka-cluster:AlterGroup",
                        "kafka-cluster:DescribeGroup"
                    ],
                    "Resource": [
                        "arn:aws:kafka:${AWS::Region}:${AWS::AccountId}:group/MSKCluster-${AWS::StackName}/*"
                    ]
                }
            ]
        }'
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonMSKFullAccess
        - arn:aws:iam::aws:policy/AWSCloudFormationReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonS3FullAccess
        - arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore
  EC2InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      InstanceProfileName: !Join
                            - '-'
                            - - 'EC2MSKCFProfile'
                              - !Ref 'AWS::StackName'
      Roles:
        - !Ref EC2Role

  # AESSLR:
  #   Type: 'AWS::IAM::ServiceLinkedRole'
  #   Properties:
  #     AWSServiceName: es.amazonaws.com
  #     Description: SLR for AES
  #     CustomSuffix: !Ref 'AWS::StackName'

  MSKCluster:
    Type: AWS::MSK::Cluster
    Properties:
      BrokerNodeGroupInfo:
        ClientSubnets: [!Ref 'PrivateSubnetMSKOne', !Ref 'PrivateSubnetMSKTwo', !Ref 'PrivateSubnetMSKThree']
        InstanceType: express.m7g.large
        SecurityGroups: [!GetAtt MSKSecurityGroup.GroupId]
      ClusterName: !Join
                    - '-'
                    - - 'MSKCluster'
                      - !Ref 'AWS::StackName'
      EncryptionInfo:
        EncryptionInTransit:
          # ClientBroker: TLS
          ClientBroker: !If [EnableSaslScram, TLS, TLS_PLAINTEXT]
          InCluster: true
        # EncryptionAtRest:
        #   DataVolumeKMSKeyId: f75bde7d-f08a-41b6-bdf0-5ecc3a0e1f80
      ClientAuthentication:
        Unauthenticated:
          Enabled: !If [EnableSaslScram, false, true]
        Sasl:
          Scram:
            Enabled: !If [EnableSaslScram, true, false]
          Iam:
            Enabled: True
      EnhancedMonitoring: PER_TOPIC_PER_BROKER
      KafkaVersion: !Ref MSKKafkaVersion
      NumberOfBrokerNodes: 3
      ConfigurationInfo:
        Arn: !Ref 'MSKConfiguration'
        Revision: 1
      OpenMonitoring:
        Prometheus:
          JmxExporter:
            EnabledInBroker: true
          NodeExporter:
            EnabledInBroker: true

Outputs:
  VPCId:
    Description: The ID of the VPC created
    Value: !Ref 'VPC'
  PublicSubnetOne:
    Description: The name of the public subnet created
    Value: !Ref 'PublicSubnetOne'
  PrivateSubnetMSKOne:
    Description: The ID of private subnet one created
    Value: !Ref 'PrivateSubnetMSKOne'
  PrivateSubnetMSKTwo:
    Description: The ID of private subnet two created
    Value: !Ref 'PrivateSubnetMSKTwo'
  PrivateSubnetMSKThree:
    Description: The ID of private subnet three created
    Value: !Ref 'PrivateSubnetMSKThree'
  MSKSecurityGroupID:
    Description: The ID of the security group created for the MSK clusters
    Value: !GetAtt MSKSecurityGroup.GroupId
  KafkaClientEC2InstancePublicDNS:
    Description: The Public DNS for the EC2 instance
    Value: !GetAtt KafkaClientEC2Instance.PublicDnsName
  KafkaClientEC2InstanceSsh:
    Description: The SSH for the EC2 instance
    Value: !Sub ssh -A ec2-user@${KafkaClientEC2Instance.PublicDnsName}
  SchemaRegistryPrivateDNS:
    Description: The Private DNS for the Schema Registry
    Value: !Sub http://${KafkaClientEC2Instance.PrivateDnsName}:8081
  MSKClusterArn:
    Description: The Arn for the MSKMMCluster1 MSK cluster
    Value: !Ref 'MSKCluster'
  KafkaClientEC2InstanceSecurityGroupId:
    Description: The security group id for the EC2 instance
    Value: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
