如何通过Terraform实现Lambda与自托管Kafka的SASL认证？

要通过Terraform配置Lambda连接自托管Kafka的SASL认证，核心是在aws_lambda_event_source_mapping资源中配置sasl_configuration块，同时配合VPC配置和IAM权限完成部署。以下是具体实现步骤和代码示例：

1. 核心配置说明 #

• SASL认证配置：通过sasl_configuration块指定认证类型（支持SCRAM-SHA-256/SCRAM-SHA-512或PLAIN），并引用存储在AWS Secrets Manager中的凭证（避免硬编码敏感信息）。
• VPC网络配置：自托管Kafka通常部署在VPC内，需为Lambda配置vpc_config块指定子网和安全组，确保Lambda能访问Kafka集群。
• IAM权限：Lambda角色需要具备访问Secrets Manager（读取SASL凭证）和VPC资源（弹性网卡、子网等）的权限，同时允许Lambda作为Kafka消费者的事件源。

2. 完整Terraform代码示例 #

# 1. 创建Lambda执行角色
resource "aws_iam_role" "lambda_kafka_role" {
  name = "lambda-kafka-consumer-role"

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

# 2. 为角色添加基础VPC访问权限
resource "aws_iam_role_policy_attachment" "lambda_kafka_policy" {
  role       = aws_iam_role.lambda_kafka_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole"
}

# 3. 为角色添加Secrets Manager访问权限
resource "aws_iam_role_policy" "lambda_secrets_access" {
  name   = "lambda-secrets-manager-access"
  role   = aws_iam_role.lambda_kafka_role.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = [
          "secretsmanager:GetSecretValue"
        ]
        Effect   = "Allow"
        Resource = aws_secretsmanager_secret.kafka_sasl_cred.arn
      }
    ]
  })
}

# 4. 存储SASL凭证到Secrets Manager
resource "aws_secretsmanager_secret" "kafka_sasl_cred" {
  name = "kafka-sasl-credentials"
}

resource "aws_secretsmanager_secret_version" "kafka_sasl_cred" {
  secret_id     = aws_secretsmanager_secret.kafka_sasl_cred.id
  secret_string = jsonencode({
    username = "your-kafka-sasl-username"
    password = "your-kafka-sasl-password"
  })
}

# 5. 创建Lambda消费者函数（替换为你的业务代码包）
resource "aws_lambda_function" "kafka_consumer" {
  filename      = "lambda_consumer.zip"
  function_name = "kafka-consumer-lambda"
  role          = aws_iam_role.lambda_kafka_role.arn
  handler       = "main.handler"
  runtime       = "python3.12"

  # VPC配置，确保Lambda能访问自托管Kafka集群
  vpc_config {
    subnet_ids         = ["subnet-xxxxxx", "subnet-yyyyyy"]
    security_group_ids = ["sg-xxxxxx"]
  }
}

# 6. 配置Lambda事件源映射（核心SASL认证配置）
resource "aws_lambda_event_source_mapping" "kafka_event_source" {
  event_source_arn = "arn:aws:kafka:us-east-1:1234567890:cluster/your-kafka-cluster/abc123-def456" # 替换为你的Kafka集群ARN
  function_name    = aws_lambda_function.kafka_consumer.arn
  starting_position = "LATEST"

  # 示例：SCRAM-SHA-256认证配置
  sasl_configuration {
    sasl_scram {
      secret_arn = aws_secretsmanager_secret.kafka_sasl_cred.arn
    }
  }

  # 如果使用PLAIN认证，替换为以下配置块：
  # sasl_configuration {
  #   sasl_plain {
  #     secret_arn = aws_secretsmanager_secret.kafka_sasl_cred.arn
  #   }
  # }
}

3. 关键细节说明 #

• Kafka集群ARN：需替换为实际的自托管Kafka集群ARN，格式为arn:aws:kafka:<region>:<account-id>:cluster/<cluster-name>/<cluster-uuid>。
• SASL凭证格式：Secrets Manager中的秘钥必须是包含username和password字段的JSON格式，Lambda会自动解析这两个字段用于认证。
• 安全组规则：Lambda的安全组需允许向Kafka集群的安全组发起出站连接（默认Kafka端口为9092或9093，根据SASL加密配置调整）。
• 部署验证：部署完成后，可在AWS控制台的Lambda事件源映射页面查看连接状态，确认是否成功关联Kafka集群。

内容的提问来源于stack exchange，提问作者0xDjole