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.
  LatestAmiId:
    Type: 'AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>'
    Default: '/aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-gp2'

Resources:
  MSKVPCStack:
    Type: AWS::CloudFormation::Stack
    Properties:
      TemplateURL: "https://reinvent2019-msk-liftandshift.s3.amazonaws.com/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

  KafkaClientInstanceSecurityGroup8081:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn: KafkaClientInstanceSecurityGroup
    Properties:
      Description: Enable access to Schema Registry inside the KafkaClientInstanceSecurityGroup
      GroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId
      IpProtocol: tcp
      FromPort: 8081
      ToPort: 8081
      SourceSecurityGroupId: !GetAtt KafkaClientInstanceSecurityGroup.GroupId

  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: !Ref LatestAmiId
      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
            yum install maven -y
            mvn -version
            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://mirrors.gigenet.com/apache/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://reinvent2019-msk-liftandshift/producer.properties_msk /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://reinvent2019-msk-liftandshift/consumer.properties /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://reinvent2019-msk-liftandshift/schema-registry.properties /tmp/kafka" -l ec2-user
            su -c "aws s3 cp s3://reinvent2019-msk-liftandshift/setup-env-sasl.py /tmp/kafka" -l ec2-user
            su -c "git -C /tmp/kafka clone https://github.com/aws-samples/sasl-scram-secrets-manager-client-for-msk.git" -l ec2-user
            su -c "cd /tmp/kafka/sasl-scram-secrets-manager-client-for-msk/ && mvn clean install -f pom.xml && cp target/SaslScramSecretsManagerClient-1.0-SNAPSHOT.jar /tmp/kafka" -l ec2-user
            su -c "cd /tmp/kafka && rm -rf sasl-scram-secrets-manager-client-for-msk" -l ec2-user
            su -c "git -C /tmp/kafka clone https://github.com/aws-samples/clickstream-producer-for-apache-kafka.git" -l ec2-user
            su -c "cd /tmp/kafka/clickstream-producer-for-apache-kafka/ && mvn clean package -f pom.xml && cp target/KafkaClickstreamClient-1.0-SNAPSHOT.jar /tmp/kafka" -l ec2-user
            su -c "cd /tmp/kafka && rm -rf clickstream-producer-for-apache-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 bash env
            su -c "echo 'export PS1=\"KafkaClientEC2Instance [\u@\h \W\\]$ \"' >> /home/ec2-user/.bash_profile" -s /bin/sh ec2-user
            su -c "echo '[ -f /tmp/kafka/setup_env ] && . /tmp/kafka/setup_env' >> /home/ec2-user/.bash_profile" -s /bin/sh ec2-user

            #setup aws Region
            su -c "mkdir -p /home/ec2-user/.aws" -s /bin/sh ec2-user
            su -c "cat > /home/ec2-user/.aws/config<<EOF
            [default]
            region = ${AWS::Region}
            EOF" -s /bin/sh ec2-user

  EC2Role: 
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument: 
        Version: 2012-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: "/"
      
  EC2InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties: 
      Roles:
        - !Ref EC2Role

  KafkaClientEC2InstancePolicy:
    Type: 'AWS::IAM::Policy'
    Properties:
      PolicyName: KafkaClientEC2InstancePolicy
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Action:
              - lambda:CreateFunction
              - s3:CreateBucket
              - iam:CreateRole
              - cloudformation:DescribeStackResource
              - s3:ListBucket
              - kafka:GetBootstrapBrokers
              - iam:AttachRolePolicy
              - cloudformation:CreateChangeSet
              - lambda:CreateEventSourceMapping
              - iam:PutRolePolicy
              - firehose:DeleteDeliveryStream
              - iam:PassRole
              - cloudformation:DescribeStackEvents
              - iam:DetachRolePolicy
              - ec2:DescribeNetworkInterfaces
              - iam:DeleteRolePolicy
              - cloudformation:UpdateStack
              - lambda:DeleteFunction
              - cloudformation:DescribeChangeSet
              - s3:DeleteObject
              - cloudformation:ExecuteChangeSet
              - s3:DeleteBucket
              - s3:PutBucketVersioning
              - iam:GetRole
              - logs:DeleteLogGroup
              - lambda:GetEventSourceMapping
              - lambda:ListEventSourceMappings
              - lambda:GetFunction
              - firehose:CreateDeliveryStream
              - kafka:DescribeCluster
              - kafka:BatchAssociateScramSecret
              - kafka:BatchDisassociateScramSecret
              - iam:DeleteRole
              - cloudformation:GetTemplateSummary
              - cloudformation:ListStackResources
              - ec2:DescribeSecurityGroups
              - cloudformation:DescribeStacks
              - firehose:DescribeDeliveryStream 
              - iam:CreatePolicy
              - s3:PutObject
              - s3:GetObject
              - ec2:DescribeVpcs
              - cloudformation:DeleteStack
              - ec2:DescribeSubnets
              - lambda:DeleteEventSourceMapping
              - iam:GetRolePolicy
              - kms:CreateKey
              - kms:ListKeys
              - kms:ListAliases
              - secretsmanager:PutResourcePolicy
              - secretsmanager:DeleteResourcePolicy
              - secretsmanager:CreateSecret
              - secretsmanager:DeleteSecret
              - secretsmanager:ListSecrets
              - glue:GetSchemaByDefinition
              - glue:CreateSchema
              - glue:RegisterSchemaVersion
              - glue:PutSchemaVersionMetadata
              - glue:GetSchemaVersion
              - ec2:CreateNetworkInterface
              - ec2:DeleteNetworkInterface
            Resource: '*'
          - Effect: Allow
            Action:
              - kms:CreateAlias
              - kms:ListGrants
            Resource:
              - !Sub arn:aws:kms:*:${AWS::AccountId}:alias/*
              - !Sub arn:aws:kms:*:${AWS::AccountId}:key/*
      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