Merge pull request #10 from rjena5300/sns-sub-eventtype-filter

SQS SNS sub eventtype filter

CHANGELOG.md

@@ -3,6 +3,10 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/) and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
+## [1.3.2] - 2020-09-18
+### Added
+- eventType filter added to SQS queue message.
+
 ## [1.3.1] - 2020-07-30
 ### Added
 - SQS queue message filter.

README.md

@@ -29,6 +29,7 @@ For more information please refer to the main [Apiary](https://github.com/Expedi
 | subnets | ECS container subnets. | list | n/a | yes |
 | target\_metastore\_uri | Target Metastore URI for Shunting Yard. | string | n/a | yes |
 | vpc\_id | VPC ID. | string | n/a | yes |
+|exclude\_event\_list | event to exclude from Shunting Yard Replication.  Supported Format: [ "DROP_PARTITION"] Wildcards are not supported, i.e. you need to specify each event explicitly. | list | `<list>` | no |
 
 ## Usage
 

cloudwatch.tf

@@ -118,22 +118,22 @@ ${join("", data.template_file.sqs_widgets.*.rendered)}
 locals {
   alerts = [
     {
-      alarm_name  = "${local.instance_alias}-cpu"
-      namespace   = "AWS/ECS"
+      alarm_name = "${local.instance_alias}-cpu"
+      namespace = "AWS/ECS"
       metric_name = "CPUUtilization"
-      threshold   = "90"
+      threshold = "90"
     },
     {
-      alarm_name  = "${local.instance_alias}-memory"
-      namespace   = "AWS/ECS"
+      alarm_name = "${local.instance_alias}-memory"
+      namespace = "AWS/ECS"
       metric_name = "MemoryUtilization"
-      threshold   = "80"
+      threshold = "80"
     },
     {
-      alarm_name  = "${local.instance_alias}-stale-messages"
-      namespace   = "AWS/SQS"
+      alarm_name = "${local.instance_alias}-stale-messages"
+      namespace = "AWS/SQS"
       metric_name = "ApproximateAgeOfOldestMessage"
-      threshold   = "${var.shuntingyard_sqs_queue_stale_messages_timeout}"
+      threshold = "${var.shuntingyard_sqs_queue_stale_messages_timeout}"
     },
   ]
 
@@ -153,17 +153,17 @@ locals {
 }
 
 resource "aws_cloudwatch_metric_alarm" "shuntingyard_alert" {
-  count               = "${length(local.alerts)}"
-  alarm_name          = "${lookup(local.alerts[count.index], "alarm_name")}"
+  count = "${length(local.alerts)}"
+  alarm_name = "${lookup(local.alerts[count.index], "alarm_name")}"
   comparison_operator = "${lookup(local.alerts[count.index], "comparison_operator", "GreaterThanOrEqualToThreshold")}"
-  metric_name         = "${lookup(local.alerts[count.index], "metric_name")}"
-  namespace           = "${lookup(local.alerts[count.index], "namespace")}"
-  period              = "${lookup(local.alerts[count.index], "period", "120")}"
-  evaluation_periods  = "${lookup(local.alerts[count.index], "evaluation_periods", "2")}"
-  statistic           = "Average"
-  threshold           = "${lookup(local.alerts[count.index], "threshold")}"
+  metric_name = "${lookup(local.alerts[count.index], "metric_name")}"
+  namespace = "${lookup(local.alerts[count.index], "namespace")}"
+  period = "${lookup(local.alerts[count.index], "period", "120")}"
+  evaluation_periods = "${lookup(local.alerts[count.index], "evaluation_periods", "2")}"
+  statistic = "Average"
+  threshold = "${lookup(local.alerts[count.index], "threshold")}"
 
   insufficient_data_actions = []
-  dimensions                = "${local.dimensions[count.index]}"
-  alarm_actions             = ["${aws_sns_topic.shuntingyard_ops_sns.arn}"]
+  dimensions = "${local.dimensions[count.index]}"
+  alarm_actions = ["${aws_sns_topic.shuntingyard_ops_sns.arn}"]
 }

common.tf

@@ -5,7 +5,7 @@
  */
 
 locals {
-  instance_alias = "${ var.instance_name == "" ? "shuntingyard" : format("shuntingyard-%s",var.instance_name) }"
+  instance_alias = "${var.instance_name == "" ? "shuntingyard" : format("shuntingyard-%s", var.instance_name)}"
 }
 
 data "aws_vpc" "shuntingyard_vpc" {

iam-policy-s3-buckets.tf

@@ -29,8 +29,8 @@ resource "aws_iam_role_policy" "s3_for_shuntingyard" {
                 "s3:PutObjectVersionTagging"
             ],
             "Resource": [
-                          "${join("\",\"", formatlist("arn:aws:s3:::%s",var.allowed_s3_buckets))}",
-                          "${join("\",\"", formatlist("arn:aws:s3:::%s/*",var.allowed_s3_buckets))}"
+                          "${join("\",\"", formatlist("arn:aws:s3:::%s", var.allowed_s3_buckets))}",
+                          "${join("\",\"", formatlist("arn:aws:s3:::%s/*", var.allowed_s3_buckets))}"
                         ]
         }
     ]

iam.tf

@@ -27,7 +27,7 @@ EOF
 }
 
 resource "aws_iam_role_policy_attachment" "task_exec_managed" {
-  role       = "${aws_iam_role.shuntingyard_task_exec.id}"
+  role = "${aws_iam_role.shuntingyard_task_exec.id}"
   policy_arn = "arn:aws:iam::aws:policy/service-role/AmazonECSTaskExecutionRolePolicy"
 }
 

sqs.tf

@@ -35,7 +35,9 @@ resource "aws_sqs_queue_policy" "shuntingyard_sqs_queue_policy" {
 POLICY
 }
 
+# default filter
 data "template_file" "sqs_hive_metastore_sns_subscription_filter" {
+  count = "${length(var.exclude_event_list) == 0 ? 1 : 0}"
   template = <<EOF
   {
     "qualifiedTableName": [ $${tables_list} ]
@@ -48,8 +50,40 @@ EOF
 }
 
 resource "aws_sns_topic_subscription" "sqs_hive_metastore_sns_subscription" {
+  count         = "${length(var.exclude_event_list) == 0 ? 1 : 0}"
   topic_arn     = "${var.metastore_events_sns_topic}"
   protocol      = "sqs"
   endpoint      = "${aws_sqs_queue.shuntingyard_sqs_queue.arn}"
-  filter_policy = "${data.template_file.sqs_hive_metastore_sns_subscription_filter.rendered}"
+  filter_policy = "${join("", data.template_file.sqs_hive_metastore_sns_subscription_filter.*.rendered)}"
+}
+
+# filter with metastore event type
+data "template_file" "sqs_hive_metastore_sns_subscription_event_filter" {
+  count    = "${length(var.exclude_event_list) != 0 ? 1 : 0}"
+  template = <<EOF
+  {
+    "qualifiedTableName": [ $${tables_list} ],
+    "eventType": [
+        {
+          "anything-but": [ $${event_list} ]
+        }
+      ]
+  }
+EOF
+
+  vars {
+    tables_list = "${join(",", formatlist("\"%s\"", var.selected_tables))}"
+  }
+  vars {
+    event_list = "${join(",", formatlist("\"%s\"", var.exclude_event_list))}"
+  }
+}
+
+
+resource "aws_sns_topic_subscription" "sqs_hive_metastore_sns_event_filter_subscription" {
+  count = "${length(var.exclude_event_list) != 0 ? 1 : 0}"
+  topic_arn = "${var.metastore_events_sns_topic}"
+  protocol = "sqs"
+  endpoint = "${aws_sqs_queue.shuntingyard_sqs_queue.arn}"
+  filter_policy = "${join("", data.template_file.sqs_hive_metastore_sns_subscription_event_filter.*.rendered)}"
 }

variables.tf

@@ -37,7 +37,7 @@ The amount of memory (in MiB) used to allocate for the Shunting Yard container.
 Valid values: https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-cpu-memory-error.html
 EOF
 
-  type    = "string"
+  type = "string"
   default = "4096"
 }
 
@@ -111,7 +111,7 @@ Supported Format: [ "database_1.table_1", "database_2.table_2" ]
 Wildcards are not supported, i.e. you need to specify each table explicitly.
 EOF
 
-  type    = "list"
+  type = "list"
   default = []
 }
 
@@ -129,3 +129,13 @@ variable "docker_registry_auth_secret_name" {
   type        = "string"
   default     = ""
 }
+
+variable "exclude_event_list" {
+  description = <<EOF
+Filter on eventtype (optional)
+Supported Format: [ "DROP_PARTITION" ]
+Wildcards are not supported, i.e. you need to specify each eventtype explicitly.
+EOF
+  type = "list"
+  default = []
+}