* Rename workflow_log to run_log to conform to latest WES draft.

* Log attachment staging and workflow_url to assist in debugging.

* Fix bug in wes-client to provide correct type for 'auth' parameter
  of WESClient object.

* Generalize --auth parameter of wes-client to support specifying
  alternate header.

Author: Peter Amstutz

wes_client/wes_client_main.py

@@ -15,7 +15,7 @@ def main(argv=sys.argv[1:]):
     parser = argparse.ArgumentParser(description="Workflow Execution Service")
     parser.add_argument("--host", type=str, default=os.environ.get("WES_API_HOST"),
                         help="Example: '--host=localhost:8080'.  Defaults to WES_API_HOST.")
-    parser.add_argument("--auth", type=str, default=os.environ.get("WES_API_AUTH"), help="Defaults to WES_API_AUTH.")
+    parser.add_argument("--auth", type=str, default=os.environ.get("WES_API_AUTH"), help="Format is 'Header: value' or just 'value'.  If header name is not provided, value goes in the 'Authorization'.  Defaults to WES_API_AUTH.")
     parser.add_argument("--proto", type=str, default=os.environ.get("WES_API_PROTO", "https"),
                         help="Options: [http, https].  Defaults to WES_API_PROTO (https).")
     parser.add_argument("--quiet", action="store_true", default=False)
@@ -49,7 +49,13 @@ def main(argv=sys.argv[1:]):
         print(u"%s %s" % (sys.argv[0], pkg[0].version))
         exit(0)
 
-    client = WESClient({'auth': args.auth, 'proto': args.proto, 'host': args.host})
+    if ": " in args.auth:
+        sp = args.auth.split(": ")
+        auth = {sp[0]: sp[1]}
+    else:
+        auth = {"Authorization": auth}
+
+    client = WESClient({'auth': auth, 'proto': args.proto, 'host': args.host})
 
     if args.list:
         response = client.list_runs()  # how to include: page_token=args.page, page_size=args.page_size ?
@@ -106,13 +112,13 @@ def main(argv=sys.argv[1:]):
 
     try:
         # TODO: Only works with Arvados atm
-        logging.info(str(s["workflow_log"]["stderr"]))
-        logs = requests.get(s["workflow_log"]["stderr"], headers={"Authorization": args.auth}).text
-        logging.info("Workflow log:\n" + logs)
+        logging.info(str(s["run_log"]["stderr"]))
+        logs = requests.get(s["run_log"]["stderr"], headers=auth).text
+        logging.info("Run log:\n" + logs)
     except InvalidSchema:
-        logging.info("Workflow log:\n" + str(s["workflow_log"]["stderr"]))
+        logging.info("Run log:\n" + str(s["run_log"]["stderr"]))
     except MissingSchema:
-        logging.info("Workflow log:\n" + str(s["workflow_log"]["stderr"]))
+        logging.info("Run log:\n" + str(s["run_log"]["stderr"]))
 
     # print the output json
     if "fields" in s["outputs"] and s["outputs"]["fields"] is None:

wes_service/arvados_wes.py

@@ -66,7 +66,7 @@ def GetServiceInfo(self):
             "workflow_type_versions": {
                 "CWL": {"workflow_type_version": ["v1.0"]}
             },
-            "supported_wes_versions": ["0.2.1"],
+            "supported_wes_versions": ["0.3.0"],
             "supported_filesystem_protocols": ["http", "https", "keep"],
             "workflow_engine_versions": {
                 "arvados-cwl-runner": stderr
@@ -108,6 +108,12 @@ def ListRuns(self, page_size=None, page_token=None, state_search=None):
             "next_page_token": workflow_list[-1]["run_id"] if workflow_list else ""
         }
 
+    def log_for_run(self, run_id, message):
+        api.logs().create(body={"log": {"object_uuid": run_id,
+                                        "event_type": "stderr",
+                                        "properties": {"text": message}}}).execute()
+
+
     def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
                           env, project_uuid,
                           tempdir):
@@ -141,9 +147,8 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
                 if proc.returncode != 0:
                     api.container_requests().update(uuid=cr_uuid, body={"priority": 0}).execute()
 
-                api.logs().create(body={"log": {"object_uuid": cr_uuid,
-                                                "event_type": "stderr",
-                                                "properties": {"text": stderrdata}}}).execute()
+                self.log_for_run(cr_uuid, stderrdata)
+
                 if tempdir:
                     shutil.rmtree(tempdir)
 
@@ -153,8 +158,6 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
 
     @catch_exceptions
     def RunWorkflow(self, **args):
-        tempdir, body = self.collect_attachments()
-
         if not connexion.request.headers.get('Authorization'):
             raise MissingAuthorization()
 
@@ -178,6 +181,8 @@ def RunWorkflow(self, **args):
                                                     "output_path": "n/a",
                                                     "priority": 500}}).execute()
 
+        tempdir, body = self.collect_attachments(cr["uuid"])
+
         workflow_url = body.get("workflow_url")
 
         project_uuid = body.get("workflow_engine_parameters", {}).get("project_uuid")
@@ -256,7 +261,7 @@ def log_object(cr):
                 "workflow_params": request["mounts"].get("/var/lib/cwl/cwl.input.json", {}).get("content", {})
             },
             "state": statemap[container["state"]],
-            "workflow_log": log_object(request),
+            "run_log": log_object(request),
             "task_logs": [log_object(t) for t in task_reqs],
             "outputs": outputobj
         }

wes_service/cwl_runner.py

@@ -142,7 +142,7 @@ def getlog(self):
             "run_id": self.run_id,
             "request": request,
             "state": state,
-            "workflow_log": {
+            "run_log": {
                 "cmd": [""],
                 "start_time": "",
                 "end_time": "",

wes_service/toil_wes.py

@@ -158,7 +158,7 @@ def getlog(self):
             "run_id": self.run_id,
             "request": request,
             "state": state,
-            "workflow_log": {
+            "run_log": {
                 "cmd": cmd,
                 "start_time": starttime,
                 "end_time": endtime,

wes_service/util.py

@@ -1,6 +1,7 @@
 import tempfile
 import json
 import os
+import logging
 
 from six import itervalues, iterlists
 import connexion
@@ -42,15 +43,20 @@ def getoptlist(self, p):
                 optlist.append(v)
         return optlist
 
-    def collect_attachments(self):
+    def log_for_run(self, run_id, message):
+        logging.info("Workflow %s: %s", run_id, message)
+
+    def collect_attachments(self, run_id=None):
         tempdir = tempfile.mkdtemp()
         body = {}
         for k, ls in iterlists(connexion.request.files):
             for v in ls:
                 if k == "workflow_attachment":
                     filename = secure_filename(v.filename)
-                    v.save(os.path.join(tempdir, filename))
-                    body[k] = "file://%s" % tempdir  # Reference to tem working dir.
+                    dest = os.path.join(tempdir, filename)
+                    self.log_for_run(run_id, "Staging attachment '%s' to '%s'" % (v.filename, dest))
+                    v.save(dest)
+                    body[k] = "file://%s" % tempdir  # Reference to temp working dir.
                 elif k in ("workflow_params", "tags", "workflow_engine_parameters"):
                     body[k] = json.loads(v.read())
                 else:
@@ -59,4 +65,6 @@ def collect_attachments(self):
         if ":" not in body["workflow_url"]:
             body["workflow_url"] = "file://%s" % os.path.join(tempdir, secure_filename(body["workflow_url"]))
 
+        self.log_for_run(run_id, "Using workflow_url '%s'" % body.get("workflow_url"))
+
         return tempdir, body