Merge branch '8784-dir-listings'

[arvados.git] / crunch_scripts / run-command

diff --git a/crunch_scripts/run-command b/crunch_scripts/run-command
index c07debd787eecfc6696c5d614d69c013f8028dd7..3fd08bf28b714cbab425c7fa7cb404946c104133 100755 (executable)
--- a/crunch_scripts/run-command
+++ b/crunch_scripts/run-command
@@ -1,4 +1,7 @@
 #!/usr/bin/env python
+# Copyright (C) The Arvados Authors. All rights reserved.
+#
+# SPDX-License-Identifier: Apache-2.0
 
 import logging
 
@@ -57,12 +60,10 @@ else:
     jobp = json.loads(args.script_parameters)
     os.environ['JOB_UUID'] = 'zzzzz-8i9sb-1234567890abcde'
     os.environ['TASK_UUID'] = 'zzzzz-ot0gb-1234567890abcde'
-    os.environ['CRUNCH_SRC'] = '/tmp/crunche-src'
+    os.environ['CRUNCH_SRC'] = '/tmp/crunch-src'
     if 'TASK_KEEPMOUNT' not in os.environ:
         os.environ['TASK_KEEPMOUNT'] = '/keep'
 
-links = []
-
 def sub_tmpdir(v):
     return os.path.join(arvados.current_task().tmpdir, 'tmpdir')
 
@@ -333,6 +334,13 @@ try:
         if not args.dry_run:
             stdoutfile = open(stdoutname, "wb")
 
+    if "task.env" in taskp:
+        env = copy.copy(os.environ)
+        for k,v in taskp["task.env"].items():
+            env[k] = subst.do_substitution(taskp, v)
+    else:
+        env = None
+
     logger.info("{}{}{}".format(' | '.join([' '.join(c) for c in cmd]), (" < " + stdinname) if stdinname is not None else "", (" > " + stdoutname) if stdoutname is not None else ""))
 
     if args.dry_run:
@@ -365,7 +373,7 @@ try:
             # this is an intermediate command in the pipeline, so its stdout should go to a pipe
             next_stdout = subprocess.PIPE
 
-        sp = subprocess.Popen(cmd[i], shell=False, stdin=next_stdin, stdout=next_stdout)
+        sp = subprocess.Popen(cmd[i], shell=False, stdin=next_stdin, stdout=next_stdout, env=env)
 
         # Need to close the FDs on our side so that subcommands will get SIGPIPE if the
         # consuming process ends prematurely.
@@ -392,12 +400,19 @@ try:
     active = 1
     pids = set([s.pid for s in subprocesses])
     while len(pids) > 0:
-        (pid, status) = os.wait()
-        pids.discard(pid)
-        if not taskp.get("task.ignore_rcode"):
-            rcode[pid] = (status >> 8)
+        try:
+            (pid, status) = os.wait()
+        except OSError as e:
+            if e.errno == errno.EINTR:
+                pass
+            else:
+                raise
         else:
-            rcode[pid] = 0
+            pids.discard(pid)
+            if not taskp.get("task.ignore_rcode"):
+                rcode[pid] = (status >> 8)
+            else:
+                rcode[pid] = 0
 
     if sig.sig is not None:
         logger.critical("terminating on signal %s" % sig.sig)
@@ -415,31 +430,27 @@ signal.signal(signal.SIGINT, signal.SIG_DFL)
 signal.signal(signal.SIGTERM, signal.SIG_DFL)
 signal.signal(signal.SIGQUIT, signal.SIG_DFL)
 
-for l in links:
-    os.unlink(l)
-
 logger.info("the following output files will be saved to keep:")
 
-subprocess.call(["find", ".", "-type", "f", "-printf", "run-command: %12.12s %h/%f\\n"], stdout=sys.stderr)
+subprocess.call(["find", "-L", ".", "-type", "f", "-printf", "run-command: %12.12s %h/%f\\n"], stdout=sys.stderr, cwd=outdir)
 
 logger.info("start writing output to keep")
 
-if "task.vwd" in taskp:
-    if "task.foreach" in jobp:
-        # This is a subtask, so don't merge with the original collection, that will happen at the end
-        outcollection = vwd.checkin(subst.do_substitution(taskp, taskp["task.vwd"]), outdir, merge=False).manifest_text()
-    else:
-        # Just a single task, so do merge with the original collection
-        outcollection = vwd.checkin(subst.do_substitution(taskp, taskp["task.vwd"]), outdir, merge=True).manifest_text()
-else:
-    outcollection = robust_put.upload(outdir, logger)
+if "task.vwd" in taskp and "task.foreach" in jobp:
+    for root, dirs, files in os.walk(outdir):
+        for f in files:
+            s = os.lstat(os.path.join(root, f))
+            if stat.S_ISLNK(s.st_mode):
+                os.unlink(os.path.join(root, f))
+
+(outcollection, checkin_error) = vwd.checkin(outdir)
 
 # Success if we ran any subprocess, and they all exited 0.
-success = rcode and all(status == 0 for status in rcode.itervalues())
+success = rcode and all(status == 0 for status in rcode.itervalues()) and not checkin_error
 
 api.job_tasks().update(uuid=arvados.current_task()['uuid'],
                                      body={
-                                         'output': outcollection,
+                                         'output': outcollection.manifest_text(),
                                          'success': success,
                                          'progress':1.0
                                      }).execute()