【AWS】GitHub ActionsでCI/CDを実現する方法について

Posted at 2026-11-03

はじめに

こんばんは、mirukyです。

GitHubへマージしたら、AWSでもそのコードが動いていてほしいですよね。今回はGitHub Actionsで、テストからdev環境への反映までを自動化します。

反映先はEC2、ECS Fargate、Lambdaの3つです。実行先のコードが今回のコミットと一致するところまで、一緒に確かめていきましょう。

今回作るCI/CD
アクセスキーなしでAWSを使いましょう
ロールは全部で6つ作ります
アプリとワークフローを用意しましょう
AWSの環境は、コンソールから作ります
マージしたら、CDが動き出します
EC2には、Run Commandで配置します
ECS Fargateでは、ロールバックも起きました
Lambdaへのデプロイ
本番で使う前に足したいこと

1. 今回作るCI/CD

1-1. ブランチからdev環境までの流れ

CI（継続的インテグレーション）は変更のたびにテストを自動で走らせ、CD（継続的デプロイメント）はテストに通ったコードを実行環境へ自動で反映します。今回のCIはfeature/**へのプッシュと、develop宛てのプルリクエスト（PR）で動きます。

featureへプッシュしたときに検査するのは、そのブランチの先頭のコミットです。PRのときは少し違って、developの先頭にfeatureの変更を重ねた一時的なマージ結果を検査します。featureが呼んでいる関数をdevelop側で先に改名していた、という場合に失敗するのはマージ結果のほうだけです。

CDはdevelopへのプッシュで始まります。まずverifyでテストをもう一度実行し、成功して、かつDEPLOY_ENABLEDがtrueのときだけ3つのAWSのジョブへ進みます。PRのマージで起きたプッシュでも、直接のプッシュでも同じです。

注意したいのは、PRのCIが検査したマージ結果は、あくまで仮の組み合わせだという点です。マージの方法や、その間のdevelopの更新によっては、実際のマージコミットと中身が変わることがあります。そこでCDはイベントのgithub.shaをDEPLOY_REVISIONに固定し、どのジョブもこのコミットを取得します。実行中にdevelopの先頭が進んでも、取得するのは最初に固定したコミットのままです。

1-2. 用意するものと、作る環境

EC2とFargateは、東京リージョンに作る専用VPCのパブリックサブネットに置き、AWSのサービスへはパブリックIPv4アドレスで接続します。セキュリティグループの受信ルールは空のままにし、ロードバランサーとNAT Gatewayを使わない構成にします。

外から接続する手段がないので、デプロイしたあとの確認にはAWSのAPIを使います。ECSのジョブが取得するのは稼働中のイメージと正常性で、HTTPの応答はコンテナ自身のヘルスチェックが検査します。

dev環境かどうかは専用のAWSリソースとDEPLOY_ENVIRONMENT=devという環境変数で表します。値の渡し方は実行先ごとに異なり、EC2はsystemdの設定から、ECSとLambdaはCloudFormationテンプレートのDeploymentEnvironmentから受け取ります。

手を動かすには、リポジトリへの書き込み権限と、IAMロールを作成できるAWSの権限が必要です。料金は参考リンクにまとめました。

2. アクセスキーなしでAWSを使いましょう

2-1. 信頼ポリシーは何を照合しているの？

OIDC（OpenID Connect）を使うと、configure-aws-credentialsがGitHubの署名付きトークンをSTSへ渡し、期限付きの認証情報を受け取ります。GitHubに長期のアクセスキーを置かずに済みます。友人に家の合鍵を渡しておく代わりに、来る日だけ使えるスマートロックの暗証番号を送る、みたいな感じです。

ここから少しややこしいですかね。id-token: writeの権限で許されるのは、GitHubにOIDCトークンを要求するところまでです。そのトークンでロールを引き受けられるかは信頼ポリシーが、引き受けたあとにどのAWSのAPIを呼べるかは権限ポリシーが決めます。

信頼ポリシーではトークンのaudとsubを照合します。今回のsubには所有者IDとリポジトリIDが含まれていたので、4-3で取得した値をそのまま条件に使います。

{
  "sub": "repo:miruky@155365058/aws-github-actions-deploy-lab@1413857726:ref:refs/heads/develop",
  "aud": "sts.amazonaws.com",
  "repository_id": "1413857726",
  "repository_owner_id": "155365058"
}

<OBSERVED_SUB>と<AWS_ACCOUNT_ID>は自分の値に置き換えてください。

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Federated": "arn:aws:iam::<AWS_ACCOUNT_ID>:oidc-provider/token.actions.githubusercontent.com"
      },
      "Action": "sts:AssumeRoleWithWebIdentity",
      "Condition": {
        "StringEquals": {
          "token.actions.githubusercontent.com:aud": "sts.amazonaws.com",
          "token.actions.githubusercontent.com:sub": "<OBSERVED_SUB>"
        }
      }
    }
  ]
}

PRのイベントでは、subの末尾がpull_requestになります。マージしたときに起きるclosedイベントでも同じです。StringEqualsはsubを全文で照合するので、リポジトリIDまで同じでも、develop用の条件には一致しません。

そこでCDはマージによって起きるdevelopへのプッシュをきっかけに始めます。

3. ロールは全部で6つ作ります

3-1. GitHubが使うロール、AWSが使うロール

GitHubからAWSへ変更を送るジョブは3つあり、操作先ごとにデプロイ用のロールを用意します。これに、EC2のインスタンス、ECSのタスクの起動、Lambdaの関数がAWS側で使うロールを加えて、作るロールは全部で6つです。

GitHub用の3つのロールは同じOIDCの信頼条件を持ち、どのロールを使うかはジョブのrole-to-assumeで指定します。ECSでAWS側が使うのはEcsExecutionRoleで、このロールを使うのはFargateのエージェントです。

今回のアプリケーションはAWSのAPIを呼ばないので、タスク定義のtaskRoleArnは空のままにしています。起動時のイメージの取得とログの送信には、EcsExecutionRoleの権限を使います。

画面でもタスクロールは-で、タスク実行ロールにだけロールが設定されています。ジョブがタスク定義を登録し直すときも、記録されているこのロールのARNをそのまま引き継ぎます。

3-2. PassRoleが要るのはECSのジョブだけです

iam:PassRoleは、AWSのサービスにロールを渡すための許可です。タスク定義に実行ロールのARNを書いて登録するECSのジョブにはこの許可が要るので、渡せる対象をEcsExecutionRoleのARNに、渡し先をecs-tasks.amazonaws.comに絞りました。

ecsジョブは取得したタスク定義のうちimageだけを差し替えて登録します。3-1で触れたとおり、executionRoleArnは取得したときの値のまま登録の要求に入ります。

PassRoleが要るのはECSのジョブだけで、ロールを指定し直さないEC2とLambdaのジョブには付けていません。また、Resourceに指定できる対象は操作によって違うため、この例では表のように設定しています。

操作	対象の指定
`ecr:GetAuthorizationToken`	`Resource: "*"`。イメージの保存・取得は別の操作でリポジトリのARNに限定
`ssm:GetCommandInvocation`	`Resource: "*"`。コマンドの送信は`SendCommand`でインスタンスとドキュメントのARNに限定
`ecs:RegisterTaskDefinition`	タスク定義のファミリーのARN
`ecs:ListTasks`	`Resource: "*"`と、`ecs:cluster`の条件を併用

Resourceが"*"の操作でも、許可されるのは列挙したActionだけです。ListTasksはecs:clusterの条件で対象クラスターのARNを照合するので、別のクラスターへの要求にはこのAllow文が当てはまりません。

4. アプリとワークフローを用意しましょう

4-1. 3つの実行先で同じJSONを返すアプリ

手元にはGit、Python 3.13、GitHub CLIを用意し、gh auth loginでリポジトリへ書き込める状態にしておきます。app.pyは3つの実行先で共通の応答を作り、コミットIDはrevision.txtから、実行先と環境名は環境変数から取得します。

最初の作業はREADME付きの非公開リポジトリの作成とクローンです。your-id/your-repositoryは自分の値に置き換えてください。

DEPLOY_LAB_REPOSITORY="your-id/your-repository"
gh repo create "$DEPLOY_LAB_REPOSITORY" --private --add-readme --clone
cd "${DEPLOY_LAB_REPOSITORY##*/}"
# 既定ブランチがmain以外の場合も、この例ではmainを使用します。
if [ "$(git branch --show-current)" != main ]; then
  git branch --move main
  git push --set-upstream origin main
  gh repo edit "$DEPLOY_LAB_REPOSITORY" --default-branch main
fi
mkdir -p .github/workflows scripts tests infra

4〜9章のコードはそれぞれ直前に書いたファイル名で保存します。全部保存したら、4-3に戻ってプッシュします。

.gitignore

# 生成した配置物や環境ごとの設定をGitの管理対象から除外します。
__pycache__/
*.pyc
build/
prepared/
revision.txt
*.zip
*.tar.gz

app.py

"""EC2、ECS、Lambdaで同じ応答を生成する処理。"""

import json
import os
from pathlib import Path


def make_response(name="miruky", target=None):
    # デプロイしたコミットと入力値を実際の応答で確認します。
    if not isinstance(name, str) or not name.strip() or len(name) > 80:
        raise ValueError("name must be a non-empty string of at most 80 characters")
    revision_file = Path(__file__).with_name("revision.txt")
    revision = revision_file.read_text().strip() if revision_file.exists() else "local"
    return {
        "message": f"Hello, {name}!",
        "revision": revision,
        "target": target or os.environ.get("DEPLOY_TARGET", "local"),
        "environment": os.environ.get("DEPLOY_ENVIRONMENT", "local"),
    }


def encode_response(name="miruky", target=None):
    return json.dumps(make_response(name, target), ensure_ascii=False).encode("utf-8")

挨拶のHello, miruky!は今回のコミットでも、前にデプロイしたコードでも同じ文字列です。そのためジョブではrevision、target、environmentも照合し、今回のコミットが目的の環境で動いていることを確認します。

EC2とFargateではserver.pyが8080番ポートでJSONを返し、Lambdaではlambda_function.pyをindex.pyとして配置します。テストでは起動したHTTPサーバーの/healthも検査します。

server.py、lambda_function.py、Dockerfile、テストの全文

server.py

"""EC2とFargateで待ち受けるHTTPサーバー。"""

import json
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse

from app import encode_response, make_response


class Handler(BaseHTTPRequestHandler):
    def do_GET(self):
        # 正常性の確認とアプリの応答を同じ処理で返します。
        parsed = urlparse(self.path)
        if parsed.path not in ("/", "/health"):
            self.send_error(404)
            return
        name = parse_qs(parsed.query).get("name", ["miruky"])[0]
        try:
            body = encode_response(name)
        except ValueError as error:
            body = json.dumps({"error": str(error)}).encode()
            self.send_response(400)
        else:
            self.send_response(200)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)


if __name__ == "__main__":
    print(json.dumps({"ready": make_response()}), flush=True)
    ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()

lambda_function.py

"""Lambdaの呼び出し入口。"""

import json

from app import make_response


def lambda_handler(event, context):
    # EC2とECSと同じ応答に、Lambdaでの実行を記録します。
    response = make_response(event.get("name", "miruky"), "lambda")
    print(json.dumps(response, ensure_ascii=False))
    return response

Dockerfile

# Fargateでも同じアプリを実行し、コミットIDを応答へ含めます。
FROM python:3.13-slim
WORKDIR /app
COPY app.py server.py ./
ARG REVISION
RUN printf '%s\n' "$REVISION" > revision.txt
ENV DEPLOY_TARGET=ecs PYTHONUNBUFFERED=1
EXPOSE 8080
USER 65534:65534
CMD ["python", "server.py"]

tests/test_app.py

"""公開する応答の内容と入力の拒否を確認する。"""

import json
import os
import threading
import unittest
import urllib.request
from http.server import ThreadingHTTPServer
from unittest.mock import patch

from app import encode_response, make_response
from lambda_function import lambda_handler
from server import Handler


class AppTests(unittest.TestCase):
    def test_message_and_target(self):
        # 実行先を応答から判断できることを確認します。
        with patch.dict(os.environ, {"DEPLOY_TARGET": "ec2", "DEPLOY_ENVIRONMENT": "dev"}):
            result = make_response("miruky")
        self.assertEqual(result["message"], "Hello, miruky!")
        self.assertEqual(result["target"], "ec2")
        self.assertEqual(result["environment"], "dev")
        self.assertTrue(result["revision"])

    def test_lambda_entry(self):
        self.assertEqual(lambda_handler({"name": "miruky"}, None)["target"], "lambda")

    def test_utf8_json(self):
        self.assertEqual(json.loads(encode_response("読者"))["message"], "Hello, 読者!")

    def test_invalid_names_are_rejected(self):
        for value in (None, 42, "", " ", "a" * 81):
            with self.subTest(value=value), self.assertRaises(ValueError):
                make_response(value)

    def test_health_response_has_runtime_identity(self):
        # 実際のHTTP応答へ、実行先とdev設定が届くことを確認します。
        with patch.dict(os.environ, {"DEPLOY_TARGET": "ecs", "DEPLOY_ENVIRONMENT": "dev"}):
            server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            try:
                address = "http://127.0.0.1:%d/health" % server.server_port
                with urllib.request.urlopen(address, timeout=3) as response:
                    result = json.load(response)
                    self.assertEqual(response.status, 200)
                self.assertEqual(result["message"], "Hello, miruky!")
                self.assertEqual(result["target"], "ecs")
                self.assertEqual(result["environment"], "dev")
                self.assertTrue(result["revision"])
            finally:
                server.shutdown()
                thread.join(timeout=3)
                server.server_close()

4-2. CIのワークフロー

.github/workflows/ci.yml

name: CI

# feature単体と、developへ取り込む結果をそれぞれ検査します。
on:
  push:
    branches: ['feature/**']
  pull_request:
    branches: [develop]
permissions:
  contents: read
concurrency:
  group: ci-${{ github.event.pull_request.number || github.ref }}
  cancel-in-progress: true
jobs:
  test:
    name: ${{ github.event_name == 'pull_request' && 'ci-pr-merge-result' || 'ci-feature-head' }}
    runs-on: ubuntu-24.04
    timeout-minutes: 5
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
        with:
          python-version: '3.13'
      - name: Record the tested revision
        run: |
          # PRでは既定のマージ結果、pushではfeatureの先頭を検査します。
          printf 'event=%s\nref=%s\nsha=%s\n' "$GITHUB_EVENT_NAME" "$GITHUB_REF" "$GITHUB_SHA"
          test "$(git rev-parse HEAD)" = "$GITHUB_SHA"
      - run: python -m unittest discover -s tests -v
      - run: python -m compileall -q app.py server.py lambda_function.py

CIはコミットIDを記録したあと、unittestでアプリの応答を、compileallでPythonの構文を検査します。各ステップは直前のステップの成功を条件に動くので、途中で失敗するとそこで止まります。

ci.ymlに与える権限はcontents: readだけです。Record the tested revisionのステップでは検査する参照とコミットIDをログに出しておきます。

実際のログでは、参照先がプッシュではrefs/heads/feature/develop-dev-flow、PRではrefs/pull/1/mergeで、SHAも別の値でした。PRのほうは、1-1で触れた一時的なマージ結果を検査しています。

concurrencyを設定しているので、同じPRやブランチに新しいコミットが来ると、実行中の古いCIは取り消されます。CDは別のグループなので、CIの取り消しとは関係なく進む設定です。

4-3. 信頼条件に使うsubを取り出してみましょう

信頼条件に使うクレームは、手動で実行するprepare.ymlで取得します。workflow_dispatchで起動するには既定のブランチにもファイルが必要なので、mainとdevelopの両方に置き、--ref developを付けて実行します。

まずコードをmainへプッシュし、同じコミットからdevelopを作ります。

# 必要な全ファイルをmainへ保存し、同じコミットからdevelopを作ります。
git add .gitignore app.py server.py lambda_function.py Dockerfile tests .github scripts infra
git commit -m "Add sample app and deployment workflows"
git push origin main
git switch -c develop
git push -u origin develop

developを作ったときにもCDは動き出しますが、DEPLOY_ENABLEDがまだtrueではないので、実行されるのはverifyだけです。

.github/workflows/prepare.yml

name: Prepare template and inspect OIDC

# AWSへアクセスする前に、コードと信頼条件の材料を確認します。
on:
  workflow_dispatch:
    inputs:
      verify_aws_role:
        description: Verify the EC2 deployment role without deploying
        type: boolean
        default: false
permissions:
  contents: read
jobs:
  inspect:
    runs-on: ubuntu-24.04
    timeout-minutes: 5
    permissions:
      contents: read
      id-token: write
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
        with:
          python-version: '3.13'
      - name: Check application behavior
        run: python -m unittest discover -s tests -v
      - name: Inspect selected OIDC claims
        run: python scripts/oidc_claims.py
      - name: Verify AWS role trust
        if: inputs.verify_aws_role
        uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
        with:
          role-to-assume: ${{ vars.EC2_DEPLOY_ROLE_ARN }}
          aws-region: ${{ vars.AWS_REGION }}
          mask-aws-account-id: true
          retry-max-attempts: 2
      - name: Verify the assumed identity
        if: inputs.verify_aws_role
        run: |
          # デプロイを行わず、ロールの取得結果だけを確認します。
          aws sts get-caller-identity --query 'Arn' --output text

scripts/oidc_claims.py

"""GitHub Actions内で必要なOIDCクレームのみ表示する。"""

import base64
import json
import os
import urllib.request

# トークン全体、認証ヘッダー、有効期限をログへ出力しません。
url = os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"] + "&audience=sts.amazonaws.com"
request = urllib.request.Request(
    url, headers={"Authorization": "Bearer " + os.environ["ACTIONS_ID_TOKEN_REQUEST_TOKEN"]}
)
with urllib.request.urlopen(request, timeout=30) as response:
    jwt = json.load(response)["value"]
payload = jwt.split(".")[1]
claims = json.loads(base64.urlsafe_b64decode(payload + "=" * (-len(payload) % 4)))
print(json.dumps({key: claims[key] for key in ("sub", "aud", "repository_id", "repository_owner_id")}, indent=2))

このスクリプトはトークンをデコードするだけで、署名の検証はしません。ロールの作成と変数の登録が済んだあとにverify_aws_role=trueで実行すると、STSでロールを引き受けられるかの確認も加わります。

# developから手動実行し、信頼条件に使うsubを取得します。
REPO='<所有者>/<リポジトリ>'
gh workflow run prepare.yml --repo "$REPO" --ref develop -f verify_aws_role=false
gh run list --repo "$REPO" --workflow prepare.yml --limit 5

ログに出すのは、subとaud、それに今回のsubに含まれる2つのIDだけで、トークン本体は伏せています。developの参照で実行して、次の出力を取得しました。

この時点ではまだロールがないのでfalseで実行し、出力されたsubを5章のGitHubSubjectに渡します。

5. AWSの環境は、コンソールから作ります

5-1. テンプレートとリソース名は先に用意しておきます

アプリの実行環境はCloudFormationのテンプレートで作り、IAMポリシーの対象はリソースの参照で絞ります。S3バケットだけは先にコンソールで作るので、スタックから適用するのはそのバケットポリシーです。

パラメーターのうち、GitHubSubjectはdevelopのsubだけを、DeploymentEnvironmentはdevだけを受け付けます。

環境テンプレートの全文

infra/environment.template.yaml

# 初期環境をコンソールから作成し、アプリの更新権限を各デプロイ先へ限定します。
AWSTemplateFormatVersion: '2010-09-09'
Description: GitHub OIDC deployment to EC2, ECS Fargate and Lambda with independent roles and no inbound network access.
Metadata:
  AWSToolsMetrics:
    AWSAgentToolkit: aws-cloudformation@3
Parameters:
  ArtifactBucketName:
    Type: String
    Default: __AWSNAME_ArtifactBucket__
    AllowedPattern: 'miruky-[a-z]{16}'
    Description: Newly created private S3 bucket for the template and application artifacts.
  GitHubSubject:
    Type: String
    AllowedPattern: '^repo:.+:ref:refs/heads/develop$'
    Description: Exact sub claim observed in the develop branch of the GitHub workflow.
  DeploymentEnvironment:
    Type: String
    Default: dev
    AllowedValues: [dev]
    Description: Dedicated development environment for this example.
  LatestAmi:
    Type: AWS::SSM::Parameter::Value<AWS::EC2::Image::Id>
    Default: /aws/service/ami-amazon-linux-latest/al2023-ami-kernel-default-x86_64
Resources:
  Vpc:
    Type: AWS::EC2::VPC
    Properties:
      CidrBlock: 10.77.0.0/24
      EnableDnsSupport: true
      EnableDnsHostnames: true
      Tags: [{Key: Name, Value: __AWSNAME_Vpc__}]
  InternetGateway:
    Type: AWS::EC2::InternetGateway
    Properties:
      Tags: [{Key: Name, Value: __AWSNAME_InternetGateway__}]
  GatewayAttachment:
    Type: AWS::EC2::VPCGatewayAttachment
    Properties:
      VpcId: !Ref Vpc
      InternetGatewayId: !Ref InternetGateway
  PublicSubnet:
    Type: AWS::EC2::Subnet
    Properties:
      VpcId: !Ref Vpc
      CidrBlock: 10.77.0.0/26
      MapPublicIpOnLaunch: true
      Tags: [{Key: Name, Value: __AWSNAME_Subnet__}]
  RouteTable:
    Type: AWS::EC2::RouteTable
    Properties:
      VpcId: !Ref Vpc
      Tags: [{Key: Name, Value: __AWSNAME_RouteTable__}]
  InternetRoute:
    Type: AWS::EC2::Route
    DependsOn: GatewayAttachment
    Properties:
      RouteTableId: !Ref RouteTable
      DestinationCidrBlock: 0.0.0.0/0
      GatewayId: !Ref InternetGateway
  SubnetRouteAssociation:
    Type: AWS::EC2::SubnetRouteTableAssociation
    Properties:
      SubnetId: !Ref PublicSubnet
      RouteTableId: !Ref RouteTable
  ComputeSecurityGroup:
    Type: AWS::EC2::SecurityGroup
    # 受信通信を許可せず、AWSサービスへ向かう送信通信だけを許可します。
    Properties:
      GroupName: __AWSNAME_SecurityGroup__
      GroupDescription: Outbound service access only
      VpcId: !Ref Vpc
      SecurityGroupIngress: []
      SecurityGroupEgress: [{IpProtocol: '-1', CidrIp: 0.0.0.0/0}]
      Tags: [{Key: Name, Value: __AWSNAME_SecurityGroup__}]
  ArtifactBucketPolicy:
    Type: AWS::S3::BucketPolicy
    Properties:
      Bucket: !Ref ArtifactBucketName
      PolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Sid: DenyNonTLS
            Effect: Deny
            Principal: '*'
            Action: s3:*
            Resource:
              - !Sub arn:${AWS::Partition}:s3:::${ArtifactBucketName}
              - !Sub arn:${AWS::Partition}:s3:::${ArtifactBucketName}/*
            Condition:
              Bool: {aws:SecureTransport: 'false'}
  GitHubProvider:
    Type: AWS::IAM::OIDCProvider
    Properties:
      Url: https://token.actions.githubusercontent.com
      ClientIdList: [sts.amazonaws.com]
  Ec2RuntimeRole:
    Type: AWS::IAM::Role
    # Systems Managerの通信と、配置物の取得をEC2自身へ許可します。
    Properties:
      RoleName: __AWSNAME_Ec2RuntimeRole__
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: {Service: ec2.amazonaws.com}
            Action: sts:AssumeRole
      ManagedPolicyArns:
        - !Sub arn:${AWS::Partition}:iam::aws:policy/AmazonSSMManagedInstanceCore
      Policies:
        - PolicyName: __AWSNAME_Ec2RuntimePolicy__
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: s3:GetObject
                Resource: !Sub arn:${AWS::Partition}:s3:::${ArtifactBucketName}/releases/*
  Ec2InstanceProfile:
    Type: AWS::IAM::InstanceProfile
    Properties:
      InstanceProfileName: __AWSNAME_Ec2InstanceProfile__
      Roles: [!Ref Ec2RuntimeRole]
  Ec2Instance:
    Type: AWS::EC2::Instance
    DependsOn: InternetRoute
    Properties:
      ImageId: !Ref LatestAmi
      InstanceType: t3.micro
      CreditSpecification: {CPUCredits: standard}
      IamInstanceProfile: !Ref Ec2InstanceProfile
      SubnetId: !Ref PublicSubnet
      SecurityGroupIds: [!Ref ComputeSecurityGroup]
      MetadataOptions: {HttpTokens: required, HttpPutResponseHopLimit: 1}
      BlockDeviceMappings:
        - DeviceName: /dev/xvda
          Ebs: {VolumeSize: 8, VolumeType: gp3, Encrypted: true, DeleteOnTermination: true}
      Tags: [{Key: Name, Value: __AWSNAME_Ec2Instance__}]
      UserData:
        Fn::Base64: |
          #!/bin/bash
          # コードの更新前にPythonとSystems Managerの実行環境を準備します。
          set -eu
          dnf install -y python3
          systemctl enable --now amazon-ssm-agent
  Ec2DeployRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: __AWSNAME_Ec2DeployRole__
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: {Federated: !Ref GitHubProvider}
            Action: sts:AssumeRoleWithWebIdentity
            Condition:
              StringEquals:
                token.actions.githubusercontent.com:aud: sts.amazonaws.com
                token.actions.githubusercontent.com:sub: !Ref GitHubSubject
      Policies:
        - PolicyName: __AWSNAME_Ec2DeployPolicy__
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: s3:PutObject
                Resource: !Sub arn:${AWS::Partition}:s3:::${ArtifactBucketName}/releases/*
              - Effect: Allow
                Action: ssm:SendCommand
                Resource:
                  - !Sub arn:${AWS::Partition}:ec2:${AWS::Region}:${AWS::AccountId}:instance/${Ec2Instance}
                  - !Sub arn:${AWS::Partition}:ssm:${AWS::Region}::document/AWS-RunShellScript
              - Effect: Allow
                Action: [ssm:GetCommandInvocation, ssm:DescribeInstanceInformation]
                Resource: '*'
  ContainerRepository:
    Type: AWS::ECR::Repository
    Properties:
      RepositoryName: __AWSNAME_ContainerRepository__
      ImageTagMutability: IMMUTABLE
      EncryptionConfiguration: {EncryptionType: AES256}
      EmptyOnDelete: true
  EcsLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      LogGroupName: __AWSNAME_EcsLogGroup__
      RetentionInDays: 1
  EcsExecutionRole:
    Type: AWS::IAM::Role
    # イメージ取得とログ送信はFargateのエージェントが担当します。
    Properties:
      RoleName: __AWSNAME_EcsExecutionRole__
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: {Service: ecs-tasks.amazonaws.com}
            Action: sts:AssumeRole
            Condition:
              StringEquals: {aws:SourceAccount: !Ref 'AWS::AccountId'}
              ArnLike: {aws:SourceArn: !Sub 'arn:${AWS::Partition}:ecs:${AWS::Region}:${AWS::AccountId}:*'}
      Policies:
        - PolicyName: __AWSNAME_EcsExecutionPolicy__
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: ecr:GetAuthorizationToken
                Resource: '*'
              - Effect: Allow
                Action: [ecr:BatchCheckLayerAvailability, ecr:GetDownloadUrlForLayer, ecr:BatchGetImage]
                Resource: !GetAtt ContainerRepository.Arn
              - Effect: Allow
                Action: [logs:CreateLogStream, logs:PutLogEvents]
                Resource: !GetAtt EcsLogGroup.Arn
  EcsCluster:
    Type: AWS::ECS::Cluster
    Properties:
      ClusterName: __AWSNAME_EcsCluster__
  EcsTaskDefinition:
    Type: AWS::ECS::TaskDefinition
    # アプリはAWS APIを呼ばないため、アプリ用のタスクロールを指定しません。
    Properties:
      Family: __AWSNAME_EcsTaskFamily__
      Cpu: '256'
      Memory: '512'
      NetworkMode: awsvpc
      RequiresCompatibilities: [FARGATE]
      ExecutionRoleArn: !GetAtt EcsExecutionRole.Arn
      RuntimePlatform: {CpuArchitecture: X86_64, OperatingSystemFamily: LINUX}
      ContainerDefinitions:
        - Name: app
          Image: !Sub ${ContainerRepository.RepositoryUri}:bootstrap
          Essential: true
          Environment:
            - Name: DEPLOY_ENVIRONMENT
              Value: !Ref DeploymentEnvironment
            - Name: DEPLOY_TARGET
              Value: ecs
          PortMappings: [{ContainerPort: 8080, Protocol: tcp}]
          HealthCheck:
            Command:
              - CMD
              - python
              - '-c'
              - >-
                import json, pathlib, urllib.request;
                response = json.load(urllib.request.urlopen('http://127.0.0.1:8080/health', timeout=3));
                assert response['environment'] == 'dev';
                assert response['target'] == 'ecs';
                assert response['revision'] == pathlib.Path('/app/revision.txt').read_text().strip()
            Interval: 10
            Timeout: 5
            Retries: 3
            StartPeriod: 30
          LogConfiguration:
            LogDriver: awslogs
            Options:
              awslogs-group: !Ref EcsLogGroup
              awslogs-region: !Ref AWS::Region
              awslogs-stream-prefix: app
  EcsService:
    Type: AWS::ECS::Service
    DependsOn: InternetRoute
    # イメージの登録前は0タスクとし、最初のワークフローで1タスクにします。
    Properties:
      ServiceName: __AWSNAME_EcsService__
      Cluster: !Ref EcsCluster
      TaskDefinition: !Ref EcsTaskDefinition
      DesiredCount: 0
      LaunchType: FARGATE
      PlatformVersion: '1.4.0'
      DeploymentConfiguration:
        MinimumHealthyPercent: 100
        MaximumPercent: 200
        DeploymentCircuitBreaker: {Enable: true, Rollback: true}
      NetworkConfiguration:
        AwsvpcConfiguration:
          AssignPublicIp: ENABLED
          Subnets: [!Ref PublicSubnet]
          SecurityGroups: [!Ref ComputeSecurityGroup]
  EcsDeployRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: __AWSNAME_EcsDeployRole__
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: {Federated: !Ref GitHubProvider}
            Action: sts:AssumeRoleWithWebIdentity
            Condition:
              StringEquals:
                token.actions.githubusercontent.com:aud: sts.amazonaws.com
                token.actions.githubusercontent.com:sub: !Ref GitHubSubject
      Policies:
        - PolicyName: __AWSNAME_EcsDeployPolicy__
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: ecr:GetAuthorizationToken
                Resource: '*'
              - Effect: Allow
                Action: [ecr:BatchCheckLayerAvailability, ecr:InitiateLayerUpload, ecr:UploadLayerPart, ecr:CompleteLayerUpload, ecr:PutImage, ecr:DescribeImages]
                Resource: !GetAtt ContainerRepository.Arn
              - Effect: Allow
                Action: ecs:RegisterTaskDefinition
                Resource: !Sub arn:${AWS::Partition}:ecs:${AWS::Region}:${AWS::AccountId}:task-definition/__AWSNAME_EcsTaskFamily__:*
              - Effect: Allow
                Action: ecs:DescribeTaskDefinition
                Resource: '*'
              - Effect: Allow
                Action: [ecs:UpdateService, ecs:DescribeServices]
                Resource: !Ref EcsService
              - Effect: Allow
                Action: ecs:ListTasks
                Resource: '*'
                Condition:
                  ArnEquals: {ecs:cluster: !GetAtt EcsCluster.Arn}
              - Effect: Allow
                Action: ecs:DescribeTasks
                Resource: !Sub arn:${AWS::Partition}:ecs:${AWS::Region}:${AWS::AccountId}:task/${EcsCluster}/*
              - Effect: Allow
                Action: iam:PassRole
                Resource: !GetAtt EcsExecutionRole.Arn
                Condition:
                  StringEquals: {iam:PassedToService: ecs-tasks.amazonaws.com}
  LambdaLogGroup:
    Type: AWS::Logs::LogGroup
    DeletionPolicy: Delete
    UpdateReplacePolicy: Delete
    Properties:
      LogGroupName: __AWSNAME_LambdaLogGroup__
      RetentionInDays: 1
  LambdaRuntimeRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: __AWSNAME_LambdaRuntimeRole__
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: {Service: lambda.amazonaws.com}
            Action: sts:AssumeRole
      Policies:
        - PolicyName: __AWSNAME_LambdaRuntimePolicy__
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: [logs:CreateLogStream, logs:PutLogEvents]
                Resource: !GetAtt LambdaLogGroup.Arn
  LambdaFunction:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: __AWSNAME_LambdaFunction__
      Runtime: python3.13
      Handler: index.lambda_handler
      Role: !GetAtt LambdaRuntimeRole.Arn
      Timeout: 10
      MemorySize: 128
      Architectures: [x86_64]
      Environment:
        Variables:
          DEPLOY_ENVIRONMENT: !Ref DeploymentEnvironment
      LoggingConfig: {LogGroup: !Ref LambdaLogGroup, LogFormat: Text}
      Code:
        ZipFile: |
          # 最初のデプロイでアプリを反映するまでの初期コードです。
          def lambda_handler(event, context):
              return {"message": "Initial function", "revision": "bootstrap", "target": "lambda"}
  LambdaDeployRole:
    Type: AWS::IAM::Role
    Properties:
      RoleName: __AWSNAME_LambdaDeployRole__
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal: {Federated: !Ref GitHubProvider}
            Action: sts:AssumeRoleWithWebIdentity
            Condition:
              StringEquals:
                token.actions.githubusercontent.com:aud: sts.amazonaws.com
                token.actions.githubusercontent.com:sub: !Ref GitHubSubject
      Policies:
        - PolicyName: __AWSNAME_LambdaDeployPolicy__
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action: [lambda:UpdateFunctionCode, lambda:GetFunctionConfiguration, lambda:InvokeFunction]
                Resource: !GetAtt LambdaFunction.Arn
Outputs:
  DeploymentEnvironment:
    Description: Deployment environment represented by these dedicated resources.
    Value: !Ref DeploymentEnvironment
  Ec2InstanceId: {Description: EC2 deployment target., Value: !Ref Ec2Instance}
  ArtifactBucket: {Description: Private artifact bucket., Value: !Ref ArtifactBucketName}
  Ec2DeployRoleArn: {Description: OIDC role for EC2 deployment., Value: !GetAtt Ec2DeployRole.Arn}
  EcsClusterName: {Description: ECS cluster., Value: !Ref EcsCluster}
  EcsServiceName: {Description: ECS service., Value: !GetAtt EcsService.Name}
  EcsTaskFamily: {Description: ECS task definition family., Value: __AWSNAME_EcsTaskFamily__}
  EcrRepository: {Description: ECR repository., Value: !Ref ContainerRepository}
  EcsDeployRoleArn: {Description: OIDC role for ECS deployment., Value: !GetAtt EcsDeployRole.Arn}
  LambdaFunctionName: {Description: Lambda deployment target., Value: !Ref LambdaFunction}
  LambdaDeployRoleArn: {Description: OIDC role for Lambda deployment., Value: !GetAtt LambdaDeployRole.Arn}

作成した直後のECRは空なので、タスクを起動してもbootstrapのイメージの取得で失敗します。そこでECSサービスは希望タスク数0で作り、デプロイを有効にしたあとの初回の反映で1に変えます。

補助スクリプトはプレースホルダーの種類ごとにランダムな名前を作り、同じ種類が出てくる箇所をすべて同じ値で置き換えます。完成したテンプレートと名前の対応は、preparedの2つのファイルに保存されます。

再実行すると新しい名前で2つのファイルが上書きされ、作成済みのAWSリソースの名前と合わなくなります。最初に生成したpreparedを、最後まで使ってください。

名前とテンプレートを準備する補助スクリプトの全文

scripts/prepare_template.py

"""新しいAWS名を生成し、コンソールへ渡すテンプレートを作成する。"""

import json
import re
import subprocess
import sys
from pathlib import Path

# 名前は同梱した生成器の出力を変更せずに使用します。
root = Path(__file__).resolve().parents[1]
template = (root / "infra/environment.template.yaml").read_text()
labels = ["Stack"] + sorted(set(re.findall(r"__AWSNAME_(\w+)__", template)))
generator = root / "scripts/aws_resource_names.py"
generated = subprocess.check_output(
    [sys.executable, str(generator), "generate", "--count", str(len(labels))], text=True
).splitlines()
if len(generated) != len(labels):
    raise RuntimeError("Unexpected resource-name count")
names = dict(zip(labels, generated))
subprocess.run([sys.executable, str(generator), "audit", *generated], check=True)
for label, value in names.items():
    template = template.replace(f"__AWSNAME_{label}__", value)
prepared = root / "prepared"
prepared.mkdir(exist_ok=True)
(prepared / "environment.yaml").write_text(template)
(prepared / "names.json").write_text(json.dumps(names, indent=2))
print(json.dumps({"stack": names["Stack"], "bucket": names["ArtifactBucket"]}, indent=2))

scripts/aws_resource_names.py

#!/usr/bin/env python3
# 乱数によるAWSリソース名の生成と、記録した名前の検査を行います。
"""Generate and audit clock-independent AWS resource-name cores."""

from __future__ import annotations

import argparse
import json
import re
import secrets
import string
import sys
from dataclasses import asdict, dataclass
from pathlib import Path

DEFAULT_RANDOM_LENGTH = 16
MIN_RANDOM_LENGTH = 12
MAX_RANDOM_LENGTH = 56
GENERATOR_ID = "aws-resource-name-v1"
FORBIDDEN_SUBSTRINGS = ("qiita", "zenn")
DATE_OR_TIME_PATTERNS = (
    re.compile(r"(?:19|20)\d{2}[-_.]?(?:0[1-9]|1[0-2])[-_.]?(?:0[1-9]|[12]\d|3[01])"),
    re.compile(r"(?<!\d)\d{8}(?!\d)"),
    re.compile(r"(?<!\d)\d{10,13}(?!\d)"),
    re.compile(r"(?:timestamp|datetime|epoch)", re.IGNORECASE),
)


@dataclass(frozen=True)
class NameRecord:
    name: str
    prefix: str
    separator: str
    random_length: int
    alphabet: str = "lowercase-ascii"
    generated_by: str = GENERATOR_ID


def _random_letters(length: int) -> str:
    while True:
        value = "".join(secrets.choice(string.ascii_lowercase) for _ in range(length))
        if not any(term in value for term in FORBIDDEN_SUBSTRINGS):
            return value


def generate_name(*, prefix: str, compact: bool, random_length: int) -> NameRecord:
    if not MIN_RANDOM_LENGTH <= random_length <= MAX_RANDOM_LENGTH:
        raise ValueError(
            f"random length must be between {MIN_RANDOM_LENGTH} and {MAX_RANDOM_LENGTH}"
        )
    if prefix not in {"miruky", "none"}:
        raise ValueError("prefix must be 'miruky' or 'none'")

    random_part = _random_letters(random_length)
    if prefix == "none":
        return NameRecord(
            name=random_part,
            prefix="",
            separator="",
            random_length=random_length,
        )

    separator = "" if compact else "-"
    return NameRecord(
        name=f"miruky{separator}{random_part}",
        prefix="miruky",
        separator=separator,
        random_length=random_length,
    )


def validate_name(name: str) -> list[str]:
    errors: list[str] = []
    lowered = name.lower()

    for term in FORBIDDEN_SUBSTRINGS:
        if term in lowered:
            errors.append(f"contains prohibited platform term: {term}")

    if any(pattern.search(name) for pattern in DATE_OR_TIME_PATTERNS):
        errors.append("contains a date-, time-, epoch-, or timestamp-shaped fragment")

    random_part = ""
    if re.fullmatch(r"miruky-[a-z]+", name):
        random_part = name.removeprefix("miruky-")
    elif re.fullmatch(r"miruky[a-z]+", name):
        random_part = name.removeprefix("miruky")
    elif re.fullmatch(r"[a-z]+", name):
        random_part = name
    else:
        errors.append(
            "must be lowercase random letters with only the optional prefix 'miruky'"
        )

    if random_part and not MIN_RANDOM_LENGTH <= len(random_part) <= MAX_RANDOM_LENGTH:
        errors.append(
            f"random part must contain {MIN_RANDOM_LENGTH}-{MAX_RANDOM_LENGTH} lowercase letters"
        )

    if random_part and any(term in random_part for term in FORBIDDEN_SUBSTRINGS):
        errors.append("random part contains a prohibited platform term")

    return list(dict.fromkeys(errors))


def _build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        description="Generate or audit random AWS resource-name cores without dates or platform names."
    )
    subparsers = parser.add_subparsers(dest="command", required=True)

    generate_parser = subparsers.add_parser("generate", help="generate compliant names")
    generate_parser.add_argument("--count", type=int, default=1)
    generate_parser.add_argument(
        "--prefix", choices=("miruky", "none"), default="miruky"
    )
    generate_parser.add_argument("--compact", action="store_true")
    generate_parser.add_argument(
        "--random-length", type=int, default=DEFAULT_RANDOM_LENGTH
    )
    generate_parser.add_argument("--json", action="store_true")

    audit_parser = subparsers.add_parser("audit", help="audit one or more name cores")
    audit_parser.add_argument("names", nargs="+")

    ledger_parser = subparsers.add_parser(
        "audit-ledger", help="audit every resource-name record in a JSON ledger"
    )
    ledger_parser.add_argument("ledger", type=Path)
    ledger_parser.add_argument(
        "--final",
        action="store_true",
        help="also require a verified absent final state for every attempted name",
    )

    return parser


def _run_generate(args: argparse.Namespace) -> int:
    if not 1 <= args.count <= 1000:
        print("ERROR: count must be between 1 and 1000", file=sys.stderr)
        return 2

    try:
        records = [
            generate_name(
                prefix=args.prefix,
                compact=args.compact,
                random_length=args.random_length,
            )
            for _ in range(args.count)
        ]
    except ValueError as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 2

    names = [record.name for record in records]
    if len(names) != len(set(names)):
        print("ERROR: generator produced a duplicate name", file=sys.stderr)
        return 1

    if args.json:
        print(
            json.dumps(
                {
                    "schema_version": 1,
                    "naming_policy": GENERATOR_ID,
                    "names": [asdict(record) for record in records],
                },
                ensure_ascii=False,
                indent=2,
            )
        )
    else:
        print("\n".join(names))
    return 0


def _run_audit(args: argparse.Namespace) -> int:
    failed = False
    for name in args.names:
        errors = validate_name(name)
        if errors:
            failed = True
            print(f"FAIL {name}")
            for error in errors:
                print(f"  - {error}")
        else:
            print(f"PASS {name}")
    return 1 if failed else 0


def _run_audit_ledger(args: argparse.Namespace) -> int:
    findings: list[str] = []
    try:
        payload = json.loads(args.ledger.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        print(f"FAIL {args.ledger}")
        print(f"  - cannot read a valid JSON ledger: {exc}")
        return 1

    if not isinstance(payload, dict):
        findings.append("ledger root must be a JSON object")
        resources: list[object] = []
        aws_generated_identifiers: list[object] = []
    else:
        if payload.get("schema_version") != 1:
            findings.append("schema_version must be 1")
        if payload.get("inventory_complete") != "PASS":
            findings.append("inventory_complete must be PASS")
        if payload.get("name_field_inventory_complete") != "PASS":
            findings.append("name_field_inventory_complete must be PASS")
        resources_value = payload.get("resources")
        if not isinstance(resources_value, list):
            findings.append("resources must be an array")
            resources = []
        else:
            resources = resources_value
        generated_value = payload.get("aws_generated_identifiers")
        if not isinstance(generated_value, list):
            findings.append("aws_generated_identifiers must be an array")
            aws_generated_identifiers = []
        else:
            aws_generated_identifiers = generated_value
        if not resources and not aws_generated_identifiers:
            findings.append(
                "resources and aws_generated_identifiers cannot both be empty"
            )

    seen_names: set[str] = set()
    seen_cores: set[str] = set()
    for index, resource in enumerate(resources, start=1):
        location = f"resources[{index}]"
        if not isinstance(resource, dict):
            findings.append(f"{location} must be an object")
            continue

        required_strings = (
            "logical_label",
            "service",
            "region",
            "resource_name",
            "name_core",
            "generated_by",
            "create_target",
            "cleanup_target",
            "absence_check",
        )
        for key in required_strings:
            if not isinstance(resource.get(key), str) or not resource[key].strip():
                findings.append(f"{location}.{key} must be a nonempty string")

        core = resource.get("name_core")
        full_name = resource.get("resource_name")
        if isinstance(core, str) and core:
            for error in validate_name(core):
                findings.append(f"{location}.name_core {error}")
            if core in seen_cores:
                findings.append(f"{location}.name_core duplicates another ledger entry")
            seen_cores.add(core)
        if isinstance(full_name, str) and isinstance(core, str) and full_name and core:
            lowered_full_name = full_name.lower()
            for term in FORBIDDEN_SUBSTRINGS:
                if term in lowered_full_name:
                    findings.append(
                        f"{location}.resource_name contains prohibited platform term: {term}"
                    )
            if any(pattern.search(full_name) for pattern in DATE_OR_TIME_PATTERNS):
                findings.append(
                    f"{location}.resource_name contains a date-, time-, epoch-, or timestamp-shaped fragment"
                )
            required_prefix = resource.get("required_prefix", "")
            required_suffix = resource.get("required_suffix", "")
            wrapper_justification = resource.get("wrapper_justification", "")
            wrapper_fields_valid = True
            for key, value in (
                ("required_prefix", required_prefix),
                ("required_suffix", required_suffix),
                ("wrapper_justification", wrapper_justification),
            ):
                if not isinstance(value, str):
                    findings.append(f"{location}.{key} must be a string when present")
                    wrapper_fields_valid = False

            if wrapper_fields_valid:
                has_wrapper = bool(required_prefix or required_suffix)
                if full_name == core:
                    if has_wrapper or wrapper_justification:
                        findings.append(
                            f"{location} must leave wrapper fields empty when resource_name equals name_core"
                        )
                else:
                    if not has_wrapper:
                        findings.append(
                            f"{location} must record every service-required prefix or suffix"
                        )
                    if not wrapper_justification.strip():
                        findings.append(
                            f"{location}.wrapper_justification must cite the service requirement"
                        )
                    if full_name != f"{required_prefix}{core}{required_suffix}":
                        findings.append(
                            f"{location}.resource_name must equal required_prefix + name_core + required_suffix"
                        )
            if full_name in seen_names:
                findings.append(f"{location}.resource_name duplicates another ledger entry")
            seen_names.add(full_name)

        if resource.get("generated_by") != GENERATOR_ID:
            findings.append(f"{location}.generated_by must be {GENERATOR_ID}")

        if args.final and resource.get("final_state") not in {
            "deleted-and-absent",
            "not-created-and-absent",
        }:
            findings.append(
                f"{location}.final_state must prove deleted-and-absent or not-created-and-absent"
            )

    for index, item in enumerate(aws_generated_identifiers, start=1):
        location = f"aws_generated_identifiers[{index}]"
        if not isinstance(item, dict):
            findings.append(f"{location} must be an object")
            continue
        for key in ("logical_label", "service", "region", "identifier_kind"):
            if not isinstance(item.get(key), str) or not item[key].strip():
                findings.append(f"{location}.{key} must be a nonempty string")
        if item.get("exemption_reason") != "no-supported-user-name-field":
            findings.append(
                f"{location}.exemption_reason must be no-supported-user-name-field"
            )

    if findings:
        print(f"FAIL {args.ledger}")
        for finding in findings:
            print(f"  - {finding}")
        return 1

    print(
        f"PASS {args.ledger} ({len(resources)} names, "
        f"{len(aws_generated_identifiers)} AWS-generated identifiers)"
    )
    return 0


def main() -> int:
    args = _build_parser().parse_args()
    if args.command == "generate":
        return _run_generate(args)
    if args.command == "audit":
        return _run_audit(args)
    return _run_audit_ledger(args)


if __name__ == "__main__":
    raise SystemExit(main())

# 名前を生成し、コンソールへ渡す完成版のテンプレートを手元に保存します。
python3 scripts/prepare_template.py

既存のGitHubのOIDCプロバイダーを使うときは、テンプレートからGitHubProviderを外し、そのARNをPrincipal.Federatedに指定します。

5-2. S3にテンプレートを置いて、スタックを作りましょう

最初に、テンプレートと配置物を置くS3バケットを作ります。名前にはprepared/names.jsonのArtifactBucketを使います。

バケット名には生成した名前を入力し、AWS リージョンはアジアパシフィック (東京) ap-northeast-1にします。あわせて汎用とグローバル名前空間を選びます。

ACL 無効 (推奨)を選び、パブリックアクセスをすべてブロックにチェックを入れます。配置物の保存と取得には、IAMロールの権限を使います。

バケットのバージョニングは有効にするにします。同じキーのファイルを上書きしても、以前のバージョンを指定して配置物を取り出せます。

暗号化はAmazon S3 マネージドキーを使用したサーバー側の暗号化 (SSE-S3)にします。SSE-S3なら、配置物を取得する側に要るのは、S3からの取得の許可だけです。

この設定でバケットを作り、prepared/environment.yamlをアップロードします。

アップロードしたファイルのオブジェクト URLを、CloudFormationのテンプレートの取得元に指定します。スタック名にはprepared/names.jsonのStackを使います。

パラメータには次の値を入力します。

入力欄	値
`ArtifactBucketName`	作成したS3バケットの名前
`GitHubSubject`	4-3で取得した値
`DeploymentEnvironment`	`dev`
`LatestAmi`	既定のAmazon Linux 2023のパラメーター

GitHubSubjectに入れるのは、4-3でdevelopから取得したsubです。ほかのブランチの値はテンプレートのパラメーター条件に合わないので、スタックの作成時に拒否されます。

テンプレートがIAMロールを作るので、AWS CloudFormation によって IAM リソースがカスタム名で作成される場合があることを承認します。にチェックを入れてから、次へで確認画面に進みます。

確認して作成の画面では、テンプレート URLがアップロードしたenvironment.yamlになっているかを確かめてから、この完成版のテンプレートでスタックを作成します。

ステータスがCREATE_COMPLETEになれば作成は完了です。5-1で書いたとおりECSサービスの希望タスク数は0なので、コンテナが起動するのは、6-2でデプロイを有効にしたあとのCDです。

出力 (11)の中身はDeploymentEnvironmentのdevや、今回のインスタンスIDを示すEc2InstanceIdなどです。DeploymentEnvironment以外の値は6-2でリポジトリ変数に登録します。

6. マージしたら、CDが動き出します

6-1. わざと失敗させてから、developへマージしましょう

ここからは、PRをマージしてCDを動かす段階です。developからfeature/develop-dev-flowを作り、READMEへの追記をPRの差分にしてプッシュしたら、develop宛てのPRを作ります。

git switch develop
git pull --ff-only origin develop
git switch -c feature/develop-dev-flow
# PRで確認する変更として、dev環境の説明をREADMEへ追記します。
printf '\nDeploy target: dev\n' >> README.md
git add README.md
git commit -m "Describe the dev deployment target"
git push -u origin feature/develop-dev-flow
gh pr create --base develop --title "Describe the dev deployment target" --body "READMEへdev環境の説明を追記します。"

両方のCIが成功したら、今度はわざと失敗させてみましょう。tests/test_app.pyのtest_message_and_targetで、メッセージの期待値だけをExpected failing greetingに書き換えてプッシュすると、PRにも同じ変更が反映されます。

# test_message_and_targetのmessageの期待値をExpected failing greetingへ書き換えてから実行します。
git add tests/test_app.py
git commit -m "Expect a wrong greeting to check CI"
git push

featureの先頭でも、PRのマージ結果でも、同じアサーションで失敗しました。

# 直前の失敗用コミットだけを打ち消してから、再びプッシュします。
git revert --no-edit HEAD
git push

期待値を戻したコミットをプッシュすると、両方のCIが成功に戻りました。

CIが成功したのを確認してから、PR #1をdevelopへマージします。

# 修正後のCIが成功したことを確認してから、PRをマージします。
gh pr checks --watch && gh pr merge --merge

今回はブランチ保護を設定していないので、CIが失敗したPRでもマージはできてしまいます。その場合でもCDのverifyが同じテストで失敗し、AWSのジョブの手前で止まります。マージそのものを止める方法は、10章の話題です。

6-2. CDのワークフローとリポジトリ変数

.github/workflows/cd-dev.yml

CDのワークフロー全文

name: CD to dev

# PRのマージでdevelopへ反映された版を、devへデプロイします。
on:
  push:
    branches: [develop]
permissions:
  contents: read
concurrency:
  group: deploy-dev-${{ github.repository }}
  cancel-in-progress: false
env:
  DEPLOY_REVISION: ${{ github.sha }}
  DEPLOY_ENVIRONMENT: dev
jobs:
  verify:
    if: github.event_name == 'push' && github.ref == 'refs/heads/develop'
    runs-on: ubuntu-24.04
    timeout-minutes: 5
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          ref: ${{ env.DEPLOY_REVISION }}
      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
        with:
          python-version: '3.13'
      - name: Verify the merged revision
        run: |
          # 実際にマージされた版を、配置する版として固定します。
          printf 'ref=%s\ndeploy_revision=%s\nenvironment=%s\n' "$GITHUB_REF" "$DEPLOY_REVISION" "$DEPLOY_ENVIRONMENT"
          test "$(git rev-parse HEAD)" = "$DEPLOY_REVISION"
          python -m unittest discover -s tests -v
          python -m compileall -q app.py server.py lambda_function.py

  ec2:
    needs: verify
    if: vars.DEPLOY_ENABLED == 'true'
    runs-on: ubuntu-24.04
    timeout-minutes: 15
    permissions:
      contents: read
      id-token: write
    env:
      ARTIFACT_BUCKET: ${{ vars.ARTIFACT_BUCKET }}
      INSTANCE_ID: ${{ vars.EC2_INSTANCE_ID }}
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          ref: ${{ env.DEPLOY_REVISION }}
      - uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
        with:
          role-to-assume: ${{ vars.EC2_DEPLOY_ROLE_ARN }}
          aws-region: ${{ vars.AWS_REGION }}
          mask-aws-account-id: true
      - name: Package and deploy to EC2 dev
        run: bash scripts/deploy_ec2.sh

  ecs:
    needs: verify
    if: vars.DEPLOY_ENABLED == 'true'
    runs-on: ubuntu-24.04
    timeout-minutes: 20
    permissions:
      contents: read
      id-token: write
    env:
      ECR_REPOSITORY: ${{ vars.ECR_REPOSITORY }}
      ECS_CLUSTER: ${{ vars.ECS_CLUSTER }}
      ECS_SERVICE: ${{ vars.ECS_SERVICE }}
      ECS_TASK_FAMILY: ${{ vars.ECS_TASK_FAMILY }}
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          ref: ${{ env.DEPLOY_REVISION }}
      - uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
        with:
          role-to-assume: ${{ vars.ECS_DEPLOY_ROLE_ARN }}
          aws-region: ${{ vars.AWS_REGION }}
          mask-aws-account-id: true
      - name: Build and push the container image
        id: image
        run: |
          # マージした版と再実行番号から、今回のイメージを特定します。
          ACCOUNT_ID=$(aws sts get-caller-identity --query Account --output text)
          REGISTRY="$ACCOUNT_ID.dkr.ecr.$AWS_REGION.amazonaws.com"
          IMAGE="$REGISTRY/$ECR_REPOSITORY:$DEPLOY_REVISION-$GITHUB_RUN_ATTEMPT"
          aws ecr get-login-password | docker login --username AWS --password-stdin "$REGISTRY"
          docker build --build-arg REVISION="$DEPLOY_REVISION" --tag "$IMAGE" .
          docker push "$IMAGE"
          printf 'image=%s\n' "$IMAGE" >> "$GITHUB_OUTPUT"
      - name: Obtain the current task definition
        run: |
          # devの実行ロールとログ設定を保持して更新します。
          aws ecs describe-task-definition --task-definition "$ECS_TASK_FAMILY" --query taskDefinition > task-definition.json
      - name: Render the new task definition
        id: task
        uses: aws-actions/amazon-ecs-render-task-definition@8d79160660ad39c402a16e5fc7cd2a32e8f7d6b7 # v1.9.1
        with:
          task-definition: task-definition.json
          container-name: app
          image: ${{ steps.image.outputs.image }}
      - name: Deploy the new task definition to dev
        uses: aws-actions/amazon-ecs-deploy-task-definition@a00d4d57ac0de1a99b39fe469f3d898d38717c2b # v2.6.4
        with:
          task-definition: ${{ steps.task.outputs.task-definition }}
          cluster: ${{ vars.ECS_CLUSTER }}
          service: ${{ vars.ECS_SERVICE }}
          desired-count: 1
          wait-for-service-stability: true
          wait-for-minutes: 15
      - name: Verify the actual running image and health
        env:
          EXPECTED_IMAGE: ${{ steps.image.outputs.image }}
        run: python3 scripts/verify_ecs.py

  lambda:
    needs: verify
    if: vars.DEPLOY_ENABLED == 'true'
    runs-on: ubuntu-24.04
    timeout-minutes: 10
    permissions:
      contents: read
      id-token: write
    env:
      FUNCTION_NAME: ${{ vars.LAMBDA_FUNCTION_NAME }}
    steps:
      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
        with:
          ref: ${{ env.DEPLOY_REVISION }}
      - uses: aws-actions/configure-aws-credentials@e1253824e5c10ff9df46874f81ed3ec929e19cfd # v6.3.0
        with:
          role-to-assume: ${{ vars.LAMBDA_DEPLOY_ROLE_ARN }}
          aws-region: ${{ vars.AWS_REGION }}
          mask-aws-account-id: true
      - name: Update and invoke the dev function
        run: bash scripts/deploy_lambda.sh

各ジョブはDEPLOY_REVISION（github.sha）をチェックアウトします。CDのverifyに与える権限はcontents: readだけで、id-token: writeはAWSへ反映する3つのジョブにだけ設定します。

4-3のprepare.ymlのinspectジョブにもこの権限を付けているのは、クレームを取得するためです。

1-1の条件どおり、vars.DEPLOY_ENABLEDがまだtrueではないマージ直後の実行では、verifyがsuccess、3つのジョブはskippedになりました。

この実行のheadShaはマージコミットと一致しています。続いて、5章の出力から実行先の名前とロールのARNをリポジトリ変数に登録します。

# 山かっこの値を、自分のスタックの出力で置き換えます。
REPO='<所有者>/<リポジトリ>'
gh variable set AWS_REGION --repo "$REPO" --body 'ap-northeast-1'
gh variable set ARTIFACT_BUCKET --repo "$REPO" --body '<ArtifactBucketの値>'
gh variable set EC2_INSTANCE_ID --repo "$REPO" --body '<Ec2InstanceIdの値>'
gh variable set EC2_DEPLOY_ROLE_ARN --repo "$REPO" --body '<Ec2DeployRoleArnの値>'
gh variable set ECR_REPOSITORY --repo "$REPO" --body '<EcrRepositoryの値>'
gh variable set ECS_CLUSTER --repo "$REPO" --body '<EcsClusterNameの値>'
gh variable set ECS_SERVICE --repo "$REPO" --body '<EcsServiceNameの値>'
gh variable set ECS_TASK_FAMILY --repo "$REPO" --body '<EcsTaskFamilyの値>'
gh variable set ECS_DEPLOY_ROLE_ARN --repo "$REPO" --body '<EcsDeployRoleArnの値>'
gh variable set LAMBDA_FUNCTION_NAME --repo "$REPO" --body '<LambdaFunctionNameの値>'
gh variable set LAMBDA_DEPLOY_ROLE_ARN --repo "$REPO" --body '<LambdaDeployRoleArnの値>'
# 設定値をすべて登録した後に、AWSへの反映を有効にします。
gh variable set DEPLOY_ENABLED --repo "$REPO" --body true

CDのconcurrencyは実行中の反映を取り消さない設定です。ただし、1つ目の反映中に2つ目と3つ目をマージすると、待っていた2つ目は取り消されます。

マージしたすべてのコミットが反映されるとは限らないので、実際に反映されたコミットは応答や稼働中のイメージで確認します。

これで、PRをdevelopへマージすると、3つのジョブがそれぞれのロールで反映を始めるようになりました。

試しに、dev環境で確認する項目をREADMEへ追記するPRを作り、CIの成功を待ってマージします。

git switch develop
git pull --ff-only origin develop
git switch -c feature/describe-dev-checks
# 反映後に確認する項目をREADMEへ追記します。
cat >> README.md <<'EOF'

## dev環境での確認

- EC2はRun Commandの出力、ECS Fargateはタスクのヘルスステータス、Lambdaは同期呼び出しの応答で確認します。
- 各応答のrevisionが、developへマージしたコミットIDと一致することを確認します。
EOF
git add README.md
git commit -m "Describe dev deployment checks in README"
git push -u origin feature/describe-dev-checks
gh pr create --base develop --title "Describe dev deployment checks" --body "dev環境での確認項目をREADMEへ追記します。"
# PRのCIが成功した場合にマージします。
gh pr checks --watch && gh pr merge --merge

verifyと3つのAWSのジョブはすべて成功しました。なお、7〜9章で確認するのは、8章のヘルスチェックの失敗から復旧したあとのコミットです。

7. EC2には、Run Commandで配置します

7-1. EC2の中で応答を確かめてみましょう

EC2のジョブは配置物を圧縮してS3に保存し、Systems ManagerのRun Commandで対象のインスタンスへ配置を指示します。指示を受け取るのはインスタンスのSSM Agentで、Agentのほうから Systems Managerへ接続するため、受信のルールは空のままで動きます。

インスタンスでは配置物のSHA-256を確認してから展開し、アプリケーションのサービスを再起動します。

scripts/deploy_ec2.sh

#!/usr/bin/env bash
# 配置物のハッシュを確認してから、専用EC2上のアプリを更新します。
set -euo pipefail
: "${ARTIFACT_BUCKET:?}" "${INSTANCE_ID:?}" "${DEPLOY_REVISION:?}" "${GITHUB_RUN_ATTEMPT:?}"
mkdir -p build
printf '%s\n' "$DEPLOY_REVISION" > revision.txt
tar -czf build/app.tar.gz app.py server.py revision.txt
KEY="releases/$DEPLOY_REVISION/$GITHUB_RUN_ATTEMPT/app.tar.gz"
aws s3api put-object --bucket "$ARTIFACT_BUCKET" --key "$KEY" \
  --body build/app.tar.gz --server-side-encryption AES256 --query ETag --output text
export ARTIFACT_KEY="$KEY"
export ARTIFACT_SHA256
ARTIFACT_SHA256=$(sha256sum build/app.tar.gz | cut -d ' ' -f 1)
python3 scripts/ec2_command.py > build/parameters.json
for attempt in $(seq 1 30); do
  ONLINE=$(aws ssm describe-instance-information \
    --filters "Key=InstanceIds,Values=$INSTANCE_ID" \
    --query 'InstanceInformationList[0].PingStatus' --output text)
  if [ "$ONLINE" = Online ]; then break; fi
  if [ "$attempt" = 30 ]; then printf 'EC2 is not online in Systems Manager\n' >&2; exit 1; fi
  sleep 10
done
COMMAND_ID=$(aws ssm send-command --document-name AWS-RunShellScript \
  --instance-ids "$INSTANCE_ID" --parameters file://build/parameters.json \
  --timeout-seconds 300 --query 'Command.CommandId' --output text)
printf 'CommandId=%s\n' "$COMMAND_ID"
aws ssm wait command-executed --command-id "$COMMAND_ID" --instance-id "$INSTANCE_ID" || {
  aws ssm get-command-invocation --command-id "$COMMAND_ID" --instance-id "$INSTANCE_ID" \
    --query '{Status:Status,Error:StandardErrorContent,Output:StandardOutputContent}'
  exit 1
}
aws ssm get-command-invocation --command-id "$COMMAND_ID" --instance-id "$INSTANCE_ID" \
  --query StandardOutputContent --output text > build/ec2-output.txt
cat build/ec2-output.txt
python3 - <<'PY'
# 応答が今回のコミットを返した場合だけ、デプロイ成功として扱います。
import json, os
from pathlib import Path
response = json.loads(Path('build/ec2-output.txt').read_text().strip().splitlines()[-1])
assert response['revision'] == os.environ['DEPLOY_REVISION'], response
assert response['target'] == 'ec2', response
assert response['environment'] == 'dev', response
print('EC2 revision verified')
PY

ジョブはコマンドが終わるのを待ち、出力された3項目が今回の値と一致したときだけ成功にします。

EC2上で実行するコマンドを作る scripts/ec2_command.py の全文

scripts/ec2_command.py

"""既知の配置先へ実行するコマンドをJSON化する。"""

import json
import os
import re

# 外部入力をシェルへ渡す前に、使える文字と配置先を制限します。
bucket, key, digest = (os.environ[x] for x in ("ARTIFACT_BUCKET", "ARTIFACT_KEY", "ARTIFACT_SHA256"))
if not re.fullmatch(r"miruky-[a-z]{16}", bucket):
    raise ValueError("Invalid artifact bucket")
if not re.fullmatch(r"releases/[0-9a-f]{40}/[0-9]+/app\.tar\.gz", key):
    raise ValueError("Invalid artifact key")
if not re.fullmatch(r"[0-9a-f]{64}", digest):
    raise ValueError("Invalid artifact digest")
commands = [
    "set -eu",
    f"aws s3 cp s3://{bucket}/{key} /tmp/deploy-app.tar.gz --only-show-errors",
    f"printf '%s  %s\\n' '{digest}' '/tmp/deploy-app.tar.gz' | sha256sum --check --status",
    "install -d -m 755 /opt/deploy-app",
    "tar -xzf /tmp/deploy-app.tar.gz -C /opt/deploy-app",
    "cat > /etc/systemd/system/deploy-app.service <<'UNIT'\n[Unit]\nDescription=Deployment example application\nAfter=network.target\n[Service]\nWorkingDirectory=/opt/deploy-app\nEnvironment=DEPLOY_TARGET=ec2\nEnvironment=DEPLOY_ENVIRONMENT=dev\nExecStart=/usr/bin/python3 /opt/deploy-app/server.py\nUser=nobody\nRestart=on-failure\n[Install]\nWantedBy=multi-user.target\nUNIT",
    "systemctl daemon-reload",
    "systemctl enable deploy-app.service >/dev/null 2>&1",
    "systemctl restart deploy-app.service",
    "for attempt in $(seq 1 20); do if curl --silent --fail http://127.0.0.1:8080/health; then echo; break; fi; if [ \"$attempt\" = 20 ]; then exit 1; fi; sleep 1; done",
]
print(json.dumps({"commands": commands}))

インスタンスの状態は実行中、インスタンスタイプはt3.microです。このインスタンスはIAM ロールに設定したEC2用のロールで、Systems ManagerとS3へ接続します。

OutputのJSONは、targetがec2、environmentがdevで、revisionも今回のコミットIDでした。レスポンスコードも0なので、インスタンスの中で応答を取得するところまで成功しています。

8. ECS Fargateでは、ロールバックも起きました

8-1. ヘルスチェックに落ちたら、どうなるの？

ECSのジョブはコミットIDと再実行番号を付けたタグでイメージをECRに保存し、appコンテナのイメージだけを差し替えたタスク定義でサービスを更新します。

ヘルスチェックで検査するのは/healthの3項目です。ジョブのほうは、今回のイメージで動くタスクが実行中で、しかも正常かを確認します。

scripts/verify_ecs.py

"""サービスの稼働数だけでなく、今回のイメージとコンテナの正常性を確認する。"""

import json
import os
import subprocess
import time


def aws(*arguments):
    return json.loads(subprocess.check_output(["aws", "ecs", *arguments], text=True))


cluster, service, expected = (os.environ[x] for x in ("ECS_CLUSTER", "ECS_SERVICE", "EXPECTED_IMAGE"))
# 更新前のタスクを成功と誤認しないよう、今回のイメージを照合します。
for attempt in range(60):
    task_arns = aws("list-tasks", "--cluster", cluster, "--service-name", service)["taskArns"]
    if task_arns:
        tasks = aws("describe-tasks", "--cluster", cluster, "--tasks", *task_arns)["tasks"]
        matching = [task for task in tasks if any(container.get("image") == expected for container in task["containers"])]
        if matching and all(task.get("lastStatus") == "RUNNING" and task.get("healthStatus") == "HEALTHY" for task in matching):
            definition = aws("describe-task-definition", "--task-definition", matching[0]["taskDefinitionArn"])["taskDefinition"]
            app = next(container for container in definition["containerDefinitions"] if container["name"] == "app")
            environment = {item["name"]: item["value"] for item in app.get("environment", [])}
            if environment.get("DEPLOY_ENVIRONMENT") != "dev":
                raise RuntimeError("The running task definition is not configured for dev")
            print(json.dumps({"status": "RUNNING", "health": "HEALTHY", "revision": os.environ["DEPLOY_REVISION"]}))
            print("ECS image and health verified")
            break
    if attempt == 59:
        raise RuntimeError("The current image did not become healthy")
    time.sleep(5)

わざわざイメージまで照合するのには理由があります。サーキットブレーカーで復旧したあとは前のリビジョンのタスクでサービスが安定することがあり、稼働数だけを見ると、今回の更新が成功したように見えてしまうからです。

実際に、HTTPサーバーを開始せずに600秒待つだけのCMDへ変えて反映してみました。ユニットテストは通っても、このコンテナはヘルスチェックに応答できません（ユニットテストだけでは気づけないのが、なかなか厄介です）。

Task failed container health checksとロールバックされましたが表示され、ターゲットはリビジョン3、ソースはリビジョン2です。前のリビジョンのタスクが1つ動いている状態に戻りました。

GitHub Actionsのほうも、更新を待つところで失敗しました。

{"state":"TIMEOUT","observedResponses":{"200: OK":7},"reason":"Waiter has timed out"}

ECSは前のリビジョンへ戻りましたが、EC2とLambdaには、ヘルスチェックに失敗したのと同じコミットが反映されたままです。そこでCMDを戻したPRをマージし、3つの実行先が同じコミットで動くことを確認しました。

同じ失敗を試すなら、DockerfileのCMDをCMD ["python", "-c", "import time; time.sleep(600)"]に変えます。失敗用の変更も復旧用の変更もfeatureブランチからのPRでマージしてください。

git switch develop
git pull --ff-only origin develop
git switch -c feature/ecs-startup-check
# DockerfileのCMDを待機する処理へ変更してからコミットします。
git add Dockerfile
git commit -m "Check an unhealthy container startup"
git push -u origin feature/ecs-startup-check
gh pr create --base develop --title "Check an unhealthy container startup" --body "ECSのヘルスチェックによる旧版への復帰を確認します。"
gh pr checks --watch && gh pr merge --merge

ECSのジョブが失敗したのを確認したら、CMDをCMD ["python", "server.py"]に戻し、同じ手順でPRをマージします。

git switch develop
git pull --ff-only origin develop
git switch -c feature/restore-server-cmd
# DockerfileのCMDをserver.pyの起動へ戻してから続けます。
git add Dockerfile
git commit -m "Restore the HTTP server command"
git push -u origin feature/restore-server-cmd
gh pr create --base develop --title "Restore the HTTP server command" --body "CMDをserver.pyの起動へ戻します。"
gh pr checks --watch && gh pr merge --merge

復旧後は、前回のステータスが実行中、ヘルスステータスが正常になりました。タスク定義リビジョン4のタスクが1つ動いていて、ヘルスチェックにも通っています。

イメージ URIの末尾は復旧用PRのマージコミットIDと再実行番号の1です。念のため、起動ログでもコミットIDと環境の値を確認します。

readyのログにはrevisionに復旧用PRのコミットID、targetにecs、environmentにdevが出ていました。ただ、これはプロセスが起動したときの記録なので、HTTPの応答が正常かどうかはヘルスチェックで判断します。

9. Lambdaへのデプロイ

9-1. 更新が終わるのを待ってから呼び出しましょう

Lambdaのジョブがコードの更新に使うのは、直前に取得したRevisionIdです。取得してから更新するまでの間にコンソールなどで関数が更新されていれば、上書きせずに拒否されます。

ZIPのルートにはlambda_function.pyをindex.pyとしてコピーしたものと、app.py、revision.txtを入れます。更新の完了を待ってから同期呼び出しをして、InvokeのStatusCodeが200でもFunctionErrorがあれば失敗です。応答の3項目も今回の値と照合します。

scripts/deploy_lambda.sh

#!/usr/bin/env bash
# 関数の実行ロールを保持したままコードを更新し、応答のコミットIDを確認します。
set -euo pipefail
: "${FUNCTION_NAME:?}" "${DEPLOY_REVISION:?}"
mkdir -p build/lambda
cp app.py build/lambda/app.py
cp lambda_function.py build/lambda/index.py
printf '%s\n' "$DEPLOY_REVISION" > build/lambda/revision.txt
(cd build/lambda && zip -q ../lambda.zip app.py index.py revision.txt)
REVISION_ID=$(aws lambda get-function-configuration --function-name "$FUNCTION_NAME" \
  --query RevisionId --output text)
aws lambda update-function-code --function-name "$FUNCTION_NAME" \
  --revision-id "$REVISION_ID" --zip-file fileb://build/lambda.zip \
  --query LastUpdateStatus --output text
# 許可したGetFunctionConfigurationで更新完了を確認します。
aws lambda wait function-updated --function-name "$FUNCTION_NAME"
aws lambda invoke --function-name "$FUNCTION_NAME" --cli-binary-format raw-in-base64-out \
  --payload '{"name":"miruky"}' build/lambda-response.json > build/lambda-invocation.json
python3 - <<'PY'
# APIの受付成功に加え、関数の例外と返却値の内容を確認します。
import json, os
from pathlib import Path
invocation = json.loads(Path('build/lambda-invocation.json').read_text())
response = json.loads(Path('build/lambda-response.json').read_text())
assert invocation.get('StatusCode') == 200, invocation
assert not invocation.get('FunctionError'), response
assert response['revision'] == os.environ['DEPLOY_REVISION'], response
assert response['target'] == 'lambda', response
assert response['environment'] == 'dev', response
print(json.dumps(response, ensure_ascii=False))
print('Lambda revision verified')
PY

ファイル一覧にはapp.py、index.py、revision.txtがあり、revision.txtの中身は復旧用PRのマージコミットでした。配置した内容がそのまま入っています。

関数が出力したJSONも、targetがlambda、environmentがdevで、revisionは今回のコミットIDと一致しました。これで、3つの実行先すべてで同じコミットが動いていることを確認できました。

10. 本番で使う前に足したいこと

今回自動化したのは、CIから3つの実行先への反映と、その照合までです。GitHub用のロールに与えたのも、既存の実行先にコードを反映する権限だけです。

CIが失敗したPRのマージを止めたり、直接のプッシュを制限したりするには、developにブランチの保護規則を追加します。使える規則はプランやリポジトリの公開範囲によって異なります。

GitHubのデプロイ環境で承認ルールを使うなら、ジョブにenvironmentの指定が必要です。そうするとsubが環境名を含む形式に変わるので、信頼ポリシーとAllowedPatternも変更します。

EC2上のアプリはサービスを再起動している間、応答が止まります。本番ではこの停止への対策が要り、ECSの前段にはロードバランサーなどが候補です。Lambdaをバージョンとエイリアスで反映、復旧したい場合は、その権限と確認の処理の追加が必要です。

おわりに

ここまでお付き合いいただきありがとうございます。

CIはfeatureの先頭とPRのマージ結果を検査し、CDはdevelopで受け取ったコミットをdev環境へ反映します。8章のようにECSだけが前のリビジョンへ戻ることもあるので、反映したあとは応答や稼働中のイメージで、どのコミットが動いているかを確認してみてください（3つの実行先のうちECSだけが自分で前のリビジョンに戻るのを見ると、サーキットブレーカーってよくできているな、と思います）。