fix(monitor): run scheduled backup jobs detached and follow them by their task UPID

.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):

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({

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/<job_id>/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])
+        upid, error = _start_vzdump(node, arguments, job_id)
-        if ok:
+        if upid:
-            found = re.search(r"UPID:[^\s\"']+", out)
+            tasks.append({"node": node, "upid": upid})
-            tasks.append({"node": node, "upid": found.group(0) if found else ""})
         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})

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():