😃
CodeDeployのdeployment_idから辿って、ECSサービスタスク内のコンテナのイメージタグを取得したくなった

職人
2025/05/10に公開
 経緯CodeDeployを使ったブルーグリーンデプロイをしている
GitHub Actionsを使ってデプロイを行っている
デプロイのワークフローではコンテナのビルドのあとにデプロイという順番で処理している
コンテナをビルドしてECRにプッシュする時にイメージタグを指定している
ブルーグリーンデプロイでデプロイされるECSサービスタスクとは別のECSサービスタスクをトラフィック切り替えのタイミングでデプロイを行いたい要件があり、訳あってイメージタグは同様のものを使いたい
新コンテナデプロイ完了後のトラフィック切り替え待ちの状況でCodeDeployからのイベントをSNSがキャッチし、Lambda関数を通じてSlackにメッセージを送信しているが、そのメッセージのボタンからトラフィック切り替えのワークフローを起動したい
ボタンを押した時に送信されたパラメータにイメージタグを仕込みたかったので、Lambda関数内でイメージタグを取得しないと、新コンテナと同様のイメージタグを使ったデプロイができない状況だった
なのでLambda関数内で今現在デプロイ中の情報から、辿ってECSサービスタスク内のコンテナのイメージタグを取得したかった

 代案普通にGitHubActionsのアーティファクトに保存し、トラフィック切り替えのワークフローでそのアーティファクトを参照すれば良かったというオチです
参考: https://docs.github.com/ja/actions/writing-workflows/choosing-what-your-workflow-does/storing-and-sharing-data-from-a-workflow#uploading-build-and-test-artifacts

Lambda関数内で頑張ってイメージタグを取得したコードを供養させてください

 実際のコード
def lambda_handler(event, context):
    # イメージタグ取得にフォーカスしているので色々と省略してます
    message = event['Records'][0]['Sns']['Message']
    message_dict = json.loads(message)
    deployment_id = message_dict['deploymentId']
    image_tag = get_image_tag(deployment_id)

def get_image_tag(deployment_id):
    try:
        codedeploy_client = boto3.client('codedeploy')
        ecs_client = boto3.client('ecs')
        # デプロイメント情報の取得
        deployment_response = codedeploy_client.get_deployment(
            deploymentId=deployment_id
        )

        # デプロイメントから必要な情報を取得
        application_name = deployment_response['deploymentInfo']['applicationName']
        revision = deployment_response['deploymentInfo']['revision']

        revision_response = codedeploy_client.get_application_revision(
            applicationName=application_name,
            revision={
                'revisionType': 'AppSpecContent',
                'appSpecContent': {
                    'sha256': revision['appSpecContent']['sha256']
                }
            }
        )

        # AppSpecの内容を取得
        appspec_content = json.loads(revision_response['revision']['appSpecContent']['content'])

        # タスク定義ARNを取得
        task_definition_arn = appspec_content['Resources'][0]['TargetService']['Properties']['TaskDefinition']

        # タスク定義の詳細を取得
        task_def_response = ecs_client.describe_task_definition(
            taskDefinition=task_definition_arn
        )

        # コンテナ定義からイメージタグを検索
        container_definitions = task_def_response['taskDefinition']['containerDefinitions']
        image_tag = None

        for container in container_definitions:
            # CONTAINER_NAME イメージタグを取得したいコンテナの名前、今回は定数で指定している
            if container['name'] == CONTAINER_NAME:
                # イメージURIからタグを抽出
                image_uri = container['image']
                # タグを抽出（例: xxx.dkr.ecr.region.amazonaws.com/image:tag から tagを取得）
                tag_match = re.search(r':([^:]+)$', image_uri)
                if tag_match:
                    image_tag = tag_match.group(1)
                break
        if image_tag is None:
            raise Exception("image_tagが取得できませんでした")
        return image_tag

    except Exception as e:
        print(f"image_tag取得処理でエラーが発生しました: {e}")
        raise e

 ざっくり解説CodeDeployのデプロイメントIDからアプリケーション名とリビジョンを取得する
リビジョンからappspecの内容を取得する
appspecにあるタスク定義を辿って、任意のコンテナのイメージタグを取得する

 参考https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/codedeploy/client/get_deployment.html

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/codedeploy/client/get_application_revision.html

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ecs/client/describe_task_definition.html
SMARTCAMP Engineer BlogPublication
SMARTCAMPに所属するエンジニアの個人記事を集めています。記事の内容は個人の見解であり、社内でのレビュー等は行っておりません。
経緯

代案

実際のコード

ざっくり解説

参考

Discussion
