diff --git a/operations/api/serializers.py b/operations/api/serializers.py index 5cb67709..39079a92 100644 --- a/operations/api/serializers.py +++ b/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() diff --git a/operations/jobs.py b/operations/jobs.py index 9a7db501..9d019f24 100644 --- a/operations/jobs.py +++ b/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 diff --git a/operations/models.py b/operations/models.py index cffb4ac9..78e46365 100644 --- a/operations/models.py +++ b/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" diff --git a/operations/static/operations/job_panel.js b/operations/static/operations/job_panel.js index 51e66da8..87568282 100644 --- a/operations/static/operations/job_panel.js +++ b/operations/static/operations/job_panel.js @@ -90,8 +90,10 @@ function jobPanel(element, urls, csrfToken, initialIds) { if (last.short_operation) { html += '
A short operation is in progress.
'; } - // 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 += '
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; } diff --git a/operations/templates/operations/job_panel_script.html b/operations/templates/operations/job_panel_script.html index e1d3b495..ad0258f2 100644 --- a/operations/templates/operations/job_panel_script.html +++ b/operations/templates/operations/job_panel_script.html @@ -1,7 +1,7 @@ {% load static %}