av
daa4379dbe
- список приложений теперь со значком статуса (забекаплено, упал дамп, бекапить нечего) и занятым местом; размеры считает dust одним вызовом - в конце уведомления — свободное место на дисках сервера, по одной строке на файловую систему
991 lines
37 KiB
Python
991 lines
37 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Backup script for all applications
|
||
Automatically discovers and runs backup scripts for all users,
|
||
then creates restic backups and sends notifications.
|
||
|
||
restic-операции разнесены на фазы с разной частотой (см. секцию [schedule] в config):
|
||
- backup, forget -- каждый прогон (forget БЕЗ --prune: только метаданные снапшотов);
|
||
- check -- структурная проверка, обычно еженедельно;
|
||
- prune -- репак/освобождение места, редко (квартально);
|
||
- verify -- check --read-data-subset, помесячно (полное покрытие за год).
|
||
Один прогон выполняет фазы строго последовательно, поэтому restic-локи между фазами
|
||
не конфликтуют. Наложение соседних прогонов предотвращается flock в cron-задаче.
|
||
|
||
Размеры приложений считает dust (ставится ролью eget в bin_prefix); если его нет
|
||
или он упал, прогон продолжается, а размеры в уведомлении просто не показываются.
|
||
"""
|
||
|
||
import argparse
|
||
import itertools
|
||
import json
|
||
import logging
|
||
import os
|
||
import pwd
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import time
|
||
import tomllib
|
||
from abc import ABC
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timedelta
|
||
from enum import Enum
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import requests
|
||
from croniter import croniter
|
||
|
||
# Default config path
|
||
CONFIG_PATH = Path("/etc/backup/config.toml")
|
||
|
||
# File name to store directories and files to back up
|
||
BACKUP_TARGETS_FILE = "backup-targets"
|
||
|
||
# Default directory fo backups (relative to app dir)
|
||
# Used when backup-targets file not exists
|
||
BACKUP_DEFAULT_DIR = "backups"
|
||
|
||
# Утилита подсчёта размеров директорий (github.com/bootandy/dust).
|
||
# Ставится ролью eget в bin_prefix, который есть в PATH cron-задачи.
|
||
DUST_BIN = "dust"
|
||
|
||
# Retention policy applied by the `forget` phase on every run.
|
||
KEEP_DAILY = "90"
|
||
KEEP_MONTHLY = "36"
|
||
|
||
# Фазы в порядке выполнения. backup и forget идут каждый прогон,
|
||
# остальные — по расписанию из config.
|
||
PHASE_BACKUP = "backup"
|
||
PHASE_FORGET = "forget"
|
||
PHASE_CHECK = "check"
|
||
PHASE_PRUNE = "prune"
|
||
PHASE_VERIFY = "verify"
|
||
|
||
ALWAYS_PHASES = [PHASE_BACKUP, PHASE_FORGET]
|
||
SCHEDULED_PHASES = [PHASE_CHECK, PHASE_PRUNE, PHASE_VERIFY]
|
||
PHASE_ORDER = ALWAYS_PHASES + SCHEDULED_PHASES
|
||
|
||
# Configure logging
|
||
logging.basicConfig(
|
||
level=logging.INFO,
|
||
format="%(asctime)s - %(levelname)s - %(message)s",
|
||
handlers=[
|
||
logging.StreamHandler(sys.stdout),
|
||
logging.FileHandler("/var/log/backup-all.log"),
|
||
],
|
||
)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
@dataclass
|
||
class Config:
|
||
host_name: str
|
||
roots: list[Path]
|
||
|
||
|
||
@dataclass
|
||
class MaintenanceOptions:
|
||
"""Параметры обслуживающих фаз (см. секцию [maintenance] в config)."""
|
||
|
||
verify_subset: str = "1/12"
|
||
prune_max_unused: str = "20%"
|
||
prune_max_repack: str = "5G"
|
||
|
||
|
||
@dataclass
|
||
class Schedule:
|
||
"""Расписание обслуживающих фаз: фаза -> cron-выражение."""
|
||
|
||
cron: dict[str, str] = field(default_factory=dict)
|
||
|
||
def due_phases(self, now: datetime) -> list[str]:
|
||
"""Фазы, которые нужно выполнить в этот прогон, в порядке PHASE_ORDER."""
|
||
phases = list(ALWAYS_PHASES)
|
||
for phase in SCHEDULED_PHASES:
|
||
expr = self.cron.get(phase)
|
||
if expr and self._due_today(expr, now):
|
||
phases.append(phase)
|
||
return phases
|
||
|
||
@staticmethod
|
||
def _due_today(expr: str, now: datetime) -> bool:
|
||
"""True, если cron-выражение срабатывает где-то в течение сегодняшних суток.
|
||
|
||
Мы не сравниваем с текущей минутой (триггер один на сутки в фиксированное
|
||
время), а проверяем, попадает ли ближайшее срабатывание выражения на сегодня.
|
||
"""
|
||
start = now.replace(hour=0, minute=0, second=0, microsecond=0)
|
||
nxt = croniter(expr, start - timedelta(minutes=1)).get_next(datetime)
|
||
return nxt.date() == now.date()
|
||
|
||
|
||
@dataclass
|
||
class Application:
|
||
path: Path
|
||
owner: str
|
||
backup_script: Path | None
|
||
backup_targets: list[Path]
|
||
|
||
|
||
class AppStatus(Enum):
|
||
"""Что случилось с приложением в этот прогон."""
|
||
|
||
DONE = "done"
|
||
FAILED = "failed"
|
||
SKIPPED = "skipped"
|
||
|
||
|
||
APP_STATUS_ICONS = {
|
||
AppStatus.DONE: "✅",
|
||
AppStatus.FAILED: "❌",
|
||
AppStatus.SKIPPED: "⏭",
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class AppRunResult:
|
||
"""Строка приложения в уведомлении: статус бекапа и занятое место."""
|
||
|
||
name: str
|
||
status: AppStatus
|
||
size: int | None = None
|
||
|
||
|
||
@dataclass
|
||
class DiskUsage:
|
||
"""Занятое и свободное место на файловой системе."""
|
||
|
||
path: Path
|
||
total: int
|
||
free: int
|
||
|
||
@property
|
||
def used(self) -> int:
|
||
return self.total - self.free
|
||
|
||
@property
|
||
def used_percent(self) -> float:
|
||
return 100.0 * self.used / self.total if self.total else 0.0
|
||
|
||
|
||
@dataclass
|
||
class BackupResult:
|
||
success: bool
|
||
error: str | None = None
|
||
|
||
|
||
@dataclass
|
||
class StorageRunResult:
|
||
name: str
|
||
success: bool
|
||
duration: float
|
||
phases: list[str]
|
||
|
||
|
||
def format_size(size: int) -> str:
|
||
"""Байты в человекочитаемый вид: 4.1 GiB, 512 MiB, 12 KiB."""
|
||
value = float(size)
|
||
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
|
||
if value < 1024 or unit == "TiB":
|
||
precision = 0 if unit == "B" or value >= 100 else 1
|
||
return f"{value:.{precision}f} {unit}"
|
||
value /= 1024
|
||
return f"{value:.1f} TiB"
|
||
|
||
|
||
def measure_app_sizes(paths: list[Path]) -> dict[str, int]:
|
||
"""Размеры директорий приложений одним вызовом dust: путь -> байты.
|
||
|
||
dust с `-o b` печатает размеры строками вида "1052672B", а при нескольких
|
||
аргументах заворачивает их в корень "(total)" — разбираем оба случая.
|
||
"""
|
||
if not paths:
|
||
return {}
|
||
|
||
cmd = [DUST_BIN, "--output-json", "--output-format", "b", "--depth", "0"]
|
||
cmd += ["--no-progress", *(str(path) for path in paths)]
|
||
try:
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
|
||
except (OSError, subprocess.TimeoutExpired) as exc:
|
||
logger.warning("Failed to run %s: %s", DUST_BIN, exc)
|
||
return {}
|
||
|
||
if result.returncode != 0:
|
||
logger.warning(
|
||
"%s exited with code %s: %s", DUST_BIN, result.returncode, result.stderr
|
||
)
|
||
return {}
|
||
|
||
try:
|
||
tree = json.loads(result.stdout)
|
||
except json.JSONDecodeError as exc:
|
||
logger.warning("Could not parse %s output: %s", DUST_BIN, exc)
|
||
return {}
|
||
|
||
sizes: dict[str, int] = {}
|
||
for node in [tree, *tree.get("children", [])]:
|
||
raw_size = str(node.get("size", "")).rstrip("B")
|
||
if not raw_size.isdigit():
|
||
continue
|
||
sizes[str(node.get("name", ""))] = int(raw_size)
|
||
return sizes
|
||
|
||
|
||
def collect_disk_usage(paths: list[Path]) -> list[DiskUsage]:
|
||
"""Занятое/свободное место по файловым системам, на которых лежат paths.
|
||
|
||
Пути с одной и той же файловой системы схлопываются: смысла показывать
|
||
/mnt/applications дважды нет.
|
||
"""
|
||
usages: list[DiskUsage] = []
|
||
seen_devices: set[int] = set()
|
||
|
||
for path in paths:
|
||
try:
|
||
device = path.stat().st_dev
|
||
if device in seen_devices:
|
||
continue
|
||
total, _used, free = shutil.disk_usage(path)
|
||
except OSError as exc:
|
||
logger.warning("Could not read disk usage for %s: %s", path, exc)
|
||
continue
|
||
seen_devices.add(device)
|
||
usages.append(DiskUsage(path=path, total=total, free=free))
|
||
|
||
return usages
|
||
|
||
|
||
def format_duration(seconds: float) -> str:
|
||
if seconds < 60:
|
||
return f"{seconds:.1f}s"
|
||
minutes = int(seconds // 60)
|
||
secs = int(seconds % 60)
|
||
if minutes < 60:
|
||
return f"{minutes}m{secs:02d}s"
|
||
hours = minutes // 60
|
||
minutes = minutes % 60
|
||
return f"{hours}h{minutes:02d}m{secs:02d}s"
|
||
|
||
|
||
class Storage(ABC):
|
||
name: str
|
||
|
||
def run(
|
||
self,
|
||
backup_dirs: list[str],
|
||
phases: list[str],
|
||
maintenance: MaintenanceOptions,
|
||
) -> BackupResult:
|
||
"""Run the requested phases against this storage."""
|
||
raise NotImplementedError()
|
||
|
||
|
||
class ResticStorage(Storage):
|
||
TYPE_NAME = "restic"
|
||
|
||
def __init__(self, name: str, params: dict[str, Any]) -> None:
|
||
self.name = name
|
||
self.restic_repository = str(params.get("restic_repository", ""))
|
||
self.restic_password = str(params.get("restic_password", ""))
|
||
|
||
env_raw = params.get("env") or {}
|
||
if not isinstance(env_raw, dict):
|
||
raise ValueError(
|
||
f"'env' must be a table for storage backend ResticStorage: '{self.name}'"
|
||
)
|
||
self.env: dict[str, str] = {str(k): str(v) for k, v in env_raw.items()}
|
||
|
||
if not self.restic_repository or not self.restic_password:
|
||
raise ValueError(
|
||
f"Missing storage configuration values for backend ResticStorage: '{self.name}'"
|
||
)
|
||
|
||
def run(
|
||
self,
|
||
backup_dirs: list[str],
|
||
phases: list[str],
|
||
maintenance: MaintenanceOptions,
|
||
) -> BackupResult:
|
||
try:
|
||
return self.__run_internal(backup_dirs, phases, maintenance)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.error("Restic process failed: %s", exc)
|
||
return BackupResult(success=False, error=str(exc))
|
||
|
||
def __build_steps(
|
||
self,
|
||
backup_dirs: list[str],
|
||
phases: list[str],
|
||
maintenance: MaintenanceOptions,
|
||
) -> list[tuple[str, list[str]]]:
|
||
"""Собрать restic-команды для запрошенных фаз в порядке PHASE_ORDER."""
|
||
steps: list[tuple[str, list[str]]] = []
|
||
|
||
for phase in PHASE_ORDER:
|
||
if phase not in phases:
|
||
continue
|
||
|
||
if phase == PHASE_BACKUP:
|
||
if not backup_dirs:
|
||
logger.warning(
|
||
"No backup directories found, skipping backup phase for '%s'",
|
||
self.name,
|
||
)
|
||
continue
|
||
steps.append(
|
||
("backup", ["restic", "backup", "--verbose"] + backup_dirs)
|
||
)
|
||
elif phase == PHASE_FORGET:
|
||
# forget БЕЗ --prune: удаляет только метаданные снапшотов, не репакует
|
||
# data-паки и не сбивает охлаждение в Intelligent Tiering.
|
||
steps.append(
|
||
(
|
||
"forget",
|
||
[
|
||
"restic",
|
||
"forget",
|
||
"--compact",
|
||
"--keep-daily",
|
||
KEEP_DAILY,
|
||
"--keep-monthly",
|
||
KEEP_MONTHLY,
|
||
],
|
||
)
|
||
)
|
||
elif phase == PHASE_CHECK:
|
||
steps.append(("check", ["restic", "check"]))
|
||
elif phase == PHASE_PRUNE:
|
||
steps.append(
|
||
(
|
||
"prune",
|
||
[
|
||
"restic",
|
||
"prune",
|
||
"--max-unused",
|
||
maintenance.prune_max_unused,
|
||
"--max-repack-size",
|
||
maintenance.prune_max_repack,
|
||
],
|
||
)
|
||
)
|
||
elif phase == PHASE_VERIFY:
|
||
steps.append(
|
||
(
|
||
"verify",
|
||
[
|
||
"restic",
|
||
"check",
|
||
f"--read-data-subset={maintenance.verify_subset}",
|
||
],
|
||
)
|
||
)
|
||
|
||
return steps
|
||
|
||
def __run_internal(
|
||
self,
|
||
backup_dirs: list[str],
|
||
phases: list[str],
|
||
maintenance: MaintenanceOptions,
|
||
) -> BackupResult:
|
||
logger.info("Starting restic run for storage '%s'", self.name)
|
||
logger.info("Destination: %s", self.restic_repository)
|
||
logger.info("Phases: %s", ", ".join(phases))
|
||
|
||
env = os.environ.copy()
|
||
env["RESTIC_REPOSITORY"] = self.restic_repository
|
||
env["RESTIC_PASSWORD"] = self.restic_password
|
||
env.update(self.env)
|
||
|
||
steps = self.__build_steps(backup_dirs, phases, maintenance)
|
||
|
||
for step, cmd in steps:
|
||
error = self.__run_step(step, cmd, env)
|
||
if error is not None:
|
||
return BackupResult(success=False, error=f"restic {step}: {error}")
|
||
|
||
return BackupResult(success=True)
|
||
|
||
def __run_step(self, step: str, cmd: list[str], env: dict[str, str]) -> str | None:
|
||
"""Run a single restic command. Return None on success or error text."""
|
||
result = subprocess.run(cmd, env=env, capture_output=True, text=True)
|
||
|
||
if result.returncode != 0:
|
||
error = result.stderr.strip() or result.stdout.strip() or "no output"
|
||
logger.error("Restic %s failed: %s", step, error)
|
||
return error
|
||
|
||
logger.info("Restic %s completed successfully", step)
|
||
return None
|
||
|
||
|
||
class Notifier(ABC):
|
||
def send(self, title: str, html_message: str) -> None:
|
||
raise NotImplementedError()
|
||
|
||
|
||
class AppriseNotifier(Notifier):
|
||
TYPE_NAME = "apprise"
|
||
|
||
def __init__(self, name: str, params: dict[str, Any]) -> None:
|
||
self.name = name
|
||
self.api_url = str(params.get("api_url", "")).rstrip("/")
|
||
self.tag = str(params.get("tag", ""))
|
||
if not self.api_url or not self.tag:
|
||
raise ValueError(
|
||
f"Missing notification configuration values for backend {name}"
|
||
)
|
||
|
||
def send(self, title: str, html_message: str) -> None:
|
||
url = f"{self.api_url}/notify/{self.tag}/"
|
||
payload = {
|
||
"title": title,
|
||
"body": html_message,
|
||
"format": "html",
|
||
}
|
||
|
||
response = requests.post(url, json=payload, timeout=30)
|
||
|
||
if response.ok:
|
||
logger.info("Apprise notification sent successfully")
|
||
else:
|
||
logger.error(
|
||
"Failed to send Apprise notification: %s - %s",
|
||
response.status_code,
|
||
response.text,
|
||
)
|
||
|
||
|
||
class ApplicationFinder:
|
||
def __init__(self, roots: list[Path]) -> None:
|
||
self.roots = roots
|
||
self.warnings: list[str] = []
|
||
|
||
def find_applications(self) -> list[Application]:
|
||
"""Discover all applications with their backup scripts and targets."""
|
||
applications: list[Application] = []
|
||
source_dirs = itertools.chain(*(root.iterdir() for root in self.roots))
|
||
|
||
for app_dir in source_dirs:
|
||
if "lost+found" in str(app_dir):
|
||
continue
|
||
if app_dir.is_dir():
|
||
try:
|
||
stat_info = app_dir.stat()
|
||
owner = pwd.getpwuid(stat_info.st_uid).pw_name
|
||
backup_script = self._find_backup_script(app_dir)
|
||
backup_targets = self._find_backup_targets(app_dir)
|
||
applications.append(
|
||
Application(
|
||
path=app_dir,
|
||
owner=owner,
|
||
backup_script=backup_script,
|
||
backup_targets=backup_targets,
|
||
)
|
||
)
|
||
except (KeyError, OSError) as e:
|
||
logger.warning("Could not get owner for %s: %s", app_dir, e)
|
||
|
||
applications.sort(key=lambda app: app.path.name)
|
||
return applications
|
||
|
||
def _find_backup_script(self, app_dir: Path) -> Path | None:
|
||
"""Find executable backup script in application directory."""
|
||
for name in ("backup.sh", "backup"):
|
||
script_path = app_dir / name
|
||
if script_path.exists():
|
||
if os.access(script_path, os.X_OK):
|
||
return script_path
|
||
logger.warning(
|
||
"Backup script %s exists but is not executable", script_path
|
||
)
|
||
return None
|
||
|
||
def _find_backup_targets(self, app_dir: Path) -> list[Path]:
|
||
"""Resolve backup target directories for an application."""
|
||
targets_file = app_dir / BACKUP_TARGETS_FILE
|
||
resolved_targets: list[Path] = []
|
||
|
||
if targets_file.exists():
|
||
for target_line in self._parse_targets_file(targets_file):
|
||
target_path = Path(target_line)
|
||
if not target_path.is_absolute():
|
||
target_path = (app_dir / target_path).resolve()
|
||
else:
|
||
target_path = target_path.resolve()
|
||
if target_path.exists():
|
||
resolved_targets.append(target_path)
|
||
else:
|
||
warning_msg = (
|
||
f"Backup target does not exist for {app_dir}: {target_path}"
|
||
)
|
||
logger.warning(warning_msg)
|
||
self.warnings.append(warning_msg)
|
||
else:
|
||
default_target = (app_dir / BACKUP_DEFAULT_DIR).resolve()
|
||
if default_target.exists():
|
||
resolved_targets.append(default_target)
|
||
else:
|
||
warning_msg = f"Default backup path does not exist for {app_dir}: {default_target}"
|
||
logger.warning(warning_msg)
|
||
self.warnings.append(warning_msg)
|
||
|
||
return resolved_targets
|
||
|
||
def _parse_targets_file(self, targets_file: Path) -> list[str]:
|
||
"""Parse backup-targets file, skipping comments and empty lines."""
|
||
targets: list[str] = []
|
||
try:
|
||
for raw_line in targets_file.read_text(encoding="utf-8").splitlines():
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
targets.append(line)
|
||
except OSError as e:
|
||
warning_msg = f"Could not read backup targets file {targets_file}: {e}"
|
||
logger.warning(warning_msg)
|
||
self.warnings.append(warning_msg)
|
||
return targets
|
||
|
||
|
||
class BackupManager:
|
||
def __init__(
|
||
self,
|
||
config: Config,
|
||
storages: list[Storage],
|
||
notifiers: list[Notifier],
|
||
schedule: Schedule,
|
||
maintenance: MaintenanceOptions,
|
||
forced_phases: list[str] | None = None,
|
||
) -> None:
|
||
self.errors: list[str] = []
|
||
self.warnings: list[str] = []
|
||
self.app_results: list[AppRunResult] = []
|
||
self.disk_usages: list[DiskUsage] = []
|
||
self.config = config
|
||
self.storages = storages
|
||
self.notifiers = notifiers
|
||
self.schedule = schedule
|
||
self.maintenance = maintenance
|
||
self.forced_phases = forced_phases
|
||
self.active_phases: list[str] = []
|
||
self.archive_duration: float = 0.0
|
||
self.storage_results: list[StorageRunResult] = []
|
||
|
||
def run_backup_process(self, applications: list[Application]) -> bool:
|
||
"""Main backup process"""
|
||
logger.info("Starting backup process")
|
||
logger.info("Found %d application directories", len(applications))
|
||
|
||
self.active_phases = self._resolve_phases()
|
||
self._run_archive_phase(applications)
|
||
backup_dirs = self._collect_backup_dirs(applications)
|
||
overall_success = self._run_storages(backup_dirs)
|
||
self._collect_usage(applications)
|
||
|
||
self._send_notification(overall_success)
|
||
|
||
logger.info("Backup process completed")
|
||
|
||
if self.errors:
|
||
logger.error("Backup completed with %d errors", len(self.errors))
|
||
return False
|
||
if self.warnings:
|
||
logger.warning("Backup completed with %d warnings", len(self.warnings))
|
||
return True
|
||
logger.info("Backup completed successfully")
|
||
return True
|
||
|
||
def _resolve_phases(self) -> list[str]:
|
||
"""Какие фазы выполняем в этот прогон: принудительно из CLI или по расписанию."""
|
||
if self.forced_phases is not None:
|
||
logger.info("Phases (forced): %s", ", ".join(self.forced_phases))
|
||
return self.forced_phases
|
||
|
||
phases = self.schedule.due_phases(datetime.now())
|
||
logger.info("Phases (scheduled): %s", ", ".join(phases))
|
||
return phases
|
||
|
||
def _run_archive_phase(self, applications: list[Application]) -> None:
|
||
"""Прогнать скрипты дампов приложений и собрать список того, что уедет в restic.
|
||
|
||
Фаза нужна только вместе с restic backup: без неё дампы делать некому и незачем.
|
||
"""
|
||
archive_start = time.monotonic()
|
||
|
||
if PHASE_BACKUP in self.active_phases:
|
||
for app in applications:
|
||
status = self._archive_app(app)
|
||
self.app_results.append(AppRunResult(name=app.path.name, status=status))
|
||
else:
|
||
logger.info("Backup phase not active, skipping per-app archive scripts")
|
||
self.app_results = [
|
||
AppRunResult(name=app.path.name, status=AppStatus.SKIPPED)
|
||
for app in applications
|
||
]
|
||
|
||
self.archive_duration = time.monotonic() - archive_start
|
||
logger.info(
|
||
"Archive phase finished in %s", format_duration(self.archive_duration)
|
||
)
|
||
|
||
def _archive_app(self, app: Application) -> AppStatus:
|
||
"""Обработать одно приложение: сделать дамп, если он предусмотрен."""
|
||
app_dir = str(app.path)
|
||
username = app.owner
|
||
|
||
if app.backup_script is None:
|
||
if app.backup_targets:
|
||
# Приложение без дампа: restic забирает его данные как есть,
|
||
# отдельный шаг архивации ему не нужен.
|
||
logger.info(
|
||
"No backup script for app: %s (user %s), "
|
||
"data directories go to restic as is",
|
||
app_dir,
|
||
username,
|
||
)
|
||
return AppStatus.DONE
|
||
|
||
warning_msg = (
|
||
f"Nothing to back up for app: {app_dir} (user {username}): "
|
||
f"no backup script and no backup targets"
|
||
)
|
||
logger.warning(warning_msg)
|
||
self.warnings.append(warning_msg)
|
||
return AppStatus.SKIPPED
|
||
|
||
logger.info("Processing backup for app: %s (user %s)", app_dir, username)
|
||
if not self._run_app_backup(str(app.backup_script), app_dir, username):
|
||
return AppStatus.FAILED
|
||
# Дамп сделан, но в restic он попадёт только если есть цели бекапа;
|
||
# об их отсутствии уже предупредил ApplicationFinder.
|
||
return AppStatus.DONE if app.backup_targets else AppStatus.SKIPPED
|
||
|
||
def _collect_usage(self, applications: list[Application]) -> None:
|
||
"""Померить размеры приложений и свободное место на их файловых системах.
|
||
|
||
Считаем после архивации, чтобы свежие дампы попали в размер, и после
|
||
restic: цифры информационные, задерживать из-за них бекап незачем.
|
||
"""
|
||
usage_start = time.monotonic()
|
||
|
||
sizes = measure_app_sizes([app.path for app in applications])
|
||
by_name = {app.path.name: str(app.path) for app in applications}
|
||
for result in self.app_results:
|
||
result.size = sizes.get(by_name.get(result.name, ""))
|
||
|
||
# Корень системы плюс диски, на которых лежат приложения: на сервере это
|
||
# разные диски, и место кончается на них независимо.
|
||
self.disk_usages = collect_disk_usage([Path("/"), *self.config.roots])
|
||
|
||
logger.info(
|
||
"Usage stats collected in %s",
|
||
format_duration(time.monotonic() - usage_start),
|
||
)
|
||
|
||
@staticmethod
|
||
def _collect_backup_dirs(applications: list[Application]) -> list[str]:
|
||
"""Собрать цели бекапа всех приложений, сохраняя порядок и убирая дубли."""
|
||
backup_dirs: list[str] = []
|
||
for app in applications:
|
||
for target in app.backup_targets:
|
||
target_str = str(target)
|
||
if target_str not in backup_dirs:
|
||
backup_dirs.append(target_str)
|
||
logger.info("Found backup directories: %s", backup_dirs)
|
||
return backup_dirs
|
||
|
||
def _run_storages(self, backup_dirs: list[str]) -> bool:
|
||
"""Прогнать активные фазы по всем хранилищам.
|
||
|
||
Хранилища независимы: падение одного не отменяет попытку для остальных.
|
||
"""
|
||
overall_success = True
|
||
|
||
for storage in self.storages:
|
||
storage_start = time.monotonic()
|
||
try:
|
||
backup_result = storage.run(
|
||
backup_dirs, self.active_phases, self.maintenance
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.error(
|
||
"Storage '%s' raised an unexpected error: %s", storage.name, exc
|
||
)
|
||
backup_result = BackupResult(success=False, error=str(exc))
|
||
storage_duration = time.monotonic() - storage_start
|
||
self.storage_results.append(
|
||
StorageRunResult(
|
||
name=storage.name,
|
||
success=backup_result.success,
|
||
duration=storage_duration,
|
||
phases=list(self.active_phases),
|
||
)
|
||
)
|
||
logger.info(
|
||
"Storage '%s' finished in %s (success=%s)",
|
||
storage.name,
|
||
format_duration(storage_duration),
|
||
backup_result.success,
|
||
)
|
||
if not backup_result.success:
|
||
error_msg = f"Storage '{storage.name}' backup failed"
|
||
if backup_result.error:
|
||
error_msg += f": {backup_result.error}"
|
||
self.errors.append(error_msg)
|
||
|
||
overall_success = overall_success and backup_result.success
|
||
|
||
return overall_success
|
||
|
||
def _run_app_backup(self, script_path: str, app_dir: str, username: str) -> bool:
|
||
"""Run backup script as the specified user"""
|
||
try:
|
||
logger.info("Running backup script %s (user %s)", script_path, username)
|
||
|
||
# Use su to run the script as the user
|
||
cmd = ["su", "--login", username, "--command", script_path]
|
||
|
||
result = subprocess.run(
|
||
cmd,
|
||
cwd=app_dir,
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=3600, # 1 hour timeout
|
||
)
|
||
|
||
if result.returncode == 0:
|
||
logger.info("Backup script for %s completed successfully", username)
|
||
return True
|
||
error_msg = f"Backup script {script_path} failed with return code {result.returncode}"
|
||
if result.stderr:
|
||
error_msg += f": {result.stderr}"
|
||
logger.error(error_msg)
|
||
self.errors.append(f"App {username}: {error_msg}")
|
||
return False
|
||
|
||
except subprocess.TimeoutExpired:
|
||
error_msg = f"Backup script {script_path} timed out"
|
||
logger.error(error_msg)
|
||
self.errors.append(f"App {username}: {error_msg}")
|
||
return False
|
||
except Exception as e:
|
||
error_msg = f"Failed to run backup script {script_path}: {str(e)}"
|
||
logger.error(error_msg)
|
||
self.errors.append(f"App {username}: {error_msg}")
|
||
return False
|
||
|
||
def _render_apps(self) -> str:
|
||
"""Список приложений: значок статуса, имя и занятое место."""
|
||
if not self.app_results:
|
||
return ""
|
||
items = ""
|
||
for result in self.app_results:
|
||
# Размер отсутствует, только если dust не отработал: тогда просто имя.
|
||
size = f" — {format_size(result.size)}" if result.size is not None else ""
|
||
items += f"<li>{APP_STATUS_ICONS[result.status]} {result.name}{size}</li>"
|
||
return f"<p>Приложения:</p><ul>{items}</ul>"
|
||
|
||
def _render_run_stats(self) -> str:
|
||
"""Фазы restic и затраченное время."""
|
||
phases_text = ", ".join(self.active_phases) if self.active_phases else "—"
|
||
block = f"<p>🔧 Фазы restic: {phases_text}</p>"
|
||
block += f"<p>⏱ Время архивации: {format_duration(self.archive_duration)}</p>"
|
||
if self.storage_results:
|
||
items = "".join(
|
||
f"<li>{'✅' if r.success else '❌'} {r.name}: {format_duration(r.duration)}</li>"
|
||
for r in self.storage_results
|
||
)
|
||
block += f"<p>⏱ Время записи в хранилища:</p><ul>{items}</ul>"
|
||
return block
|
||
|
||
def _render_disks(self) -> str:
|
||
"""Свободное место на дисках сервера."""
|
||
if not self.disk_usages:
|
||
return ""
|
||
items = "".join(
|
||
f"<li>{u.path}: свободно {format_size(u.free)} из {format_size(u.total)}"
|
||
f" (занято {u.used_percent:.0f}%)</li>"
|
||
for u in self.disk_usages
|
||
)
|
||
return f"<p>💾 Свободное место:</p><ul>{items}</ul>"
|
||
|
||
def _send_notification(self, success: bool) -> None:
|
||
"""Send notification to Notifiers"""
|
||
|
||
host = self.config.host_name
|
||
|
||
if success and not self.errors:
|
||
title = f"{host}: бекап успешно завершен"
|
||
message = f"<p><b>{host}</b>: бекап успешно завершен!</p>"
|
||
else:
|
||
title = f"{host}: бекап завершен с ошибками ({len(self.errors)})"
|
||
message = f"<p><b>{host}</b>: бекап завершен с ошибками!</p>"
|
||
|
||
message += self._render_apps()
|
||
|
||
if not (success and not self.errors):
|
||
if self.warnings:
|
||
items = "".join(f"<li>{w}</li>" for w in self.warnings)
|
||
message += f"<p>⚠️ Предупреждения:</p><ul>{items}</ul>"
|
||
|
||
if self.errors:
|
||
items = "".join(f"<li>{e}</li>" for e in self.errors)
|
||
message += f"<p>❌ Ошибки:</p><ul>{items}</ul>"
|
||
|
||
message += self._render_run_stats()
|
||
message += self._render_disks()
|
||
|
||
for notificator in self.notifiers:
|
||
try:
|
||
notificator.send(title, message)
|
||
except Exception as e:
|
||
logger.error("Failed to send notification: %s", e)
|
||
|
||
|
||
def parse_phases(raw: str) -> list[str]:
|
||
"""Разобрать CLI-список фаз, вернуть их в порядке PHASE_ORDER."""
|
||
requested = {p.strip() for p in raw.split(",") if p.strip()}
|
||
unknown = requested - set(PHASE_ORDER)
|
||
if unknown:
|
||
raise ValueError(
|
||
f"Unknown phases: {', '.join(sorted(unknown))}. "
|
||
f"Allowed: {', '.join(PHASE_ORDER)}"
|
||
)
|
||
return [p for p in PHASE_ORDER if p in requested]
|
||
|
||
|
||
def build_storages(raw_config: dict[str, Any]) -> list[Storage]:
|
||
"""Собрать хранилища из секции [storage] конфига."""
|
||
storage_raw = raw_config.get("storage") or {}
|
||
storages: list[Storage] = []
|
||
for name, params in storage_raw.items():
|
||
if not isinstance(params, dict):
|
||
raise ValueError(f"Storage config for {name} must be a table")
|
||
if params.get("type", "") == ResticStorage.TYPE_NAME:
|
||
storages.append(ResticStorage(name, params))
|
||
if not storages:
|
||
raise ValueError("At least one storage backend must be configured")
|
||
return storages
|
||
|
||
|
||
def build_notifiers(raw_config: dict[str, Any]) -> list[Notifier]:
|
||
"""Собрать нотификаторы из секции [notifier] конфига."""
|
||
notifications_raw = raw_config.get("notifier") or {}
|
||
notifiers: list[Notifier] = []
|
||
for name, params in notifications_raw.items():
|
||
if not isinstance(params, dict):
|
||
raise ValueError(f"Notificator config for {name} must be a table")
|
||
if params.get("type", "") == AppriseNotifier.TYPE_NAME:
|
||
notifiers.append(AppriseNotifier(name, params))
|
||
if not notifiers:
|
||
raise ValueError("At least one notification backend must be configured")
|
||
return notifiers
|
||
|
||
|
||
def build_schedule(raw_config: dict[str, Any]) -> Schedule:
|
||
"""Собрать расписание обслуживающих фаз из секции [schedule] конфига."""
|
||
schedule_raw = raw_config.get("schedule") or {}
|
||
if not isinstance(schedule_raw, dict):
|
||
raise ValueError("'schedule' must be a table in config.toml")
|
||
return Schedule(
|
||
cron={
|
||
phase: str(schedule_raw[phase])
|
||
for phase in SCHEDULED_PHASES
|
||
if phase in schedule_raw
|
||
}
|
||
)
|
||
|
||
|
||
def build_maintenance(raw_config: dict[str, Any]) -> MaintenanceOptions:
|
||
"""Собрать параметры обслуживания из секции [maintenance] конфига."""
|
||
maintenance_raw = raw_config.get("maintenance") or {}
|
||
if not isinstance(maintenance_raw, dict):
|
||
raise ValueError("'maintenance' must be a table in config.toml")
|
||
defaults = MaintenanceOptions()
|
||
return MaintenanceOptions(
|
||
verify_subset=str(maintenance_raw.get("verify_subset", defaults.verify_subset)),
|
||
prune_max_unused=str(
|
||
maintenance_raw.get("prune_max_unused", defaults.prune_max_unused)
|
||
),
|
||
prune_max_repack=str(
|
||
maintenance_raw.get("prune_max_repack", defaults.prune_max_repack)
|
||
),
|
||
)
|
||
|
||
|
||
def initialize(
|
||
config_path: Path,
|
||
forced_phases: list[str] | None = None,
|
||
) -> tuple[ApplicationFinder, BackupManager]:
|
||
try:
|
||
with config_path.open("rb") as config_file:
|
||
raw_config = tomllib.load(config_file)
|
||
except OSError as e:
|
||
logger.error("Failed to read config file %s: %s", config_path, e)
|
||
raise
|
||
|
||
host_name = str(raw_config.get("host_name", "unknown"))
|
||
|
||
roots_raw = raw_config.get("roots") or []
|
||
if not isinstance(roots_raw, list) or not roots_raw:
|
||
raise ValueError("roots must be a non-empty list of paths in config.toml")
|
||
roots = [Path(root) for root in roots_raw]
|
||
|
||
storages = build_storages(raw_config)
|
||
notifiers = build_notifiers(raw_config)
|
||
schedule = build_schedule(raw_config)
|
||
maintenance = build_maintenance(raw_config)
|
||
|
||
config = Config(host_name=host_name, roots=roots)
|
||
app_finder = ApplicationFinder(roots)
|
||
backup_manager = BackupManager(
|
||
config=config,
|
||
storages=storages,
|
||
notifiers=notifiers,
|
||
schedule=schedule,
|
||
maintenance=maintenance,
|
||
forced_phases=forced_phases,
|
||
)
|
||
|
||
return app_finder, backup_manager
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="Run application backups via restic")
|
||
parser.add_argument(
|
||
"--config",
|
||
type=Path,
|
||
default=CONFIG_PATH,
|
||
help=f"Path to config.toml (default: {CONFIG_PATH})",
|
||
)
|
||
parser.add_argument(
|
||
"--phases",
|
||
help=(
|
||
"Comma-separated phases to run, overriding the schedule "
|
||
f"(allowed: {', '.join(PHASE_ORDER)}). Useful for manual maintenance runs."
|
||
),
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
try:
|
||
forced_phases = parse_phases(args.phases) if args.phases else None
|
||
app_finder, backup_manager = initialize(args.config, forced_phases)
|
||
applications = app_finder.find_applications()
|
||
backup_manager.warnings.extend(app_finder.warnings)
|
||
success = backup_manager.run_backup_process(applications)
|
||
if not success:
|
||
sys.exit(1)
|
||
except KeyboardInterrupt:
|
||
logger.info("Backup process interrupted by user")
|
||
sys.exit(130)
|
||
except Exception as e:
|
||
logger.error("Unexpected error in backup process: %s", e)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|