feat: add SQS queue support for failed events storage

README.md

@@ -128,6 +128,7 @@ For complete usage examples demonstrating different configuration scenarios, see
 | dd_max_workers                    | Max concurrent workers                                 | `string` | `null`  |
 | dd_log_level                      | Log level                                              | `string` | `null`  |
 | dd_store_failed_events            | Store failed events in S3                              | `bool`   | `null`  |
+| dd_sqs_queue_url                  | SQS queue URL for failed event storage (takes priority over S3 when set, auto-enables `dd_store_failed_events`) | `string` | `null`  |
 | dd_schedule_retry_failed_events   | Periodically retry failed events (via AWS EventBridge) | `bool`   | `null`  |
 | dd_schedule_retry_interval        | Retry interval in hours for failed events              | `number` | `6`     |
 | dd_forwarder_bucket_name          | Custom S3 bucket name                                  | `string` | `null`  |

data.tf

@@ -24,7 +24,14 @@ locals {
   )
 
   # Determine if we need to create an S3 bucket for caching and failed events storage
-  create_s3_bucket = (coalesce(var.dd_fetch_log_group_tags, false) || coalesce(var.dd_fetch_lambda_tags, false) || coalesce(var.dd_fetch_s3_tags, false) || coalesce(var.dd_store_failed_events, false)) && var.dd_forwarder_existing_bucket_name == null
+  create_s3_bucket = (coalesce(var.dd_fetch_log_group_tags, false) || coalesce(var.dd_fetch_lambda_tags, false) || coalesce(var.dd_fetch_s3_tags, false) || (coalesce(var.dd_store_failed_events, false) && var.dd_sqs_queue_url == null)) && var.dd_forwarder_existing_bucket_name == null
+
+  # SQS queue ARN derived from URL for IAM policy
+  # URL format: https://sqs.{region}.amazonaws.com/{account_id}/{queue_name}
+  sqs_queue_arn = var.dd_sqs_queue_url != null ? "arn:${data.aws_partition.current.partition}:sqs:${regex("https://sqs\\.([a-z0-9-]+)\\.amazonaws\\.com", var.dd_sqs_queue_url)[0]}:${split("/", var.dd_sqs_queue_url)[3]}:${split("/", var.dd_sqs_queue_url)[4]}" : null
+
+  # Whether failed events storage is enabled (via S3 or SQS)
+  store_failed_events_enabled = var.dd_sqs_queue_url != null || (coalesce(var.dd_store_failed_events, false) && (local.create_s3_bucket || var.dd_forwarder_existing_bucket_name != null))
 
   # Account ID varies by partition
   dd_account_id = data.aws_partition.current.partition == "aws-us-gov" ? "002406178527" : "464622532012"

main.tf

@@ -24,6 +24,7 @@ module "iam" {
   dd_fetch_s3_tags                  = var.dd_fetch_s3_tags
   dd_use_vpc                        = var.dd_use_vpc
   additional_target_lambda_arns     = var.additional_target_lambda_arns != null ? split(",", var.additional_target_lambda_arns) : []
+  sqs_queue_arn                     = local.sqs_queue_arn
 }
 
 # Secrets Manager secret for Datadog API key
@@ -129,7 +130,7 @@ resource "aws_s3_bucket_public_access_block" "forwarder_bucket_pab" {
 }
 
 resource "aws_s3_bucket_logging" "forwarder_bucket_logging" {
-  count = var.dd_forwarder_buckets_access_logs_target != null ? 1 : 0
+  count = var.dd_forwarder_buckets_access_logs_target != null && local.create_s3_bucket ? 1 : 0
 
   region = local.region
 
@@ -253,7 +254,8 @@ resource "aws_lambda_function" "forwarder" {
         DD_NO_SSL                       = var.dd_no_ssl
         DD_URL                          = var.dd_url
         DD_PORT                         = var.dd_port
-        DD_STORE_FAILED_EVENTS          = coalesce(var.dd_store_failed_events, false) && (local.create_s3_bucket || var.dd_forwarder_existing_bucket_name != null) ? "true" : null
+        DD_STORE_FAILED_EVENTS          = local.store_failed_events_enabled ? "true" : null
+        DD_SQS_QUEUE_URL                = var.dd_sqs_queue_url
         REDACT_IP                       = var.redact_ip != null ? tostring(var.redact_ip) : null
         REDACT_EMAIL                    = var.redact_email != null ? tostring(var.redact_email) : null
         DD_SCRUBBING_RULE               = var.dd_scrubbing_rule
@@ -362,7 +364,7 @@ resource "aws_cloudwatch_log_group" "forwarder_log_group" {
 # Scheduled retry
 
 resource "aws_iam_role" "scheduled_retry" {
-  count = coalesce(var.dd_store_failed_events, false) && coalesce(var.dd_schedule_retry_failed_events, false) ? 1 : 0
+  count = local.store_failed_events_enabled && coalesce(var.dd_schedule_retry_failed_events, false) ? 1 : 0
 
   name = "${var.function_name}-${local.region}-retry"
 
@@ -385,7 +387,7 @@ resource "aws_iam_role" "scheduled_retry" {
 }
 
 resource "aws_iam_role_policy" "scheduled_retry" {
-  count = coalesce(var.dd_store_failed_events, false) && coalesce(var.dd_schedule_retry_failed_events, false) ? 1 : 0
+  count = local.store_failed_events_enabled && coalesce(var.dd_schedule_retry_failed_events, false) ? 1 : 0
 
   name = "${var.function_name}-${local.region}-retry-policy"
   role = aws_iam_role.scheduled_retry[0].id
@@ -405,7 +407,7 @@ resource "aws_iam_role_policy" "scheduled_retry" {
 }
 
 resource "aws_scheduler_schedule" "scheduled_retry" {
-  count = coalesce(var.dd_store_failed_events, false) && coalesce(var.dd_schedule_retry_failed_events, false) ? 1 : 0
+  count = local.store_failed_events_enabled && coalesce(var.dd_schedule_retry_failed_events, false) ? 1 : 0
 
   name                = "${var.function_name}-${local.region}-retry"
   description         = "Retry the failed events from the Datadog Lambda Forwarder ${var.function_name}"

modules/iam/main.tf

@@ -69,6 +69,20 @@ resource "aws_iam_role_policy" "forwarder_policy" {
         }
       ] : [],
 
+      # SQS permissions for failed event storage queue
+      var.sqs_queue_arn != null ? [
+        {
+          Effect = "Allow"
+          Action = [
+            "sqs:SendMessage",
+            "sqs:ReceiveMessage",
+            "sqs:DeleteMessage",
+            "sqs:ChangeMessageVisibility"
+          ]
+          Resource = var.sqs_queue_arn
+        }
+      ] : [],
+
       # S3 read access for logs
       [
         {

modules/iam/variables.tf

@@ -106,6 +106,12 @@ variable "region" {
   description = "AWS region for resource naming"
 }
 
+variable "sqs_queue_arn" {
+  type        = string
+  default     = null
+  description = "ARN of the SQS queue for failed event storage"
+}
+
 variable "account_id" {
   type        = string
   description = "AWS account ID"

outputs.tf

@@ -21,6 +21,7 @@ output "datadog_forwarder_role_name" {
 output "dd_api_key_secret_arn" {
   description = "ARN of SecretsManager Secret with Datadog API Key (only set if created by this module)"
   value       = local.should_create_secret ? aws_secretsmanager_secret.dd_api_key_secret[0].arn : null
+  sensitive   = true
 }
 
 output "forwarder_bucket_name" {

tests/s3_log_bucket_arns.tftest.hcl

@@ -53,11 +53,11 @@ run "custom_s3_log_access_restricts_buckets" {
   }
 
   variables {
-    function_name         = "TestForwarder"
-    iam_role_path         = "/"
-    partition             = "aws"
-    region                = "us-east-1"
-    account_id            = "123456789012"
+    function_name = "TestForwarder"
+    iam_role_path = "/"
+    partition     = "aws"
+    region        = "us-east-1"
+    account_id    = "123456789012"
     dd_s3_log_bucket_arns = [
       "arn:aws:s3:::my-log-bucket/*",
       "arn:aws:s3:::my-other-bucket/logs/*",

tests/sqs_failed_events.tftest.hcl

@@ -0,0 +1,233 @@
+# Test SQS queue support for failed event storage
+
+mock_provider "aws" {
+  mock_data "aws_caller_identity" {
+    defaults = {
+      account_id = "123456789012"
+    }
+  }
+
+  mock_data "aws_region" {
+    defaults = {
+      region = "us-east-1"
+    }
+  }
+
+  mock_data "aws_partition" {
+    defaults = {
+      partition = "aws"
+    }
+  }
+}
+
+# Test: SQS URL auto-enables DD_STORE_FAILED_EVENTS and skips S3 bucket creation
+run "sqs_auto_enables_store_failed_events" {
+  command = plan
+
+  variables {
+    dd_api_key       = "test-api-key-value"
+    dd_site          = "datadoghq.com"
+    dd_sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+  }
+
+  # No S3 bucket should be created when only SQS is configured
+  assert {
+    condition     = length(aws_s3_bucket.forwarder_bucket) == 0
+    error_message = "S3 bucket should not be created when dd_sqs_queue_url is set and no tag fetching is enabled"
+  }
+
+  # DD_STORE_FAILED_EVENTS should be auto-enabled
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_STORE_FAILED_EVENTS == "true"
+    error_message = "DD_STORE_FAILED_EVENTS should be automatically set to true when dd_sqs_queue_url is provided"
+  }
+
+  # DD_SQS_QUEUE_URL should be set
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_SQS_QUEUE_URL == "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    error_message = "DD_SQS_QUEUE_URL should be set to the provided SQS queue URL"
+  }
+
+  # IAM policy should include SQS permissions
+  assert {
+    condition     = length(module.iam) == 1
+    error_message = "IAM module should be created"
+  }
+}
+
+# Test: SQS with tag fetching still creates S3 bucket for caching
+run "sqs_with_tag_fetching_creates_s3" {
+  command = plan
+
+  variables {
+    dd_api_key           = "test-api-key-value"
+    dd_site              = "datadoghq.com"
+    dd_sqs_queue_url     = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    dd_fetch_lambda_tags = true
+  }
+
+  # S3 bucket should be created for tag caching
+  assert {
+    condition     = length(aws_s3_bucket.forwarder_bucket) == 1
+    error_message = "S3 bucket should be created when dd_fetch_lambda_tags is enabled, even with SQS configured"
+  }
+
+  # DD_STORE_FAILED_EVENTS should still be enabled
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_STORE_FAILED_EVENTS == "true"
+    error_message = "DD_STORE_FAILED_EVENTS should be true when dd_sqs_queue_url is provided"
+  }
+
+  # DD_SQS_QUEUE_URL should be set
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_SQS_QUEUE_URL == "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    error_message = "DD_SQS_QUEUE_URL should be set"
+  }
+}
+
+# Test: SQS with existing IAM role - no IAM module, no S3 bucket
+run "sqs_with_existing_iam_role" {
+  command = plan
+
+  variables {
+    dd_site                       = "datadoghq.com"
+    dd_sqs_queue_url              = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    existing_iam_role_arn         = "arn:aws:iam::123456789012:role/existing-datadog-role"
+    dd_api_key_ssm_parameter_name = "/datadog/api-key"
+  }
+
+  # No IAM module
+  assert {
+    condition     = length(module.iam) == 0
+    error_message = "IAM module should not be created when existing_iam_role_arn is provided"
+  }
+
+  # No S3 bucket
+  assert {
+    condition     = length(aws_s3_bucket.forwarder_bucket) == 0
+    error_message = "S3 bucket should not be created when only SQS is configured"
+  }
+
+  # DD_STORE_FAILED_EVENTS should be enabled
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_STORE_FAILED_EVENTS == "true"
+    error_message = "DD_STORE_FAILED_EVENTS should be true when dd_sqs_queue_url is provided"
+  }
+}
+
+# Test: SQS with scheduled retry creates scheduler resources
+run "sqs_with_scheduled_retry" {
+  command = plan
+
+  variables {
+    dd_api_key                      = "test-api-key-value"
+    dd_site                         = "datadoghq.com"
+    dd_sqs_queue_url                = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    dd_schedule_retry_failed_events = true
+  }
+
+  # Scheduler resources should be created
+  assert {
+    condition     = length(aws_scheduler_schedule.scheduled_retry) == 1
+    error_message = "Scheduler should be created when SQS is configured with dd_schedule_retry_failed_events"
+  }
+
+  assert {
+    condition     = length(aws_iam_role.scheduled_retry) == 1
+    error_message = "Scheduler IAM role should be created when SQS is configured with dd_schedule_retry_failed_events"
+  }
+}
+
+# Test: Invalid SQS URL validation
+run "invalid_sqs_url_fails_validation" {
+  command = plan
+
+  variables {
+    dd_api_key       = "test-api-key-value"
+    dd_site          = "datadoghq.com"
+    dd_sqs_queue_url = "not-a-valid-url"
+  }
+
+  expect_failures = [
+    var.dd_sqs_queue_url,
+  ]
+}
+
+# Test: S3 fallback unchanged - dd_store_failed_events without SQS still creates S3 bucket
+run "s3_fallback_unchanged" {
+  command = plan
+
+  variables {
+    dd_api_key             = "test-api-key-value"
+    dd_site                = "datadoghq.com"
+    dd_store_failed_events = true
+  }
+
+  # S3 bucket should be created for failed events
+  assert {
+    condition     = length(aws_s3_bucket.forwarder_bucket) == 1
+    error_message = "S3 bucket should be created when dd_store_failed_events is true and no SQS queue is configured"
+  }
+
+  # DD_STORE_FAILED_EVENTS should be enabled
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_STORE_FAILED_EVENTS == "true"
+    error_message = "DD_STORE_FAILED_EVENTS should be true when dd_store_failed_events is enabled"
+  }
+
+  # DD_SQS_QUEUE_URL should not be set
+  assert {
+    condition     = !contains(keys(aws_lambda_function.forwarder.environment[0].variables), "DD_SQS_QUEUE_URL") || aws_lambda_function.forwarder.environment[0].variables.DD_SQS_QUEUE_URL == null
+    error_message = "DD_SQS_QUEUE_URL should not be set when no SQS queue is configured"
+  }
+}
+
+# Test: SQS with old layer version fails validation
+run "sqs_with_old_layer_version_fails" {
+  command = plan
+
+  variables {
+    dd_api_key       = "test-api-key-value"
+    dd_site          = "datadoghq.com"
+    dd_sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    layer_version    = "96"
+  }
+
+  expect_failures = [
+    var.dd_sqs_queue_url,
+  ]
+}
+
+# Test: SQS with layer version 97 succeeds
+run "sqs_with_layer_version_97" {
+  command = plan
+
+  variables {
+    dd_api_key       = "test-api-key-value"
+    dd_site          = "datadoghq.com"
+    dd_sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    layer_version    = "97"
+  }
+
+  assert {
+    condition     = aws_lambda_function.forwarder.environment[0].variables.DD_SQS_QUEUE_URL == "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+    error_message = "DD_SQS_QUEUE_URL should be set with layer version 97"
+  }
+}
+
+# Test: IAM SQS permissions are included when dd_sqs_queue_url is set
+run "iam_sqs_permissions" {
+  command = plan
+
+  variables {
+    dd_api_key       = "test-api-key-value"
+    dd_site          = "datadoghq.com"
+    dd_sqs_queue_url = "https://sqs.us-east-1.amazonaws.com/123456789012/my-failed-events-queue"
+  }
+
+  # IAM module should be created with SQS permissions
+  assert {
+    condition     = length(module.iam) == 1
+    error_message = "IAM module should be created"
+  }
+}

variables.tf

@@ -450,6 +450,22 @@ variable "dd_store_failed_events" {
   description = "Set to true to enable the forwarder to store events that failed to send to Datadog."
 }
 
+variable "dd_sqs_queue_url" {
+  type        = string
+  default     = null
+  description = "URL of an existing SQS queue for failed event storage. When set, the forwarder uses SQS instead of S3 for retry storage, and DD_STORE_FAILED_EVENTS is automatically enabled. The queue must already exist. Requires forwarder layer version >= 97. Format: https://sqs.{region}.amazonaws.com/{account_id}/{queue_name}"
+
+  validation {
+    condition     = var.dd_sqs_queue_url == null || can(regex("^https://sqs\\.[a-z0-9-]+\\.amazonaws\\.com[a-z.]*/[0-9]{12}/[a-zA-Z0-9_.-]+$", var.dd_sqs_queue_url))
+    error_message = "dd_sqs_queue_url must be a valid SQS queue URL (e.g. https://sqs.us-east-1.amazonaws.com/123456789012/my-queue)."
+  }
+
+  validation {
+    condition     = var.dd_sqs_queue_url == null || var.layer_version == "latest" || (can(tonumber(var.layer_version)) && tonumber(var.layer_version) >= 97)
+    error_message = "dd_sqs_queue_url requires forwarder layer version >= 97 (forwarder 5.3.0+). Set layer_version = \"latest\" or a version >= 97."
+  }
+}
+
 variable "dd_schedule_retry_failed_events" {
   type        = bool
   default     = null