Skip to content

Instantly share code, notes, and snippets.

@nhoffman
Created October 7, 2019 03:41
Show Gist options
  • Save nhoffman/6712234cbb5594cf56160dcc1fa8fc32 to your computer and use it in GitHub Desktop.
Save nhoffman/6712234cbb5594cf56160dcc1fa8fc32 to your computer and use it in GitHub Desktop.
Show cloudwatch log events
#!/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:]))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment