Helper script to remove old ZFS snapshots

zfs-destroy-snapshots.py

#!/usr/bin/env python3
"""
Helper script to remove old ZFS snapshots
Copyright (c) Onlime GmbH, https://www.onlime.ch
"""
import argparse
from re import compile
from datetime import datetime, timedelta
from subprocess import check_output
import sys

def list_snapshots(age: int, prefix: str = None, exclude: str = None, dataset: str = None, destroy: bool = False):
    # List all snapshot names of a dataset sorted by creation date; sample command:
    # $ zfs list -H -t snapshot -o name,creation -s creation [<dataset>]
    cmd = ['zfs', 'list', '-H', '-t', 'snapshot', '-o', 'name,creation', '-s', 'creation']
    if dataset:
        cmd.append(dataset)
    snap_list = check_output(cmd).decode(sys.stdout.encoding)
    snap_lines = snap_list.splitlines()
    # reverse sort order (`zfs list` only supports asc sort order), newest snapshots on top
    snap_lines.reverse()

    for line in snap_lines:
        snapshot, creation = line.split('\t')
        if prefix and '@{}'.format(prefix) not in snapshot:
            continue
        if exclude and exclude in snapshot:
            continue
        snap_date = datetime.strptime(creation, '%a %b %d %H:%M %Y')
        if snap_date < datetime.now()-timedelta(hours=age):
            print('{}{} {}'.format(
                'DESTROYING ' if destroy else '',
                snap_date.strftime('%Y-%m-%d %H:%M'), 
                snapshot
            ))
            if destroy:
                check_output(['zfs', 'destroy', snapshot])

if __name__ == "__main__": # Only use the argument parsing if the script was called as script and not imported
    # argument parsing
    parser = argparse.ArgumentParser()
    parser.add_argument("age", type=int, help="List snapshots that are older than age (hours).")
    parser.add_argument("--prefix", help="Only list snapshot with this prefix.")
    parser.add_argument("--exclude", help="Exclude snapshots that contain this string.")
    parser.add_argument("--dataset", help="Limit to a specific dataset.")
    parser.add_argument("--destroy", action='store_true', help="Destroy snapshots that are older than age.")
    args = parser.parse_args()

    list_snapshots(args.age, args.prefix, args.exclude, args.dataset, args.destroy)