Skip to content

Instantly share code, notes, and snippets.

@Eyjafjallajokull

Eyjafjallajokull/README.md

Last active April 8, 2025 15:20
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save Eyjafjallajokull/4e917414cfb191391f9e51f6a8c3e46a to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save Eyjafjallajokull/4e917414cfb191391f9e51f6a8c3e46a to your computer and use it in GitHub Desktop.
Download ZIP

Revisions

  1. Eyjafjallajokull revised this gist Jun 30, 2019. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion README.md
    View file
    Original file line number Diff line number Diff line change
    @@ -1,6 +1,6 @@
    This script can help you find and remove unused AWS snapshots and volumes.

    The script has an encoded list of regions that it searches, adjust the value to suit your needs
    There is hardcoded list of regions that it searches, adjust the value to suit your needs.

    Use `snapshot.py snapshot-report` to generate `report.csv` containing information about all snapshots.

  2. Eyjafjallajokull revised this gist Jun 30, 2019. 2 changed files with 207 additions and 41 deletions.
    23 changes: 23 additions & 0 deletions README.md
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,23 @@
    This script can help you find and remove unused AWS snapshots and volumes.

    The script has an encoded list of regions that it searches, adjust the value to suit your needs

    Use `snapshot.py snapshot-report` to generate `report.csv` containing information about all snapshots.

    `snapshot.py snapshot-cleanup` lets you interactively delete snapshot if it finds it is referencing unexisting resources.

    ```
    ./snapshots.py --help
    Usage: snapshots.py [OPTIONS] COMMAND [ARGS]...
    Helper commands for Snapshots management.
    Options:
    --help Show this message and exit.
    Commands:
    snapshot-cleanup Find and delete unreferenced snapshots.
    snapshot-delete Delete single snapshot by id.
    snapshot-report Find unreferenced snapshots.
    volume-cleanup Find and delete unused volumes.
    ```
    225 changes: 184 additions & 41 deletions snapshots.py
    View file
    Original file line number Diff line number Diff line change
    @@ -1,74 +1,217 @@
    #!/usr/local/bin/python3
    import csv
    import re
    from collections import OrderedDict
    from pprint import pprint

    import boto3
    import csv
    import click
    from botocore.exceptions import ClientError

    ec2 = boto3.client('ec2')
    regions = ['us-west-1', 'eu-central-1']


    ec2 = None
    exists_icon = '✅'
    not_exists_icon = '❌'


    @click.group()
    def cli():
    '''
    Helper commands for Snapshots management.
    '''
    pass


    @cli.command()
    def snapshot_report():
    '''
    Find unreferenced snapshots.
    '''
    global ec2
    with open('report.csv', 'w') as csv_file:
    csv_writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
    csv_writer.writerow([
    'id',
    'volume_id',
    'volume_exists',
    'ami_id',
    'ami_exists',
    'instance_id',
    'instance_exists',
    'size',
    'start_time',
    'description' ])
    for region in regions:
    ec2 = boto3.client('ec2', region_name=region)
    for snapshot in get_snapshots():
    csv_writer.writerow([
    snapshot['id'],
    snapshot['volume_id'],
    snapshot['volume_exists'],
    snapshot['ami_id'],
    snapshot['ami_exists'],
    snapshot['instance_id'],
    snapshot['instance_exists'],
    str(snapshot['size']) + 'gb',
    str(snapshot['start_time']),
    snapshot['description']])

    @cli.command()
    def snapshot_cleanup():
    '''
    Find and delete unreferenced snapshots.
    '''
    global ec2
    print('{:22} {:23} {:23} {:23} {:>7} {:25} {:30}'.format('snapshot id', 'volume id', 'ami id',
    'instance id', 'size', 'start time', 'description'))

    for region in regions:
    ec2 = boto3.client('ec2', region_name=region)
    print('region={}'.format(region))
    for snapshot in get_snapshots():
    volume_exists = exists_icon if snapshot['volume_exists'] else not_exists_icon
    ami_exists = exists_icon if snapshot['ami_exists'] else not_exists_icon
    instance_exists = exists_icon if snapshot['instance_exists'] else not_exists_icon
    print('{:22} {:22} {:22} {:22} {:>7} {:25} {:30}'.format(
    snapshot['id'],
    snapshot['volume_id'] + volume_exists,
    snapshot['ami_id'] + ami_exists,
    snapshot['instance_id'] + instance_exists,
    str(snapshot['size']) + 'gb',
    str(snapshot['start_time']),
    snapshot['description']
    ))
    if not snapshot['volume_exists'] and not snapshot['ami_exists'] and not snapshot['instance_exists'] and click.confirm('Delete?', default=True):
    ec2.delete_snapshot(SnapshotId=snapshot['id'])


    @cli.command()
    def volume_cleanup():
    '''
    Find and delete unused volumes.
    '''
    global ec2
    print('{:23} {:20} {:>7} {:10} {:23}'.format(
    'volume id', 'status', 'size', 'created', 'snapshot id'))

    for region in regions:
    ec2 = boto3.client('ec2', region_name=region)
    print('region={}'.format(region))
    for volume in get_available_volumes():
    snapshot_exists = exists_icon if volume['snapshot_exists'] else not_exists_icon
    print('{:23} {:20} {:>7} {:10} {:22}'.format(
    volume['id'],
    volume['status'],
    str(volume['size']) + 'gb',
    volume['create_time'].strftime('%Y-%m-%d'),
    volume['snapshot_id'] + snapshot_exists

    ))
    if not volume['snapshot_exists']:
    print('Tags:')
    print(' '+('\n '.join(['{}={}'.format(click.style(key, fg='blue'), tag)
    for key, tag in volume['tags'].items()])))
    if click.confirm('Delete?', default=True):
    ec2.delete_volume(VolumeId=volume['id'])


    @cli.command()
    @click.argument('snapshot_id')
    def snapshot_delete(snapshot_id):
    '''
    Delete single snapshot by id.
    '''
    try:
    ec2.delete_snapshot(SnapshotId=snapshot_id)
    print('Deleted ' + snapshot_id)
    except ClientError as e:
    print('Failed to delete ' + snapshot_id)
    print(e)


    def get_snapshots():
    return ec2.describe_snapshots(OwnerIds=['self'])['Snapshots']
    '''
    Get all snapshots.
    '''
    for snapshot in ec2.describe_snapshots(OwnerIds=['self'])['Snapshots']:
    instance_id, image_id = parse_description(snapshot['Description'])
    yield {
    'id': snapshot['SnapshotId'],
    'description': snapshot['Description'],
    'start_time': snapshot['StartTime'],
    'size': snapshot['VolumeSize'],
    'volume_id': snapshot['VolumeId'],
    'volume_exists': volume_exists(snapshot['VolumeId']),
    'instance_id': instance_id,
    'instance_exists': instance_exists(instance_id),
    'ami_id': image_id,
    'ami_exists': image_exists(image_id),
    }


    def get_available_volumes():
    '''
    Get all volumes in 'available' state. (Volumes not attached to any instance)
    '''
    for volume in ec2.describe_volumes(Filters=[{'Name': 'status', 'Values': ['available']}])['Volumes']:
    yield {
    'id': volume['VolumeId'],
    'create_time': volume['CreateTime'],
    'status': volume['State'],
    'size': volume['Size'],
    'snapshot_id': volume['SnapshotId'],
    'snapshot_exists': str(snapshot_exists(volume['SnapshotId'])),
    'tags': OrderedDict(sorted([(tag['Key'], tag['Value']) for tag in volume['Tags']])),
    }


    def snapshot_exists(snapshot_id):
    if not snapshot_id:
    return ''
    try:
    ec2.describe_snapshots(SnapshotIds=[snapshot_id])
    return True
    except ClientError:
    return False


    def volume_exists(volume_id):
    if not volume_id: return ''
    if not volume_id:
    return False
    try:
    ec2.describe_volumes(VolumeIds=[volume_id])
    return True
    except ClientError:
    return False


    def instance_exists(instance_id):
    if not instance_id: return ''
    if not instance_id:
    return ''
    try:
    ec2.describe_instances(InstanceIds=[instance_id])
    return True
    return len(ec2.describe_instances(InstanceIds=[instance_id])['Reservations']) != 0
    except ClientError:
    return False


    def image_exists(image_id):
    if not image_id: return ''
    if not image_id:
    return ''
    try:
    requestObj = ec2.describe_images(ImageIds=[image_id,])
    if not requestObj["Images"]:
    return False
    return True
    return len(ec2.describe_images(ImageIds=[image_id])['Images']) != 0
    except ClientError:
    return False


    def parse_description(description):
    regex = r"^Created by CreateImage\((.*?)\) for (.*?) "
    matches = re.finditer(regex, description, re.MULTILINE)
    for matchNum, match in enumerate(matches):
    return match.groups()
    return '', ''

    def main():
    with open('report.csv', 'w') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow([
    'snapshot id',
    'description',
    'started',
    'size',
    'volume',
    'volume exists',
    'instance',
    'instance exists',
    'ami',
    'ami exists'])
    for snap in get_snapshots():
    instance_id, image_id = parse_description(snap['Description'])
    writer.writerow([
    snap['SnapshotId'],
    snap['Description'],
    snap['StartTime'],
    str(snap['VolumeSize']),
    snap['VolumeId'],
    str(volume_exists(snap['VolumeId'])),
    instance_id,
    str(instance_exists(instance_id)),
    image_id,
    str(image_exists(image_id)),
    ])

    if __name__ == '__main__':
    main()
    cli()
  3. Eyjafjallajokull revised this gist Jun 30, 2019. 1 changed file with 1 addition and 1 deletion.
    2 changes: 1 addition & 1 deletion snapshots.py
    View file
    Original file line number Diff line number Diff line change
    @@ -42,7 +42,7 @@ def parse_description(description):
    return '', ''

    def main():
    with open('raport.csv', 'w') as csvfile:
    with open('report.csv', 'w') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow([
    'snapshot id',
  4. Eyjafjallajokull revised this gist Oct 22, 2018. 1 changed file with 4 additions and 2 deletions.
    6 changes: 4 additions & 2 deletions snapshots.py
    View file
    Original file line number Diff line number Diff line change
    @@ -27,7 +27,9 @@ def instance_exists(instance_id):
    def image_exists(image_id):
    if not image_id: return ''
    try:
    ec2.describe_images(ImageIds=[image_id,])
    requestObj = ec2.describe_images(ImageIds=[image_id,])
    if not requestObj["Images"]:
    return False
    return True
    except ClientError:
    return False
    @@ -69,4 +71,4 @@ def main():
    ])

    if __name__ == '__main__':
    main()
    main()
  5. Paweł Olejniczak renamed this gist Nov 2, 2016. 1 changed file with 0 additions and 0 deletions.
    0 gistfile1.txt → snapshots.py
    View file
    File renamed without changes.
  6. Paweł Olejniczak created this gist Nov 2, 2016.
    72 changes: 72 additions & 0 deletions gistfile1.txt
    View file
    Original file line number Diff line number Diff line change
    @@ -0,0 +1,72 @@
    import re
    import boto3
    import csv
    from botocore.exceptions import ClientError

    ec2 = boto3.client('ec2')

    def get_snapshots():
    return ec2.describe_snapshots(OwnerIds=['self'])['Snapshots']

    def volume_exists(volume_id):
    if not volume_id: return ''
    try:
    ec2.describe_volumes(VolumeIds=[volume_id])
    return True
    except ClientError:
    return False

    def instance_exists(instance_id):
    if not instance_id: return ''
    try:
    ec2.describe_instances(InstanceIds=[instance_id])
    return True
    except ClientError:
    return False

    def image_exists(image_id):
    if not image_id: return ''
    try:
    ec2.describe_images(ImageIds=[image_id,])
    return True
    except ClientError:
    return False

    def parse_description(description):
    regex = r"^Created by CreateImage\((.*?)\) for (.*?) "
    matches = re.finditer(regex, description, re.MULTILINE)
    for matchNum, match in enumerate(matches):
    return match.groups()
    return '', ''

    def main():
    with open('raport.csv', 'w') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow([
    'snapshot id',
    'description',
    'started',
    'size',
    'volume',
    'volume exists',
    'instance',
    'instance exists',
    'ami',
    'ami exists'])
    for snap in get_snapshots():
    instance_id, image_id = parse_description(snap['Description'])
    writer.writerow([
    snap['SnapshotId'],
    snap['Description'],
    snap['StartTime'],
    str(snap['VolumeSize']),
    snap['VolumeId'],
    str(volume_exists(snap['VolumeId'])),
    instance_id,
    str(instance_exists(instance_id)),
    image_id,
    str(image_exists(image_id)),
    ])

    if __name__ == '__main__':
    main()