feat(operations): submit, cancel, finish and resolve jobs

operations/jobs.py

@@ -1,14 +1,32 @@
-"""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."""
+"""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. Submitting, cancelling, finishing and resolving jobs (spec
+sections 2-5)."""
 
+import logging
+import re
 from dataclasses import dataclass, field
+from datetime import timedelta
 
-from . import utils
-from .models import LeaseLost, Operation, fenced
+from django.core.exceptions import PermissionDenied
+from django.db import transaction
+from django.db.models import F, Value
+from django.db.models.functions import Coalesce
+from django.utils import timezone
+
+from logs.models import Logs
+from vrtManager.util import OperationError
+
+from . import claims, utils
+from .models import Claim, LeaseLost, Operation, fenced
+
+logger = logging.getLogger(__name__)
 
 KINDS = {}
 
+SUBMISSION_KEY_TTL = timedelta(hours=24)
+KEY_RE = re.compile(r"[A-Za-z0-9_-]{8,64}")
+
 
 @dataclass
 class Plan:
@@ -110,3 +128,123 @@ class Context:
 
     def set_phase(self, text):
         self._write(phase=utils.safe_text(text)[:255])
+
+
+class SubmissionConflict(Exception):
+    """A submission key reused for another request."""
+
+
+def _existing(user, key, kind_name, params):
+    op = Operation.objects.filter(user=user, submission_key=key).first()
+    if op is None:
+        return None
+    if op.created_at < timezone.now() - SUBMISSION_KEY_TTL:
+        Operation.objects.filter(pk=op.pk).update(submission_key=f"expired:{op.pk.hex}")
+        return None
+    if op.kind != kind_name or op.params != params:
+        raise SubmissionConflict("This form was already submitted for another request; reload the page")
+    return op
+
+
+def submit(kind_name, user, raw_params, submission_key):
+    if not KEY_RE.fullmatch(submission_key or ""):
+        raise OperationError("The form has no valid submission key; reload the page")
+    kind = get_kind(kind_name)
+    params = kind.normalize(raw_params)
+    utils.check_params(params)
+    if (op := _existing(user, submission_key, kind.name, params)) is not None:
+        return op
+    lock = f"submission:{user.pk}:{submission_key}"
+    try:
+        kind.validate(user, params)
+        plan = kind.plan(user, params)
+    except Exception:
+        # a concurrent duplicate may have been admitted (and even run) meanwhile
+        with transaction.atomic():
+            claims.serialize([lock])
+            if (op := _existing(user, submission_key, kind.name, params)) is not None:
+                return op
+        raise
+    op = Operation(
+        kind=kind.name, user=user, username=user.username, submission_key=submission_key,
+        resource=plan.resource, resource_label=plan.label[:255], params=params, intent=plan.intent,
+        time_limit=kind.time_limit,
+    )
+    return claims.admit_job(
+        op, plan.claims,
+        existing=lambda: _existing(user, submission_key, kind.name, params),
+        extra_locks=[lock],
+    )
+
+
+def cancel(op_id, user):
+    with transaction.atomic():
+        op = Operation.objects.select_for_update().get(pk=op_id)
+        if not utils.may_see(op, user):
+            raise PermissionDenied
+        if op.state == Operation.QUEUED:
+            Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLED, finished_at=timezone.now())
+            Claim.objects.filter(operation=op).delete()
+            if (entry := get_kind(op.kind).log_entry(op, Operation.CANCELLED)) is not None:
+                _log(op.username, *entry)
+        elif op.state == Operation.RUNNING:
+            if not get_kind(op.kind).cancellable:
+                raise OperationError("This job cannot be cancelled once it has started")
+            Operation.objects.filter(pk=op.pk).update(state=Operation.CANCELLING, cancel_requested=True)
+        elif op.state != Operation.CANCELLING:
+            raise OperationError("The job has already ended")
+    op.refresh_from_db()
+    return op
+
+
+def _log(user, host, instance, message):
+    Logs.objects.create(user=str(user)[:50], host=str(host)[:50], instance=str(instance)[:50],
+                        message=utils.safe_text(message)[:255])
+
+
+def finish(op, outcome, *, worker=None):
+    """Terminal state, claims and history together, once. With a worker,
+    only while it holds the lease and owns the job. Uncertain keeps the claims."""
+    kind = get_kind(op.kind)
+    error = utils.safe_text(outcome.error)
+    with fenced(worker.identity) if worker is not None else transaction.atomic():
+        q = Operation.objects.filter(pk=op.pk, state__in=Operation.ACTIVE)
+        if worker is not None:
+            q = q.filter(worker=worker.identity)
+        # success clears the phase and completes the progress; another end keeps the last ones
+        done = outcome.state == Operation.SUCCEEDED
+        if not q.update(state=outcome.state, result=outcome.result, error=error,
+                        phase=Value(outcome.phase[:255]) if outcome.phase else ("" if done else F("phase")),
+                        progress_done=Coalesce(F("progress_total"), F("progress_done")) if done else F("progress_done"),
+                        finished_at=timezone.now()):
+            return False
+        if outcome.state != Operation.UNCERTAIN:
+            Claim.objects.filter(operation=op).delete()
+        if (entry := kind.log_entry(op, outcome.state)) is not None:
+            _log(op.username, *entry)
+    logger.info("job %s %s: %s", op.pk, op.kind, outcome.state)
+    op.refresh_from_db()
+    return True
+
+
+def resolve(op_id, user, outcome_state, note):
+    if not user.is_superuser:
+        raise PermissionDenied
+    if outcome_state not in (Operation.SUCCEEDED, Operation.FAILED):
+        raise OperationError("Choose succeeded or failed")
+    note = note.strip()
+    if not note:
+        raise OperationError("Write a note on what was checked")
+    with transaction.atomic():
+        op = Operation.objects.select_for_update().get(pk=op_id)
+        if op.state != Operation.UNCERTAIN:
+            raise OperationError("Only an uncertain job can be resolved")
+        Operation.objects.filter(pk=op.pk).update(state=outcome_state, resolution=outcome_state,
+                                                  resolution_note=note, resolved_by=user, resolved_at=timezone.now())
+        Claim.objects.filter(operation=op).delete()
+        # the host and VM columns of the kind's own entries, so the VM's log shows it
+        entry = get_kind(op.kind).log_entry(op, outcome_state)
+        host, instance = entry[:2] if entry is not None else ("-", op.resource_label)
+        _log(user.username, host, instance, f"Job {op.pk} resolved as {outcome_state}: {note}")
+    op.refresh_from_db()
+    return op

operations/templatetags/operations.py

@@ -0,0 +1,12 @@
+import uuid
+
+from django import template
+from django.utils.html import format_html
+
+register = template.Library()
+
+
+@register.simple_tag
+def submission_key():
+    """A new key per rendered form: a resubmitted form returns its job."""
+    return format_html('<input type="hidden" name="submission_key" value="{}">', uuid.uuid4().hex)

operations/tests/test_service.py

@@ -0,0 +1,194 @@
+from datetime import timedelta
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from django.contrib.auth import get_user_model
+from django.core.exceptions import PermissionDenied
+from django.test import TransactionTestCase
+from django.utils import timezone
+from vrtManager.util import OperationError
+
+from logs.models import Logs
+from operations import claims, jobs
+from operations.jobs import Outcome
+from operations.models import Claim, LeaseLost, Operation, WorkerLease, acquire
+from operations.tests import fakes
+
+KEY = "key-00000001"
+
+
+class SubmitTestCase(TransactionTestCase):
+    def setUp(self):
+        fakes.reset()
+        self.user = get_user_model().objects.create_user("svc_user", "s@example.com", "pw")
+
+    def test_a_job_is_queued_with_its_claims(self):
+        op = jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        self.assertEqual(op.state, Operation.QUEUED)
+        self.assertEqual(list(op.claims.values_list("resource", "mode")), [("instance:u1", Claim.EXCLUSIVE)])
+
+    def test_the_same_submission_returns_the_same_job_without_validating_again(self):
+        first = jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        with patch.object(fakes.NoopKind, "validate") as validate:
+            again = jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        self.assertEqual(first.pk, again.pk)
+        validate.assert_not_called()
+
+    def test_a_reused_key_with_other_inputs_or_kind_is_refused(self):
+        jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        with self.assertRaises(jobs.SubmissionConflict):
+            jobs.submit("test.noop", self.user, {"uuid": "u2"}, KEY)
+        with self.assertRaises(jobs.SubmissionConflict):
+            jobs.submit("test.scripted", self.user, {"uuid": "u1"}, KEY)
+
+    def test_a_bad_key_is_refused(self):
+        with self.assertRaises(OperationError):
+            jobs.submit("test.noop", self.user, {}, "x")
+
+    def test_secret_params_are_refused(self):
+        with patch("operations.utils.check_params", side_effect=ValueError("secret")), self.assertRaises(ValueError):
+            jobs.submit("test.noop", self.user, {}, KEY)
+        self.assertFalse(Operation.objects.exists())
+
+    def test_a_duplicate_whose_validation_fails_returns_the_job_admitted_meanwhile(self):
+        first = jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        real = jobs._existing
+        calls = []
+
+        def unseen_first(*args):
+            calls.append(args)
+            return None if len(calls) == 1 else real(*args)  # the duplicate looked before the first was admitted
+
+        with patch.object(jobs, "_existing", side_effect=unseen_first), \
+                patch.object(fakes.NoopKind, "validate", side_effect=OperationError("gone")):
+            again = jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        self.assertEqual(again.pk, first.pk)
+
+    def test_a_conflicting_job_is_refused_and_nothing_is_saved(self):
+        jobs.submit("test.noop", self.user, {"uuid": "u1"}, KEY)
+        with self.assertRaises(claims.Busy):
+            jobs.submit("test.noop", self.user, {"uuid": "u1"}, "key-00000002")
+        self.assertEqual(Operation.objects.count(), 1)
+
+    def test_an_expired_key_starts_a_new_job(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        jobs.finish(op, Outcome(Operation.SUCCEEDED))
+        Operation.objects.filter(pk=op.pk).update(created_at=timezone.now() - timedelta(hours=25))
+        self.assertNotEqual(jobs.submit("test.noop", self.user, {}, KEY).pk, op.pk)
+
+
+class CancelFinishResolveTestCase(TransactionTestCase):
+    def setUp(self):
+        fakes.reset()
+        U = get_user_model()
+        self.user = U.objects.create_user("svc2", "s2@example.com", "pw")
+        self.admin = U.objects.create_superuser("svc_admin", "a@example.com", "pw")
+
+    def test_a_queued_job_is_cancelled_at_once_releases_its_claims_and_is_logged(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        self.assertEqual(jobs.cancel(op.pk, self.user).state, Operation.CANCELLED)
+        self.assertFalse(Claim.objects.exists())
+        self.assertEqual(list(Logs.objects.values_list("message", flat=True)), ["No-op cancelled"])
+
+    def test_the_log_entry_is_written_inside_the_finishing_transaction(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        with patch.object(fakes.NoopKind, "log_entry", side_effect=RuntimeError("x")), self.assertRaises(RuntimeError):
+            jobs.finish(op, Outcome(Operation.SUCCEEDED))
+        self.assertEqual(Operation.objects.get(pk=op.pk).state, Operation.QUEUED)  # rolled back
+        self.assertTrue(Claim.objects.exists())
+
+    def test_cancel_decides_from_the_current_state(self):
+        op = jobs.submit("test.scripted", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING)  # the worker took it meanwhile
+        self.assertEqual(jobs.cancel(op.pk, self.user).state, Operation.CANCELLING)
+        self.assertTrue(Operation.objects.get(pk=op.pk).cancel_requested)
+
+    def test_only_the_owner_or_a_superuser_cancels(self):
+        other = get_user_model().objects.create_user("svc3", "s3@example.com", "pw")
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        with self.assertRaises(PermissionDenied):
+            jobs.cancel(op.pk, other)
+        jobs.cancel(op.pk, self.admin)
+
+    def test_a_running_job_of_a_kind_that_cannot_stop_is_not_cancelled(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING)
+        with self.assertRaisesRegex(OperationError, "cannot be cancelled"):
+            jobs.cancel(op.pk, self.user)
+
+    def test_success_clears_the_phase_and_completes_the_progress(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(phase="merging", progress_done=6424625152, progress_total=6442450944)
+        jobs.finish(op, Outcome(Operation.SUCCEEDED))
+        op.refresh_from_db()
+        self.assertEqual((op.phase, op.progress_done, op.progress_total), ("", 6442450944, 6442450944))
+
+    def test_success_without_a_total_leaves_the_progress_unset(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        jobs.finish(op, Outcome(Operation.SUCCEEDED))
+        op.refresh_from_db()
+        self.assertEqual((op.progress_done, op.progress_total), (None, None))
+
+    def test_success_keeps_the_progress_when_there_is_no_total(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(progress_done=5)
+        jobs.finish(op, Outcome(Operation.SUCCEEDED))
+        op.refresh_from_db()
+        self.assertEqual((op.progress_done, op.progress_total), (5, None))
+
+    def test_a_failure_keeps_the_phase_stored_after_the_job_was_loaded(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(phase="merging")  # op in memory still has ""
+        jobs.finish(op, Outcome(Operation.FAILED, error="x"))
+        op.refresh_from_db()
+        self.assertEqual(op.phase, "merging")
+
+    def test_a_failure_keeps_the_last_phase_and_progress(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(phase="merging", progress_done=5, progress_total=10)
+        op.refresh_from_db()
+        jobs.finish(op, Outcome(Operation.FAILED, error="x"))
+        op.refresh_from_db()
+        self.assertEqual((op.phase, op.progress_done), ("merging", 5))
+
+    def test_finish_happens_once(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        self.assertTrue(jobs.finish(op, Outcome(Operation.SUCCEEDED, result={"x": 1})))
+        self.assertFalse(jobs.finish(op, Outcome(Operation.FAILED, error="late")))
+        op.refresh_from_db()
+        self.assertEqual((op.state, op.result), (Operation.SUCCEEDED, {"x": 1}))
+        self.assertEqual(Logs.objects.count(), 1)
+        self.assertFalse(Claim.objects.exists())
+
+    def test_finish_by_a_worker_without_the_lease_changes_nothing(self):
+        acquire("h1:1:a", "h1")
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, worker="h1:1:a")
+        WorkerLease.objects.update(holder="h2:1:b", host="h2")
+        with self.assertRaises(LeaseLost):
+            jobs.finish(op, Outcome(Operation.SUCCEEDED), worker=SimpleNamespace(identity="h1:1:a"))
+        self.assertEqual(Operation.objects.get(pk=op.pk).state, Operation.RUNNING)
+
+    def test_errors_and_logs_carry_no_secrets_and_fit_the_columns(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        Operation.objects.filter(pk=op.pk).update(username="u" * 150)
+        op.refresh_from_db()
+        jobs.finish(op, Outcome(Operation.FAILED, error="qemu+tcp://admin:s3cret@192.0.2.5/system failed"))
+        op.refresh_from_db()
+        self.assertNotIn("s3cret", op.error)
+        self.assertLessEqual(len(Logs.objects.get().user), 50)
+
+    def test_an_uncertain_job_keeps_its_claims_until_resolved(self):
+        op = jobs.submit("test.noop", self.user, {}, KEY)
+        jobs.finish(op, Outcome(Operation.UNCERTAIN, error="check the disks"))
+        self.assertTrue(Claim.objects.filter(operation=op).exists())
+        with self.assertRaises(PermissionDenied):
+            jobs.resolve(op.pk, self.user, Operation.FAILED, "checked")
+        with self.assertRaisesRegex(OperationError, "note"):
+            jobs.resolve(op.pk, self.admin, Operation.FAILED, "  ")
+        jobs.resolve(op.pk, self.admin, Operation.FAILED, "disks checked, nothing changed")
+        op.refresh_from_db()
+        self.assertEqual((op.state, op.resolution, op.resolved_by_id), (Operation.FAILED, Operation.FAILED, self.admin.pk))
+        self.assertFalse(Claim.objects.exists())
+        entry = Logs.objects.get(message__icontains="resolved")
+        self.assertEqual((entry.user, entry.host, entry.instance), (self.admin.username, "c", "vm"))