diff --git a/.github/scripts/tests/test_vm_backup_jobs_wiring.py b/.github/scripts/tests/test_vm_backup_jobs_wiring.py index 930c82e3..d6148e37 100644 --- a/.github/scripts/tests/test_vm_backup_jobs_wiring.py +++ b/.github/scripts/tests/test_vm_backup_jobs_wiring.py @@ -24,7 +24,10 @@ class VmBackupJobsWiring(TestCase): source = ROUTES.read_text() self.assertNotIn('shell=True', source) self.assertNotIn('os.system', source) - self.assertEqual(source.count('subprocess.run('), 1) + # pvesh, and the transient unit a backup started by hand runs in. + programs = re.findall(r'subprocess\.run\(\s*\[\s*"([a-z-]+)"', source) + self.assertEqual(len(programs), source.count('subprocess.run(')) + self.assertEqual(set(programs), {'pvesh', 'systemd-run', 'systemctl'}) self.assertIn('subprocess.run(["pvesh", *args]', source) def test_the_monitor_ships_and_registers_it(self): diff --git a/AppImage/scripts/flask_server.py b/AppImage/scripts/flask_server.py index e617932c..3fea17d1 100644 --- a/AppImage/scripts/flask_server.py +++ b/AppImage/scripts/flask_server.py @@ -99,6 +99,7 @@ from flask_notification_routes import notification_bp # noqa: E402 from flask_oci_routes import oci_bp # noqa: E402 from flask_audit_routes import audit_bp # noqa: E402 from flask_vm_backup_routes import vm_backup_bp # noqa: E402 +from flask_vm_backup_routes import active_vzdump # noqa: E402 from guest_backup_archives import (BackupArchiveError, delete_guest_backups, guest_backup_deletion_status) # noqa: E402 from notification_manager import notification_manager, resolve_notification_hostname # noqa: E402 @@ -13411,6 +13412,7 @@ def api_create_backup(vmid): pass unit = f'proxmenux-vzdump-{vmid}-{uuid.uuid4().hex[:8]}' out_path = os.path.join(run_dir, f'vzdump-{unit}.out') + since = int(time.time()) - 1 launch = subprocess.run( ['systemd-run', f'--unit={unit}', '--collect', '--quiet', f'--description=ProxMenux backup of {vm_type.upper()} {vmid}', @@ -13438,7 +13440,21 @@ def api_create_backup(vmid): capture_output=True, text=True).stdout.strip() if state not in ('active', 'activating'): break + # pvesh prints the UPID only once the backup has ended. + upid = active_vzdump(node, since) + if upid: + break time.sleep(0.3) + if upid is None: + # A short backup can finish between two reads. + try: + with open(out_path, encoding='utf-8', errors='replace') as handle: + buf = handle.read() + except OSError: + pass + m = upid_re.search(buf) + if m: + upid = m.group(0).rstrip(':') if upid is None: tail = buf.strip() or 'pvesh exited without producing a UPID' return jsonify({ diff --git a/AppImage/scripts/flask_vm_backup_routes.py b/AppImage/scripts/flask_vm_backup_routes.py index 78e77919..c7bb1a53 100644 --- a/AppImage/scripts/flask_vm_backup_routes.py +++ b/AppImage/scripts/flask_vm_backup_routes.py @@ -24,7 +24,10 @@ policy only changes the native storage configuration; it never runs pruning. import json import os import re +import shutil import subprocess +import time +import uuid from flask import Blueprint, jsonify, request @@ -395,6 +398,86 @@ def _run_arguments(job): return args +# A job of several guests leaves the id of its task empty. +UPID_RE = re.compile(r"UPID:[^:\s]+:[^:\s]+:[^:\s]+:[^:\s]+:vzdump:[^:\s]*:[^:\s]+:") +RUN_DIR = "/run/proxmenux" + + +def active_vzdump(node, since): + """The UPID of the first backup task the node started at or after `since`.""" + ok, tasks = _pvesh_json(["get", f"/nodes/{node}/tasks", "--source", "active", "--typefilter", "vzdump"]) + if not ok or not isinstance(tasks, list): + return None + started = sorted((task for task in tasks if isinstance(task, dict) + and UPID_RE.match(str(task.get("upid") or "")) + and isinstance(task.get("starttime"), int) and task["starttime"] >= since), + key=lambda task: task["starttime"]) + return started[0]["upid"].rstrip(":") if started else None + + +def _start_vzdump(node, arguments, job_id): + """Start the part of a job that runs on a node and return its task UPID. + + pvesh waits for the whole backup and prints the UPID only at the end, and + a child of the Monitor is killed with its service, so vzdump runs in its + own transient unit and its UPID is read from the node's active tasks. The + task then runs to the end on its own. Returns (upid, error).""" + if not shutil.which("systemd-run"): + return None, "systemd-run not found" + run_dir = RUN_DIR + try: + os.makedirs(run_dir, mode=0o700, exist_ok=True) + now = time.time() + for old in os.listdir(run_dir): + old_path = os.path.join(run_dir, old) + if old.startswith("vzdump-") and now - os.path.getmtime(old_path) > 86400: + os.remove(old_path) + except OSError: + pass + unit = f"proxmenux-vzdump-job-{uuid.uuid4().hex[:8]}" + out_path = os.path.join(run_dir, f"vzdump-{unit}.out") + since = int(time.time()) - 1 + try: + launch = subprocess.run( + ["systemd-run", f"--unit={unit}", "--collect", "--quiet", + f"--description=ProxMenux backup job {job_id}", + f"--property=StandardOutput=file:{out_path}", + "pvesh", "create", f"/nodes/{node}/vzdump", *arguments], + capture_output=True, text=True, timeout=15) + except subprocess.TimeoutExpired: + return None, "systemd-run timed out" + if launch.returncode != 0: + return None, (launch.stderr or launch.stdout).strip() + buf = "" + deadline = time.time() + 10.0 + while time.time() < deadline: + try: + with open(out_path, encoding="utf-8", errors="replace") as handle: + buf = handle.read() + except OSError: + buf = "" + found = UPID_RE.search(buf) + if found: + return found.group(0).rstrip(":"), None + state = subprocess.run(["systemctl", "is-active", unit], capture_output=True, text=True).stdout.strip() + if state not in ("active", "activating"): + break + upid = active_vzdump(node, since) + if upid: + return upid, None + time.sleep(0.5) + # A short backup can finish between two reads: its UPID is the last line. + try: + with open(out_path, encoding="utf-8", errors="replace") as handle: + buf = handle.read() + except OSError: + pass + found = UPID_RE.search(buf) + if found: + return found.group(0).rstrip(":"), None + return None, (buf.strip() or "pvesh exited without producing a UPID")[-800:] + + @vm_backup_bp.route("/api/vm-backup-jobs//run", methods=["POST"]) @require_admin_scope def run_job(job_id): @@ -418,12 +501,11 @@ def run_job(job_id): for node in nodes: if not NODE_RE.match(node): continue - ok, out = _pvesh(["create", f"/nodes/{node}/vzdump", *arguments]) - if ok: - found = re.search(r"UPID:[^\s\"']+", out) - tasks.append({"node": node, "upid": found.group(0) if found else ""}) + upid, error = _start_vzdump(node, arguments, job_id) + if upid: + tasks.append({"node": node, "upid": upid}) else: - failures.append({"node": node, "error": out}) + failures.append({"node": node, "error": error}) if not tasks: return _err(failures[0]["error"] if failures else "The backup could not be started", 500) return jsonify({"success": True, "tasks": tasks, "failures": failures}) diff --git a/tests/test_vm_backup_jobs_api.py b/tests/test_vm_backup_jobs_api.py index b5e50396..70699ba9 100644 --- a/tests/test_vm_backup_jobs_api.py +++ b/tests/test_vm_backup_jobs_api.py @@ -1,8 +1,11 @@ """VM/CT backup jobs API: what reaches pvesh, with pvesh itself replaced.""" import importlib import importlib.util +import contextlib import json +import subprocess import sys +import tempfile import types import unittest from pathlib import Path @@ -95,7 +98,14 @@ class VmBackupJobsApiTests(unittest.TestCase): return False, "storage is not online" return True, f'"UPID:{node}:0001:0002:0003:vzdump::root@pam:"\n' return True, "" - return patch.object(self.routes, "_pvesh", side_effect=run) + + def start(node, arguments, job_id): + ok, out = run(["create", f"/nodes/{node}/vzdump", *arguments]) + return (out.strip().strip('"').rstrip(":"), None) if ok else (None, out) + stack = contextlib.ExitStack() + stack.enter_context(patch.object(self.routes, "_pvesh", side_effect=run)) + stack.enter_context(patch.object(self.routes, "_start_vzdump", side_effect=start)) + return stack def changes(self): return [call for call in self.calls if call[0] in ("create", "set", "delete")] @@ -234,6 +244,57 @@ class VmBackupJobsApiTests(unittest.TestCase): self.assertEqual([failure["node"] for failure in answer["failures"]], ["pve2"]) self.assertNotIn("/nodes/pve3/vzdump", [call[1] for call in self.calls]) + def test_a_run_starts_vzdump_detached_and_answers_with_its_upid(self): + # pvesh waits for the whole backup; a child of the Monitor is killed + # with it. The task runs in its own unit and is followed by its UPID. + upid = "UPID:pve1:0001:0002:0003:vzdump::root@pam:" + launched = [] + + def systemd_run(command, **kwargs): + launched.append(command) + if command[0] == "systemd-run": + output = next(part.split("file:", 1)[1] for part in command if part.startswith("--property=StandardOutput=file:")) + Path(output).write_text(f"{upid}\n") + return subprocess.CompletedProcess(command, 0, "active\n", "") + with tempfile.TemporaryDirectory() as folder, patch.object(self.routes, "RUN_DIR", folder), \ + patch.object(self.routes.shutil, "which", return_value="/usr/bin/systemd-run"), \ + patch.object(self.routes.subprocess, "run", side_effect=systemd_run): + self.assertEqual(self.routes._start_vzdump("pve1", ["--vmid", "101"], "backup-nightly"), (upid.rstrip(":"), None)) + unit = launched[0] + self.assertEqual(unit[0], "systemd-run") + self.assertEqual(unit[unit.index("pvesh"):], ["pvesh", "create", "/nodes/pve1/vzdump", "--vmid", "101"]) + + def test_a_long_run_is_found_among_the_active_tasks_while_pvesh_still_waits(self): + # Until the backup ends pvesh prints only its log, never the UPID. + upid = "UPID:pve1:0001:0002:0003:vzdump:107:root@pam:" + + def systemd_run(command, **kwargs): + if command[0] == "systemd-run": + output = next(part.split("file:", 1)[1] for part in command if part.startswith("--property=StandardOutput=file:")) + Path(output).write_text("INFO: starting new backup job: vzdump 107\n") + return subprocess.CompletedProcess(command, 0, "", "") + if command[0] == "systemctl": + return subprocess.CompletedProcess(command, 0, "active\n", "") + tasks = [{"upid": "UPID:pve1:0009:0002:0001:vzdump:200:root@pam:", "starttime": 1}, + {"upid": upid, "starttime": 2_000_000_000}] + return subprocess.CompletedProcess(command, 0, json.dumps(tasks), "") + with tempfile.TemporaryDirectory() as folder, patch.object(self.routes, "RUN_DIR", folder), \ + patch.object(self.routes.shutil, "which", return_value="/usr/bin/systemd-run"), \ + patch.object(self.routes.subprocess, "run", side_effect=systemd_run): + self.assertEqual(self.routes._start_vzdump("pve1", ["--vmid", "107"], "backup-nightly"), (upid.rstrip(":"), None)) + + def test_a_run_that_ends_without_a_upid_reports_what_pvesh_said(self): + def systemd_run(command, **kwargs): + if command[0] == "systemd-run": + output = next(part.split("file:", 1)[1] for part in command if part.startswith("--property=StandardOutput=file:")) + Path(output).write_text("storage 'pbs' is not online\n") + return subprocess.CompletedProcess(command, 0, "", "") + return subprocess.CompletedProcess(command, 3, "inactive\n", "") + with tempfile.TemporaryDirectory() as folder, patch.object(self.routes, "RUN_DIR", folder), \ + patch.object(self.routes.shutil, "which", return_value="/usr/bin/systemd-run"), \ + patch.object(self.routes.subprocess, "run", side_effect=systemd_run): + self.assertEqual(self.routes._start_vzdump("pve1", [], "backup-nightly"), (None, "storage 'pbs' is not online")) + def test_a_job_whose_guests_are_on_no_online_node_starts_nothing(self): self.job["vmid"] = "999" with self.pvesh():