diff --git a/tests/conftest.py b/tests/conftest.py index 0491550..99dde35 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -35,12 +35,37 @@ DUMPXML = """ """ +FIXTURES = ROOT / "tests" / "fixtures" + +# Снимки web01: имя → (состояние, описание); список берётся из fixtures/snapshot-list.txt. +SNAPSHOTS = {"before-upd": ("shutoff", "Перед обновлением ядра"), "clean": ("running", "Чистая система после установки")} + + def fake_result(stdout: str = "", ok: bool = True, stderr: str = "") -> dict: return {"ok": ok, "code": 0 if ok else 1, "stdout": stdout, "stderr": stderr, "cmd": ""} +def option(cmd, name): + """Значение опции virsh вида --name VALUE (или '').""" + return cmd[cmd.index(name) + 1] if name in cmd and cmd.index(name) + 1 < len(cmd) else "" + + +def fake_snapshot_xml(snap: str) -> str: + state, description = SNAPSHOTS.get(snap, ("shutoff", "")) + memory = "" if state == "running" else "" + return f"{snap}{description}{state}{memory}1758794400" + + def fake_run_cmd(cmd, timeout=12, **_kwargs): joined = " ".join(cmd) + if cmd[:2] == ["virsh", "snapshot-list"]: + return fake_result((FIXTURES / "snapshot-list.txt").read_text()) + if cmd[:2] == ["virsh", "snapshot-current"]: + return fake_result("clean") + if cmd[:2] == ["virsh", "snapshot-dumpxml"]: + return fake_result(fake_snapshot_xml(option(cmd, "--snapshotname"))) + if cmd[:2] in (["virsh", "snapshot-create-as"], ["virsh", "snapshot-revert"], ["virsh", "snapshot-delete"]): + return fake_result("Domain snapshot %s created" % option(cmd, "--name")) if cmd[:2] == ["virsh", "list"]: return fake_result(" Id Name State\n-----------------------\n 1 web01 running\n - db01 shut off") if cmd[:2] == ["virsh", "dominfo"]: @@ -72,9 +97,15 @@ def fake_run_cmd(cmd, timeout=12, **_kwargs): return fake_result() +def feature_modules(): + """Загруженные модули web/features/* — им тоже подменяем run_cmd.""" + return [module for name, module in sys.modules.items() if name.startswith("features.") and hasattr(module, "run_cmd")] + + @pytest.fixture() def data_dirs(tmp_path, monkeypatch): import app + import core import host_profile import network_core import update_core @@ -95,6 +126,7 @@ def data_dirs(tmp_path, monkeypatch): monkeypatch.setattr(app, "IMAGES_DIR", dirs["images"]) monkeypatch.setattr(app, "DISK_IMAGES_DIR", dirs["disk_images"]) monkeypatch.setattr(app, "OPERATIONS_DIR", dirs["operations"]) + monkeypatch.setattr(core, "OPERATIONS_DIR", dirs["operations"]) monkeypatch.setattr(network_core, "CONFIG_DIR", dirs["config"]) monkeypatch.setattr(network_core, "NETWORK_DIR", dirs["network"]) monkeypatch.setattr(network_core, "NFT_DIR", dirs["nft"]) @@ -108,7 +140,7 @@ def data_dirs(tmp_path, monkeypatch): monkeypatch.setattr(update_core, "STATE_FILE", dirs["update"] / "state.json") monkeypatch.setattr(update_core, "LOG_FILE", dirs["update"] / "update.log") monkeypatch.setattr(update_core, "SOURCE_DIR", ROOT) - for module in (app, network_core, host_profile): + for module in (app, network_core, host_profile, core, *feature_modules()): monkeypatch.setattr(module, "run_cmd", fake_run_cmd) return dirs diff --git a/tests/dev_server.py b/tests/dev_server.py index 94dc8d8..4d21b25 100644 --- a/tests/dev_server.py +++ b/tests/dev_server.py @@ -14,16 +14,44 @@ sys.path.insert(0, str(Path(__file__).resolve().parent)) import conftest # noqa: E402 (sets sys.path and env) import app # noqa: E402 +import core # noqa: E402 import host_profile # noqa: E402 import network_core # noqa: E402 import update_core # noqa: E402 +# Снимки для демо: имя, дата, состояние, описание. +DEMO_SNAPSHOTS = [ + ("snap-20260925-1012", "2026-09-25 10:12:03 +0000", "shutoff", "Перед обновлением ядра"), + ("snap-20260924-0900", "2026-09-24 09:00:41 +0000", "running", "Настроен nginx и сертификаты"), + ("snap-20260918-1830", "2026-09-18 18:30:12 +0000", "running", "Чистая система после установки"), + ("clean", "2026-09-10 12:05:00 +0000", "shutoff", ""), +] + + +def demo_run_cmd(cmd, timeout=12, **kwargs): + """Как conftest.fake_run_cmd, но со списком снимков побогаче.""" + if cmd[:2] == ["virsh", "snapshot-list"]: + rows = "\n".join(f" {name:<20} {when} {state}" for name, when, state, _ in DEMO_SNAPSHOTS) + return conftest.fake_result(" Name Creation Time State\n" + "-" * 62 + "\n" + rows) + if cmd[:2] == ["virsh", "snapshot-current"]: + return conftest.fake_result("snap-20260924-0900") + if cmd[:2] == ["virsh", "snapshot-create-as"] and "uefi" in conftest.option(cmd, "--description").lower(): + # Описание со словом «uefi» показывает, как выглядит ошибка libvirt. + return conftest.fake_result(ok=False, stderr="error: unsupported configuration: internal snapshots of a VM with pflash based firmware are not supported") + if cmd[:2] == ["virsh", "snapshot-dumpxml"]: + snap = conftest.option(cmd, "--snapshotname") + state, description = next(((s, d) for n, _, s, d in DEMO_SNAPSHOTS if n == snap), ("shutoff", "")) + memory = "internal" if state == "running" else "no" + return conftest.fake_result(f"{snap}{description}{state}") + return conftest.fake_run_cmd(cmd, timeout, **kwargs) + def seed(root: Path) -> None: dirs = {name: root / name for name in ("iso", "images", "disk-images", "operations", "network", "config", "nft", "update")} for path in dirs.values(): path.mkdir(parents=True, exist_ok=True) app.ISO_DIR, app.IMAGES_DIR, app.DISK_IMAGES_DIR, app.OPERATIONS_DIR = dirs["iso"], dirs["images"], dirs["disk-images"], dirs["operations"] + core.OPERATIONS_DIR = dirs["operations"] network_core.CONFIG_DIR, network_core.NETWORK_DIR, network_core.NFT_DIR = dirs["config"], dirs["network"], dirs["nft"] network_core.PORT_FORWARDS_FILE = dirs["network"] / "port_forwards.json" network_core.NAT_XML_FILE = dirs["network"] / "virtuality-nat.xml" @@ -33,8 +61,8 @@ def seed(root: Path) -> None: host_profile.PROFILE_FILE = dirs["config"] / "host_profile.json" update_core.STATE_DIR, update_core.STATE_FILE, update_core.LOG_FILE = dirs["update"], dirs["update"] / "state.json", dirs["update"] / "update.log" update_core.SOURCE_DIR = Path(__file__).resolve().parents[1] - for module in (app, network_core, host_profile): - module.run_cmd = conftest.fake_run_cmd + for module in (app, network_core, host_profile, core, *conftest.feature_modules()): + module.run_cmd = demo_run_cmd app.auth.verify_password = lambda user, password: True import nodectl @@ -73,6 +101,9 @@ def seed(root: Path) -> None: op_id = str(uuid.uuid4()) app.write_operation({"id": op_id, "type": "vm_create", "title": title, "status": status, "progress": progress, "message": "Готово" if status == "success" else "qemu-img: 64%" if status == "running" else "virt-install завершился с ошибкой: 1", "created_at": app.utc_now(), "updated_at": app.utc_now(), "created_by": "tester"}) app.append_operation_log(op_id, "virt-install --name web01 --memory 2048 ...") + # Идущий снимок машины db01 — видно на /vm/db01/snapshots. + snap_op = core.new_operation("snapshot_create", "Снимок машины db01", vm_name="db01", snapshot="snap-20260925-1500", vm_running=True) + core.update_operation(snap_op, status="running", progress=15, message="Сохраняем состояние машины и памяти — она на несколько секунд замрёт…") if __name__ == "__main__": diff --git a/tests/test_snapshots.py b/tests/test_snapshots.py new file mode 100644 index 0000000..c60632c --- /dev/null +++ b/tests/test_snapshots.py @@ -0,0 +1,285 @@ +"""Снимки (точки восстановления) машины: парсеры и маршруты /vm/{name}/snapshots.""" +from datetime import datetime +from pathlib import Path +from urllib.parse import unquote + +import pytest + +import conftest +import core +from features import snapshots + +FIXTURES = Path(__file__).resolve().parent / "fixtures" + + +# ---------------------------------------------------------------- parsers +def test_parse_snapshot_list_fixture(): + rows = snapshots.parse_snapshot_list((FIXTURES / "snapshot-list.txt").read_text()) + assert [row["name"] for row in rows] == ["before-upd", "clean"] # новые сверху + assert rows[0] == {"name": "before-upd", "created_at": "2026-09-25 10:12:03", "tz": "+0000", "state": "shutoff"} + assert rows[1]["state"] == "running" + + +def test_parse_snapshot_list_without_timezone_and_garbage(): + text = " Name Creation Time State\n------\n snap1 2026-01-02 03:04:05 running\nerror: something\n\n" + rows = snapshots.parse_snapshot_list(text) + assert rows == [{"name": "snap1", "created_at": "2026-01-02 03:04:05", "tz": "", "state": "running"}] + assert snapshots.parse_snapshot_list("") == [] + + +def test_parse_snapshot_xml(): + meta = snapshots.parse_snapshot_xml(conftest.fake_snapshot_xml("clean")) + assert meta["description"] == "Чистая система после установки" + assert meta["state"] == "running" and meta["with_memory"] is True + assert snapshots.parse_snapshot_xml(conftest.fake_snapshot_xml("before-upd"))["with_memory"] is False + assert snapshots.parse_snapshot_xml("x + hvm/usr/share/OVMF/OVMF_CODE.fd/var/lib/libvirt/qemu/nvram/x.fd + + + + """ + info = snapshots.inspect_domain_xml(xml) + assert info["unsupported"] == [{"target": "vda", "format": "raw"}] + assert info["uefi"] is True + assert snapshots.inspect_domain_xml("") + return original(cmd, timeout, **kwargs) + + monkeypatch.setattr(snapshots, "run_cmd", run) + response = logged_in.post("/vm/web01/snapshots/create", data={"description": "x"}, follow_redirects=False) + assert "qcow2" in unquote(response.headers["location"]) + assert not find(recorder, "virsh", "snapshot-create-as") + + +def test_create_snapshot_surfaces_libvirt_error(logged_in, data_dirs, recorder, monkeypatch): + original = snapshots.run_cmd + + def run(cmd, timeout=12, **kwargs): + if cmd[:2] == ["virsh", "snapshot-create-as"]: + return conftest.fake_result(ok=False, stderr="error: unsupported configuration: internal snapshots of a VM with pflash based firmware are not supported") + return original(cmd, timeout, **kwargs) + + monkeypatch.setattr(snapshots, "run_cmd", run) + response = logged_in.post("/vm/web01/snapshots/create", data={}, follow_redirects=False) + location = response.headers["location"] + operation = core.read_operation(location.split("op=", 1)[1]) + assert operation["status"] == "error" + assert operation["message"].startswith("Для машин с UEFI-прошивкой") and "pflash" in operation["message"] + assert "pflash" in operation["log_tail"] + html = logged_in.get(location).text + assert "Для машин с UEFI-прошивкой" in html + + +def test_create_snapshot_waits_for_active_operation(logged_in, data_dirs, recorder): + core.new_operation("snapshot_create", "Снимок машины web01", vm_name="web01", snapshot="snap-x") + response = logged_in.post("/vm/web01/snapshots/create", data={}, follow_redirects=False) + assert "предыдущей операции" in unquote(response.headers["location"]) + assert not find(recorder, "virsh", "snapshot-create-as") + html = logged_in.get("/vm/web01/snapshots").text + assert 'data-watch-operation=' in html and "/api/operations/" in html + + +def test_revert_snapshot(logged_in, data_dirs, recorder): + response = logged_in.post("/vm/web01/snapshots/clean/revert", follow_redirects=False) + assert response.status_code == 303 + assert find(recorder, "virsh", "snapshot-revert") == [["virsh", "snapshot-revert", "--domain", "web01", "--snapshotname", "clean"]] + operation = core.read_operation(response.headers["location"].split("op=", 1)[1]) + assert operation["status"] == "success" and "возвращена" in operation["message"] and "работает" in operation["message"] + + +def test_revert_retries_with_force(logged_in, data_dirs, recorder, monkeypatch): + original = snapshots.run_cmd + + def run(cmd, timeout=12, **kwargs): + if cmd[:2] == ["virsh", "snapshot-revert"] and "--force" not in cmd: + recorder.append(list(cmd)) + return conftest.fake_result(ok=False, stderr="error: revert requires force: snapshot was created with a different configuration") + return original(cmd, timeout, **kwargs) + + monkeypatch.setattr(snapshots, "run_cmd", run) + response = logged_in.post("/vm/web01/snapshots/clean/revert", follow_redirects=False) + reverts = find(recorder, "virsh", "snapshot-revert") + assert len(reverts) == 2 and reverts[1][-1] == "--force" + assert core.read_operation(response.headers["location"].split("op=", 1)[1])["status"] == "success" + + +def test_revert_reports_error(logged_in, data_dirs, recorder, monkeypatch): + original = snapshots.run_cmd + + def run(cmd, timeout=12, **kwargs): + if cmd[:2] == ["virsh", "snapshot-revert"]: + return conftest.fake_result(ok=False, stderr="error: Domain snapshot not found: no domain snapshot with matching name 'clean'") + return original(cmd, timeout, **kwargs) + + monkeypatch.setattr(snapshots, "run_cmd", run) + response = logged_in.post("/vm/web01/snapshots/clean/revert", follow_redirects=False) + operation = core.read_operation(response.headers["location"].split("op=", 1)[1]) + assert operation["status"] == "error" and operation["message"].startswith("Такого снимка уже нет") + + +def test_delete_snapshot(logged_in, data_dirs, recorder): + response = logged_in.post("/vm/web01/snapshots/before-upd/delete", follow_redirects=False) + assert response.status_code == 303 + assert "удалён" in unquote(response.headers["location"]) + assert find(recorder, "virsh", "snapshot-delete") == [["virsh", "snapshot-delete", "--domain", "web01", "--snapshotname", "before-upd"]] + + +def test_delete_snapshot_error(logged_in, data_dirs, recorder, monkeypatch): + original = snapshots.run_cmd + + def run(cmd, timeout=12, **kwargs): + if cmd[:2] == ["virsh", "snapshot-delete"]: + return conftest.fake_result(ok=False, stderr="error: Timed out during operation: cannot acquire state change lock") + return original(cmd, timeout, **kwargs) + + monkeypatch.setattr(snapshots, "run_cmd", run) + response = logged_in.post("/vm/web01/snapshots/before-upd/delete", follow_redirects=False) + assert "snapshot_error=" in response.headers["location"] and "занята" in unquote(response.headers["location"]) + + +def test_snapshot_routes_validate_names(logged_in, data_dirs, recorder): + assert logged_in.post("/vm/web01/snapshots/-bad/revert", follow_redirects=False).status_code == 400 + assert logged_in.post("/vm/web01/snapshots/..%2Fx/delete", follow_redirects=False).status_code in (400, 404) + assert logged_in.post("/vm/-bad/snapshots/create", data={}, follow_redirects=False).status_code == 400 + assert not find(recorder, "virsh", "snapshot-delete") and not find(recorder, "virsh", "snapshot-revert") + + +def test_snapshot_routes_require_login(client): + for path in ("/vm/web01/snapshots/create", "/vm/web01/snapshots/clean/revert", "/vm/web01/snapshots/clean/delete"): + response = client.post(path, follow_redirects=False) + assert response.status_code == 303 and response.headers["location"] == "/login" diff --git a/web/app.py b/web/app.py index bf4eda1..84bdbbf 100644 --- a/web/app.py +++ b/web/app.py @@ -28,7 +28,7 @@ import auth import host_profile import network_core import presenters -from features import settings, setup +from features import settings, setup, snapshots import update_core from network_core import NetworkError @@ -2023,3 +2023,4 @@ def api_health(request: Request): app.include_router(setup.router) app.include_router(settings.router) +app.include_router(snapshots.router) diff --git a/web/features/snapshots.py b/web/features/snapshots.py new file mode 100644 index 0000000..a4ed371 --- /dev/null +++ b/web/features/snapshots.py @@ -0,0 +1,344 @@ +"""«Снимки» (точки восстановления) виртуальной машины — внутренние снимки +libvirt через `virsh snapshot-*`. + +Снимок хранится внутри qcow2-диска машины. У работающей машины в снимок +попадает и память (машина на несколько секунд замирает), у выключенной — +только диск. Создание и возврат выполняются как фоновые операции. +""" +import re +import xml.etree.ElementTree as ET +from datetime import datetime, timedelta +from typing import Any + +from fastapi import APIRouter, Form, Request +from fastapi.responses import JSONResponse, RedirectResponse + +import presenters +from core import ( + append_operation_log, + cmd_error, + finish_operation, + list_operations, + new_operation, + read_operation, + redirect_with_message, + render, + require_auth, + run_cmd, + run_operation, + update_operation, + valid_label, + valid_vm_name, +) + +router = APIRouter() + +MAX_SNAPSHOTS = 8 +DESCRIPTION_MAX = 200 +SNAPSHOT_TIMEOUT = 1800 # секунд: снимок памяти большой машины идёт минуты +OPERATION_TYPES = ("snapshot_create", "snapshot_revert") + +SNAPSHOT_STATES = { + "running": ("Работала", "success"), + "paused": ("Была на паузе", "warning"), + "shutoff": ("Была выключена", "neutral"), + "disk-snapshot": ("Только диск", "info"), +} + +MONTHS = ["января", "февраля", "марта", "апреля", "мая", "июня", "июля", "августа", "сентября", "октября", "ноября", "декабря"] + +# Известные ошибки libvirt → понятное объяснение (подстрока ищется без учёта регистра). +KNOWN_ERRORS = [ + ("pflash", "Для машин с UEFI-прошивкой эта версия libvirt не умеет делать внутренние снимки."), + ("nvram", "Для машин с UEFI-прошивкой эта версия libvirt не умеет делать внутренние снимки."), + ("unsupported for storage type", "Внутренние снимки работают только с дисками qcow2."), + ("does not support snapshot", "Внутренние снимки работают только с дисками qcow2."), + ("internal snapshot for disk", "Внутренние снимки работают только с дисками qcow2."), + ("cannot acquire state change lock", "Машина занята другой операцией. Подождите минуту и попробуйте снова."), + ("timed out during operation", "Машина занята другой операцией. Подождите минуту и попробуйте снова."), + ("snapshot not found", "Такого снимка уже нет — возможно, его удалили."), + ("already exists", "Снимок с таким именем уже есть."), + ("no space left", "На диске сервера закончилось место."), + ("is not running", "Машина выключена, поэтому это действие сейчас недоступно."), +] + +_ROW_RE = re.compile(r"^\s*(\S+)\s+(\d{4}-\d{2}-\d{2})\s+(\d{2}:\d{2}:\d{2})(?:\s+([+-]\d{4}))?\s+(\S+)\s*$") + + +# ---------------------------------------------------------------- parsers +def parse_snapshot_list(text: str) -> list[dict[str, Any]]: + """Таблица `virsh snapshot-list` → список снимков, новые сверху.""" + rows = [] + for line in (text or "").splitlines(): + match = _ROW_RE.match(line) + if not match: + continue + name, date, time, tz, state = match.groups() + rows.append({"name": name, "created_at": f"{date} {time}", "tz": tz or "", "state": state}) + return sorted(rows, key=lambda row: row["created_at"], reverse=True) + + +def parse_snapshot_xml(text: str) -> dict[str, Any]: + """`virsh snapshot-dumpxml` → описание, состояние, есть ли память.""" + try: + root = ET.fromstring(text or "") + except ET.ParseError: + return {} + memory = root.find("memory") + parent = root.find("parent/name") + return { + "description": (root.findtext("description") or "").strip(), + "state": (root.findtext("state") or "").strip(), + "with_memory": memory is not None and memory.get("snapshot", "no") not in ("no", ""), + "parent": (parent.text or "").strip() if parent is not None else "", + } + + +def inspect_domain_xml(text: str) -> dict[str, Any]: + """`virsh dumpxml` → форматы дисков и признак UEFI (для проверок перед снимком).""" + info: dict[str, Any] = {"disks": [], "unsupported": [], "uefi": False} + try: + root = ET.fromstring(text or "") + except ET.ParseError: + return info + for disk in root.iter("disk"): + if disk.get("device", "disk") != "disk": + continue + driver = disk.find("driver") + target = disk.find("target") + fmt = (driver.get("type") if driver is not None else "") or "" + dev = (target.get("dev") if target is not None else "") or "?" + info["disks"].append({"target": dev, "format": fmt or "неизвестно"}) + if fmt and fmt != "qcow2": + info["unsupported"].append({"target": dev, "format": fmt}) + os_node = root.find("os") + if os_node is not None: + loader = os_node.find("loader") + info["uefi"] = os_node.get("firmware") == "efi" or os_node.find("nvram") is not None or (loader is not None and loader.get("type") == "pflash") + return info + + +def snapshot_state(state: str) -> dict[str, str]: + label, tone = SNAPSHOT_STATES.get((state or "").strip().lower(), (state or "—", "neutral")) + return {"label": label, "tone": tone} + + +def format_when(created_at: str, now: datetime | None = None) -> str: + """'2026-09-25 10:12:03' → 'Сегодня, 10:12' / 'Вчера, 10:12' / '25 сентября 2026, 10:12'.""" + try: + moment = datetime.strptime(created_at, "%Y-%m-%d %H:%M:%S") + except (TypeError, ValueError): + return created_at or "—" + today = (now or datetime.now()).date() + clock = moment.strftime("%H:%M") + if moment.date() == today: + return f"Сегодня, {clock}" + if moment.date() == today - timedelta(days=1): + return f"Вчера, {clock}" + return f"{moment.day} {MONTHS[moment.month - 1]} {moment.year}, {clock}" + + +def explain_error(raw: str) -> str: + """Текст ошибки libvirt → понятная фраза; технический текст остаётся в скобках.""" + text = re.sub(r"^error:\s*", "", (raw or "").strip(), flags=re.IGNORECASE) + first = text.splitlines()[0] if text else "" + low = first.lower() + for needle, explanation in KNOWN_ERRORS: + if needle in low: + return f"{explanation} (libvirt: {first})" + return first or "libvirt не выполнил команду и не объяснил причину." + + +def next_snapshot_name(existing: list[str], now: datetime | None = None) -> str: + stamp = (now or datetime.now()).strftime("snap-%Y%m%d-%H%M") + name, counter = stamp, 2 + while name in existing: + name, counter = f"{stamp}-{counter}", counter + 1 + return name + + +def clean_description(text: str) -> str: + text = re.sub(r"[\x00-\x1f\x7f]+", " ", text or "") + return re.sub(r"\s+", " ", text).strip().lstrip("-").strip()[:DESCRIPTION_MAX] + + +# ---------------------------------------------------------------- virsh +def current_snapshot(vm: str) -> str: + result = run_cmd(["virsh", "snapshot-current", "--domain", vm, "--name"], timeout=10) + return result["stdout"].strip() if result["ok"] else "" + + +def list_snapshots(vm: str) -> list[dict[str, Any]]: + result = run_cmd(["virsh", "snapshot-list", "--domain", vm], timeout=20) + rows = parse_snapshot_list(result["stdout"]) if result["ok"] else [] + current = current_snapshot(vm) if rows else "" + for row in rows: + xml = run_cmd(["virsh", "snapshot-dumpxml", "--domain", vm, "--snapshotname", row["name"]], timeout=10) + meta = parse_snapshot_xml(xml["stdout"]) if xml["ok"] else {} + row["description"] = meta.get("description", "") + row["with_memory"] = meta.get("with_memory", row["state"] == "running") + row["is_current"] = row["name"] == current + row["state_info"] = snapshot_state(row["state"]) + row["when"] = format_when(row["created_at"]) + row["title"] = row["description"] or row["name"] + return rows + + +def domain_checks(vm: str) -> dict[str, Any]: + result = run_cmd(["virsh", "dumpxml", vm], timeout=10) + return inspect_domain_xml(result["stdout"] if result["ok"] else "") + + +def vm_operations(vm: str) -> list[dict[str, Any]]: + return [op for op in list_operations(50) if op.get("type") in OPERATION_TYPES and op.get("vm_name") == vm] + + +def active_operations(vm: str) -> list[dict[str, Any]]: + return [op for op in vm_operations(vm) if op.get("status") in ("queued", "running")] + + +def create_blockers(vm: str, snapshots: list[dict[str, Any]], checks: dict[str, Any]) -> list[str]: + """Почему снимок сейчас нельзя создать (пустой список — можно).""" + reasons = [] + if len(snapshots) >= MAX_SNAPSHOTS: + reasons.append(f"Уже {MAX_SNAPSHOTS} снимков — это максимум для одной машины. Удалите ненужный снимок, чтобы создать новый.") + if checks.get("unsupported"): + disks = ", ".join(f"{d['target']} ({d['format']})" for d in checks["unsupported"]) + reasons.append(f"Внутренние снимки работают только с дисками qcow2, а у этой машины диск другого формата: {disks}.") + if active_operations(vm): + reasons.append("Дождитесь окончания предыдущей операции со снимками этой машины.") + return reasons + + +def _log_result(operation: dict[str, Any], result: dict[str, Any]) -> None: + for stream in ("stdout", "stderr"): + if result.get(stream): + append_operation_log(operation["id"], result[stream]) + + +def create_worker(operation: dict[str, Any]) -> None: + vm, snap, description = operation["vm_name"], operation["snapshot"], operation.get("description", "") + cmd = ["virsh", "snapshot-create-as", "--domain", vm, "--name", snap] + if description: + cmd += ["--description", description] + cmd.append("--atomic") + append_operation_log(operation["id"], "$ " + " ".join(cmd)) + update_operation(operation, progress=15, message="Сохраняем состояние машины и памяти — она на несколько секунд замрёт…" if operation.get("vm_running") else "Сохраняем состояние диска…", cmd=" ".join(cmd)) + result = run_cmd(cmd, timeout=SNAPSHOT_TIMEOUT) + _log_result(operation, result) + if result["ok"]: + finish_operation(operation, True, f"Снимок «{snap}» создан") + else: + finish_operation(operation, False, explain_error(cmd_error(result, "virsh завершился с ошибкой"))) + + +def revert_worker(operation: dict[str, Any]) -> None: + vm, snap = operation["vm_name"], operation["snapshot"] + cmd = ["virsh", "snapshot-revert", "--domain", vm, "--snapshotname", snap] + append_operation_log(operation["id"], "$ " + " ".join(cmd)) + update_operation(operation, progress=15, message="Возвращаем машину к сохранённому состоянию…", cmd=" ".join(cmd)) + result = run_cmd(cmd, timeout=SNAPSHOT_TIMEOUT) + _log_result(operation, result) + if not result["ok"] and "force" in (result.get("stderr") or "").lower(): + # libvirt просит --force, когда конфигурация машины изменилась после снимка: пользователь уже подтвердил возврат. + append_operation_log(operation["id"], "libvirt требует подтверждения (--force): повторяем принудительно.") + cmd.append("--force") + append_operation_log(operation["id"], "$ " + " ".join(cmd)) + result = run_cmd(cmd, timeout=SNAPSHOT_TIMEOUT) + _log_result(operation, result) + if not result["ok"]: + finish_operation(operation, False, explain_error(cmd_error(result, "virsh завершился с ошибкой"))) + return + state = presenters.vm_state(run_cmd(["virsh", "domstate", vm], timeout=8)["stdout"]) + finish_operation(operation, True, f"Машина возвращена к снимку «{snap}». Сейчас она {'работает' if state['running'] else 'выключена'}.") + + +# ---------------------------------------------------------------- routes +def _guard(request: Request, name: str): + """Редирект на /login, 400 для плохого имени, редирект на / для несуществующей машины — или None + dominfo.""" + auth_redirect = require_auth(request) + if auth_redirect: + return auth_redirect, "" + if not valid_vm_name(name): + return JSONResponse({"ok": False, "error": "Invalid VM name"}, status_code=400), "" + dominfo = run_cmd(["virsh", "dominfo", name], timeout=8) + if not dominfo["ok"]: + return RedirectResponse(url="/", status_code=303), "" + return None, dominfo["stdout"] + + +@router.get("/vm/{name}/snapshots") +def snapshots_page(request: Request, name: str): + blocked, dominfo = _guard(request, name) + if blocked: + return blocked + info = presenters.parse_dominfo(dominfo) + snapshots = list_snapshots(name) + checks = domain_checks(name) + running = presenters.vm_state(info["state"])["running"] + op_id = request.query_params.get("op", "") + operation = read_operation(op_id) if op_id else None + if operation and (operation.get("vm_name") != name or operation.get("type") not in OPERATION_TYPES): + operation = None + return render(request, "vm_snapshots.html", { + "vm": {"name": name, "info": info, "dominfo": dominfo}, + "vm_running": running, + "snapshots": snapshots, + "max_snapshots": MAX_SNAPSHOTS, + "checks": checks, + "blockers": create_blockers(name, snapshots, checks), + "suggested_name": next_snapshot_name([s["name"] for s in snapshots]), + "description_max": DESCRIPTION_MAX, + "operation": operation, + "active_operations": active_operations(name), + "snapshot_message": request.query_params.get("snapshot_message", ""), + "snapshot_error": request.query_params.get("snapshot_error", ""), + "raw_list": run_cmd(["virsh", "snapshot-list", "--domain", name], timeout=20)["stdout"], + }) + + +@router.post("/vm/{name}/snapshots/create") +def snapshot_create(request: Request, name: str, description: str = Form("")): + blocked, dominfo = _guard(request, name) + if blocked: + return blocked + snapshots = list_snapshots(name) + blockers = create_blockers(name, snapshots, domain_checks(name)) + if blockers: + return redirect_with_message(f"/vm/{name}/snapshots", "snapshot_error", blockers[0]) + running = presenters.vm_state(presenters.parse_dominfo(dominfo)["state"])["running"] + snap = next_snapshot_name([s["name"] for s in snapshots]) + if not valid_label(snap): + return redirect_with_message(f"/vm/{name}/snapshots", "snapshot_error", "Не удалось подобрать имя снимка.") + operation = new_operation("snapshot_create", f"Снимок машины {name}", vm_name=name, snapshot=snap, description=clean_description(description), vm_running=running) + run_operation(operation, create_worker) + return RedirectResponse(url=f"/vm/{name}/snapshots?op={operation['id']}", status_code=303) + + +@router.post("/vm/{name}/snapshots/{snap}/revert") +def snapshot_revert(request: Request, name: str, snap: str): + blocked, _ = _guard(request, name) + if blocked: + return blocked + if not valid_label(snap): + return JSONResponse({"ok": False, "error": "Invalid snapshot name"}, status_code=400) + if active_operations(name): + return redirect_with_message(f"/vm/{name}/snapshots", "snapshot_error", "Дождитесь окончания предыдущей операции со снимками этой машины.") + operation = new_operation("snapshot_revert", f"Возврат машины {name} к снимку {snap}", vm_name=name, snapshot=snap) + run_operation(operation, revert_worker) + return RedirectResponse(url=f"/vm/{name}/snapshots?op={operation['id']}", status_code=303) + + +@router.post("/vm/{name}/snapshots/{snap}/delete") +def snapshot_delete(request: Request, name: str, snap: str): + blocked, _ = _guard(request, name) + if blocked: + return blocked + if not valid_label(snap): + return JSONResponse({"ok": False, "error": "Invalid snapshot name"}, status_code=400) + if active_operations(name): + return redirect_with_message(f"/vm/{name}/snapshots", "snapshot_error", "Дождитесь окончания предыдущей операции со снимками этой машины.") + result = run_cmd(["virsh", "snapshot-delete", "--domain", name, "--snapshotname", snap], timeout=180) + if not result["ok"]: + return redirect_with_message(f"/vm/{name}/snapshots", "snapshot_error", "Не удалось удалить снимок: " + explain_error(cmd_error(result, "virsh завершился с ошибкой"))) + return redirect_with_message(f"/vm/{name}/snapshots", "snapshot_message", f"Снимок «{snap}» удалён") diff --git a/web/static/icons.svg b/web/static/icons.svg index 12ffba4..ce2bf98 100644 --- a/web/static/icons.svg +++ b/web/static/icons.svg @@ -57,4 +57,6 @@ + + diff --git a/web/templates/vm_detail.html b/web/templates/vm_detail.html index 7764bd1..4c2d366 100644 --- a/web/templates/vm_detail.html +++ b/web/templates/vm_detail.html @@ -53,6 +53,7 @@ + {{ icon('camera', 'icon-sm') }}Снимки diff --git a/web/templates/vm_snapshots.html b/web/templates/vm_snapshots.html new file mode 100644 index 0000000..fd7724b --- /dev/null +++ b/web/templates/vm_snapshots.html @@ -0,0 +1,167 @@ +{% extends "base.html" %} +{% from "_ui.html" import icon, badge, vm_badge, op_badge, empty, alert %} +{% set page_title = 'Снимки — ' ~ vm.name %} +{% set s = vm.info.state|vm_state %} +{% set count = snapshots|length %} + +{% block page_header %} +
+
+
{{ icon('arrow-left', 'icon-sm') }} Все машины
+
+ {{ icon('monitor', 'icon-lg') }} +
+

{{ vm.name }}

{{ vm_badge(vm.info.state, 'lg') }}
+

{{ vm.info.vcpus }} {{ plural(vm.info.vcpus, 'ядро', 'ядра', 'ядер') }} · {{ vm.info.memory_mb|format_mb }} памяти · {{ count }} {{ plural(count, 'снимок', 'снимка', 'снимков') }} из {{ max_snapshots }}

+
+
+
+
+ {{ icon('camera') }}Создать снимок +
+
+{% endblock %} + +{% block alerts %} + {% if snapshot_message %}{{ alert('success', snapshot_message) }}{% endif %} + {% if snapshot_error %}{{ alert('danger', snapshot_error) }}{% endif %} + {% if operation and operation.status == 'success' %}{{ alert('success', operation.message) }}{% endif %} + {% if operation and operation.status == 'error' %}{{ alert('danger', operation.message, title='Не получилось: ' ~ operation.title|lower) }}{% endif %} +{% endblock %} + +{% block content %} + + +
+
+ {% for op in active_operations %} +
+ +
+ {{ op.title }} + {{ op.message }} +
+
+ Подробнее{{ icon('chevron-right', 'icon-sm') }} +
+ {% endfor %} + +
+
+

Точки восстановления

Сохранённые состояния машины, к которым можно вернуться

+ {{ count }} из {{ max_snapshots }} +
+ {% if snapshots %} +
+ {% for snap in snapshots %} +
+ {{ icon('camera') }} +
+
+ {{ snap.title }} + {% if snap.is_current %}{{ badge('Текущий', 'brand', dot=False, attrs='title="Машина сейчас продолжает работу от этого снимка"') }}{% endif %} +
+
{{ snap.when }} · {{ snap.state_info.label }}{% if snap.with_memory %} — сохранена вместе с памятью{% endif %}{% if snap.description %} · {{ snap.name }}{% endif %}
+
+
+ +
+
+ +
+
+ {% endfor %} +
+ {% else %} + {{ empty('camera', 'Снимков пока нет', 'Сделайте снимок перед обновлением или рискованной настройкой. Если что-то пойдёт не так, машину можно будет вернуть назад за минуту.') }} + {% endif %} +
+ +
+ {{ icon('terminal', 'icon-sm') }}Для специалистов{{ icon('chevron-down', 'icon-sm') }} +
+

Это внутренние снимки libvirt (virsh snapshot-create-as … --atomic): они хранятся внутри qcow2-файла диска. Возврат — virsh snapshot-revert, удаление — virsh snapshot-delete.

+
+
virsh snapshot-list --domain {{ vm.name }}
+ 
{{ raw_list or '—' }}
+
+ {% if checks.disks %} +

Диски: {% for d in checks.disks %}{{ d.target }} — {{ d.format }}{{ ', ' if not loop.last }}{% endfor %}{% if checks.uefi %} · прошивка UEFI{% endif %}

+ {% endif %} +
+
+
+ + +
+{% endblock %} + +{% block scripts %} +{% if active_operations %} + +{% endif %} +{% endblock %}