AWS API Gateway

フルマネージドAPI管理サービス。RESTおよびWebSocket API対応、自動スケーリング、モニタリング、セキュリティ機能を統合。サーバーレス対応。

API GatewayAWSサーバーレスLambdaREST APIHTTP APIWebSocketCloudFormation認証モニタリング

概要

AWS API Gatewayは、Amazon Web Servicesが提供するフルマネージドAPI管理サービスです。RESTful APIおよびWebSocket APIの作成、公開、保守、監視、セキュリティ確保を簡単に行えます。サーバーレスアーキテクチャとの親和性が高く、AWS Lambdaとの統合により、完全なサーバーレスAPIソリューションを構築できます。

2015年にリリースされ、AWSエコシステムでの標準的なAPI Gatewayとして広く採用されています。自動スケーリング、モニタリング、セキュリティ機能が統合されており、運用管理コストの削減と開発生産性の向上を実現します。

主要な特徴

フルマネージドサービス: サーバー管理不要の完全マネージドサービス
自動スケーリング: トラフィック量に応じた自動的なスケーリング
AWS統合: Lambda、DynamoDB、S3などとのネイティブ統合
セキュリティ: IAM、Cognito、Lambda Authorizerによる認証・認可
モニタリング: CloudWatch統合による詳細な監視とアラート

主要機能

API タイプ

REST API: 従来のRESTfulサービス、高度な機能セット
HTTP API: 高性能で低コスト、シンプルなHTTPプロキシ
WebSocket API: リアルタイム双方向通信のサポート

コア機能

リクエストルーティング: パス、メソッド、ヘッダーベースルーティング
リクエスト・レスポンス変換: JSON/XMLの変換とマッピング
認証・認可: 複数の認証方式の統合サポート
レート制限: API キー、使用量プランによるスロットリング
キャッシング: レスポンスキャッシュによる性能向上

統合オプション

Lambda統合: サーバーレス関数の直接実行
HTTP統合: 既存のHTTPエンドポイントへのプロキシ
AWS統合: DynamoDB、S3、SNSなどの直接呼び出し
Mock統合: テスト用のモック応答

インストール・セットアップ

AWS CLI セットアップ

AWS CLI インストール

# AWS CLI v2 インストール（Linux）
curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install

# macOS
brew install awscli

# 設定
aws configure
# AWS Access Key ID: YOUR_ACCESS_KEY
# AWS Secret Access Key: YOUR_SECRET_KEY  
# Default region name: us-east-1
# Default output format: json

# プロファイル確認
aws sts get-caller-identity

AWS SAM CLI インストール

# AWS SAM CLI インストール（Linux）
wget https://github.com/aws/aws-sam-cli/releases/latest/download/aws-sam-cli-linux-x86_64.zip
unzip aws-sam-cli-linux-x86_64.zip -d sam-installation
sudo ./sam-installation/install

# macOS
brew install aws-sam-cli

# 確認
sam --version

Terraform Provider設定

# main.tf
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = "us-east-1"
}

# API Gateway REST API
resource "aws_api_gateway_rest_api" "example" {
  name        = "example-api"
  description = "Example API Gateway"
  
  endpoint_configuration {
    types = ["REGIONAL"]
  }
  
  binary_media_types = [
    "application/octet-stream",
    "image/*"
  ]
}

# API Gateway Deployment
resource "aws_api_gateway_deployment" "example" {
  depends_on = [
    aws_api_gateway_method.example,
    aws_api_gateway_integration.example,
  ]
  
  rest_api_id = aws_api_gateway_rest_api.example.id
  stage_name  = "prod"
  
  lifecycle {
    create_before_destroy = true
  }
}

CloudFormation テンプレート

# api-gateway.yaml
AWSTemplateFormatVersion: '2010-09-09'
Description: 'API Gateway with Lambda integration'

Parameters:
  StageName:
    Type: String
    Default: prod
    Description: API Gateway stage name

Resources:
  # REST API
  ApiGateway:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: !Sub "${AWS::StackName}-api"
      Description: "Example API Gateway"
      EndpointConfiguration:
        Types:
          - REGIONAL
      BinaryMediaTypes:
        - "application/octet-stream"
        - "image/*"
      
  # Resource
  ApiResource:
    Type: AWS::ApiGateway::Resource
    Properties:
      RestApiId: !Ref ApiGateway
      ParentId: !GetAtt ApiGateway.RootResourceId
      PathPart: "users"
      
  # Method
  ApiMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGateway
      ResourceId: !Ref ApiResource
      HttpMethod: GET
      AuthorizationType: AWS_IAM
      RequestParameters:
        method.request.querystring.limit: false
      Integration:
        Type: AWS_PROXY
        IntegrationHttpMethod: POST
        Uri: !Sub
          - arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LambdaArn}/invocations
          - LambdaArn: !GetAtt LambdaFunction.Arn
          
  # Deployment
  ApiDeployment:
    Type: AWS::ApiGateway::Deployment
    DependsOn:
      - ApiMethod
    Properties:
      RestApiId: !Ref ApiGateway
      StageName: !Ref StageName
      StageDescription: !Sub "${StageName} stage"

Outputs:
  ApiGatewayEndpoint:
    Description: "API Gateway endpoint URL"
    Value: !Sub "https://${ApiGateway}.execute-api.${AWS::Region}.amazonaws.com/${StageName}"
    Export:
      Name: !Sub "${AWS::StackName}-ApiEndpoint"

基本的な使い方

REST API の作成

AWS CLI による作成

# REST API 作成
aws apigateway create-rest-api \
  --name "my-api" \
  --description "My REST API" \
  --endpoint-configuration types=REGIONAL

# API ID取得
export API_ID=$(aws apigateway get-rest-apis --query "items[?name=='my-api'].id" --output text)

# ルートリソースID取得
export ROOT_ID=$(aws apigateway get-resources --rest-api-id $API_ID --query "items[?path=='/'].id" --output text)

# リソース作成
aws apigateway create-resource \
  --rest-api-id $API_ID \
  --parent-id $ROOT_ID \
  --path-part users

export RESOURCE_ID=$(aws apigateway get-resources --rest-api-id $API_ID --query "items[?pathPart=='users'].id" --output text)

# メソッド作成
aws apigateway put-method \
  --rest-api-id $API_ID \
  --resource-id $RESOURCE_ID \
  --http-method GET \
  --authorization-type "NONE"

# Lambda統合設定
aws apigateway put-integration \
  --rest-api-id $API_ID \
  --resource-id $RESOURCE_ID \
  --http-method GET \
  --type AWS_PROXY \
  --integration-http-method POST \
  --uri "arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/arn:aws:lambda:us-east-1:123456789012:function:my-function/invocations"

# デプロイ
aws apigateway create-deployment \
  --rest-api-id $API_ID \
  --stage-name prod \
  --description "Production deployment"

Lambda関数の作成

# lambda_function.py
import json
import boto3
from datetime import datetime

def lambda_handler(event, context):
    """
    API Gateway Lambda プロキシ統合用のハンドラー
    """
    
    # リクエスト情報取得
    http_method = event['httpMethod']
    path = event['path']
    query_params = event.get('queryStringParameters', {}) or {}
    headers = event.get('headers', {})
    body = event.get('body')
    
    # CORS ヘッダー
    cors_headers = {
        'Access-Control-Allow-Origin': '*',
        'Access-Control-Allow-Headers': 'Content-Type,X-Amz-Date,Authorization,X-Api-Key,X-Amz-Security-Token',
        'Access-Control-Allow-Methods': 'GET,POST,PUT,DELETE,OPTIONS'
    }
    
    try:
        if http_method == 'OPTIONS':
            # CORS プリフライト
            return {
                'statusCode': 200,
                'headers': cors_headers,
                'body': json.dumps({'message': 'CORS preflight'})
            }
            
        elif http_method == 'GET' and path == '/users':
            # ユーザー一覧取得
            users = get_users(query_params.get('limit', '10'))
            
            return {
                'statusCode': 200,
                'headers': {**cors_headers, 'Content-Type': 'application/json'},
                'body': json.dumps({
                    'users': users,
                    'timestamp': datetime.utcnow().isoformat(),
                    'request_id': context.aws_request_id
                })
            }
            
        elif http_method == 'POST' and path == '/users':
            # ユーザー作成
            user_data = json.loads(body) if body else {}
            user = create_user(user_data)
            
            return {
                'statusCode': 201,
                'headers': {**cors_headers, 'Content-Type': 'application/json'},
                'body': json.dumps({
                    'user': user,
                    'message': 'User created successfully'
                })
            }
            
        else:
            # 未対応のメソッド・パス
            return {
                'statusCode': 404,
                'headers': cors_headers,
                'body': json.dumps({'error': 'Not Found'})
            }
            
    except Exception as e:
        print(f"Error: {str(e)}")
        
        return {
            'statusCode': 500,
            'headers': cors_headers,
            'body': json.dumps({
                'error': 'Internal Server Error',
                'message': str(e)
            })
        }

def get_users(limit):
    """DynamoDBからユーザー一覧を取得"""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('Users')
    
    response = table.scan(Limit=int(limit))
    return response.get('Items', [])

def create_user(user_data):
    """DynamoDBにユーザーを作成"""
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('Users')
    
    user = {
        'id': str(uuid.uuid4()),
        'name': user_data.get('name'),
        'email': user_data.get('email'),
        'created_at': datetime.utcnow().isoformat()
    }
    
    table.put_item(Item=user)
    return user

HTTP API の作成

# HTTP API 作成
aws apigatewayv2 create-api \
  --name my-http-api \
  --protocol-type HTTP \
  --description "High-performance HTTP API"

export HTTP_API_ID=$(aws apigatewayv2 get-apis --query "Items[?Name=='my-http-api'].ApiId" --output text)

# Lambda統合
aws apigatewayv2 create-integration \
  --api-id $HTTP_API_ID \
  --integration-type AWS_PROXY \
  --integration-method POST \
  --integration-uri "arn:aws:apigateway:us-east-1:lambda:path/2015-03-31/functions/arn:aws:lambda:us-east-1:123456789012:function:my-function/invocations" \
  --payload-format-version "2.0"

export INTEGRATION_ID=$(aws apigatewayv2 get-integrations --api-id $HTTP_API_ID --query "Items[0].IntegrationId" --output text)

# ルート作成
aws apigatewayv2 create-route \
  --api-id $HTTP_API_ID \
  --route-key "GET /users" \
  --target "integrations/$INTEGRATION_ID"

# ステージ作成
aws apigatewayv2 create-stage \
  --api-id $HTTP_API_ID \
  --stage-name prod \
  --auto-deploy

設定例

API Gateway + Lambda + DynamoDB 構成

# serverless.yml (Serverless Framework)
service: user-api

provider:
  name: aws
  runtime: python3.9
  region: us-east-1
  stage: ${opt:stage, 'dev'}
  environment:
    USERS_TABLE: ${self:service}-${self:provider.stage}-users
    
  iamRoleStatements:
    - Effect: Allow
      Action:
        - dynamodb:Query
        - dynamodb:Scan
        - dynamodb:GetItem
        - dynamodb:PutItem
        - dynamodb:UpdateItem
        - dynamodb:DeleteItem
      Resource: 
        - "arn:aws:dynamodb:${self:provider.region}:*:table/${self:provider.environment.USERS_TABLE}"

functions:
  api:
    handler: handler.main
    events:
      - http:
          path: /{proxy+}
          method: ANY
          cors: true
      - http:
          path: /
          method: ANY
          cors: true

resources:
  Resources:
    UsersTable:
      Type: AWS::DynamoDB::Table
      Properties:
        TableName: ${self:provider.environment.USERS_TABLE}
        AttributeDefinitions:
          - AttributeName: id
            AttributeType: S
        KeySchema:
          - AttributeName: id
            KeyType: HASH
        BillingMode: PAY_PER_REQUEST

plugins:
  - serverless-python-requirements

Request/Response マッピングテンプレート

// リクエストマッピングテンプレート
{
  "timestamp": "$context.requestTime",
  "requestId": "$context.requestId",
  "httpMethod": "$context.httpMethod",
  "resourcePath": "$context.resourcePath",
  "path": "$input.params('path')",
  "queryString": {
    #foreach($param in $input.params().querystring.keySet())
    "$param": "$util.escapeJavaScript($input.params().querystring.get($param))" #if($foreach.hasNext),#end
    #end
  },
  "headers": {
    #foreach($param in $input.params().header.keySet())
    "$param": "$util.escapeJavaScript($input.params().header.get($param))" #if($foreach.hasNext),#end
    #end
  },
  "body": $input.json('$'),
  "isBase64Encoded": false
}

// レスポンスマッピングテンプレート
#set($inputRoot = $input.path('$'))
{
  "data": $inputRoot,
  "metadata": {
    "timestamp": "$context.responseTime",
    "requestId": "$context.requestId",
    "apiId": "$context.apiId",
    "stage": "$context.stage"
  }
}

VPC 統合設定

# vpc-integration.yaml
Resources:
  VpcLink:
    Type: AWS::ApiGateway::VpcLink
    Properties:
      Name: MyVpcLink
      Description: VPC Link for private resources
      TargetArns:
        - !Ref NetworkLoadBalancer
        
  ApiMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGateway
      ResourceId: !Ref ApiResource
      HttpMethod: GET
      AuthorizationType: NONE
      Integration:
        Type: HTTP_PROXY
        IntegrationHttpMethod: GET
        Uri: http://internal-service.local:8080/{proxy}
        ConnectionType: VPC_LINK
        ConnectionId: !Ref VpcLink
        RequestParameters:
          integration.request.path.proxy: method.request.path.proxy

認証・セキュリティ

Cognito User Pool 認証

User Pool 作成

# Cognito User Pool 作成
aws cognito-idp create-user-pool \
  --pool-name "MyUserPool" \
  --policies "PasswordPolicy={MinimumLength=8,RequireUppercase=true,RequireLowercase=true,RequireNumbers=true,RequireSymbols=true}" \
  --auto-verified-attributes email \
  --username-attributes email

export USER_POOL_ID=$(aws cognito-idp list-user-pools --max-items 10 --query "UserPools[?Name=='MyUserPool'].Id" --output text)

# User Pool Client 作成
aws cognito-idp create-user-pool-client \
  --user-pool-id $USER_POOL_ID \
  --client-name "MyAppClient" \
  --generate-secret \
  --explicit-auth-flows ADMIN_NO_SRP_AUTH USER_PASSWORD_AUTH

export CLIENT_ID=$(aws cognito-idp list-user-pool-clients --user-pool-id $USER_POOL_ID --query "UserPoolClients[0].ClientId" --output text)

API Gateway Authorizer 設定

# cognito-authorizer.yaml
Resources:
  CognitoAuthorizer:
    Type: AWS::ApiGateway::Authorizer
    Properties:
      Name: CognitoUserPoolAuthorizer
      Type: COGNITO_USER_POOLS
      IdentitySource: method.request.header.Authorization
      RestApiId: !Ref ApiGateway
      ProviderARNs:
        - !Sub 
          - arn:aws:cognito-idp:${AWS::Region}:${AWS::AccountId}:userpool/${UserPoolId}
          - UserPoolId: !Ref UserPool
          
  ProtectedMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGateway
      ResourceId: !Ref ApiResource
      HttpMethod: GET
      AuthorizationType: COGNITO_USER_POOLS
      AuthorizerId: !Ref CognitoAuthorizer
      RequestParameters:
        method.request.header.Authorization: true

Lambda Authorizer の実装

# lambda_authorizer.py
import json
import jwt
import time
from jwt.exceptions import InvalidTokenError

def lambda_handler(event, context):
    """
    API Gateway Lambda Authorizer
    """
    
    token = event['authorizationToken']
    method_arn = event['methodArn']
    
    try:
        # JWT トークン検証
        decoded_token = verify_jwt_token(token)
        
        # ユーザー情報取得
        user_id = decoded_token['sub']
        user_role = decoded_token.get('role', 'user')
        
        # 認証成功時のポリシー生成
        policy = generate_policy(user_id, 'Allow', method_arn)
        
        # コンテキスト情報追加
        policy['context'] = {
            'userId': user_id,
            'userRole': user_role,
            'tokenIssuer': decoded_token.get('iss', ''),
            'tokenExpiry': str(decoded_token.get('exp', ''))
        }
        
        return policy
        
    except InvalidTokenError as e:
        print(f"Token validation error: {str(e)}")
        raise Exception('Unauthorized')
        
    except Exception as e:
        print(f"Authorizer error: {str(e)}")
        raise Exception('Unauthorized')

def verify_jwt_token(token):
    """JWT トークンの検証"""
    
    # Bearer プレフィックス除去
    if token.startswith('Bearer '):
        token = token[7:]
    
    # JWKSエンドポイントから公開鍵取得（実装例）
    # 実際には jwks-client などのライブラリを使用
    secret = get_jwt_secret()
    
    decoded_token = jwt.decode(
        token,
        secret,
        algorithms=['RS256'],
        audience='your-api-audience',
        issuer='https://your-auth-provider.com/'
    )
    
    # 追加の検証ロジック
    if decoded_token.get('exp', 0) < time.time():
        raise InvalidTokenError('Token expired')
    
    return decoded_token

def generate_policy(principal_id, effect, resource):
    """IAM ポリシー生成"""
    
    policy = {
        'principalId': principal_id,
        'policyDocument': {
            'Version': '2012-10-17',
            'Statement': [
                {
                    'Action': 'execute-api:Invoke',
                    'Effect': effect,
                    'Resource': resource
                }
            ]
        }
    }
    
    return policy

def get_jwt_secret():
    """JWT検証用シークレット取得"""
    # AWS Secrets Manager から取得する例
    import boto3
    
    client = boto3.client('secretsmanager')
    response = client.get_secret_value(SecretId='jwt-secret')
    return response['SecretString']

API Key による認証

# API Key 作成
aws apigateway create-api-key \
  --name "MyApiKey" \
  --description "API Key for client authentication" \
  --enabled

export API_KEY_ID=$(aws apigateway get-api-keys --query "items[?name=='MyApiKey'].id" --output text)

# Usage Plan 作成
aws apigateway create-usage-plan \
  --name "BasicPlan" \
  --description "Basic usage plan" \
  --throttle burstLimit=100,rateLimit=50 \
  --quota limit=10000,period=MONTH \
  --api-stages apiId=$API_ID,stage=prod

export USAGE_PLAN_ID=$(aws apigateway get-usage-plans --query "items[?name=='BasicPlan'].id" --output text)

# API Key を Usage Plan に関連付け
aws apigateway create-usage-plan-key \
  --usage-plan-id $USAGE_PLAN_ID \
  --key-id $API_KEY_ID \
  --key-type API_KEY

レート制限・トラフィック管理

Usage Plans とAPI Keys

# usage-plans.yaml
Resources:
  BasicUsagePlan:
    Type: AWS::ApiGateway::UsagePlan
    Properties:
      UsagePlanName: "Basic Plan"
      Description: "Basic usage plan with rate limiting"
      ApiStages:
        - ApiId: !Ref ApiGateway
          Stage: !Ref StageName
      Throttle:
        BurstLimit: 200
        RateLimit: 100
      Quota:
        Limit: 10000
        Period: MONTH
        
  PremiumUsagePlan:
    Type: AWS::ApiGateway::UsagePlan
    Properties:
      UsagePlanName: "Premium Plan"
      Description: "Premium usage plan with higher limits"
      ApiStages:
        - ApiId: !Ref ApiGateway
          Stage: !Ref StageName
      Throttle:
        BurstLimit: 1000
        RateLimit: 500
      Quota:
        Limit: 100000
        Period: MONTH
        
  ApiKey:
    Type: AWS::ApiGateway::ApiKey
    Properties:
      Name: "ClientApiKey"
      Description: "API Key for client authentication"
      Enabled: true
      Value: !Sub "${AWS::StackName}-${AWS::AccountId}-key"
      
  UsagePlanKey:
    Type: AWS::ApiGateway::UsagePlanKey
    Properties:
      KeyId: !Ref ApiKey
      KeyType: API_KEY
      UsagePlanId: !Ref BasicUsagePlan

Method レベルでのスロットリング

# method-throttling.yaml
Resources:
  ApiMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ApiGateway
      ResourceId: !Ref ApiResource
      HttpMethod: GET
      AuthorizationType: API_KEY
      ApiKeyRequired: true
      MethodResponses:
        - StatusCode: 200
          ResponseModels:
            application/json: Empty
        - StatusCode: 429
          ResponseModels:
            application/json: Error
      Integration:
        Type: AWS_PROXY
        IntegrationHttpMethod: POST
        Uri: !Sub
          - arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LambdaArn}/invocations
          - LambdaArn: !GetAtt LambdaFunction.Arn
        IntegrationResponses:
          - StatusCode: 200
          - StatusCode: 429
            SelectionPattern: ".*Rate exceeded.*"

  Stage:
    Type: AWS::ApiGateway::Stage
    Properties:
      StageName: !Ref StageName
      RestApiId: !Ref ApiGateway
      DeploymentId: !Ref ApiDeployment
      ThrottleSettings:
        BurstLimit: 500
        RateLimit: 250
      MethodSettings:
        - ResourcePath: "/users"
          HttpMethod: "GET"
          ThrottlingBurstLimit: 100
          ThrottlingRateLimit: 50
        - ResourcePath: "/users"
          HttpMethod: "POST"
          ThrottlingBurstLimit: 50
          ThrottlingRateLimit: 25

WAF 統合

# waf-integration.yaml
Resources:
  WebACL:
    Type: AWS::WAFv2::WebACL
    Properties:
      Name: !Sub "${AWS::StackName}-WebACL"
      Scope: REGIONAL
      DefaultAction:
        Allow: {}
      Rules:
        - Name: RateLimitRule
          Priority: 1
          Statement:
            RateBasedStatement:
              Limit: 2000
              AggregateKeyType: IP
          Action:
            Block: {}
          VisibilityConfig:
            SampledRequestsEnabled: true
            CloudWatchMetricsEnabled: true
            MetricName: RateLimitRule
            
        - Name: GeoBlockRule
          Priority: 2
          Statement:
            GeoMatchStatement:
              CountryCodes:
                - CN
                - RU
          Action:
            Block: {}
          VisibilityConfig:
            SampledRequestsEnabled: true
            CloudWatchMetricsEnabled: true
            MetricName: GeoBlockRule
            
  WebACLAssociation:
    Type: AWS::WAFv2::WebACLAssociation
    Properties:
      ResourceArn: !Sub
        - arn:aws:apigateway:${AWS::Region}::/restapis/${ApiId}/stages/${StageName}
        - ApiId: !Ref ApiGateway
          StageName: !Ref StageName
      WebACLArn: !GetAtt WebACL.Arn

モニタリング・ログ

CloudWatch 統合設定

メトリクスとアラーム

# monitoring.yaml
Resources:
  # CloudWatch アラーム
  HighErrorRateAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub "${AWS::StackName}-HighErrorRate"
      AlarmDescription: "High 4XX/5XX error rate"
      MetricName: 4XXError
      Namespace: AWS/ApiGateway
      Statistic: Sum
      Period: 300
      EvaluationPeriods: 2
      Threshold: 10
      ComparisonOperator: GreaterThanThreshold
      Dimensions:
        - Name: ApiName
          Value: !Ref ApiGateway
        - Name: Stage
          Value: !Ref StageName
      AlarmActions:
        - !Ref SNSTopic
        
  HighLatencyAlarm:
    Type: AWS::CloudWatch::Alarm
    Properties:
      AlarmName: !Sub "${AWS::StackName}-HighLatency"
      AlarmDescription: "High response latency"
      MetricName: Latency
      Namespace: AWS/ApiGateway
      Statistic: Average
      Period: 300
      EvaluationPeriods: 2
      Threshold: 5000
      ComparisonOperator: GreaterThanThreshold
      Dimensions:
        - Name: ApiName
          Value: !Ref ApiGateway
        - Name: Stage
          Value: !Ref StageName
      AlarmActions:
        - !Ref SNSTopic
        
  # SNS Topic
  SNSTopic:
    Type: AWS::SNS::Topic
    Properties:
      TopicName: !Sub "${AWS::StackName}-alerts"
      Subscription:
        - Protocol: email
          Endpoint: [email protected]

Access Logging 設定

# access-logging.yaml
Resources:
  # CloudWatch Log Group
  ApiLogGroup:
    Type: AWS::Logs::LogGroup
    Properties:
      LogGroupName: !Sub "/aws/apigateway/${AWS::StackName}"
      RetentionInDays: 14
      
  # API Gateway Account設定
  ApiGatewayAccount:
    Type: AWS::ApiGateway::Account
    Properties:
      CloudWatchRoleArn: !GetAtt ApiGatewayCloudWatchRole.Arn
      
  # CloudWatch Role
  ApiGatewayCloudWatchRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: apigateway.amazonaws.com
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - arn:aws:iam::aws:policy/service-role/AmazonAPIGatewayPushToCloudWatchLogs
        
  # Stage設定
  Stage:
    Type: AWS::ApiGateway::Stage
    Properties:
      StageName: !Ref StageName
      RestApiId: !Ref ApiGateway
      DeploymentId: !Ref ApiDeployment
      AccessLogSetting:
        DestinationArn: !Sub
          - arn:aws:logs:${AWS::Region}:${AWS::AccountId}:log-group:${LogGroup}
          - LogGroup: !Ref ApiLogGroup
        Format: >
          {
            "requestId": "$context.requestId",
            "requestTime": "$context.requestTime",
            "httpMethod": "$context.httpMethod",
            "path": "$context.path",
            "resourcePath": "$context.resourcePath",
            "status": "$context.status",
            "responseLength": "$context.responseLength",
            "responseLatency": "$context.responseLatency",
            "xrayTraceId": "$context.xrayTraceId",
            "integrationRequestId": "$context.integration.requestId",
            "functionResponseStatus": "$context.integration.status",
            "integrationLatency": "$context.integration.latency",
            "integrationServiceStatus": "$context.integration.integrationStatus",
            "ip": "$context.identity.sourceIp",
            "userAgent": "$context.identity.userAgent",
            "principalId": "$context.authorizer.principalId"
          }
      MethodSettings:
        - ResourcePath: "/*"
          HttpMethod: "*"
          LoggingLevel: INFO
          DataTraceEnabled: true
          MetricsEnabled: true

X-Ray トレーシング

# xray-tracing.yaml
Resources:
  Stage:
    Type: AWS::ApiGateway::Stage
    Properties:
      StageName: !Ref StageName
      RestApiId: !Ref ApiGateway
      DeploymentId: !Ref ApiDeployment
      TracingConfig:
        TracingEnabled: true
      MethodSettings:
        - ResourcePath: "/*"
          HttpMethod: "*"
          TracingEnabled: true

カスタムメトリクス

# custom_metrics.py
import boto3
import json
from datetime import datetime

cloudwatch = boto3.client('cloudwatch')

def lambda_handler(event, context):
    """カスタムメトリクス送信"""
    
    try:
        # ビジネスメトリクス送信
        send_custom_metric('UserRegistrations', 1, 'Count')
        send_custom_metric('ActiveUsers', get_active_user_count(), 'Count')
        
        # API レスポンス
        return {
            'statusCode': 200,
            'body': json.dumps({'message': 'Success'})
        }
        
    except Exception as e:
        # エラーメトリクス送信
        send_custom_metric('ApplicationErrors', 1, 'Count')
        
        return {
            'statusCode': 500,
            'body': json.dumps({'error': str(e)})
        }

def send_custom_metric(metric_name, value, unit):
    """CloudWatch カスタムメトリクス送信"""
    
    cloudwatch.put_metric_data(
        Namespace='MyApp/API',
        MetricData=[
            {
                'MetricName': metric_name,
                'Value': value,
                'Unit': unit,
                'Timestamp': datetime.utcnow(),
                'Dimensions': [
                    {
                        'Name': 'Environment',
                        'Value': 'Production'
                    },
                    {
                        'Name': 'Version',
                        'Value': '1.0'
                    }
                ]
            }
        ]
    )

def get_active_user_count():
    """アクティブユーザー数取得"""
    # DynamoDB から取得する例
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('UserSessions')
    
    response = table.scan(
        FilterExpression='expires_at > :now',
        ExpressionAttributeValues={
            ':now': int(datetime.utcnow().timestamp())
        }
    )
    
    return response['Count']

高度な機能

WebSocket API

# websocket-api.yaml
Resources:
  WebSocketApi:
    Type: AWS::ApiGatewayV2::Api
    Properties:
      Name: !Sub "${AWS::StackName}-websocket"
      ProtocolType: WEBSOCKET
      RouteSelectionExpression: "$request.body.action"
      
  # Connect Route
  ConnectRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref WebSocketApi
      RouteKey: $connect
      AuthorizationType: NONE
      OperationName: ConnectRoute
      Target: !Sub "integrations/${ConnectIntegration}"
      
  ConnectIntegration:
    Type: AWS::ApiGatewayV2::Integration
    Properties:
      ApiId: !Ref WebSocketApi
      Description: Connect Integration
      IntegrationType: AWS_PROXY
      IntegrationUri: !Sub
        - arn:aws:apigateway:${AWS::Region}:lambda:path/2015-03-31/functions/${LambdaArn}/invocations
        - LambdaArn: !GetAtt ConnectFunction.Arn
        
  # Disconnect Route
  DisconnectRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref WebSocketApi
      RouteKey: $disconnect
      AuthorizationType: NONE
      OperationName: DisconnectRoute
      Target: !Sub "integrations/${DisconnectIntegration}"
      
  # Send Message Route
  SendMessageRoute:
    Type: AWS::ApiGatewayV2::Route
    Properties:
      ApiId: !Ref WebSocketApi
      RouteKey: sendMessage
      AuthorizationType: NONE
      OperationName: SendMessageRoute
      Target: !Sub "integrations/${SendMessageIntegration}"
      
  # Deployment
  Deployment:
    Type: AWS::ApiGatewayV2::Deployment
    DependsOn:
      - ConnectRoute
      - DisconnectRoute
      - SendMessageRoute
    Properties:
      ApiId: !Ref WebSocketApi
      
  Stage:
    Type: AWS::ApiGatewayV2::Stage
    Properties:
      StageName: prod
      Description: Production Stage
      DeploymentId: !Ref Deployment
      ApiId: !Ref WebSocketApi

WebSocket Lambda 関数

# websocket_handler.py
import json
import boto3
import logging

logger = logging.getLogger()
logger.setLevel(logging.INFO)

# DynamoDB とAPI Gateway Management API クライアント
dynamodb = boto3.resource('dynamodb')
connections_table = dynamodb.Table('WebSocketConnections')

def lambda_handler(event, context):
    """WebSocket API ハンドラー"""
    
    route_key = event.get('requestContext', {}).get('routeKey')
    connection_id = event.get('requestContext', {}).get('connectionId')
    
    if route_key == '$connect':
        return handle_connect(connection_id, event)
    elif route_key == '$disconnect':
        return handle_disconnect(connection_id)
    elif route_key == 'sendMessage':
        return handle_send_message(connection_id, event)
    else:
        return {'statusCode': 400, 'body': 'Unknown route'}

def handle_connect(connection_id, event):
    """接続ハンドラー"""
    
    try:
        # 接続情報をDynamoDBに保存
        connections_table.put_item(
            Item={
                'connectionId': connection_id,
                'timestamp': int(time.time()),
                'userAgent': event.get('headers', {}).get('User-Agent', ''),
                'sourceIp': event.get('requestContext', {}).get('identity', {}).get('sourceIp')
            }
        )
        
        logger.info(f"Connection {connection_id} established")
        return {'statusCode': 200}
        
    except Exception as e:
        logger.error(f"Connect error: {str(e)}")
        return {'statusCode': 500}

def handle_disconnect(connection_id):
    """切断ハンドラー"""
    
    try:
        # 接続情報をDynamoDBから削除
        connections_table.delete_item(
            Key={'connectionId': connection_id}
        )
        
        logger.info(f"Connection {connection_id} disconnected")
        return {'statusCode': 200}
        
    except Exception as e:
        logger.error(f"Disconnect error: {str(e)}")
        return {'statusCode': 500}

def handle_send_message(connection_id, event):
    """メッセージ送信ハンドラー"""
    
    try:
        body = json.loads(event.get('body', '{}'))
        message = body.get('message', '')
        target = body.get('target', 'all')
        
        if target == 'all':
            # 全接続にブロードキャスト
            broadcast_message(connection_id, message, event)
        else:
            # 特定接続に送信
            send_to_connection(target, message, event)
            
        return {'statusCode': 200}
        
    except Exception as e:
        logger.error(f"Send message error: {str(e)}")
        return {'statusCode': 500}

def broadcast_message(sender_id, message, event):
    """全接続にメッセージブロードキャスト"""
    
    # API Gateway Management API クライアント
    domain_name = event.get('requestContext', {}).get('domainName')
    stage = event.get('requestContext', {}).get('stage')
    api_gateway_management_api = boto3.client(
        'apigatewaymanagementapi',
        endpoint_url=f"https://{domain_name}/{stage}"
    )
    
    try:
        # 全接続取得
        response = connections_table.scan()
        connections = response.get('Items', [])
        
        message_data = {
            'action': 'message',
            'data': {
                'senderId': sender_id,
                'message': message,
                'timestamp': int(time.time())
            }
        }
        
        for connection in connections:
            target_connection_id = connection['connectionId']
            
            # 送信者以外に送信
            if target_connection_id != sender_id:
                try:
                    api_gateway_management_api.post_to_connection(
                        ConnectionId=target_connection_id,
                        Data=json.dumps(message_data)
                    )
                except Exception as e:
                    logger.warning(f"Failed to send to {target_connection_id}: {str(e)}")
                    # 無効な接続は削除
                    connections_table.delete_item(
                        Key={'connectionId': target_connection_id}
                    )
        
    except Exception as e:
        logger.error(f"Broadcast error: {str(e)}")

API Gateway カスタムドメイン

# custom-domain.yaml
Resources:
  # SSL証明書（ACM）
  Certificate:
    Type: AWS::CertificateManager::Certificate
    Properties:
      DomainName: api.example.com
      SubjectAlternativeNames:
        - "*.api.example.com"
      ValidationMethod: DNS
      DomainValidationOptions:
        - DomainName: api.example.com
          HostedZoneId: !Ref HostedZone
          
  # カスタムドメイン
  CustomDomain:
    Type: AWS::ApiGateway::DomainName
    Properties:
      DomainName: api.example.com
      CertificateArn: !Ref Certificate
      EndpointConfiguration:
        Types:
          - REGIONAL
      SecurityPolicy: TLS_1_2
      
  # Base Path Mapping
  BasePathMapping:
    Type: AWS::ApiGateway::BasePathMapping
    Properties:
      DomainName: !Ref CustomDomain
      RestApiId: !Ref ApiGateway
      Stage: !Ref StageName
      
  # Route 53 レコード
  DNSRecord:
    Type: AWS::Route53::RecordSet
    Properties:
      HostedZoneId: !Ref HostedZone
      Name: api.example.com
      Type: A
      AliasTarget:
        DNSName: !GetAtt CustomDomain.RegionalDomainName
        HostedZoneId: !GetAtt CustomDomain.RegionalHostedZoneId

パフォーマンス最適化

キャッシング設定

# caching.yaml
Resources:
  Stage:
    Type: AWS::ApiGateway::Stage
    Properties:
      StageName: !Ref StageName
      RestApiId: !Ref ApiGateway
      DeploymentId: !Ref ApiDeployment
      CacheClusterEnabled: true
      CacheClusterSize: "0.5"
      MethodSettings:
        - ResourcePath: "/users"
          HttpMethod: "GET"
          CachingEnabled: true
          CacheTtlInSeconds: 300
          CacheKeyParameters:
            - "method.request.querystring.limit"
            - "method.request.querystring.offset"
        - ResourcePath: "/users/*"
          HttpMethod: "GET"
          CachingEnabled: true
          CacheTtlInSeconds: 600
          CacheKeyParameters:
            - "method.request.path.id"

Lambda関数の最適化

# optimized_lambda.py
import json
import boto3
import os
from functools import lru_cache

# グローバル変数（接続の再利用）
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table(os.environ['USERS_TABLE'])

# LRUキャッシュでデータベース呼び出しを最適化
@lru_cache(maxsize=128)
def get_user_by_id(user_id):
    """ユーザー情報取得（キャッシュ付き）"""
    response = table.get_item(Key={'id': user_id})
    return response.get('Item')

def lambda_handler(event, context):
    """最適化されたLambdaハンドラー"""
    
    # 早期レスポンス（ヘルスチェック）
    if event.get('path') == '/health':
        return {
            'statusCode': 200,
            'body': json.dumps({'status': 'healthy'})
        }
    
    # リクエスト情報の事前抽出
    http_method = event['httpMethod']
    path = event['path']
    path_parameters = event.get('pathParameters', {}) or {}
    
    try:
        if http_method == 'GET' and path.startswith('/users/'):
            user_id = path_parameters.get('id')
            if user_id:
                user = get_user_by_id(user_id)
                if user:
                    return create_response(200, {'user': user})
                else:
                    return create_response(404, {'error': 'User not found'})
        
        # その他のルーティング処理
        return create_response(404, {'error': 'Not found'})
        
    except Exception as e:
        print(f"Error: {str(e)}")
        return create_response(500, {'error': 'Internal server error'})

def create_response(status_code, body):
    """レスポンス作成ヘルパー"""
    return {
        'statusCode': status_code,
        'headers': {
            'Content-Type': 'application/json',
            'Access-Control-Allow-Origin': '*',
            'Access-Control-Allow-Headers': 'Content-Type,X-Amz-Date,Authorization,X-Api-Key',
            'Access-Control-Allow-Methods': 'GET,POST,PUT,DELETE,OPTIONS'
        },
        'body': json.dumps(body)
    }

Connection Pooling

# connection_pool.py
import boto3
import pymysql
from pymysql.cursors import DictCursor
import os
import json

# RDS Proxy 使用時の設定
RDS_PROXY_ENDPOINT = os.environ['RDS_PROXY_ENDPOINT']
DATABASE_NAME = os.environ['DATABASE_NAME']
USERNAME = os.environ['DB_USERNAME']

# グローバル接続オブジェクト
connection = None

def get_db_connection():
    """データベース接続取得（再利用）"""
    global connection
    
    if connection is None or not connection.open:
        # RDS Proxy経由で接続
        connection = pymysql.connect(
            host=RDS_PROXY_ENDPOINT,
            user=USERNAME,
            database=DATABASE_NAME,
            cursorclass=DictCursor,
            autocommit=True,
            connect_timeout=5
        )
    
    return connection

def lambda_handler(event, context):
    """データベース接続プール使用例"""
    
    try:
        conn = get_db_connection()
        
        with conn.cursor() as cursor:
            cursor.execute("SELECT * FROM users LIMIT 10")
            users = cursor.fetchall()
        
        return {
            'statusCode': 200,
            'body': json.dumps({'users': users})
        }
        
    except Exception as e:
        print(f"Database error: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({'error': 'Database error'})
        }

トラブルシューティング

診断とデバッグ

CloudWatch ログ分析

# ログストリーム確認
aws logs describe-log-streams \
  --log-group-name "/aws/apigateway/my-api" \
  --order-by LastEventTime \
  --descending

# 最新ログ取得
aws logs get-log-events \
  --log-group-name "/aws/apigateway/my-api" \
  --log-stream-name "最新のログストリーム名" \
  --start-from-head

# エラーログフィルタリング
aws logs filter-log-events \
  --log-group-name "/aws/apigateway/my-api" \
  --filter-pattern "ERROR"

API テストとデバッグ

# API テスト（curl）
curl -X GET "https://api-id.execute-api.us-east-1.amazonaws.com/prod/users" \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "X-API-Key: YOUR_API_KEY" \
  -v

# レスポンス時間測定
curl -w "時間: %{time_total}秒\n" \
  -o /dev/null -s \
  "https://api-id.execute-api.us-east-1.amazonaws.com/prod/users"

# ヘッダー詳細確認
curl -I "https://api-id.execute-api.us-east-1.amazonaws.com/prod/users"

よくある問題と解決法

CORS エラー

// JavaScript での CORS 対応
const response = await fetch('https://api-id.execute-api.us-east-1.amazonaws.com/prod/users', {
  method: 'GET',
  headers: {
    'Content-Type': 'application/json',
    'Authorization': 'Bearer ' + token
  },
  mode: 'cors'
});

// Lambda 関数での CORS ヘッダー設定
const corsHeaders = {
  'Access-Control-Allow-Origin': 'https://mydomain.com',
  'Access-Control-Allow-Headers': 'Content-Type,X-Amz-Date,Authorization,X-Api-Key',
  'Access-Control-Allow-Methods': 'GET,POST,PUT,DELETE,OPTIONS',
  'Access-Control-Allow-Credentials': true
};

Lambda タイムアウト問題

# Lambda タイムアウト対策
import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("Function timeout")

def lambda_handler(event, context):
    # タイムアウト設定（関数タイムアウトの90%）
    timeout_seconds = int(context.get_remaining_time_in_millis() / 1000 * 0.9)
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_seconds)
    
    try:
        # メイン処理
        result = process_request(event)
        signal.alarm(0)  # アラーム解除
        return result
        
    except TimeoutError:
        return {
            'statusCode': 503,
            'body': json.dumps({'error': 'Request timeout'})
        }

レート制限エラー対応

# リトライロジック実装
import time
import random
from botocore.exceptions import ClientError

def call_api_with_retry(api_call, max_retries=3):
    """API呼び出しリトライ"""
    
    for attempt in range(max_retries):
        try:
            return api_call()
            
        except ClientError as e:
            error_code = e.response['Error']['Code']
            
            if error_code == 'TooManyRequestsException':
                if attempt < max_retries - 1:
                    # Exponential backoff
                    delay = (2 ** attempt) + random.uniform(0, 1)
                    time.sleep(delay)
                    continue
                else:
                    raise
            else:
                raise
    
    return None

デバッグ用設定

# debug-stage.yaml
Resources:
  DebugStage:
    Type: AWS::ApiGateway::Stage
    Properties:
      StageName: debug
      RestApiId: !Ref ApiGateway
      DeploymentId: !Ref ApiDeployment
      Variables:
        debug: "true"
        loglevel: "DEBUG"
      MethodSettings:
        - ResourcePath: "/*"
          HttpMethod: "*"
          LoggingLevel: INFO
          DataTraceEnabled: true
          MetricsEnabled: true
          ThrottlingBurstLimit: 5000
          ThrottlingRateLimit: 2000

概要