---

AWSTemplateFormatVersion: '2010-09-09'
Transform: 'AWS::Serverless-2016-10-31'
Description: ''
Metadata:
  'AWS::CloudFormation::Interface':
    ParameterGroups:
      - Label:
          default: 'Hive Metastore Host Number'
        Parameters:
          - HiveMetastoreHostID
    ParameterLabels:
      HiveMetastoreHostID:
        default: Choose the Hive Metastore host number/id to use

Parameters:
  HiveMetastoreHostID:
    Description: 'Select the Hive Metastore host for multi master setup'
    Type: Number
    Default: 1
    AllowedValues:
      - 1
      - 2
      - 3

Mappings:
  VpcConfigurations:
    cidr:
      vpc: 172.32.0.0/16
      publicSubnet1: 172.32.0.0/24
      privateSubnet1: 172.32.11.0/24
      privateSubnet2: 172.32.12.0/24
  ClusterConfigurations:
    emr:
      masterInstanceType: m5.xlarge
      coreInstanceType: m5.xlarge
      masterInstanceCount: 3
      coreInstanceCount: 2
      emrReleaseVersion: emr-5.30.1
      artifactsBucketName: aws-bigdata-blog
      bootStrapScriptPath: artifacts/Access-and-Visualize-External-Hive-Tables-with-Athena-and-QuickSight/emr_bootstrap.sh
      stepActionScriptPath: artifacts/Access-and-Visualize-External-Hive-Tables-with-Athena-and-QuickSight/hive-createtables.py
    rds:
      dbName: hivemetastore
      dbInstanceType: db.m5.large
      dbEngine: mysql
      dbEngineVersion: 5.7.22
      dbPort: 3306
      dbDriveName: org.mariadb.jdbc.Driver
      dbStorageGb: 20
    hiveMetastoreFunction:
      functionSAMAppId: arn:aws:serverlessrepo:us-east-1:931394570548:applications/AthenaHiveMetastoreFunction
      functionSAMSemanticVersion: 1.0.1
      functionName: HiveMetastoreFunction
      functionMemory: 1024
      functionTimeout: 300

Conditions:
  ConditionEmrMultiMaster: !Equals [ !FindInMap [ ClusterConfigurations, emr, masterInstanceCount ], 3 ]
  ConditionEmrMaster1: !Equals [ !Ref HiveMetastoreHostID, 1 ]
  ConditionEmrMaster2: !Equals [ !Ref HiveMetastoreHostID, 2 ]
  ConditionEmrMaster3: !Equals [ !Ref HiveMetastoreHostID, 3 ]

Resources:
  VPC:
    Type: 'AWS::EC2::VPC'
    Properties:
      CidrBlock: !FindInMap [ VpcConfigurations, cidr, vpc ]
      EnableDnsSupport: true
      EnableDnsHostnames: true
      InstanceTenancy: default
      Tags:
      - Key: Name
        Value: !Sub '${AWS::StackName}-VPC'
  InternetGateway:
    Type: 'AWS::EC2::InternetGateway'
    Properties:
      Tags:
      - Key: Name
        Value: !Sub '${AWS::StackName}-IGW'
  VPCGatewayAttachment:
    Type: 'AWS::EC2::VPCGatewayAttachment'
    Properties:
      VpcId: !Ref VPC
      InternetGatewayId: !Ref InternetGateway
  SubnetPublic:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [0, !GetAZs '']
      CidrBlock: !FindInMap [ VpcConfigurations, cidr, publicSubnet1 ]
      MapPublicIpOnLaunch: true
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'public-subnet-1'
  SubnetPrivate1:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [0, !GetAZs '']
      CidrBlock: !FindInMap [ VpcConfigurations, cidr, privateSubnet1 ]
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'private-subnet-1'
  SubnetPrivate2:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select [1, !GetAZs '']
      CidrBlock: !FindInMap [ VpcConfigurations, cidr, privateSubnet2 ]
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'private-subnet-2'
  RouteTablePublic:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'public-route-table-1'
  RouteTablePrivate1:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'private-route-table-1'
  RouteTablePrivate2:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'private-route-table-2'
  RouteTableAssociationPublic:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetPublic
      RouteTableId: !Ref RouteTablePublic
  RouteTableAssociationPrivate1:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetPrivate1
      RouteTableId: !Ref RouteTablePrivate1
  RouteTableAssociationPrivate2:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      SubnetId: !Ref SubnetPrivate2
      RouteTableId: !Ref RouteTablePrivate2
  RouteTablePublicInternetRoute:
    Type: 'AWS::EC2::Route'
    DependsOn: VPCGatewayAttachment
    Properties:
      RouteTableId: !Ref RouteTablePublic
      DestinationCidrBlock: '0.0.0.0/0'
      GatewayId: !Ref InternetGateway
  NatGatewayEIP1:
    Type: 'AWS::EC2::EIP'
    DependsOn: VPCGatewayAttachment
    Properties:
      Domain: vpc
  NatGateway1:
    Type: 'AWS::EC2::NatGateway'
    Properties:
      AllocationId: !GetAtt NatGatewayEIP1.AllocationId
      SubnetId: !Ref SubnetPublic
  RouteTablePrivateInternetRoute1:
    Type: 'AWS::EC2::Route'
    Properties:
      RouteTableId: !Ref RouteTablePrivate1
      DestinationCidrBlock: '0.0.0.0/0'
      NatGatewayId: !Ref NatGateway1
  NatGatewayEIP2:
    Type: 'AWS::EC2::EIP'
    DependsOn: VPCGatewayAttachment
    Properties:
      Domain: vpc
  NatGateway2:
    Type: 'AWS::EC2::NatGateway'
    Properties:
      AllocationId: !GetAtt NatGatewayEIP2.AllocationId
      SubnetId: !Ref SubnetPublic
  RouteTablePrivateInternetRoute2:
    Type: 'AWS::EC2::Route'
    Properties:
      RouteTableId: !Ref RouteTablePrivate2
      DestinationCidrBlock: '0.0.0.0/0'
      NatGatewayId: !Ref NatGateway2
  NetworkAclPublic:
    Type: 'AWS::EC2::NetworkAcl'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'Public NACL'
  NetworkAclPrivate1:
    Type: 'AWS::EC2::NetworkAcl'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'Private NACL 1'
  NetworkAclPrivate2:
    Type: 'AWS::EC2::NetworkAcl'
    Properties:
      VpcId: !Ref VPC
      Tags:
      - Key: Name
        Value: 'Private NACL 2'
  SubnetNetworkAclAssociationPublic:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetPublic
      NetworkAclId: !Ref NetworkAclPublic
  SubnetNetworkAclAssociationPrivate1:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetPrivate1
      NetworkAclId: !Ref NetworkAclPrivate1
  SubnetNetworkAclAssociationPrivate2:
    Type: 'AWS::EC2::SubnetNetworkAclAssociation'
    Properties:
      SubnetId: !Ref SubnetPrivate2
      NetworkAclId: !Ref NetworkAclPrivate2
  NetworkAclEntryInPublicAllowAll:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPublic
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: false
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryOutPublicAllowAll:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPublic
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: true
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryInPrivateAllowVPC1:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPrivate1
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: false
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryOutPrivateAllowVP1:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPrivate1
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: true
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryInPrivateAllowVPC2:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPrivate2
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: false
      CidrBlock: '0.0.0.0/0'
  NetworkAclEntryOutPrivateAllowVP2:
    Type: 'AWS::EC2::NetworkAclEntry'
    Properties:
      NetworkAclId: !Ref NetworkAclPrivate2
      RuleNumber: 99
      Protocol: -1
      RuleAction: allow
      Egress: true
      CidrBlock: '0.0.0.0/0'
  DataBucket:
    Type: 'AWS::S3::Bucket'
  CleanupS3Bucket:
    Type: 'Custom::CleanUpBucket'
    Properties:
      ServiceToken: !GetAtt CleanupS3Function.Arn
      inputBucketName: !Ref DataBucket
  CleanupS3Function:
    DependsOn:
      - DataBucket
      - CleanupS3FunctionLambdaRolePolicy
    Type: 'AWS::Lambda::Function'
    Properties:
      Description: 'Clean-up S3 data bucket'
      Handler: index.lambda_handler
      Runtime: python3.7
      MemorySize: 128
      Timeout: 60
      Role: !GetAtt CleanupS3FunctionLambdaRole.Arn
      Code:
        ZipFile: |
          import json
          import boto3
          import urllib3

          def empty_bucket(bucket_name):
              print("Attempting to empty the bucket {0}".format(bucket_name))
              s3_client = boto3.client('s3')
              s3 = boto3.resource('s3')

              try:
                  bucket = s3.Bucket(bucket_name).load()
              except ClientError:
                  print("Bucket {0} does not exist".format(bucket_name))
                  return
              # Confirm if versioning is enabled
              version_status = s3_client.get_bucket_versioning(Bucket=bucket_name)
              status = version_status.get('Status','')
              if status == 'Enabled':
                  version_status = s3_client.put_bucket_versioning(Bucket=bucket_name,
                                                             VersioningConfiguration={'Status': 'Suspended'})
              version_paginator = s3_client.get_paginator('list_object_versions')
              version_iterator = version_paginator.paginate(
                  Bucket=bucket_name
              )

              for page in version_iterator:
                  print(page)
                  if 'DeleteMarkers' in page:
                      delete_markers = page['DeleteMarkers']
                      if delete_markers is not None:
                          for delete_marker in delete_markers:
                              key = delete_marker['Key']
                              versionId = delete_marker['VersionId']
                              s3_client.delete_object(Bucket=bucket_name, Key=key, VersionId=versionId)
                  if 'Versions' in page and page['Versions'] is not None:
                      versions = page['Versions']
                      for version in versions:
                          print(version)
                          key = version['Key']
                          versionId = version['VersionId']
                          s3_client.delete_object(Bucket=bucket_name, Key=key, VersionId=versionId)
              object_paginator = s3_client.get_paginator('list_objects_v2')
              object_iterator = object_paginator.paginate(
                  Bucket=bucket_name
              )
              for page in object_iterator:
                  if 'Contents' in page:
                      for content in page['Contents']:
                          key = content['Key']
                          s3_client.delete_object(Bucket=bucket_name, Key=content['Key'])
              print("Successfully emptied the bucket {0}".format(bucket_name))

          def lambda_handler(event, context):
              try:
                  bucket = event['ResourceProperties']['inputBucketName']
                  if event['RequestType'] == 'Delete':
                      empty_bucket(bucket)
                  sendResponse(event, context, "SUCCESS")
              except Exception as e:
                  print(e)
                  sendResponse(event, context, "FAILED")

          def sendResponse(event, context, status):
              http = urllib3.PoolManager()
              response_body = {'Status': status,
                               'Reason': 'Log stream name: ' + context.log_stream_name,
                               'PhysicalResourceId': context.log_stream_name,
                               'StackId': event['StackId'],
                               'RequestId': event['RequestId'],
                               'LogicalResourceId': event['LogicalResourceId'],
                               'Data': json.loads("{}")}
              http.request('PUT', event['ResponseURL'], body=json.dumps(response_body))
  CleanupS3FunctionLambdaRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Principal:
              Service:
                - 'lambda.amazonaws.com'
            Action: 'sts:AssumeRole'
      Path: '/'
  CleanupS3FunctionLambdaRolePolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    Properties:
      Roles:
        - !Ref CleanupS3FunctionLambdaRole
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - 's3:List*'
              - 's3:Get*'
              - 's3:DeleteObject*'
            Resource:
              - !GetAtt DataBucket.Arn
              - !Join [ '', [ !GetAtt DataBucket.Arn, '/' ]]
              - !Join [ '', [ !GetAtt DataBucket.Arn, '/*' ]]
          - Effect: 'Deny'
            Action:
              - 's3:DeleteBucket'
            Resource: '*'
          - Effect: 'Allow'
            Action:
              - ec2:CreateNetworkInterface
              - ec2:DescribeNetworkInterfaces
              - ec2:DeleteNetworkInterface
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
            Resource: '*'
  EMRHiveMetastoreCreds:
    Type: 'AWS::SecretsManager::Secret'
    Properties:
      Description: !Sub 'EMR Hive Metastore RDS credentials for ${AWS::StackName}'
      GenerateSecretString:
        SecretStringTemplate: '{"username":"hive"}'
        GenerateStringKey: "password"
        PasswordLength: 16
        ExcludePunctuation: true
  RDSHiveMetastoreSG:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: 'Security group for EMR Hive Metastore'
      VpcId: !Ref VPC
  RDSDBSubnet:
    DependsOn:
      - RouteTablePrivateInternetRoute1
      - RouteTablePrivateInternetRoute2
    Type: 'AWS::RDS::DBSubnetGroup'
    Properties:
      DBSubnetGroupDescription: 'DB Subnet group for EMR Hive Metastore'
      SubnetIds:
        - !Ref SubnetPrivate1
        - !Ref SubnetPrivate2
  RDSEMRHiveMetastore:
    Type: AWS::RDS::DBInstance
    DeletionPolicy: Delete
    Properties:
      DBInstanceIdentifier: !Sub hive-metastore-${AWS::StackName}
      DBInstanceClass: !FindInMap [ ClusterConfigurations, rds, dbInstanceType ]
      DBName: !FindInMap [ ClusterConfigurations, rds, dbName ]
      Engine: !FindInMap [ ClusterConfigurations, rds, dbEngine ]
      EngineVersion: !FindInMap [ ClusterConfigurations, rds, dbEngineVersion ]
      AllocatedStorage: !FindInMap [ ClusterConfigurations, rds, dbStorageGb ]
      AllowMajorVersionUpgrade: false
      AutoMinorVersionUpgrade: true
      Port: !FindInMap [ ClusterConfigurations, rds, dbPort ]
      PubliclyAccessible: false
      StorageType: gp2
      MasterUsername: !Join [ '', [ '{{resolve:secretsmanager:', !Ref EMRHiveMetastoreCreds, ':SecretString:username}}' ]]
      MasterUserPassword: !Join [ '', [ '{{resolve:secretsmanager:', !Ref EMRHiveMetastoreCreds, ':SecretString:password}}' ]]
      DBSubnetGroupName: !Ref RDSDBSubnet
      MultiAZ: true
      VPCSecurityGroups:
        - !Ref RDSHiveMetastoreSG
  EMRMasterSG:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: 'Security group for EMR master'
      VpcId: !Ref VPC
  EMRSlaveSG:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: 'Security group for EMR slave'
      VpcId: !Ref VPC
  EMRServiceAccessSG:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: 'Security group for EMR service'
      VpcId: !Ref VPC
  EMRAccessSG:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: 'Security group for EMR access'
      VpcId: !Ref VPC
  EMRMasterIngressSelfIcmp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: icmp
      FromPort: -1
      ToPort: -1
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRMasterIngressSelfAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRMasterIngressSelfAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRMasterIngressServiceSg:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: tcp
      FromPort: 8443
      ToPort: 8443
      SourceSecurityGroupId: !Ref EMRServiceAccessSG
  EMRMasterIngressSlaveIcmp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: icmp
      FromPort: -1
      ToPort: -1
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRMasterIngressSlaveAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRMasterIngressSlaveAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRMasterSG
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRSlaveIngressSelfIcmp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: icmp
      FromPort: -1
      ToPort: -1
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRSlaveIngressSelfAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRSlaveIngressSelfAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRSlaveIngressServiceSg:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: tcp
      FromPort: 8443
      ToPort: 8443
      SourceSecurityGroupId: !Ref EMRServiceAccessSG
  EMRSlaveIngressMasterIcmp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: icmp
      FromPort: -1
      ToPort: -1
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRSlaveIngressMasterAllTcp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRSlaveIngressMasterAllUdp:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRSlaveSG
      IpProtocol: udp
      FromPort: 0
      ToPort: 65535
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRServiceIngressMaster:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref EMRServiceAccessSG
      IpProtocol: tcp
      FromPort: 9443
      ToPort: 9443
      SourceSecurityGroupId: !Ref EMRMasterSG
  EMRServiceEgressMaster:
    Type: 'AWS::EC2::SecurityGroupEgress'
    Properties:
      GroupId: !Ref EMRServiceAccessSG
      IpProtocol: tcp
      FromPort: 8443
      ToPort: 8443
      DestinationSecurityGroupId: !Ref EMRMasterSG
  EMRServiceEgressSlave:
    Type: 'AWS::EC2::SecurityGroupEgress'
    Properties:
      GroupId: !Ref EMRServiceAccessSG
      IpProtocol: tcp
      FromPort: 8443
      ToPort: 8443
      DestinationSecurityGroupId: !Ref EMRSlaveSG
  HiveMetastoreIngressMaster:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref RDSHiveMetastoreSG
      IpProtocol: tcp
      FromPort: !FindInMap [ ClusterConfigurations, rds, dbPort ]
      ToPort: !FindInMap [ ClusterConfigurations, rds, dbPort ]
      SourceSecurityGroupId: !Ref EMRMasterSG
  HiveMetastoreIngressSlave:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      GroupId: !Ref RDSHiveMetastoreSG
      IpProtocol: tcp
      FromPort: !FindInMap [ ClusterConfigurations, rds, dbPort ]
      ToPort: !FindInMap [ ClusterConfigurations, rds, dbPort ]
      SourceSecurityGroupId: !Ref EMRSlaveSG
  EMRServiceIAMRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - 'elasticmapreduce.amazonaws.com'
            Action: 'sts:AssumeRole'
      Path: '/'
  EMRServiceIAMRolePolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    Properties:
      Roles:
        - !Ref EMRServiceIAMRole
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - ec2:AuthorizeSecurityGroupEgress
              - ec2:AuthorizeSecurityGroupIngress
              - ec2:CancelSpotInstanceRequests
              - ec2:CreateNetworkInterface
              - ec2:CreateSecurityGroup
              - ec2:CreateTags
              - ec2:DeleteNetworkInterface
              - ec2:DeleteTags
              - ec2:DescribeAvailabilityZones
              - ec2:DescribeAccountAttributes
              - ec2:DescribeDhcpOptions
              - ec2:DescribeImages
              - ec2:DescribeInstanceStatus
              - ec2:DescribeInstances
              - ec2:DescribeKeyPairs
              - ec2:DescribeNetworkAcls
              - ec2:DescribeNetworkInterfaces
              - ec2:DescribePrefixLists
              - ec2:DescribeRouteTables
              - ec2:DescribeSecurityGroups
              - ec2:DescribeSpotInstanceRequests
              - ec2:DescribeSpotPriceHistory
              - ec2:DescribeSubnets
              - ec2:DescribeTags
              - ec2:DescribeVpcAttribute
              - ec2:DescribeVpcEndpoints
              - ec2:DescribeVpcEndpointServices
              - ec2:DescribeVpcs
              - ec2:DetachNetworkInterface
              - ec2:ModifyImageAttribute
              - ec2:ModifyInstanceAttribute
              - ec2:RequestSpotInstances
              - ec2:RevokeSecurityGroupEgress
              - ec2:RunInstances
              - ec2:TerminateInstances
              - ec2:DeleteVolume
              - ec2:DescribeVolumeStatus
              - ec2:DescribeVolumes
              - ec2:DetachVolume
              - iam:GetRole
              - iam:GetRolePolicy
              - iam:ListInstanceProfiles
              - iam:ListRolePolicies
              - s3:CreateBucket
              - sdb:BatchPutAttributes
              - sdb:Select
              - cloudwatch:PutMetricAlarm
              - cloudwatch:DescribeAlarms
              - cloudwatch:DeleteAlarms
              - application-autoscaling:RegisterScalableTarget
              - application-autoscaling:DeregisterScalableTarget
              - application-autoscaling:PutScalingPolicy
              - application-autoscaling:DeleteScalingPolicy
              - application-autoscaling:Describe*
            Resource: '*'
          - Effect: 'Allow'
            Action:
              - s3:GetBucketLocation
              - s3:GetBucketCORS
              - s3:GetObjectVersionForReplication
              - s3:GetObject
              - s3:GetBucketTagging
              - s3:GetObjectVersion
              - s3:GetObjectTagging
              - s3:ListMultipartUploadParts
              - s3:ListBucketByTags
              - s3:ListBucket
              - s3:ListObjects
              - s3:ListBucketMultipartUploads
            Resource:
              - !GetAtt DataBucket.Arn
              - !Join [ '', [ !GetAtt DataBucket.Arn, '/*' ]]
          - Effect: 'Allow'
            Action:
              - sqs:CreateQueue
              - sqs:DeleteQueue
              - sqs:DeleteMessage
              - sqs:DeleteMessageBatch
              - sqs:GetQueueAttributes
              - sqs:GetQueueUrl
              - sqs:PurgeQueue
              - sqs:ReceiveMessage
            Resource:
              - !Sub arn:aws:sqs:*:${AWS::AccountId}:AWS-ElasticMapReduce-*
          - Effect: 'Allow'
            Action:
              - iam:CreateServiceLinkedRole
            Resource:
              - arn:aws:iam::*:role/aws-service-role/spot.amazonaws.com/AWSServiceRoleForEC2Spot*
            Condition:
              StringLike:
                iam:AWSServiceName: spot.amazonaws.com
          - Effect: 'Allow'
            Action:
              - iam:PassRole
            Resource:
              - arn:aws:iam::*:role/EMR_AutoScaling_DefaultRole
              - !GetAtt EMRJobFlowRole.Arn
  EMRJobFlowRole:
    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'
  EMRJobFlowRolePolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    Properties:
      Roles:
        - !Ref EMRJobFlowRole
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - ec2:Describe*
              - elasticmapreduce:Describe*
              - elasticmapreduce:ListBootstrapActions
              - elasticmapreduce:ListClusters
              - elasticmapreduce:ListInstanceGroups
              - elasticmapreduce:ListInstances
              - elasticmapreduce:ListStep
            Resource: '*'
          - Effect: 'Allow'
            Action:
              - s3:GetBucketLocation
              - s3:GetBucketCORS
              - s3:GetObjectVersionForReplication
              - s3:GetObject
              - s3:GetBucketTagging
              - s3:GetObjectVersion
              - s3:GetObjectTagging
              - s3:ListMultipartUploadParts
              - s3:ListBucketByTags
              - s3:ListBucket
              - s3:ListObjects
              - s3:ListBucketMultipartUploads
              - s3:PutObject
              - s3:PutObjectTagging
              - s3:HeadBucket
              - s3:DeleteObject
            Resource:
              - !GetAtt DataBucket.Arn
              - !Join [ '', [ !GetAtt DataBucket.Arn, '/*' ]]
          - Effect: 'Allow'
            Action:
              - sqs:CreateQueue
              - sqs:DeleteQueue
              - sqs:DeleteMessage
              - sqs:DeleteMessageBatch
              - sqs:GetQueueAttributes
              - sqs:GetQueueUrl
              - sqs:PurgeQueue
              - sqs:ReceiveMessage
            Resource:
              - !Sub arn:aws:sqs:*:${AWS::AccountId}:AWS-ElasticMapReduce-*
          - Effect: 'Allow'
            Action:
              - 's3:Get*'
              - 's3:List*'
            Resource:
              - !Join [ '', [ 'arn:aws:s3:::', !FindInMap [ ClusterConfigurations, emr, artifactsBucketName ]]]
              - !Join [ '', [ 'arn:aws:s3:::', !FindInMap [ ClusterConfigurations, emr, artifactsBucketName ], '/' ]]
              - !Join [ '', [ 'arn:aws:s3:::', !FindInMap [ ClusterConfigurations, emr, artifactsBucketName ], '/*' ]]
          - Effect: 'Allow'
            Action:
              - 's3:GetObject'
            Resource:
              - !Sub arn:aws:s3:::packages.${AWS::Region}.amazonaws.com/*
              - !Sub arn:aws:s3:::repo.${AWS::Region}.amazonaws.com/*
          - Effect: 'Allow'
            Action:
              - 's3:Put*'
              - 's3:Get*'
              - 's3:Create*'
              - 's3:Abort*'
              - 's3:List*'
            Resource:
              - !Sub arn:aws:s3:::prod.${AWS::Region}.appinfo.src/*
  EMRJobFlowInstanceProfile:
    Type: 'AWS::IAM::InstanceProfile'
    Properties:
      Roles:
        - !Ref EMRJobFlowRole
  EMRCluster:
    DependsOn:
      - RouteTablePrivateInternetRoute1
      - RouteTablePrivateInternetRoute2
      - EMRServiceIAMRolePolicy
      - EMRJobFlowRolePolicy
    Type: 'AWS::EMR::Cluster'
    Properties:
      Name: !Sub emrCluster-${AWS::StackName}
      ReleaseLabel: !FindInMap [ ClusterConfigurations, emr, emrReleaseVersion ]
      JobFlowRole: !Ref EMRJobFlowInstanceProfile
      ServiceRole: !Ref EMRServiceIAMRole
      LogUri: !Sub s3://${DataBucket}/elasticmapreduce/logs
      VisibleToAllUsers: true
      Applications:
        - Name: Hadoop
        - Name: Spark
        - Name: Hive
        - Name: Tez
      Configurations:
        - Classification: hive-site
          ConfigurationProperties:
            javax.jdo.option.ConnectionDriverName: !FindInMap [ ClusterConfigurations, rds, dbDriveName ]
            javax.jdo.option.ConnectionURL: !Join [ '', [ 'jdbc:', !FindInMap [ ClusterConfigurations, rds, dbEngine ], '://', !GetAtt RDSEMRHiveMetastore.Endpoint.Address, ':', !FindInMap [ ClusterConfigurations, rds, dbPort ], '/', !FindInMap [ ClusterConfigurations, rds, dbName ], '?createDatabaseIfNotExist=true' ]]
            javax.jdo.option.ConnectionUserName: !Join [ '', [ '{{resolve:secretsmanager:', !Ref EMRHiveMetastoreCreds, ':SecretString:username}}' ]]
            javax.jdo.option.ConnectionPassword: !Join [ '', [ '{{resolve:secretsmanager:', !Ref EMRHiveMetastoreCreds, ':SecretString:password}}' ]]
            hive.metastore.warehouse.dir: !Join [ '', [ 's3://', !Ref DataBucket, '/hive-warehouse' ]]
      BootstrapActions:
        - Name: Install Python packages
          ScriptBootstrapAction:
            Path: !Join [ '', [ 's3://', !FindInMap [ ClusterConfigurations, emr, artifactsBucketName ], '/', !FindInMap [ ClusterConfigurations, emr, bootStrapScriptPath ]]]
      Instances:
        MasterInstanceGroup:
          Name: MasterInstance
          InstanceCount: !FindInMap [ ClusterConfigurations, emr, masterInstanceCount ]
          InstanceType: !FindInMap [ ClusterConfigurations, emr, masterInstanceType ]
          Market: ON_DEMAND
        CoreInstanceGroup:
          Name: CoreInstance
          InstanceCount: !FindInMap [ ClusterConfigurations, emr, coreInstanceCount ]
          InstanceType: !FindInMap [ ClusterConfigurations, emr, coreInstanceType ]
          Market: ON_DEMAND
        TerminationProtected: false
        Ec2SubnetId: !Ref SubnetPrivate1
        EmrManagedMasterSecurityGroup: !Ref EMRMasterSG
        EmrManagedSlaveSecurityGroup: !Ref EMRSlaveSG
        ServiceAccessSecurityGroup: !Ref EMRServiceAccessSG
        AdditionalMasterSecurityGroups:
          - !Ref EMRAccessSG
  SparkSubmitStep:
    Type: 'AWS::EMR::Step'
    Properties:
      Name: Spark JH Data
      JobFlowId: !Ref EMRCluster
      ActionOnFailure: CONTINUE
      HadoopJarStep:
        Jar: command-runner.jar
        Args:
          - 'spark-submit'
          - '--deploy-mode'
          - 'client'
          - !Join [ '', [ 's3://', !FindInMap [ ClusterConfigurations, emr, artifactsBucketName ], '/', !FindInMap [ ClusterConfigurations, emr, stepActionScriptPath ]]]
          - !Ref DataBucket
  EmrMasterInstance:
    Type: 'Custom::EmrMasterInstance'
    Condition: ConditionEmrMultiMaster
    Properties:
      ServiceToken: !GetAtt EmrMasterInstanceFunction.Arn
      inputEmrId: !Ref EMRCluster
  EmrMasterInstanceFunction:
    DependsOn:
      - EMRCluster
      - EmrMasterInstanceFunctionLambdaRolePolicy
    Type: 'AWS::Lambda::Function'
    Condition: ConditionEmrMultiMaster
    Properties:
      Description: 'Locate instance details for EMR master'
      Handler: index.lambda_handler
      Runtime: python3.7
      MemorySize: 128
      Timeout: 60
      Role: !GetAtt EmrMasterInstanceFunctionLambdaRole.Arn
      Environment:
        Variables:
          REGION: !Ref 'AWS::Region'
      Code:
        ZipFile: |
          import boto3
          from os import environ
          import json
          import urllib3          

          print("[INFO]", "Initialize function")
          session = boto3.session.Session(region_name=environ["REGION"])
          ec2 = session.client('ec2')
          emr = session.client('emr')
          http = urllib3.PoolManager()          

          def findEc2InstanceDetails(emrId, regionName, returnParameter):
              print("[INFO]", "Trying to find EC2 Instances for EMR Id: " +emrId)
              try:
                  ec2Response = ec2.describe_instances(Filters=[
                    {'Name': 'tag:aws:elasticmapreduce:job-flow-id', 'Values': [emrId]},
                    {'Name': 'tag:aws:elasticmapreduce:instance-group-role', 'Values': ['MASTER']}
                    ])

                  if len(ec2Response['Reservations']) > 0:
                      print("[INFO]", "Found matching EC2 instances for EMR Id: " +emrId)
                      instanceDetails = []
                      if returnParameter == 'instanceId':
                          for reservationElement in range(len(ec2Response['Reservations'])):
                              instanceDetails.append(ec2Response['Reservations'][reservationElement]['Instances'][0]['InstanceId'])
                      elif returnParameter == 'instanceIp':
                          for reservationElement in range(len(ec2Response['Reservations'])):
                              instanceDetails.append(ec2Response['Reservations'][reservationElement]['Instances'][0]['PrivateIpAddress'])
                  else:
                      print("[INFO]", "No matching EC2 instances for EMR Id: " +emrId)
                      instanceDetails = None

                  return instanceDetails

              except Exception as e:
                  print("[ERROR]", e)

          def disableEmrTerminationProtection(emrId):
              print("[INFO]", "Trying to disable termination protection for EMR Id: " +emrId)
              try:
                  emr.set_termination_protection(JobFlowIds=[emrId], TerminationProtected=False)
              except Exception as e:
                  print("[ERROR]", e)

          def sendResponse(event, context, status, data, reason):
              response_body = {'Status': status,
                               'Reason': reason,
                               'PhysicalResourceId': context.log_stream_name,
                               'StackId': event['StackId'],
                               'RequestId': event['RequestId'],
                               'LogicalResourceId': event['LogicalResourceId'],
                               'Data': data}
              http.request('PUT', event['ResponseURL'], body=json.dumps(response_body))   

          def lambda_handler(event, context):
              print("[INFO]", "Locating EC2 Instance Id and IP")          

              try:
                  EmrId = event["ResourceProperties"]["inputEmrId"]
                  Region = environ["REGION"]

                  response_data = {}          

                  # Find EC2 Instance Id
                  if event["RequestType"] == 'Create':
                      # Disable termination protection
                      disableEmrTerminationProtection(EmrId)
                      # Locate EC2 instance id for master node
                      EmrMasterInstanceId = findEc2InstanceDetails(EmrId, Region, 'instanceId')
                      EmrMasterInstanceIp = findEc2InstanceDetails(EmrId, Region, 'instanceIp')
                      if EmrMasterInstanceId != None and EmrMasterInstanceIp != None:
                          response_status = "SUCCESS"
                          response_reason = "Located EC2 Instance"
                      elif EmrMasterInstanceId == None or EmrMasterInstanceIp == None:
                          response_status = "FAILED"
                          response_reason = "Could not find EC2 Instance"

                      response_data['emrMasterInstanceId'] = EmrMasterInstanceId
                      response_data['emrMasterInstanceIp'] = EmrMasterInstanceIp
                  elif event["RequestType"] == 'Delete':
                      # Disable termination protection
                      disableEmrTerminationProtection(EmrId)
                      response_status = "SUCCESS"
                      response_reason = "Disbaled EMR termination protection"

                  sendResponse(event, context, response_status, response_data, response_reason)
              except Exception as e:
                  print("[ERROR]", e)
                  response_data = {}
                  sendResponse(event, context, "FAILED", response_data, e)          

              print("[INFO]", "Execution complete")
  EmrMasterInstanceFunctionLambdaRole:
    Type: 'AWS::IAM::Role'
    Condition: ConditionEmrMultiMaster
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Principal:
              Service:
                - 'lambda.amazonaws.com'
            Action: 'sts:AssumeRole'
      Path: '/'
  EmrMasterInstanceFunctionLambdaRolePolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    Condition: ConditionEmrMultiMaster
    Properties:
      Roles:
        - !Ref EmrMasterInstanceFunctionLambdaRole
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - ec2:CreateNetworkInterface
              - ec2:DescribeInstances
              - ec2:DeleteNetworkInterface
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
            Resource: '*'
          - Effect: 'Allow'
            Action:
              - elasticmapreduce:SetTerminationProtection
            Resource:
              - !Join [ '', [ !Sub 'arn:aws:elasticmapreduce:${AWS::Region}:${AWS::AccountId}:cluster/', !Ref EMRCluster ]]
  HiveMetastoreFunction:
    Type: 'AWS::Serverless::Application'
    Properties:
      Location:
        ApplicationId: !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionSAMAppId ]
        SemanticVersion: !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionSAMSemanticVersion ]
      Parameters:
        HMSUris: !If [ ConditionEmrMultiMaster, !If [ ConditionEmrMaster1, !Join [ '', [ 'thrift://', !Select [ 0, !GetAtt EmrMasterInstance.emrMasterInstanceIp ], ':9083' ]], !If [ ConditionEmrMaster2, !Join [ '', [ 'thrift://', !Select [ 1, !GetAtt EmrMasterInstance.emrMasterInstanceIp ], ':9083' ]], !If [ ConditionEmrMaster3, !Join [ '', [ 'thrift://', !Select [ 2, !GetAtt EmrMasterInstance.emrMasterInstanceIp ], ':9083' ]], 'None' ]]], !Join [ '', [ 'thrift://', !GetAtt EMRCluster.MasterPublicDNS, ':9083' ]]]
        LambdaFuncName: !Join [ '-', [ !Sub '${AWS::StackName}', !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionName ]]]
        SpillLocation: !Sub '${DataBucket}/hms_spill'
        VPCSecurityGroupIds: !Ref EMRMasterSG
        VPCSubnetIds: !Join [ ', ', [ !Ref SubnetPrivate1, !Ref SubnetPrivate2 ]]
        LambdaMemory: !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionMemory ]
        LambdaTimeout: !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionTimeout ]
  DataEngineerUserCreds:
    Type: 'AWS::SecretsManager::Secret'
    Properties:
      Description: !Sub 'DataEngineering User credentials for ${AWS::StackName}'
      GenerateSecretString:
        SecretStringTemplate: !Sub '{"username":"de_user_${AWS::Region}_${AWS::StackName}"}'
        GenerateStringKey: "password"
        PasswordLength: 16
        ExcludePunctuation: true
  SalesUserCreds:
    Type: 'AWS::SecretsManager::Secret'
    Properties:
      Description: !Sub 'Sales User credentials for ${AWS::StackName}'
      GenerateSecretString:
        SecretStringTemplate: !Sub '{"username":"sales_user_${AWS::Region}_${AWS::StackName}"}'
        GenerateStringKey: "password"
        PasswordLength: 16
        ExcludePunctuation: true
  DataEngineerUser:
    DependsOn:
      - HiveMetastoreDataSourceAccessPolicy
      - DataBucketReadWriteAccessPolicy
    Type: 'AWS::IAM::User'
    Properties:
      UserName: !Sub 'de_user_${AWS::Region}_${AWS::StackName}'
      Groups:
        - !Ref DataEngineerGroup
      LoginProfile:
        Password: !Join [ '', [ '{{resolve:secretsmanager:', !Ref DataEngineerUserCreds, ':SecretString:password}}' ]]
  SalesUser:
    Type: 'AWS::IAM::User'
    Properties:
      UserName: !Sub 'sales_user_${AWS::Region}_${AWS::StackName}'
      Groups:
        - !Ref SalesGroup
      LoginProfile:
        Password: !Join [ '', [ '{{resolve:secretsmanager:', !Ref SalesUserCreds, ':SecretString:password}}' ]]
  DataEngineerGroup:
    Type: 'AWS::IAM::Group'
    Properties:
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/AmazonAthenaFullAccess'
  SalesGroup:
    Type: 'AWS::IAM::Group'
    Properties:
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/AmazonAthenaFullAccess'
  HiveMetastoreDataSourceAccessPolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    Properties:
      Groups:
        - !Ref DataEngineerGroup
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - 'lambda:GetFunction'
              - 'lambda:GetLayerVersion'
              - 'lambda:InvokeFunction'
            Resource:
              - !Join [ '', [ !Sub 'arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:', !Join [ '-', [ !Sub '${AWS::StackName}', !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionName ]]]]] 
          - Effect: 'Allow'
            Action:
              - 's3:GetBucketLocation'
              - 's3:GetObject'
              - 's3:ListBucket'
              - 's3:PutObject'
              - 's3:ListMultipartUploadParts'
              - 's3:AbortMultipartUpload'
            Resource:
              - !Sub 'arn:aws:s3:::${DataBucket}/hms_spill'
          - Effect: 'Allow'
            Action:
              - 'lambda:ListFunctions'
            Resource: '*'
  DataBucketReadWriteAccessPolicy:
    Type: 'AWS::IAM::ManagedPolicy'
    Properties:
      Groups:
        - !Ref DataEngineerGroup
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: 'Allow'
            Action:
              - 's3:Get*'
              - 's3:List*'
              - 's3:Put*'
            Resource:
              - !GetAtt DataBucket.Arn
              - !Join [ '', [ !GetAtt DataBucket.Arn, '/' ]]
              - !Join [ '', [ !GetAtt DataBucket.Arn, '/*' ]]

Outputs:
  HiveMetastoreFunctionName:
    Description: 'Lambda function to use with Athena for Hive Metastore'
    Value: !Join [ '-', [ !Sub '${AWS::StackName}', !FindInMap [ ClusterConfigurations, hiveMetastoreFunction, functionName ]]]
  DataBucket:
    Description: 'Data bucket to store data and results'
    Value: !Ref DataBucket
  DataEngineerUserCredentials:
    Description: 'Data Engineering IAM user credentials secret ARN'
    Value: !Ref DataEngineerUserCreds
  SalesUserCredentials:
    Description: 'Sales IAM user credentials secret ARN'
    Value: !Ref SalesUserCreds
