diff --git a/operations/tests/test_recovery.py b/operations/tests/test_recovery.py new file mode 100644 index 00000000..d1aae367 --- /dev/null +++ b/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()) diff --git a/operations/tests/test_worker.py b/operations/tests/test_worker.py index 77924d2a..44338a54 100644 --- a/operations/tests/test_worker.py +++ b/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() diff --git a/operations/worker.py b/operations/worker.py index a4885341..c0130984 100644 --- a/operations/worker.py +++ b/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()