Help us understand the problem. What is going on with this article?

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時間になっています。

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs.html#CloudWatchLogs.Client.filter_log_events

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ではyeildを使って、取得したログイベントを随時返却します。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))
Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした