🚶♂️
boto3,InquirerPyを使って、AWS/Lambdaで出力したログを閲覧する
背景
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 '<ログストリーム名>'
といったステップを踏む必要があります。
若干不便であるため、一度で出来るツールをpythonで作成しました。
動作イメージ
使用するライブラリ
boto3
pythonアプリケーションからAWSを操作するライブラリ
今回はcloudwatch logsから情報の取得を行います
InquirerPy
対話型のインターフェースを提供するライブラリです。
もともとはInquirer.jsというjavascriptライブラリをpython版にしたみたいです
コード
今回使用するものはこちらです
コード
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()
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
- 例えば
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で改行を削除しています。
感想
簡単に対話式ツールが作れてよかったです。
こういった複数回awscliを叩かなくてはいけない、というときにとても便利だと思います。
参考
おまけ
filter-log-eventsを使用すればログストリームをまたいで検索が出来ました。
ロググループ名が特定されている場合はこちらを使用するのがいいかもしれません。
調査不足...
Discussion