feat(operations): jobs a request runs

Request jobs (an upload's chunks) run in gunicorn: their kind is admitted
running with the request as its worker, never claimed or recovered by the
worker, cancelled in the request, and asked by every housekeeping pass to
end its jobs that went idle. The panel can follow a job the page just
started. A job a request runs does not wait for the worker, so it does not
raise the worker-unavailable warning on the Jobs page or in the panel.

operations/api/serializers.py

@@ -10,12 +10,13 @@ class OperationSerializer(serializers.ModelSerializer):
     label = serializers.SerializerMethodField()
     resource_label = serializers.SerializerMethodField()
     cancellable = serializers.SerializerMethodField()
+    in_request = serializers.SerializerMethodField()
 
     class Meta:
         model = Operation
         fields = ["id", "kind", "label", "username", "resource", "resource_label", "state", "phase",
                   "progress_done", "progress_total", "result", "error", "created_at", "started_at",
-                  "finished_at", "cancel_requested", "cancellable", "resolution", "resolution_note"]
+                  "finished_at", "cancel_requested", "cancellable", "in_request", "resolution", "resolution_note"]
         read_only_fields = fields
 
     def get_label(self, op) -> str:
@@ -28,6 +29,10 @@ class OperationSerializer(serializers.ModelSerializer):
         kind = KINDS.get(op.kind)
         return op.state == Operation.QUEUED or bool(kind and kind.cancellable and op.state == Operation.RUNNING)
 
+    def get_in_request(self, op) -> bool:
+        """A request runs it: it does not wait for the worker."""
+        return op.in_request
+
 
 class OperationSummarySerializer(serializers.Serializer):
     kind = serializers.CharField()

operations/jobs.py

@@ -73,6 +73,7 @@ class Kind:
     label = ""
     cancellable = False
     time_limit = None
+    runs_in_request = False  # its requests run it: admitted running, never the worker's
     progress_bytes = False  # progress counts bytes: the job page shows sizes
 
     def normalize(self, raw):
@@ -100,6 +101,10 @@ class Kind:
     def reconcile(self, op):
         raise NotImplementedError
 
+    def expire(self, op):
+        """Called on every housekeeping pass for an active job a request runs;
+        the kind decides whether the job is idle and ends it if so."""
+
     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
@@ -170,6 +175,9 @@ def submit(kind_name, user, raw_params, submission_key):
         resource=plan.resource, resource_label=plan.label[:255], params=params, intent=plan.intent,
         time_limit=kind.time_limit,
     )
+    if kind.runs_in_request:
+        now = timezone.now()
+        op.state, op.started_at, op.heartbeat_at, op.worker = Operation.RUNNING, now, now, Operation.REQUEST
     return claims.admit_job(
         op, plan.claims,
         existing=lambda: _existing(user, submission_key, kind.name, params),
@@ -182,6 +190,8 @@ def cancel(op_id, user):
         op = Operation.objects.select_for_update().get(pk=op_id)
         if not utils.may_see(op, user):
             raise PermissionDenied
+        # read before the update below; a repeated cancel retries the request job's stop
+        request_job = op.in_request and op.state in (Operation.RUNNING, Operation.CANCELLING)
         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()
@@ -194,6 +204,9 @@ def cancel(op_id, user):
         elif op.state != Operation.CANCELLING:
             raise OperationError("The job has already ended")
     op.refresh_from_db()
+    if request_job:  # no worker signals it: the request stops it
+        get_kind(op.kind).cancel(op)
+        op.refresh_from_db()
     return op
 
 

operations/models.py

@@ -26,6 +26,7 @@ class Operation(models.Model):
     ACTIVE = (QUEUED, RUNNING, CANCELLING)
     TERMINAL = (SUCCEEDED, FAILED, CANCELLED, UNCERTAIN)
     STATES = [(s, s) for s in ACTIVE + TERMINAL]
+    REQUEST = "request"  # the worker of a job a request runs
 
     id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
     kind = models.CharField(max_length=64, db_index=True)
@@ -49,7 +50,8 @@ class Operation(models.Model):
     started_at = models.DateTimeField(null=True)
     finished_at = models.DateTimeField(null=True)
     heartbeat_at = models.DateTimeField(null=True)
-    worker = models.CharField(max_length=128, blank=True)  # identity of the worker that owns it
+    # identity of the owning worker; REQUEST for a job a request runs, which the worker never claims or recovers
+    worker = models.CharField(max_length=128, blank=True)
     time_limit = models.PositiveIntegerField(null=True)  # seconds, fixed at admission
     cancel_requested = models.BooleanField(default=False)
     resolved_by = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL, related_name="+")
@@ -64,6 +66,10 @@ class Operation(models.Model):
     def __str__(self):
         return f"{self.kind} {self.id}"
 
+    @property
+    def in_request(self):
+        return self.worker == Operation.REQUEST
+
 
 class Claim(models.Model):
     SHARED = "shared"

operations/static/operations/job_panel.js

@@ -90,8 +90,10 @@ function jobPanel(element, urls, csrfToken, initialIds) {
         if (last.short_operation) {
             html += '<div class="alert alert-secondary mb-3" role="status">A short operation is in progress.</div>';
         }
-        // a job the worker has to start, run or stop (the Jobs page uses the same rule)
-        var waiting = last.jobs.some(function (j) { return ["queued", "running", "cancelling"].indexOf(j.state) >= 0; });
+        // a job the worker has to start, run or stop, not one a request runs (the Jobs page uses the same rule)
+        var waiting = last.jobs.some(function (j) {
+            return ["queued", "running", "cancelling"].indexOf(j.state) >= 0 && !j.in_request;
+        });
         var seen = last.worker_last_seen ? new Date(last.worker_last_seen) : null;
         if (waiting && (!seen || Date.now() - seen.getTime() > 60000)) {
             html += '<div class="alert alert-warning mb-3" role="status">Worker unavailable, last seen '
@@ -115,10 +117,12 @@ function jobPanel(element, urls, csrfToken, initialIds) {
         inflight[id] = true;
         $.getJSON(urls.job(id))
             .done(function (job) {
+                if (dismissed[id]) { delete pending[id]; return; }
+                // still active: pending until a list shows it (one older than the job may not) or it
+                // ends; the next poll asks for it again
+                if (["queued", "running", "cancelling"].indexOf(job.state) >= 0) { return; }
                 delete pending[id];
-                if (dismissed[id]) { return; }
-                // a just-submitted job that is still active shows in the list instead
-                if (["queued", "running", "cancelling"].indexOf(job.state) < 0) { outcomes[id] = job; }
+                outcomes[id] = job;
                 render();
             })
             .fail(function (xhr) {
@@ -169,5 +173,12 @@ function jobPanel(element, urls, csrfToken, initialIds) {
     }
 
     var poller = pollWhileVisible(3000, function () { return $.getJSON(urls.resource); }, onData, 0);
+    // follow a job the page just started, as a remembered one
+    poller.track = function (id) {
+        pending[id] = true;
+        render();
+        poller.stop(); // ask for the list now
+        poller.start();
+    };
     return poller;
 }

operations/templates/operations/job_panel_script.html

@@ -1,7 +1,7 @@
 {% load static %}
 <script src="{% static 'operations/job_panel.js' %}"></script>
 <script>
-    jobPanel(document.getElementById("job-panel"), {
+    window.resourceJobPanel = jobPanel(document.getElementById("job-panel"), {
         resource: "{% url 'operation-resource' %}?key={{ resource|urlencode:'' }}",
         job: function (id) { return "{% url 'operation-list' %}" + id + "/"; },
         cancel: function (id) { return "{% url 'operation-list' %}" + id + "/cancel/"; },

operations/tests/fakes.py

@@ -4,19 +4,23 @@ submitting (keyed by submission key). start_worker and stop_worker run a
 worker's loop in a thread."""
 
 import threading
+from datetime import timedelta
 
 from django.db import connection
+from django.utils import timezone
 
 from operations.jobs import Cancelled, Kind, Outcome, Plan, register
 from operations.models import Claim, Operation, acquire
 
 BARRIERS = {}  # submission key -> threading.Event the run waits on
 CANCELLED = []
+EXPIRED = []
 
 
 def reset():
     BARRIERS.clear()
     CANCELLED.clear()
+    EXPIRED.clear()
 
 
 def start_worker(worker):
@@ -100,3 +104,26 @@ class ScriptedKind(NoopKind):
 
     def reconcile(self, op):
         return Outcome(op.params["outcome"], error="scripted")
+
+
+@register
+class RequestKind(NoopKind):
+    """A job a request runs, idle after 60 s; params: fail_expire."""
+
+    name = "test.request"
+    label = "Request"
+    cancellable = True
+    runs_in_request = True
+
+    def normalize(self, raw):
+        return {"uuid": str(raw.get("uuid", "u1")), "fail_expire": bool(raw.get("fail_expire"))}
+
+    def cancel(self, op):
+        CANCELLED.append(str(op.id))
+
+    def expire(self, op):
+        if op.heartbeat_at > timezone.now() - timedelta(seconds=60):
+            return
+        if op.params["fail_expire"]:
+            raise RuntimeError("expire failed")
+        EXPIRED.append(str(op.id))

operations/tests/js/job_panel.test.js

@@ -194,6 +194,13 @@ assert.ok(!el8.innerHTML.includes("Worker unavailable"), "a fresh lease does not
 advance(3000);
 answer("/r", { ...empty, worker_last_seen: stale, jobs: [{ ...running, state: "uncertain" }] });
 assert.ok(!el8.innerHTML.includes("Worker unavailable"), "an uncertain job waits for an administrator, not the worker");
+advance(3000);
+answer("/r", { ...empty, worker_last_seen: stale, jobs: [{ ...running, in_request: true }] });
+assert.ok(!el8.innerHTML.includes("Worker unavailable"), "a job a request runs does not wait for the worker");
+advance(3000);
+answer("/r", { ...empty, worker_last_seen: stale, jobs: [{ ...running, in_request: true },
+                                                        { ...running, id: "w", state: "queued", in_request: false }] });
+assert.ok(el8.innerHTML.includes("Worker unavailable"), "a queued job beside it still warns");
 
 // cancel: the button is disabled while the request runs; a refusal or a network error shows; the list is asked again
 open = []; timers = [];
@@ -220,5 +227,39 @@ answer("/r", { ...empty, jobs: [queued] });
 click9({ target: { dataset: { cancel: "q" } } });
 answer("/c/q", { status: 0 }, false);
 assert.ok(el9.innerHTML.includes("could not be reached"), "a network error shows");
+
+// track(id): the panel follows a job the page just started and asks for the list at once
+open = []; timers = [];
+window.backgroundJobRunning = false;
+const el10 = { innerHTML: "", addEventListener() {} };
+const panel = jobPanel(el10, urls, "tok");
+advance(0);
+answer("/r", empty);
+assert.strictEqual(window.backgroundJobRunning, false);
+panel.track("t");
+assert.strictEqual(window.backgroundJobRunning, true, "a tracked job holds the page's reload");
+assert.ok(open.some((r) => r.url === "/r" && !r.settled), "the list is asked for at once");
+answer("/r", empty);
+answer("/j/t", { ...running, id: "t", label: "Upload", state: "cancelled" });
+assert.ok(el10.innerHTML.includes("Upload") && el10.innerHTML.includes('<span class="badge bg-secondary">cancelled</span>'),
+    "the tracked job's outcome shows");
+
+// a list answer older than the job, then its detail still active: the panel keeps following it
+open = []; timers = [];
+const el11 = { innerHTML: "", addEventListener() {} };
+const panel11 = jobPanel(el11, urls, "tok");
+advance(0);
+answer("/r", empty);
+panel11.track("u");
+answer("/r", empty);
+answer("/j/u", { ...running, id: "u", label: "Upload" });
+assert.ok(!open.some((r) => !r.settled && r.url === "/j/u"), "an active detail is not asked for again at once");
+assert.strictEqual(window.backgroundJobRunning, true, "still followed");
+advance(3000);
+answer("/r", empty);
+assert.strictEqual(open.filter((r) => !r.settled && r.url === "/j/u").length, 1, "asked again once per poll");
+answer("/j/u", { ...running, id: "u", label: "Upload", state: "succeeded" });
+assert.ok(el11.innerHTML.includes("Upload") && el11.innerHTML.includes('<span class="badge bg-success">succeeded</span>'),
+    "the outcome shows once the job ends");
 Date.now = realNow;
 console.log("ok");

operations/tests/test_api.py

@@ -49,6 +49,14 @@ class OperationsApiTestCase(TransactionTestCase):
         self.client.force_login(self.owner)
         self.assertTrue(self.client.get("/api/v1/operations/resource/", {"key": "instance:u2"}).json()["short_operation"])
 
+    def test_the_resource_view_marks_a_job_a_request_runs(self):
+        request_job = jobs.submit("test.request", self.owner, {"uuid": "u2"}, "key-00000002")
+        self.client.force_login(self.owner)
+        [job] = self.client.get("/api/v1/operations/resource/", {"key": "instance:u2"}).json()["jobs"]
+        self.assertEqual((job["id"], job["in_request"]), (str(request_job.pk), True))
+        # queued: it waits for the worker
+        self.assertFalse(self.client.get(f"/api/v1/operations/{self.op.pk}/").json()["in_request"])
+
     def test_cancel_through_the_api(self):
         self.client.force_login(self.owner)
         response = self.client.post(f"/api/v1/operations/{self.op.pk}/cancel/")

operations/tests/test_kinds.py

@@ -5,9 +5,10 @@ from django.contrib.auth import get_user_model
 from django.test import TestCase
 from django.utils import timezone
 
+from operations import jobs
 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
+from operations.models import TAKEOVER_AFTER, Claim, LeaseLost, Operation, WorkerLease, acquire, fenced, renew
+from operations.tests import fakes
 
 
 class RegistryTestCase(TestCase):
@@ -67,3 +68,38 @@ class ContextTestCase(TestCase):
         self.op.refresh_from_db()
         self.assertEqual((self.op.intent, self.op.effects_started), ({}, False))
         self.assertFalse(ctx.effects_started)
+
+
+class RequestJobTestCase(TestCase):
+    def setUp(self):
+        fakes.reset()
+        self.user = get_user_model().objects.create_user("req_user", "q@example.com", "pw")
+
+    def test_a_request_job_is_admitted_running_in_the_request(self):
+        op = jobs.submit("test.request", self.user, {}, "key-00000001")
+        op.refresh_from_db()
+        self.assertEqual((op.state, op.worker, op.in_request), (Operation.RUNNING, Operation.REQUEST, True))
+        self.assertIsNotNone(op.started_at)
+        self.assertEqual(op.heartbeat_at, op.started_at)
+        self.assertEqual(list(op.claims.values_list("resource", "mode")), [("instance:u1", Claim.EXCLUSIVE)])
+
+    def test_the_same_submission_returns_the_running_request_job(self):
+        op = jobs.submit("test.request", self.user, {}, "key-00000001")
+        again = jobs.submit("test.request", self.user, {}, "key-00000001")
+        self.assertEqual((again.pk, Operation.objects.count()), (op.pk, 1))
+
+    def test_cancelling_a_request_job_runs_its_cancel_in_the_request(self):
+        op = jobs.submit("test.request", self.user, {}, "key-00000001")
+        op = jobs.cancel(op.pk, self.user)
+        self.assertEqual((op.state, op.cancel_requested), (Operation.CANCELLING, True))
+        self.assertEqual(fakes.CANCELLED, [str(op.pk)])
+        jobs.cancel(op.pk, self.user)  # its stop did not end it: a repeated cancel retries
+        self.assertEqual(fakes.CANCELLED, [str(op.pk)] * 2)
+
+    def test_cancelling_a_workers_job_leaves_its_cancel_to_the_worker(self):
+        op = jobs.submit("test.scripted", self.user, {}, "key-00000001")
+        Operation.objects.filter(pk=op.pk).update(state=Operation.RUNNING, worker="h1:1:a")
+        op = jobs.cancel(op.pk, self.user)
+        self.assertEqual(op.state, Operation.CANCELLING)
+        jobs.cancel(op.pk, self.user)
+        self.assertEqual(fakes.CANCELLED, [])

operations/tests/test_recovery.py

@@ -188,6 +188,12 @@ class RecoveryTestCase(TransactionTestCase):
         self.run_due()
         self.assertEqual(self.reload(op).state, Operation.RUNNING)
 
+    def test_a_request_job_is_not_an_orphan(self):
+        op = jobs.submit("test.request", self.user, {}, "key-00000002")
+        self.run_due()
+        self.assertEqual(self.scheduler.next_try, {})
+        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")

operations/tests/test_views.py

@@ -161,10 +161,14 @@ class JobsPageTestCase(TransactionTestCase):
         self.client.force_login(self.user)
         for state, warned in ((Operation.QUEUED, True), (Operation.RUNNING, True), (Operation.CANCELLING, True),
                               (Operation.UNCERTAIN, False), (Operation.SUCCEEDED, False)):
-            Operation.objects.filter(pk=self.op.pk).update(state=state)
+            Operation.objects.filter(pk=self.op.pk).update(state=state, worker="" if state == Operation.QUEUED else "h:1:a")
             response = self.client.get(reverse("operations:jobs"))
             (self.assertContains if warned else self.assertNotContains)(response, "Worker unavailable", msg_prefix=state)
-        Operation.objects.filter(pk=self.op.pk).update(state=Operation.RUNNING)
+        # a job a request runs does not wait for the worker
+        for state in (Operation.RUNNING, Operation.CANCELLING):
+            Operation.objects.filter(pk=self.op.pk).update(state=state, worker=Operation.REQUEST)
+            self.assertNotContains(self.client.get(reverse("operations:jobs")), "Worker unavailable", msg_prefix=state)
+        Operation.objects.filter(pk=self.op.pk).update(state=Operation.RUNNING, worker="h:1:a")
         WorkerLease.objects.update(heartbeat_at=django_timezone.now())
         self.assertNotContains(self.client.get(reverse("operations:jobs")), "Worker unavailable")
         # another user's running job is not this user's concern

operations/tests/test_worker.py

@@ -6,19 +6,21 @@ import os
 import tempfile
 import threading
 import time
+from datetime import timedelta
 from unittest.mock import patch
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.db import OperationalError, connection
 from django.test import TransactionTestCase
+from django.utils import timezone
 
 from appsettings.models import AppSettings
 from appsettings.settings import app_settings
 from operations import jobs
 from operations.models import Claim, Operation, WorkerLease
 from operations.tests import fakes
-from operations.worker import Worker, data_dir, host_identity
+from operations.worker import Worker, data_dir, host_identity, housekeeping
 from vrtManager.util import ConnectionFailed
 
 
@@ -237,6 +239,45 @@ class WorkerTestCase(TransactionTestCase):
         exit_.assert_called_once_with(1)
 
 
+class RequestJobTestCase(TransactionTestCase):
+    def setUp(self):
+        fakes.reset()
+        self.user = get_user_model().objects.create_user("rq_user", "q@example.com", "pw")
+
+    def request_job(self, key, idle, **params):
+        op = jobs.submit("test.request", self.user, params, key)
+        Operation.objects.filter(pk=op.pk).update(heartbeat_at=timezone.now() - timedelta(seconds=idle))
+        return op
+
+    def test_the_worker_never_takes_a_request_job(self):
+        op = self.request_job("key-00000001", 0)
+        worker = Worker(concurrency=2, tick=0.05, heartbeat_every=0.1, progress_every=0.1, grace=1,
+                        exit_on_lost_lease=False)
+        thread = fakes.start_worker(worker)
+        try:
+            time.sleep(0.5)
+            self.assertEqual(worker.jobs, {})
+        finally:
+            fakes.stop_worker(worker, thread)
+            connection.close()
+        op.refresh_from_db()
+        self.assertEqual((op.state, op.worker), (Operation.RUNNING, Operation.REQUEST))
+
+    def test_housekeeping_lets_the_kind_expire_its_idle_request_jobs(self):
+        fresh = self.request_job("key-00000001", 30, uuid="u1")
+        idle = self.request_job("key-00000002", 61, uuid="u2")
+        housekeeping()
+        self.assertEqual(fakes.EXPIRED, [str(idle.pk)])
+        self.assertNotIn(str(fresh.pk), fakes.EXPIRED)
+
+    def test_one_failing_expire_does_not_stop_the_others(self):
+        self.request_job("key-00000001", 61, uuid="u1", fail_expire=True)
+        other = self.request_job("key-00000002", 61, uuid="u2")
+        with self.assertLogs("operations.worker", "ERROR"):
+            housekeeping()
+        self.assertEqual(fakes.EXPIRED, [str(other.pk)])
+
+
 class DataDirTestCase(TransactionTestCase):
     def test_the_data_dir_does_not_follow_the_log_file(self):
         with self.settings(WEBVIRTCLOUD_LOG_FILE="/var/log/wvc-test/webvirtcloud.log"):

operations/views.py

@@ -39,9 +39,10 @@ def jobs(request):
     # the model for its datetimes, the serialized job for its labels
     rows = [(op, OperationSerializer(op, context={"request": request}).data) for op in page.object_list]
     seen = last_seen()
-    # a job the worker has to start, run or stop (job_panel.js uses the same rule)
+    # a job the worker has to start, run or stop, not one a request runs (job_panel.js uses the same rule)
     unavailable = ((seen is None or timezone.now() - seen > timedelta(seconds=60))
-                   and _visible(request.user).filter(state__in=Operation.ACTIVE).exists())
+                   and _visible(request.user).filter(state__in=Operation.ACTIVE)
+                   .exclude(worker=Operation.REQUEST).exists())
     return render(request, "operations/jobs.html",
                   {"page": page, "rows": rows, "worker_last_seen": seen, "worker_unavailable": unavailable,
                    "filters": request.GET, "active_states": Operation.ACTIVE,

operations/worker.py

@@ -30,7 +30,7 @@ logger = logging.getLogger(__name__)
 
 MAX_SLOTS = 16
 PROGRESS_THREADS = 2
-HOUSEKEEPING_EVERY = 3600.0
+HOUSEKEEPING_EVERY = 600.0
 
 BACKOFF = (30, 60, 120, 300, 600)
 WATCH_EVERY = 15
@@ -307,6 +307,12 @@ def housekeeping():
     )
     Operation.objects.filter(pk__in=old).delete()
     Claim.objects.filter(operation__isnull=True, expires_at__lt=timezone.now()).delete()
+    # each kind ends its idle jobs; HOUSEKEEPING_EVERY bounds how late an idle request job ends
+    for op in Operation.objects.filter(state__in=ACTIVE, worker=Operation.REQUEST):
+        try:
+            jobs.get_kind(op.kind).expire(op)
+        except Exception as err:
+            logger.error("expiring job %s failed: %s", op.pk, utils.safe_text(err))
 
 
 class Scheduler:
@@ -319,7 +325,9 @@ class Scheduler:
         self.inflight = set()
 
     def schedule_orphans(self):
-        for op_id in (Operation.objects.filter(state__in=ACTIVE).exclude(worker=self.worker.identity)
+        """Other workers' active jobs; not those requests run."""
+        for op_id in (Operation.objects.filter(state__in=ACTIVE)
+                      .exclude(worker__in=[self.worker.identity, Operation.REQUEST])
                       .values_list("pk", flat=True)):
             self.next_try.setdefault(op_id, 0)