enha: added trigger to define DynamoDB stream or SQS queue as event source

kfc-manager · kfc-manager · commit 7b3990ccfa13 · 2024-04-03T10:15:29.000+02:00
diff --git a/README.md b/README.md
@@ -25,6 +25,7 @@ This module provides a Lambda function which logs to CloudWatch. If no image URI
 | ------------- | --------------------------------------------------------------------------------------------- | -------------- | -------- | :------: |
 | identifier    | Unique identifier to differentiate global resources.                                          | `string`       | n/a      |   yes    |
 | policies      | List of IAM policy ARNs for the Lambda's IAM role.                                            | `list(string)` | []       |    no    |
+| trigger       | Object to define trigger of the Lambda function.                                              | `object`       | null     |    no    |
 | vpc_config    | Object to define the subnets and security groups for the Lambda function.                     | `object`       | null     |    no    |
 | log_config    | Object to define logging configuration of the Lambda function to CloudWatch.                  | `object`       | null     |    no    |
 | image         | Object of the image which will be pulled by the Lambda function to execute.                   | `object`       | null     |    no    |
@@ -34,12 +35,23 @@ This module provides a Lambda function which logs to CloudWatch. If no image URI
 | env_variables | A map of environment variables for the Lambda function at runtime.                            | `map(string)`  | {}       |    no    |
 | tags          | A map of tags to add to all resources.                                                        | `map(string)`  | {}       |    no    |
 
+### `trigger`
+
+| Name            | Description                                                                                                                | Type     | Default | Required |
+| --------------- | -------------------------------------------------------------------------------------------------------------------------- | -------- | ------- | :------: |
+| queue_arn       | The ARN of the SQS queue, which triggers the Lambda function. Must be defined if 'stream_arn' is not defined.              | `string` | null    |    no    |
+| stream_arn      | The ARN of the DynamoDB stream, which triggers the Lambda function. Must be defined if 'queue_arn' is not defined.         | `string` | null    |    no    |
+| batch_size      | Amount of items a single Lambda invocation processes from the source.                                                      | `number` | 1       |    no    |
+| max_concurrency | Maximum amount of Lambda functions the SQS queue invokes concurrently.                                                     | `number` | 1000    |    no    |
+| max_retries     | Maximum retry attempts the Lambda function makes to process the DynamoDB stream. The value '-1' means it tries infinitely. | `number` | -1      |    no    |
+| filter          | A filter pattern of which messages the Lambda function processes. Must be in JSON format.                                  | `string` | null    |    no    |
+
 ### `vpc_config`
 
 | Name            | Description                                                  | Type           | Default | Required |
 | --------------- | ------------------------------------------------------------ | -------------- | ------- | :------: |
 | subnets         | List of subnet IDs in which the Lambda function will run in. | `list(string)` | n/a     |   yes    |
-| security_groups | List of security group IDs the Lambda function will hold.    | `list(string)` | n/a     |   yes    |
+| security_groups | List of security group IDs the Lambda function will hold.    | `list(string)` | []      |    no    |
 
 ### `log_config`
 
@@ -78,6 +90,18 @@ module "function" {
     "arn:aws:iam::aws:policy/AdministratorAccess-Amplify"
   ]
 
+  trigger = {
+    queue_arn       = "arn:aws:sqs:eu-central-1:444455556666:queue1"
+    batch_size      = 10
+    max_concurrency = 100
+    filter = jsonencode({
+      body = {
+        Temperature : [{ numeric : [">", 0, "<=", 100] }]
+        Location : ["New York"]
+      }
+    })
+  }
+
   log_config = {
     retention_in_days = 7
   }
diff --git a/main.tf b/main.tf
@@ -5,7 +5,7 @@
 resource "aws_cloudwatch_log_group" "main" {
   count             = var.log_config != null ? 1 : 0
   name              = "${var.identifier}-lambda"
-  retention_in_days = try(var.log_config["retention_in_days"], null)
+  retention_in_days = var.log_config["retention_in_days"]
 
   tags = var.tags
 }
@@ -116,7 +116,7 @@ resource "aws_lambda_function" "main" {
   function_name = var.identifier
   package_type  = "Image"
   role          = aws_iam_role.main.arn
-  image_uri     = var.image == null ? "${aws_ecr_repository.main[0].repository_url}:latest" : try(var.image["uri"], null)
+  image_uri     = var.image == null ? "${aws_ecr_repository.main[0].repository_url}:latest" : var.image["uri"]
   memory_size   = var.memory_size
   timeout       = var.timeout
   architectures = [var.architecture]
@@ -128,8 +128,8 @@ resource "aws_lambda_function" "main" {
   dynamic "vpc_config" {
     for_each = var.vpc_config != null ? [1] : []
     content {
-      subnet_ids         = try(var.vpc_config["subnets"], null)
-      security_group_ids = try(var.vpc_config["security_groups"], null)
+      subnet_ids         = var.vpc_config["subnets"]
+      security_group_ids = var.vpc_config["security_groups"]
     }
   }
 
@@ -143,3 +143,37 @@ resource "aws_lambda_function" "main" {
 
   tags = var.tags
 }
+
+resource "aws_lambda_event_source_mapping" "queue" {
+  count            = try(var.trigger["queue_arn"], null) != null ? 1 : 0
+  event_source_arn = var.trigger["queue_arn"]
+  enabled          = true
+  function_name    = aws_lambda_function.main.arn
+  batch_size       = var.trigger["batch_size"]
+
+  dynamic "scaling_config" {
+    for_each = try(var.trigger["max_concurrency"], null) != null ? [1] : []
+    content {
+      maximum_concurrency = var.trigger["max_concurrency"]
+    }
+  }
+
+  dynamic "filter_criteria" {
+    for_each = try(var.trigger["filter"], null) != null ? [1] : []
+    content {
+      filter {
+        pattern = var.trigger["filter"]
+      }
+    }
+  }
+}
+
+resource "aws_lambda_event_source_mapping" "stream" {
+  count                  = try(var.trigger["stream_arn"], null) != null ? 1 : 0
+  event_source_arn       = var.trigger["stream_arn"]
+  enabled                = true
+  function_name          = aws_lambda_function.main.arn
+  batch_size             = var.trigger["batch_size"]
+  maximum_retry_attempts = var.trigger["max_retries"]
+  starting_position      = "LATEST"
+}
diff --git a/tests/trigger.tftest.hcl b/tests/trigger.tftest.hcl
@@ -0,0 +1,148 @@
+provider "aws" {
+  region = "eu-central-1"
+  default_tags {
+    tags = {
+      Environment = "Test"
+    }
+  }
+}
+
+run "queue_and_stream_arn_defined" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      queue_arn  = "test-queue"
+      stream_arn = "test-stream"
+    }
+  }
+
+  expect_failures = [var.trigger]
+}
+
+run "queue_and_stream_arn_undefined" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger    = {}
+  }
+
+  expect_failures = [var.trigger]
+}
+
+run "filter_without_queue" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      stream_arn = "test-stream"
+      filter     = "test"
+    }
+  }
+
+  expect_failures = [var.trigger]
+}
+
+run "filter_with_queue" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      queue_arn = "test-queue"
+      filter    = "test"
+    }
+  }
+}
+
+run "max_concurrency_upper_limit" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      queue_arn       = "test-queue"
+      max_concurrency = 1001
+    }
+  }
+
+  expect_failures = [var.trigger]
+}
+
+run "max_concurrency_lower_limit" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      queue_arn       = "test-queue"
+      max_concurrency = 1
+    }
+  }
+
+  expect_failures = [var.trigger]
+}
+
+run "queue_trigger" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      queue_arn = "test-queue"
+    }
+  }
+
+  assert {
+    condition     = length(aws_lambda_event_source_mapping.queue) == 1
+    error_message = "Queue source mapping was not created"
+  }
+
+  assert {
+    condition     = length(aws_lambda_event_source_mapping.stream) == 0
+    error_message = "Stream source mapping was created unexpectedly"
+  }
+}
+
+run "stream_trigger" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger = {
+      stream_arn = "test-stream"
+    }
+  }
+
+  assert {
+    condition     = length(aws_lambda_event_source_mapping.queue) == 0
+    error_message = "Queue source mapping was created unexpectedly"
+  }
+
+  assert {
+    condition     = length(aws_lambda_event_source_mapping.stream) == 1
+    error_message = "Stream source mapping was not created"
+  }
+}
+
+run "without_trigger" {
+  command = plan
+
+  variables {
+    identifier = "abc"
+    trigger    = null
+  }
+
+  assert {
+    condition     = length(aws_lambda_event_source_mapping.queue) == 0
+    error_message = "Queue source mapping was created unexpectedly"
+  }
+
+  assert {
+    condition     = length(aws_lambda_event_source_mapping.stream) == 0
+    error_message = "Stream source mapping was created unexpectedly"
+  }
+}
diff --git a/variables.tf b/variables.tf
@@ -13,11 +13,44 @@ variable "policies" {
   default     = []
 }
 
+variable "trigger" {
+  description = "Object to define trigger of the Lambda function."
+  type = object({
+    queue_arn       = optional(string, null)
+    stream_arn      = optional(string, null)
+    batch_size      = optional(number, 1)
+    max_concurrency = optional(number, 1000)
+    max_retries     = optional(number, -1)
+    filter          = optional(string, null)
+  })
+  default = null
+  validation {
+    condition = var.trigger == null || (
+    try(var.trigger["queue_arn"], null) != null || try(var.trigger["stream_arn"], null) != null)
+    error_message = "Either 'queue_arn' or 'stream_arn' must be defined"
+  }
+  validation {
+    condition = var.trigger == null || (
+    try(var.trigger["queue_arn"], null) == null || try(var.trigger["stream_arn"], null) == null)
+    error_message = "Values 'queue_arn' and 'stream_arn' can not be both defined"
+  }
+  validation {
+    condition = !(
+    try(var.trigger["queue_arn"], null) == null && try(var.trigger["filter"], null) != null)
+    error_message = "Value 'queue_arn' must be defined if 'filter' is defined"
+  }
+  validation {
+    condition = try(var.trigger["max_concurrency"], 1000) <= 1000 && (
+    try(var.trigger["max_concurrency"], 2) >= 2)
+    error_message = "Value 'max_concurrency' must be between '2' and '1000'"
+  }
+}
+
 variable "vpc_config" {
   description = "Object to define the subnets and security groups for the Lambda function."
   type = object({
     subnets         = list(string)
-    security_groups = list(string)
+    security_groups = optional(list(string), [])
   })
   default = null
   validation {
