Amazon Data FirehoseのLambda変換で、失敗・破棄したレコードがどこへ行くのか確かめてみた

AWSTemplateFormatVersion: "2010-09-09"
Description: Amazon Data Firehose + Lambda data transformation lab (failed record behavior)

Parameters:
  ProjectName:
    Type: String
    Default: fh-lab
    AllowedPattern: "^[a-z0-9][a-z0-9-]{1,18}[a-z0-9]$"
    Description: Resource name prefix (lowercase letters, digits, hyphens, 3-20 chars)
  BackupMode:
    Type: String
    Default: Enabled
    AllowedValues:
      - Enabled
      - Disabled
    Description: Back up all source records to S3 (source/ prefix)
  LambdaRetries:
    Type: Number
    Default: 3
    MinValue: 0
    MaxValue: 300
    Description: Firehose retries when the Lambda invocation itself fails (NumberOfRetries)

Conditions:
  BackupEnabled: !Equals [!Ref BackupMode, Enabled]

Resources:
  # ---------- 配信先S3 ----------
  DestBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub ${ProjectName}-${AWS::AccountId}-${AWS::Region}
      PublicAccessBlockConfiguration:
        BlockPublicAcls: true
        BlockPublicPolicy: true
        IgnorePublicAcls: true
        RestrictPublicBuckets: true
      BucketEncryption:
        ServerSideEncryptionConfiguration:
          - ServerSideEncryptionByDefault:
              SSEAlgorithm: AES256

  # ---------- 変換用Lambda ----------
  TransformLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/lambda/${ProjectName}-transform
      RetentionInDays: 7

  TransformRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      Policies:
        - PolicyName: write-logs
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - logs:CreateLogStream
                  - logs:PutLogEvents
                Resource: !GetAtt TransformLogGroup.Arn

  TransformFunction:
    Type: AWS::Lambda::Function
    DependsOn: TransformLogGroup
    Properties:
      FunctionName: !Sub ${ProjectName}-transform
      Runtime: python3.12
      Handler: index.handler
      Role: !GetAtt TransformRole.Arn
      Timeout: 60
      MemorySize: 128
      Code:
        ZipFile: |
          import base64
          import json
          from datetime import datetime, timezone

          TAX_RATE = 1.10

          def result(record_id, status, data):
              return {"recordId": record_id, "result": status, "data": data}

          def handler(event, context):
              out = []
              counts = {"Ok": 0, "Dropped": 0, "ProcessingFailed": 0}
              for r in event["records"]:
                  rid = r["recordId"]
                  raw = base64.b64decode(r["data"])

                  # JSONとして読めないレコードは変換失敗にする
                  try:
                      body = json.loads(raw)
                  except ValueError:
                      body = None
                  if not isinstance(body, dict):
                      out.append(result(rid, "ProcessingFailed", r["data"]))
                      counts["ProcessingFailed"] += 1
                      continue

                  action = body.get("action")

                  # Lambda自体が例外で落ちるケース（バッチ全体が失敗する）
                  if action == "raise":
                      raise RuntimeError("intentional failure for lab")

                  if action == "drop":
                      out.append(result(rid, "Dropped", r["data"]))
                      counts["Dropped"] += 1
                      continue

                  if action == "fail":
                      out.append(result(rid, "ProcessingFailed", r["data"]))
                      counts["ProcessingFailed"] += 1
                      continue

                  # 正常系：税込金額と処理時刻を足す。末尾に改行を付けてJSON Linesにする
                  body["amount_with_tax"] = round(body.get("amount", 0) * TAX_RATE)
                  body["processed_at"] = datetime.now(timezone.utc).isoformat()
                  encoded = base64.b64encode(
                      (json.dumps(body, ensure_ascii=False) + "\n").encode("utf-8")
                  ).decode("utf-8")
                  out.append(result(rid, "Ok", encoded))
                  counts["Ok"] += 1

              print(json.dumps({"received": len(event["records"]), **counts}))
              return {"records": out}

  # ---------- Firehose ----------
  FirehoseLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub /aws/kinesisfirehose/${ProjectName}
      RetentionInDays: 7

  FirehoseLogStream:
    Type: AWS::Logs::LogStream
    Properties:
      LogGroupName: !Ref FirehoseLogGroup
      LogStreamName: DestinationDelivery

  FirehoseRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: "2012-10-17"
        Statement:
          - Effect: Allow
            Principal:
              Service: firehose.amazonaws.com
            Action: sts:AssumeRole
            Condition:
              StringEquals:
                sts:ExternalId: !Ref AWS::AccountId
      Policies:
        - PolicyName: deliver-to-s3-and-invoke-lambda
          PolicyDocument:
            Version: "2012-10-17"
            Statement:
              - Effect: Allow
                Action:
                  - s3:AbortMultipartUpload
                  - s3:GetBucketLocation
                  - s3:GetObject
                  - s3:ListBucket
                  - s3:ListBucketMultipartUploads
                  - s3:PutObject
                Resource:
                  - !GetAtt DestBucket.Arn
                  - !Sub ${DestBucket.Arn}/*
              - Effect: Allow
                Action:
                  - lambda:InvokeFunction
                  - lambda:GetFunctionConfiguration
                Resource: !GetAtt TransformFunction.Arn
              - Effect: Allow
                Action: logs:PutLogEvents
                Resource: !GetAtt FirehoseLogGroup.Arn

  DeliveryStream:
    Type: AWS::KinesisFirehose::DeliveryStream
    Properties:
      DeliveryStreamName: !Sub ${ProjectName}-stream
      DeliveryStreamType: DirectPut
      ExtendedS3DestinationConfiguration:
        BucketARN: !GetAtt DestBucket.Arn
        RoleARN: !GetAtt FirehoseRole.Arn
        Prefix: transformed/!{timestamp:yyyy/MM/dd/HH}/
        ErrorOutputPrefix: errors/!{firehose:error-output-type}/!{timestamp:yyyy/MM/dd/HH}/
        BufferingHints:
          IntervalInSeconds: 60
          SizeInMBs: 1
        CompressionFormat: UNCOMPRESSED
        CloudWatchLoggingOptions:
          Enabled: true
          LogGroupName: !Ref FirehoseLogGroup
          LogStreamName: !Ref FirehoseLogStream
        ProcessingConfiguration:
          Enabled: true
          Processors:
            - Type: Lambda
              Parameters:
                - ParameterName: LambdaArn
                  ParameterValue: !GetAtt TransformFunction.Arn
                - ParameterName: NumberOfRetries
                  ParameterValue: !Ref LambdaRetries
                - ParameterName: BufferSizeInMBs
                  ParameterValue: "1"
                - ParameterName: BufferIntervalInSeconds
                  ParameterValue: "60"
        S3BackupMode: !Ref BackupMode
        S3BackupConfiguration: !If
          - BackupEnabled
          - BucketARN: !GetAtt DestBucket.Arn
            RoleARN: !GetAtt FirehoseRole.Arn
            Prefix: source/!{timestamp:yyyy/MM/dd/HH}/
            ErrorOutputPrefix: backup-errors/!{firehose:error-output-type}/!{timestamp:yyyy/MM/dd/HH}/
            BufferingHints:
              IntervalInSeconds: 60
              SizeInMBs: 1
            CompressionFormat: UNCOMPRESSED
          - !Ref AWS::NoValue

Outputs:
  DeliveryStreamName:
    Value: !Ref DeliveryStream
  BucketName:
    Value: !Ref DestBucket
  FunctionName:
    Value: !Ref TransformFunction
  FirehoseLogGroupName:
    Value: !Ref FirehoseLogGroup