Closed8

boto3,InquirerPyを使って、AWS/Lambdaで出力したログを閲覧する

not75743not75743

背景

※調査が足りない可能性あり
Lambda関数を作成すると対応するcloudwatch logsグループが作成される。
そこにLambda関数のログが格納されるわけであるが、利便性のためにawscliでログを見ようとすると

  • ロググループの一覧を出力し、そこから対象となるものを探す
$ aws logs describe-log-groups | jq -r '.logGroups[].logGroupName'
  • ロググループ内のログストリームを検索する
$ aws logs describe-log-streams --log-group-name '<ロググループ名>' | jq -r '.logStreams[].logStreamName'
  • ロググループ、ログストリームからログを閲覧する
$ aws logs get-log-events --log-group-name '<ロググループ名>' --log-stream-name '<ログストリーム名>'

のステップを踏む必要があり、不便であるため

not75743not75743

今回使用するコード:

コード
app.py
import boto3
from InquirerPy import inquirer

def get_log_groups():
    client = boto3.client('logs')
    response = client.describe_log_groups()
    return [group['logGroupName'] for group in response['logGroups']]

def get_log_streams(log_group_name):
    client = boto3.client('logs')
    response = client.describe_log_streams(logGroupName=log_group_name)
    return [stream['logStreamName'] for stream in response['logStreams']]

def get_log_events(log_group_name, log_stream_name):
    client = boto3.client('logs')
    response = client.get_log_events(logGroupName=log_group_name, logStreamName=log_stream_name)
    return response['events']

def main():
    log_groups = get_log_groups()

    log_group_question = inquirer.select(
        message="Choose a log group:",
        choices=log_groups
    )

    chosen_log_group = log_group_question.execute()

    log_streams = get_log_streams(chosen_log_group)

    log_stream_question = inquirer.select(
        message="Choose a log stream:",
        choices=log_streams
    )

    chosen_log_stream = log_stream_question.execute()

    print(f'You chose: log group {chosen_log_group}, log stream {chosen_log_stream}')

    log_events = get_log_events(chosen_log_group, chosen_log_stream)

    for event in log_events:
        print(event['message'].rstrip('\n'))

if __name__ == "__main__":
    main()
not75743not75743

モジュール

import boto3
from InquirerPy import prompt

予めpipで必要なインストールしておきましょう。

$ pip install --no-cache-dir boto3 InquirerPy
not75743not75743

Logs周り

コード

def get_log_groups():
    client = boto3.client('logs')
    response = client.describe_log_groups()
    return [group['logGroupName'] for group in response['logGroups']]

def get_log_streams(log_group_name):
    client = boto3.client('logs')
    response = client.describe_log_streams(logGroupName=log_group_name)
    return [stream['logStreamName'] for stream in response['logStreams']]

def get_log_events(log_group_name, log_stream_name):
    client = boto3.client('logs')
    response = client.get_log_events(logGroupName=log_group_name, logStreamName=log_stream_name)
    return response['events']

メモ

  • それぞれCloudWatch Logsのクライアントを作成し、
    • ロググループのリストを閲覧
    • ロググループ内のログストリームのリストの閲覧
    • ログイベントの返却
      を行う関数を作成します。
  • 使い方はドキュメントを参照
not75743not75743

main関数

def main():
    log_groups = get_log_groups()

    log_group_question = inquirer.select(
        message="Choose a log group:",
        choices=log_groups
    )

    chosen_log_group = log_group_question.execute()

    log_streams = get_log_streams(chosen_log_group)

    log_stream_question = inquirer.select(
        message="Choose a log stream:",
        choices=log_streams
    )

    chosen_log_stream = log_stream_question.execute()

    print(f'You chose: log group {chosen_log_group}, log stream {chosen_log_stream}')

    log_events = get_log_events(chosen_log_group, chosen_log_stream)

    for event in log_events:
        print(event['message'].rstrip('\n'))
  • inquirer.selectで選択肢として表示する内容を指定します。
    今回はロググループ・ログストリーム一覧の配列を指定しています。
  • executeでプロンプトを動かします
  • printによる空白行を削除するため、rstripで改行を削除しています。
このスクラップは2023/05/28にクローズされました