Delete empty log streams from AWS CloudWatch Logs

del_empty_log_streams.py

#!/usr/bin/env python
"""
This script deletes empty log streams from cloudwatch log groups.

Your shell should be configured for connecting to the AWS API. This can be done
with the CLI command `aws configure`.

This script requires Python 3.6 or newer, and you must have boto3 installed.
"""

import argparse
import os
import boto3

LOG_ONLY = False


def get_log_groups():
    """Return a list containing the names of all log groups."""
    client = boto3.client('logs')

    found_all = False
    next_token = None
    group_names = []

    while not found_all:
        if next_token is not None:
            response = client.describe_log_groups(nextToken=next_token)
        else:
            response = client.describe_log_groups()
        if 'logGroups' in response:
            for group in response['logGroups']:
                if 'logGroupName' in group:
                    group_names.append(group['logGroupName'])
        next_token = response.get('nextToken', None)
        if next_token is None:
            found_all = True

    return group_names


def get_log_streams(group_name):
    """Return a list of log stream names in group `group_name`."""
    client = boto3.client('logs')

    found_all = False
    next_token = None
    stream_names = []

    while not found_all:
        if next_token is not None:
            response = client.describe_log_streams(logGroupName=group_name,
                                                   nextToken=next_token)
        else:
            response = client.describe_log_streams(logGroupName=group_name)
        if 'logStreams' in response:
            for stream in response['logStreams']:
                if 'logStreamName' in stream:
                    stream_names.append(stream['logStreamName'])
        next_token = response.get('nextToken', None)
        if next_token is None:
            found_all = True

    return stream_names


def stream_has_events(group_name, stream_name):
    """Return whether or not `stream_name` in `group_name` has any events."""
    client = boto3.client('logs')

    # Return the first event.
    response = client.get_log_events(logGroupName=group_name,
                                     logStreamName=stream_name,
                                     limit=1)

    if 'events' in response and len(response['events']) > 0:
        return True
    return False


def get_group_empty_streams(group_name):
    """Return a list of streams that are empty in `group_name`."""
    streams = get_log_streams(group_name)
    empty_streams = []
    for stream in streams:
        if not stream_has_events(group_name, stream):
            empty_streams.append(stream)

    return empty_streams


def del_empty_stream(group_name, stream_name):
    """Delete `stream_name` from `group_name`."""
    client = boto3.client('logs')
    client.delete_log_stream(logGroupName=group_name,
                             logStreamName=stream_name)

def del_empty_streams():
    """Finds all empty streams and deletes them.

    If the global variable `LOG_ONLY` is True, streams are only logged to the
    console and not deleted.
    """
    for group in get_log_groups():
        print(f'Finding empty streams in group `{group}`')
        empty_streams = get_group_empty_streams(group)
        for stream in empty_streams:
            if LOG_ONLY:
                print(
                    f'Stream `{stream}` in group `{group}` is empty. '
                    'Would be deleted...'
                )
            else:
                print(
                    f'Stream `{stream}` in group `{group}` is empty. '
                    'Deleting...'
                )
                del_empty_stream(group, stream)


def main():
    """Main handler for this script."""
    parser = argparse.ArgumentParser(description='Delete empty log streams.')
    parser.add_argument(
        '--log-only',
        action='store_true',
        help="If specified, only log to console, don't actually delete.")
    args = parser.parse_args()
    global LOG_ONLY
    LOG_ONLY = args.log_only

    del_empty_streams()


if __name__ == "__main__":
    main()