Zenn
SimpleForm Tech BlogPublicationへの投稿
🤖

Terraform/Terragrunt の自動 plan・apply ができるまで ② 〜マッピング編〜

Jun Irie
2024/12/15に公開
GitHub Actions
Terraform
IaC
CI/CD
Terragrunt
tech

本記事は SimpleForm Advent Calendar 2024 の 15 日目の記事です。

はじめに

こんにちは、シンプルフォームでインフラエンジニアをやっている入江 純 (@jirtosterone) です。

前回の記事で自動 plan について書きました。その中に、Terraform ファイルのみが修正された時にどの terragrunt.hcl を plan すれば良いかが分からない という課題があり、その解決策として 全ての terragrunt.hcl ファイルと Terraform ファイルとのマッピングを作成 しました。今回はその具体的な方法についてご紹介します。

前回の記事については以下をご覧ください。

https://zenn.dev/simpleform_blog/articles/4e2c61fadee249

対象読者

  • Terraform/Terragrunt でインフラ管理されている方
  • Terragrunt の CI/CD を運用・検討されている方

結論

時間が無い方のために結論から。

  • GitHub Actions で定期的に Terraform ファイルと terragrunt.hcl のマッピングを作成するようにした。
  • マッピングには terragrunt render-json コマンドを利用した。
  • マッピングした情報は DynamoDB に保存するようにした。
  • 保存した情報を取得して Terraform ファイルのみが更新されても対応する terragrunt.hcl に対して plan できるようにした。

構成

大きく保存と取得の 2 つの機能で構成されています。

  1. マッピング情報の保存 (upsert): Terraform ファイルと terragrunt.hclのマッピングを作成して DynamoDB に保存する。スケジュール起動にて実行される。
  2. マッピング情報の取得 (query): DynamoDB に保存したマッピング情報を取得する。自動 plan の過程で実行される。

action の構成概要

action のディレクトリ構成は以下の通りです。ここで番号を振っているコードについては 実装 で解説していきます。

マッピングを構成するワークフローと action の構成
.github
│── terragrunt-upsert-map.yml  # マッピング情報保存 action をスケジュール実行するワークフロー
├── terragrunt-plan.yml  # PR 作成時に自動 plan を行うワークフロー
└── tf_module_mapper  # マッピング情報の保存・取得を行う action
    │── upsert  # 1.1 マッピング情報を保存する action
    │   └── action.yml
    ├── query  # 2.1 マッピング情報を取得する action
    │   └── action.yml
    └── scripts  # 処理のメインを担うスクリプト (Poetry でパッケージ管理)
        ├── poetry.lock
        ├── poetry.toml
        ├── pyproject.toml
        └── src
            ├── upsert.py  # 1.2 upsert の実態
            ├── query.py  # 2.2 query の実態
            └── utils
                ├── dynamodb_handler.py  # 3.1 DynamoDB に対する処理を扱うスクリプト
                ├── terragrunt_handler.py  # 3.2 Terragrunt コマンドを扱うスクリプト
                ├── path_handler.py  # 3.3 ファイルパスの整形を扱うスクリプト
                └── logger.py

DynamoDB には以下のような情報が Terraform ファイルが格納されているディレクトリごとに保存されます。実際には属性を示すキーもありますが簡単のために省略しています。

マッピング情報の例
{
  "source": "modules/ecs/app_sample",  # primary key
  "terragruntPath": "envs/{env}/ecs/app_impl/terragrunt.hcl",
  "dependencies": [
    { "lb": "envs/{env}/load_balancer/app_impl" },
    { "network": "envs/{env}/network" }
  ]
}

実装

構成で番号を振った action とスクリプトについてコードで解説します。

1. マッピング情報の保存 (upsert)

マッピング情報を保存するための action および Python スクリプトです。

1.1 マッピング情報を保存する action

action ではセットアップと Python スクリプトの実行を行っています。

tf_module_mapper/upsert/action.yml
name: Terraform module map upsert
description: Update or insert Terraform map between HCL and modules

inputs:
  # DynamoDB にアクセスするための IAM ロール
  aws_role_arn:
    required: true
    type: string
  # マッピング情報を作成するルートパス
  search_path:
    required: true
    type: string
  # マッピング情報を保存する DynamoDB テーブル名
  dynamodb_table_name:
    required: true
    type: string

runs:
  using: "composite"
  steps:
    - uses: actions/checkout@v4

    - uses: aws-actions/configure-aws-credentials@v4
      with:
        aws-region: ap-northeast-1
        role-to-assume: ${{ inputs.aws_role_arn }}

    # Terraform のセットアップ
    - uses: hashicorp/setup-terraform@v3
      with:
        terraform_version: "1.x.x"
        terraform_wrapper: false

    # Terragrunt のセットアップ
    - uses: autero1/action-terragrunt@v3
      with:
        terragrunt-version: "0.xx.xx"

    # Poetry のセットアップ
    - run: pipx install poetry
      shell: bash

    # Python のセットアップ
    - uses: actions/setup-python@v5
      with:
        python-version: "3.xx"
        cache: "poetry"

    # 依存パッケージのインストール
    - run: poetry install
      shell: bash
      working-directory: ${{ github.workspace }}/.github/actions/tf_module_mapper/scripts

    # upsert スクリプトの実行
    - run: poetry run python src/upsert.py ${{ inputs.dynamodb_table_name }} ${{ inputs.search_path }}
      shell: bash
      env:
        PYTHONUNBUFFERED: "1"
      working-directory: ${{ github.workspace }}/.github/actions/tf_module_mapper/scripts

1.2 upsert の実態 (Python スクリプト)

次に処理の実態となる Python スクリプトを順を追って説明します。ちょいちょい出てくる src.utils.xxxx については 3. utils の正体 でまとめてご紹介します。

ルートパス配下の terragrunt.hcl ファイルが格納してあるディレクトリの一覧を取得します。

ディレクトリ一覧を取得
from os import path
from glob import glob
from src.utils.dynamodb_handler import DynamoDbHandler

TERRAGRUNT_FILE_NAME = "terragrunt.hcl"

# 探索対象のルートパスを設定
search_abs_path = path.abspath(search_path)

# DynamoDB への接続設定
dynamodb_handler = DynamoDbHandler(dynamodb_table_name)

# terragrunt.hcl ファイルを検索
tg_abs_paths = glob(f"{search_abs_path}/**/{TERRAGRUNT_FILE_NAME}", recursive=True)

得られたディレクトリ一覧の一つ一つに対して terragrunt render-json コマンドを実行して source, dependency 等を抽出して DynamoDB に保存します。

マッピング情報を保存
from src.utils.path_handler import format_path
from src.utils.terragrunt_handler import render_terragrunt

# terragrunt.hcl ファイル分繰り返し
upsert_count = 0
failed_paths = []
for tg_abs_path in tg_abs_paths:
    try:
        # terragrunt render-json コマンドで terragrunt.hcl を構造化する
        tg_config = render_terragrunt(tg_abs_path)
        # 構造化して取得したパスの必要な部分だけを抽出する
        tg_path = format_path(tg_abs_path, "envs/")

        # マッピングデータ
        mapping_item: DynamoDbHandler.Item = {
            "terragruntPath": tg_path,
            "dependencies": [],
        }

        # terraform.source に指定されたパスを抽出
        if terraform_source := tg_config.get("terraform", {}).get("source"):
            source = format_path(terraform_source, "modules/")
        else:
            msg = f"terraform.source does not exist: {tg_abs_path}"
            raise RuntimeError(msg)

        # dependency.config_path に指定されたパスを抽出
        if dependencies := tg_config.get("dependency"):
            mapping_item["dependencies"] = [
                {k: format_path(v["config_path"], "envs/")} for k, v in dependencies.items() if "config_path" in v
            ]

        key: DynamoDbHandler.PartitionKey = {"source": source}

        # DynamoDB に保存
        dynamodb_handler.upsert_dynamodb_item(key, mapping_item)
        upsert_count += 1

    except Exception as e:
        failed_paths.append(tg_path)
upsert.py の全容
import argparse
import json
from glob import glob
from os import path

from src.utils.dynamodb_handler import DynamoDbHandler
from src.utils.logger import get_logger
from src.utils.path_handler import format_path
from src.utils.terragrunt_handler import render_terragrunt

TERRAGRUNT_FILE_NAME = "terragrunt.hcl"

logger = get_logger(__name__)


def main(dynamodb_table_name: str, search_path: str) -> tuple[int, list[str]]:
    # 探索対象のルートパスを設定
    search_abs_path = path.abspath(search_path)

    # DynamoDB への接続設定
    dynamodb_handler = DynamoDbHandler(dynamodb_table_name)

    # terragrunt.hcl ファイルを検索
    tg_abs_paths = glob(f"{search_abs_path}/**/{TERRAGRUNT_FILE_NAME}", recursive=True)
    logger.info(f"Found terragrunt.hcl files: {len(tg_abs_paths)}")

    # terragrunt.hcl ファイル分繰り返し
    upsert_count = 0
    failed_paths = []
    for tg_abs_path in tg_abs_paths:
        try:
            # terragrunt render-json コマンドで terragrunt.hcl を構造化する
            logger.info(f"Target: {tg_abs_path}")
            tg_config = render_terragrunt(tg_abs_path)
            logger.info(f"Rendered data: {json.dumps(tg_config, indent=2, ensure_ascii=False)}")

            # 構造化して取得したパスの必要な部分だけを抽出する
            tg_path = format_path(tg_abs_path, "envs/")

            # マッピングデータ
            mapping_item: DynamoDbHandler.Item = {
                "terragruntPath": tg_path,
                "dependencies": [],
            }

            # terraform.source に指定されたパスを抽出
            if terraform_source := tg_config.get("terraform", {}).get("source"):
                source = format_path(terraform_source, "modules/")
            else:
                msg = f"terraform.source does not exist: {tg_abs_path}"
                raise RuntimeError(msg)

            # dependency.config_path に指定されたパスを抽出
            if dependencies := tg_config.get("dependency"):
                mapping_item["dependencies"] = [
                    {k: format_path(v["config_path"], "envs/")} for k, v in dependencies.items() if "config_path" in v
                ]

            key: DynamoDbHandler.PartitionKey = {"source": source}
            logger.info(f"Key: {json.dumps(key, indent=2, ensure_ascii=False)}")
            logger.info(f"Mappings: {json.dumps(mapping_item, indent=2, ensure_ascii=False)}")

            # DynamoDB に保存
            dynamodb_handler.upsert_dynamodb_item(key, mapping_item)
            upsert_count += 1

        except Exception as e:
            failed_paths.append(tg_path)
            logger.exception(e)

    return upsert_count, failed_paths


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("dynamodb_table_name", type=str, help="DynamoDB table name")
    parser.add_argument("search_path", type=str, help="Search path for terragrunt.hcl files")
    args = parser.parse_args()

    logger.info(f"Search path: {args.search_path}")
    upsert_count, failed_paths = main(args.dynamodb_table_name, args.search_path)
    logger.info(f"Upsert count: {upsert_count}, Failed count: {len(failed_paths)}")

    if len(failed_paths) > 0:
        logger.error("Failed path:")
        for failed_path in failed_paths:
            logger.error(f"  {failed_path}")

2. マッピング情報の取得 (query)

マッピング情報を取得するための action と処理の実態を担う Python スクリプトです。

2.1 マッピング情報を取得する action

action ではセットアップと Python スクリプトの実行を行っています。やっていることは upsert と同じです。

query/action.yml
name: Terraform module map query
description: Query HCL paths by the related Terraform modules

inputs:
  # DynamoDB にアクセスするための IAM ロール
  aws_role_arn:
    required: true
    type: string
  # 探索対象の Terraform ファイルパス
  module_paths:
    required: true
    type: string
  # マッピング情報を取得する DynamoDB テーブル名
  dynamodb_table_name:
    required: true
    type: string

outputs:
  # マッピング情報が見つかった terragrunt.hcl のパス
  found_paths:
    description: HCL paths found by the modules
    value: ${{ steps.query.outputs.found_paths }}
  # マッピング情報が見つからなかった Terraform ファイルパス
  not_found_paths:
    description: HCL paths not found by the modules
    value: ${{ steps.query.outputs.not_found_paths }}
  # マッピング情報の取得時にエラーが発生した Terraform ファイルパス
  failed_paths:
    description: module paths which failed to query
    value: ${{ steps.query.outputs.failed_paths }}

runs:
  using: "composite"
  steps:
    - uses: actions/checkout@v4

    - uses: aws-actions/configure-aws-credentials@v4
      with:
        aws-region: ap-northeast-1
        role-to-assume: ${{ inputs.aws_role_arn }}

    # Terraform のセットアップ
    - uses: hashicorp/setup-terraform@v3
      with:
        terraform_version: "1.x.x"
        terraform_wrapper: false

    # Terragrunt のセットアップ
    - uses: autero1/action-terragrunt@v3
      with:
        terragrunt-version: "0.xx.xx"

    # Poetry のセットアップ
    - run: pipx install poetry
      shell: bash

    # Python のセットアップ
    - uses: actions/setup-python@v5
      with:
        python-version: "3.xx"
        cache: "poetry"

    # 依存パッケージのインストール
    - run: poetry install
      shell: bash
      working-directory: ${{ github.workspace }}/.github/actions/tf_module_mapper/scripts

    # query スクリプトの実行
    - id: query
      run: |
        result=$(poetry run python src/query.py ${{ inputs.dynamodb_table_name }} --changed_tf_files ${{ inputs.module_paths }} --root_dir ${{ github.workspace }})
        # 出力用
        output_found_paths=$(echo "${result}" | jq .found_paths | sed "s/\/terragrunt.hcl//g") && echo "found_paths: ${output_found_paths}"
        output_not_found_paths=$(echo "${result}" | jq .not_found_paths | sed "s/\/terragrunt.hcl//g") && echo "not_found_paths: ${output_not_found_paths}"
        output_failed_paths=$(echo "${result}" | jq .failed_paths | sed "s/\/terragrunt.hcl//g") && echo "failed_paths: ${output_failed_paths}"
        # output 用
        found_paths=$(echo "${output_found_paths}" | jq -c)
        not_found_paths=$(echo "${output_not_found_paths}" | jq -c)
        failed_paths=$(echo "${output_failed_paths}" | jq -c)
        echo "found_paths=${found_paths}" >> $GITHUB_OUTPUT
        echo "not_found_paths=${not_found_paths}" >> $GITHUB_OUTPUT
        echo "failed_paths=${failed_paths}" >> $GITHUB_OUTPUT
      shell: bash
      env:
        PYTHONUNBUFFERED: "1"
      working-directory: ${{ github.workspace }}/.github/actions/tf_module_mapper/scripts

2.2 query の実態 (Python スクリプト)

upsert の逆を行います。

探索対象の絞り込み
from os import path
from typing import TypedDict

class TgPath(TypedDict):
    found_paths: list[str]
    not_found_paths: list[str]
    failed_paths: list[str]

# 戻り値の箱を作成
result: TgPath = {
    "found_paths": [],
    "not_found_paths": [],
    "failed_paths": [],
}

# 与えられたファイル群から tf ファイルのみを抽出
selected_files = [f for f in changed_tf_files if f.endswith(".tf")]

# 探索対象がない場合は空の結果を返却
if not selected_files:
    return result

# 探索対象のルートパスを設定
changed_tf_abs_paths = [path.abspath(path.join(root_dir, f)) for f in selected_files]
terragrunt.hcl ファイルを検索
from src.utils.dynamodb_handler import DynamoDbHandler
from src.utils.path_handler import format_path

# DynamoDB への接続設定
dynamodb_handler = DynamoDbHandler(dynamodb_table_name)

# 変更のあった tf ファイルから対象の terragrunt.hcl ファイルを検索
for tf_abs_path in changed_tf_abs_paths:
    try:
        # パスを整形してマッピング情報を探索
        formatted_path = format_path(path.dirname(tf_abs_path), "modules/")
        tg_path = dynamodb_handler.find_terragrunt_path_by_source(formatted_path)

        if tg_path:
            result["found_paths"].append(tg_path)
        else:
            result["not_found_paths"].append(formatted_path)

    except Exception as e:
        result["failed_paths"].append(formatted_path)

# 空文字削除
result["found_paths"] = list(filter(None, result["found_paths"]))

# 重複排除
result["found_paths"] = list(set(result["found_paths"]))
query.py の全容
query.py
import argparse
import json
from os import path
from typing import TypedDict

from src.utils.dynamodb_handler import DynamoDbHandler
from src.utils.logger import get_logger
from src.utils.path_handler import format_path

logger = get_logger(__name__)


class TgPath(TypedDict):
    found_paths: list[str]
    not_found_paths: list[str]
    failed_paths: list[str]


def main(dynamodb_table_name: str, changed_tf_files: list[str], root_dir: str = ".") -> TgPath:
    result: TgPath = {
        "found_paths": [],
        "not_found_paths": [],
        "failed_paths": [],
    }

    # 与えられたファイル群から tf ファイルのみを抽出
    logger.info(f"Changed Terraform files (Row): {changed_tf_files}")
    selected_files = [f for f in changed_tf_files if f.endswith(".tf")]

    # 探索対象がない場合は空の結果を返却
    if not selected_files:
        logger.warning("No terraform files in changed files specified")
        return result
    logger.info(f"Changed Terraform files (Selected): {selected_files}")

    # 探索対象のルートパスを設定
    changed_tf_abs_paths = [path.abspath(path.join(root_dir, f)) for f in selected_files]
    logger.info(f"Changed terraform files (Absolute path): {changed_tf_abs_paths}")

    # DynamoDB への接続設定
    dynamodb_handler = DynamoDbHandler(dynamodb_table_name)

    # 変更のあった tf ファイルから対象の terragrunt.hcl ファイルを検索
    for tf_abs_path in changed_tf_abs_paths:
        try:
            # パスを整形してマッピング情報を探索
            formatted_path = format_path(path.dirname(tf_abs_path), "modules/")
            logger.info(f"Target: {formatted_path}")
            tg_path = dynamodb_handler.find_terragrunt_path_by_source(formatted_path)
            logger.info(f"Terragrunt file: {tg_path}")

            if tg_path:
                result["found_paths"].append(tg_path)
            else:
                result["not_found_paths"].append(formatted_path)

        except Exception as e:
            result["failed_paths"].append(formatted_path)
            logger.exception(e)

    # 空文字削除
    result["found_paths"] = list(filter(None, result["found_paths"]))

    # 重複排除
    result["found_paths"] = list(set(result["found_paths"]))

    return result


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("dynamodb_table_name", type=str, help="DynamoDB table name")
    parser.add_argument(
        "--changed_tf_files", required=True, nargs="*", type=str, help="Terraform file path have been changed"
    )
    parser.add_argument("--root_dir", type=str, default=".", help="Root directory of Terraform project")
    args = parser.parse_args()

    tg_paths: TgPath = main(args.dynamodb_table_name, args.changed_tf_files, args.root_dir)
    print(json.dumps(tg_paths))

3. utils の正体

これまでにご紹介した upsert, query スクリプトで呼び出している utils の内容についてもざっと触れておきます。

なお、logger.py については特別なことはやっておらず、logging.Logger オブジェクトを作成しているだけなので割愛します。

3.1 DynamoDB に対する処理を扱うスクリプト

dynamodb_handler.py
from typing import TypedDict

import boto3


class DynamoDbHandler:
    """DynamoDB テーブルを操作するクラス"""

    class PartitionKey(TypedDict):
        source: str

    class Item(TypedDict):
        terragruntPath: str
        dependencies: list[dict[str, str]]

    def __init__(self, dynamodb_table_name: str):
        dynamodb = boto3.resource("dynamodb")
        self.table = dynamodb.Table(dynamodb_table_name)

    def upsert_dynamodb_item(self, key: PartitionKey, item: Item) -> None:
        """DynamoDB テーブルにアイテムを挿入または更新する"""
        response = self.table.get_item(Key=key)
        if response.get("Item"):
            # Update
            self.table.update_item(
                Key=key,
                UpdateExpression="SET #s=:s, #d=:d",
                ExpressionAttributeNames={
                    "#s": "terragruntPath",
                    "#d": "dependencies",
                },
                ExpressionAttributeValues={
                    ":s": item["terragruntPath"],
                    ":d": item["dependencies"],
                },
            )
        else:
            # Insert
            self.table.put_item(Item=(key | item))

    def find_terragrunt_path_by_source(self, source: str) -> str | None:
        """source に一致する terragruntPath を取得する"""
        result = None

        key: DynamoDbHandler.PartitionKey = {"source": source}
        response = self.table.get_item(Key=key)
        if response.get("Item"):
            result = response["Item"]["terragruntPath"]

        return result

3.2 Terragrunt コマンドを扱うスクリプト

terragrunt_handler.py
import json
import subprocess
from os import path

from src.utils.logger import get_logger

logger = get_logger(__name__)


def render_terragrunt(tg_path: str) -> dict:
    """terragrunt.hcl ファイルをレンダリングして JSON に変換する"""
    tg_dir_path = path.dirname(tg_path)

    ret = subprocess.run(["terragrunt", "render-json"], cwd=tg_dir_path, capture_output=True)
    if ret.returncode != 0:
        raise RuntimeError(ret.stderr.decode("utf-8"))

    logger.info(ret.stdout.decode("utf-8"))

    rendered_json_path = path.join(tg_dir_path, "terragrunt_rendered.json")
    return json.load(open(rendered_json_path, "r"))

3.3 ファイルパスの整形を扱うスクリプト

path_handler.py
def format_path(path: str, remove_position_str: str) -> str:
    """path 文字列を整形する"""
    # 不要な prefix を削除
    result = path[path.index(remove_position_str) :]

    # '//' を '/' に変換
    result = result.replace("//", "/")

    # 最後の '/' を削除
    if result.endswith("/"):
        result = result[:-1]

    return result

さいごに

Terragrunt の CI/CD は一筋縄でいかず、また事例も少ないため、苦戦しながらもこのような形で課題を解決しました。Terragrunt を運用されているどなたかの参考になれば幸いです。
ちなみに、これは数ある課題の一端に過ぎません 😇

さいごのさいごに...、本記事を書いていて思ったよりコードのボリュームが大きくなってしまいました。その内 GitHub に登録して公開したいと思います。

SimpleForm Tech Blog
SimpleForm Tech BlogPublication

リアルタイム法人調査システム「SimpleCheck」を開発・運営するシンプルフォーム株式会社の開発チームのメンバーが、日々の開発で得た知見や試してみた技術などについて発信していきます。 Publication 運用への移行前の記事は zenn.dev/simpleform からご覧ください。

Discussion