From 7ada92bb32d83a310e70d8dd9c669c2ad648a16b Mon Sep 17 00:00:00 2001 From: timeshifter Date: Fri, 3 Dec 2021 20:23:49 +0100 Subject: [PATCH] refactoring --- zfs-backup.py | 74 ++++++++++++++++++++++++++++++--------------------- 1 file changed, 44 insertions(+), 30 deletions(-) diff --git a/zfs-backup.py b/zfs-backup.py index 303707b..f60c800 100755 --- a/zfs-backup.py +++ b/zfs-backup.py @@ -30,6 +30,7 @@ SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are DISK_BY_UUID = Path("/dev/disk/by-uuid") MAPPER_PATH = Path("/dev/mapper") +# noinspection SpellCheckingInspection """ Erstes Backup muss für alle notwendigen Datasets durchgeführt werden mit # zfs send -Rv rpool/ROOT@zfs-auto-snap_daily-2018-06-14-1648 | zfs receive -dvF backup1 @@ -119,7 +120,7 @@ class Manager: local_snap for local_snap in list(cls._find_snapshots_with_backup_tag(local_dataset)) for remote_snap in list(cls._find_snapshots_with_backup_tag(remote_dataset)) - if local_snap.snapname == remote_snap.snapname + if local_snap.snapshot_name == remote_snap.snapshot_name ] common_snapshots.sort() return common_snapshots[-1] @@ -131,38 +132,38 @@ def _get_regex_matching_snapshots_with_tags(tags: list[str]): class ZFSPath: - def __init__(self, elements: list[str], snapname: str | None = None): + def __init__(self, elements: list[str], snapshot_name: str | None = None): self._elements = elements - self._snapname = snapname + self._snapshot_name = snapshot_name self._is_snapshot = False - if snapname: + if snapshot_name: self._is_snapshot = True self._sanity_check() def __repr__(self): path = "/".join(self._elements) if self._is_snapshot: - return "@".join([path, self._snapname]) + return "@".join([path, self._snapshot_name]) return path def __eq__(self, other: ZFSPath): return ( - self._elements == other._elements - and self._is_snapshot == other._is_snapshot - and self._snapname == other._snapname + self._elements == other._elements + and self._is_snapshot == other._is_snapshot + and self._snapshot_name == other._snapshot_name ) @classmethod def from_string(cls, string: str) -> ZFSPath: - snapname = None + snapshot_name = None if "@" in string: - string, snapname = string.split("@") + string, snapshot_name = string.split("@") elements: list[str] = string.split("/") - return ZFSPath(elements, snapname) + return ZFSPath(elements, snapshot_name) @property - def snapname(self) -> str: - return self._snapname + def snapshot_name(self) -> str: + return self._snapshot_name @property def is_snapshot(self) -> bool: @@ -177,13 +178,13 @@ class ZFSPath: return ZFSPath(self._elements) def replace_pool(self, pool: str) -> ZFSPath: - return ZFSPath([pool] + self._elements[1:].copy(), self.snapname) + return ZFSPath([pool] + self._elements[1:].copy(), self.snapshot_name) @property def name_without_pool(self) -> str: name = "/".join(self._elements[1:]) if self.is_snapshot: - name = "@".join([name, self.snapname]) + name = "@".join([name, self.snapshot_name]) return name def _sanity_check(self): @@ -200,17 +201,26 @@ def _assert_type(instance: Any, object_type: Any, message: str) -> None: class ZPOOL: _ZPOOL = Path("/usr/bin/zpool") + class _Subcommands: + # -N: no mount + # -d: directory to search the pool in + import_ = ["import", "-N", "-d"] + export = ["export"] + scrub = ["scrub"] + status = ["status"] + wait = ["wait", "-t", "scrub"] + @classmethod def import_from_directory(cls, pool: Pool, directory: Path) -> None: _assert_type(pool, Pool, "Can only import Pool objects") - # -N: no mount - # -d: directory to search the pool in - CommandRunner.run([cls._ZPOOL, "import", "-N", "-d", directory, pool.name]) + CommandRunner.run( + [str(cls._ZPOOL), *cls._Subcommands.import_, directory, pool.name] + ) @classmethod def export(cls, pool: Pool) -> None: _assert_type(pool, Pool, "Can only export Pool objects") - CommandRunner.run([cls._ZPOOL, "export", pool.name]) + CommandRunner.run([str(cls._ZPOOL), *cls._Subcommands.export, pool.name]) @classmethod def scrub( @@ -223,19 +233,22 @@ class ZPOOL: _assert_type(pool, Pool, "Can only scrub pools.") cls._scrub_if_necessary(pool, skip_if_recently_scrubbed) if wait_for_finish: - CommandRunner.run([cls._ZPOOL, "wait", "-t", "scrub", pool.name]) + CommandRunner.run([str(cls._ZPOOL), *cls._Subcommands.wait, pool.name]) if not cls._healthy(pool): raise IOError(f"Pool {pool.name} is not healthy.") @classmethod def _scrub_if_necessary(cls, pool: Pool, skip_if_recently_scrubbed: bool) -> None: + def do_scrub(): + CommandRunner.run([str(cls._ZPOOL), *cls._Subcommands.scrub, pool.name]) + if cls._scrub_in_progress(pool): return if not skip_if_recently_scrubbed: - CommandRunner.run([cls._ZPOOL, "scrub", pool.name]) + do_scrub() else: if not cls._recently_scrubbed(pool): - CommandRunner.run([cls._ZPOOL, "scrub", pool.name]) + do_scrub() @classmethod def _recently_scrubbed(cls, pool: Pool) -> bool: @@ -249,9 +262,9 @@ class ZPOOL: def _healthy(cls, pool: Pool) -> bool: return "ONLINE" in cls._get_pool_status_output(pool) - @staticmethod - def _get_pool_status_output(pool: Pool) -> str: - return CommandRunner.get_output([ZFS, "status", pool.name]) + @classmethod + def _get_pool_status_output(cls, pool: Pool) -> str: + return CommandRunner.get_output([ZFS, *cls._Subcommands.status, pool.name]) @classmethod def _last_scrub(cls, pool: Pool) -> datetime: @@ -276,8 +289,8 @@ class ZFS: # -R Replicate filesystem # -I send all intermediary snapshots send = ["send", "-R", "-I"] - # -d Discard the first element of the sent snapshot's file system name - # -F Force a rollback of the file system to the most recent snapshot before performing the receive operation. + # -d Discard the first element of the "send" snapshot's file system name + # -F Force a rollback of the file system to the most recent snapshot before performing the "receive". # -u File system that is associated with the received stream is not mounted. receive = ["receive", "-d", "-F", "-u"] @@ -411,6 +424,7 @@ class Cryptsetup: close = "close" status = "status" open = "open" + _CRYPTSETUP = Path("/usr/bin/cryptsetup") @classmethod @@ -578,14 +592,14 @@ class Snapshot: return cls(ZFSPath.from_string(qualified_name)) def __lt__(self, other: Snapshot) -> bool: - return self.snapname < other.snapname + return self.snapshot_name < other.snapshot_name def __repr__(self): return str(self._path) @property - def snapname(self) -> str: - return self._path.snapname + def snapshot_name(self) -> str: + return self._path.snapshot_name def matches_regex(self, regex: str) -> bool: return bool(search(regex, str(self._path)))