AWS Healthイベントを説明や影響を受けるリソースも含めて一覧で取得してみた(Boto3)

AWS Healthイベントを説明や影響を受けるリソースも含めて一覧で取得してみた(Boto3)

AWS Healthイベントの一覧をcsvで取得してみました。「説明」や「影響を受けるリソース」の情報も欲しいので、あわせて取得しています。

コンバンハ、千葉(幸)です。

皆さんはAWS Healthイベントを普段から確認されているでしょうか。AWSマネジメントコンソールの「AWS Health Dashboard」でこのように確認できるものです。

AWS_Health_Event
イベント一覧のイメージ。「イベントログ」から確認できる

イベントには、AWSアカウント固有のものもあればパブリックなもの(固有でなく一般的なもの)もあります。特に前者の「AWSアカウント固有のイベント」で、対応漏れのものが無いか/今後どのような変更が行われるかなど網羅しておきたいことがあります。イベント一覧をいい感じに確認したいです。

AWS Healthイベントをコンソールから確認する場合、以下のような困りごとがあります。

  • イベント一覧をエクスポートする機能がマネジメントコンソールにはない(2025年3月現在)

  • イベントの「説明」や「影響を受けるリソース」は一度イベントの詳細画面を開かないと確認できない

    🐓 .oO(必要な情報も含めた形で、パッと一覧で取得したい!!)

これらを解消するためのスクリプトをBoto3(AWS SDK for Python)を用いて作ってみました。

前提条件

本ブログではAWS Health APIを使用します。AWS SupportのBusiness、Enterprise On-Ramp、または Enterprise Support のいずれかを契約している必要があります。

Note

You must have a Business, Enterprise On-Ramp, or Enterprise Support plan from AWS Support to use the AWS Health API. If you call the AWS Health API from an AWS account that doesn't have a Business, Enterprise On-Ramp, or Enterprise Support plan, you receive a SubscriptionRequiredException error.

今回のスクリプトで出来上がるアウトプット

一回の実行で2つのファイルを生成します。

イベントの一覧(csv)

aws_health_events_20250318_152731
生成したcsvファイルを表計算ソフトで開いたところ

csvは以下の列を持ちます。

  1. 連番(独自に採番)
  2. サービス
  3. イベントタイプ
  4. 状態
  5. カテゴリ
  6. リージョン
  7. アベイラビリティゾーン
  8. 開始時刻
  9. 終了時刻
  10. 最終更新時刻
  11. スコープ
  12. 影響を受けるリソースの数
  13. 説明

影響を受けるリソースの一覧(txt)

以下のような形式で連番ごとの「影響を受けるリソース」のリストを作成します。

...
## 15.AWS_CLOUDSHELL_PERSISTENCE_EXPIRING

- arn:aws:sts::012345678910:assumed-role/cm-chiba.yukihiro/cm-chiba.yukihiro

## 16.AWS_LAMBDA_PLANNED_LIFECYCLE_EVENT

- arn:aws:lambda:ap-northeast-1:012345678910:function:sns-message-python
- arn:aws:lambda:ap-northeast-1:012345678910:function:NotifyBillingToSlack-HelloWorldFunction-16PVBRM43HUK9
- arn:aws:lambda:ap-northeast-1:012345678910:function:SecurityHubAccepter-LambdaFunction-pNvyVtCJmjl8

## 17.AWS_ACCOUNT_SECURITY_NOTIFICATION

- 012345678910

## 18.AWS_IAM_OPERATIONAL_ISSUE

- UNKNOWN
...

「影響を受けるリソース」の詳細もcsvの列に加えるとちょっとカオスになりそうだったので、2つのファイルに分けることにしました。

AWS HealthイベントのコンソールとAPIの違い

スクリプトの詳細を確認する前に、コンソールでの見え方とAPIによって取得できる結果の違いについて押さえておきます。知らないとびっくりしちゃいますからね。

イベントの名称はAPIでは取れない

例えばコンソールでOperational issue - EC2 (Ireland)と表示されるイベントがあります。

このイベント名はAPIで取得できません。関連する情報は以下のように取れます。

  • eventTypeCode:AWS_EC2_OPERATIONAL_ISSUE
  • region:eu-west-1

コンソールの表示は、イベントの内訳から独自にいい感じにそれっぽい名称をつけてくれているようです。

実際のステータスは3種類

APIで取得できるステータスは以下の3種類です。

  • open
  • closed
  • upcoming

コンソールでは違った表記をされることがあります。以下のような対応状況です。

API上のステータス コンソール上の表記(日本語)
open オープン、継続的、「–」
closed クローズ、完了済み
upcoming 今後

AWS Healthイベントの一覧を取得するPythonスクリプト

本題に入ります。

動作確認環境

  • mac OS

    • Python 3.13.2

    • Boto3 1.37.12

  • AWS CloudShell

    • Python 3.9.20

    • Boto3 1.35.78

コードの概要

以下の2つのファイルを生成するものです。

  • イベント一覧csvファイル(「説明」も含む)
  • 影響を受けるリソース一覧txtファイル

コードの実行イメージ

スクリプトの名称がaws_health_event_reporter.pyであるとします。

$ python aws_health_event_reporter.py 
AWS Health イベント情報の取得を開始します...
イベント一覧を取得中...

イベント取得成功: 21 件のイベント
ステータス別件数:
- closed: 13件
- open: 8件

closedステータスのイベントも出力しますか? (y/n): y

イベントの詳細情報を5つの並列処理で取得中...
処理中: 5/21 件完了
処理中: 10/21 件完了
処理中: 15/21 件完了
処理中: 20/21 件完了
処理中: 21/21 件完了

処理完了:
- イベント情報: aws_health_events_20250317_084424.csv
- リソース詳細: aws_health_affected_resources_20250317_084424.txt
合計 21 件のイベント情報を出力しました

Closedステータスのものも含めて取得するかを選択できるようにしています。

イベントが100件程度あっても10~20秒程度で完了するので、後からフィルタリングすればいいや、ということであれば全量を出してください。

コードの全量

長いので折りたたみます。

スクリプトの中身
aws_health_event_reporter.py
import boto3
import csv
from datetime import datetime
import sys
from collections import Counter
import concurrent.futures

def main():
    """AWS Health イベント情報を取得してCSV・テキストファイルに出力する"""
    print("AWS Health イベント情報の取得を開始します...")

    try:
        # イベント一覧の取得
        print("イベント一覧を取得中...")
        health_client = boto3.client('health', region_name='us-east-1')
        events = get_all_events(health_client)

        # ステータス別の件数を表示
        status_counts = Counter(event.get('statusCode', 'N/A') for event in events)
        print(f"\nイベント取得成功: {len(events)} 件のイベント")
        print("ステータス別件数:")
        for status, count in status_counts.items():
            print(f"- {status}: {count}件")

        # closedステータスを含めるか確認
        include_closed = input("\nclosedステータスのイベントも出力しますか? (y/n): ").lower().strip() in ['y', 'yes']
        if not include_closed:
            events = [e for e in events if e.get('statusCode') != 'closed']
            print(f"closedステータスを除外: {len(events)} 件のイベントを処理します")

        # イベントの詳細情報を並列取得
        print("\nイベントの詳細情報を5つの並列処理で取得中...")
        detailed_events = get_detailed_events_parallel(events)

        # ファイル出力
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        export_to_files(detailed_events, timestamp)

    except Exception as e:
        print(f"エラーが発生しました: {str(e)}")
        import traceback
        print(traceback.format_exc())
        sys.exit(1)

def get_all_events(client):
    """AWS Health APIから全イベントを取得"""
    events = []
    next_token = None

    while True:
        params = {'nextToken': next_token} if next_token else {}
        response = client.describe_events(**params)
        events.extend(response.get('events', []))
        next_token = response.get('nextToken')
        if not next_token:
            break

    return events

def get_event_details(event_with_index):
    """イベントの詳細情報を取得(並列処理用)"""
    index, event = event_with_index
    client = boto3.client('health', region_name='us-east-1')

    # 日時フォーマット用ヘルパー関数
    def format_time(dt):
        return dt.strftime('%Y-%m-%d %H:%M:%S') if isinstance(dt, datetime) else ''

    # イベント詳細取得
    try:
        details = client.describe_event_details(eventArns=[event['arn']])['successfulSet'][0]
        description = details['eventDescription']['latestDescription']
    except Exception as e:
        print(f"イベント詳細取得エラー ({event.get('eventTypeCode', 'unknown')}): {str(e)}")
        description = "詳細情報を取得できませんでした"

    # 影響リソース取得
    try:
        entities = client.describe_affected_entities(filter={'eventArns': [event['arn']]})['entities']
        resources = [entity.get('entityValue', 'N/A') for entity in entities]
    except Exception:
        resources = []

    # マッピング定義
    event_type_code = event.get('eventTypeCode', '')

    return {
        '連番': index,
        'サービス': event.get('service', ''),
        'イベントタイプ': event_type_code,
        '状態': event.get('statusCode', ''),
        'カテゴリ': event.get('eventTypeCategory', ''),
        'リージョン': event.get('region', ''),
        'アベイラビリティゾーン': event.get('availabilityZone', ''),
        '開始時刻': format_time(event.get('startTime')),
        '終了時刻': format_time(event.get('endTime')),
        '最終更新時刻': format_time(event.get('lastUpdatedTime')),
        'スコープ': event.get('eventScopeCode', ''),
        '影響を受けるリソースの数': len(resources),
        '説明': description,
        'リソース': resources,
        'EVENT_TYPE_CODE': event_type_code  # テキストファイル用
    }

def get_detailed_events_parallel(events, max_workers=5):
    """イベント詳細を並列取得"""
    indexed_events = list(enumerate(events, 1))
    detailed_events = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(get_event_details, event_idx): i 
                  for i, event_idx in enumerate(indexed_events)}

        for i, future in enumerate(concurrent.futures.as_completed(futures)):
            if (i + 1) % 5 == 0 or i + 1 == len(indexed_events):
                print(f"処理中: {i + 1}/{len(indexed_events)} 件完了")

            try:
                detailed_events.append(future.result())
            except Exception as e:
                idx = futures[future] + 1
                print(f"イベント {idx} の処理中にエラー: {str(e)}")

    # 連番順にソート
    return sorted(detailed_events, key=lambda x: x['連番'])

def export_to_files(events, timestamp):
    """イベント情報をCSVとテキストファイルに出力"""
    # CSVファイル出力
    csv_filename = f"aws_health_events_{timestamp}.csv"
    fieldnames = ['連番', 'サービス', 'イベントタイプ', '状態', 'カテゴリ', 
                 'リージョン', 'アベイラビリティゾーン', '開始時刻', '終了時刻', '最終更新時刻', 
                 'スコープ', '影響を受けるリソースの数', '説明']

    with open(csv_filename, 'w', newline='', encoding='utf-8-sig') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for event in events:
            # 必要なフィールドのみ抽出し、説明の改行を削除
            row = {k: event[k] for k in fieldnames}
            row['説明'] = row['説明'].replace('\n', ' ').replace('\r', ' ')
            writer.writerow(row)

    # テキストファイル出力
    txt_filename = f"aws_health_affected_resources_{timestamp}.txt"

    with open(txt_filename, 'w', encoding='utf-8') as textfile:
        for event in events:
            textfile.write(f"## {event['連番']}.{event['EVENT_TYPE_CODE']}\n\n")

            if not event['リソース']:
                textfile.write("- 影響を受けるリソースはありません\n")
            else:
                for resource in event['リソース']:
                    textfile.write(f"- {resource}\n")

            textfile.write("\n")

    print(f"\n処理完了:")
    print(f"- イベント情報: {csv_filename}")
    print(f"- リソース詳細: {txt_filename}")
    print(f"合計 {len(events)} 件のイベント情報を出力しました")

if __name__ == "__main__":
    main()

コードの解説

Claude 3.7 Sonnet君に書いてもらった処理の流れです。

1. 初期化と基本イベント取得

  • boto3ライブラリを使用してAWS Health APIクライアントを作成(us-east-1リージョン指定)
  • get_all_events関数を呼び出し、describe_events APIでイベント一覧を取得
  • ページネーション処理(nextToken)を使用して全イベントを収集

2. イベント情報の集計と表示

  • Counterクラスを使用してイベントのステータス(open/closed/upcoming)ごとの件数を集計
  • 取得したイベントの総数とステータス別の件数をコンソールに表示

3. ユーザー入力によるフィルタリング

  • ユーザーにclosedステータス(解決済み)のイベントを含めるか確認
  • 含めない場合は、リストから該当イベントを除外

4. 詳細情報の並列取得

  • ThreadPoolExecutorを使用して5つの並列スレッドを作成
  • 各イベントに対して以下の処理を並列実行:
    • describe_event_details APIでイベントの詳細説明を取得
    • describe_affected_entities APIで影響を受けるリソースの情報を取得
    • 日時情報のフォーマット変換
    • 必要な情報を辞書形式にまとめる
  • 処理の進捗状況を5件ごとに表示

5. ファイル出力準備

  • 現在時刻からタイムスタンプを生成(YYYYMMDD_HHMMSS形式)
  • タイムスタンプを含むCSVファイル名とテキストファイル名を作成

6. CSVファイル出力

  • DictWriterを使用してCSVファイルを作成
  • ヘッダー行を書き込み(連番、サービス、イベントタイプなど)
  • 各イベントの情報を1行ずつ書き込み
  • 説明文から改行を削除してCSV形式を保持

7. テキストファイル出力

  • 各イベントについて:
    • 連番とイベントタイプコードをヘッダーとして書き込み
    • 影響を受けるリソースがあればリスト形式で出力
    • リソースがなければその旨を記載

8. 完了通知

  • 処理完了メッセージを表示
  • 出力したファイル名と処理したイベント件数を表示

ステップバイステップの詳細な解説は、長いのでこのブログの末尾に載せておきます。

終わりに

AWS Healthイベントを「説明」付きで一覧取得するスクリプトのご紹介でした。「影響を受けるリソース」は別ファイルになりましたが、連番を振ったことで対応させやすくなっているかと思います。

イベント数が多くなってくるとコンソールからポチポチでなく一覧で取得/加工したくなりますので、なんらか参考になれば幸いです。

ここで取得した結果を生成AIツールに読み込ませて概要をまとめさせる、といった使い方もできそうですね。(生成AIツールへの情報のインプットについては各自で適切にご判断をお願いします。)

以上、チバユキ (@batchicchi)がお送りしました。

番外編:イベントの一覧をAWS CLIでサクッと取得する

今回のブログはBoto3を用いて"「説明」や「影響を受けるリソース」の情報を含めて取得する"が主題ですが、そこまで作り込むのも面倒、とりあえず一覧だけ出力できればいい!という方のためにAWS CLIを用いて情報取得するコマンドを載せておきます。

# 必要であればヘッダーを表示
echo -e "ARN\tSERVICE\tEVENT_TYPE_CODE\tEVENT_TYPE_CATEGORY\tREGION\tSTART_TIME\tEND_TIME\tLAST_UPDATED_TIME\tSTATUS_CODE\tEVENT_SCOPE_CODE"

# AWS Health イベントを取得
aws health describe-events \
  --region us-east-1 \
  --query "events[*].[
    arn,
    service,
    eventTypeCode,
    eventTypeCategory,
    region,
    startTime,
    endTime,
    lastUpdatedTime,
    statusCode,
    eventScopeCode
  ]" \
  --output text

ここでは「説明」や「影響を受けるリソース」の情報は含まれません。

ひとまず件数を確認する、といった用途で使っていました。

参考

コードの解説:詳細版

Claude 3.7 Sonnet君に書いてもらった解説です。

1. インポートと初期設定

import boto3
import csv
from datetime import datetime
import sys
from collections import Counter
import concurrent.futures
  • boto3: AWS SDKで、AWSサービスにアクセスするために使用
  • csv: CSVファイル操作用のライブラリ
  • datetime: 日時操作用のライブラリ
  • sys: システム関連の機能を提供するライブラリ
  • Counter: コレクション内の要素のカウントを容易にするクラス
  • concurrent.futures: 並列処理を実装するためのライブラリ

2. メイン関数

def main():
    """AWS Health イベント情報を取得してCSV・テキストファイルに出力する"""
    print("AWS Health イベント情報の取得を開始します...")

    try:
        # イベント一覧の取得
        print("イベント一覧を取得中...")
        health_client = boto3.client('health', region_name='us-east-1')
        events = get_all_events(health_client)
  1. まず、AWS Health APIクライアントを作成します。AWS Healthはus-east-1リージョンでのみ利用可能なグローバルサービスです。
  2. get_all_events関数を呼び出して、すべてのヘルスイベントを取得します。
        # ステータス別の件数を表示
        status_counts = Counter(event.get('statusCode', 'N/A') for event in events)
        print(f"\nイベント取得成功: {len(events)} 件のイベント")
        print("ステータス別件数:")
        for status, count in status_counts.items():
            print(f"- {status}: {count}件")
  1. 取得したイベントのステータス(open, closed, upcoming)ごとの件数をカウントして表示します。
        # closedステータスを含めるか確認
        include_closed = input("\nclosedステータスのイベントも出力しますか? (y/n): ").lower().strip() in ['y', 'yes']
        if not include_closed:
            events = [e for e in events if e.get('statusCode') != 'closed']
            print(f"closedステータスを除外: {len(events)} 件のイベントを処理します")
  1. ユーザーに「closed(解決済み)」ステータスのイベントを含めるかどうか確認し、含めない場合は除外します。
        # イベントの詳細情報を並列取得
        print("\nイベントの詳細情報を5つの並列処理で取得中...")
        detailed_events = get_detailed_events_parallel(events)

        # ファイル出力
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        export_to_files(detailed_events, timestamp)
  1. イベントの詳細情報を並列処理で取得します。
  2. 現在の日時をタイムスタンプとして取得し、ファイル名に使用します。
  3. 詳細情報をCSVとテキストファイルに出力します。
    except Exception as e:
        print(f"エラーが発生しました: {str(e)}")
        import traceback
        print(traceback.format_exc())
        sys.exit(1)
  1. エラーが発生した場合、詳細なエラー情報を表示してプログラムを終了します。

3. イベント一覧取得関数

def get_all_events(client):
    """AWS Health APIから全イベントを取得"""
    events = []
    next_token = None

    while True:
        params = {'nextToken': next_token} if next_token else {}
        response = client.describe_events(**params)
        events.extend(response.get('events', []))
        next_token = response.get('nextToken')
        if not next_token:
            break

    return events
  1. AWS Health APIのdescribe_eventsメソッドを使用して、すべてのイベントを取得します。
  2. APIはページネーション(一度に取得できる件数に制限がある)のため、nextTokenを使って全ページを取得します。
  3. 取得したイベントをリストに追加し、すべてのページを取得したら返します。

4. イベント詳細情報取得関数

def get_event_details(event_with_index):
    """イベントの詳細情報を取得(並列処理用)"""
    index, event = event_with_index
    client = boto3.client('health', region_name='us-east-1')

    # 日時フォーマット用ヘルパー関数
    def format_time(dt):
        return dt.strftime('%Y-%m-%d %H:%M:%S') if isinstance(dt, datetime) else ''
  1. イベントとそのインデックス(連番)を受け取ります。
  2. AWS Health APIクライアントを作成します。
  3. 日時をフォーマットするためのヘルパー関数を定義します。
    # イベント詳細取得
    try:
        details = client.describe_event_details(eventArns=[event['arn']])['successfulSet'][0]
        description = details['eventDescription']['latestDescription']
    except Exception as e:
        print(f"イベント詳細取得エラー ({event.get('eventTypeCode', 'unknown')}): {str(e)}")
        description = "詳細情報を取得できませんでした"
  1. describe_event_details APIを使用して、イベントの詳細説明を取得します。
  2. エラーが発生した場合は、エラーメッセージを表示し、説明を「詳細情報を取得できませんでした」に設定します。
    # 影響リソース取得
    try:
        entities = client.describe_affected_entities(filter={'eventArns': [event['arn']]})['entities']
        resources = [entity.get('entityValue', 'N/A') for entity in entities]
    except Exception:
        resources = []
  1. describe_affected_entities APIを使用して、イベントの影響を受けるリソース(EC2インスタンスやEBSボリュームなど)の情報を取得します。
  2. エラーが発生した場合は、空のリストを設定します。
    # マッピング定義
    event_type_code = event.get('eventTypeCode', '')

    return {
        '連番': index,
        'サービス': event.get('service', ''),
        'イベントタイプ': event_type_code,
        '状態': event.get('statusCode', ''),
        'カテゴリ': event.get('eventTypeCategory', ''),
        'リージョン': event.get('region', ''),
        'アベイラビリティゾーン': event.get('availabilityZone', ''),
        '開始時刻': format_time(event.get('startTime')),
        '終了時刻': format_time(event.get('endTime')),
        '最終更新時刻': format_time(event.get('lastUpdatedTime')),
        'スコープ': event.get('eventScopeCode', ''),
        '影響を受けるリソースの数': len(resources),
        '説明': description,
        'リソース': resources,
        'EVENT_TYPE_CODE': event_type_code  # テキストファイル用
    }
  1. イベントタイプコードを取得します。
  2. イベントの詳細情報を辞書形式で返します。各キーは日本語で、CSVのヘッダーとして使用されます。

5. 並列処理関数

def get_detailed_events_parallel(events, max_workers=5):
    """イベント詳細を並列取得"""
    indexed_events = list(enumerate(events, 1))
    detailed_events = []
  1. イベントリストに連番(1から始まる)を付けます。
  2. 詳細情報を格納するための空リストを作成します。
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(get_event_details, event_idx): i 
                  for i, event_idx in enumerate(indexed_events)}
  1. ThreadPoolExecutorを使用して、最大5つの並列スレッドでイベント詳細を取得します。
  2. 各イベントに対してget_event_details関数を実行するタスク(future)を作成します。
        for i, future in enumerate(concurrent.futures.as_completed(futures)):
            if (i + 1) % 5 == 0 or i + 1 == len(indexed_events):
                print(f"処理中: {i + 1}/{len(indexed_events)} 件完了")

            try:
                detailed_events.append(future.result())
            except Exception as e:
                idx = futures[future] + 1
                print(f"イベント {idx} の処理中にエラー: {str(e)}")
  1. 完了したタスクから順に結果を取得します。
  2. 5件ごと、または最後のイベントで進捗状況を表示します。
  3. タスクの結果(イベント詳細情報)をリストに追加します。
  4. エラーが発生した場合は、エラーメッセージを表示します。
    # 連番順にソート
    return sorted(detailed_events, key=lambda x: x['連番'])
  1. 最後に、連番順にソートして結果を返します。

6. ファイル出力関数

def export_to_files(events, timestamp):
    """イベント情報をCSVとテキストファイルに出力"""
    # CSVファイル出力
    csv_filename = f"aws_health_events_{timestamp}.csv"
    fieldnames = ['連番', 'サービス', 'イベントタイプ', '状態', 'カテゴリ', 
                 'リージョン', 'アベイラビリティゾーン', '開始時刻', '終了時刻', '最終更新時刻', 
                 'スコープ', '影響を受けるリソースの数', '説明']
  1. タイムスタンプを含むCSVファイル名を作成します。
  2. CSVファイルのヘッダー(列名)を定義します。
    with open(csv_filename, 'w', newline='', encoding='utf-8-sig') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for event in events:
            # 必要なフィールドのみ抽出し、説明の改行を削除
            row = {k: event[k] for k in fieldnames}
            row['説明'] = row['説明'].replace('\n', ' ').replace('\r', ' ')
            writer.writerow(row)
  1. CSVファイルを作成し、ヘッダー行を書き込みます。
  2. 各イベントについて、必要なフィールドのみを抽出します。
  3. 説明文から改行を削除して、CSVフォーマットを崩さないようにします。
  4. イベント情報をCSVファイルに書き込みます。
    # テキストファイル出力
    txt_filename = f"aws_health_affected_resources_{timestamp}.txt"

    with open(txt_filename, 'w', encoding='utf-8') as textfile:
        for event in events:
            textfile.write(f"## {event['連番']}.{event['EVENT_TYPE_CODE']}\n\n")

            if not event['リソース']:
                textfile.write("- 影響を受けるリソースはありません\n")
            else:
                for resource in event['リソース']:
                    textfile.write(f"- {resource}\n")

            textfile.write("\n")
  1. 影響を受けるリソース情報を出力するためのテキストファイル名を作成します。
  2. 各イベントについて、連番とイベントタイプコードをヘッダーとして書き込みます。
  3. 影響を受けるリソースがない場合はその旨を[1]、ある場合はリソースのリストを書き込みます。
    print(f"\n処理完了:")
    print(f"- イベント情報: {csv_filename}")
    print(f"- リソース詳細: {txt_filename}")
    print(f"合計 {len(events)} 件のイベント情報を出力しました")
  1. 処理完了のメッセージと、出力したファイル名、処理したイベント件数を表示します。

7. スクリプト実行

if __name__ == "__main__":
    main()

スクリプトが直接実行された場合(インポートではなく)、main()関数を呼び出します。

まとめ

このスクリプトは以下のことを行います:

  1. AWS Health APIを使用して、アカウントに関連するすべてのヘルスイベントを取得
  2. ユーザーの選択に応じて、解決済み(closed)のイベントを含めるかどうかを決定
  3. 各イベントの詳細情報と影響を受けるリソースを並列処理で取得
  4. イベント情報をCSVファイルに、影響を受けるリソースのリストをテキストファイルに出力

これにより、AWS環境で発生している問題や計画されたメンテナンスなどの情報を簡単に確認できます。

脚注
  1. 影響を受けるリソースがない場合は大抵「UNKNOWN」という文字列が入っておりそれをそのまま出力するので、あまり発生することは想定してないです ↩︎

Share this article

facebook logohatena logotwitter logo

© Classmethod, Inc. All rights reserved.