major rewrite

config.py

@@ -17,11 +17,12 @@ do_not_backup = [
 ]
 
 # überprüfe ob neue Snapshots mit diesen String-Teilen vorliegen
-snapshot_tag = ["znap", "monthly"]
+snapshot_tag = "znap"
+snapshot_interval = "monthly"
 
 # wie viele Snapshots von jedem Intervall sollen beibehalten werden?
 # dies sollte der Serverkonfiguration zu zfs-auto-snapshots folgen
-keep = {
+keep_snapshots_per_interval = {
     "frequent": 4,
     "hourly": 24,
     "daily": 31,

zfs-backup.py

@@ -10,6 +10,7 @@ from functools import cache
 from pathlib import Path
 from subprocess import PIPE, Popen, check_output
 from sys import exit
+from typing import Optional
 from uuid import UUID
 
 import config
@@ -27,14 +28,96 @@ def main():
     verify_running_as_root()
     local_pool = Pool(config.local_pool_to_backup)
     external_pool = ExternalPool.find_from_dict(config.backup_pools)
-    external_pool.import_()
+    manager = Manager(local_pool, external_pool)
+    manager.backup()
+
+
+class Manager:
+    def __init__(self, local_pool: Pool, external_pool: ExternalPool):
+        self._local_pool = local_pool
+        self._external_pool = external_pool
+
+    def backup(self):
+        self._external_pool.import_()
         try:
-        external_pool.scrub()
+            self._external_pool.scrub()
-        local_pool.backup_to(external_pool)
+            self._backup_all_datasets()
-        clean_old_snapshots(external_pool)
+            self._external_pool.clean_old_snapshots()
-        external_pool.scrub()
+            self._external_pool.scrub()
         finally:
-        external_pool.export()
+            self._external_pool.export()
+
+    def send(self, local_pool: Pool, external_pool: ExternalPool):
+        pass
+
+    def _backup_all_datasets(self):
+        for local_dataset in self._local_pool.datasets_to_backup:
+            self._backup_dataset(local_dataset)
+
+    def _backup_dataset(self, local_dataset: Dataset):
+        remote_dataset = self._external_pool.search_dataset_like(local_dataset)
+        last_common_snapshot = get_last_common_snapshot(remote_dataset, local_dataset)
+        last_local_snapshot: Snapshot = local_dataset.snapshots_matching_backup_tags[-1]
+        sender = ZFS.send(last_common_snapshot, last_local_snapshot)
+        ZFS.receive(sender, self._external_pool.name)
+
+
+class ZFSPath:
+    def __init__(self, elements: list[str], snapname: Optional[str] = None):
+        self._elements = elements
+        self._snapname = snapname
+        self._is_snapshot = False
+        if snapname:
+            self._is_snapshot = True
+        self._sanity_check()
+
+    def __repr__(self):
+        path = "/".join(self._elements)
+        if self._is_snapshot:
+            return "@".join([path, self._snapname])
+        return path
+
+    def __getitem__(self, item):
+        return self._qualified_name_as_list[item]
+
+    @property
+    def _qualified_name_as_list(self) -> list[str]:
+        if self.is_snapshot:
+            return self._elements + [self._snapname]
+        return self._elements
+
+    @classmethod
+    def from_string(cls, string: str) -> ZFSPath:
+        path, snapname = string.split("@")
+        elements: list[str] = path.split("/")
+        return ZFSPath(elements, snapname)
+
+    @property
+    def snapname(self) -> str:
+        return self._snapname
+
+    @property
+    def is_snapshot(self) -> bool:
+        return self._is_snapshot
+
+    def replace_pool(self, pool: str) -> ZFSPath:
+        return ZFSPath([pool] + self._elements[1:], self.snapname)
+
+    #    @property
+    #    def pool_name(self) -> str:
+    #        return self._elements[0]
+    #
+    @property
+    def name_without_pool(self) -> str:
+        name = "/".join(self._elements[1:])
+        if self.is_snapshot:
+            name = "@".join([name, self.snapname])
+        return name
+
+    def _sanity_check(self):
+        for item in self._elements:
+            if len(item) == 0:
+                raise ValueError("Not a valid ZFSPath")
 
 
 class ZFS:
@@ -56,7 +139,7 @@ class ZFS:
         )
         # we omit the first return value, which is the pool, not the dataset
         datasets_list = datasets_str.split("\n")[1:]
-        return [Dataset(d) for d in datasets_list]
+        return [Dataset(ZFSPath.from_string(d)) for d in datasets_list]
 
     @staticmethod
     def import_pool_from_directory(name, directory) -> None:
@@ -71,7 +154,7 @@ class ZFS:
     @staticmethod
     def get_snapshots(name) -> list[Snapshot]:
         return [
-            Snapshot(name)
+            Snapshot(ZFSPath.from_string(name))
             for name in get_output_of_command(
                 [
                     config.ZFS,
@@ -169,13 +252,18 @@ def get_last_common_snapshot(ds1: Dataset, ds2: Dataset) -> Snapshot:
 
 
 class Pool:
-    def __init__(self, name):
+    def __init__(self, name: str):
-        self.name = name
+        self._name = name
+        self._path: ZFSPath = ZFSPath.from_string(name)
 
     @property
     @cache
     def datasets(self) -> list[Dataset]:
-        return ZFS.get_datasets(self.name)
+        return ZFS.get_datasets(self._name)
+
+    @property
+    def name(self) -> str:
+        return self._name
 
     @property
     @cache
@@ -184,20 +272,6 @@ class Pool:
             d for d in self.datasets if d.qualified_name not in config.do_not_backup
         ]
 
-    def backup_to(self, external_pool: ExternalPool):
-        for local_dataset in self.datasets_to_backup:
-            remote_dataset = external_pool.search_dataset(
-                local_dataset.name_without_pool
-            )
-            last_common_snapshot = get_last_common_snapshot(
-                remote_dataset, local_dataset
-            )
-            last_local_snapshot: Snapshot = (
-                local_dataset.snapshots_matching_backup_tags[-1]
-            )
-            sender = ZFS.send(last_common_snapshot, last_local_snapshot)
-            ZFS.receive(sender, external_pool.name)
-
 
 class ExternalPool(Pool):
     def __init__(self, name: str, disk: Disk):
@@ -214,13 +288,13 @@ class ExternalPool(Pool):
         self._disk.encrypt()
 
     def _import_from_directory(self, directory: Path) -> None:
-        ZFS.import_pool_from_directory(self.name, directory)
+        ZFS.import_pool_from_directory(self._name, directory)
 
     def _export_pool(self) -> None:
-        ZFS.export_pool(self.name)
+        ZFS.export_pool(self._name)
 
     def scrub(self) -> None:
-        ZFS.scrub(self.name)
+        ZFS.scrub(self._name)
 
     @classmethod
     def find_from_dict(cls, backup_pools: dict[str, UUID]) -> ExternalPool:
@@ -234,32 +308,54 @@ class ExternalPool(Pool):
             print("Could not find a backup drive")
             exit(1)
 
-    def search_dataset(self, name_without_pool: str) -> Dataset:
+    @property
-        qualified_name = self.name + "/" + name_without_pool
+    @cache
-        for dataset in self.datasets_to_backup:
+    def datasets_to_backup(self) -> list[Dataset]:
-            if qualified_name == dataset.qualified_name:
+        def replace_pool_name(old_name: str):
+            "/".join([self._name, old_name.split("/", maxsplit=1)[1]])
+
+        do_not_backup = [replace_pool_name(item) for item in config.do_not_backup]
+        return [d for d in self.datasets if d.qualified_name not in do_not_backup]
+
+    def clean_old_snapshots(self):
+        for dataset in self.datasets:
+            dataset.clean_old_snapshots()
+
+    def search_dataset_like(self, remote_dataset: Dataset) -> Dataset:
+        dataset_to_search = remote_dataset.replace_pool(self.name)
+        for dataset in self.datasets:
+            if dataset == dataset_to_search:
                 return dataset
 
 
 class Dataset:
-    def __init__(self, qualified_name: str):
+    def __init__(self, zfs_path: ZFSPath):
-        self.qualified_name = qualified_name
+        self._path = zfs_path
 
     def __repr__(self):
         return self.qualified_name
 
+    def __eq__(self, other: Dataset):
+        return str(self) == str(other)
+
+    @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.qualified_name.split("/", maxsplit=1)[1]
+        return self._path.name_without_pool
 
     @property
-    @cache
     def snapshots(self) -> list[Snapshot]:
         """get qualified snapshot names that are direct children of the dataset"""
         return ZFS.get_snapshots(self.qualified_name)
 
     @property
-    @cache
     def snapshots_matching_backup_tags(self) -> list[Snapshot]:
         snapshots = [
             snapshot for snapshot in self.snapshots if snapshot.matches_backup_tags
@@ -273,49 +369,47 @@ class Dataset:
             if snapshot.matches_backup_tags:
                 return snapshot
 
+    def replace_pool(self, name: str) -> Dataset:
+        return Dataset(self._path.replace_pool(name))
+
+    def clean_old_snapshots(self):
+        for interval, number in config.keep_snapshots_per_interval.keys():
+            self._clean_old_snapshots_for_interval(interval, number)
+
+    def _clean_old_snapshots_for_interval(self, interval: str, max_count: int):
+        regex = ".*".join(["", config.snapshot_tag, interval, ""])
+        snapshots = [snapshot for snapshot in self.snapshots if snapshot.matches_regex(regex)]
+        if len(snapshots) > max_count:
+            snapshots.sort()
+            for snap in snapshots[:-max_count]:
+                snap.destroy()
+
 
 class Snapshot:
-    def __init__(self, qualified_name: str):
+    def __init__(self, path: ZFSPath):
-        self.qualified_name = qualified_name
+        self._path = path
+
+    @classmethod
+    def from_string(cls, qualified_name: str) -> Snapshot:
+        return cls(ZFSPath.from_string(qualified_name))
 
     def __lt__(self, other: Snapshot) -> bool:
         return self.snapname < other.snapname
 
     @property
     def snapname(self) -> str:
-        return self.qualified_name.split("@")[1]
+        return self._path.snapname
 
     @property
     def matches_backup_tags(self) -> bool:
-        return bool(
+        regex = "@" + ".*".join([config.snapshot_tag, config.snapshot_interval])
-            re.search("@" + ".*".join(config.snapshot_tag), self.qualified_name)
+        return self.matches_regex(regex)
-        )
 
-    @property
+    def matches_regex(self, regex: str) -> bool:
-    def name_without_pool(self) -> str:
+        return bool(re.search(regex, str(self._path)))
-        return self.qualified_name.split("/", maxsplit=1)[1]
 
-
+    def destroy(self):
-# def clean_old_snapshots(dataset, pool):
+        ZFS.destroy(str(self))
-#    snapshots = get_snapshots(pool)  # full name like 'pool/dataset@snapshot'
-#    # get just the dataset name without the pool
-#    actual_dataset = dataset.split("/")[-1]
-#    # print('actual_dataset', actual_dataset)
-#    for interval in config.keep.keys():
-#        # print('interval', interval)
-#        # e.g. 'data@zfs-auto-snap_daily', 'ubuntu@zfs-auto-snap_hourly', ...
-#        pattern = actual_dataset + "@" + config.snapshot_tag[0] + ".*" + interval
-#        snapshots_in_interval = []
-#        for snapshot in snapshots:
-#            result = re.search(pattern, snapshot)
-#            if result is not None:
-#                snapshots_in_interval.append(snapshot)
-#        snapshots_in_interval.sort()
-#        for i in range(len(snapshots_in_interval) - config.keep[interval]):
-#            to_delete = snapshots_in_interval[i]
-#            print("  Lösche veralteten Snapshot {}".format(to_delete))
-#            zfs_destroy(to_delete, recursive=True)
-#
 
 
 def run_command(cmdline: list[str]):
@@ -334,9 +428,5 @@ def verify_running_as_root():
         exit(1)
 
 
-def clean_old_snapshots(external_pool: ExternalPool) -> None:  # TODO
-    ...
-
-
 if __name__ == "__main__":
     main()