Dump AWS logs entries by in timestamp range

aws_log_dump.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Dump AWS logs entries by in timestamp range"""
import argparse
from datetime import datetime
import logging
import json

import boto3

logging.basicConfig()
LOG = logging.getLogger(__name__)
LOG.setLevel(logging.INFO)


LOG_GROUP_NAME = "/aws/lambda/whatever-the-path-is"


def _iter_steam_names(cli, args):
    start_dt = datetime.strptime(args.start, "%Y-%m-%d %H:%M")
    end_dt = datetime.strptime(args.end, "%Y-%m-%d %H:%M")
    pager = cli.get_paginator("describe_log_streams")
    response = pager.paginate(
        logGroupName=LOG_GROUP_NAME,
        orderBy="LastEventTime",
        descending=True,
    )
    for page in response:
        for stream in page["logStreams"]:
            stream_name = stream["logStreamName"]
            last_ts = stream["lastEventTimestamp"]
            first_ts = stream["firstEventTimestamp"]
            #  LOG.info(
            #      "stream_name: %s, last_ts: %s, first_ts: %s",
            #      stream_name,
            #      last_ts,
            #      first_ts,
            #  )
            if last_ts < (start_dt.timestamp() * 1000) or first_ts > (
                end_dt.timestamp() * 1000
            ):
                # LOG.info("Skipping stream %s", stream_name)
                continue
            yield stream_name


def run(args):
    """Run."""
    session = boto3.Session(region_name="us-east-2")
    cli = session.client("logs")
    names = list(_iter_steam_names(cli, args))
    LOG.info("Found %s streams", len(names))
    pager = cli.get_paginator("filter_log_events")
    response = pager.paginate(
        logGroupName=LOG_GROUP_NAME,
        logStreamNames=names,
        filterPattern="",
    )
    for page in response:
        for event in page["events"]:
            message = event["message"]
            try:
                message = json.loads(message)
                LOG.info("%s\n", json.dumps(message, indent=2))
            except json.JSONDecodeError:
                LOG.info("%s\n", message)


def main():
    """Run main function."""
    parser = argparse.ArgumentParser(
        formatter_class=argparse.RawDescriptionHelpFormatter,
        description=__doc__,
    )
    parser.add_argument(
        "--start",
        help="Start time (YYYY-MM-DD HH:MM)",
        required=True,
        type=str,
    )
    parser.add_argument(
        "--end",
        help="End time, (YYYY-MM-DD HH:MM)",
        required=True,
        type=str,
    )
    args = parser.parse_args()
    #  args = parser.parse_args(
    #      [
    #          "--start",
    #          "2022-06-16 12:30",
    #          "--end",
    #          "2022-06-16 12:45",
    #      ]
    #  )
    run(args)


if __name__ == "__main__":
    main()