pythonを使ってAWSのログのインサイトを実行し、結果をCSVファイルで出力する #Python

背景、目的

AWS環境を使用したWebサイトを提供しており、日次の運用としてエラー有無、内容の調査を行います。
エラーが出ている場合（基本はなにかしらのエラーが出ている）は、どのようなエラーであるか、対応が必要であるかを判断します。

その前段のエラー有無を取得するためにCloudWatch Logsを使用しているのですが、毎回以下の操作を行います。

ブラウザを開いてコンソールへログイン
CloudWatchを開いてログのインサイトを選択
対象日時を範囲選択
保存済みのクエリから対象を選択
実行して結果が出るのを待つ
結果をダウンロード

「保存済みのクエリから対象を選択」以降は複数回実行します。
この後にエラー有無と対応要否を判断します。

毎日やる作業と考えると結構しんどいです。

環境

key	value
OS	mac 11.6.1
language	Python 3.9.1
AWS CLI	2.7.9

前提

操作、設定

AWSのアカウントを持っている
AWSのCloudWatchのログのインサイトを利用している

実行環境

CLIを使用したAWSへの接続環境が整っている
- CloudWatch Logs、CLIを使用する権限を持っている
- CLIのインストールが完了している
- 端末へプロファイルの設定が完了している
Pythonを実行できる環境である

実践

登場するファイル

file	in-out	note
GetAWSCloudWatchInsight.py	in	実行するファイル本体
GetAWSCloudWatchInsight.json	in	使用するプロファイルやログのインサイトで使用する情報を記載した設定ファイル
{logs-insights-results_xxx.csv}	out	jsonファイルへ記載した名前で、ログのインサイトの結果を出力
{summary.txt}	out	jsonファイルへ記載した名前で、ログのインサイトの統計情報をサマリ情報として出力

ロジック

実際に書いたロジックです。
作ることを最優先したのでベタ書き．．．
言い訳です、すみません。

GetAWSCloudWatchInsight.py

from sqlite3 import Timestamp
from traceback import print_tb
import boto3
from datetime import datetime, timedelta
import time
import json
from dateutil import tz
JST = tz.gettz('Asia/Tokyo')
UTC = tz.gettz("UTC")


input_json_file_name = "GetAWSCloudWatchInsight.json"
with open(input_json_file_name) as f:
  data = json.loads(f.read())

# 対象プロファイル
aws_profile = data["aws_profile"]
summary_file_name = data["summary_file_name"]

# 日時指定
start_datetime = data['start_datetime']
start_time_input = datetime(start_datetime["year"], start_datetime["month"] , start_datetime["day"], start_datetime["hour"], start_datetime["minute"], start_datetime["second"], tzinfo=JST)
start_time = int(datetime.timestamp(start_time_input))
end_datetime = data['end_datetime']
end_time_input = datetime(end_datetime["year"], end_datetime["month"] , end_datetime["day"], end_datetime["hour"], end_datetime["minute"], end_datetime["second"], tzinfo=JST)
end_time = int(datetime.timestamp(end_time_input))

# 設定対象を処理
print(f"start: {start_time_input}")
print(f"end  : {end_time_input}\n")
with open(summary_file_name, mode="a") as f:
    f.write(f"start: {start_time_input}\n")
    f.write(f"end  : {end_time_input}\n")

for target in data["targets"]:
    target_name = target['name']
    print(f"■start {target_name}")

    query = target['query']

    log_groups = []
    for log_group in target["log_groups"]:
        log_groups.append(log_group["value"])

    output_file_name = target["output"]

    # セッション生成
    my_session = boto3.Session(profile_name = aws_profile)
    client = my_session.client('logs')

    # クエリ実行
    start_query_response = client.start_query(
        logGroupNames=log_groups,
        startTime=start_time,
        endTime=end_time,
        queryString=query,
    )
    
    query_id = start_query_response['queryId']
    print(f'query_id: {query_id}')

    # クエリ実行待機
    response = None
    while response == None or response['status'] == 'Running':
        print('Waiting for query to complete ...')
        time.sleep(1)
        response = client.get_query_results(
            queryId=query_id
        )

    # 統計情報出力
    with open(summary_file_name, mode="a") as f:
        f.write(f"{target_name}\n")
        f.write(json.dumps(response['statistics']) + "\n\n")

    # ファイル出力
    with open(output_file_name, mode="w") as f:
        f.write(f"@timestamp,@message\n")
        for result in response['results']:
            message = str(result[1]['value'].replace('"', '""'))
            f.write(f"{result[0]['value']},\"{message}\"\n")

設定ファイル（json）

次はjsonファイルのサンプルです。
先頭部分は共通で使用する設定、「targets」がログのインサイトで指定する内容を設定したものです。

GetAWSCloudWatchInsight.json

{
  "aws_profile": "xxx_dev",
  "start_datetime": {
    "year": 2022,
    "month": 9,
    "day": 3,
    "hour": 0,
    "minute": 0,
    "second":  0
  },
  "end_datetime": {
    "year": 2022,
    "month": 9,
    "day": 3,
    "hour": 23,
    "minute": 59,
    "second":  59
  },
  "summary_file_name": "summary.txt",
  "targets": [
    {
      "name": "app-error",
      "query": "Fields @timestamp, @message | filter @logStream like 'xxx/xxx/' | filter @message like 'ERROR' | sort @timestamp desc | limit 10000",
      "log_groups": [
        {
          "value": "/xxx/dev-xxx"
        },
        {
          "value": "/xxx/dev-yyy"
        },
        {
          "value": "/xxx/dev-zzz"
        }
      ],
      "output": "logs-insights-results_app-error.csv"
    }
  ]
}

解説

jsonファイルについて、基本的にはログのインサイトで操作する項目であるため、説明は省略します。
ロジックについても、コメントを記載しているので解説は最小限にさせていただきます。

時間指定
時間指定はJSTを意図的に指定しています。
リクエストで使用する開始時間、終了時間の指定がエポック秒であるため、datetime.timestampを使用して変換しています。
エポック秒はこちらのサイトの説明がわかりやすいです。
AWSのクエリについて
- start_query
  クエリの実行を依頼して、実行したクエリIDを取得します。実行には時間がかかるため、完了まで待機します。
- get_query_results
  クエリIDをキーにしてリクエストし、実行状況と実行結果を取得します。
  実行結果がCompleteになったら完了ですが、参考にした情報がRunningの間は待機する、となっていたのでそのまま採用。
実行結果のファイル出力について
CSVファイル（カンマ区切り、ダブルコーテーション囲み）で出力するため、実行結果にダブルコーテーションが含まれていた場合は、ダブルコーテーションを重ねています。
これにより、エクセルで開いたときに正しく表示することができます。

終わりに

前提条件を書いているときに思ったのですが、PythonやAWS CLI等の環境面がいろいろあることに気づきました。
自分の環境ではある程度インストールが完了した状態だったので気になりませんでしたが、まっさらな状態からだと厳しいかもしれません。
それでも、一度実行してしまえばあとは楽だし、Pythonはインストールしておいて損はないので億劫にならずに実践していただければ。

改善事項はたくさんあると思いますので、ご意見いただければと思います。

pythonを使ってAWSのログのインサイトを実行し、結果をCSVファイルで出力する