Refactor job.py to allow modification of job environment w/shell commands.

Author: jmchilton

cwltool/job.py

@@ -7,6 +7,7 @@
 import yaml
 import logging
 import sys
+import string
 import requests
 from . import docker
 from .process import get_feature, empty_subtree
@@ -25,6 +26,9 @@
 
 needs_shell_quoting_re = re.compile(r"""(^$|[\s|&;()<>\'"$@])""")
 
+FORCE_SHELLED_POPEN = os.getenv("CWLTOOL_FORCE_SHELL_POPEN", "1") == "1"
+
+
 def deref_links(outputs):  # type: (Any) -> None
     if isinstance(outputs, dict):
         if outputs.get("class") == "File":
@@ -177,50 +181,36 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
                     raise Exception("Unhandled type %s", type(entry))
 
             if self.stdin:
-                stdin = open(self.pathmapper.mapper(self.stdin)[0], "rb")
+                stdin_path = self.pathmapper.mapper(self.stdin)[0]
             else:
-                stdin = subprocess.PIPE
+                stdin_path = None
 
             if self.stderr:
                 abserr = os.path.join(self.outdir, self.stderr)
                 dnerr = os.path.dirname(abserr)
                 if dnerr and not os.path.exists(dnerr):
                     os.makedirs(dnerr)
-                stderr = open(abserr, "wb")
+                stderr_path = abserr
             else:
-                stderr = sys.stderr
+                stderr_path = None
 
             if self.stdout:
                 absout = os.path.join(self.outdir, self.stdout)
                 dn = os.path.dirname(absout)
                 if dn and not os.path.exists(dn):
                     os.makedirs(dn)
-                stdout = open(absout, "wb")
+                stdout_path = absout
             else:
-                stdout = sys.stderr
-
-            sp = subprocess.Popen([str(x) for x in runtime + self.command_line],
-                                  shell=False,
-                                  close_fds=True,
-                                  stdin=stdin,
-                                  stderr=stderr,
-                                  stdout=stdout,
-                                  env=env,
-                                  cwd=self.outdir)
-
-            if sp.stdin:
-                sp.stdin.close()
-
-            rcode = sp.wait()
-
-            if isinstance(stdin, file):
-                stdin.close()
+                stdout_path = None
 
-            if stderr is not sys.stderr:
-                stderr.close()
-
-            if stdout is not sys.stderr:
-                stdout.close()
+            rcode = shelled_popen(
+                [str(x) for x in runtime + self.command_line],
+                stdin_path=stdin_path,
+                stdout_path=stdout_path,
+                stderr_path=stderr_path,
+                env=env,
+                cwd=self.outdir,
+            )
 
             if self.successCodes and rcode in self.successCodes:
                 processStatus = "success"
@@ -274,3 +264,152 @@ def run(self, dry_run=False, pull_image=True, rm_container=True,
         if move_outputs and empty_subtree(self.outdir):
             _logger.debug(u"[job %s] Removing empty output directory %s", self.name, self.outdir)
             shutil.rmtree(self.outdir, True)
+
+
+SHELL_COMMAND_TEMPLATE = string.Template("""#!/bin/bash
+$prefix
+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, "rd")
+    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,
+                          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 shelled_popen(commands,
+                  stdin_path,
+                  stdout_path,
+                  stderr_path,
+                  env,
+                  cwd,
+                  prefix=None):
+    if prefix is None and not FORCE_SHELLED_POPEN:
+        if stdin_path is not None:
+            stdin = open(stdin_path, "rd")
+        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()
+
+        return rcode
+    else:
+        template_kwds = dict(
+            prefix=prefix or '',
+        )
+        job_script_contents = SHELL_COMMAND_TEMPLATE.substitute(
+            **template_kwds
+        )
+        job_dir = tempfile.mkdtemp(prefix="cwltooljob")
+        job_description = dict(
+            commands=commands,
+            cwd=cwd,
+            env=env,
+            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],
+                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
+        finally:
+            shutil.rmtree(job_dir)