4
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

私は現場でAWS CloudWatch に出力されるエラーログの定期的な確認と集計を担当しています。
手動でのログ確認は手間がかかるので、作業を効率化するためにCloudWatch Logs InsightsでエクスポートしたJSONを元に、エラー内容と発生箇所(path)ごとにログをグループ化し、件数を集計する Pythonスクリプトを作成しました。
ログ分析を効率化したい方の参考になれば幸いです!

準備

まずは CloudWatch Logs Insights を使って、必要なログを検索します。

ロググループを選択し、クエリを実行した後に、右上の 「結果をエクスポート」ボタンをクリック → 「テーブルをダウンロード(JSON)」を選択 すると、検索結果を JSON 形式でエクスポートできます。
保存されたファイルは通常 logs-insights-results.json というファイル名になります。

目的

今回はAWS CloudWatch Logs Insights からエクスポートしたログファイル(JSON形式)を対象に、以下の3つの処理を行うPythonスクリプトを紹介します。
・エラーメッセージとpathの組み合わせでグルーピング

・各グループごとに件数をカウント

・不正な(不完全な)ログはスキップして内容を確認

Pythonスクリプト

group_logs.py
import json
from collections import defaultdict

# JSONファイルを開く
with open("logs-insights-results.json", "r", encoding="utf-8") as f:
    data = json.load(f)

# data はリスト形式
results = data

# エラーの種類と path でグループ化
group_counts = defaultdict(int)

for entry in results:
    try:
        message = entry["@message"]
        context = message.get("context", {})
        detail = context.get("detail", {})

        error_message = detail.get("message")
        error_path_raw = detail.get("path")

        # message や path が無効の場合はスキップ
        if not error_message or not isinstance(error_path_raw, list):
            raise ValueError("message または path が無効")

        error_path = tuple(error_path_raw)
        group_key = (error_message, error_path)
        group_counts[group_key] += 1

    except Exception as e:
        # timestamp 付きでスキップログを出力
        timestamp_str = entry.get("@timestamp", None)
        if timestamp_str:
            formatted_timestamp = timestamp_str
        else:
            formatted_timestamp = "timestampなし"

        print(f"スキップされたログ(理由: {e} / timestamp: {formatted_timestamp}")
        continue

# 件数の多い順にソートして出力
sorted_group_counts = sorted(group_counts.items(), key=lambda x: x[1], reverse=True)

for (msg, path), count in sorted_group_counts:
    print(f"[{count}件] {msg} | path: {path}")

スクリプトの実行方法(初心者向け)

Pythonの初心者でも実行できるように、手順を簡単にまとめました。

1.Pythonのインストール
 公式サイトからPythonをインストールしてください
https://www.python.org/
 インストール時に 「Add Python to PATH」 にチェックを入れてください!

「Add Python to PATH」にチェックを入れずにPythonをインストールすると、コマンドライン(ターミナルやPowerShell)から python が使えなくなります。

2.スクリプトとJSONファイルを準備
 上記のPythonコードを group_logs.py という名前で保存します。
 logs-insights-results.json を同じフォルダに置きます。

3.コマンドラインで実行

cd スクリプトのあるフォルダ
python group_logs.py

結果がコンソールに表示されます。

実行結果

スキップされたログ(理由: message または path が無効 / timestamp: 2025-06-21 03:38:48.707)
[15件] User not found | path: ('user', 'profile')
[8件] Invalid token | path: ('auth', 'validate')
[3件] Database timeout | path: ('orders', 'submit')

不完全なログは「スキップされたログ」として表示され、内容とタイムスタンプが記録されます。
スキップされてしまったログはタイムスタンプをもとに、あとから手動で確認できます。

おわりに

CloudWatch のエラーログを手動で確認・集計するのは、件数が多くなるととても大変です。
しかし、このような簡易集計スクリプトを作っておくと、エラーの発生件数や新規エラーの検出が格段に楽になります。

小さな改善ですが、日々の運用を効率化したい方の参考になれば幸いです!

4
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
4
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?