feat(operations): recover interrupted jobs from libvirt evidence

operations/tests/test_recovery.py

@@ -0,0 +1,197 @@
+from concurrent.futures import ThreadPoolExecutor
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.test import TransactionTestCase
+
+from operations import jobs
+from operations.jobs import Outcome
+from operations.models import Claim, Operation, WorkerLease, acquire
+from operations.tests import fakes
+from operations.worker import ABANDONED, STALL_LIMIT, STALLED, WAITING, Scheduler
+
+
+class FakeWorker:
+    host = "this-host"
+    identity = "this-host:2:new"
+    lost_lease = False
+
+    def __init__(self):
+        self.control = ThreadPoolExecutor(1)
+
+
+class RecoveryTestCase(TransactionTestCase):
+    def setUp(self):
+        fakes.reset()
+        self.user = get_user_model().objects.create_user("rec_user", "r@example.com", "pw")
+        self.worker = FakeWorker()
+        acquire(self.worker.identity, self.worker.host)
+        self.scheduler = Scheduler(self.worker)
+
+    def tearDown(self):
+        self.worker.control.shutdown(wait=True)
+
+    def orphan(self, params, worker="this-host:1:old", effects=True):
+        op = jobs.submit("test.scripted", self.user, params, "key-00000001")
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, worker=worker,
+                                                  intent={"step": "effect"}, effects_started=effects)
+        return op
+
+    def run_due(self):
+        self.scheduler.schedule_orphans()
+        self.scheduler.due()
+        self.worker.control.shutdown(wait=True)
+        self.worker.control = ThreadPoolExecutor(1)
+
+    def reload(self, op):
+        op.refresh_from_db()
+        return op
+
+    def test_a_proven_success_is_released(self):
+        op = self.orphan({"outcome": "succeeded"})
+        self.run_due()
+        self.assertEqual(self.reload(op).state, Operation.SUCCEEDED)
+        self.assertFalse(Claim.objects.exists())
+
+    def test_a_proven_success_does_not_carry_the_raised_error(self):
+        op = self.orphan({}, worker=self.worker.identity)
+        self.scheduler.inspect_now(op.pk, error="failed after effects")
+        with patch.object(fakes.ScriptedKind, "reconcile", return_value=Outcome(Operation.SUCCEEDED)):
+            self.run_due()
+        op = self.reload(op)
+        self.assertEqual((op.state, op.error), (Operation.SUCCEEDED, ""))
+
+    def test_an_uncertain_outcome_keeps_the_error_the_job_raised(self):
+        op = self.orphan({}, worker=self.worker.identity)
+        self.scheduler.inspect_now(op.pk, error="resource busy: X")
+        with patch.object(fakes.ScriptedKind, "reconcile",
+                          return_value=Outcome(Operation.UNCERTAIN, error="check the VM")):
+            self.run_due()
+        op = self.reload(op)
+        self.assertEqual((op.state, op.error), (Operation.UNCERTAIN, "resource busy: X. check the VM"))
+
+    def test_uncertain_keeps_the_claims(self):
+        op = self.orphan({"outcome": "uncertain"})
+        self.run_due()
+        self.assertEqual(self.reload(op).state, Operation.UNCERTAIN)
+        self.assertTrue(Claim.objects.filter(operation=op).exists())
+
+    def test_still_running_in_libvirt_keeps_watching(self):
+        op = self.orphan({"outcome": "running"})
+        self.run_due()
+        op = self.reload(op)
+        self.assertEqual((op.state, op.worker), (Operation.RUNNING, self.worker.identity))
+        self.assertIn(op.pk, self.scheduler.next_try)
+
+    def run_at(self, now, outcome):
+        with patch("operations.worker.time.monotonic", return_value=now), \
+                patch.object(fakes.ScriptedKind, "reconcile", return_value=outcome):
+            self.run_due()
+
+    def test_watching_stores_the_progress_libvirt_reports(self):
+        op = self.orphan({})
+        self.run_at(1000, Outcome(Operation.RUNNING, phase="merging", progress=(3, 8)))
+        op = self.reload(op)
+        self.assertEqual((op.phase, op.progress_done, op.progress_total), ("merging", 3, 8))
+        self.scheduler.next_try[op.pk] = 0
+        self.run_at(1015, Outcome(Operation.RUNNING, progress=(5, 8)))
+        op = self.reload(op)
+        self.assertEqual((op.phase, op.progress_done, op.progress_total), ("merging", 5, 8))
+
+    def test_watching_writes_no_progress_without_the_lease(self):
+        op = self.orphan({}, worker=self.worker.identity)
+        WorkerLease.objects.update(holder="other-host:9:z")
+        self.run_at(1000, Outcome(Operation.RUNNING, progress=(3, 8)))
+        self.assertIsNone(self.reload(op).progress_done)
+
+    def test_a_stalled_merge_is_watched_until_the_stall_limit(self):
+        op = self.orphan({})
+        self.run_at(1000, Outcome(Operation.RUNNING, stalled=True))
+        self.run_at(1000 + STALL_LIMIT - 1, Outcome(Operation.RUNNING, stalled=True))
+        self.assertEqual(self.reload(op).state, Operation.RUNNING)
+        self.assertIn(op.pk, self.scheduler.next_try)
+
+    def test_a_merge_stalled_past_the_limit_ends_uncertain_and_keeps_the_claims(self):
+        op = self.orphan({})
+        self.run_at(1000, Outcome(Operation.RUNNING, stalled=True))
+        self.run_at(1000 + STALL_LIMIT, Outcome(Operation.RUNNING, stalled=True))
+        op = self.reload(op)
+        self.assertEqual((op.state, op.error), (Operation.UNCERTAIN, STALLED))
+        self.assertTrue(Claim.objects.filter(operation=op).exists())
+        self.assertNotIn(op.pk, self.scheduler.stalled_since)
+
+    def test_a_stalled_merge_keeps_the_error_the_job_raised(self):
+        op = self.orphan({}, worker=self.worker.identity)
+        self.scheduler.inspect_now(op.pk, error="resource busy: X")
+        self.run_at(1000, Outcome(Operation.RUNNING, stalled=True))
+        self.scheduler.next_try[op.pk] = 0
+        self.run_at(1000 + STALL_LIMIT, Outcome(Operation.RUNNING, stalled=True))
+        self.assertEqual(self.reload(op).error, f"resource busy: X. {STALLED}")
+
+    def test_a_raised_error_ending_with_a_dot_is_not_doubled(self):
+        op = self.orphan({}, worker=self.worker.identity)
+        self.scheduler.inspect_now(op.pk, error="refused.")
+        with patch.object(fakes.ScriptedKind, "reconcile", return_value=Outcome(Operation.UNCERTAIN, error="check")):
+            self.run_due()
+        self.assertEqual(self.reload(op).error, "refused. check")
+
+    def test_an_outcome_without_an_error_stores_the_raised_one(self):
+        op = self.orphan({}, worker=self.worker.identity)
+        self.scheduler.inspect_now(op.pk, error="resource busy: X")
+        with patch.object(fakes.ScriptedKind, "reconcile", return_value=Outcome(Operation.FAILED)):
+            self.run_due()
+        self.assertEqual(self.reload(op).error, "resource busy: X")
+
+    def test_progress_resets_the_stall_timer(self):
+        op = self.orphan({})
+        self.run_at(1000, Outcome(Operation.RUNNING, stalled=True))
+        self.run_at(1300, Outcome(Operation.RUNNING))
+        self.run_at(1400, Outcome(Operation.RUNNING, stalled=True))
+        self.run_at(1000 + STALL_LIMIT + 100, Outcome(Operation.RUNNING, stalled=True))
+        self.assertEqual(self.reload(op).state, Operation.RUNNING)
+
+    def test_an_unreachable_host_waits_with_growing_intervals(self):
+        op = self.orphan({})
+        with patch.object(fakes.ScriptedKind, "reconcile", side_effect=OSError("host down")):
+            self.run_due()
+        op = self.reload(op)
+        self.assertEqual((op.state, op.phase), (Operation.RUNNING, WAITING))
+        self.assertEqual(self.scheduler.attempts[op.pk], 1)
+
+    def test_a_job_that_never_recorded_its_intent_fails(self):
+        op = self.orphan({}, effects=False)
+        self.run_due()
+        self.assertEqual(self.reload(op).state, Operation.FAILED)
+
+    def test_another_hosts_orphan_ends_uncertain_without_inspection(self):
+        op = self.orphan({"outcome": "succeeded"}, worker="other-host:1:old")
+        with patch.object(fakes.ScriptedKind, "reconcile") as reconcile, \
+                self.assertLogs("operations.worker", "WARNING") as logs:
+            self.run_due()
+        reconcile.assert_not_called()
+        op = self.reload(op)
+        # the stored error is shown to the job's owner: the host goes to the worker log only
+        self.assertEqual((op.state, op.error), (Operation.UNCERTAIN, ABANDONED))
+        self.assertNotIn("other-host", op.error)
+        self.assertIn("other-host", "\n".join(logs.output))
+        self.assertTrue(Claim.objects.filter(operation=op).exists())
+
+    def test_another_hosts_orphan_stays_theirs_if_it_cannot_end_uncertain(self):
+        op = self.orphan({"outcome": "succeeded"}, worker="other-host:1:old")
+        with patch.object(jobs, "finish", side_effect=RuntimeError("database is down")):
+            self.run_due()
+        op = self.reload(op)
+        self.assertEqual((op.state, op.worker), (Operation.RUNNING, "other-host:1:old"))
+
+    def test_this_workers_own_jobs_are_not_orphans(self):
+        op = self.orphan({"outcome": "succeeded"}, worker=self.worker.identity)
+        self.run_due()
+        self.assertEqual(self.reload(op).state, Operation.RUNNING)
+
+    def test_an_own_job_that_raised_before_its_effects_fails_and_releases(self):
+        op = self.orphan({"outcome": "uncertain"}, worker=self.worker.identity, effects=False)
+        self.scheduler.inspect_now(op.pk, error="refused before effects")
+        self.run_due()
+        op = self.reload(op)
+        self.assertEqual(op.state, Operation.FAILED)
+        self.assertFalse(Claim.objects.exists())

operations/tests/test_worker.py

@@ -200,12 +200,13 @@ class WorkerTestCase(TransactionTestCase):
         self.worker._drain()
         self.assertEqual(state(op), Operation.CANCELLED)
 
-    def test_a_job_thread_that_raised_unfinished_goes_to_inspection(self):
+    def test_a_job_thread_that_raised_before_its_effects_fails(self):
         op = self.submit("test.noop", {}, "key-00000001")
         with patch("operations.jobs.Context", side_effect=RuntimeError("no context")):
             self.start()
-            self.assertTrue(wait_for(lambda: state(op) == Operation.UNCERTAIN))
+            self.assertTrue(wait_for(lambda: state(op) == Operation.FAILED))
         self.assertIn("no context", Operation.objects.get(pk=op.pk).error)
+        self.assertFalse(Claim.objects.filter(operation=op).exists())
 
     def test_a_failing_manager_loop_ends_the_process(self):
         directory = tempfile.TemporaryDirectory()

operations/worker.py

@@ -1,8 +1,8 @@
 """The job worker (spec section 3). The manager loop never calls libvirt: job
 threads run kinds; control threads run cancel hooks and recovery; progress
-threads read progress. Every write is fenced by the lease. Recovery of jobs
-the worker cannot finish itself. Minimal: a job that raised after its effects
-started goes straight to inspection (uncertain)."""
+threads read progress. Every write is fenced by the lease. Recovery of
+interrupted jobs (spec section 4): inspection decides from intent and libvirt;
+nothing is re-run; uncertain keeps its claims."""
 
 import fcntl
 import logging
@@ -19,6 +19,7 @@ from pathlib import Path
 from django.conf import settings
 from django.db import close_old_connections
 from django.utils import timezone
+from django.utils.translation import gettext_noop
 
 from . import claims, jobs, utils
 from .models import Claim, LeaseLost, Operation, acquire, fenced, renew
@@ -29,6 +30,16 @@ MAX_SLOTS = 16
 PROGRESS_THREADS = 2
 HOUSEKEEPING_EVERY = 3600.0
 
+BACKOFF = (30, 60, 120, 300, 600)
+WATCH_EVERY = 15
+STALL_LIMIT = 600  # a block job left at cur == end after a libvirtd restart never ends
+STALLED = ("libvirt's merge finished copying but never completed (for example after a libvirtd restart); "
+           "check the VM before resolving")
+ABANDONED = ("The worker that ran this job stopped responding; an administrator must make sure it is no longer "
+             "running, then resolve this job")
+WAITING = gettext_noop("Waiting for recovery")
+ACTIVE = (Operation.RUNNING, Operation.CANCELLING)
+
 
 def data_dir():
     """The installation's persistent data directory (the Docker volume)."""
@@ -298,13 +309,107 @@ def housekeeping():
 class Scheduler:
     def __init__(self, worker):
         self.worker = worker
+        self.next_try = {}
+        self.attempts = {}
+        self.errors = {}
+        self.stalled_since = {}
+        self.inflight = set()
 
     def schedule_orphans(self):
-        pass
-
-    def due(self):
-        pass
+        for op_id in (Operation.objects.filter(state__in=ACTIVE).exclude(worker=self.worker.identity)
+                      .values_list("pk", flat=True)):
+            self.next_try.setdefault(op_id, 0)
 
     def inspect_now(self, op_id, error=""):
-        op = Operation.objects.get(pk=op_id)
-        jobs.finish(op, jobs.Outcome(Operation.UNCERTAIN, error=error), worker=self.worker)
+        self.errors[op_id] = error
+        self.next_try[op_id] = 0
+
+    def due(self):
+        now = time.monotonic()
+        for op_id, at in list(self.next_try.items()):
+            if at <= now and op_id not in self.inflight:
+                self.inflight.add(op_id)
+                self.worker.control.submit(self._inspect, op_id)
+
+    def _done(self, op_id):
+        for table in (self.next_try, self.attempts, self.errors, self.stalled_since):
+            table.pop(op_id, None)
+
+    def _take_over(self, op):
+        """True if op is now this worker's. Another host's job ends uncertain
+        in the same transaction, so a failed finish leaves its owner as it was."""
+        previous = op.worker
+        previous_host = previous.split(":", 1)[0]
+        with fenced(self.worker.identity):
+            if not (Operation.objects.filter(pk=op.pk, worker=previous, state__in=ACTIVE)
+                    .update(worker=self.worker.identity)):
+                self._done(op.pk)
+                return False
+            op.refresh_from_db()
+            if previous_host != self.worker.host:
+                # the error is shown to the job's owner: the host goes to the worker log only
+                self._finish(op, jobs.Outcome(Operation.UNCERTAIN, error=ABANDONED))
+                logger.warning("job %s ended uncertain: its worker %s stopped responding", op.pk, previous)
+                return False
+        return True
+
+    def _finish(self, op, outcome):
+        jobs.finish(op, outcome, worker=self.worker)
+        self._done(op.pk)
+
+    def _error(self, op_id, finding):
+        """What an unproven end stores: the error the job raised, then what libvirt shows."""
+        raised = self.errors.get(op_id, "")
+        return f"{raised.rstrip('.')}. {finding}" if raised and finding else finding or raised
+
+    def _phase(self, op_id, text, progress=None):
+        fields = {"phase": utils.safe_text(text)[:255]} if text else {}
+        if progress is not None:
+            fields.update(progress_done=progress[0], progress_total=progress[1])
+        if not fields:
+            return
+        with fenced(self.worker.identity):
+            Operation.objects.filter(pk=op_id, worker=self.worker.identity, state__in=ACTIVE).update(**fields)
+
+    def _inspect(self, op_id):
+        try:
+            op = Operation.objects.get(pk=op_id)
+            if op.state not in ACTIVE:
+                self._done(op_id)
+                return
+            if op.worker != self.worker.identity and not self._take_over(op):
+                return
+            if not op.effects_started:
+                self._finish(op, jobs.Outcome(Operation.FAILED, error=self.errors.get(op_id)
+                                         or "The worker stopped before the job changed anything"))
+                return
+            try:
+                outcome = jobs.get_kind(op.kind).reconcile(op)
+            except Exception as err:
+                n = self.attempts.get(op_id, 0)
+                self.attempts[op_id] = n + 1
+                self.next_try[op_id] = time.monotonic() + BACKOFF[min(n, len(BACKOFF) - 1)]
+                self._phase(op_id, WAITING)
+                logger.warning("recovery of job %s waits: %s", op_id, utils.safe_text(err))
+                return
+            if outcome.state == Operation.RUNNING:
+                now = time.monotonic()
+                if not outcome.stalled:
+                    self.stalled_since.pop(op_id, None)
+                elif now - self.stalled_since.setdefault(op_id, now) >= STALL_LIMIT:
+                    self._finish(op, jobs.Outcome(Operation.UNCERTAIN, error=self._error(op_id, STALLED)))
+                    return
+                self.next_try[op_id] = now + WATCH_EVERY
+                self._phase(op_id, outcome.phase, outcome.progress)
+                return
+            # the error the job raised does not describe a proven success
+            error = outcome.error if outcome.state == Operation.SUCCEEDED else self._error(op_id, outcome.error)
+            self._finish(op, jobs.Outcome(outcome.state, outcome.result, error, outcome.phase))
+        except LeaseLost:
+            self._done(op_id)
+        except Exception as err:
+            logger.error("recovery of job %s failed: %s", op_id, utils.safe_text(err))
+            self.next_try[op_id] = time.monotonic() + BACKOFF[-1]
+        finally:
+            self.inflight.discard(op_id)
+            close_old_connections()