Support for v1.1 TimeLimit, WorkReuse and NetworkAccess (#718)

* Kill jobs that exceed timeout (TimeLimit)
* Enable/disable network (NetworkAccess)
* Enable/disable caching (WorkReuse)
* Support v1.1 and as a prefixed backport in v1.0

Author: tetron

cwltool/argparser.py

@@ -208,12 +208,9 @@ def arg_parser():  # type: () -> argparse.ArgumentParser
                         help="Specify a default docker container that will be used if the workflow fails to specify one.")
     parser.add_argument("--no-match-user", action="store_true",
                         help="Disable passing the current uid to `docker run --user`")
-    parser.add_argument("--disable-net", action="store_true",
-                        help="Use docker's default networking for containers;"
-                             " the default is to enable networking.")
     parser.add_argument("--custom-net", type=Text,
-                        help="Will be passed to `docker run` as the '--net' "
-                             "parameter. Implies '--enable-net'.")
+                        help="Passed to `docker run` as the '--net' "
+                             "parameter when NetworkAccess is true.")
     parser.add_argument("--disable-validate", dest="do_validate",
                         action="store_false", default=True,
                         help=argparse.SUPPRESS)

cwltool/command_line_tool.py

@@ -250,8 +250,15 @@ def job(self,
             ):
         # type: (...) -> Generator[Union[JobBase, CallbackJob], None, None]
 
+        require_prefix = ""
+        if self.metadata["cwlVersion"] == "v1.0":
+            require_prefix = "http://commonwl.org/cwltool#"
+
+        workReuse = self.get_requirement(require_prefix+"WorkReuse")[0]
+        enableReuse = workReuse.get("enableReuse", True) if workReuse else True
+
         jobname = uniquename(kwargs.get("name", shortname(self.tool.get("id", "job"))))
-        if kwargs.get("cachedir"):
+        if kwargs.get("cachedir") and enableReuse:
             cacheargs = kwargs.copy()
             cacheargs["outdir"] = "/out"
             cacheargs["tmpdir"] = "/tmp"
@@ -487,6 +494,22 @@ def register_reader(f):
             adjustDirObjs(builder.files, register_reader)
             adjustDirObjs(builder.bindings, register_reader)
 
+        timelimit = self.get_requirement(require_prefix+"TimeLimit")[0]
+        if timelimit:
+            with SourceLine(timelimit, "timelimit", validate.ValidationException, debug):
+                j.timelimit = builder.do_eval(timelimit["timelimit"])
+                if not isinstance(j.timelimit, int) or j.timelimit < 0:
+                    raise Exception("timelimit must be an integer >= 0, got: %s" % j.timelimit)
+
+        if self.metadata["cwlVersion"] == "v1.0":
+            j.networkaccess = True
+        networkaccess = self.get_requirement(require_prefix+"NetworkAccess")[0]
+        if networkaccess:
+            with SourceLine(networkaccess, "networkAccess", validate.ValidationException, debug):
+                j.networkaccess = builder.do_eval(networkaccess["networkAccess"])
+                if not isinstance(j.networkaccess, bool):
+                    raise Exception("networkAccess must be a boolean, got: %s" % j.networkaccess)
+
         j.environment = {}
         evr = self.get_requirement("EnvVarRequirement")[0]
         if evr:

cwltool/docker.py

@@ -267,9 +267,10 @@ def create_runtime(self, env, rm_container=True, record_container_id=False, cidf
             if not kwargs.get("no_read_only"):
                 runtime.append(u"--read-only=true")
 
-            if kwargs.get("custom_net", None) is not None:
-                runtime.append(u"--net={0}".format(kwargs.get("custom_net")))
-            elif kwargs.get("disable_net", None):
+            if self.networkaccess:
+                if kwargs.get("custom_net"):
+                    runtime.append(u"--net={0}".format(kwargs["custom_net"]))
+            else:
                 runtime.append(u"--net=none")
 
             if self.stdout:

cwltool/extensions.yml

@@ -53,4 +53,85 @@ $graph:
         which will be deliberately obscured from logging.
       jsonldPredicate:
         "_type": "@id"
-        refScope: 0
+        refScope: 0
+
+
+- type: record
+  name: TimeLimit
+  inVocab: false
+  extends: cwl:ProcessRequirement
+  doc: |
+    Set an upper limit on the execution time of a CommandLineTool or
+    ExpressionTool.  A tool execution which exceeds the time limit may
+    be preemptively terminated and considered failed.  May also be
+    used by batch systems to make scheduling decisions.
+  fields:
+    - name: class
+      type: string
+      doc: "Always 'TimeLimit'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    - name: timelimit
+      type: [long, string]
+      doc: |
+        The time limit, in seconds.  A time limit of zero means no
+        time limit.  Negative time limits are an error.
+
+
+- type: record
+  name: WorkReuse
+  inVocab: false
+  extends: cwl:ProcessRequirement
+  doc: |
+    For implementations that support reusing output from past work (on
+    the assumption that same code and same input produce same
+    results), control whether to enable or disable the reuse behavior
+    for a particular tool or step (to accomodate situations where that
+    assumption is incorrect).  A reused step is not executed but
+    instead returns the same output as the original execution.
+
+    If `enableReuse` is not specified, correct tools should assume it
+    is enabled by default.
+  fields:
+    - name: class
+      type: string
+      doc: "Always 'WorkReuse'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    - name: enableReuse
+      type: [boolean, string]
+      #default: true
+
+
+- type: record
+  name: NetworkAccess
+  inVocab: false
+  extends: cwl:ProcessRequirement
+  doc: |
+    Indicate whether a process requires outgoing IPv4/IPv6 network
+    access.  Choice of IPv4 or IPv6 is implementation and site
+    specific, correct tools must support both.
+
+    If `networkAccess` is false or not specified, tools must not
+    assume network access, except for localhost (the loopback device).
+
+    If `networkAccess` is true, the tool must be able to make outgoing
+    connections to network resources.  Resources may be on a private
+    subnet or the public Internet.  However, implementations and sites
+    may apply their own security policies to restrict what is
+    accessible by the tool.
+
+    Enabling network access does not imply a publically routable IP
+    address or the ability to accept inbound connections.
+
+  fields:
+    - name: class
+      type: string
+      doc: "Always 'NetworkAccess'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    - name: networkAccess
+      type: [boolean, string]

cwltool/job.py

@@ -11,6 +11,7 @@
 import stat
 import sys
 import tempfile
+import threading
 from abc import ABCMeta, abstractmethod
 from io import open
 from threading import Lock
@@ -154,6 +155,8 @@ def __init__(self):  # type: () -> None
         self.generatefiles = None  # type: Dict[Text, Union[List[Dict[Text, Text]], Dict[Text, Text], Text]]
         self.stagedir = None  # type: Text
         self.inplace_update = None  # type: bool
+        self.timelimit = None  # type: int
+        self.networkaccess = False  # type: bool
 
     def _setup(self, kwargs):  # type: (Dict) -> None
         if not os.path.exists(self.outdir):
@@ -235,9 +238,17 @@ def _execute(self,
             if builder is not None:
                 job_script_contents = builder.build_job_script(commands)
             rcode = _job_popen(
-                commands, stdin_path, stdout_path, stderr_path, env,
-                self.outdir, tempfile.mkdtemp(prefix=tmp_outdir_prefix),
-                job_script_contents)
+                commands,
+                stdin_path=stdin_path,
+                stdout_path=stdout_path,
+                stderr_path=stderr_path,
+                env=env,
+                cwd=self.outdir,
+                job_dir=tempfile.mkdtemp(prefix=tmp_outdir_prefix),
+                job_script_contents=job_script_contents,
+                timelimit=self.timelimit,
+                name=self.name
+            )
 
             if self.successCodes and rcode in self.successCodes:
                 processStatus = "success"
@@ -429,7 +440,10 @@ def _job_popen(
         cwd,                       # type: Text
         job_dir,                   # type: Text
         job_script_contents=None,  # type: Text
+        timelimit=None,            # type: int
+        name=None                  # type: Text
        ):  # type: (...) -> int
+
     if not job_script_contents and not FORCE_SHELLED_POPEN:
 
         stdin = None  # type: Union[IO[Any], int]
@@ -463,8 +477,22 @@ def _job_popen(
         if sp.stdin:
             sp.stdin.close()
 
+        tm = None
+        if timelimit:
+            def terminate():
+                try:
+                    _logger.warn(u"[job %s] exceeded time limit of %d seconds and will be terminated", name, timelimit)
+                    sp.terminate()
+                except OSError:
+                    pass
+            tm = threading.Timer(timelimit, terminate)
+            tm.start()
+
         rcode = sp.wait()
 
+        if tm:
+            tm.cancel()
+
         if isinstance(stdin, io.IOBase):
             stdin.close()
 

cwltool/pathmapper.py

@@ -17,7 +17,7 @@
 from schema_salad.sourceline import SourceLine
 from six.moves import urllib
 
-from .utils import convert_pathsep_to_unix
+from .utils import convert_pathsep_to_unix, visit_class
 
 from .stdfsaccess import StdFsAccess, abspath
 
@@ -38,17 +38,6 @@ def adjustFiles(rec, op):  # type: (Any, Union[Callable[..., Any], partial[Any]]
         for d in rec:
             adjustFiles(d, op)
 
-def visit_class(rec, cls, op):  # type: (Any, Iterable, Union[Callable[..., Any], partial[Any]]) -> None
-    """Apply a function to with "class" in cls."""
-
-    if isinstance(rec, dict):
-        if "class" in rec and rec.get("class") in cls:
-            op(rec)
-        for d in rec:
-            visit_class(rec[d], cls, op)
-    if isinstance(rec, list):
-        for d in rec:
-            visit_class(d, cls, op)
 
 def adjustFileObjs(rec, op):  # type: (Any, Union[Callable[..., Any], partial[Any]]) -> None
     """Apply an update function to each File object in the object `rec`."""

cwltool/process.py

@@ -71,6 +71,12 @@ def filter(self, record):
                                 "StepInputExpressionRequirement",
                                 "ResourceRequirement",
                                 "InitialWorkDirRequirement",
+                                "TimeLimit",
+                                "WorkReuse",
+                                "NetworkAccess",
+                                "http://commonwl.org/cwltool#TimeLimit",
+                                "http://commonwl.org/cwltool#WorkReuse",
+                                "http://commonwl.org/cwltool#NetworkAccess",
                                 "http://commonwl.org/cwltool#LoadListingRequirement",
                                 "http://commonwl.org/cwltool#InplaceUpdateRequirement"]
 

cwltool/schemas/v1.1.0-dev1/CommandLineTool.yml

@@ -990,3 +990,81 @@ $graph:
     - name: outdirMax
       type: ["null", long, string, Expression]
       doc: Maximum reserved filesystem based storage for the designated output directory, in mebibytes (2**20)
+
+
+- type: record
+  name: TimeLimit
+  extends: ProcessRequirement
+  doc: |
+    Set an upper limit on the execution time of a CommandLineTool or
+    ExpressionTool.  A tool execution which exceeds the time limit may
+    be preemptively terminated and considered failed.  May also be
+    used by batch systems to make scheduling decisions.
+  fields:
+    - name: class
+      type: string
+      doc: "Always 'TimeLimit'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    - name: timelimit
+      type: [long, string, Expression]
+      doc: |
+        The time limit, in seconds.  A time limit of zero means no
+        time limit.  Negative time limits are an error.
+
+
+- type: record
+  name: WorkReuse
+  extends: ProcessRequirement
+  doc: |
+    For implementations that support reusing output from past work (on
+    the assumption that same code and same input produce same
+    results), control whether to enable or disable the reuse behavior
+    for a particular tool or step (to accomodate situations where that
+    assumption is incorrect).  A reused step is not executed but
+    instead returns the same output as the original execution.
+
+    If `enableReuse` is not specified, correct tools should assume it
+    is enabled by default.
+  fields:
+    - name: class
+      type: string
+      doc: "Always 'WorkReuse'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    - name: enableReuse
+      type: [boolean, string, Expression]
+      default: true
+
+
+- type: record
+  name: NetworkAccess
+  extends: ProcessRequirement
+  doc: |
+    Indicate whether a process requires outgoing IPv4/IPv6 network
+    access.  Choice of IPv4 or IPv6 is implementation and site
+    specific, correct tools must support both.
+
+    If `networkAccess` is false or not specified, tools must not
+    assume network access, except for localhost (the loopback device).
+
+    If `networkAccess` is true, the tool must be able to make outgoing
+    connections to network resources.  Resources may be on a private
+    subnet or the public Internet.  However, implementations and sites
+    may apply their own security policies to restrict what is
+    accessible by the tool.
+
+    Enabling network access does not imply a publically routable IP
+    address or the ability to accept inbound connections.
+
+  fields:
+    - name: class
+      type: string
+      doc: "Always 'NetworkAccess'"
+      jsonldPredicate:
+        "_id": "@type"
+        "_type": "@vocab"
+    - name: networkAccess
+      type: [boolean, string, Expression]

cwltool/update.py

@@ -13,7 +13,7 @@
 from ruamel.yaml.comments import CommentedMap, CommentedSeq
 from schema_salad.ref_resolver import Loader
 
-from .utils import aslist
+from .utils import aslist, visit_class
 
 def findId(doc, frg):  # type: (Any, Any) -> Dict
     if isinstance(doc, dict):
@@ -112,6 +112,16 @@ def v1_0dev4to1_0(doc, loader, baseuri):
 def v1_0to1_1_0dev1(doc, loader, baseuri):
     # type: (Any, Loader, Text) -> Tuple[Any, Text]
     """Public updater for v1.0 to v1.1.0-dev1."""
+
+    def add_networkaccess(t):
+        t.setdefault("requirements", [])
+        t["requirements"].append({
+            "class": "NetworkAccess",
+            "networkAccess": True
+            })
+
+    visit_class(doc, ("CommandLineTool",), add_networkaccess)
+
     return (doc, "v1.1.0-dev1")