Commands for creating and deleting lvm snapshots

lvmsnapshot.py

+#!/usr/bin/env python3
+
+# create:
+# xfs_freeze -f <parent mountpoint>
+# lvcreate -s -n <snapshot name> -pr <parent volume>
+# xfs_freeze -u <parent mountpoint>
+# lvchange -ay -K <snapshot volume>
+# mount <snapshot device> <snapshot mountpoint> -nouuid,ro
+
+# delete:
+# umount <snapshot mountpoint>
+# lvchange -an -L <snapshot volume>
+# lvremove <snapshot volume>
+
+# naming scheme:
+# snapshot name: <parent mountpoint>-snapshot-<yyyy-mm-dd-hh-MM>
+# snapshot mountpoint: <base dir>/<parent mountpoint>/<yyyy-mm-dd-hh-MM>
+
+# how many snapshots to keep:
+# give number of snapshots for a given period
+# sort by period
+# mark snapshots to keep, from oldest within period to new
+# delete unmarked snapshots
+# periods are cumulative
+
+import os
+import re
+from datetime import datetime
+import subprocess as sp
+import json
+from contextlib import contextmanager
+
+SNAPSHOT_BASE_DIR = "/snapshots"
+TIMESTAMP_FORMAT = "%Y-%m-%d-%H-%M"
+
+@contextmanager
+def xfs_freeze(mountpoint):
+    freeze_xfs(True)
+    yield
+    freeze_xfs(False)
+
+def freeze_xfs(mountpoint, freeze):
+    command = [
+        "/usr/sbin/xfs_freeze",
+        "-f" if freeze else "-u",
+        mountpoint
+    ]
+    sp.run(command, check=True)
+
+class Snapshot:
+    def __init__(self, volume_group, parent_volume, timestamp):
+        self.volume_group = volume_group
+        self.parent_volume = parent_volume
+        self.timestamp = timestamp
+
+    def get_full_parent_volume(self):
+        return "{}/{}".format(self.volume_group, self.volume)
+
+    def get_timestamp_str(self):
+        return self.timestamp.strftime(TIMESTAMP_FORMAT)
+
+    def get_name(self):
+        return "{}-snapshot-{}".format(
+            self.parent_volume,
+            self.get_timestamp_str())
+
+    def get_full_volume(self):
+        return "{}/{}".format(self.volume_group, self.get_name())
+
+    def get_mountpoint(self):
+        return os.path.join([
+            SNAPSHOT_BASE_DIR, self.parent_volume, self.get_timestamp_str()
+        ])
+
+    def get_parent_mapper_device(self):
+        return "/dev/mapper/{}-{}".format(self.volume_group, self.parent_volume)
+
+    def get_mapper_device(self):
+        return "/dev/mapper/{}-{}".format(self.volume_group, self.get_name())
+
+    def get_parent_mountpoint(self):
+        command = [
+            "/bin/findmnt",
+            "--source", self.get_parent_mapper_device(),
+            "--json"
+        ]
+        result = sp.check_output(command)
+        data = json.loads(result)
+        filesystems = data["filesystems"]
+        if len(filesystems) < 1:
+            raise Exception("No parent device {} found!".format(self.get_parent_mapper_device()))
+        return filesystems[0]["target"]
+
+    def create(self):
+        parent_mountpoint = self.get_parent_mountpoint()
+        with xfs_freeze(mountpoint):
+            create_command = [
+                "/sbin/lvcreate",
+                "--snapshot",
+                "--name", self.get_name(),
+                "--permission", "r",
+                self.get_full_parent_volume()
+            ]
+            sp.run(create_command, check=True)
+        self.mount()
+
+    def mount(self):
+        activate_command = [
+            "/sbin/lvchange",
+            "--activate", "y",
+            "--ignoreactivationskip",
+            self.get_full_volume()
+        ]
+        sp.run(activate_command, check=True)
+        mount_command = [
+            "/bin/mount",
+            self.get_mapper_device(),
+            self.get_mountpoint(),
+            "-onouuid,ro" # nouuid is necessary for xfs snapshots
+        ]
+        sp.run(mount_command, check=True)
+
+    def remove(self):
+        self.unmount()
+        remove_command = [
+            "/sbin/lvremove",
+            self.get_full_volume()
+        ]
+        sp.run(remove_command, check=True)
+
+    def unmount(self):
+        unmount_command = [
+            "/bin/umount",
+            self.get_mountpoint()
+        ]
+        sp.run(unmount_command, check=True)
+        deactivate_command = [
+            "/sbin/lvchange",
+            "--activate", "n",
+            "--ignoreactivationskip",
+            self.get_full_volume()
+        ]
+        sp.run(deactivate_command, check=True)
+