IaC Generatorで使用されるリソーススキャンをboto3から使ってAWSリソース一覧をCSVで出力してみた

IaC Generatorで使用されるリソーススキャンをboto3から使ってAWSリソース一覧をCSVで出力してみた

Clock Icon2025.02.13

こんちには。

データ事業本部 インテグレーション部 機械学習チームの中村( @nokomoro3 )です。

今回は、IaC Generatorで使用されるリソーススキャンをboto3から使って、リソース一覧をCSVファイルで出力してみたいと思います。

IaC Generatorとは

IaC Generatorとは、AWS上にデプロイされている既存リソースからCloudFormationのテンプレートを作成する機能です。

https://dev.classmethod.jp/articles/cloudformation-iac-generate/

実際には、CloudFormationのリソーススキャンをした後に、リソースを選択してテンプレートを作成していくのですが、今回はリソース一覧を取得したいだけですので、リソーススキャンを使っていく感じになります。

実際のコード

リージョンごとに取得する前提となっていますので、リソースを取得したいAWS_PROFILEとAWS_REGIONをそれぞれ指定した後に以下を実行されてください。

import time
import boto3
import json
import polars as pl
from functools import wraps
import time

def wait_until_complete(timeout=3600, interval=10):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            start_time = time.time()
            while True:
                response = func(*args, **kwargs)
                elapsed_time = int(time.time() - start_time)

                if response["Status"] == "COMPLETE":
                    print(f"{elapsed_time=} [sec] finished!! (status={response['Status']})")
                    return response

                if timeout and elapsed_time >= timeout:
                    raise TimeoutError(f"Timeout after {timeout} seconds")

                print(f"{elapsed_time} [sec] waiting... (status={response['Status']})")
                time.sleep(interval)
        return wrapper
    return decorator

def main():

    client = boto3.client("cloudformation")

    # スキャン開始
    response = client.start_resource_scan()
    ResourceScanId = response["ResourceScanId"]
    print(f"{ResourceScanId=}")

    @wait_until_complete()
    def check_scan_status(client, ResourceScanId):
        return client.describe_resource_scan(ResourceScanId=ResourceScanId)
    response = check_scan_status(client=client, ResourceScanId=ResourceScanId)

    # リソース取得
    data = []
    paginator = client.get_paginator("list_resource_scan_resources")
    for response in paginator.paginate(ResourceScanId=ResourceScanId):
        resources = response.get("Resources", [])
        resources = [{**i, "ResourceIdentifier": json.dumps(i["ResourceIdentifier"])} for i in resources]
        data.extend(resources)

    # csv出力
    df = pl.DataFrame(data=data)
    df.write_csv("resource_scan.csv")

if __name__ == "__main__":
    main()

スキャン待ち処理やページネーターもあるのですが、基本的には以下のような流れで処理されます。

  • start_resource_scanでリソースをスキャン
  • list_resource_scan_resourcesでスキャン結果を取得
  • データフレームに詰めてCSV出力

結果の例

結果は、以下のようになりました。

ResourceType,ResourceIdentifier,ManagedByStack
AWS::Athena::WorkGroup,"{""Name"": ""primary""}",false
AWS::CloudFront::CachePolicy,"{""Id"": ""xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx""}",false
AWS::CloudFront::CachePolicy,"{""Id"": ""xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx""}",false
AWS::CloudFront::CachePolicy,"{""Id"": ""xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx""}",false
AWS::CloudFront::CachePolicy,"{""Id"": ""xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx""}",false
AWS::CloudFront::CachePolicy,"{""Id"": ""xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx""}",false
AWS::CloudFront::CachePolicy,"{""Id"": ""xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx""}",false
...

ResourceTypeがそのままリソースタイプ、ResourceIdentifierが識別子となっているようです。
ResourceIdentifierはリソースタイプごとに異なり、複数のキーを持つ場合もあるようです。

ManagedByStackは、既にCloudFormationでスタック管理されているかを示していると考えられます。

注意書き

以下にいくつかの制限事項があります。

特に以下あたりは留意しておく必要がありそうです。スキャン回数や最大リソース数は確認した上で活用しましょう。

Name Value
アカウントスキャンで処理可能な最大リソース数 100,000
1日あたりのスキャン回数(10,000リソース未満のアカウント) 3
1日あたりのスキャン回数(10,000リソース以上のアカウント) 1

また、以下の但し書きもあります。

IaC generator only supports resources that are supported by Cloud Control API in your Region. For more information, see Determining if a resource type supports Cloud Control API in the Cloud Control API User Guide

記載の通りリソースタイプがCloud Control APIをサポートしている必要がありますので、その判断は以下を参考にされてください。

まとめ

いかがでしたでしょうか。本記事がリソース管理のご参考になれば幸いです。

Share this article

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.