diff --git a/zfs-backup.py b/zfs-backup.py index 93093c2..4821bc2 100755 --- a/zfs-backup.py +++ b/zfs-backup.py @@ -29,7 +29,6 @@ def main(): # TODO tests -# TODO docstrings # TODO --skip-import # TODO skip post-backup scrub if nothing was sent (except when pre-scrub was also skipped) # TODO print time estimation while scrub is in progress @@ -70,6 +69,7 @@ class Manager: ZFS.receive(stream, self._external_pool) def _get_user_filtered_dataset_backup_map(self) -> _map_type: + """Get the dataset map, let user confirm new datasets.""" return self._user_confirm_new_datasets(self._get_raw_dataset_map()) def _user_confirm_new_datasets(self, dataset_map: _map_type) -> _map_type: @@ -185,6 +185,10 @@ def _get_regex_matching_snapshots_with_tags(tags: list[str]): class ZFSPath: + """Represents a path in ZFS. + + Used to represent the path for objects of Pool, Dataset and Snapshot classes.""" + def __init__(self, elements: list[str], snapshot_name: str | None = None): self._elements = elements self._snapshot_name = snapshot_name @@ -194,6 +198,7 @@ class ZFSPath: self._sanity_check() def __repr__(self): + """Return the path in the same manner as `zfs`.""" path = "/".join(self._elements) if self._is_snapshot: return "@".join([path, self._snapshot_name]) @@ -208,6 +213,7 @@ class ZFSPath: @classmethod def from_string(cls, string: str) -> ZFSPath: + """Take a string from the `zfs` cli, return a ZFSPath object.""" snapshot_name = None if "@" in string: string, snapshot_name = string.split("@") @@ -215,7 +221,8 @@ class ZFSPath: return ZFSPath(elements, snapshot_name) @property - def snapshot_name(self) -> str: + def snapshot_name(self) -> str | None: + """Return the name of the snapshot, or None if ZFSPath does not point to a snapshot.""" return self._snapshot_name @property @@ -223,18 +230,22 @@ class ZFSPath: return self._is_snapshot @property - def pool(self) -> str: + def pool_name(self) -> str: + """Return the name.""" return self._elements[0] @property def dataset(self) -> ZFSPath: + """Return the `zfs` path down to the dataset (omit snapshot).""" return ZFSPath(self._elements) def replace_pool(self, pool: str) -> ZFSPath: + """Return a new ZFSPath. Replace the old pool with a pool with new name.""" return ZFSPath([pool] + self._elements[1:].copy(), self.snapshot_name) @property def name_without_pool(self) -> str: + """Return the `zfs` path, but without the pool name.""" name = "/".join(self._elements[1:]) if self.is_snapshot: name = "@".join([name, self.snapshot_name]) @@ -285,11 +296,13 @@ class ZPool(CommandInterface): @classmethod def import_from_directory(cls, pool: Pool, directory: Path) -> None: + """Search `directory` for `pool` and import it.""" _assert_type(pool, Pool, "Can only import Pool objects") cls._run_command(*cls._Subcommands.import_, str(directory), pool.name) @classmethod def export(cls, pool: Pool) -> None: + """Export the pool.""" _assert_type(pool, Pool, "Can only export Pool objects") cls._run_command(*cls._Subcommands.export, pool.name) @@ -301,6 +314,9 @@ class ZPool(CommandInterface): skip_if_recently_scrubbed: bool, wait_for_finish: bool = True, ) -> None: + """Scrub the pool. + + Waits for the scrub to finish. Raise an IOError if the pool reports as not healthy after the scrub.""" _assert_type(pool, Pool, "Can only scrub pools.") cls._scrub_if_necessary(pool, skip_if_recently_scrubbed) if wait_for_finish: @@ -310,6 +326,8 @@ class ZPool(CommandInterface): @classmethod def _scrub_if_necessary(cls, pool: Pool, skip_if_recently_scrubbed: bool) -> None: + """Scrub the pool. If""" + def do_scrub(): cls._run_command(*cls._Subcommands.scrub, pool.name) @@ -323,6 +341,7 @@ class ZPool(CommandInterface): @classmethod def _recently_scrubbed(cls, pool: Pool) -> bool: + """Is the interval after the last scrub smaller than the minimum timedelta?""" return datetime.now() - cls._last_scrub(pool) < config.recent_scrub_timedelta @classmethod @@ -335,10 +354,12 @@ class ZPool(CommandInterface): @classmethod def _get_pool_status_output(cls, pool: Pool) -> str: + """Return output of `zpool status [pool]`.""" return cls._get_output(*cls._Subcommands.status, pool.name) @classmethod def _last_scrub(cls, pool: Pool) -> datetime: + """Get time of the last scrub.""" return Time.get_last_scrub_from_status_output(cls._get_pool_status_output(pool)) @@ -373,6 +394,9 @@ class ZFS(CommandInterface): @classmethod def get_first_level_datasets(cls, pool: Pool) -> list[Dataset]: + """Get the first level datasets of a pool. + + If a pool `p` has the datasets `p/a`, `p/a/b` and `p/c`, return `[p/a, p/c]`.""" _assert_type(pool, Pool, "Can only get datasets from Pool objects") datasets_str = cls._get_output(*cls._Subcommands.List.dataset, pool.name) # we omit the first return value, which is the pool, not the dataset @@ -381,6 +405,7 @@ class ZFS(CommandInterface): @classmethod def get_snapshots(cls, dataset: Dataset) -> list[Snapshot]: + """Get list of snapshots that are associated with a dataset.""" _assert_type(dataset, Dataset, "Can only get snapshots from Dataset object") command = [*cls._Subcommands.List.snapshot, dataset.qualified_name] snapshot_names: list[str] = cls._get_output(*command).split("\n") @@ -388,6 +413,7 @@ class ZFS(CommandInterface): @classmethod def destroy_snapshot(cls, snapshot: Snapshot) -> None: + """Non-recursively destroy a snapshot.""" _assert_type(snapshot, Snapshot, "Can only destroy snapshots.") if config.local_pool_to_backup == snapshot.pool: raise Exception(Messages.DELETE_IN_LOCAL_POOL) @@ -397,6 +423,7 @@ class ZFS(CommandInterface): def send_incremental( cls, old_snapshot: Snapshot, new_snapshot: Snapshot ) -> Popen[str]: + """Send the incremental replicating stream between two snapshots.""" cls._pre_send_sanity_checks(old_snapshot, new_snapshot) print(f" {old_snapshot} -> {new_snapshot}") return cls._open_stream( @@ -424,11 +451,13 @@ class ZFS(CommandInterface): @classmethod def receive(cls, stream: Popen[str], pool: Pool): + """Receive a stream generated with `zfs send` into a pool.""" _assert_type(pool, Pool, "Can only receive into pools.") cls._receive_stream(*cls._Subcommands.receive, pool.name, stream=stream) @classmethod def send_absolute(cls, snapshot: Snapshot) -> Popen: + """Send the non-incremental replicating stream of the snapshot.""" print(f" {snapshot} -> [new]") return cls._open_stream(*cls._Subcommands.Send.absolute, str(snapshot)) @@ -448,10 +477,12 @@ class Time: @classmethod def _extract_datetime_segment(cls, line_with_datetime: str) -> str: + """From the line in `zpool status` that includes the date and time, extract only the part with date and time.""" return line_with_datetime.split(" errors on ")[-1] @classmethod def _get_line_with_datetime_from_status_output(cls, output: str) -> str: + """From `zpool status`, extract the line that contains the date and time of the last scrub.""" lines = output.split("\n") for line in lines: if line.startswith(" scan: scrub repaired"): @@ -461,6 +492,7 @@ class Time: @classmethod def get_last_scrub_from_status_output(cls, output: str) -> datetime: + """Given the output of `zfs status`, return the `datetime` object of the last scrub.""" line_with_datetime = cls._get_line_with_datetime_from_status_output(output) date_str = cls._extract_datetime_segment(line_with_datetime) return datetime.strptime(date_str, cls._get_date_format()) @@ -474,6 +506,7 @@ class Disk: self._name = name def decrypt(self) -> None: + """Open a LUKS-encrypted disk with `cryptsetup`. Silently pass if it was already decrypted.""" if self._decrypted: raise ValueError(Messages.ALREADY_DECRYPTED, self._name) self._was_already_decrypted = Cryptsetup.decrypt(self._path, self._mapper_entry) @@ -481,13 +514,16 @@ class Disk: @property def _path(self) -> Path: + """Unique path of the disk using /dev/disk/by-uuid.""" return DISK_BY_UUID / str(self.uuid) @property def _mapper_entry(self) -> str: + """Name under which the decrypted disk will appear in /dev/mapper.""" return f"crypt-{self._name}" def encrypt(self) -> None: + """Close a LUKS-encrypted container with `cryptsetup`.""" if not self._decrypted: raise ValueError(Messages.NOT_DECRYPTED, self._name) if self._was_already_decrypted: @@ -506,10 +542,12 @@ class Cryptsetup(CommandInterface): @classmethod def encrypt(cls, mapper_entry: str) -> None: + """Close a previously open LUKS container.""" cls._run_command(cls._Subcommands.close, mapper_entry) @classmethod def _status(cls, name: str) -> str: + """Return `cryptsetup status` of a device under `/dev/mapper`""" return cls._get_output(cls._Subcommands.status, name) @classmethod @@ -529,12 +567,17 @@ class Cryptsetup(CommandInterface): @classmethod def _get_device_by_uuid_from_status_output(cls, status: str) -> Path: - device = cls._get_device_name(status) + """Given the output of `cryptsetup status [device]`, + get the unique name of the disk as path in /dev/disk/by-uuid.""" + device = cls._get_device_path(status) result = cls._resolve_uuid_of_device(device) return result @classmethod def _resolve_uuid_of_device(cls, device): + """Given a device like `/dev/sda`, resolve its entry under `/dev/disk/by-uuid`. + + Exit if none can be found.""" for item in DISK_BY_UUID.iterdir(): if item.readlink() == device: result = item @@ -546,12 +589,14 @@ class Cryptsetup(CommandInterface): return result @classmethod - def _get_device_name(cls, status): + def _get_device_path(cls, status) -> Path: + """Given the output of `cryptsetup status`, return the path of the device.""" for line in status.split("\n"): if line.startswith(" device:"): device = Path(line.split(":")[-1].strip()) break else: + # TODO better error message print("panic") exit(EXIT_ERROR) # noinspection PyUnboundLocalVariable @@ -559,6 +604,8 @@ class Cryptsetup(CommandInterface): @classmethod def _already_decrypted(cls, path: Path, mapper_entry: str) -> bool: + """Given a path in `/dev/disk/by-uuid` and the corresponding name under `/dev/mapper`, + check if the disk is already decrypted.""" decrypted_disk_path = cls._get_device_by_uuid_from_status_output( cls._status(mapper_entry) ) @@ -566,6 +613,7 @@ class Cryptsetup(CommandInterface): class Pool: + """Represents an internal pool.""" def __init__(self, name: str): self._name = name self._path: ZFSPath = ZFSPath.from_string(name) @@ -580,12 +628,14 @@ class Pool: class ExternalPool(Pool): + """Represents an external pool.""" def __init__(self, name: str, disk: Disk): super().__init__(name) self._imported: bool = False self._disk = disk def import_(self) -> None: + """Import a pool. Raise an error if the pool was previously imported.""" if self._imported: raise ValueError(Messages.ALREADY_IMPORTED, self.name) self._disk.decrypt() @@ -593,22 +643,21 @@ class ExternalPool(Pool): self._imported = True def export(self) -> None: + """Export a previously imported pool. Raise an error if the pool was not previously imported.""" if not self._imported: raise ValueError(Messages.NOT_IMPORTED, self.name) - self._export_pool() + ZPool.export(self) self._disk.encrypt() self._imported = False print(f"Exported {self.name}. Disk {self._disk.uuid} can be removed.") - def _export_pool(self) -> None: - ZPool.export(self) - def scrub(self, skip_if_recently_scrubbed: bool) -> None: print(f"Scrubbing {self._name}") ZPool.scrub(self, skip_if_recently_scrubbed=skip_if_recently_scrubbed) @classmethod def find_from_dict(cls, backup_pools: dict[str, UUID]) -> ExternalPool: + """Using the dict in config.py, find the backup pool.""" present_uuids = [item.name for item in DISK_BY_UUID.iterdir()] for pool_name, disk_uuid in backup_pools.items(): if str(disk_uuid) in present_uuids: @@ -698,6 +747,7 @@ class Snapshot: return self._path.snapshot_name def matches_regex(self, regex: str) -> bool: + """Check if the qualified name of the snapshot matches a regular expression.""" return bool(search(regex, str(self._path))) def destroy(self) -> None: @@ -705,7 +755,7 @@ class Snapshot: @property def pool(self) -> str: - return self._path.pool + return self._path.pool_name @property def dataset(self) -> Dataset: @@ -721,30 +771,35 @@ class CommandRunner: @staticmethod def _to_list(command: str | list[str]) -> list[str]: + """If the command is a str, return it as list. Otherwise return the command.""" if isinstance(command, str): return [command] return command @classmethod def run(cls, cmdline: str | list[str]) -> None: + """Run the command, wait for it to return.""" cmdline = cls._to_list(cmdline) cls._sanity_check(cmdline) call(cmdline) @classmethod def get_output(cls, cmdline: str | list[str]) -> str: + """Run the command, return the output of the command.""" cmdline = cls._to_list(cmdline) cls._sanity_check(cmdline) return check_output(cmdline).strip().decode() @classmethod def receive_from_stream(cls, cmdline: str | list[str], stream: Popen) -> None: + """Run the command, receive data from a pipe to stdin.""" cmdline = cls._to_list(cmdline) cls._sanity_check(cmdline) check_output(cmdline, stdin=stream.stdout) @classmethod def send_stream(cls, cmdline: str | list[str]) -> Popen: + """Run the command, redirect stdout to a pipe.""" cmdline = cls._to_list(cmdline) cls._sanity_check(cmdline) return Popen(cmdline, stdout=PIPE)