#!/usr/bin/python3 # -*- coding: utf-8 -*- """ Модуль ZFS Backup — централизованное управление ZFS-бэкапами удалённых серверов. Интегрирован с SSH-модулем. Cron: ежедневно 20:00. Запуск: python zfs_backup.py --config=config.yaml """ import argparse import logging import os import re import sys from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Tuple import yaml # Добавляем корень проекта в path для импорта modules SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) if SCRIPT_DIR not in sys.path: sys.path.insert(0, SCRIPT_DIR) from modules.ssh_base import SSHBase # Константы DEFAULT_LOG_FILE = "/var/log/zfs_backup.log" FALLBACK_LOG_FILE = "zfs_backup.log" MAX_RETRIES = 3 SNAPSHOT_DATE_FMT = "%d-%m-%Y" # dd-mm-yyyy def setup_logging(log_file: Optional[str] = None) -> logging.Logger: """Настройка логирования в файл и консоль.""" logger = logging.getLogger("zfs_backup") logger.setLevel(logging.INFO) if logger.handlers: return logger formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) # Файл path = log_file or DEFAULT_LOG_FILE try: fh = logging.FileHandler(path, encoding="utf-8") fh.setFormatter(formatter) logger.addHandler(fh) except OSError: path = os.path.join(SCRIPT_DIR, FALLBACK_LOG_FILE) fh = logging.FileHandler(path, encoding="utf-8") fh.setFormatter(formatter) logger.addHandler(fh) logger.warning("Не удалось писать в %s, используется %s", DEFAULT_LOG_FILE, path) # Консоль ch = logging.StreamHandler(sys.stdout) ch.setFormatter(formatter) logger.addHandler(ch) return logger def load_config(config_path: str) -> Dict[str, Any]: """Загрузка конфигурации из YAML → словарь Python (формат v1.1: servers[].pools[]).""" with open(config_path, "r", encoding="utf-8") as f: config = yaml.safe_load(f) if not config or "servers" not in config: raise ValueError("В config.yaml должен быть раздел 'servers'") for i, srv in enumerate(config["servers"]): if "pools" not in srv or not srv["pools"]: raise ValueError(f"Сервер [{i}] (name={srv.get('name')}) должен содержать непустой раздел 'pools'") for j, pool in enumerate(srv["pools"]): for key in ("source_pool", "target_pool", "datasets"): if key not in pool: raise ValueError(f"Сервер [{i}] pool [{j}]: отсутствует поле '{key}'") return config def execute(ssh: SSHBase, command: str, log: logging.Logger) -> Tuple[str, str]: """ Выполнение команды по SSH. При ошибке (stderr или ненулевой код) выбрасывает исключение. Возвращает (stdout, stderr). """ stdout, stderr = ssh.cmd(command) if stderr and stderr.strip(): raise RuntimeError(f"Команда завершилась с ошибкой: {stderr.strip()}") return stdout, stderr def create_snapshot( ssh: SSHBase, full_dataset: str, date_str: str, log: logging.Logger, ) -> bool: """ Создать ZFS-снапшот. При сообщении "dataset already exists" логируем и пропускаем (успех). Возвращает True при успехе (создан или уже существует), False при ошибке (для retry). """ cmd = f"/usr/sbin/zfs snapshot {full_dataset}@{date_str}" stdout, stderr = ssh.cmd(cmd, suppress_warnings=True) stderr_lower = (stderr or "").lower() if "dataset already exists" in stderr_lower: pool_name = full_dataset.split("/")[0] snapshot_name = "/".join(full_dataset.split("/")[1:]) log.info( "В пуле %s снимок %s@%s уже существует, действие пропущено", pool_name, snapshot_name, date_str, ) return True if stderr and stderr.strip(): log.error("❌ Создание снимка failed: %s", stderr.strip()) return False log.info("✅ Снимок %s@%s создан", full_dataset, date_str) return True def replicate_snapshot( ssh: SSHBase, source_dataset: str, target_dataset: str, date_str: str, prev_date: Optional[str], log: logging.Logger, ) -> bool: """ Репликация снапшота: zfs send [ -i @prev ] source@date | zfs recv target. При ошибке "not an earlier snapshot from the same fs" логируем и пропускаем (успех). Возвращает True при успехе (репликация выполнена или снимок уже есть на target), False при ошибке (для retry). """ if prev_date: send_cmd = f"/usr/sbin/zfs send -i @{prev_date} {source_dataset}@{date_str}" else: send_cmd = f"/usr/sbin/zfs send {source_dataset}@{date_str}" full_cmd = f"{send_cmd} | /usr/sbin/zfs recv -F {target_dataset}" stdout, stderr = ssh.cmd(full_cmd, suppress_warnings=True) if stderr and "not an earlier snapshot from the same fs" in stderr: pool_name = target_dataset.split("/")[0] dataset_name = "/".join(target_dataset.split("/")[1:]) log.info( "В пуле %s снимок %s@%s уже существует", pool_name, dataset_name, date_str, ) return True if stderr and stderr.strip(): log.error("❌ Репликация failed: %s", stderr.strip()) return False log.info("✅ Репликация %s → %s", source_dataset, target_dataset) return True def get_previous_snapshot_date( ssh: SSHBase, target_dataset: str, log: logging.Logger, ) -> Optional[str]: """ Получить дату последнего снапшота для target_dataset (формат dd-mm-yyyy). Если снапшотов нет — возвращает None. """ cmd = f"/usr/sbin/zfs list -t snapshot -H -o name {target_dataset} 2>/dev/null || true" stdout, stderr = ssh.cmd(cmd) lines = [s.strip() for s in stdout.splitlines() if s.strip()] pattern = re.compile(r"@(\d{2}-\d{2}-\d{4})$") dates: List[str] = [] for line in lines: m = pattern.search(line) if m: dates.append(m.group(1)) if not dates: return None def parse_d(s: str) -> datetime: return datetime.strptime(s, "%d-%m-%Y") dates.sort(key=parse_d) return dates[-1] def run_with_retry( log: logging.Logger, server_name: str, operation_name: str, func, *args, **kwargs, ) -> Any: """Выполнить операцию до MAX_RETRIES раз при ошибке.""" last_exc = None for attempt in range(1, MAX_RETRIES + 1): try: return func(*args, **kwargs) except Exception as e: last_exc = e log.warning( "❌ %s: failed %s (retry %s/%s): %s", server_name, operation_name, attempt, MAX_RETRIES, e, ) if attempt == MAX_RETRIES: raise raise last_exc def backup_server( server: Dict[str, Any], ssh_defaults: Dict[str, Any], log: logging.Logger, ) -> bool: """ Выполнить бэкап для одного сервера: снапшоты по пулам/datasets, send/recv, очистка. Конфиг v1.1: server['pools'] — список пулов с source_pool, datasets, target_pool. """ name = server["name"] pools = server["pools"] retention_days = int(server.get("retention_days", 30)) port = ssh_defaults.get("port", 22) username = ssh_defaults.get("username", "root") pkey_file = ssh_defaults.get("pkey_file", "/root/.ssh/id_rsa") host_keys = ssh_defaults.get("host_keys", "~/.ssh/known_hosts") date_str = datetime.now().strftime(SNAPSHOT_DATE_FMT) log.info("Сервер %s: дата снапшотов %s", name, date_str) ssh = SSHBase(hostname=name, port=port, username=username, pkey_file=pkey_file, host_keys=host_keys) ssh.connect() pool_counts: Dict[str, int] = {} total_datasets = 0 try: # 1. Фаза снапшотов for pool_config in pools: source_pool = pool_config["source_pool"] datasets = pool_config["datasets"] target_pool = pool_config["target_pool"] pool_counts[source_pool] = len(datasets) total_datasets += len(datasets) for dataset in datasets: full_dataset = f"{source_pool}/{dataset}" def do_snapshot(fd=full_dataset, d=date_str): if not create_snapshot(ssh, fd, d, log): raise RuntimeError(f"Snapshot {fd}@{d} failed") run_with_retry(log, name, f"snapshot {full_dataset}", do_snapshot) # 2. Фаза send/recv for pool_config in pools: source_pool = pool_config["source_pool"] datasets = pool_config["datasets"] target_pool = pool_config["target_pool"] for dataset in datasets: full_dataset = f"{source_pool}/{dataset}" target_dataset = f"{target_pool}/{dataset}" def do_send_recv(fd=full_dataset, td=target_dataset, d=date_str): prev_date = get_previous_snapshot_date(ssh, td, log) if prev_date: log.info("Инкрементальная передача %s (от %s)", fd, prev_date) else: log.info("Полная передача %s", fd) if not replicate_snapshot(ssh, fd, td, d, prev_date, log): raise RuntimeError(f"Replicate {fd}@{d} → {td} failed") run_with_retry(log, name, f"send/recv {full_dataset}", do_send_recv) # 3. Cleanup cutoff = datetime.now() - timedelta(days=retention_days) for pool_config in pools: source_pool = pool_config["source_pool"] datasets = pool_config["datasets"] target_pool = pool_config["target_pool"] for dataset in datasets: for pool, ds in [(source_pool, dataset), (target_pool, dataset)]: full_ds = f"{pool}/{ds}" cmd = f"/usr/sbin/zfs list -t snapshot -H -o name {full_ds} 2>/dev/null || true" stdout, _ = ssh.cmd(cmd) pattern = re.compile(r"@(\d{2}-\d{2}-\d{4})$") for line in stdout.splitlines(): line = line.strip() if not line: continue m = pattern.search(line) if not m: continue try: snap_date = datetime.strptime(m.group(1), "%d-%m-%Y") if snap_date < cutoff: destroy_cmd = f"/usr/sbin/zfs destroy {line}" execute(ssh, destroy_cmd, log) log.info("Удалён старый снапшот: %s", line) except ValueError: continue parts = ", ".join(f"{p}:{c}" for p, c in sorted(pool_counts.items())) log.info("✅ %s: %s datasets backed up (%s)", name, total_datasets, parts) return True finally: ssh.close() def main() -> int: parser = argparse.ArgumentParser(description="ZFS Backup — бэкап ZFS датасетов по config.yaml") parser.add_argument("--config", default="config.yaml", help="Путь к config.yaml") parser.add_argument("--log-file", default=None, help="Путь к лог-файлу (по умолчанию /var/log/zfs_backup.log)") args = parser.parse_args() log = setup_logging(args.log_file) config_path = args.config if not os.path.isabs(config_path): config_path = os.path.join(SCRIPT_DIR, config_path) if not os.path.isfile(config_path): log.error("Файл конфигурации не найден: %s", config_path) return 1 try: config = load_config(config_path) except Exception as e: log.exception("Ошибка загрузки config.yaml: %s", e) return 1 ssh_defaults = config.get("ssh_defaults") or {} servers = config["servers"] for server in servers: try: backup_server(server, ssh_defaults, log) except Exception as e: log.exception("❌ %s: бэкап завершился ошибкой после %s попыток: %s", server["name"], MAX_RETRIES, e) return 1 return 0 if __name__ == "__main__": sys.exit(main())