controlshift · sjwmoveon · Jan 13, 2020 · Jan 15, 2020 · May 6, 2020 · Jun 5, 2020
diff --git a/config_item.json b/config_item.json
@@ -23,5 +23,5 @@
   "failureTopicARN": {"S": "${failure_topic_arn}"},
   "batchSize": {"N": "1"},
   "currentBatch": {"S": "${current_batch}"},
-  "compress": {"S": "${compress}"}
+  "compression": {"S": "${compress}"}
 }
diff --git a/config_table.tf b/config_table.tf
@@ -13,12 +13,12 @@ resource "aws_dynamodb_table" "loader_config" {
 }
 
 resource "aws_dynamodb_table_item" "load_config_full_items" {
-  for_each = toset([for table in local.parsed_bulk_data_schemas["tables"] : table["table"]["name"]])
+  for_each = toset(local.table_names)
 
   table_name = aws_dynamodb_table.loader_config.name
   hash_key   = aws_dynamodb_table.loader_config.hash_key
 
-  item = data.template_file.loader_config_full_item[each.key].rendered
+  item = local.loader_config_full_items[each.key]
 
   lifecycle {
     ignore_changes = [
@@ -33,39 +33,13 @@ resource "aws_dynamodb_table_item" "load_config_full_items" {
   }
 }
 
-data "template_file" "loader_config_full_item" {
-  for_each = toset([for table in local.parsed_bulk_data_schemas["tables"] : table["table"]["name"]])
-
-  template = "${file("${path.module}/config_item.json")}"
-  vars = {
-    kind = "full"
-    bulk_data_table = each.key
-    redshift_endpoint = data.aws_redshift_cluster.sync_data_target.endpoint
-    redshift_database_name: var.redshift_database_name
-    redshift_port = data.aws_redshift_cluster.sync_data_target.port
-    redshift_username = var.redshift_username
-    redshift_password = aws_kms_ciphertext.redshift_password.ciphertext_blob
-    schema = var.redshift_schema
-    s3_bucket = "agra-data-exports-${var.controlshift_environment}"
-    manifest_bucket = aws_s3_bucket.manifest.bucket
-    manifest_prefix = var.manifest_prefix
-    failed_manifest_prefix = var.failed_manifest_prefix
-    success_topic_arn = aws_sns_topic.success_sns_topic.arn
-    failure_topic_arn = aws_sns_topic.failure_sns_topic.arn
-    current_batch = random_id.current_batch.b64_url
-    column_list = data.http.column_list[each.key].body
-    truncate_target = true
-    compress = try(local.parsed_bulk_data_schemas["settings"]["compression_format"], "")
-  }
-}
-
 resource "aws_dynamodb_table_item" "load_config_incremental_items" {
-  for_each = toset([for table in local.parsed_bulk_data_schemas["tables"] : table["table"]["name"]])
+  for_each = toset(local.table_names)
 
   table_name = aws_dynamodb_table.loader_config.name
   hash_key   = aws_dynamodb_table.loader_config.hash_key
 
-  item = data.template_file.loader_config_incremental_item[each.key].rendered
+  item = local.loader_config_incremental_items[each.key]
 
   lifecycle {
     ignore_changes = [
@@ -80,29 +54,53 @@ resource "aws_dynamodb_table_item" "load_config_incremental_items" {
   }
 }
 
-data "template_file" "loader_config_incremental_item" {
-  for_each = toset([for table in local.parsed_bulk_data_schemas["tables"] : table["table"]["name"]])
-
-  template = "${file("${path.module}/config_item.json")}"
-  vars = {
-    kind = "incremental"
-    bulk_data_table = each.key
-    redshift_endpoint = data.aws_redshift_cluster.sync_data_target.endpoint
-    redshift_database_name: var.redshift_database_name
-    redshift_port = data.aws_redshift_cluster.sync_data_target.port
-    redshift_username = var.redshift_username
-    redshift_password = aws_kms_ciphertext.redshift_password.ciphertext_blob
-    schema = var.redshift_schema
-    s3_bucket = "agra-data-exports-${var.controlshift_environment}"
-    manifest_bucket = aws_s3_bucket.manifest.bucket
-    manifest_prefix = var.manifest_prefix
-    failed_manifest_prefix = var.failed_manifest_prefix
-    success_topic_arn = aws_sns_topic.success_sns_topic.arn
-    failure_topic_arn = aws_sns_topic.failure_sns_topic.arn
-    current_batch = random_id.current_batch.b64_url
-    column_list = data.http.column_list[each.key].body
-    truncate_target = false
-    compress = try(local.parsed_bulk_data_schemas["settings"]["compression_format"], "")
+locals {
+  table_names = [for table in local.parsed_bulk_data_schemas["tables"] : table["table"]["name"]]
+
+  loader_config_full_items = {
+    for name in local.table_names : name => templatefile("${path.module}/config_item.json", {
+      kind = "full"
+      bulk_data_table = name
+      redshift_endpoint = data.aws_redshift_cluster.sync_data_target.endpoint
+      redshift_database_name = var.redshift_database_name
+      redshift_port = data.aws_redshift_cluster.sync_data_target.port
+      redshift_username = var.redshift_username
+      redshift_password = aws_kms_ciphertext.redshift_password.ciphertext_blob
+      schema = var.redshift_schema
+      s3_bucket = "agra-data-exports-${var.controlshift_environment}"
+      manifest_bucket = aws_s3_bucket.manifest.bucket
+      manifest_prefix = var.manifest_prefix
+      failed_manifest_prefix = var.failed_manifest_prefix
+      success_topic_arn = aws_sns_topic.success_sns_topic.arn
+      failure_topic_arn = aws_sns_topic.failure_sns_topic.arn
+      current_batch = random_id.current_batch.b64_url
+      column_list = data.http.column_list[name].body
+      truncate_target = true
+      compress = try(local.parsed_bulk_data_schemas["settings"]["compression_format"], "")
+    })
+  }
+
+  loader_config_incremental_items = {
+    for name in local.table_names : name => templatefile("${path.module}/config_item.json", {
+      kind = "incremental"
+      bulk_data_table = name
+      redshift_endpoint = data.aws_redshift_cluster.sync_data_target.endpoint
+      redshift_database_name = var.redshift_database_name
+      redshift_port = data.aws_redshift_cluster.sync_data_target.port
+      redshift_username = var.redshift_username
+      redshift_password = aws_kms_ciphertext.redshift_password.ciphertext_blob
+      schema = var.redshift_schema
+      s3_bucket = "agra-data-exports-${var.controlshift_environment}"
+      manifest_bucket = aws_s3_bucket.manifest.bucket
+      manifest_prefix = var.manifest_prefix
+      failed_manifest_prefix = var.failed_manifest_prefix
+      success_topic_arn = aws_sns_topic.success_sns_topic.arn
+      failure_topic_arn = aws_sns_topic.failure_sns_topic.arn
+      current_batch = random_id.current_batch.b64_url
+      column_list = data.http.column_list[name].body
+      truncate_target = false
+      compress = try(local.parsed_bulk_data_schemas["settings"]["compression_format"], "")
+    })
   }
 }
 
@@ -134,11 +132,11 @@ data "http" "bulk_data_schemas" {
 }
 
 locals {
-  parsed_bulk_data_schemas = jsondecode(data.http.bulk_data_schemas.body)
+  parsed_bulk_data_schemas = jsondecode(data.http.bulk_data_schemas.response_body)
 }
 
 data "http" "column_list" {
-  for_each = toset([for table in local.parsed_bulk_data_schemas["tables"] : table["table"]["name"]])
+  for_each = toset(local.table_names)
 
   url = "https://${var.controlshift_hostname}/api/bulk_data/schema/columns?table=${each.key}"
 }
diff --git a/glue_job.tf b/glue_job.tf
@@ -10,6 +10,14 @@ resource "aws_glue_crawler" "signatures_crawler" {
   database_name = aws_glue_catalog_database.catalog_db.name
   name = "${var.controlshift_environment}_full_signatures"
   role = aws_iam_role.glue_service_role.arn
+  configuration = jsonencode(
+      {
+        Grouping = {
+            TableGroupingPolicy = "CombineCompatibleSchemas"
+          }
+        Version  = 1
+      }
+  )
 
   s3_target {
     path = local.signatures_s3_path
@@ -18,34 +26,69 @@ resource "aws_glue_crawler" "signatures_crawler" {
 
 resource "aws_s3_bucket" "glue_resources" {
   bucket = var.glue_scripts_bucket_name
-  region = var.aws_region
+}
+
+# Ownership controls block is required to support ACLs.
+resource "aws_s3_bucket_ownership_controls" "glue_resources" {
+  bucket = aws_s3_bucket.glue_resources.id
+  rule {
+    object_ownership = "ObjectWriter"
+  }
+}
+
+resource "aws_s3_bucket_acl" "glue_resources" {
+  depends_on = [aws_s3_bucket_ownership_controls.glue_resources]
 
+  bucket = aws_s3_bucket.glue_resources.id
   acl = "private"
-  server_side_encryption_configuration {
-    rule {
-      apply_server_side_encryption_by_default {
-        sse_algorithm = "AES256"
-      }
+}
+
+resource "aws_s3_bucket_server_side_encryption_configuration" "glue_resources" {
+  bucket = aws_s3_bucket.glue_resources.id
+
+  rule {
+    apply_server_side_encryption_by_default {
+      sse_algorithm = "AES256"
     }
   }
 }
 
-data "template_file" "signatures_script" {
-  template = file("${path.module}/templates/signatures_job.py.tpl")
-  vars = {
+resource "aws_s3_bucket_lifecycle_configuration" "glue_resources" {
+  bucket = aws_s3_bucket.glue_resources.id
+
+  rule {
+    id = "Remove temp files over a week old"
+    status = "Enabled"
+
+    filter {
+      prefix = "production/temp/"
+    }
+
+    expiration {
+      days = 7
+    }
+
+    abort_incomplete_multipart_upload {
+      days_after_initiation = 7 # Note: must be greater than 0
+    }
+  }
+}
+
+locals {
+  signatures_script = templatefile("${path.module}/templates/signatures_job.py.tpl", {
     catalog_database_name = aws_glue_catalog_database.catalog_db.name
     redshift_database_name = var.redshift_database_name
     redshift_schema = var.redshift_schema
     redshift_connection_name = aws_glue_connection.redshift_connection.name
-  }
+  })
 }
 
-resource "aws_s3_bucket_object" "signatures_script" {
+resource "aws_s3_object" "signatures_script" {
   bucket = aws_s3_bucket.glue_resources.id
   key = "${var.controlshift_environment}/signatures_job.py"
   acl = "private"
 
-  content = data.template_file.signatures_script.rendered
+  content = local.signatures_script
 }
 
 resource "aws_iam_role" "glue_service_role" {
@@ -134,6 +177,8 @@ resource "aws_glue_job" "signatures_full" {
   name = "cs-${var.controlshift_environment}-signatures-full"
   connections = [ aws_glue_connection.redshift_connection.name ]
   glue_version = "3.0"
+  number_of_workers = 9
+  worker_type = "G.1X"
   default_arguments = {
     "--TempDir": "s3://${aws_s3_bucket.glue_resources.bucket}/${var.controlshift_environment}/temp",
     "--job-bookmark-option": "job-bookmark-disable",

diff --git a/iam.tf b/iam.tf
@@ -34,6 +34,22 @@ data "aws_iam_policy_document" "receiver_execution_policy" {
     resources = ["arn:aws:sqs:${var.aws_region}:*:${aws_sqs_queue.receiver_queue.name}",
                  "arn:aws:sqs:${var.aws_region}:*:${aws_sqs_queue.receiver_queue_glue.name}"]
   }
+
+  # allow the receiver lambda to send messages to Firehose streams
+  statement {
+      effect = "Allow"
+      actions = [
+          "firehose:DeleteDeliveryStream",
+          "firehose:PutRecord",
+          "firehose:StartDeliveryStreamEncryption",
+          "firehose:CreateDeliveryStream",
+          "firehose:PutRecordBatch",
+          "firehose:StopDeliveryStreamEncryption",
+          "firehose:UpdateDestination"
+      ]
+      resources = ["arn:aws:firehose:${var.aws_region}:*:deliverystream/${var.email_open_firehose_stream}",
+      "arn:aws:firehose:${var.aws_region}:*:deliverystream/${var.email_click_firehose_stream}"]
+  }
 }
 
 resource "aws_iam_role_policy" "lambda_receiver" {

diff --git a/lambdas/receiver.js b/lambdas/receiver.js
@@ -1,13 +1,28 @@
 'use strict';
 
 const AWS = require('aws-sdk');
-
 // Set the region
 AWS.config.update({region: process.env.AWS_REGION});
 
 // Create an SQS service object
 const sqs = new AWS.SQS();
 
+// Create a Firehose service object
+const firehose = new AWS.Firehose();
+
+function putFirehose(data, stream) {
+  let params = {
+    DeliveryStreamName: stream,
+    Record:{
+      Data: data
+    }
+  };
+  firehose.putRecord(params, function(err, data) {
+    if (err) console.log(err, err.stack);
+    else     console.log('Record added:',data);
+  });
+}
+
 async function enqueueTask(receivedData, kind) {
   console.log("Processing: " + receivedData.url);
 
@@ -21,7 +36,7 @@ async function enqueueTask(receivedData, kind) {
 
   messageBody['kind'] =  kind;
 
-  const jsonMessageBody = JSON.stringify(messageBody)
+  const jsonMessageBody = JSON.stringify(messageBody);
 
   const loaderQueueParams = {
     MessageBody: jsonMessageBody,
@@ -69,6 +84,12 @@ exports.handler = async (event) => {
     } else if(receivedJSON.type === 'data.incremental_table_exported'){
       await enqueueTask(receivedJSON.data, 'incremental');
       return sendResponse({"status": "processed"});
+    } else if(receivedJSON.type === 'email.open' && process.env.EMAIL_OPEN_FIREHOSE_STREAM !== null && process.env.EMAIL_OPEN_FIREHOSE_STREAM !== ''){
+      await putFirehose(JSON.stringify(receivedJSON.data), process.env.EMAIL_OPEN_FIREHOSE_STREAM);
+      return sendResponse({"status": "processed"});
+    } else if(receivedJSON.type === 'email.click' && process.env.EMAIL_CLICK_FIREHOSE_STREAM !== null && process.env.EMAIL_CLICK_FIREHOSE_STREAM !== ''){
+      await putFirehose(JSON.stringify(receivedJSON.data), process.env.EMAIL_CLICK_FIREHOSE_STREAM);
+      return sendResponse({"status": "processed"});
     } else {
       return Promise.resolve(sendResponse({"status": "skipped", "payload": receivedJSON}));
     }

diff --git a/loader.tf b/loader.tf
@@ -1,10 +1,10 @@
 resource "aws_lambda_function" "loader" {
   s3_bucket = local.lambda_buckets[var.aws_region]
-  s3_key = "LambdaRedshiftLoader/AWSLambdaRedshiftLoader-2.7.8.zip"
+  s3_key = "LambdaRedshiftLoader/AWSLambdaRedshiftLoader-2.8.3.zip"
   function_name = "controlshift-redshift-loader"
   role          = aws_iam_role.loader_lambda_role.arn
   handler       = "index.handler"
-  runtime       = "nodejs12.x"
+  runtime       = "nodejs16.x"
   timeout       = 900
 
   vpc_config {

diff --git a/receiver.tf b/receiver.tf
@@ -9,16 +9,23 @@ resource "aws_lambda_function" "receiver_lambda" {
   function_name = "controlshift-webhook-handler"
   role          = aws_iam_role.receiver_lambda_role.arn
   handler       = "receiver.handler"
-  runtime       = "nodejs12.x"
+  runtime       = "nodejs16.x"
   timeout       = var.receiver_timeout
   source_code_hash = data.archive_file.receiver_zip.output_base64sha256
 
   environment {
     variables = {
       SQS_QUEUE_URL = aws_sqs_queue.receiver_queue.id
       GLUE_SQS_QUEUE_URL = aws_sqs_queue.receiver_queue_glue.id
+      EMAIL_OPEN_FIREHOSE_STREAM = var.email_open_firehose_stream
+      EMAIL_CLICK_FIREHOSE_STREAM = var.email_click_firehose_stream
     }
   }
+
+  // This prevents noisy logs from cluttering up datadog
+  tags = {
+    datadog = "exclude"
+  }
 }
 
 resource "aws_api_gateway_rest_api" "receiver" {

diff --git a/run_glue_crawler.tf b/run_glue_crawler.tf
@@ -10,7 +10,7 @@ resource "aws_lambda_function" "glue_crawler_lambda" {
   function_name = "controlshift-run-glue-crawler"
   role          = aws_iam_role.run_glue_crawler_lambda_role.arn
   handler       = "run-glue-crawler.handler"
-  runtime       = "nodejs12.x"
+  runtime       = "nodejs16.x"
   timeout       = 60
   source_code_hash = data.archive_file.run_glue_crawler_zip.output_base64sha256
 

diff --git a/run_glue_job.tf b/run_glue_job.tf
@@ -10,7 +10,7 @@ resource "aws_lambda_function" "glue_job_lambda" {
   function_name = "controlshift-run-glue-job"
   role          = aws_iam_role.run_glue_job_lambda_role.arn
   handler       = "run-glue-job.handler"
-  runtime       = "nodejs12.x"
+  runtime       = "nodejs16.x"
   timeout       = 60
   source_code_hash = data.archive_file.run_glue_job_zip.output_base64sha256