processing-s3-uploads-with-step-functions

Step Functions ワークフロー：S3 アップロードを Lambda または Fargate にルーティングする

概要

このスキルは、AWS CLI を使用してイベント駆動型ワークフローをデプロイします。S3 バケットにファイルがアップロードされると、EventBridge が Step Functions ステートマシンをトリガーします。ステートマシンはファイルサイズを確認し、処理を Lambda 関数（6 MB 以下のファイル）または Fargate タスク（6 MB を超えるファイル）にルーティングします。

アーキテクチャには、次の要素が含まれます。

• EventBridge 通知が有効になっている S3 バケット
• S3 オブジェクトの作成時に Step Functions をトリガーする EventBridge ルール
• ルーティング用の Choice 状態を含む Step Functions ステートマシン
• 小さなファイルを処理する Lambda 関数
• 大きなファイルを処理する ECS Fargate タスク
• 2 つのサブネット、インターネットゲートウェイ、セキュリティグループを備えた VPC
• Fargate コンテナイメージ用の ECR リポジトリ
• Lambda、Step Functions、ECS タスク用にスコープを限定した IAM ロール

次の場合にこのスキルを使用します。

• ファイルサイズに応じて異なるコンピューティングリソースで S3 アップロードを処理する必要がある
• 小さなファイルと大きなファイルの両方に対応できるサーバーレスワークフローが必要である
• Lambda と Fargate を使用した Step Functions によるオーケストレーションが必要である

次の場合はこのスキルを使用しないでください。

• すべてのファイルが Lambda で処理できるサイズである（S3 → Lambda を直接使用してください）
• リアルタイムストリーミングが必要である（Kinesis を使用してください）
• ファイルサイズに応じたルーティングが不要である

前提条件

1. AWS CLI v2 — インストールおよび設定済みであること。aws sts get-caller-identity で確認してください。
2. Python 3.12 — Lambda 関数のランタイム用です。
3. Docker — Fargate コンテナイメージのビルドとプッシュ用です。

パラメーター

• bucket_name（必須）：S3 バケット名（グローバルで一意、小文字、3～63 文字）
• region（必須）：すべてのリソースに使用する AWS リージョン
• ecr_repo_name（必須）：ECR リポジトリ名
• state_machine_name（必須）：Step Functions ステートマシン名
• kms_key_arn（任意）：CloudWatch Logs の暗号化に使用する KMS キーの ARN。指定しない場合は、aws kms create-key --description "Key for CloudWatch Logs encryption" --region {region} で作成します

パラメーター取得に関する制約：

• すべての必須パラメーターを、最初のプロンプトでまとめて必ず確認してください
• 複数の入力方法（直接入力、ファイルパス、URL）に必ず対応してください
• すべてのパラメーターを正常に取得したことを、先に進む前に必ず確認してください
• bucket_name が S3 の命名規則に従っていることを必ず検証してください

手順

ステップ 0：依存関係を確認する

制約：

• 次のツールが利用可能であることを必ず確認してください：aws-cli、python3（3.12 以降）、docker
• 不足しているツールがある場合は、明確なメッセージでユーザーに必ず知らせてください
• 不足しているツールがあっても続行するかどうか、ユーザーに必ず確認してください
• どの時点でも、処理を中止するというお客様の判断に必ず従ってください
• どのステップを実行しているのか、その理由、および呼び出すツールを、お客様に必ず説明してください

ステップ 1：AWS アカウント ID を取得する

制約：

• 次のコマンドでアカウント ID を必ず取得してください：aws sts get-caller-identity --query 'Account' --output text
• 後続のすべてのステップで使用できるよう、取得結果を {account_id} として必ず保存してください
• 認証情報が設定されていない場合は、必ず処理を中止してください

ステップ 2：デフォルト VPC とネットワークを取得する

制約：

• 次のコマンドでデフォルト VPC ID を取得する必要があります： aws ec2 describe-vpcs --filters Name=isDefault,Values=true --query 'Vpcs[0].VpcId' --output text --region {region}
• デフォルト VPC が存在しない場合は、aws ec2 create-default-vpc --region {region} を使用して作成するか、VPC ID を手動で指定する必要があることをユーザーに伝えます
• 次のコマンドでデフォルト VPC から 2 つのサブネット ID を取得する必要があります： aws ec2 describe-subnets --filters Name=vpc-id,Values={vpc_id} --query 'Subnets[0:2].SubnetId' --output text --region {region}
• デフォルト VPC にセキュリティグループを作成する必要があります： aws ec2 create-security-group --group-name fargate-sg --description "Security group for Fargate tasks" --vpc-id {vpc_id} --region {region}
• HTTPS と DNS のアウトバウンドのみを許可するように、セキュリティグループのエグレスルールを設定する必要があります。まず、デフォルトのすべて許可のエグレスルールを取り消します： aws ec2 revoke-security-group-egress --group-id {sg_id} --ip-permissions IpProtocol=-1,IpRanges='[{CidrIp=0.0.0.0/0}]' --region {region} 次に、対象を限定したルールを追加します： aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol tcp --port 443 --cidr 0.0.0.0/0 --region {region} と aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol udp --port 53 --cidr 0.0.0.0/0 --region {region}
• 本番ワークロードでは、インターネット経由のトラフィックを避け、広範なエグレスルールを不要にするために、S3 と CloudWatch Logs の VPC エンドポイントを推奨する必要があります
• 後続の手順で使用するため、{vpc_id}、{subnet1_id}、{subnet2_id}、{sg_id} を記録する必要があります

ステップ 3：ECR リポジトリを作成する

制約：

• 次のコマンドでリポジトリを作成する必要があります： aws ecr create-repository --repository-name {ecr_repo_name} --region {region}
• レスポンスから repositoryUri を記録する必要があります

ステップ 4：コンテナイメージをビルドしてプッシュする

制約：

• docker --version を実行して Docker がインストールされていることを確認する必要があります。Docker がインストールされていない場合は、https://docs.docker.com/get-docker/ からインストールするようユーザーに案内し、利用可能になるまで中止します

• 次のコマンドで Docker を ECR に認証する必要があります： aws ecr get-login-password --region {region} | docker login --username AWS --password-stdin {account_id}.dkr.ecr.{region}.amazonaws.com

• Dockerfile とプロセッサーコードは scripts/Dockerfile と scripts/fargate_processor.py にあります

• scripts ディレクトリからイメージをビルドしてプッシュする必要があります：

  cd scripts
  docker build --platform linux/amd64 -t {ecr_repo_name} .
  docker tag {ecr_repo_name}:latest {account_id}.dkr.ecr.{region}.amazonaws.com/{ecr_repo_name}:latest
  docker push {account_id}.dkr.ecr.{region}.amazonaws.com/{ecr_repo_name}:latest
  cd ..
  

ステップ 5：IAM ロールを作成する

references/iam-roles.md の詳細な手順に従って、すべての IAM ロール（Lambda、ECS タスク実行、ECS タスク、Step Functions、EventBridge の各ロール）を作成します。

• IAM ロールの伝播を待つため、少なくとも 10 秒待機する必要があります

ステップ 6：Lambda 関数を作成する

制約：

• 関数コードは scripts/lambda_function.py にあります

• パッケージ化と関数の作成を行う前に、スキルのルートディレクトリに移動する必要があります

• 次のコマンドでパッケージ化する必要があります：python3 -c "import zipfile,io; z=io.BytesIO(); f=zipfile.ZipFile(z,'w'); f.writestr('lambda_function.py', open('scripts/lambda_function.py').read()); f.close(); open('/tmp/lambda_function.zip','wb').write(z.getvalue())"

• 次のコマンドで関数を作成する必要があります：

  aws lambda create-function \
      --function-name sfn-file-processor \
      --runtime python3.12 \
      --handler lambda_function.lambda_handler \
      --role arn:aws:iam::{account_id}:role/sfn-lambda-role \
      --zip-file fileb:///tmp/lambda_function.zip \
      --timeout 60 \
      --architectures x86_64 \
      --region {region}
  

• 次のコマンドで関数が作成されたことを確認する必要があります： aws lambda get-function --function-name sfn-file-processor --region {region}

ステップ 7：CloudWatch ロググループを作成する

制約：

• 次のコマンドで Fargate 用のロググループを作成する必要があります： aws logs create-log-group --log-group-name /StepFunctionFargateTask --region {region}
• 次のコマンドでロググループを KMS キーで暗号化する必要があります： aws logs associate-kms-key --log-group-name /StepFunctionFargateTask --kms-key-arn {kms_key_arn} --region {region}

ステップ 8：ECS クラスターとタスク定義を作成する

references/ecs-task-definition.md の詳細な手順に従って、ECS クラスターを作成し、Fargate タスク定義を登録します。

• レスポンスからタスク定義 ARN を記録する必要があります

ステップ 9：EventBridge 通知を設定して S3 バケットを作成する

制約：

• 次のコマンドでバケットを作成する必要があります。 aws s3api create-bucket --bucket {bucket_name} --region {region} --create-bucket-configuration LocationConstraint={region}
• リージョンが us-east-1 の場合は、--create-bucket-configuration を指定してはなりません
• 次のコマンドで、バケットの EventBridge 通知を有効にする必要があります。 aws s3api put-bucket-notification-configuration --bucket {bucket_name} --notification-configuration '{"EventBridgeConfiguration": {}}' --region {region}
• 次のコマンドで、バケットのデフォルト暗号化を有効にする必要があります。 aws s3api put-bucket-encryption --bucket {bucket_name} --server-side-encryption-configuration '{"Rules":[{"ApplyServerSideEncryptionByDefault":{"SSEAlgorithm":"aws:kms"}}]}' --region {region}

ステップ 10：Step Functions ステートマシンを作成する

制約：

• ステートマシンの定義は scripts/statemachine.asl.json にあります

• 作業用コピーを作成し、すべてのプレースホルダーを置き換える必要があります。

  sed -e 's|${LambdaFunction}|arn:aws:lambda:{region}:{account_id}:function:sfn-file-processor|g' \
      -e 's|${Cluster}|arn:aws:ecs:{region}:{account_id}:cluster/sfn-cluster|g' \
      -e 's|${TaskDefinition}|{task_definition_arn}|g' \
      -e 's|${Subnet1}|{subnet1_id}|g' \
      -e 's|${Subnet2}|{subnet2_id}|g' \
      -e 's|${SecurityGroup}|{sg_id}|g' \
      scripts/statemachine.asl.json > /tmp/statemachine.asl.json
  

• 次のコマンドでステートマシンを作成する必要があります。

  aws stepfunctions create-state-machine \
      --name {state_machine_name} \
      --definition file:///tmp/statemachine.asl.json \
      --role-arn arn:aws:iam::{account_id}:role/sfn-state-machine-role \
      --type STANDARD \
      --region {region}
  

• 応答から stateMachineArn を取得する必要があります

ステップ 11：EventBridge ルールを作成する

制約：

• S3 オブジェクトの作成をトリガーとする EventBridge ルールを作成する必要があります。

  aws events put-rule \
      --name s3-to-stepfunctions \
      --event-pattern '{
        "source": ["aws.s3"],
        "detail-type": ["Object Created"],
        "detail": {
          "bucket": {
            "name": ["{bucket_name}"]
          }
        }
      }' \
      --region {region}
  

• ステートマシンをターゲットとして追加する必要があります。

  aws events put-targets \
      --rule s3-to-stepfunctions \
      --targets '[{
        "Id": "StepFunctionsTarget",
        "Arn": "{state_machine_arn}",
        "RoleArn": "arn:aws:iam::{account_id}:role/sfn-eventbridge-role"
      }]' \
      --region {region}
  

ステップ 12：モニタリングを設定する

制約：

• EventBridge の呼び出しに失敗した場合に備えて、デッドレターキューを作成する必要があります。 aws sqs create-queue --queue-name s3-to-stepfunctions-dlq --region {region}

• DLQ を関連付けるために、EventBridge ターゲットを更新する必要があります。

  aws events put-targets \
      --rule s3-to-stepfunctions \
      --targets '[{
        "Id": "StepFunctionsTarget",
        "Arn": "{state_machine_arn}",
        "RoleArn": "arn:aws:iam::{account_id}:role/sfn-eventbridge-role",
        "DeadLetterConfig": {
          "Arn": "arn:aws:sqs:{region}:{account_id}:s3-to-stepfunctions-dlq"
        }
      }]' \
      --region {region}
  

• Step Functions の実行失敗に対する CloudWatch アラームを作成する必要があります。 aws cloudwatch put-metric-alarm --alarm-name sfn-execution-failures --metric-name ExecutionsFailed --namespace AWS/States --statistic Sum --period 300 --threshold 1 --comparison-operator GreaterThanOrEqualToThreshold --evaluation-periods 1 --dimensions Name=StateMachineArn,Value={state_machine_arn} --region {region}

ステップ 13：検証する

制約：

• Lambda の処理を確認するため、小さなファイル（6 MB 未満）でテストする必要があります。

  echo 'test data' > /tmp/small-file.txt
  aws s3 cp /tmp/small-file.txt s3://{bucket_name}/small-file.txt --region {region}
  

• 15 秒待ってから、次のコマンドで Step Functions の実行を確認する必要があります。 aws stepfunctions list-executions --state-machine-arn {state_machine_arn} --region {region}

• 実行が成功し、Lambda にルーティングされたことを確認する必要があります

• 作成したすべてのリソースの概要を提示する必要があります。これには、VPC ID、サブネット ID、セキュリティグループ ID、ECR リポジトリ URI、ECS クラスター ARN、タスク定義 ARN、Lambda 関数 ARN、ステートマシン ARN、バケット名、EventBridge ルール名を含めてください

トラブルシューティング

EventBridge ルールがトリガーされない

• バケットで EventBridge 通知が有効になっていることを確認します。aws s3api get-bucket-notification-configuration --bucket {bucket_name}
• ルールが存在することを確認します。aws events describe-rule --name s3-to-stepfunctions --region {region}
• ターゲットに正しいステートマシン ARN とロールが設定されていることを確認します

Fargate タスクで Step Functions の実行が失敗する

• コンテナイメージが ECR に存在することを確認します。aws ecr describe-images --repository-name {ecr_repo_name} --region {region}
• サブネットからインターネットにアクセスできることを確認します（IGW を含むルートテーブル）
• セキュリティグループでアウトバウンドトラフィックが許可されていることを確認します
• /StepFunctionFargateTask の CloudWatch Logs を確認します

Lambda の呼び出しに失敗する

• CloudWatch Logs を確認します。aws logs tail /aws/lambda/sfn-file-processor --region {region}
• Step Functions ロールに lambda:InvokeFunction のアクセス許可があることを確認します

IAM PassRole エラー

• Step Functions ロールには、ECS 実行ロールとタスクロールの両方の ARN に対する iam:PassRole が必要です

Fargate タスクが PROVISIONING のまま進まない

• サブネットでパブリック IP の自動割り当てが有効になっていることを確認します
• インターネットゲートウェイがアタッチされ、ルートテーブルに 0.0.0.0/0 のルートがあることを確認します

セキュリティに関する考慮事項

• パブリック IP を持つ Fargate タスクはインターネットに公開されます。デフォルトの全許可の送信ルールを取り消し、送信先を限定して設定してください。aws ec2 revoke-security-group-egress --group-id {sg_id} --ip-permissions IpProtocol=-1,IpRanges='[{CidrIp=0.0.0.0/0}]' を実行してから、aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol tcp --port 443 --cidr 0.0.0.0/0 と aws ec2 authorize-security-group-egress --group-id {sg_id} --protocol udp --port 53 --cidr 0.0.0.0/0 を追加してください。本番環境では、インターネット経由のトラフィックの代わりに、S3 と CloudWatch Logs に VPC エンドポイントを使用することを検討してください。
• ECR にプッシュする前に、コンテナイメージの脆弱性をスキャンしてください。次のコマンドで ECR イメージスキャンを有効にします。aws ecr put-image-scanning-configuration --repository-name {ecr_repo_name} --image-scanning-configuration scanOnPush=true --region {region}
• 認証情報には IAM ロールを使用してください。コンテナコードにアクセスキーをハードコードしないでください。
• S3 バケットの保存時の暗号化を有効にします。aws s3api put-bucket-encryption --bucket {bucket_name} --server-side-encryption-configuration '{"Rules":[{"ApplyServerSideEncryptionByDefault":{"SSEAlgorithm":"aws:kms"}}]}'
• Fargate コンテナログの CloudWatch Logs 暗号化を有効にします。aws logs associate-kms-key --log-group-name /StepFunctionFargateTask --kms-key-arn <KMS_KEY_ARN>
• 呼び出しが失敗した場合に備えて、EventBridge ルールにデッドレターキューを設定してください。
• 運用状況を把握できるよう、Step Functions の実行失敗に対する CloudWatch アラームを設定してください。

バージョン情報

• AWS CLI：2.x
• Python ランタイム：3.12
• 最終検証日：2026-04-27

関連リソース

• Step Functions 開発者ガイド
• EventBridge の S3 イベント
• Fargate タスク定義
• ECR へのイメージのプッシュ
• Step Functions と Fargate の統合