feat: added support for on_failure destinations to event source mappings

README.md

@@ -116,7 +116,8 @@ module "lambda" {
 ### with event source mappings
 
 [Event Source Mappings](https://www.terraform.io/docs/providers/aws/r/lambda_event_source_mapping.html) to trigger your Lambda function by DynamoDb,
-Kinesis and SQS can be declared inline. The module will add the required IAM permissions depending on the event source type to the function role automatically.
+Kinesis and SQS can be declared inline. The module will add the required read-only IAM permissions depending on the event source type to
+the function role automatically. In addition, permissions to send discarded batches to SNS or SQS will be added automatically, if `destination_arn_on_failure` is configured.
 
 see [examples](examples/with-event-source-mappings) for details
 
@@ -133,6 +134,11 @@ module "lambda" {
       batch_size        = 50
       starting_position = "LATEST"
 
+      // optionally configure a SNS or SQS destination for discarded batches, required IAM
+      // permissions will be added automatically by this module,
+      // see https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html
+      destination_arn_on_failure = aws_sqs_queue.errors.arn
+
       // optionally overwrite function_name in case an alias should be used in the
       // event source mapping, see https://docs.aws.amazon.com/lambda/latest/dg/configuration-aliases.html
       function_name = aws_lambda_alias.example.arn
@@ -329,7 +335,7 @@ MINOR, and PATCH versions on each release to indicate any incompatibilities.
 | <a name="input_description"></a> [description](#input\_description) | Description of what your Lambda Function does. | `string` | `""` | no |
 | <a name="input_environment"></a> [environment](#input\_environment) | Environment (e.g. env variables) configuration for the Lambda function enable you to dynamically pass settings to your function code and libraries | <pre>object({<br>    variables = map(string)<br>  })</pre> | `null` | no |
 | <a name="input_event"></a> [event](#input\_event) | (deprecated - use `cloudwatch_event_rules` [EventBridge/CloudWatch Events], `event_source_mappings` [DynamoDb, Kinesis, SQS] or `sns_subscriptions` [SNS] instead) Event source configuration which triggers the Lambda function. Supported events: cloudwatch-scheduled-event, dynamodb, kinesis, s3, sns, sqs | `map(string)` | `{}` | no |
-| <a name="input_event_source_mappings"></a> [event\_source\_mappings](#input\_event\_source\_mappings) | Creates event source mappings to allow the Lambda function to get events from Kinesis, DynamoDB and SQS. The IAM role of this Lambda function will be enhanced with necessary minimum permissions to get those events. | `map(any)` | `{}` | no |
+| <a name="input_event_source_mappings"></a> [event\_source\_mappings](#input\_event\_source\_mappings) | Creates event source mappings to allow the Lambda function to get events from Kinesis, DynamoDB and SQS. The IAM role of this Lambda function will be enhanced with necessary minimum permissions to get those events. | `any` | `{}` | no |
 | <a name="input_filename"></a> [filename](#input\_filename) | The path to the function's deployment package within the local filesystem. If defined, The s3\_-prefixed options and image\_uri cannot be used. | `string` | `null` | no |
 | <a name="input_function_name"></a> [function\_name](#input\_function\_name) | A unique name for your Lambda Function. | `string` | n/a | yes |
 | <a name="input_handler"></a> [handler](#input\_handler) | The function entrypoint in your code. | `string` | `""` | no |

docs/part1.md

@@ -116,7 +116,8 @@ module "lambda" {
 ### with event source mappings
 
 [Event Source Mappings](https://www.terraform.io/docs/providers/aws/r/lambda_event_source_mapping.html) to trigger your Lambda function by DynamoDb,
-Kinesis and SQS can be declared inline. The module will add the required IAM permissions depending on the event source type to the function role automatically.
+Kinesis and SQS can be declared inline. The module will add the required read-only IAM permissions depending on the event source type to
+the function role automatically. In addition, permissions to send discarded batches to SNS or SQS will be added automatically, if `destination_arn_on_failure` is configured.
 
 see [examples](examples/with-event-source-mappings) for details
 
@@ -133,6 +134,11 @@ module "lambda" {
       batch_size        = 50
       starting_position = "LATEST"
 
+      // optionally configure a SNS or SQS destination for discarded batches, required IAM
+      // permissions will be added automatically by this module,
+      // see https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html
+      destination_arn_on_failure = aws_sqs_queue.errors.arn
+
       // optionally overwrite function_name in case an alias should be used in the
       // event source mapping, see https://docs.aws.amazon.com/lambda/latest/dg/configuration-aliases.html
       function_name = aws_lambda_alias.example.arn

docs/part2.md

@@ -61,7 +61,7 @@
 | <a name="input_description"></a> [description](#input\_description) | Description of what your Lambda Function does. | `string` | `""` | no |
 | <a name="input_environment"></a> [environment](#input\_environment) | Environment (e.g. env variables) configuration for the Lambda function enable you to dynamically pass settings to your function code and libraries | <pre>object({<br>    variables = map(string)<br>  })</pre> | `null` | no |
 | <a name="input_event"></a> [event](#input\_event) | (deprecated - use `cloudwatch_event_rules` [EventBridge/CloudWatch Events], `event_source_mappings` [DynamoDb, Kinesis, SQS] or `sns_subscriptions` [SNS] instead) Event source configuration which triggers the Lambda function. Supported events: cloudwatch-scheduled-event, dynamodb, kinesis, s3, sns, sqs | `map(string)` | `{}` | no |
-| <a name="input_event_source_mappings"></a> [event\_source\_mappings](#input\_event\_source\_mappings) | Creates event source mappings to allow the Lambda function to get events from Kinesis, DynamoDB and SQS. The IAM role of this Lambda function will be enhanced with necessary minimum permissions to get those events. | `map(any)` | `{}` | no |
+| <a name="input_event_source_mappings"></a> [event\_source\_mappings](#input\_event\_source\_mappings) | Creates event source mappings to allow the Lambda function to get events from Kinesis, DynamoDB and SQS. The IAM role of this Lambda function will be enhanced with necessary minimum permissions to get those events. | `any` | `{}` | no |
 | <a name="input_filename"></a> [filename](#input\_filename) | The path to the function's deployment package within the local filesystem. If defined, The s3\_-prefixed options and image\_uri cannot be used. | `string` | `null` | no |
 | <a name="input_function_name"></a> [function\_name](#input\_function\_name) | A unique name for your Lambda Function. | `string` | n/a | yes |
 | <a name="input_handler"></a> [handler](#input\_handler) | The function entrypoint in your code. | `string` | `""` | no |

event_source_mappings.tf

@@ -1,11 +1,33 @@
 locals {
-  dynamodb_event_sources = [for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:dynamodb:.*", lookup(v, "event_source_arn", null))) > 0]
-  kinesis_event_sources  = [for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:kinesis:.*", lookup(v, "event_source_arn", null))) > 0]
-  sqs_event_sources      = [for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:sqs:.*", lookup(v, "event_source_arn", null))) > 0]
+  // compute all event source mappings for DynamoDb
+  dynamodb_event_sources = [
+    for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:dynamodb:.*", lookup(v, "event_source_arn", null))) > 0
+  ]
+
+  // compute all event source mappings for Kinesis
+  kinesis_event_sources = [
+    for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:kinesis:.*", lookup(v, "event_source_arn", null))) > 0
+  ]
+
+  // compute all event source mappings for SQS
+  sqs_event_sources = [
+    for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:sqs:.*", lookup(v, "event_source_arn", null))) > 0
+  ]
+
+  // compute SQS destination ARNs for discarded batches
+  on_failure_sqs_destination_arns = [
+    for k, v in var.event_source_mappings : lookup(v, "destination_arn_on_failure", "") if length(regexall(".*:sqs:.*", lookup(v, "destination_arn_on_failure", ""))) > 0
+  ]
+
+  // compute SNS destination ARNs for discarded batches
+  on_failure_sns_destination_arns = [
+    for k, v in var.event_source_mappings : lookup(v, "destination_arn_on_failure", "") if length(regexall(".*:sns:.*", lookup(v, "destination_arn_on_failure", ""))) > 0
+  ]
 }
 
 resource "aws_lambda_event_source_mapping" "event_source" {
-  for_each = var.event_source_mappings
+  for_each   = var.event_source_mappings
+  depends_on = [module.lambda]
 
   batch_size                         = lookup(each.value, "batch_size", null)
   bisect_batch_on_function_error     = lookup(each.value, "bisect_batch_on_function_error", null)
@@ -18,14 +40,23 @@ resource "aws_lambda_event_source_mapping" "event_source" {
   parallelization_factor             = lookup(each.value, "parallelization_factor", null)
   starting_position                  = lookup(each.value, "starting_position", length(regexall(".*:sqs:.*", lookup(each.value, "event_source_arn", null))) > 0 ? null : "TRIM_HORIZON")
   starting_position_timestamp        = lookup(each.value, "starting_position_timestamp", null)
+
+  dynamic "destination_config" {
+    for_each = lookup(each.value, "destination_arn_on_failure", null) != null ? [true] : []
+    content {
+      on_failure {
+        destination_arn = each.value["destination_arn_on_failure"]
+      }
+    }
+  }
 }
 
 // type specific minimal permissions for supported event_sources,
 // see https://github.com/awslabs/serverless-application-model/blob/develop/samtranslator/policy_templates_data/policy_templates.json
 data "aws_iam_policy_document" "event_sources" {
   count = length(var.event_source_mappings) > 0 ? 1 : 0
 
-  // SQS permissions
+  // SQS event source mapping permissions
   dynamic "statement" {
     for_each = length(local.sqs_event_sources) > 0 ? [true] : []
     content {
@@ -44,7 +75,7 @@ data "aws_iam_policy_document" "event_sources" {
     }
   }
 
-  // DynamoDb permissions
+  // DynamoDb event source mapping permissions
   dynamic "statement" {
     for_each = length(local.dynamodb_event_sources) > 0 ? [true] : []
     content {
@@ -61,7 +92,7 @@ data "aws_iam_policy_document" "event_sources" {
     }
   }
 
-  // Kinesis permissions
+  // Kinesis event source mapping permissions
   dynamic "statement" {
     for_each = length(local.kinesis_event_sources) > 0 ? [true] : []
     content {
@@ -93,6 +124,34 @@ data "aws_iam_policy_document" "event_sources" {
       ]
     }
   }
+
+  // SQS permission for on-failure destinations
+  dynamic "statement" {
+    for_each = length(local.on_failure_sqs_destination_arns) > 0 ? [true] : []
+    content {
+      actions = [
+        "sqs:SendMessage"
+      ]
+
+      resources = [
+        for arn in local.on_failure_sqs_destination_arns : arn
+      ]
+    }
+  }
+
+  // SNS permission for on-failure destinations
+  dynamic "statement" {
+    for_each = length(local.on_failure_sns_destination_arns) > 0 ? [true] : []
+    content {
+      actions = [
+        "sns:Publish"
+      ]
+
+      resources = [
+        for arn in local.on_failure_sns_destination_arns : arn
+      ]
+    }
+  }
 }
 
 resource "aws_iam_policy" "event_sources" {

examples/with-event-source-mappings/dynamodb-with-alias/main.tf

@@ -54,11 +54,16 @@ module "lambda" {
 
   event_source_mappings = {
     table_1 = {
-      // optionally overwrite arguments like 'batch_size'
-      // from https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping
-      batch_size        = 50
-      event_source_arn  = aws_dynamodb_table.table_1.stream_arn
-      starting_position = "LATEST"
+      event_source_arn = aws_dynamodb_table.table_1.stream_arn
+
+      // optionally overwrite arguments from https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping
+      batch_size             = 50
+      maximum_retry_attempts = 3
+
+      // optionally configure a SNS or SQS destination for discarded batches, required IAM
+      // permissions will be added automatically by this module,
+      // see https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventsourcemapping.html
+      destination_arn_on_failure = aws_sqs_queue.errors.arn
 
       // optionally overwrite function_name in case an alias should be used in the
       // event source mapping, see https://docs.aws.amazon.com/lambda/latest/dg/configuration-aliases.html
@@ -71,3 +76,7 @@ module "lambda" {
     }
   }
 }
+
+resource "aws_sqs_queue" "errors" {
+  name = "${module.lambda.function_name}-processing-errors"
+}

examples/with-event-source-mappings/dynamodb-with-alias/outputs.tf

@@ -10,7 +10,7 @@ output "arn" {
 
 output "event_source_arns" {
   description = "The Amazon Resource Names (ARNs) identifying the event sources."
-  value       = list(aws_dynamodb_table.table_1.stream_arn, aws_dynamodb_table.table_2.stream_arn)
+  value       = [aws_dynamodb_table.table_1.stream_arn, aws_dynamodb_table.table_2.stream_arn]
 }
 
 output "function_name" {

examples/with-event-source-mappings/kinesis/main.tf

@@ -30,19 +30,19 @@ module "lambda" {
 
   event_source_mappings = {
     stream_1 = {
-      // optionally overwrite arguments like 'batch_size'
-      // from https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping
-      batch_size       = 50
       event_source_arn = aws_kinesis_stream.stream_1.arn
 
+      // optionally overwrite arguments from https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping
+      batch_size        = 50
+      starting_position = "LATEST" // optionally overwrite default 'starting_position'
+
       // optionally overwrite function_name in case an alias should be used in the
       // event source mapping, see https://docs.aws.amazon.com/lambda/latest/dg/configuration-aliases.html
       // function_name    = aws_lambda_alias.example.arn
     }
 
     stream_2 = {
-      event_source_arn  = aws_kinesis_stream.stream_2.arn
-      starting_position = "LATEST" // optionally overwrite default 'starting_position'
+      event_source_arn = aws_kinesis_stream.stream_2.arn
     }
   }
 }

examples/with-event-source-mappings/kinesis/outputs.tf

@@ -5,7 +5,7 @@ output "arn" {
 
 output "event_source_arns" {
   description = "The Amazon Resource Names (ARNs) identifying the event sources."
-  value       = list(aws_kinesis_stream.stream_1.arn, aws_kinesis_stream.stream_2.arn)
+  value       = [aws_kinesis_stream.stream_1.arn, aws_kinesis_stream.stream_2.arn]
 }
 
 output "function_name" {

examples/with-event-source-mappings/sqs/main.tf

@@ -28,11 +28,11 @@ module "lambda" {
 
   event_source_mappings = {
     queue_1 = {
-      // optionally overwrite arguments like 'batch_size'
-      // from https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping
-      batch_size       = 5
       event_source_arn = aws_sqs_queue.queue_1.arn
 
+      // optionally overwrite arguments from https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/lambda_event_source_mapping
+      batch_size = 5
+
       // optionally overwrite function_name in case an alias should be used in the
       // event source mapping, see https://docs.aws.amazon.com/lambda/latest/dg/configuration-aliases.html
       // function_name    = aws_lambda_alias.example.arn

examples/with-event-source-mappings/sqs/outputs.tf

@@ -5,7 +5,7 @@ output "arn" {
 
 output "event_source_arns" {
   description = "The Amazon Resource Names (ARNs) identifying the event sources."
-  value       = list(aws_sqs_queue.queue_1.arn, aws_sqs_queue.queue_2.arn)
+  value       = [aws_sqs_queue.queue_1.arn, aws_sqs_queue.queue_2.arn]
 }
 
 output "function_name" {