Description: "Creates the Lambda functions for the Bucket Rekognition Backfill application."
Parameters:
  JobDefinitionVersion:
    Type: String
    Description: The job defintion version number
    MinLength: 1
    MaxLength: 2
    Default: "1"

  EsEndpoint:
    Type: String
    Description: The Elasticsearch domain endpoint

  ImagesBucket:
    Type: String
    Description: The name of the S3 bucket where the images are stored

Resources:

  RekognitionLabelerRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Path: "/"
      Policies:
        -
          PolicyName: image-labeler-policy
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              -
                Effect: "Allow"
                Action: "rekognition:DetectLabels"
                Resource: "*"
              -
                Effect: Allow
                Action:
                    - es:ESHttp*
                Resource: "*"
              -
                Effect: Allow
                Action:
                    - "s3:GetObject"
                    - "s3:GetObjectVersion"
                Resource: !Sub arn:aws:s3:::${ImagesBucket}/*
              -
                Effect: "Allow"
                Action:
                  - "logs:CreateLogGroup"
                  - "logs:CreateLogStream"
                  - "logs:PutLogEvents"
                Resource: "arn:aws:logs:*:*:*"

  RekognitionLabelerFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      Code:
        S3Bucket: aws-bigdata-blog
        S3Key: artifacts/RekognitionBatch/lambda/src/RekognitionLabeler.zip
      Description: Calls Rekognition when a new image is uploaded to S3 and saves labels to an Elasticsearch cluster.
      Environment:
        Variables:
          ES_ENDPOINT: !Ref EsEndpoint
      FunctionName: RekognitionLabeler
      Handler: image_labeler.lambda_handler
      MemorySize: 512
      Role: !GetAtt RekognitionLabelerRole.Arn
      Runtime: python2.7
      Timeout: 60

  S3InventoryProcessorFunctionRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Path: "/"
      Policies:
        -
          PolicyName: s3-inventory-processor-policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: "Allow"
                Action:
                  - "batch:SubmitJob"
                Resource: "*"
              -
                Effect: "Allow"
                Action:
                  - "logs:CreateLogGroup"
                  - "logs:CreateLogStream"
                  - "logs:PutLogEvents"
                Resource: "arn:aws:logs:*:*:*"
              -
                Effect: "Allow"
                Action:
                  - s3:GetObject
                  - s3:GetObjectVersion
                  - s3:PutBucketNotification
                  - s3:GetBucketNotification
                Resource: "*"

  S3InventoryProcessorFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      Code:
        S3Bucket: aws-bigdata-blog
        S3Key: artifacts/RekognitionBatch/lambda/src/S3InventoryProcessor.zip
      Description: Creates an AWS Batch job for every image in the bucket inventory file when created by the S3 Inventory service.
      Environment:
        Variables:
          AWS_BATCH_JOB_QUEUE: image_labeler_queue
          AWS_BATCH_JOB_DEFINITION: !Sub image_labeler:${JobDefinitionVersion}
      FunctionName: S3InventoryProcessor
      Handler: inventory_processor.lambda_handler
      MemorySize: 512
      Role: !GetAtt S3InventoryProcessorFunctionRole.Arn
      Runtime: python2.7
      Timeout: 300

  CodeBuildTriggerFunctionRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Path: "/"
      Policies:
        -
          PolicyName: codebuild-policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: "*"
              -
                Effect: Allow
                Action:
                  - codebuild:ListBuilds
                  - codebuild:ListProjects
                  - codebuild:StartBuild
                  - codebuild:StopBuild
                Resource: "*"
              -
                Effect: Allow
                Action:
                  - s3:PutObject
                Resource: !Sub arn:aws:s3:::rekdemo-inventory-${AWS::AccountId}-${AWS::Region}/*

  CodeBuildTriggerFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: CodeBuildTriggerResource
      Handler: index.handler
      Role: !GetAtt CodeBuildTriggerFunctionRole.Arn
      Description: Custom CFN resource to starts the build of a CodeBuild project.
      Runtime: nodejs4.3
      Timeout: 30
      Code:
        ZipFile: |
          var AWS = require('aws-sdk');
          var response = require('cfn-response');
          var http = require('https');
          var url = 'https://s3.amazonaws.com/aws-bigdata-blog/artifacts/RekognitionBatch/ecs/src/RekognitionBatchImage.zip';
          exports.handler = function(event, context) {
              console.log('REQUEST RECEIVED');
              console.log(event);
              if (event.RequestType != 'Create') {
                  response.send(event, context, response.SUCCESS);
                  return;
              }
              http.get(url, function(res) { // download the source code
                var data = [];
                var contentType = res.headers['content-type'];
                var contentLength = res.headers['content-length'];
                res.on('data', function(chunk) {
                  data.push(chunk);
                });
                res.on('end', function() {
                  var s3 = new AWS.S3();
                  var buffer = Buffer.concat(data);
                  var params = {Bucket: event.ResourceProperties.Bucket, Key: event.ResourceProperties.Key, Body: buffer, ContentType: contentType, ContentLength: contentLength};
                  s3.upload(params, function(err, data) { // save the source code to the inventory bucket
                    if (err) {
                        console.log(err, err.stack); // an error occurred
                        response.send(event, context, response.FAILED, err);
                    } else {
                      console.log(data);
                      var codebuild = new AWS.CodeBuild();
                      var params = { projectName: event.ResourceProperties.ProjectName };
                      codebuild.startBuild(params, function(err, data) {
                        if (err) {
                          console.log(err, err.stack); // an error occurred
                          response.send(event, context, response.FAILED, err);
                        }
                        else {
                          console.log(data);           // successful response
                          response.send(event, context, response.SUCCESS, data);
                        }
                      });
                    }
                  });
                });
              });
          }

  S3NotificationConfigFunctionRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Path: "/"
      Policies:
        -
          PolicyName: s3-notification-config-policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: "*"
              -
                Effect: Allow
                Action:
                  - s3:PutBucketNotification
                  - s3:GetBucketNotification
                Resource: "*"

  S3NotificationConfigFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: S3NotificationConfigResource
      Handler: index.handler
      Role: !GetAtt S3NotificationConfigFunctionRole.Arn
      Description: Custom CFN resource to configure the S3 Notification Configuration of a bucket
      Runtime: nodejs4.3
      Timeout: 30
      Code:
        ZipFile: |
          var AWS = require('aws-sdk');
          var response = require('cfn-response');
          exports.handler = function(event, context) {
              console.log('REQUEST RECEIVED');
              console.log(event);
              var s3 = new AWS.S3();
              if (event.RequestType == 'Create') {
                var params = {
                    Bucket: event.ResourceProperties.Bucket,
                    NotificationConfiguration: {
                      LambdaFunctionConfigurations: [
                          {
                              Id: "1",
                              LambdaFunctionArn: event.ResourceProperties.LambdaArn,
                              Events: [
                                  "s3:ObjectCreated:*",
                                  "s3:ObjectRemoved:*"
                              ],
                              Filter: {
                                  Key: {
                                      FilterRules: [
                                          {
                                              Name: "suffix",
                                              Value: ".jpg"
                                          }
                                      ]
                                  }
                              }
                          },
                          {
                              Id: "2",
                              LambdaFunctionArn: event.ResourceProperties.LambdaArn,
                              Events: [
                                  "s3:ObjectCreated:*",
                                  "s3:ObjectRemoved:*"
                              ],
                              Filter: {
                                  Key: {
                                      FilterRules: [
                                          {
                                              Name: "suffix",
                                              Value: ".jpeg"
                                          }
                                      ]
                                  }
                              }
                          },
                          {
                              Id: "3",
                              LambdaFunctionArn: event.ResourceProperties.LambdaArn,
                              Events: [
                                  "s3:ObjectCreated:*",
                                  "s3:ObjectRemoved:*"
                              ],
                              Filter: {
                                  Key: {
                                      FilterRules: [
                                          {
                                              Name: "suffix",
                                              Value: ".png"
                                          }
                                      ]
                                  }
                              }
                          }
                      ]
                    }
                };
                s3.putBucketNotificationConfiguration(params, function(err, data) {
                    if (err) {
                      console.log(err, err.stack); // an error occurred
                      response.send(event, context, response.FAILED, err);
                    }
                    else {
                      console.log(data);           // successful response
                      response.send(event, context, response.SUCCESS, data);
                    }
                });
              } else if (event.RequestType == 'Delete') {
                var params = {
                    Bucket: event.ResourceProperties.Bucket,
                    NotificationConfiguration: {
                      LambdaFunctionConfigurations: []
                    }
                };
                s3.putBucketNotificationConfiguration(params, function(err, data) {
                  if (err) {
                    console.log(err, err.stack); // an error occurred
                    response.send(event, context, response.FAILED, err);
                  }
                  else {
                    console.log(data);           // successful response
                    response.send(event, context, response.SUCCESS, data);
                  }
                });
              } else {
                  response.send(event, context, response.SUCCESS);
                  return;
              }
          }

  S3InventoryFunctionRole:
    Type: "AWS::IAM::Role"
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          -
            Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Path: "/"
      Policies:
        -
          PolicyName: s3-inventory-config-policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: "*"
              -
                Effect: Allow
                Action:
                  - s3:PutInventoryConfiguration
                  - s3:GetInventoryConfiguration
                Resource: "*"

  S3InventoryConfigFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: S3InventoryConfigResource
      Handler: index.handler
      Role: !GetAtt S3InventoryFunctionRole.Arn
      Description: Custom CFN resource to configure the S3 Inventory Configuration of a bucket
      Runtime: nodejs4.3
      Timeout: 30
      Code:
        ZipFile: |
          var AWS = require('aws-sdk');
          var response = require('cfn-response');
          exports.handler = function(event, context) {
              console.log('REQUEST RECEIVED');
              console.log(event);
              var s3 = new AWS.S3();
              if (event.RequestType == 'Create') {
                var params = {
                    Bucket: event.ResourceProperties.ImagesBucket,
                    Id: 'images-inventory',
                    InventoryConfiguration: {
                            Destination: {
                                S3BucketDestination: {
                                        AccountId: event.ResourceProperties.AccountId,
                                        Bucket: 'arn:aws:s3:::' + event.ResourceProperties.InventoryBucket,
                                        Format: 'CSV'
                                }
                            },
                            IsEnabled: true,
                            OptionalFields: [],
                            Id: 'images-inventory',
                            IncludedObjectVersions: 'Current',
                            Schedule: {
                                    Frequency: 'Daily'
                            }
                    }
                };
                s3.putBucketInventoryConfiguration(params, function(err, data) {
                    if (err) {
                      console.log(err, err.stack); // an error occurred
                      response.send(event, context, response.FAILED, err);
                    }
                    else {
                      console.log(data);           // successful response
                      response.send(event, context, response.SUCCESS, data);
                    }
                });
              } else if (event.RequestType == 'Delete') {
                var params = {
                    Bucket: event.ResourceProperties.ImagesBucket,
                    Id: 'images-inventory'
                };
                s3.deleteBucketInventoryConfiguration(params, function(err, data) {
                  if (err) {
                    console.log(err, err.stack); // an error occurred
                    response.send(event, context, response.FAILED, err);
                  }
                  else {
                    console.log(data);           // successful response
                    response.send(event, context, response.SUCCESS, data);
                  }
                });
              } else {
                  response.send(event, context, response.SUCCESS);
                  return;
              }
          }

  AWSBatchFunctionRole:
    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/AWSBatchFullAccess
      Policies:
        -
          PolicyName: awsbatch-policy
          PolicyDocument:
            Version: 2012-10-17
            Statement:
              -
                Effect: Allow
                Action:
                  - logs:CreateLogGroup
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: "*"
              -
                Effect: Allow
                Action:
                  - iam:PassRole
                Resource: "*"

  AWSBatchComputeEnvFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: AWSBatchComputeEnvResource
      Handler: index.handler
      Role: !GetAtt AWSBatchFunctionRole.Arn
      Description: Custom CFN resource to creates and delets an AWS Batch Compute Environment
      Runtime: nodejs4.3
      Timeout: 300
      Code:
        ZipFile: |
          var AWS = require('aws-sdk');
          var response = require('cfn-response');
          exports.handler = function(event, context) {
              console.log('REQUEST RECEIVED');
              console.log(event);
              var batch = new AWS.Batch();
              if (event.RequestType == 'Create') {
                var subnetParam;
                if (event.ResourceProperties.Subnets instanceof Array) {
                  subnetParam = event.ResourceProperties.Subnets;
                } else {
                  subnetParam = event.ResourceProperties.Subnets.split(',');
                }
                var securityGroupParam;
                if (event.ResourceProperties.SecurityGroups instanceof Array) {
                  securityGroupParam = event.ResourceProperties.SecurityGroups;
                } else {
                  securityGroupParam = event.ResourceProperties.SecurityGroups.split(',');
                }
                var params = {
                    type: 'Managed',
                    computeEnvironmentName: event.ResourceProperties.ComputeEnvName,
                    serviceRole: event.ResourceProperties.ServiceRole,
                    state: 'ENABLED',
                    computeResources: {
                        type: 'EC2',
                        maxvCpus: event.ResourceProperties.MaxvCpus,
                        desiredvCpus: event.ResourceProperties.DesiredvCpus,
                        minvCpus: event.ResourceProperties.MinvCpus,
                        instanceRole: event.ResourceProperties.InstanceProfile,
                        instanceTypes: ['optimal'],
                        subnets: subnetParam,
                        securityGroupIds: securityGroupParam,
                        tags: {
                          "Name": "AWS Batch Instance"
                        }
                    }
                };
                batch.createComputeEnvironment(params, function(err, data) {
                    if (err) {
                      console.log(err, err.stack); // an error occurred
                      response.send(event, context, response.FAILED, err);
                    }
                    else {
                      console.log(data);           // successful response
                      response.send(event, context, response.SUCCESS, data);
                    }
                });
              } else if (event.RequestType == 'Delete') {
                var params = {
                    computeEnvironment: event.ResourceProperties.ComputeEnvName,
                    state: 'DISABLED'
                };
                batch.updateComputeEnvironment(params, function(err, data) {
                  if (err) {
                    console.log(err, err.stack); // an error occurred
                    response.send(event, context, response.FAILED, err);
                  }
                  else {
                    console.log(data);           // successful response
                    console.log("Wait 90 secs to allow job queue to be deleted");
                    setTimeout(function(){
                      var params = {
                          computeEnvironment: event.ResourceProperties.ComputeEnvName
                      };
                      batch.deleteComputeEnvironment(params, function(err, data) {
                        if (err) {
                          console.log(err, err.stack); // an error occurred
                          response.send(event, context, response.FAILED, err);
                        }
                        else {
                          console.log("Wait 120 sec to send response to allow resources to be deleted");
                          setTimeout(function(){
                            response.send(event, context, response.SUCCESS, data);
                          }, 120000);
                        }
                      });
                    }, 90000);
                  }
                });
              } else {
                  response.send(event, context, response.SUCCESS);
                  return;
              }
          }

  AWSBatchJobQueueFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: AWSBatchJobQueueResource
      Handler: index.handler
      Role: !GetAtt AWSBatchFunctionRole.Arn
      Description: Custom CFN resource to create and delete AWS Batch Job Queues
      Runtime: nodejs4.3
      Timeout: 180
      Code:
        ZipFile: |
          var AWS = require('aws-sdk');
          var response = require('cfn-response');
          exports.handler = function(event, context) {
              console.log('REQUEST RECEIVED');
              console.log(event);
              var batch = new AWS.Batch();
              if (event.RequestType == 'Create') {
                var params = {
                    computeEnvironmentOrder: [
                        {
                            computeEnvironment: event.ResourceProperties.ComputeEnvName,
                            order: 1
                        }
                    ],
                    jobQueueName: event.ResourceProperties.JobQueueName,
                    priority: 10,
                    state: 'ENABLED'
                };
                batch.createJobQueue(params, function(err, data) {
                    if (err) {
                      console.log(err, err.stack); // an error occurred
                      response.send(event, context, response.FAILED, err);
                    }
                    else {
                      console.log(data);           // successful response
                      response.send(event, context, response.SUCCESS, data);
                    }
                });
              } else if (event.RequestType == 'Delete') {
                var params = {
                    jobQueue: event.ResourceProperties.JobQueueName,
                    state: 'DISABLED'
                };
                batch.updateJobQueue(params, function(err, data) {
                  if (err) {
                    console.log(err, err.stack); // an error occurred
                    response.send(event, context, response.FAILED, err);
                  }
                  else {
                    console.log(data);           // successful response
                    console.log("Sleeping for 60 secs before deleting");
                    setTimeout(function(){
                      var params = { jobQueue: event.ResourceProperties.JobQueueName };
                      batch.deleteJobQueue(params, function(err, data) {
                        if (err) {
                          console.log(err, err.stack); // an error occurred
                          response.send(event, context, response.FAILED, err);
                        }
                        else {
                          console.log(data);           // successful response
                          response.send(event, context, response.SUCCESS, data);
                        }
                      });
                    }, 60000);
                  }
                });
              } else {
                  response.send(event, context, response.SUCCESS);
                  return;
              }
          }

  AWSBatchJobDefinitionFunction:
    Type: "AWS::Lambda::Function"
    Properties:
      FunctionName: AWSBatchJobDefinitionResource
      Handler: index.handler
      Role: !GetAtt AWSBatchFunctionRole.Arn
      Description: Custom CFN resource to create and delete AWS Batch Job Definitions
      Runtime: nodejs4.3
      Timeout: 60
      Code:
        ZipFile: |
          var AWS = require('aws-sdk');
          var response = require('cfn-response');
          exports.handler = function(event, context) {
              console.log('REQUEST RECEIVED');
              console.log(event);
              var batch = new AWS.Batch();
              if (event.RequestType == 'Create') {
                var params = {
                    type: "container",
                    containerProperties: {
                        environment: [
                            {
                                name: "ES_ENDPOINT",
                                value: event.ResourceProperties.EsEndpoint
                            }
                        ],
                        jobRoleArn: event.ResourceProperties.JobRoleArn,
                        image: event.ResourceProperties.ImageName,
                        memory: 50,
                        vcpus: 1
                    },
                    jobDefinitionName: event.ResourceProperties.JobDefinitionName
                };
                batch.registerJobDefinition(params, function(err, data) {
                    if (err) {
                      console.log(err, err.stack); // an error occurred
                      response.send(event, context, response.FAILED, err);
                    }
                    else {
                      console.log(data);           // successful response
                      response.send(event, context, response.SUCCESS, data);
                    }
                });
              } else if (event.RequestType == 'Delete') {
                batch.describeJobDefinitions({ jobDefinitionName: event.ResourceProperties.JobDefinitionName}, function(err, data) {
                  if (err) {
                    console.log(err, err.stack); // an error occurred
                    response.send(event, context, response.FAILED, err);
                  } else {
                    console.log(data);
                    data.jobDefinitions.forEach(function(element) {
                      console.log("Deregistering JobDefinitionArn: " + element.jobDefinitionArn);
                      batch.deregisterJobDefinition({ jobDefinition: element.jobDefinitionArn }, function(err, data) {
                        if (err) {
                          console.log(err, err.stack); // an error occurred
                          response.send(event, context, response.FAILED, err);
                        }
                        else {
                          console.log(data);           // successful response
                          response.send(event, context, response.SUCCESS, data);
                        }
                      });
                    });
                  }
                });
              } else {
                  response.send(event, context, response.SUCCESS);
                  return;
              }
          }

Outputs:
  RekognitionLabelerFunction:
    Description: The Rekognition Labeler lambda function
    Value: !GetAtt RekognitionLabelerFunction.Arn

  S3InventoryConfigFunction:
    Description: S3 Inventory Config Lambda Function ARN
    Value: !GetAtt S3InventoryConfigFunction.Arn

  CodeBuildTriggerFunction:
    Description: CodeBuild Trigger Lambda Function ARN
    Value: !GetAtt CodeBuildTriggerFunction.Arn

  AWSBatchComputeEnvFunction:
    Description: AWS Batch Compute Enviornment Lambda Function ARN
    Value: !GetAtt AWSBatchComputeEnvFunction.Arn

  AWSBatchJobQueueFunction:
    Description: AWS Batch Job Queue Lambda Function ARN
    Value: !GetAtt AWSBatchJobQueueFunction.Arn

  AWSBatchJobDefinitionFunction:
    Description: AWS Batch Job Definition Lambda Function ARN
    Value: !GetAtt AWSBatchJobDefinitionFunction.Arn

  S3InventoryProcessorFunction:
    Description: S3 Inventory Processor Lambda Function ARN
    Value: !GetAtt S3InventoryProcessorFunction.Arn

  S3NotificationConfigFunction:
    Description: S3 Notification Configuration Lambda Function ARN
    Value: !GetAtt S3NotificationConfigFunction.Arn
