Record logs from arvados-cwl-runner to report them back to WES.

Author: Peter Amstutz

wes_client/__init__.py

@@ -139,7 +139,7 @@ def fixpaths(d):
 
     s = client.WorkflowExecutionService.GetWorkflowLog(
         workflow_id=r["workflow_id"]).result()
-    logging.info(s["workflow_log"]["stderr"])
+    logging.info("Workflow log:\n"+s["workflow_log"]["stderr"])
 
     if "fields" in s["outputs"] and s["outputs"]["fields"] is None:
         del s["outputs"]["fields"]

wes_service/arvados_wes.py

@@ -82,20 +82,29 @@ def ListWorkflows(self):
         }
 
     def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params, env):
+        api = arvados.api_from_config(version="v1", apiconfig={
+            "ARVADOS_API_HOST": env["ARVADOS_API_HOST"],
+            "ARVADOS_API_TOKEN": env['ARVADOS_API_TOKEN'],
+            "ARVADOS_API_HOST_INSECURE": env["ARVADOS_API_HOST_INSECURE"]  # NOQA
+        })
+
         try:
             with tempfile.NamedTemporaryFile() as inputtemp:
                 json.dump(workflow_params, inputtemp)
                 inputtemp.flush()
-                workflow_id = subprocess.check_output(["arvados-cwl-runner", "--submit-request-uuid="+cr_uuid, # NOQA
+                proc = subprocess.Popen(["arvados-cwl-runner", "--submit-request-uuid="+cr_uuid, # NOQA
                                                        "--submit", "--no-wait", "--api=containers",     # NOQA
-                                                       workflow_url, inputtemp.name], env=env).strip()  # NOQA
+                                                       workflow_url, inputtemp.name], env=env,
+                                 stdout=subprocess.PIPE, stderr=subprocess.PIPE)  # NOQA
+                (stdoutdata, stderrdata) = proc.communicate()
+                if proc.returncode != 0:
+                    api.container_requests().update(uuid=cr_uuid, body={"priority": 0,
+                                                                        "properties": {"arvados-cwl-runner-log": stderrdata}}).execute()
+                else:
+                    api.container_requests().update(uuid=cr_uuid, body={"properties": {"arvados-cwl-runner-log": stderrdata}}).execute()
         except subprocess.CalledProcessError as e:
-            api = arvados.api_from_config(version="v1", apiconfig={
-                "ARVADOS_API_HOST": env["ARVADOS_API_HOST"],
-                "ARVADOS_API_TOKEN": env['ARVADOS_API_TOKEN'],
-                "ARVADOS_API_HOST_INSECURE": env["ARVADOS_API_HOST_INSECURE"]  # NOQA
-            })
-            request = api.container_requests().update(uuid=cr_uuid, body={"priority": 0}).execute()  # NOQA
+            api.container_requests().update(uuid=cr_uuid, body={"priority": 0,
+                                                                "properties": {"arvados-cwl-runner-log": str(e)}}).execute()
 
     @catch_exceptions
     def RunWorkflow(self, body):
@@ -132,6 +141,8 @@ def GetWorkflowLog(self, workflow_id):
         else:
             container = {"state": "Queued", "exit_code": None}
 
+        stderr = request["properties"].get("arvados-cwl-runner-log", "")
+
         outputobj = {}
         if request["output_uuid"]:
             c = arvados.collection.CollectionReader(request["output_uuid"], api_client=api)
@@ -144,12 +155,11 @@ def keepref(d):
 
                 visit(outputobj, keepref)
 
-        stderr = ""
         if request["log_uuid"]:
             c = arvados.collection.CollectionReader(request["log_uuid"], api_client=api)
             if "stderr.txt" in c:
                 with c.open("stderr.txt") as f:
-                    stderr = f.read()
+                    stderr += f.read()
 
         r = {
             "workflow_id": request["uuid"],