feat(forwarder): relax filedrop requirement (#144)

Allow forwarder to be set up with an HTTP endpoint. This commit
introduces a `uri` field in `destination`, and makes all fields within
the object optional.

This commit deviates from the cloudformation equivalent parameters. In
cloudformation, we have a single `DestinationUri` parameter which
accepts both `s3` and `https` schemes. In Terraform, we prioritize
ergonomics for passing in an `observe_filedrop` object. As a result,
`uri` is reserved solely for the case where we pass in an `https` URL.

examples/forwarder-kms/README.md

@@ -10,7 +10,7 @@ decrypt objects.
 
 | Name | Version |
 |------|---------|
-| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | >= 1.2 |
+| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | >= 1.3 |
 | <a name="requirement_aws"></a> [aws](#requirement\_aws) | >= 5.0 |
 | <a name="requirement_observe"></a> [observe](#requirement\_observe) | ~> 0.14 |
 | <a name="requirement_random"></a> [random](#requirement\_random) | >= 3.0.0 |

examples/forwarder-kms/versions.tf

@@ -1,5 +1,5 @@
 terraform {
-  required_version = ">= 1.2"
+  required_version = ">= 1.3"
 
   required_providers {
     aws = {

modules/forwarder/README.md

@@ -58,7 +58,7 @@ module "forwarder" {
 
 | Name | Version |
 |------|---------|
-| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | >= 1.2 |
+| <a name="requirement_terraform"></a> [terraform](#requirement\_terraform) | >= 1.3 |
 | <a name="requirement_aws"></a> [aws](#requirement\_aws) | >= 5.0 |
 | <a name="requirement_random"></a> [random](#requirement\_random) | >= 3.0.0 |
 
@@ -101,7 +101,7 @@ No modules.
 |------|-------------|------|---------|:--------:|
 | <a name="input_content_type_overrides"></a> [content\_type\_overrides](#input\_content\_type\_overrides) | A list of key value pairs. The key is a regular expression which is<br>applied to the S3 source (<bucket>/<key>) of forwarded files. The value<br>is the content type to set for matching files. For example,<br>`\.json$=application/x-ndjson` would forward all files ending in `.json`<br>as newline delimited JSON | <pre>list(object({<br>    pattern      = string<br>    content_type = string<br>  }))</pre> | `[]` | no |
 | <a name="input_debug_endpoint"></a> [debug\_endpoint](#input\_debug\_endpoint) | Endpoint to send debugging telemetry to. Sets the OTEL\_EXPORTER\_OTLP\_ENDPOINT environment variable for the lambda function. | `string` | `""` | no |
-| <a name="input_destination"></a> [destination](#input\_destination) | Destination filedrop | <pre>object({<br>    arn    = string<br>    bucket = string<br>    prefix = string<br>  })</pre> | n/a | yes |
+| <a name="input_destination"></a> [destination](#input\_destination) | Destination filedrop | <pre>object({<br>    arn    = optional(string, "")<br>    bucket = optional(string, "")<br>    prefix = optional(string, "")<br>    # exclusively for backward compatible HTTP endpoint<br>    uri = optional(string, "")<br>  })</pre> | n/a | yes |
 | <a name="input_lambda_env_vars"></a> [lambda\_env\_vars](#input\_lambda\_env\_vars) | Environment variables to be passed into lambda. | `map(string)` | `{}` | no |
 | <a name="input_lambda_memory_size"></a> [lambda\_memory\_size](#input\_lambda\_memory\_size) | Memory size for lambda function. | `number` | `128` | no |
 | <a name="input_lambda_timeout"></a> [lambda\_timeout](#input\_lambda\_timeout) | Timeout in seconds for lambda function. | `number` | `300` | no |

modules/forwarder/iam.tf

@@ -6,8 +6,8 @@ resource "aws_iam_role" "this" {
     for_each = {
       for k, v in {
         logging = data.aws_iam_policy_document.logging.json
-        writer  = data.aws_iam_policy_document.writer.json
         queue   = data.aws_iam_policy_document.queue.json
+        writer  = var.destination.bucket != "" ? data.aws_iam_policy_document.writer.json : null
         reader  = length(var.source_bucket_names) > 0 ? data.aws_iam_policy_document.reader.json : null
         kms     = length(var.source_kms_key_arns) > 0 ? data.aws_iam_policy_document.kms.json : null
       } : k => v if v != null
@@ -53,7 +53,7 @@ data "aws_iam_policy_document" "writer" {
       "s3:PutObjectTagging",
     ]
 
-    resources = var.destination.arn != "" ? ["*"] : ["arn:${data.aws_partition.current.id}:s3:::${var.destination.bucket}"]
+    resources = var.destination.arn != "" ? ["*"] : ["arn:${data.aws_partition.current.id}:s3:::${var.destination.bucket}/${var.destination.prefix}*"]
 
     dynamic "condition" {
       for_each = var.destination.arn != "" ? [1] : []

modules/forwarder/lambda.tf

@@ -11,7 +11,7 @@ resource "aws_lambda_function" "this" {
 
   environment {
     variables = merge({
-      DESTINATION_URI             = "s3://${var.destination.bucket}/${var.destination.prefix}"
+      DESTINATION_URI             = local.destination_uri
       MAX_FILE_SIZE               = var.max_file_size
       CONTENT_TYPE_OVERRIDES      = join(",", [for o in var.content_type_overrides : "${o["pattern"]}=${o["content_type"]}"])
       SOURCE_BUCKET_NAMES         = join(",", var.source_bucket_names)

modules/forwarder/main.tf

@@ -1,6 +1,7 @@
 locals {
-  s3_uri        = one([for item in csvdecode(file("${path.module}/uris.csv")) : item["code_uri"] if item["region"] == data.aws_region.current.name])
-  parsed_s3_uri = regex("s3://(?P<bucket>[^/]+)/(?P<key>.+)", local.s3_uri)
+  s3_uri          = one([for item in csvdecode(file("${path.module}/uris.csv")) : item["code_uri"] if item["region"] == data.aws_region.current.name])
+  parsed_s3_uri   = regex("s3://(?P<bucket>[^/]+)/(?P<key>.+)", local.s3_uri)
+  destination_uri = var.destination.uri != "" ? var.destination.uri : "s3://${var.destination.bucket}/${var.destination.prefix}"
 }
 
 data "aws_caller_identity" "current" {}

modules/forwarder/tests/forwarder.tftest.hcl

@@ -23,9 +23,7 @@ run "install_forwarder" {
   variables {
     name = run.setup.id
     destination = {
-      arn    = ""
       bucket = run.create_bucket.id
-      prefix = ""
     }
     source_bucket_names = [for source in ["sns", "sqs", "eventbridge"] : "${run.setup.short}-${source}"]
     source_topic_arns   = ["arn:aws:sns:${run.setup.region}:${run.setup.account_id}:*"]
@@ -37,3 +35,14 @@ run "install_forwarder" {
     ]
   }
 }
+
+run "update_forwarder" {
+  variables {
+    name = run.setup.id
+    destination = {
+      uri = "https://localhost:8080"
+    }
+    source_bucket_names = [for source in ["sns", "sqs", "eventbridge"] : "${run.setup.short}-${source}"]
+    source_topic_arns   = ["arn:aws:sns:${run.setup.region}:${run.setup.account_id}:*"]
+  }
+}

modules/forwarder/variables.tf

@@ -14,12 +14,34 @@ variable "name" {
 
 variable "destination" {
   type = object({
-    arn    = string
-    bucket = string
-    prefix = string
+    arn    = optional(string, "")
+    bucket = optional(string, "")
+    prefix = optional(string, "")
+    # exclusively for backward compatible HTTP endpoint
+    uri = optional(string, "")
   })
   nullable    = false
   description = "Destination filedrop"
+
+  validation {
+    condition     = !(var.destination.uri != "" && "${var.destination.arn}${var.destination.bucket}${var.destination.prefix}" != "")
+    error_message = "URI is mutually exclusive with S3 attributes"
+  }
+
+  validation {
+    condition     = !(var.destination.bucket == "" && var.destination.prefix != "")
+    error_message = "Prefix cannot be set without bucket"
+  }
+
+  validation {
+    condition     = !(var.destination.bucket == "" && var.destination.arn != "")
+    error_message = "Access point ARN cannot be set without bucket"
+  }
+
+  validation {
+    condition     = var.destination.uri == "" || can(regex("^https://.*", var.destination.uri))
+    error_message = "URI must have https scheme"
+  }
 }
 
 variable "source_bucket_names" {

modules/forwarder/versions.tf

@@ -1,5 +1,5 @@
 terraform {
-  required_version = ">= 1.2"
+  required_version = ">= 1.3"
 
   required_providers {
     aws = {

modules/stack/README.md

@@ -100,7 +100,7 @@ module "collection_stack" {
 | <a name="input_config"></a> [config](#input\_config) | Variables for AWS Config collection. | <pre>object({<br>    include_resource_types        = list(string)<br>    exclude_resource_types        = optional(list(string))<br>    delivery_frequency            = optional(string)<br>    include_global_resource_types = optional(bool)<br>  })</pre> | `null` | no |
 | <a name="input_configsubscription"></a> [configsubscription](#input\_configsubscription) | Variables for AWS Config subscription. | <pre>object({<br>    delivery_bucket_name = string<br>  })</pre> | `null` | no |
 | <a name="input_debug_endpoint"></a> [debug\_endpoint](#input\_debug\_endpoint) | Endpoint to send debugging telemetry to. Sets OTEL\_EXPORTER\_OTLP\_ENDPOINT environment variable for supported lambda functions. | `string` | `null` | no |
-| <a name="input_destination"></a> [destination](#input\_destination) | Destination filedrop | <pre>object({<br>    arn    = string<br>    bucket = string<br>    prefix = string<br>  })</pre> | n/a | yes |
+| <a name="input_destination"></a> [destination](#input\_destination) | Destination filedrop | <pre>object({<br>    arn    = optional(string, "")<br>    bucket = optional(string, "")<br>    prefix = optional(string, "")<br>    # exclusively for backward compatible HTTP endpoint<br>    uri = optional(string, "")<br>  })</pre> | n/a | yes |
 | <a name="input_forwarder"></a> [forwarder](#input\_forwarder) | Variables for forwarder module. | <pre>object({<br>    source_bucket_names                      = optional(list(string), [])<br>    source_topic_arns                        = optional(list(string), [])<br>    content_type_overrides                   = optional(list(object({ pattern = string, content_type = string })), [])<br>    max_file_size                            = optional(number)<br>    lambda_memory_size                       = optional(number)<br>    lambda_timeout                           = optional(number)<br>    lambda_env_vars                          = optional(map(string))<br>    retention_in_days                        = optional(number)<br>    queue_max_receive_count                  = optional(number)<br>    queue_delay_seconds                      = optional(number)<br>    queue_message_retention_seconds          = optional(number)<br>    queue_batch_size                         = optional(number)<br>    queue_maximum_batching_window_in_seconds = optional(number)<br>  })</pre> | `{}` | no |
 | <a name="input_logwriter"></a> [logwriter](#input\_logwriter) | Variables for AWS CloudWatch Logs collection. | <pre>object({<br>    log_group_name_patterns = optional(list(string))<br>    log_group_name_prefixes = optional(list(string))<br>    buffering_interval      = optional(number)<br>    buffering_size          = optional(number)<br>    filter_name             = optional(string)<br>    filter_pattern          = optional(string)<br>    num_workers             = optional(number)<br>    discovery_rate          = optional(string, "24 hours")<br>    lambda_memory_size      = optional(number)<br>    lambda_timeout          = optional(number)<br>  })</pre> | `null` | no |
 | <a name="input_metricstream"></a> [metricstream](#input\_metricstream) | Variables for AWS CloudWatch Metrics Stream collection. | <pre>object({<br>    include_filters    = optional(list(object({ namespace = string, metric_names = optional(list(string)) })))<br>    exclude_filters    = optional(list(object({ namespace = string, metric_names = optional(list(string)) })))<br>    buffering_interval = optional(number)<br>    buffering_size     = optional(number)<br>  })</pre> | `null` | no |

modules/stack/variables.tf

@@ -14,9 +14,11 @@ variable "name" {
 
 variable "destination" {
   type = object({
-    arn    = string
-    bucket = string
-    prefix = string
+    arn    = optional(string, "")
+    bucket = optional(string, "")
+    prefix = optional(string, "")
+    # exclusively for backward compatible HTTP endpoint
+    uri = optional(string, "")
   })
   nullable    = false
   description = "Destination filedrop"