Setting up Amazon Kinesis with Terraform

Learn how to provision Kinesis Data Streams and Kinesis Data Firehose with Terraform for real-time data ingestion and delivery to S3

Kinesis Data Streams gives you a durable, ordered, replayable log for real-time event ingestion - the AWS-native equivalent of a Kafka topic - and Kinesis Data Firehose handles the unglamorous but common case of buffering that stream and loading it into S3, without you managing any consumer code. This guide covers both with Terraform.

Prerequisites

  • AWS CLI configured with appropriate permissions
  • Terraform installed (version 1.0.0 or later)

Project Structure

aws-kinesis-terraform/
├── main.tf
├── variables.tf
├── outputs.tf
└── terraform.tfvars

Kinesis Data Stream

# main.tf
provider "aws" {
  region = var.aws_region
}

resource "aws_kms_key" "kinesis" {
  description             = "KMS key for Kinesis stream encryption"
  deletion_window_in_days = 7
  enable_key_rotation     = true
}

resource "aws_kinesis_stream" "events" {
  name             = "${var.project_name}-events"
  retention_period = 24

  stream_mode_details {
    stream_mode = "ON_DEMAND"
  }

  encryption_type = "KMS"
  kms_key_id      = aws_kms_key.kinesis.arn

  tags = {
    Environment = var.environment
  }
}

ON_DEMAND mode scales shard count automatically based on throughput and is the simplest starting point. For predictable, high-volume workloads where you want to control cost precisely, switch to provisioned mode with an explicit shard count instead:

resource "aws_kinesis_stream" "provisioned" {
  name        = "${var.project_name}-provisioned"
  shard_count = var.shard_count

  stream_mode_details {
    stream_mode = "PROVISIONED"
  }

  encryption_type = "KMS"
  kms_key_id      = aws_kms_key.kinesis.arn

  retention_period = 24

  tags = {
    Environment = var.environment
  }
}

Each shard supports up to 1 MB/sec or 1,000 records/sec of ingest, and 2 MB/sec of output - size shard_count from your expected peak throughput, not average.

Producer and Consumer IAM Policies

resource "aws_iam_policy" "kinesis_producer" {
  name = "${var.project_name}-kinesis-producer"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "kinesis:PutRecord",
          "kinesis:PutRecords",
          "kinesis:DescribeStreamSummary"
        ]
        Resource = aws_kinesis_stream.events.arn
      },
      {
        Effect   = "Allow"
        Action   = ["kms:GenerateDataKey", "kms:Decrypt"]
        Resource = aws_kms_key.kinesis.arn
      }
    ]
  })
}

resource "aws_iam_policy" "kinesis_consumer" {
  name = "${var.project_name}-kinesis-consumer"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "kinesis:GetRecords",
          "kinesis:GetShardIterator",
          "kinesis:DescribeStreamSummary",
          "kinesis:ListShards",
          "kinesis:SubscribeToShard",
          "kinesis:DescribeStreamConsumer"
        ]
        Resource = aws_kinesis_stream.events.arn
      },
      {
        Effect   = "Allow"
        Action   = ["kms:Decrypt"]
        Resource = aws_kms_key.kinesis.arn
      }
    ]
  })
}

A consumer built with the Kinesis Client Library (KCL) also needs a DynamoDB table for checkpointing and lease tracking, plus dynamodb:* permissions scoped to that table - KCL creates the table itself on first run if the IAM role allows dynamodb:CreateTable.

Lambda Consumer via Event Source Mapping

For simple stream processing, skip KCL entirely and let Lambda poll the stream directly:

resource "aws_lambda_event_source_mapping" "stream_processor" {
  event_source_arn                  = aws_kinesis_stream.events.arn
  function_name                     = aws_lambda_function.processor.arn
  starting_position                 = "LATEST"
  batch_size                        = 100
  maximum_batching_window_in_seconds = 5
  parallelization_factor            = 1

  bisect_batch_on_function_error = true

  destination_config {
    on_failure {
      destination_arn = aws_sqs_queue.stream_dlq.arn
    }
  }
}

resource "aws_sqs_queue" "stream_dlq" {
  name = "${var.project_name}-stream-dlq"
}

resource "aws_lambda_function" "processor" {
  filename         = "processor.zip"
  function_name    = "${var.project_name}-stream-processor"
  role             = aws_iam_role.lambda.arn
  handler          = "index.handler"
  runtime          = "nodejs20.x"
  timeout          = 60

  tags = {
    Environment = var.environment
  }
}

resource "aws_iam_role" "lambda" {
  name = "${var.project_name}-stream-processor-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "lambda.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy_attachment" "lambda_kinesis" {
  role       = aws_iam_role.lambda.name
  policy_arn = aws_iam_policy.kinesis_consumer.arn
}

resource "aws_iam_role_policy_attachment" "lambda_basic" {
  role       = aws_iam_role.lambda.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
}

Kinesis Data Firehose: Stream to S3

Firehose buffers records and writes them to a destination for you - no consumer code needed at all for the common “land everything in S3, query with Athena” pattern.

resource "aws_s3_bucket" "firehose_destination" {
  bucket = "${var.project_name}-firehose-data"
}

resource "aws_iam_role" "firehose" {
  name = "${var.project_name}-firehose-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "firehose.amazonaws.com"
        }
      }
    ]
  })
}

resource "aws_iam_role_policy" "firehose" {
  name = "${var.project_name}-firehose-policy"
  role = aws_iam_role.firehose.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:AbortMultipartUpload",
          "s3:GetBucketLocation",
          "s3:GetObject",
          "s3:ListBucket",
          "s3:ListBucketMultipartUploads",
          "s3:PutObject"
        ]
        Resource = [
          aws_s3_bucket.firehose_destination.arn,
          "${aws_s3_bucket.firehose_destination.arn}/*"
        ]
      },
      {
        Effect = "Allow"
        Action = [
          "kinesis:DescribeStream",
          "kinesis:GetShardIterator",
          "kinesis:GetRecords",
          "kinesis:ListShards"
        ]
        Resource = aws_kinesis_stream.events.arn
      }
    ]
  })
}

resource "aws_cloudwatch_log_group" "firehose" {
  name              = "/aws/kinesisfirehose/${var.project_name}"
  retention_in_days = 30
}

resource "aws_kinesis_firehose_delivery_stream" "to_s3" {
  name        = "${var.project_name}-to-s3"
  destination = "extended_s3"

  kinesis_source_configuration {
    kinesis_stream_arn = aws_kinesis_stream.events.arn
    role_arn            = aws_iam_role.firehose.arn
  }

  extended_s3_configuration {
    role_arn           = aws_iam_role.firehose.arn
    bucket_arn         = aws_s3_bucket.firehose_destination.arn
    prefix              = "year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
    error_output_prefix = "errors/!{firehose:error-output-type}/"
    buffering_size      = 64
    buffering_interval  = 300
    compression_format  = "GZIP"

    cloudwatch_logging_options {
      enabled         = true
      log_group_name  = aws_cloudwatch_log_group.firehose.name
      log_stream_name = "S3Delivery"
    }
  }

  tags = {
    Environment = var.environment
  }
}

buffering_size (MB) and buffering_interval (seconds) are both upper bounds - Firehose flushes whichever limit is hit first, so a low-traffic stream still delivers within buffering_interval even if it never reaches buffering_size.

Monitoring

resource "aws_cloudwatch_metric_alarm" "iterator_age" {
  alarm_name          = "${var.project_name}-consumer-lag"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "2"
  metric_name         = "GetRecords.IteratorAgeMilliseconds"
  namespace           = "AWS/Kinesis"
  period              = "300"
  statistic           = "Maximum"
  threshold           = "60000"
  alarm_description   = "Consumer is falling behind the stream"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = {
    StreamName = aws_kinesis_stream.events.name
  }
}

resource "aws_cloudwatch_metric_alarm" "write_throttled" {
  alarm_name          = "${var.project_name}-write-throttled"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = "1"
  metric_name         = "WriteProvisionedThroughputExceeded"
  namespace           = "AWS/Kinesis"
  period              = "300"
  statistic           = "Sum"
  threshold           = "0"
  alarm_description   = "Producers are being throttled - add shards or switch to ON_DEMAND"
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = {
    StreamName = aws_kinesis_stream.events.name
  }
}

resource "aws_sns_topic" "alerts" {
  name = "${var.project_name}-kinesis-alerts"
}

GetRecords.IteratorAgeMilliseconds is the single most useful Kinesis metric to alarm on: it directly measures how far behind real-time your consumer has fallen, regardless of whether the cause is a slow consumer or an under-shard stream.

Variables Configuration

variable "aws_region" {
  description = "AWS region"
  type        = string
  default     = "us-west-2"
}

variable "project_name" {
  description = "Project name"
  type        = string
}

variable "environment" {
  description = "Environment name"
  type        = string
  default     = "dev"
}

variable "shard_count" {
  description = "Shard count for provisioned-mode streams"
  type        = number
  default     = 2
}

Best Practices

  1. Stream Mode

    • Start with ON_DEMAND unless you have a well-understood, steady throughput profile - it removes an entire class of “did we under-provision shards” incidents
    • Switch to PROVISIONED only once you can justify the cost tradeoff with real traffic data

  2. Consumers

    • Prefer enhanced fan-out (aws_kinesis_stream_consumer + SubscribeToShard) over standard polling when you have multiple consumer applications reading the same stream - standard throughput (2 MB/sec/shard) is shared across all polling consumers, enhanced fan-out gives each registered consumer its own 2 MB/sec
    • Set bisect_batch_on_function_error on Lambda event source mappings so one malformed record doesn’t block an entire batch indefinitely

  3. Firehose

    • Keep buffering_interval as high as your latency requirements allow - fewer, larger S3 objects are both cheaper and faster to query with Athena than many small ones

Conclusion

Kinesis Data Streams and Firehose solve different problems that are easy to conflate: Streams is for when you need multiple independent consumers reading the same ordered log in near real time, Firehose is for when you just need the data to land somewhere durable with minimal operational effort. Many pipelines legitimately use both - a stream for real-time processing, with Firehose attached as a second consumer purely for the S3/Athena archive.

Remember to:

  • Alarm on iterator age, not just error counts
  • Size shards (or pick ON_DEMAND) based on peak, not average, throughput
  • Keep the KMS key’s grants scoped per producer/consumer role rather than reused broadly