IaC AWS

Terraform Module: Amazon Data Firehose (Kinesis Firehose) — Encrypted, Partitioned S3 Delivery Without the Footguns

Jun 23, 2026 11 min read Intermediate Vinod

Quick take — A reusable hashicorp/aws ~> 5.0 Terraform module for aws_kinesis_firehose_delivery_stream to Amazon S3: GZIP compression, CMK server-side encryption, CloudWatch logging, optional Parquet conversion via Glue, and dynamic partitioning — production defaults baked in. New here? Jump to the Quickstart below to deploy it in minutes; read on for how it works and when to reach for it.

Quickstart (copy-paste)

Minimal, runnable configuration — drop this in a .tf file and fill in the "..." placeholders (each required input is commented):

provider "aws" {
  region = "us-east-1"
}

module "firehose" {
  source = "git::https://dev.azure.com/teknohut/kloudvin/_git/terraform-modules//terraform-module-aws-kinesis-firehose?ref=v1.0.0"

  name           = "..."  # Delivery stream name (unique per region/account).
  bucket_arn     = "..."  # ARN of the destination S3 bucket (e.g. arn:aws:s3:::my-lake).
  kms_key_arn    = "..."  # Customer-managed KMS CMK ARN for at-rest encryption…
}

Then terraform init && terraform apply. Every other input has a sensible default — see Inputs below to override behaviour.

What this module is

Amazon Data Firehose (formerly Kinesis Data Firehose) is a fully managed service that captures streaming records — from Direct PUT producers, a Kinesis data stream, or an Amazon MSK topic — buffers them, optionally transforms them, and lands them in a destination like Amazon S3, Redshift, or OpenSearch. The aws_kinesis_firehose_delivery_stream resource with destination = "extended_s3" is the workhorse for building a data lake: it writes batched, compressed objects into S3 on a size/time buffer, with optional record-format conversion to Parquet and dynamic partitioning of the S3 prefix.

The defaults, however, are the ones that bite you. Out of the box compression_format is UNCOMPRESSED, so you pay full S3 storage and scan cost on raw JSON. There is no server-side encryption unless you explicitly configure it, and no CloudWatch error logging — so when delivery fails, records vanish into the void with no breadcrumbs. The IAM role wiring is fiddly: Firehose needs precise S3, KMS, Glue, and CloudWatch permissions, and a too-broad inline policy is the single most common review finding.

This module wraps the resource so the correct defaults ship by default: GZIP compression on, a customer-managed CMK enforced for server_side_encryption, an error-output prefix and CloudWatch logging configured, and a tightly scoped IAM role with exactly the permissions the chosen features need. Optional Parquet conversion (via a Glue Data Catalog table) and dynamic partitioning are flag-gated so you opt into complexity only when you want columnar, query-optimized output.

When to use it

Reach for Kinesis Data Streams + a custom consumer (or a managed Flink application) instead when you need sub-second latency, replay, or stateful stream processing. Firehose is near-real-time (a 60-second minimum buffer) and is the right tool when “land it cheaply and queryably in S3” is the goal — which covers the majority of analytics ingestion.

Module structure

terraform-module-aws-kinesis-firehose/
├── versions.tf      # provider + Terraform version pins
├── main.tf          # IAM role/policy, delivery stream, S3 + encryption wiring
├── variables.tf     # var-driven inputs with validations
└── outputs.tf       # stream ARN/name, role ARN, log group

versions.tf

terraform {
  required_version = ">= 1.5.0"

  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

main.tf

data "aws_caller_identity" "current" {}
data "aws_region" "current" {}
data "aws_partition" "current" {}

locals {
  tags = merge(
    {
      "Name"      = var.name
      "ManagedBy" = "terraform"
      "Module"    = "terraform-module-aws-kinesis-firehose"
    },
    var.tags,
  )

  log_group_name  = "/aws/kinesisfirehose/${var.name}"
  log_stream_name = "S3Delivery"

  # Glue catalog ARNs the delivery role needs when Parquet conversion is on.
  glue_table_arns = var.data_format_conversion_enabled ? [
    "arn:${data.aws_partition.current.partition}:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:catalog",
    "arn:${data.aws_partition.current.partition}:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:database/${var.glue_database_name}",
    "arn:${data.aws_partition.current.partition}:glue:${data.aws_region.current.name}:${data.aws_caller_identity.current.account_id}:table/${var.glue_database_name}/${var.glue_table_name}",
  ] : []
}

# CloudWatch log group for delivery errors. Without this, failed puts to S3
# have no diagnostic trail — a common production blind spot.
resource "aws_cloudwatch_log_group" "this" {
  name              = local.log_group_name
  retention_in_days = var.log_retention_in_days
  kms_key_id        = var.log_group_kms_key_arn
  tags              = local.tags
}

resource "aws_cloudwatch_log_stream" "this" {
  name           = local.log_stream_name
  log_group_name = aws_cloudwatch_log_group.this.name
}

# ---- IAM: assume role for the Firehose service ----
data "aws_iam_policy_document" "assume" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]

    principals {
      type        = "Service"
      identifiers = ["firehose.amazonaws.com"]
    }

    # Confused-deputy guard: only this account's Firehose may assume the role.
    condition {
      test     = "StringEquals"
      variable = "sts:ExternalId"
      values   = [data.aws_caller_identity.current.account_id]
    }
  }
}

resource "aws_iam_role" "this" {
  name_prefix        = "firehose-${substr(var.name, 0, 20)}-"
  assume_role_policy = data.aws_iam_policy_document.assume.json
  tags               = local.tags
}

# Least-privilege delivery policy: S3 put/list on the target bucket+prefix,
# KMS for encryption, CloudWatch for logging, and Glue only when converting.
data "aws_iam_policy_document" "delivery" {
  statement {
    sid    = "S3Delivery"
    effect = "Allow"
    actions = [
      "s3:AbortMultipartUpload",
      "s3:GetBucketLocation",
      "s3:GetObject",
      "s3:ListBucket",
      "s3:ListBucketMultipartUploads",
      "s3:PutObject",
    ]
    resources = [
      var.bucket_arn,
      "${var.bucket_arn}/*",
    ]
  }

  statement {
    sid    = "KmsEncrypt"
    effect = "Allow"
    actions = [
      "kms:Decrypt",
      "kms:GenerateDataKey",
    ]
    resources = [var.kms_key_arn]

    condition {
      test     = "StringEquals"
      variable = "kms:ViaService"
      values   = ["s3.${data.aws_region.current.name}.amazonaws.com"]
    }
  }

  statement {
    sid    = "CloudWatchLogs"
    effect = "Allow"
    actions = [
      "logs:PutLogEvents",
      "logs:CreateLogStream",
    ]
    resources = ["${aws_cloudwatch_log_group.this.arn}:*"]
  }

  dynamic "statement" {
    for_each = var.data_format_conversion_enabled ? [1] : []
    content {
      sid    = "GlueCatalog"
      effect = "Allow"
      actions = [
        "glue:GetTable",
        "glue:GetTableVersion",
        "glue:GetTableVersions",
      ]
      resources = local.glue_table_arns
    }
  }
}

resource "aws_iam_role_policy" "this" {
  name_prefix = "firehose-delivery-"
  role        = aws_iam_role.this.id
  policy      = data.aws_iam_policy_document.delivery.json
}

resource "aws_kinesis_firehose_delivery_stream" "this" {
  name        = var.name
  destination = "extended_s3"

  extended_s3_configuration {
    role_arn   = aws_iam_role.this.arn
    bucket_arn = var.bucket_arn

    # Buffer until either threshold is hit; larger objects = cheaper queries.
    buffering_size     = var.buffering_size
    buffering_interval = var.buffering_interval
    compression_format = var.compression_format

    # Time-format + custom prefixes; error_output_prefix isolates failed records.
    prefix              = var.prefix
    error_output_prefix = var.error_output_prefix

    cloudwatch_logging_options {
      enabled         = true
      log_group_name  = aws_cloudwatch_log_group.this.name
      log_stream_name = aws_cloudwatch_log_stream.this.name
    }

    # Optional: convert incoming JSON to columnar Parquet using a Glue table.
    dynamic "data_format_conversion_configuration" {
      for_each = var.data_format_conversion_enabled ? [1] : []
      content {
        input_format_configuration {
          deserializer {
            open_x_json_ser_de {}
          }
        }
        output_format_configuration {
          serializer {
            parquet_ser_de {}
          }
        }
        schema_configuration {
          role_arn      = aws_iam_role.this.arn
          database_name = var.glue_database_name
          table_name    = var.glue_table_name
          region        = data.aws_region.current.name
        }
      }
    }

    # Optional: route records into partitioned S3 prefixes from record content.
    dynamic "dynamic_partitioning_configuration" {
      for_each = var.dynamic_partitioning_enabled ? [1] : []
      content {
        enabled = true
      }
    }

    # When dynamic partitioning is on, a MetadataExtraction processor pulls
    # partition keys out of each record via a JQ query.
    dynamic "processing_configuration" {
      for_each = var.dynamic_partitioning_enabled ? [1] : []
      content {
        enabled = true
        processors {
          type = "MetadataExtraction"
          parameters {
            parameter_name  = "JsonParsingEngine"
            parameter_value = "JQ-1.6"
          }
          parameters {
            parameter_name  = "MetadataExtractionQuery"
            parameter_value = var.partition_keys_jq_query
          }
        }
      }
    }
  }

  # At-rest encryption of the delivery stream with a customer-managed CMK.
  server_side_encryption {
    enabled  = true
    key_type = "CUSTOMER_MANAGED_CMK"
    key_arn  = var.kms_key_arn
  }

  tags = local.tags
}

variables.tf

variable "name" {
  description = "Delivery stream name (unique per region/account)."
  type        = string

  validation {
    condition     = can(regex("^[a-zA-Z0-9_.-]{1,64}$", var.name))
    error_message = "name must be 1-64 chars: letters, digits, underscore, period, or hyphen."
  }
}

variable "bucket_arn" {
  description = "ARN of the destination S3 bucket."
  type        = string

  validation {
    condition     = can(regex("^arn:aws[a-z-]*:s3:::", var.bucket_arn))
    error_message = "bucket_arn must be a valid S3 bucket ARN (arn:aws:s3:::name)."
  }
}

variable "kms_key_arn" {
  description = "Customer-managed KMS CMK ARN used for stream and S3 object encryption."
  type        = string

  validation {
    condition     = can(regex("^arn:aws[a-z-]*:kms:", var.kms_key_arn))
    error_message = "kms_key_arn must be a valid KMS key ARN."
  }
}

variable "buffering_size" {
  description = "Buffer size in MiB before flushing to S3 (min 64 when format conversion is enabled)."
  type        = number
  default     = 64

  validation {
    condition     = var.buffering_size >= 1 && var.buffering_size <= 128
    error_message = "buffering_size must be between 1 and 128 MiB."
  }
}

variable "buffering_interval" {
  description = "Buffer interval in seconds before flushing to S3."
  type        = number
  default     = 300

  validation {
    condition     = var.buffering_interval >= 0 && var.buffering_interval <= 900
    error_message = "buffering_interval must be between 0 and 900 seconds."
  }
}

variable "compression_format" {
  description = "Compression for delivered objects: GZIP, Snappy, ZIP, HADOOP_SNAPPY, or UNCOMPRESSED."
  type        = string
  default     = "GZIP"

  validation {
    condition     = contains(["GZIP", "Snappy", "ZIP", "HADOOP_SNAPPY", "UNCOMPRESSED"], var.compression_format)
    error_message = "compression_format must be GZIP, Snappy, ZIP, HADOOP_SNAPPY, or UNCOMPRESSED."
  }
}

variable "prefix" {
  description = "S3 key prefix for delivered objects (supports !{timestamp:...} and partitionKeyFromQuery expressions)."
  type        = string
  default     = "data/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/"
}

variable "error_output_prefix" {
  description = "S3 key prefix for records Firehose fails to process or deliver."
  type        = string
  default     = "errors/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/!{firehose:error-output-type}/"
}

variable "log_retention_in_days" {
  description = "CloudWatch Logs retention for the delivery error log group."
  type        = number
  default     = 30
}

variable "log_group_kms_key_arn" {
  description = "Optional KMS key ARN to encrypt the CloudWatch log group. Null uses the default CloudWatch key."
  type        = string
  default     = null
}

variable "data_format_conversion_enabled" {
  description = "Convert incoming JSON records to Parquet using a Glue Data Catalog table."
  type        = bool
  default     = false
}

variable "glue_database_name" {
  description = "Glue Data Catalog database holding the target table. Required when data_format_conversion_enabled is true."
  type        = string
  default     = null
}

variable "glue_table_name" {
  description = "Glue Data Catalog table describing the output schema. Required when data_format_conversion_enabled is true."
  type        = string
  default     = null
}

variable "dynamic_partitioning_enabled" {
  description = "Enable dynamic partitioning so S3 prefixes are derived from record content."
  type        = bool
  default     = false
}

variable "partition_keys_jq_query" {
  description = "JQ-1.6 MetadataExtractionQuery mapping record fields to partition keys, e.g. '{customer_id:.customer_id}'."
  type        = string
  default     = "{customer_id:.customer_id}"
}

variable "tags" {
  description = "Additional tags merged onto every resource."
  type        = map(string)
  default     = {}
}

outputs.tf

output "arn" {
  description = "ARN of the Firehose delivery stream."
  value       = aws_kinesis_firehose_delivery_stream.this.arn
}

output "name" {
  description = "Name of the Firehose delivery stream."
  value       = aws_kinesis_firehose_delivery_stream.this.name
}

output "role_arn" {
  description = "ARN of the IAM role Firehose assumes to deliver records."
  value       = aws_iam_role.this.arn
}

output "role_name" {
  description = "Name of the delivery IAM role."
  value       = aws_iam_role.this.name
}

output "log_group_name" {
  description = "CloudWatch log group capturing delivery errors."
  value       = aws_cloudwatch_log_group.this.name
}

output "log_group_arn" {
  description = "ARN of the delivery error CloudWatch log group."
  value       = aws_cloudwatch_log_group.this.arn
}

How to use it

module "events_firehose" {
  source = "git::https://dev.azure.com/teknohut/kloudvin/_git/terraform-modules//terraform-module-aws-kinesis-firehose?ref=v1.0.0"

  name        = "clickstream-events-prod"
  bucket_arn  = aws_s3_bucket.lake.arn
  kms_key_arn = aws_kms_key.lake.arn

  buffering_size     = 128
  buffering_interval = 300
  compression_format = "GZIP"

  # Convert JSON to Parquet for cheap Athena scans.
  data_format_conversion_enabled = true
  glue_database_name             = aws_glue_catalog_database.analytics.name
  glue_table_name                = aws_glue_catalog_table.clickstream.name

  # Partition by customer and event date pulled from each record.
  dynamic_partitioning_enabled = true
  partition_keys_jq_query      = "{customer_id:.customer_id}"
  prefix                       = "clickstream/customer_id=!{partitionKeyFromQuery:customer_id}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"
  error_output_prefix          = "clickstream-errors/!{firehose:error-output-type}/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/"

  log_retention_in_days = 90

  tags = {
    Environment = "prod"
    Team        = "analytics"
    CostCenter  = "ANL-3310"
  }
}

# Downstream: let an application role write records via Direct PUT.
resource "aws_iam_role_policy" "app_put_records" {
  name = "clickstream-firehose-put"
  role = aws_iam_role.ingest_app.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect   = "Allow"
      Action   = ["firehose:PutRecord", "firehose:PutRecordBatch"]
      Resource = module.events_firehose.arn
    }]
  })
}

Pin the module with ?ref=<tag> so a stack never silently picks up a breaking module change. To ingest from a Kinesis data stream or an MSK topic instead of Direct PUT, extend the module with a kinesis_source_configuration / msk_source_configuration block — the S3 delivery wiring above is unchanged.

With Terragrunt

Terragrunt keeps this module DRY across environments — define the backend and provider once in a root config, then a thin terragrunt.hcl per environment supplies only the inputs that differ.

1. Root config — live/terragrunt.hcl (inherited by every module):

remote_state {
  backend = "s3"
  generate = { path = "backend.tf", if_exists = "overwrite" }
  config = {
    # ...s3 state bucket/container + key per path...
  }
}

2. Module config — live/prod/firehose/terragrunt.hcl:

include "root" {
  path = find_in_parent_folders()
}

terraform {
  source = "git::https://dev.azure.com/teknohut/kloudvin/_git/terraform-modules//terraform-module-aws-kinesis-firehose?ref=v1.0.0"
}

inputs = {
  name = "..."
  bucket_arn = "..."
  kms_key_arn = "..."
}

3. Deploy one environment, or roll out all modules together:

cd live/prod/firehose && terragrunt apply        # this module
terragrunt run-all apply                      # every module under live/prod

Why Terragrunt here: the backend and provider live in one place instead of being copy-pasted into every module; inputs is overridden per environment (dev / stage / prod) without forking the module; and run-all orchestrates dependencies across modules. Reach for it once you have more than one environment or more than a handful of modules — for a single stack, the plain Quickstart above is enough.

Inputs

Name Type Default Required Description
name string — Yes Delivery stream name (unique per region/account).
bucket_arn string — Yes ARN of the destination S3 bucket.
kms_key_arn string — Yes Customer-managed KMS CMK for stream + S3 encryption.
buffering_size number 64 No Buffer size in MiB before flushing (≥64 when converting format).
buffering_interval number 300 No Buffer interval in seconds (0–900).
compression_format string GZIP No GZIP, Snappy, ZIP, HADOOP_SNAPPY, or UNCOMPRESSED.
prefix string time-format prefix No S3 key prefix (supports timestamp/partition expressions).
error_output_prefix string error prefix No S3 key prefix for failed records.
log_retention_in_days number 30 No CloudWatch Logs retention for the error log group.
log_group_kms_key_arn string null No KMS key to encrypt the CloudWatch log group.
data_format_conversion_enabled bool false No Convert JSON to Parquet via a Glue table.
glue_database_name string null No Glue database (required when converting format).
glue_table_name string null No Glue table describing output schema (required when converting).
dynamic_partitioning_enabled bool false No Derive S3 prefixes from record content.
partition_keys_jq_query string {customer_id:.customer_id} No JQ-1.6 metadata extraction query for partition keys.
tags map(string) {} No Additional tags merged onto every resource.

Outputs

Name Description
arn ARN of the Firehose delivery stream.
name Name of the Firehose delivery stream.
role_arn ARN of the IAM role Firehose assumes to deliver records.
role_name Name of the delivery IAM role.
log_group_name CloudWatch log group capturing delivery errors.
log_group_arn ARN of the delivery error CloudWatch log group.

Enterprise scenario

An ad-tech platform ingests roughly two million clickstream events per minute across dev, staging, and prod. The data team publishes this module at v1.0.0 so every event type — impressions, clicks, conversions — gets its own delivery stream with GZIP compression, a customer-managed CMK, and a 90-day error log group enforced by default. Production streams flip on Parquet conversion against a shared Glue Data Catalog and dynamic partitioning by customer_id, so Athena queries prune to a single tenant’s partitions and scan columnar Parquet instead of raw JSON — cutting query cost by an order of magnitude. Because the delivery role is scoped to exactly the target bucket prefix plus the specific Glue table, a quarterly security review confirms no Firehose role can read another team’s data lake or touch an unrelated catalog.

Best practices

Part of the KloudVin Terraform module library. Pair this with the S3 Bucket, KMS Key, and Glue Catalog modules — they supply the bucket_arn, kms_key_arn, and Glue table this module delivers into.

TerraformAWSKinesis FirehoseModuleIaC
Need this built for real?

Vinod is a Senior Cloud Architect (22+ yrs) — available for Azure / AWS / GCP architecture, landing zones, and migrations.

Work with me
VH

Written by Vinod

Senior Cloud Architect with 22+ years of experience. Sharing practical knowledge from production environments.

Comments

// more articles

Keep Reading

Terraform Module: The Full Azure Landing Zones (ALZ) Policy Library as Code

IaC 24 min read

Terraform Module: AWS Private CA (ACM PCA) — A Root or Subordinate Certificate Authority as Code

IaC 12 min read

Terraform Module: AWS CloudWatch Dashboard — Typed Widgets, jsonencode, and Dashboards as Code

IaC 11 min read