diff --git a/operations/jobs.py b/operations/jobs.py index 19045297..52ce5c35 100644 --- a/operations/jobs.py +++ b/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 diff --git a/operations/templatetags/__init__.py b/operations/templatetags/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/operations/templatetags/operations.py b/operations/templatetags/operations.py new file mode 100644 index 00000000..476b3cee --- /dev/null +++ b/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('', uuid.uuid4().hex) diff --git a/operations/tests/test_service.py b/operations/tests/test_service.py new file mode 100644 index 00000000..6d6d9539 --- /dev/null +++ b/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"))