AWSTemplateFormatVersion: '2010-09-09'
Description:  This template deploys components to replicate data from Aurora postgres source to S3 Datalake using DMS and submit AWS Glue jobs using Apache Iceberg Connector for AWS Glue
Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      -
        Label:
          default: "VPC Configuration"
        Parameters:
          - VpcCIDR
          - PublicSubnetCIDR
          - PrivateSubnet1CIDR
          - PrivateSubnet2CIDR
      -
        Label:
          default: "Access Configuration"
        Parameters:
          - DBMasterUserPassword
          - KeyName
          - ClientIPCIDR
    ParameterLabels:
      VpcCIDR:
        default: "VPC CIDR"
      PublicSubnetCIDR:
        default: "Public Subnet"
      PrivateSubnet1CIDR:
        default: "Private Subnet 1"
      PrivateSubnet2CIDR:
        default: "Private Subnet 2"
      DBMasterUserPassword:
        default: "Aurora DB Password"
      KeyName:
        default: "Key Pair Name"
      ClientIPCIDR:
        default: "My Client IP"

Parameters:

  VpcCIDR:
    Description: The IP range (CIDR notation) for this VPC
    Type: String
    Default: 10.0.0.0/16
    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

  PrivateSubnet1CIDR:
    Description: The IP range (CIDR notation) for the private subnet in the first Availability Zone
    Type: String
    Default: 10.0.10.0/24
    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
    
  PrivateSubnet2CIDR:
    Description: The IP range (CIDR notation) for the private subnet in the second Availability Zone
    Type: String
    Default: 10.0.20.0/24
    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
    
  PublicSubnetCIDR:
    Description: The IP range (CIDR notation) for the public subnet in the first Availability Zone
    Type: String
    Default: 10.0.30.0/24
    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
    
  DBMasterUserPassword:
    Type: String
    Description: Default value:Awsuser1. Aurora database master user password. This will be used for Aurora cluster. Must have atleast 8 characters, one upper case, one lower case and one number. Special characters are not allowed
    AllowedPattern: '^(?=.*[A-Za-z])(?=.*[0-9])[A-Za-z0-9]{8,}$'
    ConstraintDescription: Must have atleast 8 characters, one upper case, one lower case and one number. Special characters are not allowed
    NoEcho: true

  KeyName:
    Type: AWS::EC2::KeyPair::KeyName
    Description: The KeyPair to be used in your EC2 instance. This EC2 instance will be used as a proxy to connect from your SQL client to Aurora postgres source database.
    ConstraintDescription: Please select a key to be used to login to the EC2 instance
    
  ClientIPCIDR:
    Type: String
    Description: The IP Address of your SQL Client that will be used to connect to the Aurora Source DB. On Mac, you can run the following command to get your IP in CIDR notation using - curl ipecho.net/plain ; echo /32
    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

  EC2ImageId:
    Type: AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
    Description: Latest AMI ID of Amazon Linux 2. You can keep the default value for this.
    Default: /aws/service/ami-amazon-linux-latest/amzn2-ami-hvm-x86_64-gp2

Resources:

  VPC:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: !Ref VpcCIDR
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags:
        -
          Key: Project
          Value: Lakehouse

  InternetGateway:
    Type: AWS::EC2::InternetGateway
    DependsOn:
      - VPC     
    Properties:
      Tags:
        -
          Key: Project
          Value: Lakehouse  
      
  VPCInternetGatewayAttachment:
    Type: AWS::EC2::VPCGatewayAttachment
    DependsOn:
      - InternetGateway
      - VPC      
    Properties:
      VpcId: !Ref VPC
      InternetGatewayId: !Ref InternetGateway
      
  PublicSubnet:
    Type: AWS::EC2::Subnet
    DependsOn:
      - VPC    
    Properties:
      AvailabilityZone: !Select [0, !GetAZs '']
      CidrBlock: !Ref PublicSubnetCIDR
      MapPublicIpOnLaunch: true
      VpcId: !Ref VPC
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  PrivateSubnet1:
    Type: AWS::EC2::Subnet
    DependsOn:
      - VPC    
    Properties:
      AvailabilityZone: !Select [ 0, !GetAZs  '' ]
      CidrBlock: !Ref PrivateSubnet1CIDR
      MapPublicIpOnLaunch: false
      VpcId: !Ref VPC      
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  PrivateSubnet2:
    Type: AWS::EC2::Subnet
    DependsOn:
      - VPC     
    Properties:
      AvailabilityZone: !Select [ 2, !GetAZs  '' ]
      CidrBlock: !Ref PrivateSubnet2CIDR
      MapPublicIpOnLaunch: false
      VpcId: !Ref VPC      
      Tags:
        -
          Key: Project
          Value: Lakehouse

  PublicRouteTable:
    Type: AWS::EC2::RouteTable
    DependsOn:
      - VPC    
    Properties:
      VpcId: !Ref VPC
      Tags:
        -
          Key: Project
          Value: Lakehouse

  DefaultPublicRoute:
    Type: AWS::EC2::Route
    DependsOn:
      - VPCInternetGatewayAttachment 
    Properties:
      RouteTableId: !Ref PublicRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway

  PublicSubnetRouteTableAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation  
    Properties:
      RouteTableId: !Ref PublicRouteTable    
      SubnetId: !Ref PublicSubnet

  NatGatewayEIP:
    Type: AWS::EC2::EIP
    DependsOn:
      - VPC
      - VPCInternetGatewayAttachment
    Properties:
      Domain: vpc

  NatGateway:
    Type: AWS::EC2::NatGateway
    DependsOn:
      - VPC
      - NatGatewayEIP
      - PublicSubnet    
    Properties:
      AllocationId: !GetAtt NatGatewayEIP.AllocationId
      SubnetId: !Ref PublicSubnet

  PrivateRouteTable:
    Type: AWS::EC2::RouteTable
    DependsOn:
      - VPC    
    Properties:
      VpcId: !Ref VPC
      Tags:
        -
          Key: Project
          Value: Lakehouse

  DefaultPrivateRoute:
    Type: AWS::EC2::Route   
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      DestinationCidrBlock: 0.0.0.0/0
      NatGatewayId: !Ref NatGateway

  PrivateSubnetRouteTableAssociation1:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnet1

  PrivateSubnetRouteTableAssociation2:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      RouteTableId: !Ref PrivateRouteTable
      SubnetId: !Ref PrivateSubnet2

  LakehouseSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    DependsOn:
      - VPC    
    Properties:
      GroupName: Lakehouse-sg
      GroupDescription: Security group to protect Lakehouse resources
      VpcId: !Ref VPC
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  SecurityGroupIngressForSelfRef:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn:
      - VPC
      - LakehouseSecurityGroup
    Properties:
      GroupId: !Ref LakehouseSecurityGroup
      IpProtocol: tcp
      FromPort: 0
      ToPort: 65535
      Description: Self-referencing rule
      SourceSecurityGroupId: !GetAtt LakehouseSecurityGroup.GroupId

  SecurityGroupIngressForEC2Tunnel:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn:
      - VPC
      - LakehouseSecurityGroup
    Properties:
      GroupId: !Ref LakehouseSecurityGroup
      IpProtocol: tcp
      FromPort: 22
      ToPort: 22
      Description: Rule to allow connection from the SQL client to the EC2 instance used as bastion host for SSH tunnel
      CidrIp: !Ref ClientIPCIDR

  SecurityGroupIngressForPrivateSubnet1:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn:
      - VPC
      - LakehouseSecurityGroup
    Properties:
      GroupId: !Ref LakehouseSecurityGroup
      IpProtocol: tcp
      FromPort: 443
      ToPort: 443
      Description: Rule to allow connection from EMR Notebook to Glue VPCEndpoint
      CidrIp: !Ref PrivateSubnet1CIDR

  SecurityGroupIngressForPrivateSubnet2:
    Type: AWS::EC2::SecurityGroupIngress
    DependsOn:
      - VPC
      - LakehouseSecurityGroup
    Properties:
      GroupId: !Ref LakehouseSecurityGroup
      IpProtocol: tcp
      FromPort: 443
      ToPort: 443
      Description: Rule to allow connection from EMR Notebook to Glue VPCEndpoint
      CidrIp: !Ref PrivateSubnet2CIDR

  SecurityGroupEgressForSelfRef:
    Type: AWS::EC2::SecurityGroupEgress
    DependsOn:
      - VPC
      - LakehouseSecurityGroup
    Properties:
      GroupId: !Ref LakehouseSecurityGroup
      IpProtocol: "-1"
      FromPort: 0
      ToPort: 65535
      Description: Rule to allow connection to Self SecurityGroup
      DestinationSecurityGroupId: !GetAtt LakehouseSecurityGroup.GroupId

  SecurityGroupEgressForEC2Tunnel:
    Type: AWS::EC2::SecurityGroupEgress
    DependsOn:
      - VPC
      - LakehouseSecurityGroup
    Properties:
      GroupId: !Ref LakehouseSecurityGroup
      IpProtocol: "-1"
      FromPort: 0
      ToPort: 65535
      Description: Rule to allow connection to EC2Tunnel
      CidrIp: !Ref ClientIPCIDR

  S3VPCEndpoint:
    Type: AWS::EC2::VPCEndpoint
    DependsOn:
      - PrivateRouteTable
      - VPC
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Action: '*'
            Effect: Allow
            Resource: '*'
            Principal: '*'
      RouteTableIds:
        - !Ref PrivateRouteTable
      ServiceName: !Sub com.amazonaws.${AWS::Region}.s3
      VpcId: !Ref VPC      

  GlueVPCEndpoint:
    Type: AWS::EC2::VPCEndpoint
    DependsOn:
      - PrivateSubnet1
      - PrivateSubnet2
      - LakehouseSecurityGroup
      - VPC    
    Properties:
      ServiceName: !Sub com.amazonaws.${AWS::Region}.glue
      VpcEndpointType: Interface
      VpcId: !Ref VPC
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
      SecurityGroupIds:
        - !GetAtt LakehouseSecurityGroup.GroupId
      PrivateDnsEnabled: true

  ExecuteLambdaFnsRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: LakehouseLambdaExecutionRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonSSMReadOnlyAccess
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  TriggerLambdaFns:
    Type: AWS::IAM::ManagedPolicy
    DependsOn:
      - ExecuteLambdaFnsRole
      - ScriptsAndTempS3Bucket
      - CuratedS3Bucket
      - RawS3Bucket
    Properties:
      ManagedPolicyName: LakehouseTriggerLambdaFns
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - 
            Effect: Allow
            Action:
              - glue:DeleteDatabase
              - dms:TestConnection
              - dms:DescribeConnections
              - dms:StopReplicationTask
              - dms:DescribeReplicationTasks
              - dms:StartReplicationTask
            Resource: '*'
          -
            Effect: Allow
            Action: 
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
              - logs:AssociateKmsKey
            Resource:
              - !Sub arn:aws:logs:*:${AWS::AccountId}:log-group:/aws/lambda/create-dms-role-check
              - !Sub arn:aws:logs:*:${AWS::AccountId}:log-group:/aws/lambda/create-dms-role-check:log-stream:* 
              - !Sub arn:aws:logs:*:${AWS::AccountId}:log-group:/aws/lambda/invoke-dms-tasks
              - !Sub arn:aws:logs:*:${AWS::AccountId}:log-group:/aws/lambda/invoke-dms-tasks:log-stream:*                
          -
            Effect: Allow
            Action:
              - s3:PutObject
              - s3:GetObject
              - s3:ListBucket
              - s3:DeleteObject
            Resource:
              - !Join ['', ["arn:aws:s3:::", !Ref ScriptsAndTempS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref ScriptsAndTempS3Bucket, "/*"]]  
              - !Join ['', ["arn:aws:s3:::", !Ref CuratedS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref CuratedS3Bucket, "/*"]]  
              - !Join ['', ["arn:aws:s3:::", !Ref RawS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref RawS3Bucket, "/*"]]               
          -
            Effect: Allow
            Action:
                - s3:GetObject
                - s3:ListBucket
            Resource:
                - arn:aws:s3:::aws-bigdata-blog
                - arn:aws:s3:::aws-bigdata-blog/artifacts/iceberg-on-glue/*
          -
            Effect: Allow
            Action:
              - iam:GetRole
              - iam:DetachRolePolicy
              - iam:UntagRole
              - iam:ListRoleTags
              - iam:TagRole
              - iam:CreateRole
              - iam:DeleteRole
              - iam:AttachRolePolicy
            Resource: !Sub arn:aws:iam::${AWS::AccountId}:role/dms-vpc-role
      Roles:
        - LakehouseLambdaExecutionRole
  
  CreateDMSRoleLambdaFn:
    Type: AWS::Lambda::Function
    DependsOn:
      - LakehouseKMSKey
      - TriggerLambdaFns
      - ExecuteLambdaFnsRole
    Properties:
      Code:
        ZipFile: |
          import boto3
          import json
          import urllib3
          import time
            
          iam_client = boto3.client('iam')
          http = urllib3.PoolManager()
            
          def lambda_handler(event, context):
            print('Event: %s' % json.dumps(event))
            response = { 'StackId': event['StackId'], 'RequestId': event['RequestId'], 'LogicalResourceId': event['LogicalResourceId'] }
            if event['RequestType'] == 'Create':
              try:
                iam_client.get_role(RoleName='dms-vpc-role')
                response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'dms-vpc-roleExists' })
                print('dms-vpc-role exists. No action taken.')
              except Exception as e:
                if e.response['Error']['Code'] == 'NoSuchEntity':
                  iam_client.create_role(Path='/', RoleName='dms-vpc-role', AssumeRolePolicyDocument='{ "Version": "2012-10-17", "Statement": [ { "Sid": "", "Effect": "Allow", "Principal": { "Service": "dms.amazonaws.com" }, "Action": "sts:AssumeRole" } ] }')
                  iam_client.attach_role_policy(PolicyArn='arn:aws:iam::aws:policy/service-role/AmazonDMSVPCManagementRole', RoleName='dms-vpc-role')
                  iam_client.tag_role(RoleName='dms-vpc-role', Tags=[{ 'Key': 'Project', 'Value': 'Lakehouse' }])
                  response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'dms-vpc-roleCreated' })
                  print('dms-vpc-role created.')
                  time.sleep(90)
                else:
                  response.update({ 'Status': 'FAILED', 'Reason': str(e), 'PhysicalResourceId': 'dms-vpc-roleCreationFailed' })
                  print('dms-vpc-role creation failed.')
            if event['RequestType'] == 'Update':
              response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'dms-vpc-roleNothingChangedOnUpdate' })
              print('dms-vpc-role was not updated. Nothing changed.')
            if event['RequestType'] == 'Delete':
              dmsVPCRoleTags = iam_client.list_role_tags(RoleName='dms-vpc-role')['Tags']
              if len(dmsVPCRoleTags) == 0:
                print('Role has not been created by this stack or the tags have been manually edited/removed. No action taken.')
                response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'dms-vpc-roleNothingChangedOnDelete' })
              else:
                if dmsVPCRoleTags[0]['Value'] == 'Lakehouse':
                  try:
                    iam_client.detach_role_policy(RoleName='dms-vpc-role', PolicyArn='arn:aws:iam::aws:policy/service-role/AmazonDMSVPCManagementRole')
                    iam_client.delete_role(RoleName='dms-vpc-role')
                    response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'dms-vpc-roleDeleted' })
                    print('dms-vpc-role deleted.')
                  except Exception as e:
                    response.update({ 'Status': 'FAILED', 'Reason': str(e), 'PhysicalResourceId': 'dms-vpc-roleDeleteFailed' })
                    print('dms-vpc-role delete failed.')
                else:
                  print('Role has not been created by this stack or the tags have been manually edited/removed. No action taken.')
                  response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'dms-vpc-roleNothingChangedOnDelete' })                          
            resp = http.request('PUT', event['ResponseURL'], body = json.dumps(response), headers = {'Content-Type': 'application/json'}, retries = False)
            print('http resp:' + resp.data.decode('utf-8'))
            print(response)
            return(response)
      FunctionName: create-dms-role-check
      KmsKeyArn: !GetAtt LakehouseKMSKey.Arn
      MemorySize: 128
      Runtime: python3.6
      Description: Lambda function that creates dms-vpc-role.
      Handler: index.lambda_handler
      Role: !GetAtt ExecuteLambdaFnsRole.Arn
      Timeout: 900      
      Tags:
        -
          Key: Project
          Value: Lakehouse
          
  CreateDMSRole:
    Type: Custom::CreateDMSRole
    DependsOn:
      - CreateDMSRoleLambdaFn
    Properties:
      ServiceToken: !GetAtt CreateDMSRoleLambdaFn.Arn

  DMSReplicateDataToS3:
    Type: AWS::IAM::ManagedPolicy
    DependsOn:
      - DMSWriteToRawS3Role
      - RawS3Bucket
    Properties:
      ManagedPolicyName: LakehouseDMSReplicateDataToS3
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - 
            Effect: Allow
            Action:
              - s3:PutObject
              - s3:GetObject
              - s3:ListBucket
              - s3:DeleteObject
            Resource:
              - !Join ['', ["arn:aws:s3:::", !Ref RawS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref RawS3Bucket, "/*"]]               
      Roles:
        - LakehouseDMSWriteToS3Role

  DMSWriteToRawS3Role:
    Type: AWS::IAM::Role
    Properties:
      RoleName: LakehouseDMSWriteToS3Role
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - dms.amazonaws.com
            Action:
              - sts:AssumeRole
      Path: /
      Tags:
        -
          Key: Project
          Value: Lakehouse

  ExecuteGlueJobRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: LakehouseGlueJobRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - glue.amazonaws.com
            Action:
              - sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AmazonSSMReadOnlyAccess
        - arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly
        - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
        - arn:aws:iam::aws:policy/AWSGlueConsoleFullAccess

      Path: /
      Tags:
        -
          Key: Project
          Value: Lakehouse

  GlueJobToExecute:
    Type: AWS::IAM::ManagedPolicy
    DependsOn: ExecuteGlueJobRole
    Properties:
      ManagedPolicyName: LakehouseGlueJobPolicy
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - 
            Effect: Allow
            Action:
                - glue:CreateTable
                - glue:GetTable
                - glue:GetTables                
                - glue:UpdateTable
                - glue:DeleteTable
                - glue:BatchDeleteTable
                - glue:GetTableVersion
                - glue:GetTableVersions
                - glue:DeleteTableVersion
                - glue:BatchDeleteTableVersion
                - glue:CreatePartition
                - glue:BatchCreatePartition
                - glue:GetPartition
                - glue:GetPartitions
                - glue:BatchGetPartition
                - glue:UpdatePartition
                - glue:DeletePartition
                - glue:BatchDeletePartition
                - glue:TagResource
                - glue:UntagResource
            Resource: '*'
          - 
            Effect: Allow
            Action:
              - logs:CreateLogGroup
              - logs:CreateLogStream
              - logs:PutLogEvents
              - logs:AssociateKmsKey
            Resource: arn:aws:logs:*:*:/aws-glue/*    
          - 
            Effect: Allow
            Action:
              - cloudwatch:PutMetricData
            Resource: '*'
          - 
            Effect: Allow
            Action:
              - ec2:DescribeVpcEndpoints
              - ec2:DescribeRouteTables
              - ec2:DescribeNetworkInterfaces
              - ec2:DescribeSecurityGroups
              - ec2:DescribeSubnets
              - ec2:DescribeVpcAttribute
              - ec2:CreateTags
              - ec2:DeleteTags
              - ec2:CreateNetworkInterface
              - ec2:DeleteNetworkInterface
            Resource: '*'
          - 
            Effect: Allow
            Action:
              - dynamodb:BatchGetItem
              - dynamodb:DescribeStream
              - dynamodb:DescribeTable
              - dynamodb:GetItem
              - dynamodb:Query
              - dynamodb:Scan
              - dynamodb:BatchWriteItem
              - dynamodb:CreateTable
              - dynamodb:DeleteTable
              - dynamodb:UpdateTable
              - dynamodb:UpdateItem
              - dynamodb:PutItem
            Resource: !Sub arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/*
          - 
            Effect: Allow
            Action:
              - iam:GetRole
            Resource: '*'
          - 
            Effect: Allow
            Action:
              - s3:GetBucketLocation
              - s3:ListBucket
              - s3:GetBucketAcl
              - s3:GetObject
              - s3:PutObject
              - s3:DeleteObject
            Resource:
              - !Join ['', ["arn:aws:s3:::", !Ref ScriptsAndTempS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref ScriptsAndTempS3Bucket, "/*"]]  
              - !Join ['', ["arn:aws:s3:::", !Ref CuratedS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref CuratedS3Bucket, "/*"]]  
              - !Join ['', ["arn:aws:s3:::", !Ref RawS3Bucket]]
              - !Join ['', ["arn:aws:s3:::", !Ref RawS3Bucket, "/*"]]                                 
      Roles:
        - LakehouseGlueJobRole

  LakehouseKMSKey:
    Type: AWS::KMS::Key
    DependsOn:
      - ExecuteGlueJobRole
      - EMREC2Role
      - EMRRole
    Properties: 
      Description: Key to encrypt in Lakehouse
      Enabled: true
      EnableKeyRotation: true
      KeyPolicy: 
        Version: 2012-10-17
        Statement:
          - 
            Sid: Enable IAM User Permissions
            Effect: Allow
            Principal:
              AWS: !Sub arn:aws:iam::${AWS::AccountId}:root
            Action: 
              - kms:*
            Resource: '*'
          - 
            Sid: Allow use of the key
            Effect: Allow
            Principal: 
              AWS:
                - !Sub arn:aws:iam::${AWS::AccountId}:role/LakehouseGlueJobRole
                - !GetAtt EMREC2Role.Arn
              Service: !Sub logs.${AWS::Region}.amazonaws.com
            Action:
              - kms:Encrypt
              - kms:Decrypt
              - kms:ReEncrypt*
              - kms:GenerateDataKey
              - kms:GenerateDataKeyWithoutPlaintext
              - kms:GenerateDataKeyPairWithoutPlaintext
              - kms:GenerateDataKeyPair
              - kms:DescribeKey
            Resource: '*'            
      Tags:
        -
          Key: Project
          Value: Lakehouse

  LakehouseKMSKeyAlias:
    Type: AWS::KMS::Alias
    DependsOn: LakehouseKMSKey
    Properties:
      AliasName: alias/Lakehouse-kms-key
      TargetKeyId: !Ref LakehouseKMSKey

  RawS3Bucket:
    Type: AWS::S3::Bucket
    DependsOn: LakehouseKMSKey    
    DeletionPolicy: Delete
    Properties:
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: aws:kms
              KMSMasterKeyID: !GetAtt LakehouseKMSKey.Arn              
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  CuratedS3Bucket:
    Type: AWS::S3::Bucket
    DependsOn: LakehouseKMSKey     
    DeletionPolicy: Delete
    Properties:
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: aws:kms
              KMSMasterKeyID: !GetAtt LakehouseKMSKey.Arn
      Tags:
        -
          Key: Project
          Value: Lakehouse

  ScriptsAndTempS3Bucket:
    Type: AWS::S3::Bucket
    DeletionPolicy: Delete    
    Properties:
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256       
      Tags:
        -
          Key: Project
          Value: Lakehouse      
      
  RDSDBClusterParameterGroup:
    Type: AWS::RDS::DBClusterParameterGroup
    Properties:
      Description: "Aurora PG Cluster Parameter Group for Lakehouse-db-source"
      Family: aurora-postgresql11
      Parameters:
        rds.logical_replication: 1
        wal_sender_timeout: 0
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  AuroraSubnetGroup:
    Type: AWS::RDS::DBSubnetGroup
    DependsOn:
      - PrivateSubnet1
      - PrivateSubnet2
      - VPC
    Properties:
      DBSubnetGroupDescription: "Subnet group for Aurora source database"
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  AuroraCluster:        
    Type: AWS::RDS::DBCluster
    DependsOn:
      - LakehouseSecurityGroup
      - RDSDBClusterParameterGroup
      - AuroraSubnetGroup
      - LakehouseKMSKey
      - VPC
    DeletionPolicy: Delete
    Properties:
      Engine: aurora-postgresql
      EngineVersion: 11.15
      DatabaseName: lakehouse_source_db
      DBClusterIdentifier: lakehouse-db-source
      Port: 5432
      MasterUsername: postgres
      MasterUserPassword: !Ref DBMasterUserPassword
      VpcSecurityGroupIds:
        - !Ref LakehouseSecurityGroup  
      DBClusterParameterGroupName: !Ref RDSDBClusterParameterGroup
      StorageEncrypted:  true
      KmsKeyId: !GetAtt LakehouseKMSKey.Arn
      DBSubnetGroupName: !Ref AuroraSubnetGroup
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  AuroraDBInstance1:
    Type: AWS::RDS::DBInstance
    DependsOn:
      - AuroraCluster
      - AuroraSubnetGroup
      - LakehouseKMSKey
      - VPC
    DeletionPolicy: Delete
    Properties:
      CopyTagsToSnapshot: true  
      DBInstanceClass: db.r5.large
      DBClusterIdentifier: !Ref AuroraCluster
      DBInstanceIdentifier: lakehouse-db-source-instance-1     
      Engine: aurora-postgresql
      EngineVersion: 11.15      
      AutoMinorVersionUpgrade: true
      PubliclyAccessible: false
      EnablePerformanceInsights: true
      PerformanceInsightsRetentionPeriod: 7
      DBSubnetGroupName: !Ref AuroraSubnetGroup
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  EC2InstanceForTunnel:
    Type: AWS::EC2::Instance
    DependsOn:
      - LakehouseSecurityGroup
      - PublicSubnet
      - LakehouseKMSKey
      - VPC  
    Properties:
      InstanceType: t3.small
      KeyName: !Ref KeyName
      ImageId: !Ref EC2ImageId
      SecurityGroupIds:
        - !GetAtt LakehouseSecurityGroup.GroupId
      SubnetId: !Ref PublicSubnet
      BlockDeviceMappings: 
      - DeviceName: /dev/xvda
        Ebs:
          Encrypted: true
          KmsKeyId: !GetAtt LakehouseKMSKey.Arn
          VolumeType: gp2
          DeleteOnTermination: true
          VolumeSize: 8
      Tags:
        -
          Key: Project
          Value: Lakehouse

  DMSSubnetGroup:
    Type: AWS::DMS::ReplicationSubnetGroup
    DependsOn:
      - PrivateSubnet1
      - PrivateSubnet2
      - CreateDMSRole
      - VPC
    Properties:
      ReplicationSubnetGroupDescription: "Subnet group for DMS replication instance"
      ReplicationSubnetGroupIdentifier: Lakehouse-dms-subnet-group
      SubnetIds:
        - !Ref PrivateSubnet1
        - !Ref PrivateSubnet2
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  DMSReplicationInstance:
    Type: AWS::DMS::ReplicationInstance
    DependsOn:
      - DMSSubnetGroup
      - LakehouseSecurityGroup
      - AuroraSourceConnection
      - LakehouseKMSKey
      - VPC
    Properties:
      ReplicationInstanceClass: dms.t2.small
      PubliclyAccessible: false
      ReplicationSubnetGroupIdentifier: !Ref DMSSubnetGroup
      VpcSecurityGroupIds:
        - !GetAtt LakehouseSecurityGroup.GroupId
      EngineVersion: 3.4.6
      AllocatedStorage: 5
      KmsKeyId: !GetAtt LakehouseKMSKey.Arn
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  DMSSourceDBEndpoint:
    Type: AWS::DMS::Endpoint
    DependsOn:
      - AuroraCluster
      - LakehouseKMSKey
      - VPC
    Properties:
      DatabaseName: lakehouse_source_db
      EndpointType: source
      KmsKeyId: !GetAtt LakehouseKMSKey.Arn
      EngineName: aurora-postgresql
      EndpointIdentifier: lakehouse-db-source
      Password: !Ref DBMasterUserPassword
      Port: 5432
      ServerName: !GetAtt AuroraCluster.Endpoint.Address
      Username: postgres
      Tags:
        -
          Key: Project
          Value: Lakehouse
        
  DMSTargetS3Endpoint:
    Type: AWS::DMS::Endpoint
    DependsOn:
      - RawS3Bucket
      - DMSWriteToRawS3Role
      - LakehouseKMSKey
      - VPC
    Properties:
      EndpointType: target
      KmsKeyId: !GetAtt LakehouseKMSKey.Arn
      EngineName: s3
      EndpointIdentifier: lakehouse-s3-target
      S3Settings:
        BucketFolder: full-load
        BucketName: !Ref RawS3Bucket
        ServiceAccessRoleArn: !GetAtt DMSWriteToRawS3Role.Arn     
      ExtraConnectionAttributes: !Join ['', ["LakehouseDMSWriteToS3Role=", ";DataFormat=parquet;ParquetTimestampInMillisecond=true;TimestampColumnName=update_ts_dms;"]]
      Tags:
        -
          Key: Project
          Value: Lakehouse 

  CDCTargetS3Endpoint:
    Type: AWS::DMS::Endpoint
    DependsOn:
      - RawS3Bucket
      - DMSWriteToRawS3Role
      - LakehouseKMSKey
      - VPC
    Properties:
      EndpointType: target
      KmsKeyId: !GetAtt LakehouseKMSKey.Arn
      EngineName: s3
      EndpointIdentifier: lakehouse-s3-cdc
      S3Settings:
        BucketFolder: cdc-load
        BucketName: !Ref RawS3Bucket
        ServiceAccessRoleArn: !GetAtt DMSWriteToRawS3Role.Arn     
      ExtraConnectionAttributes: !Join ['', ["LakehouseDMSWriteToS3Role=", ";DataFormat=parquet;ParquetTimestampInMillisecond=true;TimestampColumnName=update_ts_dms;"]]
      Tags:
        -
          Key: Project
          Value: Lakehouse

  DMSReplicationTask:
    Type: AWS::DMS::ReplicationTask
    DependsOn:
      - DMSReplicationInstance
      - DMSSourceDBEndpoint
      - DMSTargetS3Endpoint
      - VPC
    Properties:
      MigrationType: full-load
      ReplicationInstanceArn: !Ref DMSReplicationInstance
      ReplicationTaskIdentifier: lakehouse-db-to-s3-full
      SourceEndpointArn: !Ref DMSSourceDBEndpoint
      TargetEndpointArn: !Ref DMSTargetS3Endpoint
      TableMappings: "{\"rules\":[{\"rule-type\":\"selection\",\"rule-id\":\"1\",\"rule-name\":\"1\",\"object-locator\":{\"schema-name\":\"human_resources\",\"table-name\":\"%\"},\"rule-action\":\"include\",\"filters\":[]},{\"rule-type\": \"transformation\",\"rule-id\": \"2\",\"rule-name\": \"2\",\"rule-target\": \"column\",\"object-locator\": {\"schema-name\": \"%\",\"table-name\": \"%\"},\"rule-action\": \"add-column\",\"value\": \"schema_name\",\"expression\": \"$SCHEMA_NAME_VAR\",\"data-type\": {\"type\": \"string\",\"length\": 50}},{\"rule-type\": \"transformation\",\"rule-id\": \"3\",\"rule-name\": \"3\",\"rule-target\": \"column\",\"object-locator\": {\"schema-name\": \"%\",\"table-name\": \"%\"},\"rule-action\": \"add-column\",\"value\": \"table_name\",\"expression\": \"$TABLE_NAME_VAR\",\"data-type\": {\"type\": \"string\",\"length\": 50}}]}"
      ReplicationTaskSettings: "{\"TargetMetadata\":{\"TargetSchema\":\"\",\"SupportLobs\":true,\"FullLobMode\":false,\"LobChunkSize\":0,\"LimitedSizeLobMode\":true,\"LobMaxSize\":32,\"InlineLobMaxSize\":0,\"LoadMaxFileSize\":0,\"ParallelLoadThreads\":0,\"ParallelLoadBufferSize\":0,\"BatchApplyEnabled\":false,\"TaskRecoveryTableEnabled\":false,\"ParallelLoadQueuesPerThread\":0,\"ParallelApplyThreads\":0,\"ParallelApplyBufferSize\":0,\"ParallelApplyQueuesPerThread\":0},\"FullLoadSettings\":{\"TargetTablePrepMode\":\"DROP_AND_CREATE\",\"CreatePkAfterFullLoad\":false,\"StopTaskCachedChangesApplied\":false,\"StopTaskCachedChangesNotApplied\":false,\"MaxFullLoadSubTasks\":8,\"TransactionConsistencyTimeout\":600,\"CommitRate\":10000},\"Logging\":{\"EnableLogging\":false,\"LogComponents\":[{\"Id\":\"TRANSFORMATION\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"SOURCE_UNLOAD\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"IO\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TARGET_LOAD\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"PERFORMANCE\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"SOURCE_CAPTURE\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"SORTER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"REST_SERVER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"VALIDATOR_EXT\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TARGET_APPLY\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TASK_MANAGER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TABLES_MANAGER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"METADATA_MANAGER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"FILE_FACTORY\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"COMMON\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"ADDONS\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"DATA_STRUCTURE\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"COMMUNICATION\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"FILE_TRANSFER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"}]},\"ControlTablesSettings\":{\"historyTimeslotInMinutes\":5,\"ControlSchema\":\"\",\"HistoryTimeslotInMinutes\":5,\"HistoryTableEnabled\":false,\"SuspendedTablesTableEnabled\":false,\"StatusTableEnabled\":false},\"StreamBufferSettings\":{\"StreamBufferCount\":3,\"StreamBufferSizeInMB\":8,\"CtrlStreamBufferSizeInMB\":5},\"ChangeProcessingDdlHandlingPolicy\":{\"HandleSourceTableDropped\":true,\"HandleSourceTableTruncated\":true,\"HandleSourceTableAltered\":true},\"ErrorBehavior\":{\"DataErrorPolicy\":\"LOG_ERROR\",\"DataTruncationErrorPolicy\":\"LOG_ERROR\",\"DataErrorEscalationPolicy\":\"SUSPEND_TABLE\",\"DataErrorEscalationCount\":0,\"TableErrorPolicy\":\"SUSPEND_TABLE\",\"TableErrorEscalationPolicy\":\"STOP_TASK\",\"TableErrorEscalationCount\":0,\"RecoverableErrorCount\":-1,\"RecoverableErrorInterval\":5,\"RecoverableErrorThrottling\":true,\"RecoverableErrorThrottlingMax\":1800,\"RecoverableErrorStopRetryAfterThrottlingMax\":false,\"ApplyErrorDeletePolicy\":\"IGNORE_RECORD\",\"ApplyErrorInsertPolicy\":\"LOG_ERROR\",\"ApplyErrorUpdatePolicy\":\"LOG_ERROR\",\"ApplyErrorEscalationPolicy\":\"LOG_ERROR\",\"ApplyErrorEscalationCount\":0,\"ApplyErrorFailOnTruncationDdl\":false,\"FullLoadIgnoreConflicts\":true,\"FailOnTransactionConsistencyBreached\":false,\"FailOnNoTablesCaptured\":false},\"ChangeProcessingTuning\":{\"BatchApplyPreserveTransaction\":true,\"BatchApplyTimeoutMin\":1,\"BatchApplyTimeoutMax\":30,\"BatchApplyMemoryLimit\":500,\"BatchSplitSize\":0,\"MinTransactionSize\":1000,\"CommitTimeout\":1,\"MemoryLimitTotal\":1024,\"MemoryKeepTime\":60,\"StatementCacheSize\":50},\"PostProcessingRules\":null,\"CharacterSetSettings\":null,\"LoopbackPreventionSettings\":null,\"BeforeImageSettings\":null}"
      Tags:
        -
          Key: Project
          Value: Lakehouse

  CDCReplicationTask:
    Type: AWS::DMS::ReplicationTask
    DependsOn:
      - DMSReplicationInstance
      - DMSSourceDBEndpoint
      - CDCTargetS3Endpoint
      - VPC
    Properties:
      MigrationType: cdc
      ReplicationInstanceArn: !Ref DMSReplicationInstance
      ReplicationTaskIdentifier: lakehouse-db-to-s3-cdc
      SourceEndpointArn: !Ref DMSSourceDBEndpoint
      TargetEndpointArn: !Ref CDCTargetS3Endpoint
      TableMappings: "{\"rules\":[{\"rule-type\":\"selection\",\"rule-id\":\"1\",\"rule-name\":\"1\",\"object-locator\":{\"schema-name\":\"human_resources\",\"table-name\":\"%\"},\"rule-action\":\"include\",\"filters\":[]},{\"rule-type\": \"transformation\",\"rule-id\": \"2\",\"rule-name\": \"2\",\"rule-target\": \"column\",\"object-locator\": {\"schema-name\": \"%\",\"table-name\": \"%\"},\"rule-action\": \"add-column\",\"value\": \"schema_name\",\"expression\": \"$SCHEMA_NAME_VAR\",\"data-type\": {\"type\": \"string\",\"length\": 50}},{\"rule-type\": \"transformation\",\"rule-id\": \"3\",\"rule-name\": \"3\",\"rule-target\": \"column\",\"object-locator\": {\"schema-name\": \"%\",\"table-name\": \"%\"},\"rule-action\": \"add-column\",\"value\": \"table_name\",\"expression\": \"$TABLE_NAME_VAR\",\"data-type\": {\"type\": \"string\",\"length\": 50}}]}"
      ReplicationTaskSettings: "{\"TargetMetadata\":{\"TargetSchema\":\"\",\"SupportLobs\":true,\"FullLobMode\":false,\"LobChunkSize\":0,\"LimitedSizeLobMode\":true,\"LobMaxSize\":32,\"InlineLobMaxSize\":0,\"LoadMaxFileSize\":0,\"ParallelLoadThreads\":0,\"ParallelLoadBufferSize\":0,\"BatchApplyEnabled\":false,\"TaskRecoveryTableEnabled\":false,\"ParallelLoadQueuesPerThread\":0,\"ParallelApplyThreads\":0,\"ParallelApplyBufferSize\":0,\"ParallelApplyQueuesPerThread\":0},\"FullLoadSettings\":{\"TargetTablePrepMode\":\"DROP_AND_CREATE\",\"CreatePkAfterFullLoad\":false,\"StopTaskCachedChangesApplied\":false,\"StopTaskCachedChangesNotApplied\":false,\"MaxFullLoadSubTasks\":8,\"TransactionConsistencyTimeout\":600,\"CommitRate\":10000},\"Logging\":{\"EnableLogging\":false,\"LogComponents\":[{\"Id\":\"TRANSFORMATION\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"SOURCE_UNLOAD\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"IO\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TARGET_LOAD\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"PERFORMANCE\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"SOURCE_CAPTURE\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"SORTER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"REST_SERVER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"VALIDATOR_EXT\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TARGET_APPLY\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TASK_MANAGER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"TABLES_MANAGER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"METADATA_MANAGER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"FILE_FACTORY\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"COMMON\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"ADDONS\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"DATA_STRUCTURE\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"COMMUNICATION\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"},{\"Id\":\"FILE_TRANSFER\",\"Severity\":\"LOGGER_SEVERITY_DEFAULT\"}],\"CloudWatchLogGroup\":null,\"CloudWatchLogStream\":null},\"ControlTablesSettings\":{\"historyTimeslotInMinutes\":5,\"ControlSchema\":\"\",\"HistoryTimeslotInMinutes\":5,\"HistoryTableEnabled\":false,\"SuspendedTablesTableEnabled\":false,\"StatusTableEnabled\":false},\"StreamBufferSettings\":{\"StreamBufferCount\":3,\"StreamBufferSizeInMB\":8,\"CtrlStreamBufferSizeInMB\":5},\"ChangeProcessingDdlHandlingPolicy\":{\"HandleSourceTableDropped\":true,\"HandleSourceTableTruncated\":true,\"HandleSourceTableAltered\":true},\"ErrorBehavior\":{\"DataErrorPolicy\":\"LOG_ERROR\",\"DataTruncationErrorPolicy\":\"LOG_ERROR\",\"DataErrorEscalationPolicy\":\"SUSPEND_TABLE\",\"DataErrorEscalationCount\":0,\"TableErrorPolicy\":\"SUSPEND_TABLE\",\"TableErrorEscalationPolicy\":\"STOP_TASK\",\"TableErrorEscalationCount\":0,\"RecoverableErrorCount\":-1,\"RecoverableErrorInterval\":5,\"RecoverableErrorThrottling\":true,\"RecoverableErrorThrottlingMax\":1800,\"RecoverableErrorStopRetryAfterThrottlingMax\":false,\"ApplyErrorDeletePolicy\":\"IGNORE_RECORD\",\"ApplyErrorInsertPolicy\":\"LOG_ERROR\",\"ApplyErrorUpdatePolicy\":\"LOG_ERROR\",\"ApplyErrorEscalationPolicy\":\"LOG_ERROR\",\"ApplyErrorEscalationCount\":0,\"ApplyErrorFailOnTruncationDdl\":false,\"FullLoadIgnoreConflicts\":true,\"FailOnTransactionConsistencyBreached\":false,\"FailOnNoTablesCaptured\":false},\"ChangeProcessingTuning\":{\"BatchApplyPreserveTransaction\":true,\"BatchApplyTimeoutMin\":1,\"BatchApplyTimeoutMax\":30,\"BatchApplyMemoryLimit\":500,\"BatchSplitSize\":0,\"MinTransactionSize\":1000,\"CommitTimeout\":1,\"MemoryLimitTotal\":1024,\"MemoryKeepTime\":60,\"StatementCacheSize\":50},\"PostProcessingRules\":null,\"CharacterSetSettings\":null,\"LoopbackPreventionSettings\":null,\"BeforeImageSettings\":null}"
      Tags:
        -
          Key: Project
          Value: Lakehouse

  AuroraSourceConnection:
    Type: AWS::Glue::Connection
    DependsOn:
      - PrivateSubnet1
      - PrivateSubnet2
      - LakehouseSecurityGroup
      - AuroraCluster
      - VPC
    Properties:
      CatalogId: !Ref AWS::AccountId
      ConnectionInput:
        ConnectionType: JDBC
        Description: This connection is used in Glue job definitions
        ConnectionProperties:
          JDBC_CONNECTION_URL: !Join ['', ["jdbc:postgresql://", !GetAtt AuroraCluster.Endpoint.Address, ":5432/lakehouse_source_db"]]
          USERNAME: postgres
          PASSWORD: !Ref DBMasterUserPassword
        Name: Lakehouse-db-connection
        PhysicalConnectionRequirements:
          SubnetId: !Ref PrivateSubnet2
          AvailabilityZone: !GetAtt PrivateSubnet2.AvailabilityZone
          SecurityGroupIdList:
            - !GetAtt LakehouseSecurityGroup.GroupId

  ExecuteStarterTasksLambdaFn:
    Type: AWS::Lambda::Function
    DependsOn:
      - ExecuteLambdaFnsRole
      - TriggerLambdaFns
      - LakehouseKMSKey
      - VPC
    Properties:
      Code:
        ZipFile: |
          from time import sleep
          import boto3
          import json
          import urllib3
          
          dms = boto3.client('dms')
          glue = boto3.client('glue')
          s3 = boto3.resource('s3')
          s3_client = boto3.client('s3')
          http = urllib3.PoolManager()
          
          def lambda_handler(event, context):
            print('Event: %s' % json.dumps(event))
            response = { 'StackId': event['StackId'], 'RequestId': event['RequestId'], 'LogicalResourceId': event['LogicalResourceId'] }
            resProps = event['ResourceProperties']
            if event['RequestType'] == 'Create':
              try:
                sleep(30)
                testConnResp = dms.test_connection(ReplicationInstanceArn=resProps['ReplInst'], EndpointArn=resProps['SrcEP'])
                while True:
                  descConnResp = dms.describe_connections(Filters=[{'Name': 'endpoint-arn', 'Values': [resProps['SrcEP']] }])
                  descConnStatus = descConnResp['Connections'][0]['Status']
                  sleep(5)
                  print('Polling Src conn check. Current status is:' + descConnStatus)
                  if descConnStatus == 'successful':
                    break
                print('Start copying jars from aws-bigdata-blog')
                for i in s3_client.list_objects_v2(Bucket='aws-bigdata-blog',Prefix='artifacts/iceberg-on-glue/')['Contents']:
                  src = {'Bucket': 'aws-bigdata-blog','Key': i['Key']}
                  print('Source object list:' + i['Key'])
                  s3.meta.client.copy(src, resProps['S3Dep'], i['Key'])
                  print('Copy complated :' + i['Key'])
                response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'InitSetupDone' })
              except Exception as e:
                response.update({ 'Status': 'FAILED', 'Reason': str(e), 'PhysicalResourceId': 'InitSetupFailed' })
            if event['RequestType'] == 'Delete':
              try:
                s3.Bucket(resProps['S3Dep']).objects.delete()
                s3.Bucket(resProps['S3Raw']).objects.delete()
                s3.Bucket(resProps['S3Curated']).objects.delete()
                descReplTaskResp = dms.describe_replication_tasks(Filters=[{'Name': 'replication-task-arn', 'Values': [resProps['ReplTsk']] }])
                dmsTaskStatus = descReplTaskResp['ReplicationTasks'][0]['Status']
                if dmsTaskStatus == 'running':
                  stopTaskResp = dms.stop_replication_task(ReplicationTaskArn=resProps['ReplTsk'])
                while True:
                  descReplTaskResp = dms.describe_replication_tasks(Filters=[{'Name': 'replication-task-arn', 'Values': [resProps['ReplTsk']] }])
                  dmsTaskStatus = descReplTaskResp['ReplicationTasks'][0]['Status']
                  sleep(5)
                  print('Polling DMS task. Current status is:' + dmsTaskStatus)
                  if dmsTaskStatus == 'stopped':
                    break
                  elif dmsTaskStatus == 'ready':
                    break
                try:
                  glue.delete_database(Name='human_resources')
                except Exception as e:
                  if e.response['Error']['Code'] == 'EntityNotFoundException':
                    pass
                response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'InitSetupDelete' })
                sleep(90)
              except Exception as e:
                response.update({ 'Status': 'FAILED', 'Reason': str(e), 'PhysicalResourceId': 'InitSetupDeleteFailed' })
            if event['RequestType'] == 'Update':
              response.update({ 'Status': 'SUCCESS', 'PhysicalResourceId': 'InitSetupNothingChanged' })
            resp = http.request('PUT', event['ResponseURL'], body = json.dumps(response), headers = {'Content-Type': 'application/json'}, retries = False)
            print('http resp:' + resp.data.decode('utf-8'))
            print(response)
            return(response)
      FunctionName: invoke-dms-tasks
      KmsKeyArn: !GetAtt LakehouseKMSKey.Arn
      MemorySize: 128
      Runtime: python3.6
      Description: Lambda function that executes starter tasks such as testing the DMS aurora connection and copy the jar files from the public bucket to the local bucket for iceberg glue job to use.
      Handler: index.lambda_handler
      Role: !GetAtt ExecuteLambdaFnsRole.Arn
      Timeout: 900      
      Tags:
        -
          Key: Project
          Value: Lakehouse

  ExecuteStarterTasks:
    Type: Custom::ExecuteStarterTasks
    DependsOn:
      - AuroraDBInstance1
      - AuroraCluster
      - ExecuteStarterTasksLambdaFn
      - ScriptsAndTempS3Bucket
      - RawS3Bucket
      - CuratedS3Bucket
      - DMSReplicationInstance
      - DMSSourceDBEndpoint
      - ExecuteLambdaFnsRole
      - TriggerLambdaFns
      - DMSTargetS3Endpoint
      - DMSReplicationTask
      - CDCTargetS3Endpoint
      - CDCReplicationTask
      - VPC   
    Properties:
      ServiceToken: !GetAtt ExecuteStarterTasksLambdaFn.Arn
      S3Dep: !Ref ScriptsAndTempS3Bucket
      S3Raw: !Ref RawS3Bucket
      S3Curated: !Ref CuratedS3Bucket
      ReplInst: !Ref DMSReplicationInstance
      SrcEP: !Ref DMSSourceDBEndpoint
      ReplTsk: !Ref DMSReplicationTask

  GlueJobsSecurityConf:
    Type: AWS::Glue::SecurityConfiguration
    Properties: 
      EncryptionConfiguration: 
        CloudWatchEncryption:
          CloudWatchEncryptionMode: SSE-KMS
          KmsKeyArn: !GetAtt LakehouseKMSKey.Arn
        JobBookmarksEncryption:
          JobBookmarksEncryptionMode: CSE-KMS
          KmsKeyArn: !GetAtt LakehouseKMSKey.Arn
        S3Encryptions:
          - 
            S3EncryptionMode: SSE-KMS
            KmsKeyArn: !GetAtt LakehouseKMSKey.Arn
      Name: GlueJobsSecurityConf

  EMRSparkIcebergCluster:
    DependsOn:
      - EMRRole
      - EMREC2Role
      - EMREC2InstanceProfile
    Type: AWS::EMR::Cluster
    Properties:
      Name: Iceberg Spark EMR Cluster
      JobFlowRole: !Ref EMREC2InstanceProfile
      # JobFlowRole: "EMR_EC2_DefaultRole"
      ReleaseLabel: emr-6.5.0
      ScaleDownBehavior: TERMINATE_AT_TASK_COMPLETION
      LogUri: !Join ['', ["s3://", !Ref ScriptsAndTempS3Bucket, "/elasticmapreduce/"]]
      ServiceRole: !Ref EMRRole
      # ServiceRole: "EMR_DefaultRole"
      # AutoScalingRole: "EMR_AutoScaling_DefaultRole"
      Tags:
        - Key: 'Project'
          Value: 'Lakehouse'
      VisibleToAllUsers: true
      Applications:
        - Name: Spark
        # - Name: Hive
        - Name: JupyterEnterpriseGateway
      Configurations:
        - Classification: iceberg-defaults
          ConfigurationProperties:
            "iceberg.enabled": "true"      
      EbsRootVolumeSize: 10
      Instances:  
        Ec2KeyName: !Ref KeyName
        Ec2SubnetId: !Ref PrivateSubnet2
        MasterInstanceGroup:
          EbsConfiguration:
            EbsBlockDeviceConfigs:
              - VolumeSpecification:
                  SizeInGB: 64
                  VolumeType: gp2
          InstanceCount: 1
          InstanceType: m5.xlarge
          Market: ON_DEMAND
          Name: 'Master instance group'
        #CoreInstanceGroup:
        #  EbsConfiguration:
        #    EbsBlockDeviceConfigs:
        #      - VolumeSpecification:
        #          SizeInGB: 100
        #          VolumeType: gp2
        #  InstanceCount: 3
        #  InstanceType: m5.xlarge
        #  Market: ON_DEMAND
        #  Name: 'Core instance group'

  EMRRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2008-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: elasticmapreduce.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: /
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole'
  EMREC2Role:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: 2008-10-17
        Statement:
          - Sid: ''
            Effect: Allow
            Principal:
              Service: ec2.amazonaws.com
            Action: 'sts:AssumeRole'
      Path: /
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role'
        # - 'arn:aws:iam::aws:policy/service-role/AmazonEC2RoleforSSM'
        - 'arn:aws:iam::aws:policy/AmazonSSMManagedInstanceCore'
        # - 'arn:aws:iam::aws:policy/AmazonSSMFullAccess'
        # - 'arn:aws:iam::aws:policy/service-role/AmazonSSMAutomationRole'
        # - 'arn:aws:iam::aws:policy/AmazonSSMPatchAssociation'        
  EMREC2InstanceProfile:
    Type: 'AWS::IAM::InstanceProfile'
    Properties:
      Path: /
      Roles:
        - !Ref EMREC2Role
Outputs:
  PublicIPOfEC2InstanceForTunnel:
    Description: Public IP of the EC2 instance that can be used as a jump box
    Value: !GetAtt EC2InstanceForTunnel.PublicIp

  EndpointOfAuroraCluster:
    Description: Endpoint address of the source Aurora cluster
    Value: !GetAtt AuroraCluster.Endpoint.Address

  RawS3Bucket:
    Description: Amazon S3 Bucket for the raw data
    Value: !Ref RawS3Bucket

  CuratedS3Bucket:
    Description: Amazon S3 Bucket for the curated data
    Value: !Ref CuratedS3Bucket 

  ScriptsAndTempS3Bucket:
    Description: Amazon S3 Bucket for the script files
    Value: !Ref ScriptsAndTempS3Bucket

  ExecuteGlueJobRole:
    Description: AWS Glue Job Role Name
    Value: !Ref ExecuteGlueJobRole

  EmrSparkClusterId:
    Value: !Ref EMRSparkIcebergCluster    