feat(kinesis): support dedicated-throughput consumer with enhanced fan-out.

Author: moritzzimmer

README.md

@@ -92,7 +92,7 @@ module "lambda" {
 
       // optionally overwrite `cloudwatch_event_target_arn` in case an alias should be used for the event rule
       cloudwatch_event_target_arn = aws_lambda_alias.example.arn
-      
+
       // optionally add `cloudwatch_event_target_input` for event input
       cloudwatch_event_target_input = jsonencode({"key": "value"})
     }
@@ -114,10 +114,14 @@ module "lambda" {
 
 [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 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.
+the function role automatically (including support for [dedicated-throughput consumers](https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html#services-kinesis-configure) aka. enhanced fan-out).
+
+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
 
+#### DynamoDb
+
 ```hcl
 module "lambda" {
   // see above
@@ -155,6 +159,22 @@ module "lambda" {
 }
 ```
 
+#### Kinesis
+
+```hcl
+module "lambda" {
+  // see above
+
+  event_source_mappings = {
+    enhanced_fan_out = {
+      // To use a consumer (enhanced fan-out), specify the consumer's ARN instead of the stream's ARN, see https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html#services-kinesis-configure
+      event_source_arn = aws_kinesis_stream_consumer.this.arn
+    }
+  }
+}
+
+```
+
 ### with SNS subscriptions
 
 [SNS Topic Subscriptions](https://www.terraform.io/docs/providers/aws/r/sns_topic_subscription.html) to trigger your Lambda function by SNS can de declared inline.

docs/part1.md

@@ -92,7 +92,7 @@ module "lambda" {
 
       // optionally overwrite `cloudwatch_event_target_arn` in case an alias should be used for the event rule
       cloudwatch_event_target_arn = aws_lambda_alias.example.arn
-      
+
       // optionally add `cloudwatch_event_target_input` for event input
       cloudwatch_event_target_input = jsonencode({"key": "value"})
     }
@@ -114,10 +114,14 @@ module "lambda" {
 
 [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 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.
+the function role automatically (including support for [dedicated-throughput consumers](https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html#services-kinesis-configure) aka. enhanced fan-out).
+
+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
 
+#### DynamoDb
+
 ```hcl
 module "lambda" {
   // see above
@@ -155,6 +159,22 @@ module "lambda" {
 }
 ```
 
+#### Kinesis
+
+```hcl
+module "lambda" {
+  // see above
+
+  event_source_mappings = {
+    enhanced_fan_out = {
+      // To use a consumer (enhanced fan-out), specify the consumer's ARN instead of the stream's ARN, see https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html#services-kinesis-configure
+      event_source_arn = aws_kinesis_stream_consumer.this.arn
+    }
+  }
+}
+
+```
+
 ### with SNS subscriptions
 
 [SNS Topic Subscriptions](https://www.terraform.io/docs/providers/aws/r/sns_topic_subscription.html) to trigger your Lambda function by SNS can de declared inline.

event_source_mappings.tf

@@ -9,6 +9,11 @@ locals {
     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 Kinesis consumers for enhanced fan-out
+  kinesis_consumers = [
+    for k, v in var.event_source_mappings : lookup(v, "event_source_arn", null) if length(regexall(".*:kinesis:.*/consumer/.*", 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
@@ -113,7 +118,7 @@ data "aws_iam_policy_document" "event_sources" {
       resources = [
         // extracting 'arn:${Partition}:kinesis:${Region}:${Account}:stream/' from the kinesis stream ARN
         // see https://docs.aws.amazon.com/IAM/latest/UserGuide/list_amazonkinesis.html#amazonkinesis-resources-for-iam-policies
-        length(regexall("arn.*\\/", local.kinesis_event_sources[0])) > 0 ? "${regex("arn.*\\/", local.kinesis_event_sources[0])}*" : ""
+        length(regexall("arn:.*:kinesis:.*:.*:stream/", local.kinesis_event_sources[0])) > 0 ? "${regex("arn:.*:kinesis:.*:.*:stream/", local.kinesis_event_sources[0])}*" : ""
       ]
     }
   }
@@ -130,11 +135,19 @@ data "aws_iam_policy_document" "event_sources" {
       ]
 
       resources = [
-        for arn in local.kinesis_event_sources : arn
+        for arn in local.kinesis_event_sources : replace(arn, "/\\/consumer.*/", "")
       ]
     }
   }
 
+  dynamic "statement" {
+    for_each = length(local.kinesis_consumers) > 0 ? [true] : []
+    content {
+      actions   = ["kinesis:SubscribeToShard"]
+      resources = [for arn in local.kinesis_consumers : arn]
+    }
+  }
+
   // SQS permission for on-failure destinations
   dynamic "statement" {
     for_each = length(local.on_failure_sqs_destination_arns) > 0 ? [true] : []

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

@@ -42,7 +42,14 @@ module "lambda" {
     }
 
     stream_2 = {
-      event_source_arn = aws_kinesis_stream.stream_2.arn
+      // To use a consumer (enhanced fan-out), specify the consumer's ARN instead of the stream's ARN, see https://docs.aws.amazon.com/lambda/latest/dg/with-kinesis.html#services-kinesis-configure
+      event_source_arn = aws_kinesis_stream_consumer.this.arn
     }
   }
 }
+
+resource "aws_kinesis_stream_consumer" "this" {
+  name       = module.lambda.function_name
+  stream_arn = aws_kinesis_stream.stream_2.arn
+}
+