Add an extension point allowing building scripts for cwltool jobs.

Author: jmchilton

cwltool/builder.py

@@ -35,6 +35,7 @@ def __init__(self):  # type: () -> None
         self.pathmapper = None  # type: PathMapper
         self.stagedir = None  # type: Text
         self.make_fs_access = None  # type: Type[StdFsAccess]
+        self.build_job_script = None  # type: Callable[[List[str]], Text]
 
     def bind_input(self, schema, datum, lead_pos=[], tail_pos=[]):
         # type: (Dict[Text, Any], Any, Union[int, List[int]], List[int]) -> List[Dict[Text, Any]]

cwltool/job.py

@@ -14,6 +14,7 @@
 import stat
 import re
 import shellescape
+import string
 from .docker_uid import docker_vm_uid
 from .builder import Builder
 from typing import (Any, Callable, Union, Iterable, Mapping, MutableMapping,
@@ -25,6 +26,58 @@
 
 needs_shell_quoting_re = re.compile(r"""(^$|[\s|&;()<>\'"$@])""")
 
+FORCE_SHELLED_POPEN = os.getenv("CWLTOOL_FORCE_SHELL_POPEN", "1") == "1"
+
+SHELL_COMMAND_TEMPLATE = """#!/bin/bash
+python "run_job.py" "job.json"
+"""
+
+PYTHON_RUN_SCRIPT = """
+import json
+import sys
+import subprocess
+
+with open(sys.argv[1], "r") as f:
+    popen_description = json.load(f)
+    commands = popen_description["commands"]
+    cwd = popen_description["cwd"]
+    env = popen_description["env"]
+    stdin_path = popen_description["stdin_path"]
+    stdout_path = popen_description["stdout_path"]
+    stderr_path = popen_description["stderr_path"]
+    if stdin_path is not None:
+        stdin = open(stdin_path, "rb")
+    else:
+        stdin = subprocess.PIPE
+    if stdout_path is not None:
+        stdout = open(stdout_path, "wb")
+    else:
+        stdout = sys.stderr
+    if stderr_path is not None:
+        stderr = open(stderr_path, "wb")
+    else:
+        stderr = sys.stderr
+    sp = subprocess.Popen(commands,
+                          shell=False,
+                          close_fds=True,
+                          stdin=stdin,
+                          stdout=stdout,
+                          stderr=stderr,
+                          env=env,
+                          cwd=cwd)
+    if sp.stdin:
+        sp.stdin.close()
+    rcode = sp.wait()
+    if isinstance(stdin, file):
+        stdin.close()
+    if stdout is not sys.stderr:
+        stdout.close()
+    if stderr is not sys.stderr:
+        stderr.close()
+    sys.exit(rcode)
+"""
+
+
 def deref_links(outputs):  # type: (Any) -> None
     if isinstance(outputs, dict):
         if outputs.get("class") == "File":
@@ -207,13 +260,15 @@ def linkoutdir(src, tgt):
                     os.makedirs(dn)
                 stdout_path = absout
 
+            build_job_script = self.builder.build_job_script  # type: Callable[[List[str]], Text]
             rcode = _job_popen(
                 [Text(x).encode('utf-8') for x in runtime + self.command_line],
                 stdin_path=stdin_path,
                 stdout_path=stdout_path,
                 stderr_path=stderr_path,
                 env=env,
                 cwd=self.outdir,
+                build_job_script=build_job_script,
             )
 
             if self.successCodes and rcode in self.successCodes:
@@ -282,49 +337,102 @@ def _job_popen(
     stderr_path,
     env,
     cwd,
+    job_dir=None,
+    build_job_script=None,
 ):
-    # type: (List[str], Text, Text, Text, Union[MutableMapping[Text, Text], MutableMapping[str, str]], Text) -> int
+    # type: (List[str], Text, Text, Text, Union[MutableMapping[Text, Text], MutableMapping[str, str]], Text, Text, Callable[[List[str]], Text]) -> int
 
-    stdin = None  # type: Union[IO[Any], int]
-    stderr = None  # type: IO[Any]
-    stdout = None  # type: IO[Any]
+    job_script_contents = None  # type: Text
+    if build_job_script:
+        job_script_contents = build_job_script(commands)
 
-    if stdin_path is not None:
-        stdin = open(stdin_path, "rb")
-    else:
-        stdin = subprocess.PIPE
+    if not job_script_contents and not FORCE_SHELLED_POPEN:
 
-    if stdout_path is not None:
-        stdout = open(stdout_path, "wb")
-    else:
-        stdout = sys.stderr
+        stdin = None  # type: Union[IO[Any], int]
+        stderr = None  # type: IO[Any]
+        stdout = None  # type: IO[Any]
 
-    if stderr_path is not None:
-        stderr = open(stderr_path, "wb")
-    else:
-        stderr = sys.stderr
+        if stdin_path is not None:
+            stdin = open(stdin_path, "rb")
+        else:
+            stdin = subprocess.PIPE
 
-    sp = subprocess.Popen(commands,
-                          shell=False,
-                          close_fds=True,
-                          stdin=stdin,
-                          stdout=stdout,
-                          stderr=stderr,
-                          env=env,
-                          cwd=cwd)
+        if stdout_path is not None:
+            stdout = open(stdout_path, "wb")
+        else:
+            stdout = sys.stderr
 
-    if sp.stdin:
-        sp.stdin.close()
+        if stderr_path is not None:
+            stderr = open(stderr_path, "wb")
+        else:
+            stderr = sys.stderr
 
-    rcode = sp.wait()
+        sp = subprocess.Popen(commands,
+                              shell=False,
+                              close_fds=True,
+                              stdin=stdin,
+                              stdout=stdout,
+                              stderr=stderr,
+                              env=env,
+                              cwd=cwd)
 
-    if isinstance(stdin, file):
-        stdin.close()
+        if sp.stdin:
+            sp.stdin.close()
 
-    if stdout is not sys.stderr:
-        stdout.close()
+        rcode = sp.wait()
 
-    if stderr is not sys.stderr:
-        stderr.close()
+        if isinstance(stdin, file):
+            stdin.close()
+
+        if stdout is not sys.stderr:
+            stdout.close()
+
+        if stderr is not sys.stderr:
+            stderr.close()
+
+        return rcode
+    else:
+        if job_dir is None:
+            job_dir = tempfile.mkdtemp(prefix="cwltooljob")
+
+        if not job_script_contents:
+            job_script_contents = SHELL_COMMAND_TEMPLATE
+
+        env_copy = {}
+        for key in env:
+            key = key.encode("utf-8")
+            env_copy[key] = env[key]
+
+        job_description = dict(
+            commands=commands,
+            cwd=cwd,
+            env=env_copy,
+            stdout_path=stdout_path,
+            stderr_path=stderr_path,
+            stdin_path=stdin_path,
+        )
+        with open(os.path.join(job_dir, "job.json"), "w") as f:
+            json.dump(job_description, f)
+        try:
+            job_script = os.path.join(job_dir, "run_job.bash")
+            with open(job_script, "w") as f:
+                f.write(job_script_contents)
+            job_run = os.path.join(job_dir, "run_job.py")
+            with open(job_run, "w") as f:
+                f.write(PYTHON_RUN_SCRIPT)
+            sp = subprocess.Popen(
+                ["bash", job_script.encode("utf-8")],
+                shell=False,
+                cwd=job_dir,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                stdin=subprocess.PIPE,
+            )
+            if sp.stdin:
+                sp.stdin.close()
+
+            rcode = sp.wait()
 
-    return rcode
+            return rcode
+        finally:
+            shutil.rmtree(job_dir)