Kryιk

実践LESSON 19

この教材で学ぶこと Lambdaを中心に、非同期処理と障害対策を設計します。

AWS サーバーレス — Lambda・API Gateway・Step Functions

対象読者: Lambdaの基本操作から、設計とトラブル対処へ理解を進めたい人。

1. Lambda の内部動作（コールドスタートを理解する）

リクエスト 1（コールドスタート）:
  [AWS インフラ確保] → [コンテナ起動] → [ランタイム初期化] → [ハンドラ実行]
  ↑ここが遅い（数百ms〜数秒）

リクエスト 2（ウォームスタート）:
  [既存コンテナ再利用] → [ハンドラ実行]
  ↑速い（ミリ秒単位）

対策:
  1. Provisioned Concurrency（常に起動済みコンテナを確保）
     → コスト増。本番の重要エンドポイントに使う
  2. ハンドラ外で初期化（DB接続など）
  3. Lambda SnapStart（Java 向け。スナップショット起動）

2. Lambda ハンドラの書き方

import json
import boto3
import os
from typing import Any

# ✅ ハンドラ外で初期化（ウォームスタート時に再利用される）
s3 = boto3.client("s3")
TABLE_NAME = os.environ["TABLE_NAME"]

def handler(event: dict, context: Any) -> dict:
    """
    event の構造はトリガーによって変わる:
      - API Gateway: {"body": "...", "headers": {...}, "pathParameters": {...}}
      - SQS:         {"Records": [{"body": "...", "receiptHandle": "..."}]}
      - EventBridge: {"source": "myapp", "detail-type": "OrderPlaced", "detail": {...}}
    """
    # ログ（CloudWatch Logs に出力）
    print(json.dumps({"event": event, "requestId": context.aws_request_id}))

    try:
        # メイン処理
        result = process(event)
        return {
            "statusCode": 200,
            "body": json.dumps(result),
            "headers": {"Content-Type": "application/json"}
        }
    except ValueError as e:
        return {"statusCode": 400, "body": json.dumps({"error": str(e)})}
    except Exception as e:
        # 予期しないエラーは再スロー → Lambda がリトライ
        raise

def process(event: dict) -> dict:
    body = json.loads(event.get("body", "{}"))
    return {"message": "OK", "data": body}

3. 同時実行数とスロットリング

Lambda の同時実行制限:
  - リージョンごとのデフォルト上限: 1000
  - 1関数に予約枠を設定可能（Reserved Concurrency）

スロットリングが起こる条件:
  同時実行数がリージョン上限 or 関数の予約枠を超えた場合

エラー: TooManyRequestsException (HTTP 429)

対策:
  SQS → Lambda: SQS がバッファリング → スロットリングを自動リトライ
  API Gateway → Lambda: クライアントに 429 を返す → クライアント側でリトライ実装が必要
# 予約済み同時実行数の設定
aws lambda put-function-concurrency \
  --function-name myapp \
  --reserved-concurrent-executions 50

# Provisioned Concurrency（コールドスタート対策）
aws lambda put-provisioned-concurrency-config \
  --function-name myapp \
  --qualifier production \
  --provisioned-concurrent-executions 5

4. API Gateway + Lambda（REST API vs HTTP API）

REST API:
  ✅ 高度なリクエスト変換（マッピングテンプレート）
  ✅ API キー・使用量プラン
  ✅ キャッシュ
  ❌ 高コスト ($3.5/百万リクエスト)
  → 複雑な要件

HTTP API:
  ✅ 低コスト ($1.0/百万リクエスト)
  ✅ JWT 認証（Cognito/Auth0 との連携が容易）
  ✅ 低レイテンシ
  → 新規では基本こちらを選ぶ
# SAM (Serverless Application Model) でのシンプルな定義
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Resources:
  MyApi:
    Type: AWS::Serverless::Api
    Properties:
      StageName: production
      Auth:
        DefaultAuthorizer: CognitoAuthorizer
        Authorizers:
          CognitoAuthorizer:
            UserPoolArn: !GetAtt UserPool.Arn

  MyFunction:
    Type: AWS::Serverless::Function
    Properties:
      CodeUri: ./src/
      Handler: handler.handler
      Runtime: python3.12
      Environment:
        Variables:
          TABLE_NAME: !Ref DynamoTable
      Events:
        GetItem:
          Type: Api
          Properties:
            RestApiId: !Ref MyApi
            Path: /items/{id}
            Method: GET
      Policies:
        - DynamoDBReadPolicy:
            TableName: !Ref DynamoTable

5. Step Functions — ワークフロー設計

用途: 複数のLambdaを組み合わせた複雑な処理フロー
例: 注文処理 → 在庫確認 → 決済 → 配送手配 → 通知

Express vs Standard:
  Express: 高スループット・短時間(5分以内)・価格安
  Standard: 長時間(1年まで)・待機中は課金なし・イベント履歴
{
  "Comment": "注文処理ワークフロー",
  "StartAt": "CheckInventory",
  "States": {
    "CheckInventory": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:check-inventory",
      "Next": "ProcessPayment",
      "Catch": [{
        "ErrorEquals": ["InsufficientStock"],
        "Next": "NotifyOutOfStock"
      }]
    },
    "ProcessPayment": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:process-payment",
      "Retry": [{
        "ErrorEquals": ["PaymentTimeout"],
        "IntervalSeconds": 2,
        "MaxAttempts": 3,
        "BackoffRate": 2.0
      }],
      "Next": "ParallelShipping"
    },
    "ParallelShipping": {
      "Type": "Parallel",
      "Branches": [
        {"StartAt": "ArrangeDelivery", "States": {"ArrangeDelivery": {"Type": "Task", "Resource": "...", "End": true}}},
        {"StartAt": "SendConfirmEmail", "States": {"SendConfirmEmail": {"Type": "Task", "Resource": "...", "End": true}}}
      ],
      "End": true
    },
    "NotifyOutOfStock": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:...:function:notify-user",
      "End": true
    }
  }
}

6. Lambda の権限設計

2種類のロール:
  1. 実行ロール（Execution Role）: Lambda が他の AWS サービスを呼ぶための権限
  2. リソースベースポリシー: 他のサービスが Lambda を呼び出す権限

例: API Gateway → Lambda を許可
# API Gateway が Lambda を実行できるようにする（自動設定されるが確認方法）
aws lambda get-policy --function-name myapp

# Lambdaが S3・DynamoDB・Secrets Manager を使えるように
# → IAM ロールに以下のポリシーをアタッチ
{
  "Effect": "Allow",
  "Action": ["s3:GetObject", "s3:PutObject"],
  "Resource": "arn:aws:s3:::my-bucket/*"
}

7. 観測性（Observability）の設計

# Lambda Powertools（Python）で構造化ログ・トレース
from aws_lambda_powertools import Logger, Tracer, Metrics
from aws_lambda_powertools.metrics import MetricUnit

logger = Logger()
tracer = Tracer()
metrics = Metrics(namespace="MyApp")

@logger.inject_lambda_context
@tracer.capture_lambda_handler
@metrics.log_metrics
def handler(event, context):
    metrics.add_metric(name="OrderProcessed", unit=MetricUnit.Count, value=1)
    logger.info("Processing order", order_id=event["orderId"])
    
    with tracer.capture_method:
        result = heavy_computation()
    
    return result

🔗 次のステップ