Adding kill switch trigger. If a job's processStatus != "success" and runtimeContext.on_error = "kill", then the switch is activated. WorkflowKillSwitch is raised so it can be handled at the workflow and executor levels

Author: AlexTate

cwltool/errors.py

@@ -18,3 +18,14 @@ class UnsupportedRequirement(WorkflowException):
 
 class ArgumentException(Exception):
     """Mismatched command line arguments provided."""
+
+
+class WorkflowKillSwitch(Exception):
+    """When processStatus != "success" and on-error=kill, raise this exception."""
+
+    def __init__(self, job_id, rcode):
+        self.job_id = job_id
+        self.rcode = rcode
+
+    def __str__(self):
+        return f'[job {self.job_id}] activated kill switch with return code {self.rcode}'

cwltool/job.py

@@ -30,7 +30,7 @@
 from .builder import Builder
 from .context import RuntimeContext
 from .cuda import cuda_check
-from .errors import UnsupportedRequirement, WorkflowException
+from .errors import UnsupportedRequirement, WorkflowException, WorkflowKillSwitch
 from .loghandler import _logger
 from .pathmapper import MapperEnt, PathMapper
 from .process import stage_files
@@ -348,14 +348,19 @@ def stderr_stdout_log_path(
                 processStatus = "permanentFail"
 
             if processStatus != "success":
-                if rcode < 0:
+                if runtimeContext.kill_switch.is_set():
+                    return
+                elif rcode < 0:
                     _logger.warning(
                         "[job %s] was terminated by signal: %s",
                         self.name,
                         signal.Signals(-rcode).name,
                     )
                 else:
                     _logger.warning("[job %s] exited with status: %d", self.name, rcode)
+                if runtimeContext.on_error == "kill":
+                    runtimeContext.kill_switch.set()
+                    raise WorkflowKillSwitch(self.name, rcode)
 
             if "listing" in self.generatefiles:
                 if self.generatemapper:
@@ -386,6 +391,8 @@ def stderr_stdout_log_path(
         except WorkflowException as err:
             _logger.error("[job %s] Job error:\n%s", self.name, str(err))
             processStatus = "permanentFail"
+        except WorkflowKillSwitch:
+            raise
         except Exception:
             _logger.exception("Exception while running job")
             processStatus = "permanentFail"