【Claude CodeとAWSで作るAIエージェントの土台 第10回】ECS から Lambda へ、同じコンテナのまま移す：切り替えの手順と3つの落とし穴

# infra/stacks/api_stack.py
"""API を動かす部分（小さく始める構成）。

compute で、API を何で動かすかを選ぶ（同じコンテナを使う）
- ecs：ALB＋ECS Fargate（第4回）。常に動く。確認用は running=false で止める（第8回）
- lambda：API Gateway＋Lambda。呼ばれたときだけ動く。VPC・ALB・NAT は作らない
- both：移す間だけ両方を作る。DNS が向いている方が使われる

共通
- トークンの表（DynamoDB）：ハッシュだけを置く。どちらからも同じ表を読む
- WordPress のパスワード：Secrets Manager の <prefix>/wp/<サイト> を API だけが読む（CDK では作らない）
- WAF（設定で）：トークンなしの呼び出しは 5分で100回、全体は 5分で3,000回まで。ALB にも API Gateway にも付ける
- 監視：トークン違い（401）が 5分で20回を超えたら知らせる（ECS・Lambda の記録を合わせて数える）

ECS 側の部品は、名前（ID）を第8回から変えていない。変えると作り直しになり、ALB の DNS 名が変わるため。
"""

from pathlib import Path

import aws_cdk as cdk
from aws_cdk import aws_apigateway as apigw
from aws_cdk import aws_certificatemanager as acm
from aws_cdk import aws_cloudwatch as cw
from aws_cdk import aws_cloudwatch_actions as cw_actions
from aws_cdk import aws_dynamodb as ddb
from aws_cdk import aws_ec2 as ec2
from aws_cdk import aws_ecr_assets as ecr_assets
from aws_cdk import aws_ecs as ecs
from aws_cdk import aws_elasticloadbalancingv2 as elbv2
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as lambda_
from aws_cdk import aws_logs as logs
from aws_cdk import aws_sns as sns
from aws_cdk import aws_sns_subscriptions as subs
from aws_cdk import aws_wafv2 as wafv2
from constructs import Construct

CODE_DIR = Path(__file__).resolve().parents[2]  # Dockerfile のある場所
PORT = 8080
COMPUTE = ("ecs", "lambda", "both")


class ApiStack(cdk.Stack):
    def __init__(
        self,
        scope: Construct,
        cid: str,
        *,
        stage: str,
        domain_name: str,
        certificate_arn: str,
        waf_enabled: bool,
        running: bool,
        alarm_email: str = "",
        compute: str = "ecs",
        **kwargs,
    ):
        super().__init__(scope, cid, **kwargs)
        if not certificate_arn:
            raise ValueError('HTTPS の証明書の ARN を -c certificateArn=... で渡してください（ACM で先に作っておく）')
        if compute not in COMPUTE:
            raise ValueError(f"compute は {' / '.join(COMPUTE)} のどれかにしてください（今は {compute}）")
        prefix = f"ai-agent-kiban/{stage}"
        cert = acm.Certificate.from_certificate_arn(self, "Cert", certificate_arn)

        # --- トークンの表 ---
        tokens = ddb.TableV2(
            self,
            "Tokens",
            partition_key=ddb.Attribute(name="token_hash", type=ddb.AttributeType.STRING),
            removal_policy=cdk.RemovalPolicy.RETAIN,
        )
        env = {"STAGE": stage, "TOKENS_TABLE": tokens.table_name, "SECRET_PREFIX": prefix}

        def grant(role: iam.IRole):
            """API の権限：トークンの表を読む・自分の環境のパスワードを読む、だけ（ECS と Lambda で同じ）"""
            tokens.grant_read_data(role)
            role.add_to_principal_policy(
                iam.PolicyStatement(
                    actions=["secretsmanager:GetSecretValue"],
                    resources=[f"arn:{self.partition}:secretsmanager:{self.region}:{self.account}:secret:{prefix}/*"],
                )
            )

        acl = self._web_acl(cid) if waf_enabled else None
        log_groups = {}
        if compute in ("ecs", "both"):
            log_groups["AuthFailed"] = self._ecs(domain_name, cert, env, grant, running, acl)
        if compute in ("lambda", "both"):
            log_groups["AuthFailedLambda"] = self._lambda(domain_name, cert, env, grant, acl)

        # --- 監視：トークン違い（401）が続いたら知らせる（ECS・Lambda の記録を同じ数字に足す） ---
        for filter_id, log_group in log_groups.items():
            logs.MetricFilter(
                self,
                filter_id,
                log_group=log_group,
                filter_pattern=logs.FilterPattern.string_value("$.event", "=", "auth_failed"),
                metric_namespace="ai-agent-kiban",
                metric_name=f"{stage}-auth-failed",
                metric_value="1",
            )
        topic = sns.Topic(self, "Alarms")
        if alarm_email:
            topic.add_subscription(subs.EmailSubscription(alarm_email))
        cw.Alarm(
            self,
            "AuthFailedAlarm",
            alarm_description="トークン違い（401）が続いている",
            metric=cw.Metric(
                namespace="ai-agent-kiban",
                metric_name=f"{stage}-auth-failed",
                period=cdk.Duration.minutes(5),
                statistic="Sum",
            ),
            threshold=20,
            evaluation_periods=1,
            comparison_operator=cw.ComparisonOperator.GREATER_THAN_THRESHOLD,
            treat_missing_data=cw.TreatMissingData.NOT_BREACHING,
        ).add_alarm_action(cw_actions.SnsAction(topic))

        cdk.CfnOutput(
            self, "TokensTable", value=tokens.table_name, description="tools/issue_token.py の --table に渡す"
        )

    def _ecs(self, domain_name, cert, env, grant, running, acl) -> logs.LogGroup:
        """ALB＋ECS Fargate（第4回・第8回と同じ。ID も同じ）"""
        # --- ネットワーク（NAT なし） ---
        vpc = ec2.Vpc(
            self,
            "Vpc",
            max_azs=2,
            nat_gateways=0,
            subnet_configuration=[ec2.SubnetConfiguration(name="public", subnet_type=ec2.SubnetType.PUBLIC)],
        )
        # DynamoDB へはインターネットを通らない無料の出入口（ゲートウェイ型の VPC エンドポイント）で行く
        vpc.add_gateway_endpoint("DynamoDb", service=ec2.GatewayVpcEndpointAwsService.DYNAMODB)

        # --- API（ECS Fargate） ---
        cluster = ecs.Cluster(self, "Cluster", vpc=vpc)
        task = ecs.FargateTaskDefinition(self, "Task", cpu=256, memory_limit_mib=512)
        logs_group = logs.LogGroup(self, "Logs", retention=logs.RetentionDays.THREE_MONTHS)
        task.add_container(
            "api",
            image=ecs.ContainerImage.from_asset(str(CODE_DIR)),
            port_mappings=[ecs.PortMapping(container_port=PORT)],
            logging=ecs.LogDrivers.aws_logs(stream_prefix="api", log_group=logs_group),
            environment=env,
        )
        grant(task.task_role)

        alb_sg = ec2.SecurityGroup(self, "AlbSg", vpc=vpc, allow_all_outbound=False)
        alb_sg.add_ingress_rule(ec2.Peer.any_ipv4(), ec2.Port.tcp(443))
        alb_sg.add_ingress_rule(ec2.Peer.any_ipv4(), ec2.Port.tcp(80))
        task_sg = ec2.SecurityGroup(self, "TaskSg", vpc=vpc, allow_all_outbound=True)
        task_sg.add_ingress_rule(alb_sg, ec2.Port.tcp(PORT), "from ALB only")
        alb_sg.add_egress_rule(task_sg, ec2.Port.tcp(PORT))

        service = ecs.FargateService(
            self,
            "Service",
            cluster=cluster,
            task_definition=task,
            desired_count=1 if running else 0,
            assign_public_ip=True,
            security_groups=[task_sg],
            circuit_breaker=ecs.DeploymentCircuitBreaker(enable=True, rollback=True),
        )

        # --- 入口（ALB） ---
        alb = elbv2.ApplicationLoadBalancer(
            self, "Alb", vpc=vpc, internet_facing=True, security_group=alb_sg, drop_invalid_header_fields=True
        )
        listener = alb.add_listener(
            "Https", port=443, open=False, ssl_policy=elbv2.SslPolicy.RECOMMENDED_TLS, certificates=[cert]
        )
        listener.add_targets(
            "Api",
            port=PORT,
            protocol=elbv2.ApplicationProtocol.HTTP,
            targets=[service],
            health_check=elbv2.HealthCheck(path="/healthz"),
        )
        alb.add_listener(
            "Http",
            port=80,
            open=False,
            default_action=elbv2.ListenerAction.redirect(protocol="HTTPS", port="443", permanent=True),
        )
        if acl:
            wafv2.CfnWebACLAssociation(self, "WafAlb", resource_arn=alb.load_balancer_arn, web_acl_arn=acl.attr_arn)

        cdk.CfnOutput(
            self,
            "AlbDnsName",
            value=alb.load_balancer_dns_name,
            description=f"DNS で {domain_name} の CNAME の向け先にする（ECS で動かすとき）",
        )
        return logs_group

    def _lambda(self, domain_name, cert, env, grant, acl) -> logs.LogGroup:
        """API Gateway（REST API）＋Lambda。ECS と同じコンテナを、Lambda Web Adapter で動かす"""
        # 反映に失敗して作り直すとき、残った記録の置き場と重ならないよう、スタックと一緒に消す
        logs_group = logs.LogGroup(
            self, "FnLogs", retention=logs.RetentionDays.THREE_MONTHS, removal_policy=cdk.RemovalPolicy.DESTROY
        )
        fn = lambda_.DockerImageFunction(
            self,
            "Fn",
            # Lambda は1つの CPU の種類に向けたイメージしか読めない。x86_64 に固定して作る
            code=lambda_.DockerImageCode.from_image_asset(str(CODE_DIR), platform=ecr_assets.Platform.LINUX_AMD64),
            architecture=lambda_.Architecture.X86_64,
            memory_size=1024,  # メモリを増やすと CPU も増え、起動（コールドスタート）が速くなる
            timeout=cdk.Duration.seconds(30),  # API Gateway の上限（29秒）より少し長く
            environment=env,  # AWS_REGION は Lambda が自分で入れる
            log_group=logs_group,
        )
        grant(fn.role)

        # 入口：WAF を付けられる REST API を使う（HTTP API には付けられない）
        api = apigw.LambdaRestApi(
            self,
            "Gateway",
            handler=fn,
            proxy=True,
            endpoint_types=[apigw.EndpointType.REGIONAL],
            binary_media_types=["*/*"],  # 画像などもそのまま通す
            deploy_options=apigw.StageOptions(
                stage_name="api", throttling_rate_limit=20, throttling_burst_limit=40
            ),  # 1秒あたりの上限（超えた分は 429）
            cloud_watch_role=False,
        )
        # 独自ドメイン：ALB と同じ証明書を使う（同じリージョンの ACM の証明書）
        domain = apigw.DomainName(
            self,
            "Domain",
            domain_name=domain_name,
            certificate=cert,
            endpoint_type=apigw.EndpointType.REGIONAL,
            security_policy=apigw.SecurityPolicy.TLS_1_2,
        )
        domain.add_base_path_mapping(api, stage=api.deployment_stage)
        if acl:
            wafv2.CfnWebACLAssociation(
                self, "WafGateway", resource_arn=api.deployment_stage.stage_arn, web_acl_arn=acl.attr_arn
            )

        cdk.CfnOutput(
            self,
            "ApiGatewayDomainTarget",
            value=domain.domain_name_alias_domain_name,
            description=f"DNS で {domain_name} の CNAME の向け先にする（Lambda に切り替えるとき）",
        )
        cdk.CfnOutput(
            self,
            "ApiGatewayUrl",
            value=api.url,
            description="DNS を切り替える前に確かめる用（末尾に healthz を付ける）",
        )
        return logs_group

    def _web_acl(self, cid) -> wafv2.CfnWebACL:
        """WAF（第8回と同じ決まり）。同じ1つを ALB にも API Gateway にも付けられる"""

        def visibility(name):
            return wafv2.CfnWebACL.VisibilityConfigProperty(
                cloud_watch_metrics_enabled=True, metric_name=f"{cid}-{name}", sampled_requests_enabled=True
            )

        def rate(name, priority, limit, scope_down=None):
            return wafv2.CfnWebACL.RuleProperty(
                name=name,
                priority=priority,
                action=wafv2.CfnWebACL.RuleActionProperty(block={}),
                statement=wafv2.CfnWebACL.StatementProperty(
                    rate_based_statement=wafv2.CfnWebACL.RateBasedStatementProperty(
                        limit=limit,
                        aggregate_key_type="IP",
                        evaluation_window_sec=300,
                        scope_down_statement=scope_down,
                    )
                ),
                visibility_config=visibility(name),
            )

        without_token = wafv2.CfnWebACL.StatementProperty(
            not_statement=wafv2.CfnWebACL.NotStatementProperty(
                statement=wafv2.CfnWebACL.StatementProperty(
                    byte_match_statement=wafv2.CfnWebACL.ByteMatchStatementProperty(
                        field_to_match=wafv2.CfnWebACL.FieldToMatchProperty(single_header={"Name": "authorization"}),
                        positional_constraint="STARTS_WITH",
                        search_string="Bearer th_",
                        text_transformations=[wafv2.CfnWebACL.TextTransformationProperty(priority=0, type="NONE")],
                    )
                )
            )
        )
        return wafv2.CfnWebACL(
            self,
            "Waf",
            scope="REGIONAL",
            default_action=wafv2.CfnWebACL.DefaultActionProperty(allow={}),
            visibility_config=visibility("all"),
            rules=[rate("rate-without-token", 0, 100, without_token), rate("rate-all", 1, 3000)],
        )