Delete Old ZFS Snapshots

The script generates a list of existing snapshots, which should be removed from a dataset. For every month DELCOUNT = 3 snapshots are preserved for deleting. The snapshots from the current month are also protected. The snapshots needs to be in the following string format: pool/dataset@%Y-%m-%d_%H%M, for example storage/documents@2023-04-24_0625.

# deleteOldZFSSnapshots.py

 1| import libs.mockdata as md
 2| import libs.datemagic as dm
 3| import libs.zfsmagic as zm
 4| import subprocess
 5| import datetime
 6| import sys
 7| 
 8| # dummydates = md.mockDates
 9| 
10| DATASET = sys.argv[1]
11| 
12| DATENOW = datetime.datetime.now()
13| DELCOUNT = 3
14| 
15| result = subprocess.run(["zfs", "list", "-t", "snapshot", "-o", "name", DATASET], capture_output=True)
16| #print(result.stdout.splitlines()[2].decode("utf-8"))
17| snapshotentries = result.stdout.splitlines()
18| print("Found snapshotentries:", len(snapshotentries))
19| zfsdict = dm.sortDates(zm.snapOutToDict(snapshotentries))
20| print("Found zfsdict        :", dm.countsnaps(zfsdict))
21| delSmaps = zm.removeProtectedSnaps(zfsdict, DATENOW, DELCOUNT)
22| print("Size cleanup         :", dm.countsnaps(delSmaps))
23| 
24| dm.pSnapsAsLine("zfs destroy", zfsdict)
25| #dm.ppDates(zfsdict)


# libs/datemagic.py

 1| def sortDates(inDict):
 2|     inDict = dict(sorted(inDict.items()))
 3| 
 4|     for year, months in inDict.items():
 5|         # print("Year:", year)
 6|         months = dict(sorted(months.items()))
 7|         inDict[year] = months
 8| 
 9|         for month, days in months.items():
10|             # print("Month:", month)
11|             days = dict(sorted(days.items()))
12|             inDict[year][month] = days
13| 
14|     return inDict
15| 
16| def countsnaps(inDict):
17|     count = 0
18|     for year, months in inDict.items():
19|         for month, days in months.items():
20|             for day, daystring in days.items():
21|                 count = count + 1
22| 
23|     return count
24| 
25| def ppDates(inDict):
26|     ycount = 0
27|     mcount = 0
28|     dcount = 0
29|     for year, months in inDict.items():
30|         ycount = ycount + 1
31|         mcount = 0
32|         print("", year)
33|         for month, days in months.items():
34|             mcount = mcount + 1
35|             dcount = 0
36|             print("  ", month)
37|             for day, daystring in days.items():
38|                 dcount = dcount + 1
39|                 print("  ", "  ", year, month, day,":",daystring)
40|             print("  ", "  ", "Days:", dcount, "for", month, year, "")
41|         print("  ", "Months:", mcount, "for", year, "")
42|     print("Years:", ycount, "")
43| 
44| def pSnapsAsLine(precommand: str, inDict: {}):
45|     for year, months in inDict.items():
46|         for month, days in months.items():
47|             for day, daystring in days.items():
48|                 print(precommand, daystring)


# libs/zfsmagic.py

 1| import datetime
 2| import libs.datemagic as dm
 3| import random
 4| 
 5| 
 6| def snapOutToDict(strin):
 7|     returnDict = {}
 8|     #print(type(returnDict), returnDict)
 9|     #print(type(strin))
10|     for k in strin[1:]:
11|         l = k.decode("UTF-8").split("@")
12|         #print(k, l[0], l[1])
13|         datetime = l[1].split("_")
14|         date = datetime[0].split("-")
15|         #print(date[0], date[1], date[2], k.decode("UTF-8"))
16|         datedict = {date[2]: k.decode("UTF-8")}
17| 
18| 
19|         itm = {date[0]: {date[1]: {date[2]: k.decode("UTF-8")}}}
20|         returnDict = addItemToDict(returnDict, itm)
21| 
22|     return returnDict
23| 
24| def addItemToDict(inDict: {}, item: {}):
25|     year = list(item)[0]
26|     month = list(item[year])[0]
27|     day = list(item[year][month])[0]
28|     daystr = item[year][month][day]
29| 
30|     if year not in inDict:
31|         inDict[year] = {}
32|     if month not in inDict[year]:
33|         inDict[year][month] = {}
34|     if day not in inDict[year][month]:
35|         inDict[year][month][day] = {}
36| 
37|     inDict[year][month][day] = daystr
38| 
39|     return inDict
40| 
41| def removeProtectedSnaps(dictin: {}, today: datetime.date, count: int):
42|     monthstring = str(f"{today.month:02d}")
43|     yearstring = str(today.year)
44|     try:
45|         del dictin[yearstring][monthstring]
46|     except KeyError:
47|         print("Not in Dictionary:", yearstring, monthstring)
48| 
49|     for y in dictin:
50|         for m in dictin[y]:
51|             #print(dictin[y][m])
52|             dictin[y][m] = randomDelMonth(dictin[y][m], count)
53| 
54|     return dictin
55| 
56| def randomDelMonth(dictin: {}, count: int):
57|     lenmonth = len(dictin)
58| 
59|     #print(lenmonth, count)
60|     if lenmonth <= count:
61|         return {}
62| 
63|     for i in range(count):
64|         rndindex = random.randint(0, lenmonth - 1 - i)
65|         tmpkey = list(dictin.keys())[rndindex]
66|         del dictin[tmpkey]
67| 
68|     return dictin


# libs/mockdata.py

 1| mockDates = {
 2|     "2020": {
 3|         "05": {
 4|             "02": "2022-05-02",
 5|             "01": "2022-05-01"
 6|             },
 7|         "03": {
 8|             "06": "2022-03-06",
 9|             "02": "2022-03-02"
10|             },
11|     },
12|     "1990": {
13|         "06": {
14|             "02": "1999-06-02",
15|             "01": "1999-06-01"
16|             },
17|         "02": {
18|             "06": "1999-02-06",
19|             "02": "1999-02-02"
20|             },
21|     }
22| }