External/joulenap
1
0
Fork 0
mirror of https://github.com/Joulenap/joulenap.git synced 2026-08-11 13:21:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(jobs): multi-source, cluster-aware backup route cycle

Browse Source 
One run now executes one backup route: N source PVEs (each possibly a
cluster) onto one PBS target. Wake and power-off stay with the power
lease, so the cycle starts with the box awake and never touches its
power.

- PveClient lists guests cluster-wide via /cluster/resources (identical
  on a standalone node), tags each with its node and drops templates;
  vzdump takes a node, and the task endpoints read theirs from the UPID
  so one client can drive several nodes.
- Sources are isolated: a broken one leaves its backup:<pve-id> step
  failed and the run continues, finishing failed and naming it. GC and
  verify still run - the box is awake and the other snapshots are real.
- The guest tally aggregates across sources; failed guest names keep
  working.
- Cache writes carry real ids: datastore_stats keyed by the target,
  guest_backups attributing each vmid to the PVE that backed it up.
- CycleDeps gains device-shaped connect_pve/connect_pbs beside the
  config-shaped pair, and RunRecorder.step takes a label.

The 0.9 cycle and the flat pve/pbs/backup config sections stay for now:
their other consumers (the GC/verify/monitor cycles, the scheduler, the
API routers) are ported in the next milestones and delete both halves
together.
This commit is contained in:
Catubba
2026-08-02 17:59:08 +02:00
parent e5d0d5cf58
commit 9e858df5b1
8 changed files with 951 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/connectors/pve.py
+54 -6
View File
@@ -27,6 +27,10 @@ class Guest:
name: str
type: str # "qemu" (VM) or "lxc" (CT)
status: str # "running" | "stopped" | ...
# Which cluster node holds it. Empty from the per-node listing (the caller knows the
# node already); filled in by :meth:`PveClient.list_cluster_guests`, which is what lets
# a route group its vzdump calls per node.
node: str = ""
@property
def is_ct(self) -> bool:
@@ -51,9 +55,12 @@ class PveClient:
def __init__(
self,
host: str,
node: str,
token_id: str,
token_secret: str,
# A route's PVE device has no node (nodes are discovered at runtime): the node-scoped
# endpoints below are the 0.9 path and the wizard's, and the task endpoints read the
# node from the UPID instead.
node: str = "",
port: int = 8006,
verify_tls: bool = False,
timeout: float = 30.0,
@@ -110,6 +117,32 @@ class PveClient:
guests.sort(key=lambda g: g.vmid)
return guests
def list_cluster_guests(self) -> list[Guest]:
"""Every VM/CT this endpoint knows about, each tagged with the node holding it.
One call covers a whole cluster — the endpoint proxies its nodes — and works just as
well against a standalone node, which the API simply reports as a one-node cluster.
Templates are dropped: they are never backed up, so counting them would inflate the
guest tally and naming one explicitly would fail the vzdump task.
"""
rows = self._api.request("GET", "/cluster/resources", params={"type": "vm"}) or []
guests: list[Guest] = []
for r in rows:
if r.get("template") or r.get("vmid") is None:
continue
kind = r.get("type") or ""
guests.append(
Guest(
vmid=int(r["vmid"]),
name=r.get("name") or f"{kind}-{r['vmid']}",
type=kind,
status=r.get("status", "unknown"),
node=r.get("node", ""),
)
)
guests.sort(key=lambda g: g.vmid)
return guests
# --- backup --------------------------------------------------------------
def vzdump(
@@ -121,10 +154,13 @@ class PveClient:
mode: str = "snapshot",
prune_backups: str | None = None,
bwlimit: int = 0,
node: str = "",
) -> str:
"""Start a vzdump backup; returns the task UPID to poll with :meth:`wait_task`.
Either pass ``vmids`` (explicit selection) or ``all_guests=True``.
Either pass ``vmids`` (explicit selection) or ``all_guests=True``. ``node`` picks the
cluster node to run it on (a backup route starts one task per node); it defaults to
the client's own node.
"""
params: dict[str, Any] = {"storage": storage, "mode": mode}
if all_guests:
@@ -135,12 +171,24 @@ class PveClient:
params["prune-backups"] = prune_backups
if bwlimit:
params["bwlimit"] = bwlimit
return self._api.request("POST", f"/nodes/{self.node}/vzdump", data=params)
return self._api.request("POST", f"/nodes/{node or self.node}/vzdump", data=params)
# --- tasks ---------------------------------------------------------------
def _task_node(self, upid: str) -> str:
"""The node a task runs on, read from its own UPID (``UPID:<node>:<pid>:…``).
One client can drive tasks on several nodes (a cluster route starts one vzdump per
node), so the node has to come from the task rather than from the client. Falls back
to the client's node if the string isn't a UPID.
"""
parts = upid.split(":")
if len(parts) > 2 and parts[0] == "UPID" and parts[1]:
return parts[1]
return self.node
def task_status(self, upid: str) -> dict[str, Any]:
return self._api.request("GET", f"/nodes/{self.node}/tasks/{upid}/status")
return self._api.request("GET", f"/nodes/{self._task_node(upid)}/tasks/{upid}/status")
def task_log(self, upid: str, start: int = 0, limit: int = 5000) -> list[LogLine]:
"""Fetch task-log lines starting at offset ``start``, as ``(line_no, text)`` pairs.
@@ -150,7 +198,7 @@ class PveClient:
"""
data = self._api.request(
"GET",
f"/nodes/{self.node}/tasks/{upid}/log",
f"/nodes/{self._task_node(upid)}/tasks/{upid}/log",
params={"start": start, "limit": limit},
)
return [(int(e["n"]), e.get("t") or "") for e in (data or [])]
@@ -161,7 +209,7 @@ class PveClient:
Without this a cancelled backup would keep running on the PVE host after Joulenap
stopped watching it, and the next run would collide with it.
"""
self._api.request("DELETE", f"/nodes/{self.node}/tasks/{upid}")
self._api.request("DELETE", f"/nodes/{self._task_node(upid)}/tasks/{upid}")
def wait_task(
self,
backend/app/db/models.py
+4 -2
View File
@@ -144,7 +144,9 @@ class RunStep(Base):
id: Mapped[int] = mapped_column(primary_key=True)
run_id: Mapped[int] = mapped_column(ForeignKey("runs.id", ondelete="CASCADE"))
name: Mapped[str] = mapped_column(String(16))
# A StepName value, optionally suffixed with what it applied to: a backup route records
# one step per source PVE, ``backup:pve-alpha``.
name: Mapped[str] = mapped_column(String(64))
status: Mapped[str] = mapped_column(String(16), default=StepStatus.RUNNING)
started_at: Mapped[datetime] = mapped_column(UtcDateTime(), default=_utcnow)
@@ -188,7 +190,7 @@ class TaskLogLine(Base):
id: Mapped[int] = mapped_column(primary_key=True)
run_id: Mapped[int] = mapped_column(ForeignKey("runs.id", ondelete="CASCADE"))
step: Mapped[str] = mapped_column(String(16)) # StepName value (backup/gc/verify)
step: Mapped[str] = mapped_column(String(64)) # RunStep.name (e.g. gc, backup:pve-alpha)
source: Mapped[str] = mapped_column(String(8)) # "pve" | "pbs"
line_no: Mapped[int] = mapped_column() # the task's own 1-based line number
text: Mapped[str] = mapped_column(Text)
backend/app/jobs/backup_cycle.py
+298 -3
View File
@@ -15,7 +15,7 @@ import time
from collections.abc import Callable
from typing import Any, Protocol
from ..config import Config
from ..config import Config, PbsDevice, Route, RouteGuests, RouteSource
from ..connectors.errors import TaskCancelled
from ..connectors.pbs import DatastoreStatus
from ..connectors.pve import Guest, build_prune_string
@@ -63,7 +63,7 @@ def _guest_watcher(summary: GuestSummary, names: dict[int, str]):
return watch
def _tailer(recorder: RunRecorder, step: StepName, source: str, watch=None):
def _tailer(recorder: RunRecorder, step: str, source: str, watch=None):
"""A ``wait_task(on_log=...)`` callback that persists each task-log batch.
Best-effort: a failure to store a log line must never fail an otherwise-fine backup,
@@ -106,7 +106,7 @@ def _wait_or_stop(
upid: str,
recorder: RunRecorder,
deps: CycleDeps,
step: StepName,
step: str,
source: str,
watch: Callable[[list[tuple[int, str]]], None] | None = None,
) -> None:
@@ -652,3 +652,298 @@ def _notify_result(
deps.notify(config, recorder.run, datastore, guests, deps.next_run())
except Exception as exc:
recorder.log(LogLevel.WARN, f"notification failed: {exc}")
# --- v1.0 route cycle --------------------------------------------------------
#
# One run executes one **backup route**: N source PVEs (each possibly a cluster) -> one PBS
# target. Wake and power-off are deliberately absent — ``JobService._execute`` takes a
# :class:`~.lease.PowerLease` on the target around the job (M04), so the cycle starts with
# the box already awake and ends without touching its power. Everything else below mirrors
# the 0.9 cycle above, which it replaces once the flat ``pve:``/``pbs:``/``backup:`` config
# sections and their other consumers are gone (M06/M07).
def _find_device(devices, device_id: str):
return next((d for d in devices if d.id == device_id), None)
def _guests_by_node(selection: RouteGuests, guests: list[Guest]) -> dict[str, list[int]]:
"""Group the guests this source wants by the cluster node holding them — a route runs
one vzdump per node, and only on nodes that actually have something to back up."""
wanted = None if selection.mode == "all" else set(selection.list)
picked: dict[str, list[int]] = {}
for guest in guests:
if wanted is None or guest.vmid in wanted:
picked.setdefault(guest.node, []).append(guest.vmid)
return picked
def _route_backup_source(
config: Config,
route: Route,
source: RouteSource,
recorder: RunRecorder,
deps: CycleDeps,
summary: GuestSummary,
step,
) -> set[int]:
"""Back up one source PVE onto the route's target: one vzdump per cluster node.
Returns the vmids this source covered, so the last-backup cache can attribute each guest
to the PVE it came from. Raises on failure — the caller records that against this
source's step and moves on to the next source.
"""
pve = _find_device(config.pves, source.pve)
if pve is None:
raise CycleAbort(f"source pve '{source.pve}' no longer exists")
storage = pve.storages.get(route.target)
if not storage:
raise CycleAbort(
f"pve '{pve.id}' has no storage mapping for pbs '{route.target}' "
"(Datacenter > Storage)"
)
prune = build_prune_string(route.retention.model_dump())
step_name = f"{StepName.BACKUP.value}:{source.pve}"
upids: list[str] = []
covered: set[int] = set()
with deps.connect_pve(pve) as client:
# One cluster-wide listing feeds all three needs: the per-node grouping, the guest
# tally and the {vmid: name} map the notification names failed guests with.
guests = client.list_cluster_guests()
names = {g.vmid: g.name for g in guests}
per_node = _guests_by_node(source.guests, guests)
# "all" stays vzdump's own ``all`` flag rather than the vmids we just listed, so PVE
# keeps deciding: a guest marked *exclude from backup* is honoured, and one created
# since the listing is still covered.
all_guests = source.guests.mode == "all"
if not all_guests:
missing = sorted(set(source.guests.list) - {g.vmid for g in guests})
if missing:
recorder.log(
LogLevel.WARN,
f"pve '{pve.id}': selected guest(s) {missing} are not on it "
"(deleted, a template, or migrated away); skipping them",
)
if not per_node:
raise CycleAbort(f"pve '{pve.id}': no guests selected for backup")
for node, vmids in per_node.items():
upid = client.vzdump(
storage,
node=node,
vmids=None if all_guests else vmids,
all_guests=all_guests,
mode=route.options.mode,
prune_backups=prune,
bwlimit=route.options.bwlimit,
)
upids.append(upid)
step.detail = ", ".join(upids)
# Counted before the wait: a task that dies still set out to back these up.
summary.total += len(vmids)
done_before = summary.ok
_wait_or_stop(
client,
upid,
recorder,
deps,
step_name,
"pve",
_guest_watcher(summary, names),
)
# The task exited OK, so every guest it covered was backed up whatever the log
# parse made of it — a vzdump wording change must never report "0/14" on a good
# run. Only on success, so a failed node doesn't advertise guests as backed up.
summary.ok = done_before + len(vmids)
covered |= set(vmids)
return covered
def _route_preflight(
route: Route, target: PbsDevice, recorder: RunRecorder, deps: CycleDeps
) -> None:
"""Abort before any vzdump if the target datastore is below the route's free-space floor.
No-op when ``min_free_percent`` is 0 (the default), so the step only appears when the
user opted in. An abort here leaves the PBS on for inspection (the lease's failure
policy), matching the other failure paths.
"""
threshold = route.options.min_free_percent
if threshold <= 0:
return
with recorder.step(StepName.PRECHECK) as step:
with deps.connect_pbs(target) as pbs:
ds = pbs.datastore_status()
_cache_route_datastore(target, recorder, ds)
step.detail = f"{ds.avail_pct:.1f}% free ({ds.avail / 1_000_000_000:.0f} GB)"
if ds.avail_pct < threshold:
raise CycleAbort(
f"PBS '{target.id}' datastore {target.datastore!r} only {ds.avail_pct:.1f}% "
f"free (need >= {threshold}%); skipping backup"
)
def _route_gc_step(target: PbsDevice, recorder: RunRecorder, deps: CycleDeps) -> None:
"""Garbage-collect the route's target datastore while the PBS is still awake."""
with recorder.step(StepName.GC) as step:
with deps.connect_pbs(target) as pbs:
upid = pbs.start_gc()
step.detail = upid
_wait_or_stop(pbs, upid, recorder, deps, StepName.GC.value, "pbs")
def _route_verify_step(
target: PbsDevice, recorder: RunRecorder, deps: CycleDeps, *, outdated_after: int | None
) -> None:
"""Verify snapshots on the route's target. ``outdated_after=None`` -> only never-verified
(i.e. this run's new) snapshots; an int -> also re-verify ones older than that many days
(0 -> everything), which is what a Verify route will ask for (M06)."""
with recorder.step(StepName.VERIFY) as step:
with deps.connect_pbs(target) as pbs:
if outdated_after is not None and outdated_after <= 0:
upid = pbs.start_verify(ignore_verified=False)
else:
upid = pbs.start_verify(ignore_verified=True, outdated_after=outdated_after)
step.detail = upid
_wait_or_stop(pbs, upid, recorder, deps, StepName.VERIFY.value, "pbs")
def _cache_route_datastore(
target: PbsDevice, recorder: RunRecorder, ds: DatastoreStatus
) -> None:
"""Persist the target's usage so the UI can show it while the PBS sleeps. Best-effort:
a cache-write failure must never fail the run."""
try:
with session_scope() as session:
upsert_datastore_stat(session, target.id, target.datastore, ds.total, ds.used)
except Exception as exc:
recorder.log(LogLevel.WARN, f"could not cache datastore usage: {exc}")
def _route_read_datastore(
target: PbsDevice, recorder: RunRecorder, deps: CycleDeps
) -> DatastoreStatus | None:
"""Read the target's usage for the notification, while it is still awake. Best-effort:
a read failure only costs the notification its usage line."""
try:
with deps.connect_pbs(target) as pbs:
ds = pbs.datastore_status()
except Exception as exc:
recorder.log(LogLevel.WARN, f"could not read datastore usage: {exc}")
return None
recorder.log(LogLevel.INFO, f"PBS '{target.id}' datastore {ds.used_pct}% used")
_cache_route_datastore(target, recorder, ds)
return ds
def _refresh_route_backup_cache(
target: PbsDevice,
covered: dict[str, set[int]],
recorder: RunRecorder,
deps: CycleDeps,
) -> None:
"""Cache each guest's latest snapshot time per (source pve, target pbs), so the dashboard
can show last-backup dates once the PBS sleeps again.
One datastore lists snapshots by vmid with no idea which PVE they came from, so each
source claims the vmids it actually backed up. Two PVEs sharing a vmid both get a row
with the same time — that is a real PBS namespace collision, not something to fix here.
Best-effort: the backup already succeeded, so a read/write failure is logged and dropped.
"""
try:
with deps.connect_pbs(target) as pbs:
latest = pbs.latest_backups()
cached = 0
with session_scope() as session:
for pve_id, vmids in covered.items():
mine = {vmid: ts for vmid, ts in latest.items() if vmid in vmids}
upsert_last_backups(session, pve_id, target.id, mine)
cached += len(mine)
except Exception as exc:
recorder.log(LogLevel.WARN, f"could not refresh last-backup cache: {exc}")
return
recorder.log(LogLevel.INFO, f"cached last-backup times for {cached} guest(s)")
def run_route_backup(
config: Config, route: Route, recorder: RunRecorder, deps: CycleDeps
) -> None:
"""Execute one backup route, recording each step. Sets the final run status itself.
Sources run in order and are isolated from each other: one unreachable PVE fails the run
but the remaining sources still get their backup, because a shared target and a shared
wake window are exactly what makes a multi-source route worth having.
"""
datastore: DatastoreStatus | None = None
# Owned here, not by the per-source step: a guest failing takes the whole vzdump task
# down and unwinds that frame, and the failed run is exactly the one whose tally we want.
guests = GuestSummary()
covered: dict[str, set[int]] = {}
failed: list[str] = []
try:
target = _find_device(config.pbss, route.target)
if target is None:
raise CycleAbort(f"route '{route.id}': target pbs '{route.target}' no longer exists")
_route_preflight(route, target, recorder, deps)
for source in route.sources:
# A cancel that lands between sources must not start the next one — the task
# waits check the flag themselves, this covers the gaps between them.
if deps.cancelled():
raise CycleCancelled("Run cancelled")
try:
with recorder.step(StepName.BACKUP, label=source.pve) as step:
covered[source.pve] = _route_backup_source(
config, route, source, recorder, deps, guests, step
)
except CycleCancelled:
raise
except Exception as exc:
# The step row already carries the failure; keep going so one broken source
# doesn't cost the others their backup.
failed.append(source.pve)
recorder.log(LogLevel.ERROR, f"source '{source.pve}' failed: {exc}")
recorder.run.guests_ok = guests.ok
if deps.cancelled():
raise CycleCancelled("Run cancelled")
# GC and verify still run when a source failed: the PBS is awake and the snapshots
# the other sources wrote are real.
if route.options.gc:
_route_gc_step(target, recorder, deps)
else:
recorder.skip_step(StepName.GC, "GC disabled for this route")
if deps.cancelled():
raise CycleCancelled("Run cancelled")
if route.options.verify_after:
_route_verify_step(target, recorder, deps, outdated_after=None)
else:
recorder.skip_step(StepName.VERIFY, "verify disabled for this route")
datastore = _route_read_datastore(target, recorder, deps)
_refresh_route_backup_cache(target, covered, recorder, deps)
if failed:
recorder.finish(
RunStatus.FAILURE, error=f"backup failed for source(s): {', '.join(failed)}"
)
else:
recorder.finish(RunStatus.SUCCESS)
except CycleCancelled:
# No notification: the user pressed Stop and is standing at the UI — a "backup
# aborted" push would just be noise about their own click.
recorder.finish(RunStatus.ABORTED, error="Cancelled by user")
return
except CycleAbort as exc:
recorder.finish(RunStatus.ABORTED, error=str(exc))
except Exception as exc: # connector/task failures: the lease leaves the PBS on
recorder.finish(RunStatus.FAILURE, error=str(exc))
_notify_result(config, recorder, deps, datastore, guests)
backend/app/jobs/deps.py
+37 -1
View File
@@ -12,7 +12,7 @@ from collections.abc import Callable
from dataclasses import dataclass
from datetime import datetime
from ..config import Config
from ..config import Config, PbsDevice, PveDevice
from ..connectors import net, tls
from ..connectors.pbs import DatastoreStatus, PbsClient
from ..connectors.power import PbsPower
@@ -51,6 +51,37 @@ def _build_pbs(config: Config) -> PbsClient:
)
# --- device-shaped connectors (the v1.0 route model) -------------------------
# The two above are bound to the 0.9 single-PVE/PBS Config; a route points at devices from
# ``pves[]``/``pbss[]`` instead. ponytail: both shapes coexist while the 0.9 cycle, the
# wizard and the status routers still run on the flat sections — M07 deletes the pair above.
def _connect_pve(pve: PveDevice) -> PveClient:
# No node: a route lists guests cluster-wide and names the node per vzdump call.
return PveClient(
host=pve.host,
token_id=pve.api_token_id,
token_secret=pve.api_token_secret,
port=pve.port,
verify_tls=pve.verify_tls,
)
def _connect_pbs(pbs: PbsDevice) -> PbsClient:
verify: bool | ssl.SSLContext = False
if pbs.fingerprint:
verify = tls.pinned_ssl_context(pbs.host, pbs.port, pbs.fingerprint)
return PbsClient(
host=pbs.host,
datastore=pbs.datastore,
token_id=pbs.api_token_id,
token_secret=pbs.api_token_secret,
port=pbs.port,
verify=verify,
)
def _build_power(config: Config) -> PbsPower:
p = config.pbs
return PbsPower(host=p.host, user=p.ssh_user, key_path=p.ssh_key_path)
@@ -104,6 +135,11 @@ class CycleDeps:
# (config, run, datastore, guests, next_at) -> None. Untyped args because the test fakes
# take fewer of them; the production implementation is ``_notify`` above.
notify: Callable[..., None]
# The same two connectors, built from a route's devices instead of the flat 0.9 config.
# Defaulted so every existing CycleDeps(...) construction keeps working; the config-shaped
# pair above goes away with its last consumers in M07.
connect_pve: Callable[[PveDevice], PveClient] = _connect_pve
connect_pbs: Callable[[PbsDevice], PbsClient] = _connect_pbs
# True once the user has asked to stop the in-flight run. Wired by JobService to its own
# cancel event and read live, so the cycle can check it without knowing about the service.
# Default: nothing ever cancels (tests and direct callers that don't care).
backend/app/jobs/recorder.py
+16 -9
View File
@@ -74,17 +74,19 @@ class RunRecorder:
self._session.add(LogEvent(run_id=self.run.id, level=level, message=message))
self._session.commit()
def task_log(self, step: StepName, source: str, lines: list[tuple[int, str]]) -> None:
def task_log(self, step: str, source: str, lines: list[tuple[int, str]]) -> None:
"""Append a batch of raw task-log lines for the live Task-log panel.
``lines`` is a list of ``(line_no, text)`` pairs from the PVE/PBS task tailer;
committed per batch so an in-flight task streams to ``GET /api/tasklog``.
committed per batch so an in-flight task streams to ``GET /api/tasklog``. ``step`` is
the owning :class:`RunStep`'s name — a plain :class:`StepName` works, and so does the
``backup:<pve-id>`` form a multi-source route records.
"""
for line_no, text in lines:
self._session.add(
TaskLogLine(
run_id=self.run.id,
step=step.value,
step=step,
source=source,
line_no=line_no,
text=text,
@@ -95,20 +97,25 @@ class RunRecorder:
# --- steps ---------------------------------------------------------------
@contextmanager
def step(self, name: StepName) -> Iterator[RunStep]:
def step(self, name: StepName, label: str | None = None) -> Iterator[RunStep]:
"""Run a step: persisted RUNNING on entry, SUCCESS on clean exit, FAILURE (and
re-raised) on exception. The yielded row can carry a ``detail`` (e.g. task UPID)."""
step = RunStep(run_id=self.run.id, name=name, status=StepStatus.RUNNING)
re-raised) on exception. The yielded row can carry a ``detail`` (e.g. task UPID).
``label`` qualifies a step that happens more than once in a run — a backup route
records one per source PVE, named ``backup:pve-alpha``.
"""
full = f"{name.value}:{label}" if label else name.value
step = RunStep(run_id=self.run.id, name=full, status=StepStatus.RUNNING)
self._session.add(step)
self._session.commit()
self.log(LogLevel.INFO, f"{name.value}: started")
self.log(LogLevel.INFO, f"{full}: started")
try:
yield step
except Exception as exc:
step.status = StepStatus.FAILURE
step.finished_at = _utcnow()
step.detail = str(exc)
self.log(LogLevel.ERROR, f"{name.value}: {exc}")
self.log(LogLevel.ERROR, f"{full}: {exc}")
self._session.commit()
raise
else:
@@ -118,7 +125,7 @@ class RunRecorder:
# setting step.status = FAILURE and returning normally.
if step.status == StepStatus.RUNNING:
step.status = StepStatus.SUCCESS
self.log(LogLevel.OK, f"{name.value}: done")
self.log(LogLevel.OK, f"{full}: done")
self._session.commit()
def skip_step(self, name: StepName, detail: str | None = None) -> None:
backend/tests/fakes.py
+25 -2
View File
@@ -23,6 +23,9 @@ class UnreachablePve:
def list_guests(self):
raise ConnectorError("connection refused")
def list_cluster_guests(self):
raise ConnectorError("connection refused")
class FakePve:
def __init__(
@@ -34,7 +37,8 @@ class FakePve:
self.guests = guests or []
self.fail_task = fail_task
self.log_lines = log_lines or []
self.vzdump_args: dict | None = None
self.vzdump_args: dict | None = None # the last call (0.9 single-vzdump cycle)
self.vzdump_calls: list[dict] = [] # every call, in order (a route backs up per node)
self.stopped: list[str] = [] # upids passed to stop_task
def __enter__(self) -> FakePve:
@@ -46,6 +50,9 @@ class FakePve:
def list_guests(self) -> list[Guest]:
return self.guests
def list_cluster_guests(self) -> list[Guest]:
return self.guests
def vzdump(
self,
storage,
@@ -55,6 +62,7 @@ class FakePve:
mode="snapshot",
prune_backups=None,
bwlimit=0,
node="",
) -> str:
self.vzdump_args = {
"storage": storage,
@@ -63,8 +71,12 @@ class FakePve:
"mode": mode,
"prune_backups": prune_backups,
"bwlimit": bwlimit,
"node": node,
}
return "UPID:pve:backup"
self.vzdump_calls.append(self.vzdump_args)
# A real UPID names the node the task runs on, which is how the client routes its
# task calls — and how a per-node test tells the tasks apart.
return f"UPID:{node or 'pve'}:backup"
def wait_task(
self, upid: str, poll_interval=None, on_log=None, should_cancel=None, **_
@@ -237,7 +249,16 @@ def make_deps(
pbs_idle: bool | Callable[[], bool] = True,
wol=None,
notify=None,
pves: dict[str, object] | None = None,
pbss: dict[str, object] | None = None,
) -> tuple[CycleDeps, FakePve, FakePbs, FakePower]:
"""Build a :class:`CycleDeps` wired to in-memory fakes.
``pve``/``pbs``/``power`` serve the 0.9 config-shaped builders. ``pves``/``pbss`` map a
*device id* to its fake for the route-shaped ``connect_pve``/``connect_pbs``; a route
test with several sources gives one entry per device, and everything else falls back to
the single fake so no existing caller changes.
"""
pve = pve or FakePve()
pbs = pbs or FakePbs()
power = power or FakePower()
@@ -255,5 +276,7 @@ def make_deps(
wait_reachable=lambda _c, _cancel=None: wait(),
wait_pbs_idle=lambda _c: idle(),
notify=notify or (lambda *_a: None),
connect_pve=lambda device: (pves or {}).get(device.id, pve),
connect_pbs=lambda device: (pbss or {}).get(device.id, pbs),
)
return deps, pve, pbs, power
backend/tests/test_pve.py
+52
View File
@@ -62,6 +62,58 @@ def test_list_guests_merges_and_sorts():
assert guests[1].type == "qemu" and not guests[1].is_ct
def test_list_cluster_guests_tags_each_guest_with_its_node():
def handler(request: httpx.Request) -> httpx.Response:
assert request.url.path.endswith("/cluster/resources")
assert parse_qs(request.url.query.decode())["type"] == ["vm"]
return json_data([
{"vmid": 200, "name": "mail", "type": "qemu", "status": "running", "node": "n2"},
{"vmid": 100, "name": "web", "type": "qemu", "status": "running", "node": "n1"},
{"vmid": 101, "name": "db", "type": "lxc", "status": "stopped", "node": "n1"},
])
guests = make_client(handler).list_cluster_guests()
assert [(g.vmid, g.node) for g in guests] == [(100, "n1"), (101, "n1"), (200, "n2")]
def test_list_cluster_guests_drops_templates():
"""A template is never backed up: counting it would inflate the guest tally, and naming
it in an explicit vmid list would fail the vzdump task."""
def handler(_request: httpx.Request) -> httpx.Response:
return json_data([
{"vmid": 100, "name": "web", "type": "qemu", "status": "running", "node": "n1"},
{"vmid": 900, "name": "tpl", "type": "qemu", "status": "stopped", "node": "n1",
"template": 1},
])
assert [g.vmid for g in make_client(handler).list_cluster_guests()] == [100]
def test_task_calls_follow_the_node_named_in_the_upid():
"""One client drives tasks on several cluster nodes, so the node comes from the task."""
seen: list[str] = []
def handler(request: httpx.Request) -> httpx.Response:
seen.append(request.url.path)
return json_data({"status": "stopped", "exitstatus": "OK"})
client = make_client(handler) # its own node is "pve"
client.task_status("UPID:n3:0001:vzdump::")
assert seen[-1].startswith("/api2/json/nodes/n3/tasks/")
def test_vzdump_runs_on_the_named_node():
seen = {}
def handler(request: httpx.Request) -> httpx.Response:
seen["path"] = request.url.path
return json_data("UPID:n2:0001:vzdump::")
make_client(handler).vzdump(storage="pbs", vmids=[100], node="n2")
assert seen["path"] == "/api2/json/nodes/n2/vzdump"
def test_vzdump_builds_params_and_returns_upid():
captured = {}
backend/tests/test_route_backup.py
+465
View File
@@ -0,0 +1,465 @@
"""The v1.0 backup route cycle: several PVE sources (one a cluster) onto one PBS target.
Everything here runs through the connector fakes — no real PVE/PBS — and the route cycle
itself never touches power: the lease does (see test_lease/test_queue), which is why the
end-to-end cases at the bottom go through ``JobService.enqueue``.
"""
from __future__ import annotations
import time
from fakes import FakeBox, FakePbs, FakePve, UnreachablePve, make_deps
from sqlalchemy import select
from app.config import Config, PbsDevice, PveDevice, Route, RouteGuests, RouteSource
from app.connectors.pve import Guest
from app.core.config_store import ConfigStore
from app.db import session_scope
from app.db.models import (
DatastoreStat,
GuestBackup,
LogEvent,
LogLevel,
Run,
RunKind,
RunStatus,
RunTrigger,
)
from app.jobs.backup_cycle import run_route_backup
from app.jobs.recorder import RunRecorder
from app.jobs.service import JobService
# pve-alpha is a cluster: four guests spread over three nodes. pve-beta is a standalone
# node. Both back up to the same PBS, each through its own storage name.
ALPHA_GUESTS = [
Guest(vmid=100, name="web", type="qemu", status="running", node="n1"),
Guest(vmid=101, name="db", type="lxc", status="running", node="n1"),
Guest(vmid=200, name="mail", type="qemu", status="running", node="n2"),
Guest(vmid=300, name="dns", type="lxc", status="stopped", node="n3"),
]
BETA_GUESTS = [Guest(vmid=500, name="nas", type="qemu", status="running", node="beta")]
def _config(**route_kwargs) -> Config:
config = Config()
config.pves = [
PveDevice(id="pve-alpha", host="192.0.2.10", storages={"pbs1": "pbs-alpha"}),
PveDevice(id="pve-beta", host="192.0.2.11", storages={"pbs1": "pbs-beta"}),
]
config.pbss = [
PbsDevice(id="pbs1", host="192.0.2.20", datastore="backup", mac="00:11:22:33:44:55")
]
sources = route_kwargs.pop(
"sources", [RouteSource(pve="pve-alpha"), RouteSource(pve="pve-beta")]
)
config.routes = [
Route(id="nightly", name="Nightly", kind="backup", target="pbs1", sources=sources,
**route_kwargs)
]
return config
def _deps(*, alpha=None, beta=None, pbs=None, notify=None, cancelled=None):
alpha = alpha if alpha is not None else FakePve(guests=list(ALPHA_GUESTS))
beta = beta if beta is not None else FakePve(guests=list(BETA_GUESTS))
pbs = pbs or FakePbs()
deps, *_ = make_deps(
pves={"pve-alpha": alpha, "pve-beta": beta},
pbss={"pbs1": pbs},
notify=notify,
)
if cancelled is not None:
deps.cancelled = cancelled
return deps, alpha, beta, pbs
def _run(config: Config, deps, route_id: str = "nightly") -> int:
route = next(r for r in config.routes if r.id == route_id)
with RunRecorder(
RunKind.CYCLE, RunTrigger.MANUAL, route_id=route.id, route_name=route.name
) as recorder:
run_route_backup(config, route, recorder, deps)
return recorder.run_id
def _load(run_id: int) -> tuple[str, dict[str, str]]:
"""Return (run status, {step name: step status})."""
with session_scope() as session:
run = session.get(Run, run_id)
return run.status, {s.name: s.status for s in run.steps}
def _logs(run_id: int, level: LogLevel | None = None) -> list[str]:
with session_scope() as session:
rows = session.scalars(select(LogEvent).where(LogEvent.run_id == run_id)).all()
return [r.message for r in rows if level is None or r.level == level]
# --- per-node vzdump grouping ------------------------------------------------
def test_cluster_and_standalone_each_get_one_vzdump_per_node(temp_db):
config = _config()
deps, alpha, beta, _pbs = _deps()
status, steps = _load(_run(config, deps))
assert status == RunStatus.SUCCESS
# The cluster's three nodes, each once; the standalone node once.
assert [c["node"] for c in alpha.vzdump_calls] == ["n1", "n2", "n3"]
assert [c["node"] for c in beta.vzdump_calls] == ["beta"]
# Each PVE names the same PBS its own way.
assert {c["storage"] for c in alpha.vzdump_calls} == {"pbs-alpha"}
assert beta.vzdump_calls[0]["storage"] == "pbs-beta"
assert steps["backup:pve-alpha"] == "success"
assert steps["backup:pve-beta"] == "success"
def test_all_mode_uses_vzdumps_own_all_flag(temp_db):
config = _config()
deps, alpha, _beta, _pbs = _deps()
_run(config, deps)
# Not the vmids we just listed: PVE keeps deciding, so a guest excluded from backup on
# the host is still honoured.
assert all(c["all_guests"] is True and c["vmids"] is None for c in alpha.vzdump_calls)
def test_include_mode_groups_the_selection_per_node(temp_db):
config = _config(
sources=[
RouteSource(pve="pve-alpha", guests=RouteGuests(mode="include", list=[100, 300])),
RouteSource(pve="pve-beta"),
]
)
deps, alpha, _beta, _pbs = _deps()
status, _steps = _load(_run(config, deps))
assert status == RunStatus.SUCCESS
# n2 holds nothing selected, so it is never woken up with a task.
assert [(c["node"], c["vmids"]) for c in alpha.vzdump_calls] == [("n1", [100]), ("n3", [300])]
assert all(c["all_guests"] is False for c in alpha.vzdump_calls)
def test_route_options_and_retention_drive_the_vzdump_arguments(temp_db):
config = _config()
route = config.routes[0]
route.options.mode = "stop"
route.options.bwlimit = 50_000
route.retention.keep_last = 3
route.retention.keep_daily = 0
route.retention.keep_weekly = 0
route.retention.keep_monthly = 0
deps, alpha, _beta, _pbs = _deps()
_run(config, deps)
call = alpha.vzdump_calls[0]
assert call["mode"] == "stop"
assert call["bwlimit"] == 50_000
assert call["prune_backups"] == "keep-last=3"
def test_selected_guest_that_is_gone_is_warned_about_and_skipped(temp_db):
config = _config(
sources=[RouteSource(pve="pve-alpha", guests=RouteGuests(mode="include", list=[100, 999]))]
)
deps, alpha, _beta, _pbs = _deps()
run_id = _run(config, deps)
status, _steps = _load(run_id)
assert status == RunStatus.SUCCESS
assert [c["vmids"] for c in alpha.vzdump_calls] == [[100]]
assert any("999" in m for m in _logs(run_id, LogLevel.WARN))
def test_source_left_with_no_guest_fails_that_source(temp_db):
config = _config(
sources=[RouteSource(pve="pve-alpha", guests=RouteGuests(mode="include", list=[999]))]
)
deps, alpha, _beta, _pbs = _deps()
status, steps = _load(_run(config, deps))
assert status == RunStatus.FAILURE
assert steps["backup:pve-alpha"] == "failure"
assert alpha.vzdump_calls == []
# --- per-source failure isolation --------------------------------------------
def test_one_broken_source_still_lets_the_others_run(temp_db):
config = _config()
deps, alpha, _beta, _pbs = _deps(beta=UnreachablePve())
run_id = _run(config, deps)
status, steps = _load(run_id)
assert status == RunStatus.FAILURE
assert steps["backup:pve-alpha"] == "success"
assert steps["backup:pve-beta"] == "failure"
assert len(alpha.vzdump_calls) == 3 # the healthy source got its full backup
with session_scope() as session:
assert "pve-beta" in session.get(Run, run_id).error
def test_a_failing_node_task_fails_only_its_source(temp_db):
config = _config()
deps, _alpha, _beta, _pbs = _deps(beta=FakePve(guests=list(BETA_GUESTS), fail_task=True))
status, steps = _load(_run(config, deps))
assert status == RunStatus.FAILURE
assert steps["backup:pve-alpha"] == "success"
assert steps["backup:pve-beta"] == "failure"
assert steps["gc"] == "success" # the PBS is awake and the other source's data is real
# --- guest tally --------------------------------------------------------------
def _summary(config, deps_kwargs=None):
seen: dict = {}
def notify(_config, _run, datastore, guests, _next_at=None):
seen["datastore"] = datastore
seen["guests"] = guests
deps, alpha, beta, pbs = _deps(notify=notify, **(deps_kwargs or {}))
_run(config, deps)
return seen, alpha, beta, pbs
def test_guest_tally_aggregates_across_sources(temp_db):
seen, _alpha, _beta, _pbs = _summary(_config())
assert seen["guests"].total == 5 # 4 on the cluster + 1 standalone
assert seen["guests"].ok == 5
assert seen["guests"].failed == []
def test_failed_guests_are_named_across_sources(temp_db):
alpha = FakePve(
guests=list(ALPHA_GUESTS),
fail_task=True,
log_lines=["INFO: Finished Backup of VM 100", "ERROR: Backup of VM 101 failed - boom"],
)
seen, _alpha, _beta, _pbs = _summary(_config(), {"alpha": alpha})
assert seen["guests"].failed == ["db"] # the guest's name, not its vmid
# 100 from the failed task's own log (the tally survives the raise) + pve-beta's guest,
# which still ran because sources are isolated.
assert seen["guests"].ok == 2
# --- GC / verify / preflight --------------------------------------------------
def test_gc_runs_once_per_route(temp_db):
config = _config()
deps, _alpha, _beta, pbs = _deps()
_load(_run(config, deps))
assert pbs.gc_started is True
def test_gc_disabled_skips_the_step(temp_db):
config = _config()
config.routes[0].options.gc = False
deps, _alpha, _beta, pbs = _deps()
status, steps = _load(_run(config, deps))
assert status == RunStatus.SUCCESS
assert steps["gc"] == "skipped"
assert pbs.gc_started is False
def test_verify_after_runs_only_the_new_snapshots(temp_db):
config = _config()
config.routes[0].options.verify_after = True
deps, _alpha, _beta, pbs = _deps()
status, steps = _load(_run(config, deps))
assert status == RunStatus.SUCCESS
assert steps["verify"] == "success"
assert pbs.verify_args == {"ignore_verified": True, "outdated_after": None}
def test_verify_is_skipped_by_default(temp_db):
_status, steps = _load(_run(_config(), _deps()[0]))
assert steps["verify"] == "skipped"
def test_preflight_aborts_before_any_backup_when_the_datastore_is_full(temp_db):
config = _config()
config.routes[0].options.min_free_percent = 50
deps, alpha, _beta, _pbs = _deps(
pbs=FakePbs(total=8_000_000_000, used=7_000_000_000, avail=1_000_000_000)
)
status, steps = _load(_run(config, deps))
assert status == RunStatus.ABORTED
assert steps["precheck"] == "failure"
assert alpha.vzdump_calls == []
def test_preflight_passes_when_there_is_room(temp_db):
config = _config()
config.routes[0].options.min_free_percent = 50
deps, alpha, _beta, _pbs = _deps()
status, steps = _load(_run(config, deps))
assert status == RunStatus.SUCCESS
assert steps["precheck"] == "success"
assert len(alpha.vzdump_calls) == 3
def test_no_precheck_step_when_the_guard_is_off(temp_db):
_status, steps = _load(_run(_config(), _deps()[0]))
assert "precheck" not in steps
# --- caches -------------------------------------------------------------------
def test_last_backup_cache_is_attributed_to_each_source(temp_db):
config = _config()
# The datastore holds snapshots for guests of both PVEs; each source claims its own.
deps, _alpha, _beta, _pbs = _deps(
pbs=FakePbs(snapshots={100: 1_700_000_000, 500: 1_700_000_100})
)
_run(config, deps)
with session_scope() as session:
rows = {(r.pve_id, r.vmid, r.pbs_id) for r in session.scalars(select(GuestBackup))}
assert rows == {("pve-alpha", 100, "pbs1"), ("pve-beta", 500, "pbs1")}
def test_datastore_stat_is_keyed_by_the_target_pbs(temp_db):
config = _config()
deps, _alpha, _beta, _pbs = _deps()
_run(config, deps)
with session_scope() as session:
row = session.get(DatastoreStat, ("pbs1", "backup"))
assert row is not None
assert (row.total, row.used) == (8_000_000_000, 2_000_000_000)
def test_a_cache_failure_never_fails_the_run(temp_db):
config = _config()
class NoSnapshots(FakePbs):
def latest_backups(self):
raise RuntimeError("snapshot listing exploded")
deps, _alpha, _beta, _pbs = _deps(pbs=NoSnapshots())
status, _steps = _load(_run(config, deps))
assert status == RunStatus.SUCCESS
# --- cancellation -------------------------------------------------------------
def test_cancel_stops_the_running_vzdump_and_aborts(temp_db):
config = _config()
notified: list[object] = []
# False at the between-sources check, then True from inside the first task's wait.
answers = iter([False])
deps, alpha, beta, _pbs = _deps(
cancelled=lambda: next(answers, True), notify=lambda *a: notified.append(a)
)
status, steps = _load(_run(config, deps))
assert status == RunStatus.ABORTED
assert alpha.stopped == ["UPID:n1:backup"] # the task it had actually started
assert steps["backup:pve-alpha"] == "failure"
assert "backup:pve-beta" not in steps # a cancel stops the route, not just one source
assert beta.vzdump_calls == []
assert notified == [] # the user is standing at the UI; no push about their own click
def test_cancel_before_the_first_source_starts_nothing(temp_db):
config = _config()
deps, alpha, _beta, _pbs = _deps(cancelled=lambda: True)
status, steps = _load(_run(config, deps))
assert status == RunStatus.ABORTED
assert alpha.vzdump_calls == []
assert steps == {}
# --- end to end through the queue (the lease owns the power) ------------------
def _service(config_setup) -> tuple[JobService, FakeBox, dict]:
store = ConfigStore.load_or_create()
template = _config()
store.config.pves = template.pves
store.config.pbss = template.pbss
store.config.routes = template.routes
fakes = {"alpha": FakePve(guests=list(ALPHA_GUESTS)), "beta": config_setup}
deps, *_ = make_deps(
pves={"pve-alpha": fakes["alpha"], "pve-beta": fakes["beta"]},
pbss={"pbs1": FakePbs()},
)
# Found asleep, up after the first magic packet — so the wake is observable.
box = FakeBox(reachable=[False, True])
return JobService(store, deps=deps, lease_deps=box.deps()), box, fakes
def _enqueue_and_drain(service: JobService) -> None:
def job(config, recorder, deps) -> None:
route = next(r for r in config.routes if r.id == "nightly")
run_route_backup(config, route, recorder, deps)
service.enqueue("nightly", RunTrigger.MANUAL, job)
deadline = time.monotonic() + 5
while time.monotonic() < deadline:
if service.current() is None and not service.pending() and not service.is_running:
return
time.sleep(0.01)
raise AssertionError("queue did not drain")
def test_a_route_run_wakes_the_target_once_and_powers_it_off_once(temp_config, temp_db):
service, box, _fakes = _service(FakePve(guests=list(BETA_GUESTS)))
_enqueue_and_drain(service)
assert box.wol == ["pbs1"]
assert box.poweroffs == ["pbs1"]
with session_scope() as session:
run = session.scalars(select(Run)).one()
assert run.status == RunStatus.SUCCESS
assert run.route_id == "nightly"
def test_a_failed_source_leaves_the_pbs_on(temp_config, temp_db):
service, box, fakes = _service(UnreachablePve())
_enqueue_and_drain(service)
assert box.wol == ["pbs1"]
assert box.poweroffs == [] # left on for inspection
assert len(fakes["alpha"].vzdump_calls) == 3
with session_scope() as session:
assert session.scalars(select(Run)).one().status == RunStatus.FAILURE