From e5be5d4f71d0db1777a4dfc4255fd8f6381c0ea2 Mon Sep 17 00:00:00 2001 From: catborise Date: Tue, 13 Oct 2026 17:56:24 +0300 Subject: [PATCH] feat(operations): kind contract, job context and worker lease --- operations/jobs.py | 102 ++++++++++++++++++++++++++++++++- operations/models.py | 51 ++++++++++++++++- operations/tests/fakes.py | 72 +++++++++++++++++++++++ operations/tests/test_kinds.py | 69 ++++++++++++++++++++++ 4 files changed, 290 insertions(+), 4 deletions(-) create mode 100644 operations/tests/fakes.py create mode 100644 operations/tests/test_kinds.py diff --git a/operations/jobs.py b/operations/jobs.py index fd822fbc..19045297 100644 --- a/operations/jobs.py +++ b/operations/jobs.py @@ -1,8 +1,41 @@ -"""Kinds of jobs by name (filled by each app's jobs.py via register).""" +"""The contract every kind of job follows (spec section 5), the kinds by +name (filled by each app's jobs.py via register) and the context a running +job reports through.""" + +from dataclasses import dataclass, field + +from . import utils +from .models import LeaseLost, Operation, fenced KINDS = {} +@dataclass +class Plan: + claims: dict + resource: str + label: str + intent: dict = field(default_factory=dict) + + +@dataclass +class Outcome: + """What a run or an inspection decided. RUNNING: libvirt still works; + stalled: its work reports nothing left to do but does not end; + progress: (done, total) of that work, when libvirt tells.""" + + state: str + result: dict = field(default_factory=dict) + error: str = "" + phase: str = "" + stalled: bool = False + progress: tuple | None = None + + +class Cancelled(Exception): + """Raised by a kind's run when a cancel stopped its work.""" + + def get_kind(name): return KINDS[name] @@ -10,3 +43,70 @@ def get_kind(name): def label_of(name): kind = KINDS.get(name) return kind.label if kind is not None else name + + +def register(cls): + KINDS[cls.name] = cls() + return cls + + +class Kind: + name = "" + label = "" + cancellable = False + time_limit = None + progress_bytes = False # progress counts bytes: the job page shows sizes + + def normalize(self, raw): + """Plain, allowlisted values from the request, without libvirt.""" + raise NotImplementedError + + def validate(self, user, params): + """In the request: permissions and libvirt state. Raise to refuse.""" + + def plan(self, user, params): + raise NotImplementedError + + def precheck(self, op, ctx): + """In the worker, before any effect: permission and state again.""" + + def run(self, op, ctx): + raise NotImplementedError + + def progress(self, op): + return None + + def cancel(self, op): + raise NotImplementedError + + def reconcile(self, op): + raise NotImplementedError + + def log_entry(self, op, state): + """(host, instance, message) of the Logs row written with each end of + the job (in the same transaction); its host and instance also go on + the row of a resolution. None: no row.""" + return None + + +class Context: + def __init__(self, op, worker): + self.op = op + self.worker = worker + self.effects_started = False + + def _write(self, **fields): + with fenced(self.worker.identity): + if not Operation.objects.filter(pk=self.op.pk, worker=self.worker.identity).update(**fields): + raise LeaseLost("the job is no longer this worker's") + for name, value in fields.items(): + setattr(self.op, name, value) + + def record_intent(self, **values): + """Commit what run is about to change; the first libvirt effect may + follow only after this returns.""" + self._write(intent={**self.op.intent, **values}, effects_started=True) + self.effects_started = True + + def set_phase(self, text): + self._write(phase=utils.safe_text(text)[:255]) diff --git a/operations/models.py b/operations/models.py index e27b337b..cffb4ac9 100644 --- a/operations/models.py +++ b/operations/models.py @@ -1,10 +1,18 @@ -"""Long-running operations: the queue, the history and the resources jobs -hold (spec 2026-10-13, sections 1-2).""" +"""Long-running operations: the queue, the history and the resources jobs hold +(spec 2026-10-13, sections 1-2). One worker per installation, and fencing for +every write a worker makes. Same host (the same data directory, which holds +the worker's flock): the flock proves the previous holder is gone, so the +lease is taken over at once. Another host: only after a stale heartbeat.""" import uuid +from contextlib import contextmanager +from datetime import timedelta from django.conf import settings -from django.db import models +from django.db import models, transaction +from django.utils import timezone + +TAKEOVER_AFTER = timedelta(minutes=2) class Operation(models.Model): @@ -74,3 +82,40 @@ class WorkerLease(models.Model): holder = models.CharField(max_length=128) host = models.CharField(max_length=255) heartbeat_at = models.DateTimeField() + + +class LeaseLost(Exception): + """This worker no longer holds the lease or owns the job.""" + + +def acquire(holder, host): + now = timezone.now() + with transaction.atomic(): + lease, created = WorkerLease.objects.select_for_update().get_or_create( + name="worker", defaults={"holder": holder, "host": host, "heartbeat_at": now} + ) + if created or lease.holder == holder: + return True + if lease.host == host or lease.heartbeat_at < now - TAKEOVER_AFTER: + WorkerLease.objects.filter(name="worker").update(holder=holder, host=host, heartbeat_at=now) + return True + return False + + +def renew(holder): + return bool(WorkerLease.objects.filter(name="worker", holder=holder).update(heartbeat_at=timezone.now())) + + +def last_seen(): + lease = WorkerLease.objects.filter(name="worker").first() + return lease.heartbeat_at if lease else None + + +@contextmanager +def fenced(holder): + """A transaction in which holder still holds the lease; a takeover locks + the same row, so it waits for this transaction or makes it fail.""" + with transaction.atomic(): + if not WorkerLease.objects.select_for_update().filter(name="worker", holder=holder).exists(): + raise LeaseLost(holder) + yield diff --git a/operations/tests/fakes.py b/operations/tests/fakes.py new file mode 100644 index 00000000..db547013 --- /dev/null +++ b/operations/tests/fakes.py @@ -0,0 +1,72 @@ +"""Kinds used by the operations tests; registered on import. test.scripted +takes its behaviour from params and from barriers that tests install before +submitting (keyed by submission key).""" + +from operations.jobs import Cancelled, Kind, Outcome, Plan, register +from operations.models import Claim, Operation + +BARRIERS = {} # submission key -> threading.Event the run waits on +CANCELLED = [] + + +def reset(): + BARRIERS.clear() + CANCELLED.clear() + + +@register +class NoopKind(Kind): + name = "test.noop" + label = "No-op" + + def normalize(self, raw): + return {"uuid": str(raw.get("uuid", "u1"))} + + def plan(self, user, params): + key = "instance:" + params["uuid"] + return Plan(claims={key: Claim.EXCLUSIVE}, resource=key, label="c/vm") + + def run(self, op, ctx): + ctx.record_intent(step="effect") + return {"done": True} + + def reconcile(self, op): + return Outcome(Operation.SUCCEEDED, result={"recovered": True}) + + def log_entry(self, op, state): + return ("c", "vm", f"No-op {state}") + + +@register +class ScriptedKind(NoopKind): + """params: fail_before, fail_after, outcome (what reconcile returns).""" + + name = "test.scripted" + label = "Scripted" + cancellable = True + + def normalize(self, raw): + return {"uuid": str(raw.get("uuid", "u1")), "fail_before": bool(raw.get("fail_before")), + "fail_after": bool(raw.get("fail_after")), "outcome": str(raw.get("outcome", "uncertain"))} + + def run(self, op, ctx): + if op.params["fail_before"]: + raise RuntimeError("refused before effects") + ctx.record_intent(step="effect") + if op.params["fail_after"]: + raise RuntimeError("failed after effects") + barrier = BARRIERS.get(op.submission_key) + if barrier is not None: + barrier.wait(30) + if str(op.id) in CANCELLED: + raise Cancelled() + return {"ran": True} + + def cancel(self, op): + CANCELLED.append(str(op.id)) + barrier = BARRIERS.get(op.submission_key) + if barrier is not None: + barrier.set() + + def reconcile(self, op): + return Outcome(op.params["outcome"], error="scripted") diff --git a/operations/tests/test_kinds.py b/operations/tests/test_kinds.py new file mode 100644 index 00000000..2f5fe71e --- /dev/null +++ b/operations/tests/test_kinds.py @@ -0,0 +1,69 @@ +from datetime import timedelta +from types import SimpleNamespace + +from django.contrib.auth import get_user_model +from django.test import TestCase +from django.utils import timezone + +from operations.jobs import KINDS, Context, get_kind, label_of +from operations.models import TAKEOVER_AFTER, LeaseLost, Operation, WorkerLease, acquire, fenced, renew +from operations.tests import fakes # noqa: F401 + + +class RegistryTestCase(TestCase): + def test_kinds_register_by_name(self): + self.assertIs(get_kind("test.noop"), KINDS["test.noop"]) + self.assertEqual((label_of("test.noop"), label_of("unknown.kind")), ("No-op", "unknown.kind")) + + +class LeaseTestCase(TestCase): + def test_the_first_worker_takes_the_lease_and_renews_it(self): + self.assertTrue(acquire("h1:1:a", "h1")) + self.assertTrue(renew("h1:1:a")) + self.assertFalse(renew("h1:2:b")) + + def test_the_same_host_takes_over_at_once(self): + acquire("h1:1:a", "h1") + self.assertTrue(acquire("h1:2:b", "h1")) # the caller holds the host's flock + self.assertFalse(renew("h1:1:a")) + + def test_another_host_waits_until_the_lease_is_stale(self): + acquire("h1:1:a", "h1") + self.assertFalse(acquire("h2:1:c", "h2")) + WorkerLease.objects.update(heartbeat_at=timezone.now() - TAKEOVER_AFTER - timedelta(seconds=1)) + self.assertTrue(acquire("h2:1:c", "h2")) + + def test_fenced_refuses_a_worker_without_the_lease(self): + acquire("h1:1:a", "h1") + with fenced("h1:1:a"): + pass + with self.assertRaises(LeaseLost): + with fenced("h9:9:z"): + pass + + +class ContextTestCase(TestCase): + def setUp(self): + user = get_user_model().objects.create_user("ctx_user", "x@example.com", "pw") + acquire("h1:1:a", "h1") + self.worker = SimpleNamespace(identity="h1:1:a") + self.op = Operation.objects.create(kind="test.noop", user=user, username="ctx_user", submission_key="k", + resource="r", resource_label="r", state=Operation.RUNNING, + worker="h1:1:a") + + def test_intent_is_committed_at_once(self): + ctx = Context(self.op, self.worker) + self.assertFalse(ctx.effects_started) + ctx.record_intent(path="/pool/x.qcow2") + self.op.refresh_from_db() + self.assertTrue(ctx.effects_started and self.op.effects_started) + self.assertEqual(self.op.intent, {"path": "/pool/x.qcow2"}) + + def test_a_worker_without_the_lease_records_nothing(self): + WorkerLease.objects.update(holder="h2:1:b", host="h2") + ctx = Context(self.op, self.worker) + with self.assertRaises(LeaseLost): + ctx.record_intent(path="/pool/x.qcow2") + self.op.refresh_from_db() + self.assertEqual((self.op.intent, self.op.effects_started), ({}, False)) + self.assertFalse(ctx.effects_started)