AWSTemplateFormatVersion: 2010-09-09
Parameters:
  CFNJobName1:
    Type: String
    Default: multiple_job_bookmark_keys
  CFNJobName2:
    Type: String
    Default: parameterised_job_bookmark_keys
  KmsKeyId:
    Type: String
    Default: alias/aws/s3
  TrafficType:
    Description: 'The type of traffic to log.'
    Type: String
    Default: REJECT
    AllowedValues:
    - ACCEPT
    - REJECT
    - ALL
Resources:
  BlogVPC:
    Type: 'AWS::EC2::VPC'
    Properties:
      CidrBlock: 10.2.0.0/24
      EnableDnsHostnames: 'true'
      EnableDnsSupport: 'true'
      Tags:
        - Key: Name
          Value: 'glueblog'
 
  BlogSubnet1:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select
        - '0'
        - !GetAZs ''
      CidrBlock: 10.2.0.0/26
      VpcId: !Ref BlogVPC
      Tags:
        - Key: Name
          Value: 'glueblog-subnet1'
    DependsOn: BlogVPC
  BlogSubnet2:
    Type: 'AWS::EC2::Subnet'
    Properties:
      AvailabilityZone: !Select
        - '1'
        - !GetAZs ''
      CidrBlock: 10.2.0.128/26
      VpcId: !Ref BlogVPC
      Tags:
        - Key: Name
          Value: 'glueblog-subnet2'
    DependsOn: BlogVPC
  MyDBSubnetGroup:
    Type: 'AWS::RDS::DBSubnetGroup'
    Properties:
      DBSubnetGroupDescription: Subnet available for the DB Instance
      SubnetIds:
        - !Ref BlogSubnet1
        - !Ref BlogSubnet2
    DependsOn: [BlogSubnet1, BlogSubnet2]
 

  RouteTable1:
    Type: 'AWS::EC2::RouteTable'
    Properties:
      Tags:
        - Key: Name
          Value: 'glueblog-routetable'
      VpcId: !Ref BlogVPC
    DependsOn: BlogVPC
  SubnetRouteTableAssociation1:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      RouteTableId: !Ref RouteTable1
      SubnetId: !Ref BlogSubnet1
    DependsOn: [BlogSubnet1, RouteTable1]
 
  RouteTable2: #For this Route Table S3 endpoint won't be added!!!
    Type: 'AWS::EC2::RouteTable'
    Properties:
      Tags:
        - Key: Name
          Value: 'glueblog-routetable-cloud9'
      VpcId: !Ref BlogVPC
    DependsOn: BlogVPC
  SubnetRouteTableAssociation2:
    Type: 'AWS::EC2::SubnetRouteTableAssociation'
    Properties:
      RouteTableId: !Ref RouteTable2
      SubnetId: !Ref BlogSubnet2
    DependsOn: [BlogSubnet2, RouteTable2]
 
  Role:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service: 'vpc-flow-logs.amazonaws.com'
          Action: 'sts:AssumeRole'
      Policies:
      - PolicyName: 'flowlogs-policy'
        PolicyDocument:
          Version: '2012-10-17'
          Statement:
          - Effect: Allow
            Action:
            - 'logs:CreateLogStream'
            - 'logs:PutLogEvents'
            - 'logs:DescribeLogGroups'
            - 'logs:DescribeLogStreams'
            Resource: !GetAtt 'LogGroup.Arn'
  VPCFlowLogs:
    Type: AWS::EC2::FlowLog
    Properties:
      DeliverLogsPermissionArn: !GetAtt 'Role.Arn'
      LogGroupName: !Ref LogGroup
      ResourceId: !Ref BlogVPC
      ResourceType: VPC
      TrafficType: !Ref TrafficType
     
  LogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      RetentionInDays: 30
 
  S3Endpoint:
    Type: 'AWS::EC2::VPCEndpoint'
    Properties:
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal: '*'
            Action: '*'
            Resource: '*'
      ServiceName: !Sub 'com.amazonaws.${AWS::Region}.s3'
      RouteTableIds: [!Ref RouteTable1]
      VpcEndpointType: Gateway
      VpcId: !Ref BlogVPC
    DependsOn: [BlogVPC, RouteTable1] #Added only to Route1/RouteTable1
 
  DefaultVPCSecurityGroup:
    Type: 'AWS::EC2::SecurityGroup'
    Properties:
      GroupDescription: Default Security Group for the VPC and .
      VpcId: !Ref BlogVPC
      SecurityGroupEgress:
        - CidrIp: 0.0.0.0/0
          IpProtocol: '-1'
      Tags:
        - Key: Name
          Value: 'glueblog-sg'
    DependsOn: BlogVPC
 
  DefaultVPCSecurityGroupSelfRefIngress:
    Type: 'AWS::EC2::SecurityGroupIngress'
    Properties:
      SourceSecurityGroupId: !Ref DefaultVPCSecurityGroup
      Description : 'Self referencing security group to allow traffic from AWS Glue jobs'
      IpProtocol: '-1'
      GroupId: !Ref DefaultVPCSecurityGroup
    DependsOn: DefaultVPCSecurityGroup
 
  GlueVpcEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
      ServiceName: !Join
        - ''
        - - com.amazonaws.
          - !Ref 'AWS::Region'
          - .glue
      SubnetIds:
        - !Ref BlogSubnet1
        - !Ref BlogSubnet2
      VpcId: !Ref BlogVPC
      VpcEndpointType: Interface
      SecurityGroupIds:
        - !Ref DefaultVPCSecurityGroup
      PrivateDnsEnabled: true
    DependsOn: [BlogVPC, BlogSubnet1, BlogSubnet2, DefaultVPCSecurityGroup]
   
  MyRDSSecret:
    Type: AWS::SecretsManager::Secret
    Properties:
      Name: glueblog/rdspostgres/credentials
      Description: This is a Secrets Manager secret for a RDS
      GenerateSecretString:
        SecretStringTemplate: '{"username": "postgres"}'
        GenerateStringKey: password
        PasswordLength: 16
        ExcludeCharacters: "\"'@/\\"
 
  RDSPostgreSQL:
    Type: AWS::RDS::DBInstance
    Properties:
      DBName: 'postgres'
      Engine: 'postgres'
      StorageEncrypted: true
      DeletionProtection: true
      EngineVersion: 14.6
      MasterUsername:
        Fn::Sub: "{{resolve:secretsmanager:${MyRDSSecret}::username}}"
      MasterUserPassword:
        Fn::Sub: "{{resolve:secretsmanager:${MyRDSSecret}::password}}"
      DBInstanceClass: 'db.t3.micro'
      AllocatedStorage: 50
      StorageType: gp2
      PubliclyAccessible: false
      AvailabilityZone: !GetAtt
        - BlogSubnet1
        - AvailabilityZone
      DBSubnetGroupName: !Ref MyDBSubnetGroup
      VPCSecurityGroups:
        - !Ref DefaultVPCSecurityGroup
    DependsOn: [DefaultVPCSecurityGroup, MyDBSubnetGroup, BlogSubnet1, MyRDSSecret]
 
  SecretRDSAttachment:
    Type: AWS::SecretsManager::SecretTargetAttachment
    Properties:
      SecretId:
        Ref: MyRDSSecret
      TargetId:
        Ref: RDSPostgreSQL
      TargetType: AWS::RDS::DBInstance
    DependsOn: [RDSPostgreSQL, MyRDSSecret]
 
  GlueRDSConnection:
    Type: AWS::Glue::Connection
    Properties:
      CatalogId: !Ref AWS::AccountId
      ConnectionInput:
        Name: RDSPostgresConnection
        ConnectionProperties:
          Type: JDBC
          JDBC_CONNECTION_URL: !Sub "jdbc:postgresql://${RDSPostgreSQL.Endpoint.Address}:${RDSPostgreSQL.Endpoint.Port}/${RDSPostgreSQL.DBName}"
          SECRET_ID: {"Fn::Select": ["0", {"Fn::Split": ["-", {"Fn::Select": ["6", { "Fn::Split": [ ":", { "Ref": "MyRDSSecret" }  ]  } ] }  ] } ] }
        ConnectionType: JDBC
        PhysicalConnectionRequirements:
          AvailabilityZone: !GetAtt
          - BlogSubnet1 #HAS TO BE SUBNET1 WHICH HAS S3 ENDPOINT IN ITS ROUTE!!!
          - AvailabilityZone
          SecurityGroupIdList:
            - !Ref DefaultVPCSecurityGroup
          SubnetId: !Ref BlogSubnet1
    DependsOn: [RDSPostgreSQL]
 
  MyS3Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName:
        Fn::Join:
          - "-"
          - - job-bookmark-keys-demo
            - Ref: AWS::AccountId
      LoggingConfiguration:
        DestinationBucketName: !Ref LoggingBucket
        LogFilePrefix: blog-logs
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: aws:kms
              KMSMasterKeyID: !Ref KmsKeyId
    DeletionPolicy: Delete
   
  MyS3BucketPolicy:
    Type: AWS::S3::BucketPolicy
    Properties:
      Bucket:  !Ref MyS3Bucket
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Deny
            Principal: "*"
            Action: [
                     "s3:GetObject",
                     "s3:PutObject",
                     "s3:DeleteObject"]
            Resource:
                - !Sub "arn:aws:s3:::${MyS3Bucket}/*"
            Condition:
                Bool:
                  aws:SecureTransport: false
          - Effect: Allow
            Principal:
                Service:
                  - glue.amazonaws.com
                  - lambda.amazonaws.com
            Action: [
                     "s3:GetObject",
                     "s3:PutObject",
                     "s3:DeleteObject"]
            Resource:
                - !Sub "arn:aws:s3:::${MyS3Bucket}/*"
  LoggingBucket:
      Type: 'AWS::S3::Bucket'
  S3BucketPolicy:
    Type: 'AWS::S3::BucketPolicy'
    Properties:
      Bucket: !Ref LoggingBucket
      PolicyDocument:
        Version: 2012-10-17
        Statement:
          - Action:
              - 's3:PutObject'
            Effect: Allow
            Principal:
              Service: logging.s3.amazonaws.com
            Resource: !Join
              - ''
              - - 'arn:aws:s3:::'
                - !Ref LoggingBucket
                - /*
            Condition:
              ArnLike:
                'aws:SourceArn': !GetAtt
                  - MyS3Bucket
                  - Arn
              StringEquals:
                'aws:SourceAccount': !Sub '${AWS::AccountId}'
 
  GlueServiceRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service:
                  - glue.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
      - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
      - arn:aws:iam::aws:policy/SecretsManagerReadWrite
      Policies:
      - PolicyName: 'ats-permission-policy'
        PolicyDocument:
            Version: 2012-10-17
            Statement:
              - Effect: Allow
                Action:
                  - s3:GetObject
                  - s3:PutObject
                  - iam:PassRole
                Resource:
                  - !Sub 'arn:aws:s3:::job-bookmark-keys-demo-${AWS::AccountId}/*'
  GlueJob1:
    Type: AWS::Glue::Job
    Properties:
      Role: !Ref GlueServiceRole
      Description: Job created with CloudFormation
      GlueVersion: "3.0"
      Command:
        Name: glueetl
        ScriptLocation: !Sub s3://job-bookmark-keys-demo-${AWS::AccountId}/scenario_1_job.py
      ExecutionProperty:
        MaxConcurrentRuns: 1
      Name: !Ref CFNJobName1
      NumberOfWorkers: 2
      WorkerType: "G.1X"
      Timeout: 5
      Connections:
        Connections:
          - !Ref GlueRDSConnection
      DefaultArguments:
        "--s3_bucket": !Sub "job-bookmark-keys-demo-${AWS::AccountId}"
        "--enable-glue-datacatalog": true
        "--job-bookmark-option": 'job-bookmark-enable'
  GlueSenDataDetectJob2:
    Type: AWS::Glue::Job
    Properties:
      Role: !Ref GlueServiceRole
      Description: Job created with CloudFormation
      GlueVersion: "3.0"
      Command:
        Name: glueetl
        ScriptLocation: !Sub s3://job-bookmark-keys-demo-${AWS::AccountId}/scenario_2_job.py
      ExecutionProperty:
        MaxConcurrentRuns: 1
      Name: !Ref CFNJobName2
      NumberOfWorkers: 2
      WorkerType: "G.1X"
      Timeout: 5
      Connections:
        Connections:
          - !Ref GlueRDSConnection
      DefaultArguments:
        "--s3_bucket": !Sub "job-bookmark-keys-demo-${AWS::AccountId}"
        "--enable-glue-datacatalog": true
        "--table_name": address
        "--bookmarkkey": address_id
        "--job-bookmark-option": 'job-bookmark-enable'
       
 
  LambdaVpcEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
      ServiceName: !Join
        - ''
        - - com.amazonaws.
          - !Ref 'AWS::Region'
          - .lambda
      SubnetIds:
        - !Ref BlogSubnet1
        - !Ref BlogSubnet2
      VpcId: !Ref BlogVPC
      VpcEndpointType: Interface
      SecurityGroupIds:
        - !Ref DefaultVPCSecurityGroup
      PrivateDnsEnabled: true
    DependsOn: [BlogVPC, BlogSubnet1, BlogSubnet2, DefaultVPCSecurityGroup]
   
  SecretsManagerVpcEndpoint:
    Type: AWS::EC2::VPCEndpoint
    Properties:
      ServiceName: !Join
        - ''
        - - com.amazonaws.
          - !Ref 'AWS::Region'
          - .secretsmanager
      SubnetIds:
        - !Ref BlogSubnet1
        - !Ref BlogSubnet2
      VpcId: !Ref BlogVPC
      VpcEndpointType: Interface
      SecurityGroupIds:
        - !Ref DefaultVPCSecurityGroup
      PrivateDnsEnabled: true
    DependsOn: [BlogVPC, BlogSubnet1, BlogSubnet2, DefaultVPCSecurityGroup]
       
        
  LambdaLayer:
   Type: AWS::Lambda::LayerVersion
   Properties:
    CompatibleArchitectures:
       - arm64
       - x86_64
    CompatibleRuntimes:
       - python3.9
    Content:
       S3Bucket: aws-blogs-artifacts-public
       S3Key: artifacts/BDB-2907/lambdalayer/psycopg2layer.zip
    Description: artifacts/BDB-2907/lambdalayer/psycopg2layer.zip
    LayerName: psycopg2
 
  LambdaRDSDDLExecuteRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: MyLambdaExecutionRole
      AssumeRolePolicyDocument:
        Version: 2012-10-17
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
         - arn:aws:iam::aws:policy/AWSLambdaExecute
         - arn:aws:iam::aws:policy/SecretsManagerReadWrite
         - arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole
      Policies:
        -
          PolicyName: S3Policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - 's3:GetBucketLocation'
                  - 's3:ListBucket'
                  - 's3:GetBucketAcl'
                  - 's3:GetObject'
                  - 's3:PutObject'
                  - 's3:DeleteObject'
                Resource:
                  - !Sub 'arn:aws:s3:::job-bookmark-keys-demo-${AWS::AccountId}'
                  - !Sub 'arn:aws:s3:::job-bookmark-keys-demo-${AWS::AccountId}/*'
               
  LambdaRDSDDLExecute:
    Type: AWS::Lambda::Function
    DependsOn: LambdaRDSDDLExecuteRole
    Properties:
      FunctionName: LambdaRDSDDLExecute
      Runtime: python3.9
      Handler: index.handler
      Layers:
        - !Ref LambdaLayer
      Role: !GetAtt LambdaRDSDDLExecuteRole.Arn
      VpcConfig:
        SecurityGroupIds:
          - !Ref DefaultVPCSecurityGroup
        SubnetIds:
          - !Ref BlogSubnet1
          - !Ref BlogSubnet2
      Code:
        ZipFile: |
            import sys
            import sys
            import logging
            #import rds_config
            import psycopg2
            import boto3
            import json
 
            logger = logging.getLogger()
            logger.setLevel(logging.INFO)
 
            def get_secret():
                secret_name = "glueblog/rdspostgres/credentials"
                region_name = "us-east-1"
                # Create a Secrets Manager client
                session = boto3.session.Session()
                client = session.client(
                    service_name='secretsmanager',
                    region_name=region_name
                )
 
                get_secret_value_response = client.get_secret_value(
                        SecretId=secret_name
                    )
                # Decrypts secret using the associated KMS key.
                secret = get_secret_value_response['SecretString']
               
                return json.loads(secret)
 
            secretvalue=get_secret()
 
            try:
 
                rds_username=secretvalue["username"]
                rds_host=secretvalue["host"]
                rds_user_pwd=secretvalue["password"]
                rds_db_name=secretvalue["dbname"]
                conn_string = "host=%s user=%s password=%s dbname=%s" % (rds_host, rds_username, rds_user_pwd, rds_db_name)
                conn = psycopg2.connect(conn_string)
 
            except:
                logger.error("ERROR: Could not connect to Postgres instance.")
                sys.exit()
 
                logger.info("SUCCESS: Connection to RDS Postgres instance succeeded")
 
            def handler(event, context):
                """
                This function fetches content from RDS instance
                """
 
                item_count = 0
 
                with conn.cursor() as cur:
                    cur.execute("DROP TABLE IF EXISTS product")
                    cur.execute("CREATE TABLE IF NOT EXISTS product ( product_id integer, version integer, product_name character varying(50), product_description character varying(100) )")
                    print('product table created')
                    cur.execute("insert into product(product_id,version,product_name,product_description) Select 44,1,'V-Neck T-Shirt','Stylish V Neck Tshirt' union all Select 45,1,'Hoodie','Hoodie to keep warm during Winter'")
                    print('rows inserted to product table')
                    cur.execute("DROP TABLE IF EXISTS address")
                    cur.execute("CREATE TABLE IF NOT EXISTS address ( address_id integer, address character varying(100), district character varying(50), postcode character varying(10) )")
                    print('address table created')
                    cur.execute("insert into address(address_id,address,district,postcode) Select 1,'28 MySQL Boulevard','QLD','17886' union all Select 2,'48 Mysakila Drive','Alberta','35200'")
                    print('rows inserted to address table')
                    conn.commit()
 
                    cur.execute("select 'Product',* from product")
                    for row in cur:
                        item_count += 1
                        #logger.info(row)
                        print(row)
   
                    cur.execute("select 'Address',* from address")
                    for row in cur:
                        item_count += 1
                        #logger.info(row)
                        print(row)
                conn.commit()
 
                return "Added items from RDS table"
   
  LambdaRDSDDLExecuteIncremental:
    Type: AWS::Lambda::Function
    DependsOn: LambdaRDSDDLExecuteRole
    Properties:
      FunctionName: LambdaRDSDDLExecute_incremental
      Runtime: python3.9
      Handler: index.handler
      Layers:
        - !Ref LambdaLayer
      Role: !GetAtt LambdaRDSDDLExecuteRole.Arn
      VpcConfig:
        SecurityGroupIds:
          - !Ref DefaultVPCSecurityGroup
        SubnetIds:
          - !Ref BlogSubnet1
          - !Ref BlogSubnet2
      Code:
        ZipFile: |
            import sys
            import sys
            import logging
            #import rds_config
            import psycopg2
            import boto3
            import json
 
            logger = logging.getLogger()
            logger.setLevel(logging.INFO)
 
            def get_secret():
                secret_name = "glueblog/rdspostgres/credentials"
                region_name = "us-east-1"
                # Create a Secrets Manager client
                session = boto3.session.Session()
                client = session.client(
                    service_name='secretsmanager',
                    region_name=region_name
                )
 
                get_secret_value_response = client.get_secret_value(
                        SecretId=secret_name
                    )
                # Decrypts secret using the associated KMS key.
                secret = get_secret_value_response['SecretString']
           
                return json.loads(secret)
 
            secretvalue=get_secret()
 
            try:
 
                rds_username=secretvalue["username"]
                rds_host=secretvalue["host"]
                rds_user_pwd=secretvalue["password"]
                rds_db_name=secretvalue["dbname"]
                conn_string = "host=%s user=%s password=%s dbname=%s" % (rds_host, rds_username, rds_user_pwd, rds_db_name)
                conn = psycopg2.connect(conn_string)
 
            except:
                logger.error("ERROR: Could not connect to Postgres instance.")
                sys.exit()
 
                logger.info("SUCCESS: Connection to RDS Postgres instance succeeded")
 
            def handler(event, context):
                """
                This function fetches content from RDS instance
                """
 
                item_count = 0
 
                with conn.cursor() as cur:
 
                    cur.execute("insert into product(product_id,version,product_name,product_description) Select 45,2,'V-Neck T-Shirt','Stylish V Neck Tshirt with Logo'")
                    print('rows inserted to product table')
                    conn.commit()
   
                    cur.execute("select 'product',* from product")
                    for row in cur:
                        item_count += 1
                        #logger.info(row)
                        print(row)
       
                conn.commit()
 
                return "Added items from RDS table"