AWSTemplateFormatVersion: '2010-09-09'
Parameters:
  KeyName:
    Description: Name of an existing EC2 KeyPair to enable SSH access to the instance
    Type: AWS::EC2::KeyPair::KeyName
    ConstraintDescription: Can contain only ASCII characters.

Mappings:
  RegionAMI:
      us-east-1:
        HVM64: ami-00dc79254d0461090
      us-west-2:
        HVM64: ami-0a85857bfc5345c38
      eu-west-1:
        HVM64: ami-040ba9174949f6de4
      # us-east-1:
      #   HVM64: ami-00dc79254d0461090
      # us-west-2:
      #   HVM64: ami-0a85857bfc5345c38
      # eu-west-1:
      #   HVM64: ami-040ba9174949f6de4

Resources:
  MSKVPCStack:
    Type: AWS::CloudFormation::Stack
    Properties:
      TemplateURL: "https://aws-streaming-artifacts.s3.amazonaws.com/msk-lab-resources/MSKLabs/MSKPrivateVPCOnly.yml"

  KafkaClientInstanceSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    Properties:
      GroupDescription: Enable SSH access via port 22 from BastionHostSecurityGroup
      VpcId: !GetAtt MSKVPCStack.Outputs.VPCId
      SecurityGroupIngress:
      - IpProtocol: tcp
        FromPort: 22
        ToPort: 22
        CidrIp: 10.0.0.0/24

  Cloud9EC2Bastion:
    Type: AWS::Cloud9::EnvironmentEC2
    Properties: 
      AutomaticStopTimeMinutes: 600
      Description: "Cloud9 EC2 environment"
      InstanceType: m5.large
      Name: !Sub "${AWS::StackName}-Cloud9EC2Bastion"
      SubnetId: !GetAtt MSKVPCStack.Outputs.PublicSubnetOne
      Tags: 
        - Key: 'Purpose'
          Value: 'Cloud9EC2BastionHostInstance'

  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: !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKOne
      SecurityGroupIds: [!GetAtt KafkaClientInstanceSecurityGroup.GroupId]
      ImageId: !FindInMap ['RegionAMI', !Ref 'AWS::Region', 'HVM64']
      Tags:
        - Key: 'Name'
          Value: 'KafkaClientInstance'
      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
            amazon-linux-extras install docker -y
            service docker start
            usermod -a -G docker ec2-user

            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
            su -c "/home/ec2-user/.local/bin/pip3 install kafka-python --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

            # Create dirs, get Apache Kafka 2.3.1, 2.4.0 and unpack it
            su -c "mkdir -p kafka231 kafka241 confluent" -s /bin/sh ec2-user
            cd kafka231
            su -c "wget https://archive.apache.org/dist/kafka/2.3.1/kafka_2.12-2.3.1.tgz" -s /bin/sh ec2-user
            su -c "tar -xzf kafka_2.12-2.3.1.tgz --strip 1" -s /bin/sh ec2-user

            cd /home/ec2-user
            ln -s /home/ec2-user/kafka241 /home/ec2-user/kafka
            cd kafka241
            su -c "wget http://archive.apache.org/dist/kafka/2.4.1/kafka_2.12-2.4.1.tgz" -s /bin/sh ec2-user
            su -c "tar -xzf kafka_2.12-2.4.1.tgz --strip 1" -s /bin/sh ec2-user

            # Get Confluent Community and unpack it
            cd /home/ec2-user
            cd confluent
            su -c "wget http://packages.confluent.io/archive/5.4/confluent-community-5.4.1-2.12.tar.gz" -s /bin/sh ec2-user
            su -c "tar -xzf confluent-community-5.4.1-2.12.tar.gz --strip 1" -s /bin/sh ec2-user
            
            # Initialize the Kafka cert trust store
            su -c 'find /usr/lib/jvm/ -name "cacerts" -exec cp {} /tmp/kafka.client.truststore.jks \;' -s /bin/sh ec2-user

            cd /tmp
            su -c "mkdir -p kafka" -s /bin/sh ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/producer.properties_msk /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/KafkaClickstreamClient-1.0-SNAPSHOT.jar/KafkaClickstreamClient-1.0-SNAPSHOT.jar /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/MSKLabs/schema-registry-ssl/schema-registry.properties /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/generatePropertiesFiles.py /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/generateStartupFile.py /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/AuthMSK-1.0-SNAPSHOT.jar /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/connect-distributed.properties /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://aws-streaming-artifacts/msk-lab-resources/kafka-consumer-python.py /tmp/kafka" -l ec2-user

            # Setup unit in systemd for Schema Registry
            echo -n "
            [Unit]
            Description=Confluent Schema Registry
            After=network.target

            [Service]
            Type=simple
            User=ec2-user
            ExecStart=/bin/sh -c '/home/ec2-user/confluent/bin/schema-registry-start /tmp/kafka/schema-registry.properties > /tmp/kafka/schema-registry.log 2>&1'
            ExecStop=/home/ec2-user/confluent/bin/schema-registry-stop
            Restart=on-abnormal

            [Install]
            WantedBy=multi-user.target" > /etc/systemd/system/confluent-schema-registry.service

            # Setup unit in systemd for Kafka Connect
            echo -n "
            [Unit]
            Description=Kafka Connect
            After=network.target

            [Service]
            Type=simple
            User=ec2-user
            ExecStart=/bin/sh -c '/home/ec2-user/kafka/bin/connect-distributed.sh /tmp/kafka/connect-distributed.properties > /tmp/kafka/kafka-connect.log 2>&1'
            Restart=on-abnormal

            [Install]
            WantedBy=multi-user.target" > /etc/systemd/system/kafka-connect.service

            #setup Amazon MSK environment

            su -c "echo 'export region=${AWS::Region}' >> /tmp/kafka/setup_env" -s /bin/sh ec2-user
            su -c "echo 'export stackname=\$1' >> /tmp/kafka/setup_env" -s /bin/sh ec2-user
            su -c "echo 'export mskclusterarn=\`aws cloudformation describe-stacks --stack-name \$stackname --region \$region|jq '\''.Stacks[0].Outputs[3].OutputValue'\''|cut -d \"\\\"\" -f 2\`' >> /tmp/kafka/setup_env" -s /bin/sh ec2-user
            su -c "echo 'export zoo=\`aws kafka describe-cluster --cluster-arn \$mskclusterarn --region \$region|jq '\''.ClusterInfo.ZookeeperConnectString'\''|cut -d \"\\\"\" -f 2\`' >> /tmp/kafka/setup_env" -s /bin/sh ec2-user
            su -c "echo 'export brokers=\`aws kafka get-bootstrap-brokers --cluster-arn \$mskclusterarn --region \$region|jq '\''.BootstrapBrokerString'\''|cut -d \"\\\"\" -f 2\`' >> /tmp/kafka/setup_env" -s /bin/sh ec2-user
            su -c "echo 'export brokerstls=\`aws kafka get-bootstrap-brokers --cluster-arn \$mskclusterarn --region \$region|jq '\''.BootstrapBrokerStringTls'\''|cut -d \"\\\"\" -f 2\`' >> /tmp/kafka/setup_env" -s /bin/sh ec2-user
            chmod +x /tmp/kafka/setup_env

  EC2Role: 
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument: 
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: "/"
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonMSKFullAccess
        - arn:aws:iam::aws:policy/AWSCloudFormationReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonS3FullAccess
        - arn:aws:iam::aws:policy/AWSCertificateManagerPrivateCAFullAccess
  EC2InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties: 
      InstanceProfileName: !Join
                            - '-'
                            - - 'EC2MMMSKCFProfile'
                              - !Ref 'AWS::StackName'
      Roles:
        - !Ref EC2Role

Outputs:
  VPCId: 
    Description: The ID of the VPC created
    Value: !GetAtt MSKVPCStack.Outputs.VPCId
  PublicSubnetOne: 
    Description: The name of the public subnet created
    Value: !GetAtt MSKVPCStack.Outputs.PublicSubnetOne
  PrivateSubnetMSKOne: 
    Description: The ID of private subnet one created
    Value: !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKOne
  PrivateSubnetMSKTwo: 
    Description: The ID of private subnet two created
    Value: !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKTwo
  PrivateSubnetMSKThree: 
    Description: The ID of private subnet three created
    Value: !GetAtt MSKVPCStack.Outputs.PrivateSubnetMSKThree
  KafkaClientEC2InstancePrivateDNS:
    Description: The Public DNS for the EC2 instance
    Value: !GetAtt KafkaClientEC2Instance.PrivateDnsName
  SSHKafkaClientEC2Instance:
    Description: SSH command for Kafka the EC2 instance
    Value: !Sub ssh -A ec2-user@${KafkaClientEC2Instance.PrivateDnsName}
    Export:
      Name: !Sub "${AWS::StackName}-SSHKafkaClientEC2Instance"
  KafkaClientEC2InstanceSecurityGroupId:
    Description: The security group id for the EC2 instance
    Value: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
    Export:
      Name: !Sub "${AWS::StackName}-KafkaClientEC2InstanceSecurityGroupId"
  SchemaRegistryUrl:
    Description: The url for the Schema Registry
    Value: !Sub http://${KafkaClientEC2Instance.PrivateDnsName}:8081
    Export:
      Name: !Sub "${AWS::StackName}-SchemaRegistryUrl"
  VPCStackName:
    Description: The name of the VPC Stack
    Value: !GetAtt MSKVPCStack.Outputs.VPCStackName
