CloudWatch Logsのイベントを期間指定で取得する

CloudWatch Logs

CloudWatchに保存されたログはAWSのコンソールで確認する事ができて便利ですが、ログを解析する場合にS3にエクスポートするなど一手間かかります。ログの解析には、私はPythonスクリプトで行う事が多いので、PythonスクリプトでCloudWatchからログを取得するスクリプトの作成を目指します。なお、以下のコードはpython3(3.7.5)で動作させています。

$ python3 --version
Python 3.7.5

PythonでのCloudWatch Logsの取得方法

他のAWSの制御と同様でboto3を利用します。boto3が入っていない場合は、インストールをして下さい。

$ pip3 install boto3

boto3を利用するためには、AWSのIAMアカウントのアクセスキーとシークレットを利用する必要があります。そちらについては、aws configureで事前に設定しておいて下さい。設定されていれば、boto3が環境変数を読み取リますので、pythonスクリプトでアクセスキー、シークレットに指定は必要ありません。
https://docs.aws.amazon.com/ja_jp/cli/latest/userguide/cli-chap-configure.html

CloudWatchのログを取得する場合は、'logs'を指定してboto3のクライアントを取得します。

client = boto3.client('logs')

CloudWatch Logsのboto3のクライアントのI/Fは以下に記述されています。
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs.html#CloudWatchLogs.Client.get_log_events

CloudWatch Logsはロググループがあり、その中にログストリームが複数存在する構成となっています。get_log_events()ではロググループとログストリームを指定してログイベントを取得します。
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs.html#CloudWatchLogs.Client.get_log_events

今回はロググループの指定だけで、ログイベントを取得したいので、filter_log_events()を利用します。全てのデータを取得したいので、フィルターは指定しません。startTime, endTimeで期間を指定する事ができます。こちらの時間はUNIX時間になっています。

response = client.filter_log_events(
                logGroupName=log_group_name,
                startTime=start_time,
                endTime=end_time
            )

ログイベントは最大1Mbyte分まで一度の呼び出しで取得できます。続きを取得する場合は、responseに含まれるnextTokenを利用して再度呼び出す必要があります。

next_token = response['nextToken']
response = client.filter_log_events(
                logGroupName=log_group_name,
                startTime=start_time,
                endTime=end_time,
                nextToken=next_token
            )

サンプルコード

現在の時間の日付だけを取り出し、昨日から今日までの24時間のログを取得するサンプルになります。retrieve_eventsではyieldを使って、取得したログイベントを随時返却します。main関数では、取得したログイベントをfor文で随時取得して、ログイベントの数を表示しています。

def retrieve_events(log_group_name, start_time, end_time):
    client = boto3.client('logs')
    next_token = ''
    response = {}
    while True:
        if next_token == '':
            response = client.filter_log_events(
                logGroupName=log_group_name,
                startTime=start_time,
                endTime=end_time
            )
        else:
            response = client.filter_log_events(
                logGroupName=log_group_name,
                startTime=start_time,
                endTime=end_time,
                nextToken=next_token
            )
        if 'nextToken' in response:
            next_token = response['nextToken']
            yield response['events']
        else:
            break

if __name__ == '__main__':
    now = datetime.datetime.now()
    today = datetime.datetime(now.year, now.month, now.day, 0, 0, 0, 0)
    yesterday = today - datetime.timedelta(days=1)
    today_unix = int(today.timestamp()*1000)
    yesterday_unix = int(yesterday.timestamp()*1000)

    for events in retrieve_events(LOG_GROUP_NAME, yesterday_unix, today_unix):
        print(len(events))