fix: use dlq to avoid sqs bottleneck + ignore both None & empty array in validation

cumulus_lambda_functions/daac_archiver/daac_archiver_logic.py

@@ -82,7 +82,7 @@ def send_to_daac_internal(self, uds_cnm_json: dict):
         granule_identifier = UdsCollections.decode_identifier(uds_cnm_json['identifier'])  # This is normally meant to be for collection. Since our granule ID also has collection id prefix. we can use this.
         self.__archive_index_logic.set_tenant_venue(granule_identifier.tenant, granule_identifier.venue)
         daac_config = self.__archive_index_logic.percolate_document(uds_cnm_json['identifier'])
-        if daac_config is None:
+        if daac_config is None or len(daac_config) < 1:
             LOGGER.debug(f'uds_cnm_json is not configured for archival. uds_cnm_json: {uds_cnm_json}')
             return
         daac_config = daac_config[0]  # TODO This is currently not supporting more than 1 daac.

tf-module/unity-cumulus/daac_archiver.tf

@@ -69,34 +69,13 @@ resource "aws_sns_topic_policy" "daac_archiver_response_policy" {
   })
 }
 
-resource "aws_sqs_queue" "daac_archiver_response" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
-  name                      = "${var.prefix}-daac_archiver_response"
-  delay_seconds             = 0
-  max_message_size          = 262144
-  message_retention_seconds = 345600
-  visibility_timeout_seconds = var.granules_cnm_ingester__sqs_visibility_timeout_seconds  // Used as cool off time in seconds. It will wait for 5 min if it fails
-  receive_wait_time_seconds = 0
-  policy = templatefile("${path.module}/sqs_policy.json", {
-    region: var.aws_region,
-    roleArn: var.lambda_processing_role_arn,
-    accountId: local.account_id,
-    sqsName: "${var.prefix}-daac_archiver_response",
-  })
-  tags = var.tags
-}
-
-resource "aws_sns_topic_subscription" "daac_archiver_response_subscription" { // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic_subscription
-  topic_arn = aws_sns_topic.daac_archiver_response.arn
-  protocol  = "sqs"
-  endpoint  = aws_sqs_queue.daac_archiver_response.arn
-#  filter_policy_scope = "MessageBody"  // MessageAttributes. not using attributes
-#  filter_policy = templatefile("${path.module}/ideas_api_job_results_filter_policy.json", {})
-}
-
+module "daac_archiver_response" {
+  source = "./sqs--sns-lambda-connector"
 
-resource "aws_lambda_event_source_mapping" "daac_archiver_response_queue_lambda_trigger" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping#sqs
-  event_source_arn = aws_sqs_queue.daac_archiver_response.arn
-  function_name    = aws_lambda_function.daac_archiver_response.arn
-  batch_size = 1
-  enabled = true
+  account_id                 = local.account_id
+  lambda_arn                 = aws_lambda_function.daac_archiver_response.arn
+  lambda_processing_role_arn = var.lambda_processing_role_arn
+  name                       = "daac_archiver_response"
+  prefix                     = var.prefix
+  sns_arn                    = aws_sns_topic.daac_archiver_response.arn
 }

tf-module/unity-cumulus/granules_cnm_ingester.tf

@@ -44,64 +44,18 @@ resource "aws_sns_topic_policy" "granules_cnm_ingester_policy" {
   })
 }
 
-resource "aws_sqs_queue" "dead_letter_granules_cnm_ingester" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
-  // TODO how to notify admin for failed ingestion?
-  tags = var.tags
-  name                      = "${var.prefix}-dead_letter_granules_cnm_ingester"
-  delay_seconds             = 0
-  max_message_size          = 262144
-  message_retention_seconds = 345600
-  visibility_timeout_seconds = 300
-  receive_wait_time_seconds = 0
-  policy = templatefile("${path.module}/sqs_policy.json", {
-    region: var.aws_region,
-    roleArn: var.lambda_processing_role_arn,
-    accountId: local.account_id,
-    sqsName: "${var.prefix}-dead_letter_granules_cnm_ingester",
-  })
-//  redrive_policy = jsonencode({
-//    deadLetterTargetArn = aws_sqs_queue.terraform_queue_deadletter.arn
-//    maxReceiveCount     = 4
-//  })
-//  tags = {
-//    Environment = "production"
-//  }
-}
 
-resource "aws_sqs_queue" "granules_cnm_ingester" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
-  name                      = "${var.prefix}-granules_cnm_ingester"
-  delay_seconds             = 0
-  max_message_size          = 262144
-  message_retention_seconds = 345600
-  visibility_timeout_seconds = var.granules_cnm_ingester__sqs_visibility_timeout_seconds  // Used as cool off time in seconds. It will wait for 5 min if it fails
-  receive_wait_time_seconds = 0
-  policy = templatefile("${path.module}/sqs_policy.json", {
-    region: var.aws_region,
-    roleArn: var.lambda_processing_role_arn,
-    accountId: local.account_id,
-    sqsName: "${var.prefix}-granules_cnm_ingester",
-  })
-  redrive_policy = jsonencode({
-    deadLetterTargetArn = aws_sqs_queue.dead_letter_granules_cnm_ingester.arn
-    maxReceiveCount     = var.granules_cnm_ingester__sqs_retried_count  // How many times it will be retried.
-  })
-  tags = var.tags
-}
+module "granules_cnm_ingester" {
+  source = "./sqs--sns-lambda-connector"
 
-resource "aws_sns_topic_subscription" "granules_cnm_ingester_topic_subscription" { // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic_subscription
-  topic_arn = aws_sns_topic.granules_cnm_ingester.arn
-  protocol  = "sqs"
-  endpoint  = aws_sqs_queue.granules_cnm_ingester.arn
-#  filter_policy_scope = "MessageBody"  // MessageAttributes. not using attributes
-#  filter_policy = templatefile("${path.module}/ideas_api_job_results_filter_policy.json", {})
+  account_id                 = local.account_id
+  lambda_arn                 = aws_lambda_function.granules_cnm_ingester.arn
+  lambda_processing_role_arn = var.lambda_processing_role_arn
+  name                       = "granules_cnm_ingester"
+  prefix                     = var.prefix
+  sns_arn                    = aws_sns_topic.granules_cnm_ingester.arn
 }
 
-resource "aws_lambda_event_source_mapping" "granules_cnm_ingester_queue_lambda_trigger" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping#sqs
-  event_source_arn = aws_sqs_queue.granules_cnm_ingester.arn
-  function_name    = aws_lambda_function.granules_cnm_ingester.arn
-  batch_size = 1
-  enabled = true
-}
 ################# << CNM Response Writer >> ########################
 
 resource "aws_lambda_function" "granules_cnm_response_writer" {
@@ -131,33 +85,13 @@ data "aws_sns_topic" "granules_cnm_response_topic" {  // https://registry.terraf
   name              = var.granules_cnm_response_topic
 }
 
-resource "aws_sqs_queue" "granules_cnm_response_writer" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
-  name                      = "${var.prefix}-granules_cnm_response_writer"
-  delay_seconds             = 0
-  max_message_size          = 262144
-  message_retention_seconds = 345600
-  visibility_timeout_seconds = var.granules_cnm_ingester__sqs_visibility_timeout_seconds  // Used as cool off time in seconds. It will wait for 5 min if it fails
-  receive_wait_time_seconds = 0
-  policy = templatefile("${path.module}/sqs_policy.json", {
-    region: var.aws_region,
-    roleArn: var.lambda_processing_role_arn,
-    accountId: local.account_id,
-    sqsName: "${var.prefix}-granules_cnm_response_writer",
-  })
-  tags = var.tags
-}
+module "granules_cnm_response_writer" {
+  source = "./sqs--sns-lambda-connector"
 
-resource "aws_sns_topic_subscription" "granules_cnm_response_writer_subscription" { // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic_subscription
-  topic_arn = data.aws_sns_topic.granules_cnm_response_topic.arn
-  protocol  = "sqs"
-  endpoint  = aws_sqs_queue.granules_cnm_response_writer.arn
-#  filter_policy_scope = "MessageBody"  // MessageAttributes. not using attributes
-#  filter_policy = templatefile("${path.module}/ideas_api_job_results_filter_policy.json", {})
+  account_id                 = local.account_id
+  lambda_arn                 = aws_lambda_function.granules_cnm_response_writer.arn
+  lambda_processing_role_arn = var.lambda_processing_role_arn
+  name                       = "granules_cnm_response_writer"
+  prefix                     = var.prefix
+  sns_arn                    = data.aws_sns_topic.granules_cnm_response_topic.arn
 }
-
-resource "aws_lambda_event_source_mapping" "granules_cnm_response_writer_lambda_trigger" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping#sqs
-  event_source_arn = aws_sqs_queue.granules_cnm_response_writer.arn
-  function_name    = aws_lambda_function.granules_cnm_response_writer.arn
-  batch_size = 1
-  enabled = true
-}

tf-module/unity-cumulus/sqs--sns-lambda-connector/sqs-sns-lambda.tf

@@ -0,0 +1,58 @@
+resource "aws_sqs_queue" "dlq" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
+  // TODO how to notify admin for failed ingestion?
+  tags = var.tags
+  name                      = "${var.prefix}-dlq-${var.name}"
+  delay_seconds             = 0
+  max_message_size          = 262144
+  message_retention_seconds = 345600
+  visibility_timeout_seconds = 300
+  receive_wait_time_seconds = 0
+  policy = templatefile("${path.module}/sqs_policy.json", {
+    region: var.aws_region,
+    roleArn: var.lambda_processing_role_arn,
+    accountId: var.account_id,
+    sqsName: "${var.prefix}-dlq-${var.name}",
+  })
+//  redrive_policy = jsonencode({
+//    deadLetterTargetArn = aws_sqs_queue.terraform_queue_deadletter.arn
+//    maxReceiveCount     = 4
+//  })
+//  tags = {
+//    Environment = "production"
+//  }
+}
+
+resource "aws_sqs_queue" "main_sqs" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
+  name                      = "${var.prefix}-${var.name}"
+  delay_seconds             = 0
+  max_message_size          = 262144
+  message_retention_seconds = 345600
+  visibility_timeout_seconds = var.cool_off  // Used as cool off time in seconds. It will wait for 5 min if it fails
+  receive_wait_time_seconds = 0
+  policy = templatefile("${path.module}/sqs_policy.json", {
+    region: var.aws_region,
+    roleArn: var.lambda_processing_role_arn,
+    accountId: var.account_id,
+    sqsName: "${var.prefix}-${var.name}",
+  })
+  redrive_policy = jsonencode({
+    deadLetterTargetArn = aws_sqs_queue.dlq.arn
+    maxReceiveCount     = var.retried_count  // How many times it will be retried.
+  })
+  tags = var.tags
+}
+
+resource "aws_sns_topic_subscription" "granules_cnm_ingester_topic_subscription" { // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic_subscription
+  topic_arn = var.sns_arn
+  protocol  = "sqs"
+  endpoint  = aws_sqs_queue.main_sqs.arn
+#  filter_policy_scope = "MessageBody"  // MessageAttributes. not using attributes
+#  filter_policy = templatefile("${path.module}/ideas_api_job_results_filter_policy.json", {})
+}
+
+resource "aws_lambda_event_source_mapping" "granules_cnm_ingester_queue_lambda_trigger" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping#sqs
+  event_source_arn = aws_sqs_queue.main_sqs.arn
+  function_name    = var.lambda_arn
+  batch_size = var.sqs_batch_size
+  enabled = true
+}

tf-module/unity-cumulus/sqs--sns-lambda-connector/variables.tf

@@ -0,0 +1,41 @@
+variable "prefix" {
+  type = string
+}
+variable "name" {
+  type = string
+}
+variable "sns_arn" {
+  type = string
+}
+variable "lambda_arn" {
+  type = string
+}
+variable "sqs_batch_size" {
+  type = number
+  default = 1
+}
+variable "retried_count" {
+  type = number
+  default = 3
+}
+variable "cool_off" {
+  type = number
+  default = 300
+  description = "visibility time out for sqs. in seconds"
+}
+variable "aws_region" {
+  type    = string
+  default = "us-west-2"
+}
+variable "tags" {
+  description = "Tags to be applied to Cumulus resources that support tags"
+  type        = map(string)
+  default     = {}
+}
+variable "lambda_processing_role_arn" {
+  type = string
+}
+variable "account_id" {
+  type = string
+  description = "AWS Account ID"
+}

tf-module/unity-cumulus/sqs-sns.tf

@@ -2,40 +2,13 @@ data "aws_sns_topic" "report_granules_topic" {  // https://registry.terraform.io
   name              = var.report_granules_topic
 }
 
-resource "aws_sqs_queue" "granules_to_es_queue" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sqs_queue
-  name                      = "${var.prefix}-granules_to_es_queue"
-  delay_seconds             = 0
-  max_message_size          = 262144
-  message_retention_seconds = 345600
-  visibility_timeout_seconds = 310
-  receive_wait_time_seconds = 0
-  policy = templatefile("${path.module}/sqs_policy.json", {
-    region: var.aws_region,
-    roleArn: var.lambda_processing_role_arn,
-    accountId: local.account_id,
-    sqsName: "${var.prefix}-granules_to_es_queue",
-  })
-//  redrive_policy = jsonencode({
-//    deadLetterTargetArn = aws_sqs_queue.terraform_queue_deadletter.arn
-//    maxReceiveCount     = 4
-//  })
-//  tags = {
-//    Environment = "production"
-//  }
-  tags = var.tags
-}
+module "granules_to_es" {
+  source = "./sqs--sns-lambda-connector"
 
-resource "aws_sns_topic_subscription" "report_granules_topic_subscription" { // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/sns_topic_subscription
-  topic_arn = data.aws_sns_topic.report_granules_topic.arn
-  protocol  = "sqs"
-  endpoint  = aws_sqs_queue.granules_to_es_queue.arn
-#  filter_policy_scope = "MessageBody"  // MessageAttributes. not using attributes
-#  filter_policy = templatefile("${path.module}/ideas_api_job_results_filter_policy.json", {})
+  account_id                 = local.account_id
+  lambda_arn                 = aws_lambda_function.granules_to_es.arn
+  lambda_processing_role_arn = var.lambda_processing_role_arn
+  name                       = "granules_to_es"
+  prefix                     = var.prefix
+  sns_arn                    = data.aws_sns_topic.report_granules_topic.arn
 }
-
-resource "aws_lambda_event_source_mapping" "granules_to_es_queue_lambda_trigger" {  // https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping#sqs
-  event_source_arn = aws_sqs_queue.granules_to_es_queue.arn
-  function_name    = aws_lambda_function.granules_to_es.arn
-  batch_size = 1
-  enabled = true
-}