【Lambda Powertools】S3 イベント通知の二重処理を防いでみた

AWSTemplateFormatVersion: '2010-09-09'
Description: S3 Event Deduplication with Powertools Idempotency
Parameters:
  ProjectName:
    Type: String
    Default: s3-event-dedup
    Description: Project name used as prefix for resource names
  TtlSeconds:
    Type: Number
    Default: 7200
    Description: TTL for idempotency records (seconds). Default 2 hours.
Resources:
  # ============================================================
  # S3 Bucket
  # ============================================================
  SourceBucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: !Sub '${ProjectName}-source-${AWS::AccountId}'
      NotificationConfiguration:
        EventBridgeConfiguration:
          EventBridgeEnabled: true
      VersioningConfiguration:
        Status: Enabled
  # ============================================================
  # DynamoDB Table (Idempotency)
  # ============================================================
  IdempotencyTable:
    Type: AWS::DynamoDB::Table
    DeletionPolicy: Delete
    Properties:
      TableName: !Sub '${ProjectName}-idempotency'
      BillingMode: PAY_PER_REQUEST
      AttributeDefinitions:
        - AttributeName: id
          AttributeType: S
      KeySchema:
        - AttributeName: id
          KeyType: HASH
      TimeToLiveSpecification:
        AttributeName: expiration
        Enabled: true
  # ============================================================
  # IAM Role - Lambda
  # ============================================================
  LambdaRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: !Sub '${ProjectName}-lambda-role'
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
      Policies:
        - PolicyName: IdempotencyPolicy
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - dynamodb:PutItem
                  - dynamodb:GetItem
                  - dynamodb:UpdateItem
                  - dynamodb:DeleteItem
                Resource: !GetAtt IdempotencyTable.Arn
              - Effect: Allow
                Action:
                  - s3:GetObject
                Resource: !Sub '${SourceBucket.Arn}/*'
  # ============================================================
  # CloudWatch Logs - Lambda Log Group
  # ============================================================
  DedupFunctionLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    Properties:
      LogGroupName: !Sub '/aws/lambda/${ProjectName}-handler'
      RetentionInDays: 14
  # ============================================================
  # Lambda Function
  # ============================================================
  DedupFunction:
    Type: AWS::Lambda::Function
    DependsOn: DedupFunctionLogGroup
    Properties:
      FunctionName: !Sub '${ProjectName}-handler'
      Runtime: python3.13
      Handler: index.lambda_handler
      Role: !GetAtt LambdaRole.Arn
      Timeout: 60
      MemorySize: 256
      LoggingConfig:
        LogGroup: !Ref DedupFunctionLogGroup
      Layers:
        - !Sub 'arn:aws:lambda:${AWS::Region}:017000801446:layer:AWSLambdaPowertoolsPythonV3-python313-x86_64:7'
      Environment:
        Variables:
          IDEMPOTENCY_TABLE_NAME: !Ref IdempotencyTable
          TTL_SECONDS: !Ref TtlSeconds
          POWERTOOLS_SERVICE_NAME: !Sub '${ProjectName}'
      Code:
        ZipFile: |
          # プレースホルダー: ステップ2でコードを設定します
          def lambda_handler(event, context):
              return {'statusCode': 200}
  # ============================================================
  # EventBridge Rule
  # ============================================================
  S3EventRule:
    Type: AWS::Events::Rule
    Properties:
      Name: !Sub '${ProjectName}-s3-object-created'
      Description: Catch S3 ObjectCreated events and invoke Lambda
      State: ENABLED
      EventPattern:
        source:
          - aws.s3
        detail-type:
          - Object Created
        detail:
          bucket:
            name:
              - !Ref SourceBucket
      Targets:
        - Id: DedupFunctionTarget
          Arn: !GetAtt DedupFunction.Arn
  # ============================================================
  # Lambda Permission for EventBridge
  # ============================================================
  LambdaPermission:
    Type: AWS::Lambda::Permission
    Properties:
      FunctionName: !Ref DedupFunction
      Action: lambda:InvokeFunction
      Principal: events.amazonaws.com
      SourceArn: !GetAtt S3EventRule.Arn
Outputs:
  SourceBucketName:
    Value: !Ref SourceBucket
    Description: Source S3 bucket name
    Export:
      Name: !Sub '${AWS::StackName}-SourceBucketName'
  IdempotencyTableName:
    Value: !Ref IdempotencyTable
    Description: DynamoDB idempotency table name
    Export:
      Name: !Sub '${AWS::StackName}-IdempotencyTableName'
  DedupFunctionName:
    Value: !Ref DedupFunction
    Description: Lambda function name
    Export:
      Name: !Sub '${AWS::StackName}-DedupFunctionName'

"""
Amazon S3 イベント重複制御 AWS Lambda
- Powertools for AWS Lambda の Idempotency ユーティリティを使用
- Amazon EventBridge 経由で Amazon S3 ObjectCreated イベントを受信
- sequencer + ETag + versionId の組み合わせで重複判定
- ユニークイベントのみ本処理を実行
"""
import json
import os
import boto3
from aws_lambda_powertools import Logger
from aws_lambda_powertools.utilities.idempotency import (
    DynamoDBPersistenceLayer,
    IdempotencyConfig,
    idempotent
)
# ロガー
logger = Logger()
# 環境変数
IDEMPOTENCY_TABLE_NAME = os.environ['IDEMPOTENCY_TABLE_NAME']
TTL_SECONDS = int(os.environ['TTL_SECONDS'])
# Powertools Idempotency の設定
# DynamoDB を永続化レイヤーとして使用
persistence_layer = DynamoDBPersistenceLayer(
    table_name=IDEMPOTENCY_TABLE_NAME
)
# Idempotency 設定
# event_key_jmespath: イベントからユニークキーを抽出する JMESPath 式
# → detail.object 内の key, etag, version-id, sequencer を "#" で結合
idempotency_config = IdempotencyConfig(
    expires_after_seconds=TTL_SECONDS,
    event_key_jmespath='detail.object.[key, etag, "version-id", sequencer] | join(`#`, @)'
)
# S3 クライアント
s3_client = boto3.client('s3')
@idempotent(config=idempotency_config, persistence_store=persistence_layer)
def lambda_handler(event, context):
    """
    Powertools Idempotency により、同一イベントに対して1回だけ実行される。
    動作:
    - 初回（ユニーク）: 本処理を実行し、結果を DynamoDB に保存
    - 2回目以降（重複）: 処理をスキップし、前回の結果を返却
    """
    # EventBridge 経由の S3 イベントから情報を抽出
    detail = event.get('detail', {})
    bucket_name = detail.get('bucket', {}).get('name', '')
    obj = detail.get('object', {})
    object_key = obj.get('key', '')
    etag = obj.get('etag', '')
    version_id = obj.get('version-id', 'null')
    sequencer = obj.get('sequencer', '')
    # 受信イベント全体をログ出力（生のイベント JSON を確認するため）
    logger.info("Received event", extra={"event": event})
    logger.info("Processing unique event", extra={
        "bucket": bucket_name,
        "key": object_key,
        "etag": etag,
        "version_id": version_id,
        "sequencer": sequencer
    })
    # ================================================================
    # ここから本処理（ビジネスロジック）
    # ================================================================
    # サンプル: オブジェクトのメタデータを取得
    try:
        response = s3_client.head_object(
            Bucket=bucket_name,
            Key=object_key
        )
        content_length = response.get('ContentLength', 0)
        content_type = response.get('ContentType', 'unknown')
        logger.info("Object metadata retrieved", extra={
            "size_bytes": content_length,
            "content_type": content_type
        })
    except Exception as e:
        logger.error("Failed to get object metadata", extra={"error": str(e)})
        raise
    # ================================================================
    # 本処理ここまで
    # 実際のビジネスロジックに置き換えてください
    # 例: オブジェクトの加工、別バケットへのコピー、DB 書き込みなど
    # ================================================================
    result = {
        'statusCode': 200,
        'body': {
            'result': 'success',
            'bucket': bucket_name,
            'key': object_key,
            'size': content_length,
            'contentType': content_type
        }
    }
    logger.info("Processing completed", extra={"result": result})
    return result

"""
Amazon S3 イベント重複制御 AWS Lambda（Powertools を使わない手動実装版）
- Amazon EventBridge 経由で Amazon S3 ObjectCreated イベントを受信
- key + etag + version-id + sequencer の組み合わせで重複判定
- ユニークイベントのみ本処理を実行
"""
import os
import time
import json
import boto3
from botocore.exceptions import ClientError
# 環境変数
DEDUP_TABLE_NAME = os.environ['IDEMPOTENCY_TABLE_NAME']
TTL_SECONDS = int(os.environ['TTL_SECONDS'])
# AWS クライアント
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(DEDUP_TABLE_NAME)
s3_client = boto3.client('s3')
def lambda_handler(event, context):
    # 受信イベント全体をログ出力（生のイベント JSON を確認するため）
    print(json.dumps(event, ensure_ascii=False, default=str))
    # EventBridge 経由の S3 イベントから情報を抽出
    detail = event.get('detail', {})
    bucket_name = detail.get('bucket', {}).get('name', '')
    obj = detail.get('object', {})
    object_key = obj.get('key', '')
    etag = obj.get('etag', '')
    version_id = obj.get('version-id', 'null')
    sequencer = obj.get('sequencer', '')
    # ユニークキーを組み立てる
    # key + etag + version-id + sequencer を "#" 区切りで結合
    dedup_key = '#'.join([object_key, etag, version_id, sequencer])
    print(f"dedup_key: {dedup_key}")
    # TTL（現在時刻 + TTL_SECONDS 秒後の Unix エポック秒）
    ttl_epoch = int(time.time()) + TTL_SECONDS
    # 重複チェック（条件付き書き込み）
    try:
        table.put_item(
            Item={'id': dedup_key, 'expiration': ttl_epoch},
            ConditionExpression='attribute_not_exists(id)'
        )
    except ClientError as e:
        if e.response['Error']['Code'] == 'ConditionalCheckFailedException':
            # 既にレコードが存在する = 重複イベント → 処理をスキップ
            print(f"Duplicate event. Skipped. dedup_key: {dedup_key}")
            return {'statusCode': 200, 'body': {'result': 'duplicate'}}
        raise
    # ================================================================
    # ここから本処理（ユニークイベントのみ実行される）
    # 実際のビジネスロジックに置き換えてください
    # ================================================================
    response = s3_client.head_object(Bucket=bucket_name, Key=object_key)
    print(f"Processed: s3://{bucket_name}/{object_key}, "
          f"size={response.get('ContentLength', 0)}")
    return {'statusCode': 200, 'body': {'result': 'success'}}