0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@cozy_666_tech
in

AWS CDK入門 — インフラをPythonで書けるのが面白かった

0
Posted at

はじめに

ECSやRDSの設定をAWS CLIやマネジメントコンソールで手作業でやっていたが、「インフラをコードで管理したい」と思うようになってきた。

選択肢はTerraformとAWS CDKが主要どころ。CDKはPythonで書けるということで試してみた。「インフラをPythonのクラスとして定義する」という発想が面白かった。

CDKとTerraformの違い

Terraform:
  HCL（HashiCorp Configuration Language）で書く
  マルチクラウド対応
  Stateファイルで現在の状態を管理
  大きいエコシステム・情報が多い

AWS CDK:
  Python/TypeScript/Javaなど普通のプログラム言語で書く
  AWSのみ（AWS専用）
  CloudFormationを通じてデプロイ
  ループや条件分岐を普通のコードで書ける

Terraformのほうが情報が多く実績もある。CDKはAWSに特化している分、AWSのサービスとの統合がスムーズで、PythonやTypeScriptが書ける人にはとっつきやすい。

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

# CDK CLIをインストール
npm install -g aws-cdk

# バージョン確認
cdk --version

# Python用の依存関係
pip install aws-cdk-lib constructs

# プロジェクトを初期化
mkdir my-infra && cd my-infra
cdk init app --language python

# 仮想環境を有効化
source .venv/bin/activate
pip install -r requirements.txt

生成されるディレクトリ：

my-infra/
├── app.py               # エントリーポイント
├── my_infra/
│   └── my_infra_stack.py # スタックの定義
├── tests/
├── cdk.json             # CDKの設定
├── requirements.txt
└── .venv/

基本概念

App（アプリ）:
  CDKプロジェクトのルート
  複数のStackを持てる

Stack（スタック）:
  デプロイの単位
  CloudFormationのスタックに対応
  例: VPCStack, ECSStack, RDSStack

Construct（コンストラクト）:
  AWSリソースを表すクラス
  Stack内で定義する
  例: VPC, Cluster, DatabaseInstance

最初のスタック — S3バケットを作る

# my_infra/my_infra_stack.py
import aws_cdk as cdk
from aws_cdk import (
    Stack,
    aws_s3 as s3,
    RemovalPolicy,
)
from constructs import Construct

class MyInfraStack(Stack):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        **kwargs,
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # S3バケットの作成
        bucket = s3.Bucket(
            self,
            "MyBucket",
            bucket_name     = "my-app-uploads-2024",
            versioned       = True,
            encryption      = s3.BucketEncryption.S3_MANAGED,
            removal_policy  = RemovalPolicy.RETAIN,  # スタック削除時にバケットを保持
            block_public_access = s3.BlockPublicAccess.BLOCK_ALL,
            lifecycle_rules = [
                s3.LifecycleRule(
                    id      = "DeleteOldVersions",
                    enabled = True,
                    noncurrent_version_expiration = cdk.Duration.days(30),
                )
            ],
        )

        # OutputとしてバケットARNを出力
        cdk.CfnOutput(
            self,
            "BucketArn",
            value       = bucket.bucket_arn,
            description = "S3バケットのARN",
        )

# app.py
import aws_cdk as cdk
from my_infra.my_infra_stack import MyInfraStack

app = cdk.App()

MyInfraStack(
    app,
    "MyInfraStack",
    env = cdk.Environment(
        account = "123456789012",
        region  = "ap-northeast-1",
    ),
)

app.synth()

# CloudFormationテンプレートを生成して確認
cdk synth

# 差分を確認（何が変わるか）
cdk diff

# デプロイ
cdk deploy

# 削除
cdk destroy

VPCとサブネットの構築

# my_infra/network_stack.py
import aws_cdk as cdk
from aws_cdk import (
    Stack,
    aws_ec2 as ec2,
)
from constructs import Construct

class NetworkStack(Stack):
    def __init__(
        self,
        scope: Construct,
        construct_id: str,
        **kwargs,
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # VPCの作成
        self.vpc = ec2.Vpc(
            self,
            "AppVpc",
            ip_addresses    = ec2.IpAddresses.cidr("10.0.0.0/16"),
            max_azs         = 2,
            nat_gateways    = 1,
            subnet_configuration = [
                ec2.SubnetConfiguration(
                    name        = "Public",
                    subnet_type = ec2.SubnetType.PUBLIC,
                    cidr_mask   = 24,
                ),
                ec2.SubnetConfiguration(
                    name        = "Private",
                    subnet_type = ec2.SubnetType.PRIVATE_WITH_EGRESS,
                    cidr_mask   = 24,
                ),
                ec2.SubnetConfiguration(
                    name        = "Isolated",
                    subnet_type = ec2.SubnetType.PRIVATE_ISOLATED,
                    cidr_mask   = 24,
                ),
            ],
        )

        # VPC IDをOutputとして出力
        cdk.CfnOutput(
            self,
            "VpcId",
            value       = self.vpc.vpc_id,
            export_name = "AppVpcId",
        )

self.vpcとして保持することで他のスタックから参照できる。

ECSとFargateの構築

# my_infra/ecs_stack.py
import aws_cdk as cdk
from aws_cdk import (
    Stack,
    aws_ec2      as ec2,
    aws_ecs      as ecs,
    aws_ecs_patterns as ecs_patterns,
    aws_ecr      as ecr,
    aws_iam      as iam,
    aws_logs     as logs,
    aws_secretsmanager as secretsmanager,
)
from constructs import Construct

class EcsStack(Stack):
    def __init__(
        self,
        scope:         Construct,
        construct_id:  str,
        vpc:           ec2.Vpc,
        **kwargs,
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # ECSクラスターの作成
        cluster = ecs.Cluster(
            self,
            "AppCluster",
            cluster_name        = "production",
            vpc                 = vpc,
            container_insights  = True,  # Container Insightsを有効化
        )

        # ECRリポジトリの参照（既存のリポジトリ）
        repository = ecr.Repository.from_repository_name(
            self,
            "FastApiRepo",
            "fastapi-app",
        )

        # Secrets Managerからシークレットを参照
        db_secret = secretsmanager.Secret.from_secret_name_v2(
            self,
            "DbSecret",
            "production/database",
        )

        # タスク定義
        task_definition = ecs.FargateTaskDefinition(
            self,
            "FastApiTaskDef",
            cpu            = 512,
            memory_limit_mib = 1024,
        )

        # S3へのアクセス権限をタスクロールに追加
        task_definition.task_role.add_to_policy(
            iam.PolicyStatement(
                actions   = ["s3:GetObject", "s3:PutObject"],
                resources = ["arn:aws:s3:::my-app-bucket/*"],
            )
        )

        # コンテナの定義
        container = task_definition.add_container(
            "FastApi",
            image  = ecs.ContainerImage.from_ecr_repository(
                repository,
                tag = "latest",
            ),
            logging = ecs.LogDrivers.aws_logs(
                stream_prefix = "ecs",
                log_group     = logs.LogGroup(
                    self,
                    "FastApiLogGroup",
                    log_group_name      = "/ecs/fastapi-app",
                    retention           = logs.RetentionDays.ONE_MONTH,
                    removal_policy      = cdk.RemovalPolicy.DESTROY,
                ),
            ),
            environment = {
                "APP_ENV":   "production",
                "LOG_LEVEL": "info",
            },
            secrets = {
                "DATABASE_URL": ecs.Secret.from_secrets_manager(
                    db_secret,
                    field = "DATABASE_URL",
                ),
            },
            health_check = ecs.HealthCheck(
                command      = ["CMD-SHELL", "curl -f http://localhost:8000/health || exit 1"],
                interval     = cdk.Duration.seconds(30),
                timeout      = cdk.Duration.seconds(5),
                retries      = 3,
                start_period = cdk.Duration.seconds(60),
            ),
        )

        container.add_port_mappings(
            ecs.PortMapping(container_port=8000)
        )

        # ALB + ECSサービスをまとめて作成（ApplicationLoadBalancedFargateService）
        fargate_service = ecs_patterns.ApplicationLoadBalancedFargateService(
            self,
            "FastApiService",
            cluster         = cluster,
            task_definition = task_definition,
            desired_count   = 2,
            public_load_balancer = True,
            task_subnets    = ec2.SubnetSelection(
                subnet_type = ec2.SubnetType.PRIVATE_WITH_EGRESS,
            ),
            health_check_grace_period = cdk.Duration.seconds(60),
        )

        # オートスケーリングの設定
        scaling = fargate_service.service.auto_scale_task_count(
            min_capacity = 2,
            max_capacity = 10,
        )
        scaling.scale_on_cpu_utilization(
            "CpuScaling",
            target_utilization_percent = 70,
            scale_in_cooldown          = cdk.Duration.seconds(60),
            scale_out_cooldown         = cdk.Duration.seconds(30),
        )

        # ALBのDNS名を出力
        cdk.CfnOutput(
            self,
            "LoadBalancerDNS",
            value       = fargate_service.load_balancer.load_balancer_dns_name,
            description = "ALBのDNS名",
        )

ApplicationLoadBalancedFargateServiceというL3コンストラクトを使うとALB + ECSサービスを数行で作れる。CLIで個別に設定していた作業がまとまって書ける。

RDSの構築

# my_infra/rds_stack.py
import aws_cdk as cdk
from aws_cdk import (
    Stack,
    aws_ec2  as ec2,
    aws_rds  as rds,
    aws_secretsmanager as secretsmanager,
)
from constructs import Construct

class RdsStack(Stack):
    def __init__(
        self,
        scope:        Construct,
        construct_id: str,
        vpc:          ec2.Vpc,
        **kwargs,
    ) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # RDSのセキュリティグループ
        db_security_group = ec2.SecurityGroup(
            self,
            "DbSecurityGroup",
            vpc         = vpc,
            description = "RDS security group",
        )

        # ECSのセキュリティグループからのアクセスを許可（後から追加することも可）
        # db_security_group.add_ingress_rule(
        #     ecs_security_group,
        #     ec2.Port.tcp(5432),
        # )

        # Aurora Serverless v2 クラスターの作成
        self.cluster = rds.DatabaseCluster(
            self,
            "AuroraCluster",
            engine = rds.DatabaseClusterEngine.aurora_postgres(
                version = rds.AuroraPostgresEngineVersion.VER_15_4,
            ),
            writer = rds.ClusterInstance.serverless_v2(
                "Writer",
                auto_pause = False,
            ),
            serverless_v2_min_capacity = 0.5,
            serverless_v2_max_capacity = 4.0,
            vpc          = vpc,
            vpc_subnets  = ec2.SubnetSelection(
                subnet_type = ec2.SubnetType.PRIVATE_ISOLATED,
            ),
            security_groups  = [db_security_group],
            default_database_name = "mydb",
            backup           = rds.BackupProps(
                retention        = cdk.Duration.days(7),
                preferred_window = "03:00-04:00",  # JST 12:00-13:00
            ),
            deletion_protection = True,
            removal_policy      = cdk.RemovalPolicy.RETAIN,
            credentials         = rds.Credentials.from_generated_secret(
                "admin",
                secret_name = "production/database",
            ),
        )

        # DB接続情報を出力
        cdk.CfnOutput(
            self,
            "DbEndpoint",
            value       = self.cluster.cluster_endpoint.hostname,
            description = "Auroraクラスターのエンドポイント",
        )

複数スタックの組み合わせ

# app.py
import aws_cdk as cdk
from my_infra.network_stack import NetworkStack
from my_infra.rds_stack     import RdsStack
from my_infra.ecs_stack     import EcsStack

app = cdk.App()

env = cdk.Environment(
    account = "123456789012",
    region  = "ap-northeast-1",
)

# ネットワーク（他のスタックの基盤）
network_stack = NetworkStack(app, "NetworkStack", env=env)

# RDS（VPCに依存）
rds_stack = RdsStack(
    app,
    "RdsStack",
    vpc = network_stack.vpc,
    env = env,
)

# ECS（VPCに依存）
ecs_stack = EcsStack(
    app,
    "EcsStack",
    vpc = network_stack.vpc,
    env = env,
)

# 依存関係を明示
ecs_stack.add_dependency(rds_stack)

app.synth()

# 全スタックを順番にデプロイ
cdk deploy --all

# 特定のスタックだけデプロイ
cdk deploy EcsStack

# 差分を確認
cdk diff EcsStack

環境（dev/staging/prod）の切り替え

# app.py
import aws_cdk as cdk
from my_infra.ecs_stack import EcsStack

app  = cdk.App()
env_name = app.node.try_get_context("env") or "dev"

# 環境ごとの設定
config = {
    "dev": {
        "cpu":           256,
        "memory":        512,
        "desired_count": 1,
        "min_capacity":  1,
        "max_capacity":  2,
    },
    "production": {
        "cpu":           512,
        "memory":        1024,
        "desired_count": 2,
        "min_capacity":  2,
        "max_capacity":  10,
    },
}[env_name]

EcsStack(
    app,
    f"EcsStack-{env_name}",
    config = config,
    env    = cdk.Environment(
        account = "123456789012",
        region  = "ap-northeast-1",
    ),
)

app.synth()

# 開発環境にデプロイ
cdk deploy --context env=dev

# 本番環境にデプロイ
cdk deploy --context env=production

Terraformのworkspaceに相当することをCDKのContextで実現できる。Pythonの辞書で環境ごとの設定を管理できるのが便利。

テストを書く

# tests/test_my_infra_stack.py
import aws_cdk as cdk
from aws_cdk.assertions import Template, Match
from my_infra.my_infra_stack import MyInfraStack

def test_s3_bucket_created():
    app   = cdk.App()
    stack = MyInfraStack(app, "TestStack")

    template = Template.from_stack(stack)

    # S3バケットが作成されることを検証
    template.has_resource_properties(
        "AWS::S3::Bucket",
        {
            "VersioningConfiguration": {"Status": "Enabled"},
            "BucketEncryption": {
                "ServerSideEncryptionConfiguration": [
                    {"ServerSideEncryptionByDefault": {"SSEAlgorithm": "AES256"}}
                ]
            },
        }
    )

def test_s3_bucket_public_access_blocked():
    app   = cdk.App()
    stack = MyInfraStack(app, "TestStack")

    template = Template.from_stack(stack)

    template.has_resource_properties(
        "AWS::S3::Bucket",
        {
            "PublicAccessBlockConfiguration": {
                "BlockPublicAcls":       True,
                "BlockPublicPolicy":     True,
                "IgnorePublicAcls":      True,
                "RestrictPublicBuckets": True,
            }
        }
    )

pytest tests/

インフラのコードにテストを書けるのはCDKの大きなメリット。「S3バケットのパブリックアクセスがブロックされているか」のようなセキュリティ要件をテストとして記述できる。Terraformだと別途Terratestなどを使う必要がある。

CDKとTerraformの比較

項目 AWS CDK（Python） Terraform
記述言語 Python/TypeScript等 HCL（独自言語）
マルチクラウド AWSのみ 対応
ループ・条件分岐 Pythonで書ける for_each / countで制限あり
テスト pytestで書ける Terratest（Go）が必要
状態管理 CloudFormationが管理 tfstateファイル
学習コスト 低い（Python知ってれば） HCL独自の学習が必要
エコシステム AWSに限定 大きい
L3コンストラクト 高レベルパターンが豊富 モジュールを自分で探す

GitHubActionsでCDKをデプロイ

# .github/workflows/cdk-deploy.yml
name: CDK Deploy

on:
  push:
    branches: [main]
    paths:
      - 'infra/**'

jobs:
  deploy:
    runs-on: ubuntu-latest
    permissions:
      id-token: write
      contents: read

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.12'

      - name: Set up Node.js（CDK CLI用）
        uses: actions/setup-node@v4
        with:
          node-version: '20'

      - name: Install CDK
        run: npm install -g aws-cdk

      - name: Install Python dependencies
        run: |
          cd infra
          pip install -r requirements.txt

      - name: Configure AWS credentials（OIDC）
        uses: aws-actions/configure-aws-credentials@v4
        with:
          role-to-assume: arn:aws:iam::123456789012:role/cdk-deploy-role
          aws-region:     ap-northeast-1

      - name: CDK diff
        run: |
          cd infra
          cdk diff --all

      - name: CDK deploy
        run: |
          cd infra
          cdk deploy --all --require-approval never

まとめ

  • CDKはPythonでAWSリソースを定義できる
  • App → Stack → Constructの階層でリソースを管理する
  • L3コンストラクト（ApplicationLoadBalancedFargateServiceなど）で複数リソースを一括定義できる
  • cdk diffで変更差分を確認してからcdk deployする
  • テストをpytestで書けるのはTerraformにない強み
  • Contextで環境（dev/prod）を切り替えられる
  • CLIコマンド一つひとつだったインフラ構築がコードで再現可能になった

Pythonを覚えてからCDKを使えるようになるまでの学習コストが低かった。インフラの変更履歴をGitで管理できるようになったのが一番大きな変化。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?