Merge pull request 'backups: add clanCore backup & clan borgbackup module' (#605) from lassulus-backups into main

pkgs/clan-cli/clan_cli/__init__.py

@@ -6,7 +6,7 @@ from pathlib import Path
 from types import ModuleType
 from typing import Any
 
-from . import config, flakes, machines, secrets, vms, webui
+from . import backups, config, flakes, machines, secrets, vms, webui
 from .custom_logger import setup_logging
 from .dirs import get_clan_flake_toplevel, is_clan_flake
 from .ssh import cli as ssh_cli
@@ -81,6 +81,11 @@ def create_parser(prog: str | None = None) -> argparse.ArgumentParser:
 
     subparsers = parser.add_subparsers()
 
+    parser_backups = subparsers.add_parser(
+        "backups", help="manage backups of clan machines"
+    )
+    backups.register_parser(parser_backups)
+
     parser_flake = subparsers.add_parser(
         "flakes", help="create a clan flake inside the current directory"
     )

pkgs/clan-cli/clan_cli/backups/__init__.py

@@ -0,0 +1,25 @@
+# !/usr/bin/env python3
+import argparse
+
+from .create import register_create_parser
+from .list import register_list_parser
+from .restore import register_restore_parser
+
+
+# takes a (sub)parser and configures it
+def register_parser(parser: argparse.ArgumentParser) -> None:
+    subparser = parser.add_subparsers(
+        title="command",
+        description="the command to run",
+        help="the command to run",
+        required=True,
+    )
+
+    list_parser = subparser.add_parser("list", help="list backups")
+    register_list_parser(list_parser)
+
+    create_parser = subparser.add_parser("create", help="create backups")
+    register_create_parser(create_parser)
+
+    restore_parser = subparser.add_parser("restore", help="restore backups")
+    register_restore_parser(restore_parser)

pkgs/clan-cli/clan_cli/backups/create.py

@@ -0,0 +1,45 @@
+import argparse
+import json
+import subprocess
+from typing import Optional
+
+from ..errors import ClanError
+from ..machines.machines import Machine
+
+
+def create_backup(machine: Machine, provider: Optional[str] = None) -> None:
+    backup_scripts = json.loads(
+        machine.eval_nix(f"nixosConfigurations.{machine.name}.config.clanCore.backups")
+    )
+    if provider is None:
+        for provider in backup_scripts["providers"]:
+            proc = subprocess.run(
+                ["bash", "-c", backup_scripts["providers"][provider]["start"]],
+            )
+            if proc.returncode != 0:
+                raise ClanError("failed to start backup")
+            else:
+                print("successfully started backup")
+    else:
+        if provider not in backup_scripts["providers"]:
+            raise ClanError(f"provider {provider} not found")
+        proc = subprocess.run(
+            ["bash", "-c", backup_scripts["providers"][provider]["start"]],
+        )
+        if proc.returncode != 0:
+            raise ClanError("failed to start backup")
+        else:
+            print("successfully started backup")
+
+
+def create_command(args: argparse.Namespace) -> None:
+    machine = Machine(name=args.machine, flake_dir=args.flake)
+    create_backup(machine=machine, provider=args.provider)
+
+
+def register_create_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine", type=str, help="machine in the flake to create backups of"
+    )
+    parser.add_argument("--provider", type=str, help="backup provider to use")
+    parser.set_defaults(func=create_command)

pkgs/clan-cli/clan_cli/backups/list.py

@@ -0,0 +1,63 @@
+import argparse
+import pprint
+from pathlib import Path
+from typing import Optional
+
+from ..errors import ClanError
+
+
+def list_backups(
+    flake_dir: Path, machine: str, provider: Optional[str] = None
+) -> dict[str, dict[str, list[dict[str, str]]]]:
+    dummy_data = {
+        "testhostname": {
+            "borgbackup": [
+                {"date": "2021-01-01T00:00:00Z", "id": "1"},
+                {"date": "2022-01-01T00:00:00Z", "id": "2"},
+                {"date": "2023-01-01T00:00:00Z", "id": "3"},
+            ],
+            "restic": [
+                {"date": "2021-01-01T00:00:00Z", "id": "1"},
+                {"date": "2022-01-01T00:00:00Z", "id": "2"},
+                {"date": "2023-01-01T00:00:00Z", "id": "3"},
+            ],
+        },
+        "another host": {
+            "borgbackup": [
+                {"date": "2021-01-01T00:00:00Z", "id": "1"},
+                {"date": "2022-01-01T00:00:00Z", "id": "2"},
+                {"date": "2023-01-01T00:00:00Z", "id": "3"},
+            ],
+        },
+    }
+
+    if provider is not None:
+        new_data = {}
+        for machine_ in dummy_data:
+            if provider in dummy_data[machine_]:
+                new_data[machine_] = {provider: dummy_data[machine_][provider]}
+        dummy_data = new_data
+
+    if machine is None:
+        return dummy_data
+    else:
+        return {machine: dummy_data[machine]}
+
+
+def list_command(args: argparse.Namespace) -> None:
+    if args.flake is None:
+        raise ClanError("Could not find clan flake toplevel directory")
+    backups = list_backups(
+        Path(args.flake), machine=args.machine, provider=args.provider
+    )
+    if len(backups) > 0:
+        pp = pprint.PrettyPrinter(depth=4)
+        pp.pprint(backups)
+
+
+def register_list_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine", type=str, help="machine in the flake to show backups of"
+    )
+    parser.add_argument("--provider", type=str, help="backup provider to filter by")
+    parser.set_defaults(func=list_command)

pkgs/clan-cli/clan_cli/backups/restore.py

@@ -0,0 +1,42 @@
+import argparse
+from pathlib import Path
+from typing import Optional
+
+from ..errors import ClanError
+
+
+def restore_backup(
+    flake_dir: Path,
+    machine: str,
+    provider: str,
+    backup_id: str,
+    service: Optional[str] = None,
+) -> None:
+    if service is None:
+        print("would restore backup", machine, provider, backup_id)
+    else:
+        print(
+            "would restore backup", machine, provider, backup_id, "of service:", service
+        )
+
+
+def restore_command(args: argparse.Namespace) -> None:
+    if args.flake is None:
+        raise ClanError("Could not find clan flake toplevel directory")
+    restore_backup(
+        Path(args.flake),
+        machine=args.machine,
+        provider=args.provider,
+        backup_id=args.backup_id,
+        service=args.service,
+    )
+
+
+def register_restore_parser(parser: argparse.ArgumentParser) -> None:
+    parser.add_argument(
+        "machine", type=str, help="machine in the flake to create backups of"
+    )
+    parser.add_argument("provider", type=str, help="backup provider to use")
+    parser.add_argument("backup_id", type=str, help="id of the backup to restore")
+    parser.add_argument("--service", type=str, help="name of the service to restore")
+    parser.set_defaults(func=restore_command)