diff --git a/backup.py b/backup.py new file mode 100755 index 0000000..30c9e01 --- /dev/null +++ b/backup.py @@ -0,0 +1,38 @@ +#!/usr/bin/python + +# -*- coding: utf-8 -*- + +from __future__ import annotations + +from sys import exit + +import config +from zfs_backup.misc import verify_running_as_root +from zfs_backup import Manager, Pool, ExternalPool + +EXIT_ERROR = 1 + + +def main(): + verify_running_as_root() + local_pool = Pool(config.local_pool_to_backup) + external_pool = ExternalPool.find_from_dict(config.backup_pools) + manager = Manager(local_pool, external_pool) + manager.backup() + + +# TODO tests +# 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 +# TODO logging + + +if __name__ == "__main__": + try: + main() + except KeyboardInterrupt: + exit(EXIT_ERROR) + except Exception as e: + print(e) + exit(EXIT_ERROR) diff --git a/test_zfs_backup.py b/test_zfs_backup.py index f7ed358..2260a64 100644 --- a/test_zfs_backup.py +++ b/test_zfs_backup.py @@ -2,7 +2,9 @@ import pytest from unittest import mock -from zfs_backup import ZFSPath, Dataset, Snapshot +from zfs_backup.snapshot import Snapshot +from zfs_backup.dataset import Dataset +from zfs_backup.zfs_path import ZFSPath class TestZFSPath: diff --git a/zfs_backup.py b/zfs_backup.py deleted file mode 100755 index 7b729c9..0000000 --- a/zfs_backup.py +++ /dev/null @@ -1,856 +0,0 @@ -#!/usr/bin/python - -# -*- coding: utf-8 -*- - -from __future__ import annotations - -from datetime import datetime -from os import getuid -from pathlib import Path -from re import search -from subprocess import PIPE, Popen, call, check_output -from sys import exit -from typing import Any -from uuid import UUID - -import config - -EXIT_ERROR = 1 - -DISK_BY_UUID = Path("/dev/disk/by-uuid") -MAPPER_PATH = Path("/dev/mapper") - - -def main(): - _verify_running_as_root() - local_pool = Pool(config.local_pool_to_backup) - external_pool = ExternalPool.find_from_dict(config.backup_pools) - manager = Manager(local_pool, external_pool) - manager.backup() - - -# TODO tests -# 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 -# TODO logging - - -class Manager: - """Controls the overall backup process.""" - - def __init__(self, local_pool: Pool, external_pool: ExternalPool): - self._local_pool = local_pool - self._external_pool = external_pool - - def backup(self) -> None: - """Backup workflow.""" - self._external_pool.import_() - try: - self._external_pool.scrub(skip_if_recently_scrubbed=True) - self._backup_all_datasets() - self._external_pool.clean_old_snapshots() - self._external_pool.scrub(skip_if_recently_scrubbed=False) - finally: - self._external_pool.export() - - def _backup_all_datasets(self) -> None: - """Send snapshots of local datasets to the backup pool.""" - print("Backing up datasets") - for ( - local_dataset, - remote_dataset, - ) in self._get_user_filtered_dataset_backup_map().items(): - if remote_dataset is None: - stream = self._send_absolute(local_dataset) - else: - stream = self._send_incremental(local_dataset, remote_dataset) - if stream is None: - continue - ZFS.receive(stream, self._external_pool) - - def _get_user_filtered_dataset_backup_map(self) -> dict[Dataset, Dataset | None]: - """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: dict[Dataset, Dataset | None]) -> dict[Dataset, Dataset | None]: - """For datasets that exist only locally, ask the user if he wants to back it up. - - If he doesn't, remove it from the dict. - """ - return { - local_dataset: remote_dataset - for local_dataset, remote_dataset in dataset_map.items() - if remote_dataset is not None or self._ask_user_confirmation(local_dataset) - } - - def _get_raw_dataset_map(self) -> dict[Dataset, Dataset | None]: - """Return map from local to remote datasets. - - If the local dataset has no remote match, set the remote dataset to None. - """ - return { - local_dataset: self._search_matching_dataset_in_remote_pool( - local_dataset, self._external_pool - ) - for local_dataset in self._datasets_to_backup - } - - def _send_incremental( - self, local_dataset: Dataset, remote_dataset: Dataset - ) -> Popen | None: - """Send incremental stream between two datasets to remote, return the pipe. If local and remote snapshot are identical, perform nothing and return None.""" - start_snapshot = self._get_newest_common_snapshot_with_backup_tags( - local_dataset, remote_dataset - ) - end_snapshot = self._get_newest_snapshot_with_backup_tag(local_dataset) - if start_snapshot == end_snapshot: - return None - stream = ZFS.send_incremental(start_snapshot, end_snapshot) - return stream - - def _send_absolute(self, local_dataset: Dataset) -> Popen: - """Send absolute stream to remote.""" - snapshot = self._get_newest_snapshot_with_backup_tag(local_dataset) - return ZFS.send_absolute(snapshot) - - def _get_newest_snapshot_with_backup_tag(self, dataset: Dataset) -> Snapshot: - """From all snapshots in the dataset, get the newest one that matches the backup tags in the config.""" - snapshots = self._find_snapshots_with_backup_tag(dataset) - snapshots.sort() - return snapshots[-1] - - @property - def _datasets_to_backup(self) -> list[Dataset]: - """Return local datasets to consider for a backup.""" - return [ - dataset - for dataset in self._local_pool.datasets - if dataset.qualified_name not in config.do_not_backup - ] - - @staticmethod - def _find_snapshots_with_backup_tag(dataset: Dataset) -> list[Snapshot]: - """Find snapshots of a dataset that carry the backup tags.""" - regex = _get_regex_matching_snapshots_with_tags( - [config.snapshot_tag, config.snapshot_interval] - ) - return [ - snapshot for snapshot in dataset.snapshots if snapshot.matches_regex(regex) - ] - - @staticmethod - def _search_matching_dataset_in_remote_pool( - dataset: Dataset, pool: Pool - ) -> Dataset | None: - """For a dataset in the local pool, get the matching dataset in the remote pool. - - If there is no match, return None. - """ - dataset_to_search = dataset.replace_pool(pool.name) - for dataset in pool.datasets: - if dataset == dataset_to_search: - return dataset - - @classmethod - def _get_newest_common_snapshot_with_backup_tags( - cls, local_dataset: Dataset, remote_dataset: Dataset - ) -> Snapshot: - """In a local and a remote dataset, find the newest snapshot with backup tags which exists in both datasets. - - Return the local snapshot.""" - common_snapshots = [ - 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.snapshot_name == remote_snap.snapshot_name - ] - common_snapshots.sort() - return common_snapshots[-1] - - @staticmethod - def _ask_user_confirmation(dataset: Dataset) -> bool: # type: ignore (false positive) - msg = f"Dataset {dataset.qualified_name} does not exist in backup pool. Backup? [y/n] " - reply = "" - while not reply: - reply = input(msg).lower() - if reply == "y": - return True - elif reply == "n": - return False - else: - reply = "" - - -def _get_regex_matching_snapshots_with_tags(tags: list[str]): - return "@" + ".*".join(tags) - - -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 - self._is_snapshot = False - if snapshot_name: - self._is_snapshot = True - self._sanity_check(self._elements) - - def __repr__(self): - """Return the path in the same manner as `zfs`.""" - path = "/".join(self._elements) - if self._is_snapshot: - assert self._snapshot_name is not None - 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._snapshot_name == other._snapshot_name - ) - - def __hash__(self): - return str(self).__hash__() - - @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("@") - elements: list[str] = string.split("/") - return ZFSPath(elements, snapshot_name) - - @property - 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 - def is_snapshot(self) -> bool: - return self._is_snapshot - - @property - 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: - assert self.snapshot_name - name = "@".join([name, self.snapshot_name]) - return name - - @staticmethod - def _sanity_check(elements: list[str]) -> None: - for item in elements: - if len(item) == 0: - raise ValueError(Messages.NOT_VALID_ZFS_PATH) - - -def _assert_type(instance: Any, object_type: Any, message: str) -> None: - if not isinstance(instance, object_type): - raise TypeError(message) - - -class CommandInterface: - _BINARY: Path - - @classmethod - def _run_command(cls, *args: str) -> None: - CommandRunner.run([str(cls._BINARY), *args]) - - @classmethod - def _get_output(cls, *args: str) -> str: - return CommandRunner.get_output([str(cls._BINARY), *args]) - - @classmethod - def _open_stream(cls, *args: str) -> Popen: - return CommandRunner.send_stream([str(cls._BINARY), *args]) - - @classmethod - def _receive_stream(cls, *args: str, stream: Popen) -> None: - CommandRunner.receive_from_stream([str(cls._BINARY), *args], stream=stream) - - -class ZPool(CommandInterface): - _BINARY = 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: - """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) - - @classmethod - def scrub( - cls, - pool: Pool, - *, - 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: - cls._run_command(*cls._Subcommands.wait, pool.name) - if not cls._is_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: - """Scrub the pool. If""" - - def do_scrub(): - cls._run_command(*cls._Subcommands.scrub, pool.name) - - if cls._scrub_in_progress(pool): - return - if not skip_if_recently_scrubbed: - do_scrub() - else: - if not cls._recently_scrubbed(pool): - do_scrub() - - @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 - def _scrub_in_progress(cls, pool: Pool) -> bool: - return "scrub in progress" in cls._get_pool_status_output(pool) - - @classmethod - def _is_healthy(cls, pool: Pool) -> bool: - return "ONLINE" in cls._get_pool_status_output(pool) - - @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)) - - -class ZFS(CommandInterface): - - """Interface of `zfs` commands""" - - class _Subcommands: - class List: - # -H Scripting mode, omit headers - # -d1 maximum depth of 1 - # -o name output column - # -t object type - # -r recursive - _base = ["list", "-H", "-d1", "-o", "name"] - dataset = [*_base, "-t", "filesystem"] - snapshot = [*_base, "-t", "snapshot", "-r"] - - class Send: - # -R Replicate filesystem - # -I send all intermediary snapshots - absolute = ["send", "-R"] - incremental = [*absolute, "-I"] - - destroy = ["destroy"] - # -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"] - - _BINARY = Path("/usr/bin/zfs") - - @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 - datasets_list = datasets_str.split("\n")[1:] - return [Dataset(ZFSPath.from_string(d)) for d in datasets_list] - - @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") - return [Snapshot(ZFSPath.from_string(name)) for name in snapshot_names] - - @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) - cls._run_command(*cls._Subcommands.destroy, str(snapshot)) - - @classmethod - 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( - *cls._Subcommands.Send.incremental, - str(old_snapshot), - str(new_snapshot), - ) - - @classmethod - def _pre_send_sanity_checks( - cls, old_snapshot: Snapshot, new_snapshot: Snapshot - ) -> None: - for snapshot in [old_snapshot, new_snapshot]: - _assert_type( - snapshot, - Snapshot, - f"Cannot send stream, object is not a Snapshot: {snapshot}", - ) - if old_snapshot.dataset != new_snapshot.dataset: - raise ValueError(Messages.SAME_DATASET) - if old_snapshot.newer_than(new_snapshot): - raise ValueError("Old snapshot is newer than new snapshot.") - if old_snapshot == new_snapshot: - raise ValueError("Cannot send stream, snapshots are identical.") - - @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)) - - -class Time: - @staticmethod - def _get_date_format() -> str: - """date format as used by zpool status - - %a: day of week, short version - %b: month, short version - %d: day of month, zero-padded - %H, %M, %S: hour, minute, second, zero-padded - %Y: year - """ - return "%a %b %d %H:%M:%S %Y" - - @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"): - return line.strip() - else: - raise ValueError("Could not find the correct line") - - @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()) - - -class Disk: - def __init__(self, uuid: UUID, name: str): - self.uuid = uuid - self._decrypted: bool = False - self._was_already_decrypted: bool = False - 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) - self._decrypted = True - - @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: - return - Cryptsetup.encrypt(self._mapper_entry) - self._decrypted = False - - -class Cryptsetup(CommandInterface): - class _Subcommands: - close = "close" - status = "status" - open = "open" - - _BINARY = Path("/usr/bin/cryptsetup") - - @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 - def decrypt(cls, path: Path, mapper_entry: str) -> bool: - """Decrypt a disk, return whether it was already decrypted.""" - if cls._mapper_entry_in_use(mapper_entry): - if cls._already_decrypted(path, mapper_entry): - return True - print(Messages.MAPPER_ENTRY_ALREADY_EXISTS) - exit(EXIT_ERROR) - cls._run_command(cls._Subcommands.open, str(path), mapper_entry) - return False - - @classmethod - def _mapper_entry_in_use(cls, mapper_entry: str) -> bool: - return (MAPPER_PATH / mapper_entry).exists() - - @classmethod - def _get_device_by_uuid_from_status_output(cls, status: str) -> Path: - """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 - break - else: - print(f"Could not resolve UUID of cryptsetup device {device}") - exit(EXIT_ERROR) - # noinspection PyUnboundLocalVariable - return result - - @classmethod - 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: - print(Messages.CANNOT_FIND_DEVICE_PATH) - exit(EXIT_ERROR) - # noinspection PyUnboundLocalVariable - return device - - @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) - ) - return decrypted_disk_path == path - - -class Pool: - """Represents an internal pool.""" - - def __init__(self, name: str): - self._name = name - self._path: ZFSPath = ZFSPath.from_string(name) - - @property - def datasets(self) -> list[Dataset]: - return ZFS.get_first_level_datasets(self) - - @property - def name(self) -> str: - return self._name - - -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() - ZPool.import_from_directory(self, MAPPER_PATH) - 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) - ZPool.export(self) - self._disk.encrypt() - self._imported = False - print(f"Exported {self.name}. Disk {self._disk.uuid} can be removed.") - - 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: - disk = Disk(disk_uuid, pool_name) - print(f"Found disk {pool_name}, {disk_uuid}") - return cls(pool_name, disk) - else: - print(Messages.CANNOT_FIND_BACKUP_DRIVE) - exit(EXIT_ERROR) - - def clean_old_snapshots(self) -> None: - print("Cleaning old snapshots") - for dataset in self.datasets: - dataset.clean_old_snapshots() - - -class Dataset: - def __init__(self, zfs_path: ZFSPath): - self._path = zfs_path - - def __repr__(self): - return self.qualified_name - - def __eq__(self, other: Dataset): - return self._path == other._path - - def __hash__(self): - return self._path.__hash__() - - @classmethod - def from_string(cls, qualified_name: str) -> Dataset: - return cls(ZFSPath.from_string(qualified_name)) - - @property - def qualified_name(self) -> str: - return str(self._path) - - @property - def name_without_pool(self) -> str: - return self._path.name_without_pool - - @property - def snapshots(self) -> list[Snapshot]: - """get qualified snapshots that are direct children of the dataset""" - return ZFS.get_snapshots(self) - - def replace_pool(self, name: str) -> Dataset: - return Dataset(self._path.replace_pool(name)) - - def clean_old_snapshots(self) -> None: - for interval, number in config.keep_snapshots_per_interval.items(): - self._clean_old_snapshots_for_interval(interval, number) - - def _clean_old_snapshots_for_interval(self, interval: str, max_count: int) -> None: - regex = _get_regex_matching_snapshots_with_tags([config.snapshot_tag, interval]) - snapshots = [ - snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex) - ] - old_snapshots = self._get_oldest_snapshots_exceeding_max_count( - snapshots, max_count - ) - for snapshot in old_snapshots: - snapshot.destroy() - - @staticmethod - def _get_oldest_snapshots_exceeding_max_count( - snapshots: list[Snapshot], max_count: int - ) -> list[Snapshot]: - snapshots.sort() - count = len(snapshots) - max_count - if count < 0: - count = 0 - return snapshots[:count] - - -class Snapshot: - def __init__(self, path: ZFSPath): - self._path = path - - @classmethod - def from_string(cls, qualified_name: str) -> Snapshot: - return cls(ZFSPath.from_string(qualified_name)) - - def __gt__(self, other: Snapshot) -> bool: - return self.snapshot_name > other.snapshot_name - - def __repr__(self): - return str(self._path) - - def __eq__(self, other: Snapshot): - return self._path == other._path - - def newer_than(self, other: Snapshot) -> bool: - return self > other - - @property - def snapshot_name(self) -> str: - assert self._path.snapshot_name - 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: - ZFS.destroy_snapshot(self) - - @property - def pool(self) -> str: - return self._path.pool_name - - @property - def dataset(self) -> Dataset: - return Dataset(self._path.dataset) - - -class CommandRunner: - @staticmethod - def _sanity_check(cmdline: list[str]) -> None: - _assert_type(cmdline, list, "Command line must be a list") - for item in cmdline: - _assert_type(item, str, "Only strings allowed in command line") - - @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) - - -class Messages: - ALREADY_DECRYPTED = "Cannot decrypt, already decrypted" - ALREADY_IMPORTED = "Cannot import, already imported" - CANNOT_FIND_DEVICE_PATH = "Cannot find device path from output of `cryptsetup status`" - CANNOT_FIND_BACKUP_DRIVE = "Could not find a backup drive" - DELETE_IN_LOCAL_POOL = "ALERT! Tried to delete in local pool!" - MAPPER_ENTRY_ALREADY_EXISTS = "mapper entry already exists" - NOT_DECRYPTED = "Cannot encrypt, not decrypted" - NOT_IMPORTED = "Cannot export, not imported" - NOT_VALID_ZFS_PATH = "Not a valid ZFSPath" - RUN_AS_ROOT = "Run as root." - SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset" - - -def _verify_running_as_root() -> None: - if getuid() > 0: - print(Messages.RUN_AS_ROOT) - exit(EXIT_ERROR) - - -if __name__ == "__main__": - try: - main() - except KeyboardInterrupt: - exit(EXIT_ERROR) - except Exception as e: - print(e) - exit(EXIT_ERROR) diff --git a/zfs_backup/__init__.py b/zfs_backup/__init__.py new file mode 100644 index 0000000..f30e637 --- /dev/null +++ b/zfs_backup/__init__.py @@ -0,0 +1,2 @@ +from .core import Manager +from .pool import Pool, ExternalPool diff --git a/zfs_backup/commands.py b/zfs_backup/commands.py new file mode 100644 index 0000000..cd734f3 --- /dev/null +++ b/zfs_backup/commands.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from pathlib import Path +from subprocess import Popen, call, check_output, PIPE + +from misc import assert_type + + +class CommandInterface: + _BINARY: Path + + @classmethod + def _run_command(cls, *args: str) -> None: + CommandRunner.run([str(cls._BINARY), *args]) + + @classmethod + def _get_output(cls, *args: str) -> str: + return CommandRunner.get_output([str(cls._BINARY), *args]) + + @classmethod + def _open_stream(cls, *args: str) -> Popen: + return CommandRunner.send_stream([str(cls._BINARY), *args]) + + @classmethod + def _receive_stream(cls, *args: str, stream: Popen) -> None: + CommandRunner.receive_from_stream([str(cls._BINARY), *args], stream=stream) + + +class CommandRunner: + @staticmethod + def _sanity_check(cmdline: list[str]) -> None: + assert_type(cmdline, list, "Command line must be a list") + for item in cmdline: + assert_type(item, str, "Only strings allowed in command line") + + @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) diff --git a/zfs_backup/core.py b/zfs_backup/core.py new file mode 100644 index 0000000..48c72a8 --- /dev/null +++ b/zfs_backup/core.py @@ -0,0 +1,161 @@ +from __future__ import annotations + +from subprocess import Popen + +import config +from dataset import Dataset +from pool import Pool, ExternalPool +from snapshot import Snapshot +from zfs import ZFS + + +class Manager: + """Controls the overall backup process.""" + + def __init__(self, local_pool: Pool, external_pool: ExternalPool): + self._local_pool = local_pool + self._external_pool = external_pool + + def backup(self) -> None: + """Backup workflow.""" + self._external_pool.import_() + try: + self._external_pool.scrub(skip_if_recently_scrubbed=True) + self._backup_all_datasets() + self._external_pool.clean_old_snapshots() + self._external_pool.scrub(skip_if_recently_scrubbed=False) + finally: + self._external_pool.export() + + def _backup_all_datasets(self) -> None: + """Send snapshots of local datasets to the backup pool.""" + print("Backing up datasets") + for ( + local_dataset, + remote_dataset, + ) in self._get_user_filtered_dataset_backup_map().items(): + if remote_dataset is None: + stream = self._send_absolute(local_dataset) + else: + stream = self._send_incremental(local_dataset, remote_dataset) + if stream is None: + continue + ZFS.receive(stream, self._external_pool) + + def _get_user_filtered_dataset_backup_map(self) -> dict[Dataset, Dataset | None]: + """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: dict[Dataset, Dataset | None] + ) -> dict[Dataset, Dataset | None]: + """For datasets that exist only locally, ask the user if he wants to back it up. + + If he doesn't, remove it from the dict. + """ + return { + local_dataset: remote_dataset + for local_dataset, remote_dataset in dataset_map.items() + if remote_dataset is not None or self._ask_user_confirmation(local_dataset) + } + + def _get_raw_dataset_map(self) -> dict[Dataset, Dataset | None]: + """Return map from local to remote datasets. + + If the local dataset has no remote match, set the remote dataset to None. + """ + return { + local_dataset: self._search_matching_dataset_in_remote_pool( + local_dataset, self._external_pool + ) + for local_dataset in self._datasets_to_backup + } + + def _send_incremental( + self, local_dataset: Dataset, remote_dataset: Dataset + ) -> Popen | None: + """Send incremental stream between two datasets to remote, return the pipe. If local and remote snapshot are identical, perform nothing and return None.""" + start_snapshot = self._get_newest_common_snapshot_with_backup_tags( + local_dataset, remote_dataset + ) + end_snapshot = self._get_newest_snapshot_with_backup_tag(local_dataset) + if start_snapshot == end_snapshot: + return None + stream = ZFS.send_incremental(start_snapshot, end_snapshot) + return stream + + def _send_absolute(self, local_dataset: Dataset) -> Popen: + """Send absolute stream to remote.""" + snapshot = self._get_newest_snapshot_with_backup_tag(local_dataset) + return ZFS.send_absolute(snapshot) + + def _get_newest_snapshot_with_backup_tag(self, dataset: Dataset) -> Snapshot: + """From all snapshots in the dataset, get the newest one that matches the backup tags in the config.""" + snapshots = self._find_snapshots_with_backup_tag(dataset) + snapshots.sort() + return snapshots[-1] + + @property + def _datasets_to_backup(self) -> list[Dataset]: + """Return local datasets to consider for a backup.""" + return [ + dataset + for dataset in self._local_pool.datasets + if dataset.qualified_name not in config.do_not_backup + ] + + @staticmethod + def _find_snapshots_with_backup_tag(dataset: Dataset) -> list[Snapshot]: + """Find snapshots of a dataset that carry the backup tags.""" + regex = _get_regex_matching_snapshots_with_tags( + [config.snapshot_tag, config.snapshot_interval] + ) + return [ + snapshot for snapshot in dataset.snapshots if snapshot.matches_regex(regex) + ] + + @staticmethod + def _search_matching_dataset_in_remote_pool( + dataset: Dataset, pool: Pool + ) -> Dataset | None: + """For a dataset in the local pool, get the matching dataset in the remote pool. + + If there is no match, return None. + """ + dataset_to_search = dataset.replace_pool(pool.name) + for dataset in pool.datasets: + if dataset == dataset_to_search: + return dataset + + @classmethod + def _get_newest_common_snapshot_with_backup_tags( + cls, local_dataset: Dataset, remote_dataset: Dataset + ) -> Snapshot: + """In a local and a remote dataset, find the newest snapshot with backup tags which exists in both datasets. + + Return the local snapshot.""" + common_snapshots = [ + 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.snapshot_name == remote_snap.snapshot_name + ] + common_snapshots.sort() + return common_snapshots[-1] + + @staticmethod + def _ask_user_confirmation(dataset: Dataset) -> bool: # type: ignore + msg = f"Dataset {dataset.qualified_name} does not exist in backup pool. Backup? [y/n] " + reply = "" + while not reply: + reply = input(msg).lower() + if reply == "y": + return True + elif reply == "n": + return False + else: + reply = "" + + +def _get_regex_matching_snapshots_with_tags(tags: list[str]): + return "@" + ".*".join(tags) diff --git a/zfs_backup/cryptsetup.py b/zfs_backup/cryptsetup.py new file mode 100644 index 0000000..733074b --- /dev/null +++ b/zfs_backup/cryptsetup.py @@ -0,0 +1,85 @@ +from __future__ import annotations + +from pathlib import Path + +from .messages import MAPPER_ENTRY_ALREADY_EXISTS, CANNOT_FIND_DEVICE_PATH +from .commands import CommandInterface +from .misc import DISK_BY_UUID, MAPPER_PATH + + +class Cryptsetup(CommandInterface): + class _Subcommands: + close = "close" + status = "status" + open = "open" + + _BINARY = Path("/usr/bin/cryptsetup") + + @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 + def decrypt(cls, path: Path, mapper_entry: str) -> bool: + """Decrypt a disk, return whether it was already decrypted.""" + if cls._mapper_entry_in_use(mapper_entry): + if cls._already_decrypted(path, mapper_entry): + return True + print(MAPPER_ENTRY_ALREADY_EXISTS) + raise FileExistsError() + cls._run_command(cls._Subcommands.open, str(path), mapper_entry) + return False + + @classmethod + def _mapper_entry_in_use(cls, mapper_entry: str) -> bool: + return (MAPPER_PATH / mapper_entry).exists() + + @classmethod + def _get_device_by_uuid_from_status_output(cls, status: str) -> Path: + """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 + break + else: + raise FileNotFoundError( + f"Could not resolve UUID of cryptsetup device {device}" + ) + return result + + @classmethod + 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: + raise FileNotFoundError(CANNOT_FIND_DEVICE_PATH) + # noinspection PyUnboundLocalVariable + return device + + @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) + ) + return decrypted_disk_path == path diff --git a/zfs_backup/dataset.py b/zfs_backup/dataset.py new file mode 100644 index 0000000..f4084b1 --- /dev/null +++ b/zfs_backup/dataset.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +import config +from zfs import ZFS +from zfs_backup.core import _get_regex_matching_snapshots_with_tags +from snapshot import Snapshot +from zfs_path import ZFSPath + + +class Dataset: + def __init__(self, zfs_path: ZFSPath): + self._path = zfs_path + + def __repr__(self): + return self.qualified_name + + def __eq__(self, other: Dataset): + return self._path == other._path + + def __hash__(self): + return self._path.__hash__() + + @classmethod + def from_string(cls, qualified_name: str) -> Dataset: + return cls(ZFSPath.from_string(qualified_name)) + + @property + def qualified_name(self) -> str: + return str(self._path) + + @property + def name_without_pool(self) -> str: + return self._path.name_without_pool + + @property + def snapshots(self) -> list[Snapshot]: + """get qualified snapshots that are direct children of the dataset""" + return ZFS.get_snapshots(self) + + def replace_pool(self, name: str) -> Dataset: + return Dataset(self._path.replace_pool(name)) + + def clean_old_snapshots(self) -> None: + for interval, number in config.keep_snapshots_per_interval.items(): + self._clean_old_snapshots_for_interval(interval, number) + + def _clean_old_snapshots_for_interval(self, interval: str, max_count: int) -> None: + regex = _get_regex_matching_snapshots_with_tags([config.snapshot_tag, interval]) + snapshots = [ + snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex) + ] + old_snapshots = self._get_oldest_snapshots_exceeding_max_count( + snapshots, max_count + ) + for snapshot in old_snapshots: + snapshot.destroy() + + @staticmethod + def _get_oldest_snapshots_exceeding_max_count( + snapshots: list[Snapshot], max_count: int + ) -> list[Snapshot]: + snapshots.sort() + count = len(snapshots) - max_count + if count < 0: + count = 0 + return snapshots[:count] diff --git a/zfs_backup/disk.py b/zfs_backup/disk.py new file mode 100644 index 0000000..330a325 --- /dev/null +++ b/zfs_backup/disk.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from pathlib import Path +from uuid import UUID + +from .messages import ALREADY_DECRYPTED, NOT_DECRYPTED +from .cryptsetup import Cryptsetup +from .misc import DISK_BY_UUID + + +class Disk: + def __init__(self, uuid: UUID, name: str): + self.uuid = uuid + self._decrypted: bool = False + self._was_already_decrypted: bool = False + 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(ALREADY_DECRYPTED, self._name) + self._was_already_decrypted = Cryptsetup.decrypt(self._path, self._mapper_entry) + self._decrypted = True + + @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(NOT_DECRYPTED, self._name) + if self._was_already_decrypted: + return + Cryptsetup.encrypt(self._mapper_entry) + self._decrypted = False diff --git a/zfs_backup/messages.py b/zfs_backup/messages.py new file mode 100644 index 0000000..648f7bb --- /dev/null +++ b/zfs_backup/messages.py @@ -0,0 +1,13 @@ +from __future__ import annotations + + +ALREADY_DECRYPTED = "Cannot decrypt, already decrypted" +ALREADY_IMPORTED = "Cannot import, already imported" +CANNOT_FIND_DEVICE_PATH = "Cannot find device path from output of `cryptsetup status`" +CANNOT_FIND_BACKUP_DRIVE = "Could not find a backup drive" +DELETE_IN_LOCAL_POOL = "ALERT! Tried to delete in local pool!" +MAPPER_ENTRY_ALREADY_EXISTS = "mapper entry already exists" +NOT_DECRYPTED = "Cannot encrypt, not decrypted" +NOT_IMPORTED = "Cannot export, not imported" +NOT_VALID_ZFS_PATH = "Not a valid ZFSPath" +SAME_DATASET = "Cannot send incremental snapshots if start and end snapshot are not based on the same dataset" diff --git a/zfs_backup/misc.py b/zfs_backup/misc.py new file mode 100644 index 0000000..090bb67 --- /dev/null +++ b/zfs_backup/misc.py @@ -0,0 +1,19 @@ +from __future__ import annotations + +from os import getuid +from pathlib import Path + +from typing import Any + +DISK_BY_UUID = Path("/dev/disk/by-uuid") +MAPPER_PATH = Path("/dev/mapper") + + +def assert_type(instance: Any, object_type: Any, message: str) -> None: + if not isinstance(instance, object_type): + raise TypeError(message) + + +def verify_running_as_root() -> None: + if getuid() > 0: + raise PermissionError("Run as root.") diff --git a/zfs_backup/pool.py b/zfs_backup/pool.py new file mode 100644 index 0000000..fa787f4 --- /dev/null +++ b/zfs_backup/pool.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +from uuid import UUID + +from .disk import Disk +from .messages import CANNOT_FIND_BACKUP_DRIVE, ALREADY_IMPORTED, NOT_IMPORTED +from .misc import MAPPER_PATH, DISK_BY_UUID +from .zfs import ZFS +from .dataset import Dataset +from .zfs_path import ZFSPath +from .zpool import ZPool + + +class Pool: + """Represents an internal pool.""" + + def __init__(self, name: str): + self._name = name + self._path: ZFSPath = ZFSPath.from_string(name) + + @property + def datasets(self) -> list[Dataset]: + return ZFS.get_first_level_datasets(self) + + @property + def name(self) -> str: + return self._name + + +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(ALREADY_IMPORTED, self.name) + self._disk.decrypt() + ZPool.import_from_directory(self, MAPPER_PATH) + 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(NOT_IMPORTED, self.name) + ZPool.export(self) + self._disk.encrypt() + self._imported = False + print(f"Exported {self.name}. Disk {self._disk.uuid} can be removed.") + + 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: + disk = Disk(disk_uuid, pool_name) + print(f"Found disk {pool_name}, {disk_uuid}") + return cls(pool_name, disk) + else: + raise FileNotFoundError(CANNOT_FIND_BACKUP_DRIVE) + + def clean_old_snapshots(self) -> None: + print("Cleaning old snapshots") + for dataset in self.datasets: + dataset.clean_old_snapshots() diff --git a/zfs_backup/snapshot.py b/zfs_backup/snapshot.py new file mode 100644 index 0000000..797788e --- /dev/null +++ b/zfs_backup/snapshot.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from re import search + +from dataset import Dataset +from zfs import ZFS +from zfs_path import ZFSPath + + +class Snapshot: + def __init__(self, path: ZFSPath): + self._path = path + + @classmethod + def from_string(cls, qualified_name: str) -> Snapshot: + return cls(ZFSPath.from_string(qualified_name)) + + def __gt__(self, other: Snapshot) -> bool: + return self.snapshot_name > other.snapshot_name + + def __repr__(self): + return str(self._path) + + def __eq__(self, other: Snapshot): + return self._path == other._path + + def newer_than(self, other: Snapshot) -> bool: + return self > other + + @property + def snapshot_name(self) -> str: + assert self._path.snapshot_name + 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: + ZFS.destroy_snapshot(self) + + @property + def pool(self) -> str: + return self._path.pool_name + + @property + def dataset(self) -> Dataset: + return Dataset(self._path.dataset) diff --git a/zfs_backup/time.py b/zfs_backup/time.py new file mode 100644 index 0000000..27dd55a --- /dev/null +++ b/zfs_backup/time.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from datetime import datetime + + +class Time: + @staticmethod + def _get_date_format() -> str: + """date format as used by zpool status + + %a: day of week, short version + %b: month, short version + %d: day of month, zero-padded + %H, %M, %S: hour, minute, second, zero-padded + %Y: year + """ + return "%a %b %d %H:%M:%S %Y" + + @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"): + return line.strip() + else: + raise ValueError("Could not find the correct line") + + @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()) diff --git a/zfs_backup/zfs.py b/zfs_backup/zfs.py new file mode 100644 index 0000000..528ab81 --- /dev/null +++ b/zfs_backup/zfs.py @@ -0,0 +1,112 @@ +from __future__ import annotations + +from pathlib import Path +from subprocess import Popen + +import config +from .messages import DELETE_IN_LOCAL_POOL, SAME_DATASET +from .misc import assert_type +from .commands import CommandInterface +from .snapshot import Snapshot +from .dataset import Dataset +from .pool import Pool +from .zfs_path import ZFSPath + + +class ZFS(CommandInterface): + + """Interface of `zfs` commands""" + + class _Subcommands: + class List: + # -H Scripting mode, omit headers + # -d1 maximum depth of 1 + # -o name output column + # -t object type + # -r recursive + _base = ["list", "-H", "-d1", "-o", "name"] + dataset = [*_base, "-t", "filesystem"] + snapshot = [*_base, "-t", "snapshot", "-r"] + + class Send: + # -R Replicate filesystem + # -I send all intermediary snapshots + absolute = ["send", "-R"] + incremental = [*absolute, "-I"] + + destroy = ["destroy"] + # -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"] + + _BINARY = Path("/usr/bin/zfs") + + @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 + datasets_list = datasets_str.split("\n")[1:] + return [Dataset(ZFSPath.from_string(d)) for d in datasets_list] + + @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") + return [Snapshot(ZFSPath.from_string(name)) for name in snapshot_names] + + @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(DELETE_IN_LOCAL_POOL) + cls._run_command(*cls._Subcommands.destroy, str(snapshot)) + + @classmethod + 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( + *cls._Subcommands.Send.incremental, + str(old_snapshot), + str(new_snapshot), + ) + + @classmethod + def _pre_send_sanity_checks( + cls, old_snapshot: Snapshot, new_snapshot: Snapshot + ) -> None: + for snapshot in [old_snapshot, new_snapshot]: + assert_type( + snapshot, + Snapshot, + f"Cannot send stream, object is not a Snapshot: {snapshot}", + ) + if old_snapshot.dataset != new_snapshot.dataset: + raise ValueError(SAME_DATASET) + if old_snapshot.newer_than(new_snapshot): + raise ValueError("Old snapshot is newer than new snapshot.") + if old_snapshot == new_snapshot: + raise ValueError("Cannot send stream, snapshots are identical.") + + @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)) diff --git a/zfs_backup/zfs_path.py b/zfs_backup/zfs_path.py new file mode 100644 index 0000000..68c26f0 --- /dev/null +++ b/zfs_backup/zfs_path.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +from zfs_backup.messages import NOT_VALID_ZFS_PATH + + +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 + self._is_snapshot = False + if snapshot_name: + self._is_snapshot = True + self._sanity_check(self._elements) + + def __repr__(self): + """Return the path in the same manner as `zfs`.""" + path = "/".join(self._elements) + if self._is_snapshot: + assert self._snapshot_name is not None + 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._snapshot_name == other._snapshot_name + ) + + def __hash__(self): + return str(self).__hash__() + + @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("@") + elements: list[str] = string.split("/") + return ZFSPath(elements, snapshot_name) + + @property + 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 + def is_snapshot(self) -> bool: + return self._is_snapshot + + @property + 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: + assert self.snapshot_name + name = "@".join([name, self.snapshot_name]) + return name + + @staticmethod + def _sanity_check(elements: list[str]) -> None: + for item in elements: + if len(item) == 0: + raise ValueError(NOT_VALID_ZFS_PATH) diff --git a/zfs_backup/zpool.py b/zfs_backup/zpool.py new file mode 100644 index 0000000..45735e0 --- /dev/null +++ b/zfs_backup/zpool.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +from datetime import datetime, timedelta +from pathlib import Path + +from .misc import assert_type +from .commands import CommandInterface +from .pool import Pool +from .time import Time + + +class ZPool(CommandInterface): + _BINARY = 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: + """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) + + def scrub( + self, + pool: Pool, + *, + skip_if_recently_scrubbed: bool, + recent_scrub_timedelta: timedelta, + 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.") + self._scrub_if_necessary( + pool, skip_if_recently_scrubbed, recent_scrub_timedelta + ) + if wait_for_finish: + self._run_command(*self._Subcommands.wait, pool.name) + if not self._is_healthy(pool): + raise IOError(f"Pool {pool.name} is not healthy.") + + def _scrub_if_necessary( + self, + pool: Pool, + skip_if_recently_scrubbed: bool, + recent_scrub_timedelta: timedelta, + ) -> None: + """Scrub the pool. If""" + + def do_scrub(): + self._run_command(*self._Subcommands.scrub, pool.name) + + if self._scrub_in_progress(pool): + return + if not skip_if_recently_scrubbed: + do_scrub() + else: + if not self._recently_scrubbed(pool, recent_scrub_timedelta): + do_scrub() + + @classmethod + def _recently_scrubbed(cls, pool: Pool, recent_scrub_timedelta: timedelta) -> bool: + """Is the interval after the last scrub smaller than the minimum timedelta?""" + return datetime.now() - cls._last_scrub(pool) < recent_scrub_timedelta + + @classmethod + def _scrub_in_progress(cls, pool: Pool) -> bool: + return "scrub in progress" in cls._get_pool_status_output(pool) + + @classmethod + def _is_healthy(cls, pool: Pool) -> bool: + return "ONLINE" in cls._get_pool_status_output(pool) + + @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))