Show cloudwatch log events

cwlog.py

#!/usr/bin/env python3

"""Show cloudwatch log events.

Examples:

Show log stream containing the most recent event for a log group:

  cwlog.py -g /aws/lambda/lambda-labreport

List log streams for a log group:

  cwlog.py -g /aws/lambda/lambda-labreport -l

List a specific log stream:

  cwlog.py -g /aws/lambda/lambda-labreport \
    -s '2019/10/07/[$LATEST]c1535335102d479b90ce28e6061cb506'

"""

import sys
import argparse
from datetime import datetime
import io

import boto3


def main(arguments):

    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument('-g', '--log-group')
    parser.add_argument('-s', '--log-stream',
                        help="""Optional log stream name (use stream
                        containing most recent event by default)""")
    parser.add_argument('-l', '--list-streams', action='store_true', default=False,
                        help="list log streams for the specified group and exit")
    parser.add_argument('-e', '--errors-only', action='store_true', default=False)

    args = parser.parse_args(arguments)

    client = boto3.client('logs')

    if not args.log_group:
        groups = client.describe_log_groups()
        for g in groups['logGroups']:
            print(g['logGroupName'])
        return 0

    streams = client.describe_log_streams(
        logGroupName=args.log_group,
        orderBy='LastEventTime',
        descending=True,
    )

    if args.list_streams:
        for s in streams['logStreams']:
            print(s['logStreamName'])
        return 0

    # first should be most recent
    stream_name = streams['logStreams'][0]['logStreamName']

    events = client.get_log_events(
        logGroupName=args.log_group,
        logStreamName=stream_name,
    )

    for event in events['events']:
        ts = int(str(event['timestamp'])[:10])
        dt = datetime.utcfromtimestamp(ts)
        msg = io.StringIO(event['message'], newline=None).read()
        if args.errors_only and not msg.startswith('[ERROR]'):
            continue

        for line in msg.splitlines():
            print(dt, line)


if __name__ == '__main__':
    sys.exit(main(sys.argv[1:]))