boto3とは？

pythonアプリケーションからAWSを操作するライブラリ
今回はcloudwatch logsの操作、閲覧を行います

InquirerPy

対話型のインターフェースを提供するライブラリです。
もともとはInquirer.jsというjavascriptライブラリをpython版にしたみたいです

not75743

背景

※調査が足りない可能性あり
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 '<ログストリーム名>'

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

not75743

今回使用するコード:

コード

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()

not75743

モジュール

import boto3
from InquirerPy import prompt

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

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

not75743

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のクライアントを作成し、
- ロググループのリストを閲覧
- ロググループ内のログストリームのリストの閲覧
- ログイベントの返却
  を行う関数を作成します。
使い方はドキュメントを参照
- 例えばdescribe_log_streamsを例にだすと
  - logGroupNameかlogGroupIdentifierがリクエストで必須である
  - Response SyntaxよりResponseがJson形式である
- ことなどがわかる
- https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/logs/client/describe_log_streams.html

not75743

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で改行を削除しています。

not75743

参考

ドキュメント見ないとわからないです

not75743

このスクラップは2023/05/28にクローズされました