AWS BatchのジョブIDからログを特定し印字するpythonスクリプト

AWS Batchのログをコマンドラインから見たかったので、AWS BatchのAPIを叩いてCloud Watchのグループ名とストリーム名を特定し、標準出力に印字するスクリプト作りました

"""
AWS BatchのジョブIDからログを特定し印字するpythonスクリプト
"""
import datetime
import sys

import boto3
import click


@click.command()
@click.option("--job_id", "-j", help="Job ID of AWS Batch", required=True)
def main(job_id: str) -> None:
    client = boto3.client("batch")
    response = client.describe_jobs(jobs=[job_id])
    if len(response["jobs"]) == 0:
        print("Jobs not found")
        sys.exit(1)
    log_stream_name = response["jobs"][0]["container"]["logStreamName"]
    log_group_name = response["jobs"][0]["container"]["logConfiguration"]["options"]["awslogs-group"]
    print(f"log_group_name = {log_group_name}")
    print(f"log_stream_name = {log_stream_name}")
    print("-------------------------------------------------")
    client = boto3.client("logs")
    next_token = None
    while True:
        args = {"logGroupName": log_group_name, "logStreamName": log_stream_name, "startFromHead": True}
        if next_token is not None:
            args["nextToken"] = next_token
        log_events = client.get_log_events(**args)
        events = log_events["events"]
        if len(events) == 0:
            break  # 書き込むログがなくなったのでwhileループを抜ける
        for log_line in events:
            dt = datetime.datetime.fromtimestamp(log_line["timestamp"] / 1000)
            message = log_line["message"]
            print(f"[{dt.isoformat()}] {message}")
        next_token = log_events["nextForwardToken"]
    print("-------------------------------------------------")


if __name__ == "__main__":
    main()

実行例

$ python print_awsbatch_log.py -j 9ffbad12-b57b-4ae2-81ad-c17045fa9f10

log_group_name = /aws/batch/my_log_group
log_stream_name = my_job_def/default/e79f3ac231ff4042840e3a5d93c4a357
-------------------------------------------------
[2022-12-19T17:07:59.908000] ログの１行目
[2022-12-19T17:08:00.050000] ログの２行目
[2022-12-19T17:08:00.054000] ログの３行目
-------------------------------------------------