feat: support executors[].ignore_error

docs/api.md

@@ -107,9 +107,27 @@ TESK accepts the fields that TES 1.1 added to tasks:
   `400 Bad Request`.
 - `inputs[].streamable`: accepted and ignored, as TES allows: inputs are
   always downloaded before the executors start.
-- `executors[].ignore_error`: only `false` is accepted. TESK stops at the
-  first executor that fails, so `true` is refused with `400 Bad Request`
-  rather than run differently than asked.
+- `executors[].ignore_error`: with `true`, a failure of the executor
+  doesn't end the task.
+    - The executor's job gets the annotation `ignore-error: "true"`, and
+      the taskmaster goes on with the next executor and the outputs filer,
+      as if the executor had completed. The task ends `COMPLETE` if the
+      other jobs complete, including the outputs filer if the task has
+      outputs: a taskmaster that stopped at the failure without uploading
+      them gives `SYSTEM_ERROR`. The executor log keeps the executor's
+      `exit_code` (and, with `FULL`, `stdout`), so clients see the failure
+      there.
+    - Failures that the executor's pods report are ignored: a non-zero exit
+      code, but also an eviction, preemption or OOM kill.
+    - Failures of jobs that the taskmaster ended itself aren't, as the
+      executor may never have run. A pod that couldn't start (e.g. an image
+      that can't be pulled) fails its job with the reason
+      `DeadlineExceeded`, which gives `EXECUTOR_ERROR`. An executor whose
+      pod wasn't scheduled in time usually has its job deleted, and the
+      missing executor gives `SYSTEM_ERROR`.
+    - It needs a taskmaster that supports the annotation
+      ([tesk-core#142](https://github.com/elixir-cloud-aai/tesk-core/issues/142));
+      see [Upgrading](upgrading.md) for what older ones do.
 - `resources.backend_parameters`: TESK supports none (service info lists
   none, see [Service info](#service-info)).
     - With `backend_parameters_strict` set to `true`, for which TES asks to
@@ -131,10 +149,10 @@ Service info reports the TES version 1.1.0 (`type.version`).
 TESK API implements TES 1.1 with these differences, which clients should
 know about:
 
-- `executors[].ignore_error: true` is refused with `400 Bad Request` (see
-  [TES 1.1 fields](#tes-11-fields);
-  [#159](https://github.com/elixir-cloud-aai/tesk-api/issues/159) is about
-  supporting it).
+- `executors[].ignore_error: true` ignores only the failures that the
+  executor's pods report, not those of executors that the taskmaster ended
+  because their pod couldn't start or be scheduled (see [TES 1.1
+  fields](#tes-11-fields)).
 - `resources.preemptible` and `resources.zones` are accepted and returned,
   but have no effect on where the task runs, as TES allows.
 - `inputs[].streamable` is accepted and ignored (see [TES 1.1

docs/index.md

@@ -96,12 +96,18 @@ applies gives the state:
    as active (see below); `CANCELED` once none has.
 2. An executor's job failed, whichever executor it is: `PREEMPTED` if every
    failed executor's job was disrupted (see below), otherwise
-   `EXECUTOR_ERROR`.
+   `EXECUTOR_ERROR`. Executors that ignore errors (`ignore_error: true`,
+   see [TES 1.1 fields](api.md#tes-11-fields)) don't count here, unless the
+   taskmaster ended their job itself (it has `activeDeadlineSeconds`, or
+   the reason `DeadlineExceeded`).
 3. The taskmaster's job or the outputs filer's job failed: `SYSTEM_ERROR`.
 4. The taskmaster's job completed: `COMPLETE` if the jobs of all the task's
-   executors exist and completed, otherwise `SYSTEM_ERROR` (e.g., the inputs
+   executors exist and completed (or failed, for executors whose errors are
+   ignored as in rule 2), otherwise `SYSTEM_ERROR` (e.g., the inputs
    filer failed, so that no executor ran, or an executor's job is missing).
-   The number of executors comes from the task stored with the taskmaster's
+   After an ignored failure, a task with outputs (or whose stored task
+   can't be read) also needs its outputs filer's job to have completed;
+   otherwise it's `SYSTEM_ERROR`. The number of executors comes from the task stored with the taskmaster's
    job; without it, the executor jobs that exist count as all executors.
 5. A pod of the taskmaster's job is pending: `QUEUED`.
 6. The taskmaster's job is active: `INITIALIZING` until the first executor's

docs/upgrading.md

@@ -29,6 +29,15 @@ notice changed answers:
   are more (see [Listing tasks](api.md#listing-tasks)). Before, it returned
   all tasks at once. Clients that don't pass `page_size` must follow
   `next_page_token`, or pass a `page_size` of up to 2048.
+- **`ignore_error`.** `executors[].ignore_error: true` is accepted rather
+  than refused with `400 Bad Request` (see [TES 1.1
+  fields](api.md#tes-11-fields)). It needs a taskmaster that goes on after
+  such an executor fails
+  ([tesk-core#142](https://github.com/elixir-cloud-aai/tesk-core/issues/142)):
+  upgrade the taskmaster image first (`TESK_API_TASKMASTER_IMAGE_VERSION`,
+  see [Configuration](configuration.md)). An older taskmaster stops at the
+  failure: the task then ends `SYSTEM_ERROR`, unless the failed executor was
+  the last one and the task has no outputs, which completes as asked.
 
 ## To 1.2.0
 

src/main/java/uk/ac/ebi/tsc/tesk/k8s/constant/Constants.java

@@ -47,6 +47,16 @@ public class Constants {
     /** Key of the annotation, that stores whole input TES task serialized to JSON */
     public static final String ANN_JSON_INPUT_KEY = "json-input";
 
+    /**
+     * Key of the annotation of an executor's job that ignores errors (TES {@code ignore_error}),
+     * with the value {@code "true"}: the taskmaster goes on after the executor fails, and the
+     * failure doesn't fail the task (needs a taskmaster with elixir-cloud-aai/tesk-core#142)
+     */
+    public static final String ANN_IGNORE_ERROR_KEY = "ignore-error";
+
+    /** Value of the annotation {@link #ANN_IGNORE_ERROR_KEY} of an executor that ignores errors */
+    public static final String ANN_IGNORE_ERROR_VALUE = "true";
+
     /**
      * Key of the annotation of the taskmaster's job with the keys of the task's backend parameters,
      * which TESK doesn't support and therefore doesn't store (a JSON array of strings)

src/main/java/uk/ac/ebi/tsc/tesk/k8s/convert/TaskStateResolver.java

@@ -1,10 +1,13 @@
 package uk.ac.ebi.tsc.tesk.k8s.convert;
 
+import static uk.ac.ebi.tsc.tesk.k8s.constant.Constants.ANN_IGNORE_ERROR_KEY;
+import static uk.ac.ebi.tsc.tesk.k8s.constant.Constants.ANN_IGNORE_ERROR_VALUE;
 import static uk.ac.ebi.tsc.tesk.k8s.constant.Constants.LABEL_TASKSTATE_KEY;
 import static uk.ac.ebi.tsc.tesk.k8s.constant.Constants.LABEL_TASKSTATE_VALUE_CANC;
 
 import io.kubernetes.client.openapi.models.V1Job;
 import io.kubernetes.client.openapi.models.V1JobCondition;
+import io.kubernetes.client.openapi.models.V1JobSpec;
 import io.kubernetes.client.openapi.models.V1JobStatus;
 import io.kubernetes.client.openapi.models.V1ObjectMeta;
 import io.kubernetes.client.openapi.models.V1Pod;
@@ -144,6 +147,24 @@ private static boolean isDisrupted(V1Pod pod) {
      * @return the TES state of the task
      */
     public static TesState resolve(Task task, Supplier<Integer> expectedExecutors) {
+        return resolve(task, expectedExecutors, () -> null);
+    }
+
+    /**
+     * Derives the state of a task.
+     *
+     * @param task the task's jobs and their pods
+     * @param expectedExecutors supplies the number of executors of the task, or {@code null} if
+     *     unknown (then the executor jobs that exist count as all executors); only called once the
+     *     taskmaster's job completed
+     * @param hasOutputs supplies whether the task has outputs, or {@code null} if unknown; only
+     *     called once the taskmaster's job completed after an executor whose failure is ignored
+     *     (then, with outputs or if unknown, the task completed only if the outputs filer did, as a
+     *     taskmaster can stop at such a failure without uploading the outputs)
+     * @return the TES state of the task
+     */
+    public static TesState resolve(
+            Task task, Supplier<Integer> expectedExecutors, Supplier<Boolean> hasOutputs) {
         Job taskmaster = task.getTaskmaster();
         V1Job taskmasterJob = taskmaster.getJob();
         if (isCancelled(taskmasterJob)) {
@@ -161,6 +182,7 @@ public static TesState resolve(Task task, Supplier<Integer> expectedExecutors) {
         List<Job> failedExecutors =
                 executors.stream()
                         .filter(executor -> hasFailed(executor.getJob(), taskmasterEnded))
+                        .filter(executor -> !ignoresError(executor.getJob()))
                         .collect(Collectors.toList());
         if (!failedExecutors.isEmpty()) {
             return failedExecutors.stream().allMatch(TaskStateResolver::wasDisrupted)
@@ -180,8 +202,18 @@ public static TesState resolve(Task task, Supplier<Integer> expectedExecutors) {
             boolean allCompleted =
                     expected > 0
                             && executors.size() == expected
-                            && executors.stream()
-                                    .allMatch(executor -> hasSucceeded(executor.getJob()));
+                            && executors.stream().allMatch(executor -> isDone(executor.getJob()));
+            // After an ignored failure, a taskmaster can stop without uploading the outputs
+            // (one without support for ignore_error, or one whose failed executor's pods didn't
+            // end in time): the outputs filer must have completed, if there are outputs
+            boolean ignoredFailure =
+                    executors.stream().anyMatch(executor -> !hasSucceeded(executor.getJob()));
+            if (allCompleted
+                    && ignoredFailure
+                    && !Boolean.FALSE.equals(hasOutputs.get())
+                    && !outputFiler.map(filer -> hasSucceeded(filer.getJob())).orElse(false)) {
+                return TesState.SYSTEM_ERROR;
+            }
             return allCompleted ? TesState.COMPLETE : TesState.SYSTEM_ERROR;
         }
         if (hasPendingPod(taskmaster)) {
@@ -209,6 +241,55 @@ public static TesState resolve(Task task, Supplier<Integer> expectedExecutors) {
         return TesState.SYSTEM_ERROR;
     }
 
+    /**
+     * Whether a failure of an executor's job is ignored (TES {@code ignore_error}, issue #159): the
+     * job has the annotation {@code ignore-error} with the value {@code "true"}, and the taskmaster
+     * didn't end it itself. The taskmaster ends a job whose pod can't start or be scheduled by
+     * setting its {@code activeDeadlineSeconds}, so that Kubernetes fails it with the reason {@code
+     * DeadlineExceeded}; the executor may then never have run, and the taskmaster stops the task.
+     */
+    private static boolean ignoresError(V1Job job) {
+        boolean annotated =
+                Optional.ofNullable(job.getMetadata())
+                        .map(V1ObjectMeta::getAnnotations)
+                        .map(
+                                annotations ->
+                                        ANN_IGNORE_ERROR_VALUE.equals(
+                                                annotations.get(ANN_IGNORE_ERROR_KEY)))
+                        .orElse(false);
+        return annotated && !endedByDeadline(job);
+    }
+
+    /**
+     * Whether an executor's job is done, once the taskmaster's job has completed: it succeeded, or
+     * it failed and its failure is ignored ({@link #ignoresError}).
+     */
+    private static boolean isDone(V1Job job) {
+        // The taskmaster's job has ended, so failed pods alone count as a failure
+        boolean taskmasterEnded = true;
+        return hasSucceeded(job) || (ignoresError(job) && hasFailed(job, taskmasterEnded));
+    }
+
+    /**
+     * Whether the taskmaster ended the job itself: it fails a job whose pod can't start or be
+     * scheduled by setting {@code activeDeadlineSeconds} (to 1, {@code FAIL_DEADLINE} in
+     * tesk-core's {@code job.py}), so that Kubernetes fails it with the reason {@code
+     * DeadlineExceeded}. TESK API sets no deadline on executor jobs.
+     */
+    private static boolean endedByDeadline(V1Job job) {
+        boolean hasDeadline =
+                Optional.ofNullable(job.getSpec())
+                        .map(V1JobSpec::getActiveDeadlineSeconds)
+                        .isPresent();
+        List<V1JobCondition> conditions =
+                Optional.ofNullable(job.getStatus())
+                        .map(V1JobStatus::getConditions)
+                        .orElse(Collections.emptyList());
+        return hasDeadline
+                || conditions.stream()
+                        .anyMatch(condition -> "DeadlineExceeded".equals(condition.getReason()));
+    }
+
     private static boolean isCancelled(V1Job job) {
         return isLabelledCancelled(job.getMetadata());
     }

src/main/java/uk/ac/ebi/tsc/tesk/k8s/convert/TesKubernetesConverter.java

@@ -263,6 +263,10 @@ public V1Job fromTesExecutorToK8sJob(
         job.getMetadata()
                 .putLabelsItem(LABEL_EXECNO_KEY, Integer.valueOf(executorIndex).toString());
         job.getMetadata().putAnnotationsItem(ANN_TESTASK_NAME_KEY, tesTaskName);
+        // The taskmaster goes on after such an executor fails (issue #159)
+        if (Boolean.TRUE.equals(executor.getIgnoreError())) {
+            job.getMetadata().putAnnotationsItem(ANN_IGNORE_ERROR_KEY, ANN_IGNORE_ERROR_VALUE);
+        }
         CreatorLabels.put(job.getMetadata(), user.getUsername());
         V1Container container = job.getSpec().getTemplate().getSpec().getContainers().get(0);
         // Tries to convert potential TRS URI into docker image. If conversion unsuccessful ->
@@ -342,9 +346,26 @@ public TesCreateTaskResponse fromK8sJobToTesCreateTaskResponse(V1Job job) {
      */
     public TesState extractStateFromK8sJobs(Task taskmasterWithExecutors) {
         // the stored task is read only if needed: once the taskmaster's job completed
+        V1Job taskmasterJob = taskmasterWithExecutors.getTaskmaster().getJob();
         return TaskStateResolver.resolve(
                 taskmasterWithExecutors,
-                () -> this.extractExecutorCount(taskmasterWithExecutors.getTaskmaster().getJob()));
+                () -> this.extractExecutorCount(taskmasterJob),
+                () -> this.extractHasOutputs(taskmasterJob));
+    }
+
+    /**
+     * Reads whether the task stored with the taskmaster's job (annotation {@code json-input}) has
+     * outputs.
+     *
+     * @param taskMasterJob - taskmaster's job
+     * @return whether the task has outputs, or null if the task isn't stored or can't be read
+     */
+    Boolean extractHasOutputs(V1Job taskMasterJob) {
+        if (this.readStoredTaskField(taskMasterJob, "executors") == null) {
+            return null;
+        }
+        JsonNode outputs = this.readStoredTaskField(taskMasterJob, "outputs");
+        return outputs != null && outputs.isArray() && outputs.size() > 0;
     }
 
     /**

src/main/java/uk/ac/ebi/tsc/tesk/tes/model/TesExecutor.java

@@ -254,10 +254,11 @@ public TesExecutor ignoreError(Boolean ignoreError) {
 
   /**
    * Whether to run the next executor, and finish the task, after this executor fails (TES 1.1).
-   * TESK stops at the first executor that fails, so it accepts only `false`.
+   * With `true`, the executor's job gets the annotation `ignore-error`, so that the taskmaster goes
+   * on after it fails (see `TaskStateResolver`).
    * @return ignoreError
   */
-  @ApiModelProperty(value = "Whether errors of this executor are ignored, and the next one runs. TESK doesn't support it: only false is accepted.")
+  @ApiModelProperty(value = "Whether errors of this executor are ignored: if it fails, the next executor runs, and the task can still complete.")
   public Boolean getIgnoreError() {
     return ignoreError;
   }
@@ -266,16 +267,6 @@ public void setIgnoreError(Boolean ignoreError) {
     this.ignoreError = ignoreError;
   }
 
-  /**
-   * Refuses `ignore_error: true`, rather than running the task differently than asked.
-   * @return whether the executor doesn't ask to ignore its errors
-   */
-  @JsonIgnore
-  @AssertTrue(message = "ignore_error: true is not supported; TESK stops at the first executor that fails")
-  public boolean isIgnoreErrorUnset() {
-    return ignoreError == null || !ignoreError;
-  }
-
   public TesExecutor env(Map<String, String> env) {
     this.env = env;
     return this;