AWSTemplateFormatVersion: "2010-09-09"
Description: "CloudFormation Template to deploy ETL Orchestration using Lambda and Redshift Data API for Amazon Redshift"
Parameters:
  inputGlueETLJob:
    Description: Glue job name which will be invoked from lambda
    Type: String
    Default: 'bq-migration-ETL'
  inputGlueMetaJob:
    Description: Glue job name which will be invoked from lambda
    Type: String
    Default: 'bq-get-metadata'
  inputBucketName:
    Description: Bucket name where data and metadata will be stored
    Type: String
  SourceS3BucketName:
    Description: Bucket name where the blog artifact are stored do not change the default!
    Type: String
    Default: 'aws-blogs-artifacts-public'
  StateMachineName:
    Description: AWS StepFunctions State Machine name to be used 
    Type: String
    Default: 'bq-to-s3-migration-state-machine'
  inputGlueConnectionName:
    Description: Connection name of the BQ connector
    Type: String
    Default: 'glue-bq-connector-24'
  inputDynamoDBTablePrefix:
    Description: Dynamo DB table name
    Type: String
    Default: 'bq_to_s3'
  GlueS3PathURI:
    Description: AWS Glue Script path (S3 URI) on Amazon S3
    Type: String
    Default: 's3://aws-glue-scripts-${AWS::Account}-${AWS::Region}/admin/'
  InputBQSecret:
    Description: The secret name of the AWS Secret Manager secret in which you stored the Google BigQuery Credential
    Type: String
  InputBQProjectName:
    Description: The name of your project in bigquery in which you want to store temp tables you need write permissions on it
    Type: String
  InputMaxParallelism:
    Description: The maximum number of table you want to migrate concurrently
    Type: Number
    Default: 30
Metadata:
  AWS::CloudFormation::Interface:
    ParameterGroups:
      -
        Label:
          default: "Input Parameters"
        Parameters:
        - inputGlueConnectionName
        - GlueS3PathURI
        - inputGlueETLJob
        - inputGlueMetaJob
        - InputMaxParallelism
        - SourceS3BucketName
        - StateMachineName

Resources:


#Roles
#Role for StepFunction state machine
  StateMachineExecutionRole:
      Type: 'AWS::IAM::Role'
      Properties:
        Description : IAM Role for the state machine in step function to run
        AssumeRolePolicyDocument:
          Version: 2012-10-17
          Statement:
            - Effect: Allow
              Principal:
                Service:
                  - !Sub 'states.${AWS::Region}.amazonaws.com'
              Action: 'sts:AssumeRole'
        Path: /
        Policies:

          -
            PolicyName: DynamoDBAccessPolicy
            PolicyDocument:
              Version: 2012-10-17
              Statement:
                -
                  Effect: Allow
                  Action:
                   - dynamodb:DescribeStream
                   - dynamodb:GetRecords
                   - dynamodb:GetShardIterator
                   - dynamodb:ListStreams
                   - dynamodb:GetItem
                   - dynamodb:PutItem
                   - dynamodb:Scan
                   - dynamodb:Query
                   - dynamodb:UpdateItem
                   - dynamodb:BatchWriteItem
                   - dynamodb:DeleteItem
                  Resource: !Sub 'arn:aws:dynamodb:*:*:table/${inputDynamoDBTablePrefix}*'
          -
            PolicyName: s3EventNotification
            PolicyDocument:
              Version: 2012-10-17
              Statement:
                -
                  Effect: Allow
                  Action:
                    - 's3:GetBucketNotification'
                    - 's3:PutBucketNotification'
                    - 's3:ListObjectsV2'
                    - 's3:GetObject'
                    - 's3:PutObject'
                  Resource:
                    - !Sub 'arn:aws:s3:::${inputBucketName}'
                    - !Sub 'arn:aws:s3:::${inputBucketName}/*'
                -
                  Effect: Allow
                  Action:
                    - 'logs:CreateLogGroup'
                    - 'logs:CreateLogStream'
                    - 'logs:PutLogEvents'
                  Resource: 'arn:aws:logs:*:*:*'
                -
                  Effect: Allow
                  Action:
                    - 'lambda:InvokeFunction'
                  Resource:
                    - !Sub arn:aws:lambda:${AWS::Region}:${AWS::AccountId}:function:*
          -
             PolicyName: GlueAccessPolicy
             PolicyDocument:
              Version: 2012-10-17
              Statement:
                -
                  Effect: Allow
                  Action:
                    - 'glue:StartJobRun'
                    - 'glue:GetJobRun'
                    - 'glue:GetJobRuns'
                    - 'glue:BatchStopJobRun'
                  Resource: "*"

#Role for Glue jobs
  GlueExecutionRole:
      Type: AWS::IAM::Role
      Properties:
        Description : IAM Role for AWS Glue Jobs
        ManagedPolicyArns:
          - arn:aws:iam::aws:policy/AmazonEC2ContainerRegistryReadOnly
          - arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
        AssumeRolePolicyDocument:
          Version: "2012-10-17"
          Statement:
            -
              Effect: "Allow"
              Principal:
                Service:
                  - "glue.amazonaws.com"
              Action:
                - "sts:AssumeRole"
        Path: "/"
        Policies:
            -
              PolicyName: GetDescribeSecret
              PolicyDocument:
                Version: 2012-10-17
                Statement:
                  -
                    Effect: Allow
                    Action:
                      - 'secretsmanager:GetResourcePolicy'
                      - 'secretsmanager:GetSecretValue'
                      - 'secretsmanager:DescribeSecret'
                      - 'secretsmanager:ListSecretVersionIds'
                    Resource: 
                      - !Sub arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:*
            -
              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:::${inputBucketName}'
                      - !Sub 'arn:aws:s3:::${inputBucketName}/*'
            -
              PolicyName: DynamoDBAccessPolicy
              PolicyDocument:
                Version: 2012-10-17
                Statement:
                  -
                    Effect: Allow
                    Action:
                    - dynamodb:DescribeStream
                    - dynamodb:GetRecords
                    - dynamodb:GetShardIterator
                    - dynamodb:ListStreams
                    - dynamodb:GetItem
                    - dynamodb:PutItem
                    - dynamodb:Scan
                    - dynamodb:Query
                    - dynamodb:UpdateItem
                    - dynamodb:BatchWriteItem
                    - dynamodb:DeleteItem
                    Resource: "arn:aws:dynamodb:*:*:table/bq*"

  
  # Setup Lambda execution Role
  LambdaExecutionRole:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service:
                - lambda.amazonaws.com
            Action:
              - 'sts:AssumeRole'
      Path: /
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/AWSLambdaExecute
        - arn:aws:iam::aws:policy/SecretsManagerReadWrite
        - arn:aws:iam::aws:policy/AmazonVPCFullAccess 
      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:::${inputBucketName}'
                      - !Sub 'arn:aws:s3:::${inputBucketName}/*'

# Setup EventBridge Role  
  RoleEventBridgeStartStateMachine:
    Type: 'AWS::IAM::Role'
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
        - Effect: Allow
          Principal:
            Service: 'events.amazonaws.com'
          Action: 'sts:AssumeRole'
      Policies:
      - PolicyName: StepFunctionsStateMachine-BQ-Migration-Trigger
        PolicyDocument:
          Statement:
          - Effect: Allow
            Action:
              - 'states:StartExecution'
            Resource:
              - !GetAtt 'MigrationETLStepFunction.Arn'

# EventBridge Rule
  EventBridgeRuleNewDataArrival:
    Type: 'AWS::Events::Rule'
    Properties:
      Description: Listen to object creation and update events inside the S3 bucket for migration config file.
      EventBusName: default
      State: ENABLED
      EventPattern:
        source:
          - aws.s3
        detail-type:
          - Object Created
        detail:
          bucket:
            name:
            - !Ref 'inputBucketName'
          object:
            key:
              - prefix: 'bq-mig-config/'
      Targets:
        - Arn: !GetAtt 'MigrationETLStepFunction.Arn'
          Id: NewConfigArrivalForMigrationTask
          RoleArn: !GetAtt 'RoleEventBridgeStartStateMachine.Arn'

#DynamoDB tables
#Metadata DDB table
  DynamoDBTablesBQToS3Tracking:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
        -
          AttributeName: "table_name"
          AttributeType: "S"
      KeySchema:
        -
          AttributeName: "table_name"
          KeyType: "HASH"
      BillingMode: PAY_PER_REQUEST
      TableName: !Sub ${inputDynamoDBTablePrefix}_tracking

#Metadata DDB table
  DynamoDBTablesMaxStreams:
    Type: AWS::DynamoDB::Table
    Properties:
      AttributeDefinitions:
        -
          AttributeName: "exec_id"
          AttributeType: "S"
      KeySchema:
        -
          AttributeName: "exec_id"
          KeyType: "HASH"
      BillingMode: PAY_PER_REQUEST
      TableName: !Sub ${inputDynamoDBTablePrefix}_maxstreams

#AWS Glue 

#Migration Glue Job
  GlueJobBQToS3:
      Type: AWS::Glue::Job
      DependsOn:
        - GlueExecutionRole
      Properties:
        Command:
          Name: glueetl
          ScriptLocation: !Sub ${GlueS3PathURI}bdb2244_bq_migrate_data_ETL.py
        ExecutionProperty:
          MaxConcurrentRuns: 40
        MaxRetries: 1
        Connections:
          Connections: 
          - !Ref inputGlueConnectionName
        Name: !Ref inputGlueETLJob
        WorkerType: G.1X
        NumberOfWorkers: 3
        GlueVersion: 4.0
        Timeout: 180
        DefaultArguments:
          '--job-bookmark-option': job-bookmark-disable
          '--enable-metrics': "true"
          '--enable-auto-scaling': "true"
          "--connectionname" : !Ref inputGlueConnectionName
          "--bucket" : !Ref inputBucketName
          "--parent_project" : !Ref InputBQProjectName
          "--tablename" : bq-demo-table
        Role: !Ref GlueExecutionRole

#Metadata 
  GlueJobBQMetadata:
    Type: AWS::Glue::Job
    Properties:
      Name: !Ref inputGlueMetaJob
      Role:
        Fn::GetAtt: [GlueExecutionRole, Arn]
      MaxCapacity: 0.0625
      GlueVersion: 1.0
      Command:
        Name: pythonshell
        PythonVersion: 3.9
        ScriptLocation: !Sub ${GlueS3PathURI}bdb2244_bq_get_metadata.py
      DefaultArguments:
        '--TempDir': !Sub
          - s3://${bucket}/glue_tmp/
          - {bucket: !Ref inputBucketName}
        "--prebuilt-library-option" : "prebuilt-library-enable"
        "--secret" : !Ref InputBQSecret
        "--additional-python-modules" : google-cloud-bigquery
        "--object_key" : bq-mig-config/bq-mig-config.json
        "--dynamodb_table" : !Ref DynamoDBTablesBQToS3Tracking
        "--enable-metrics" : ""
        "--region": ${AWS::Region}
        "--connectionname" : !Ref inputGlueConnectionName
        "--bq_project" : !Ref InputBQProjectName

    DependsOn:
      - GlueExecutionRole

  MigrationETLStepFunction:
    Type: 'AWS::StepFunctions::StateMachine'
    Properties:
      StateMachineName: !Ref StateMachineName
      DefinitionString: !Sub
        - |-
          {
            "Comment": "State Machine to Migrate Google BigQuery datasets to AmazonS3",
            "StartAt": "getMetaData",
            "States": {
              "getMetaData": {
                "Type": "Task",
                "Resource": "arn:aws:states:::glue:startJobRun.sync",
                "Parameters": {
                  "JobName": "${glueMetaJob}",
                  "Arguments": {
                    "--additional-python-modules": "google-cloud-bigquery",
                    "--dynamodb_table": "${dynamoTrackingTBL}",
                    "--secret": "${secretName}",
                    "--bucket.$": "$.detail.bucket.name",
                    "--object_key.$": "$.detail.object.key",
                    "--region.$": "$.region",
                    "--connectionname" : "${glueConnection}",
                    "--bq_project" : "${bq_project}"
                  }
                },
                "Next": "Set Max streams for current execution"
              },
              "Set Max streams for current execution": {
                "Type": "Task",
                "Resource": "arn:aws:states:::dynamodb:putItem",
                "Parameters": {
                  "TableName": "${dynamoStreamsTBL}",
                  "Item": {
                    "exec_id": {
                      "S.$": "$$.Execution.Name"
                    },
                    "max_streams": {
                      "N": "1000"
                    }
                  }
                },
                "Next": "ScanTrackingTable"
              },
              "ScanTrackingTable": {
                "Type": "Task",
                "Next": "check if there are still table to migrate",
                "Parameters": {
                  "TableName": "${dynamoTrackingTBL}",
                  "Select": "ALL_ATTRIBUTES",
                  "FilterExpression": "load_status = :status",
                  "ExpressionAttributeValues": {
                    ":status": {
                      "S": "waiting"
                    }
                  }
                },
                "Resource": "arn:aws:states:::aws-sdk:dynamodb:scan"
              },
              "check if there are still table to migrate": {
                "Type": "Choice",
                "Choices": [
                  {
                    "Variable": "$.Count",
                    "NumericEquals": 0,
                    "Next": "Scan metadata table for Errors in excecution batch"
                  }
                ],
                "Default": "Migrate tables in Parallel"
              },
              "Scan metadata table for Errors in excecution batch": {
                "Type": "Task",
                "Next": "Check if there have been errors",
                "Parameters": {
                  "TableName": "${dynamoTrackingTBL}",
                  "Select": "ALL_ATTRIBUTES",
                  "FilterExpression": "load_status = :status and exec_id = :e",
                  "ExpressionAttributeValues": {
                    ":status": {
                      "S": "FAILED"
                    },
                    ":e": {
                      "S.$": "$$.Execution.Name"
                    }
                  }
                },
                "Resource": "arn:aws:states:::aws-sdk:dynamodb:scan"
              },
              "Check if there have been errors": {
                "Type": "Choice",
                "Choices": [
                  {
                    "Variable": "$.Count",
                    "NumericGreaterThan": 0,
                    "Next": "Fail"
                  }
                ],
                "Default": "Success"
              },
              "Fail": {
                "Type": "Fail"
              },
              "Success": {
                "Type": "Succeed"
              },
              "Migrate tables in Parallel": {
                "Type": "Map",
                "ItemProcessor": {
                  "ProcessorConfig": {
                    "Mode": "INLINE"
                  },
                  "StartAt": "Check Priority",
                  "States": {
                    "Check Priority": {
                      "Type": "Choice",
                      "Choices": [
                        {
                          "Variable": "$.priority.N",
                          "StringEquals": "1",
                          "Next": "TransactWriteItems"
                        }
                      ],
                      "Default": "Update Priority"
                    },
                    "TransactWriteItems": {
                      "Type": "Task",
                      "Next": "Migrate BQ query to S3",
                      "Parameters": {
                        "TransactItems": [
                          {
                            "Update": {
                              "TableName": "${dynamoStreamsTBL}",
                              "Key": {
                                "exec_id": {
                                  "S.$": "$$.Execution.Name"
                                }
                              },
                              "UpdateExpression": "set max_streams = max_streams - :s",
                              "ExpressionAttributeValues": {
                                ":s": {
                                  "N.$": "$.number_of_streams.N"
                                }
                              }
                            }
                          },
                          {
                            "Update": {
                              "TableName": "${dynamoTrackingTBL}",
                              "Key": {
                                "table_name": {
                                  "S.$": "$.table_name.S"
                                }
                              },
                              "UpdateExpression": "set updated_date = :r, load_status = :p, exec_id = :e",
                              "ExpressionAttributeValues": {
                                ":r": {
                                  "S.$": "$$.State.EnteredTime"
                                },
                                ":p": {
                                  "S": "STARTING"
                                },
                                ":e": {
                                  "S.$": "$$.Execution.Name"
                                }
                              }
                            }
                          }
                        ]
                      },
                      "Resource": "arn:aws:states:::aws-sdk:dynamodb:transactWriteItems",
                      "ResultPath": null,
                      "Retry": [
                        {
                          "ErrorEquals": [
                            "States.TaskFailed"
                          ],
                          "BackoffRate": 2,
                          "IntervalSeconds": 2,
                          "MaxAttempts": 15,
                          "Comment": "catch-transaction-cancelled"
                        }
                      ]
                    },
                    "Update Priority": {
                      "Type": "Task",
                      "Resource": "arn:aws:states:::dynamodb:updateItem",
                      "Parameters": {
                        "TableName": "${dynamoTrackingTBL}",
                        "Key": {
                          "table_name": {
                            "S.$": "$.table_name.S"
                          }
                        },
                        "UpdateExpression": "set priority = priority - :p",
                        "ExpressionAttributeValues": {
                          ":p": {
                            "N": "1"
                          }
                        }
                      },
                      "End": true
                    },
                    "Migrate BQ query to S3": {
                      "Type": "Task",
                      "Resource": "arn:aws:states:::glue:startJobRun.sync",
                      "Parameters": {
                        "JobName": "${glueETLJob}",
                        "Arguments": {
                          "--tablename.$": "$.table_name.S",
                          "--dynamodb_table": "${dynamoTrackingTBL}",
                          "--bucket": "${bucket}",
                          "--connectionname": "${glueConnection}",
                          "--parent_project.$": "$.project_name.S",
                          "--s3_path.$": "$.s3_path.S"
                        },
                        "NumberOfWorkers.$": "States.StringToJson($.workers_needed.N)",
                        "WorkerType": "G.1X"
                      },
                      "Next": "Update Metadata Colmpleted",
                      "Catch": [
                        {
                          "ErrorEquals": [
                            "States.TaskFailed"
                          ],
                          "Next": "Update Metadata Failed",
                          "ResultPath": "$.error_path.errot_details"
                        }
                      ],
                      "ResultPath": "$.myresult"
                    },
                    "Update Metadata Colmpleted": {
                      "Type": "Task",
                      "Resource": "arn:aws:states:::dynamodb:updateItem",
                      "Parameters": {
                        "TableName": "${dynamoTrackingTBL}",
                        "Key": {
                          "table_name": {
                            "S.$": "$.table_name.S"
                          }
                        },
                        "UpdateExpression": "set updated_date = :r, load_status = :p , glue_job_run_id = :i,glue_job_exec_time_sec = :t",
                        "ExpressionAttributeValues": {
                          ":r": {
                            "S.$": "$$.State.EnteredTime"
                          },
                          ":p": {
                            "S.$": "$.myresult.JobRunState"
                          },
                          ":i": {
                            "S.$": "$.myresult.Id"
                          },
                          ":t": {
                            "N.$": "States.JsonToString($.myresult.ExecutionTime)"
                          }
                        }
                      },
                      "Next": "Update Max streams for current execution",
                      "ResultPath": null
                    },
                    "Update Max streams for current execution": {
                      "Type": "Task",
                      "Resource": "arn:aws:states:::dynamodb:updateItem",
                      "Parameters": {
                        "TableName": "${dynamoStreamsTBL}",
                        "Key": {
                          "exec_id": {
                            "S.$": "$$.Execution.Name"
                          }
                        },
                        "UpdateExpression": "set max_streams = max_streams + :s",
                        "ExpressionAttributeValues": {
                          ":s": {
                            "N.$": "$.number_of_streams.N"
                          }
                        }
                      },
                      "End": true
                    },
                    "Update Metadata Failed": {
                      "Type": "Task",
                      "Resource": "arn:aws:states:::dynamodb:updateItem",
                      "Parameters": {
                        "TableName": "${dynamoTrackingTBL}",
                        "Key": {
                          "table_name": {
                            "S.$": "$.table_name.S"
                          }
                        },
                        "UpdateExpression": "set updated_date = :r, load_status = :p, glue_job_run_id = :i,details = :e",
                        "ExpressionAttributeValues": {
                          ":r": {
                            "S.$": "$$.State.EnteredTime"
                          },
                          ":p": {
                            "S": "FAILED"
                          },
                          ":i": {
                            "S": "see details"
                          },
                          ":e": {
                            "S.$": "$.errorPath"
                          }
                        }
                      },
                      "Next": "Update Max streams for current execution",
                      "ResultPath": null
                    }
                  }
                },
                "ItemsPath": "$.Items",
                "Next": "ScanTrackingTable",
                "MaxConcurrency": ${maxConcurrency}
              }
            }
          }
        - {bucket: !Ref inputBucketName,
          glueConnection: !Ref inputGlueConnectionName,
          glueETLJob: !Ref inputGlueETLJob,
          glueMetaJob: !Ref inputGlueMetaJob,
          maxConcurrency: !Ref InputMaxParallelism,
          dynamoTrackingTBL: !Ref DynamoDBTablesBQToS3Tracking,
          dynamoStreamsTBL: !Ref DynamoDBTablesMaxStreams,
          secretName: !Ref InputBQSecret,
          bq_project : !Ref InputBQProjectName
            }

      RoleArn: !GetAtt StateMachineExecutionRole.Arn


  SetupFunction:
    Type: AWS::Lambda::Function
    Properties:
      Role: !GetAtt 'LambdaExecutionRole.Arn'
      FunctionName: !Join ['-', [!Ref 'AWS::StackName','copy_scripts' ]]
      MemorySize: 2048
      Runtime: python3.7
      Timeout: 300
      Handler: index.handler
      Code:
        ZipFile: 
          Fn::Sub:
          - |-
           import json
           import boto3
           import os
           import logging
           import cfnresponse

           LOGGER = logging.getLogger()
           LOGGER.setLevel(logging.INFO)

           SOURCE_BUCKET = '${SourceS3Bucket}'
           DEST_BUCKET= '${S3Path}'.split('/',3)[2]
           DEST_PATH= '${S3Path}'.split('/',3)[3]
           ROLE_ARN = '${Role}'

           def handler(event, context):

               # Get CloudFormation-specific parameters, if they exist
               cfn_stack_id = event.get('StackId')
               cfn_request_type = event.get('RequestType')
               
               # Was the function triggered by a CloudFormation resource deletion?
               # If so, delete all objects in dest bucket and return
               if cfn_stack_id and cfn_request_type == 'Delete':
                   s3 = boto3.resource('s3')
                   try:
                       message = 'Nothing to do'
                   except botocore.exceptions.ClientError as e:
                       # If a client error is thrown, then check that it was a 404 error.
                       # If it was a 404 error, then the bucket does not exist.
                       error_code = int(e.response['Error']['Code'])
                       if error_code == 404:
                           message = 'bucket does not exist.'
                    
                   cfnresponse.send(event, context, cfnresponse.SUCCESS, {
                           'Message': message
                            },
                            context.log_stream_name)

                   return {
                        'statusCode': 200,
                        'body': json.dumps(message)
                    }
                        
               s3 = boto3.client('s3') 
               #copy python script files directly
               s3.download_file(SOURCE_BUCKET, 'artifacts/BDB-2244/python/bdb2244_bq_get_metadata.py', '/tmp/bdb2244_bq_get_metadata.py')
               s3.upload_file('/tmp/bdb2244_bq_get_metadata.py', DEST_BUCKET, DEST_PATH+'bdb2244_bq_get_metadata.py', )
               s3.download_file(SOURCE_BUCKET, 'artifacts/BDB-2244/python/bdb2244_bq_migrate_data_ETL.py', '/tmp/bdb2244_bq_migrate_data_ETL.py')
               s3.upload_file('/tmp/bdb2244_bq_migrate_data_ETL.py', DEST_BUCKET, DEST_PATH+'bdb2244_bq_migrate_data_ETL.py', )
                
               # Send a response to CloudFormation pre-signed URL
               kv_output = {"message": "success"}
               cfnresponse.send(event, context, cfnresponse.SUCCESS, kv_output, context.log_stream_name)
                
               return {
                   'statusCode': 200,
                   'body': json.dumps('Copied Files')
               }

          - {
            SourceS3Bucket: !Ref SourceS3BucketName,
            S3Path : !Ref GlueS3PathURI,
            Role : !GetAtt LambdaExecutionRole.Arn
            } 
  
  EnvSetup:
    Type: 'Custom::EnvSetup'
    DependsOn:
      - LambdaExecutionRole
    Properties:
      ServiceToken: !GetAtt 
        - SetupFunction
        - Arn