【Python】CloudWatch logsをlocalファイルに保存する方法

2019年8月23日2022年6月7日

CloudWatchのログってS3にエクスポートできたり、cliでいろいろできるけどやっぱりlogファイルとして欲しいときに面倒なのでそれをPythonに頑張ってもらうやつです。

手順
①.pyと同階層にlogsフォルダを作成
②access_key, secret_access_keyを設定
③regionを設定
④logを取得したいロググループ名(group_name)を設定
⑤logの開始と終了時刻を設定(from_time, to_time)
⑥logファイルに書き込む内容を絞り込む場合は grep に任意の文字を設定
➆logファイル名を設定
⑧実行

python3.6にて動作確認

import boto3
from datetime import datetime

# access_key, secret_access_key
access_key = "[自分のアクセスキー]"
secret_access_key = "[自分のシークレットアクセスキー]"

# region
region = "ap-northeast-1"

# ロググループ名
group_name = '/aws/lambda/hoge'

# ミリ秒表記のUNIX time
from_time = 1526470510000
to_time = 1526471300000

# grep text
grep = ""

# file name
file_name = 'hoge.log'

# ログストリーム一覧を取得
client = boto3.client(
    'logs',
    aws_access_key_id=access_key,
    aws_secret_access_key=secret_access_key,
    region_name=region
)

next_token = ""
stream_list = []
while(1):
    if next_token == "":
        response = client.describe_log_streams(
            logGroupName=group_name,
            orderBy='LastEventTime',
            descending=True
        )
    else:
        response = client.describe_log_streams(
            logGroupName=group_name,
            orderBy='LastEventTime',
            descending=True,
            nextToken=next_token
        )

    for stream in response['logStreams']:
        stream_list.append(stream['logStreamName'])
    if "nextToken" in response:
        next_token = response["nextToken"]
    else:
        break

# 全てのストリームのログを取得
for stream in stream_list:
    next_token = ""
    while(1):
        if next_token == "":
            # ログを取得
            logs = client.get_log_events(
                logGroupName=group_name,
                logStreamName=stream,
                startTime=from_time,
                endTime=to_time,
                startFromHead=True
            )
        else:
            logs = client.get_log_events(
                logGroupName=group_name,
                logStreamName=stream,
                startTime=from_time,
                endTime=to_time,
                startFromHead=True,
                nextToken=next_token
            )

        # 'timestamp' と 'message' をログファイルに出力
        body = logs['events']
        with open("logs/"+file_name, 'a') as f:
            for line in body:
                if grep in line['message']:
                    time = int(str(line['timestamp'])[:10])
                    # if from_time <= time and time <= to_time:
                    message = '[{}] {}'.format(datetime.fromtimestamp(time), line['message'])
                    f.write(message)

        if "nextForwardToken" in logs:
            if logs["nextForwardToken"] == next_token:
                break
            next_token = logs["nextForwardToken"]
        else:
            break