Merge pull request #61 from common-workflow-language/run_log

Rename workflow_log to run_log to conform to latest WES draft.

Author: tetron

setup.py

@@ -11,7 +11,7 @@
     long_description = readmeFile.read()
 
 setup(name='wes-service',
-      version='2.7',
+      version='2.8',
       description='GA4GH Workflow Execution Service reference implementation',
       long_description=long_description,
       author='GA4GH Containers and Workflows task team',

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,15 @@ 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})
+    auth = {}
+    if args.auth:
+        if ": " in args.auth:
+            sp = args.auth.split(": ")
+            auth[sp[0]] = sp[1]
+        else:
+            auth["Authorization"] = args.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 ?
@@ -79,15 +87,15 @@ def main(argv=sys.argv[1:]):
         logging.error("Missing json/yaml file.")
         return 1
 
-    modify_jsonyaml_paths(args.job_order)
+    job_order = modify_jsonyaml_paths(args.job_order)
 
     if args.quiet:
         logging.basicConfig(level=logging.WARNING)
     else:
         logging.basicConfig(level=logging.INFO)
 
     args.attachments = "" if not args.attachments else args.attachments.split(',')
-    r = client.run(args.workflow_url, args.job_order, args.attachments)
+    r = client.run(args.workflow_url, job_order, args.attachments)
 
     if args.wait:
         logging.info("Workflow run id is %s", r["run_id"])
@@ -106,13 +114,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

@@ -19,11 +19,13 @@ class MissingAuthorization(Exception):
     pass
 
 
-def get_api():
-    if not connexion.request.headers.get('Authorization'):
-        raise MissingAuthorization()
-    authtoken = connexion.request.headers['Authorization']
-    if authtoken.startswith("Bearer ") or authtoken.startswith("OAuth2 "):
+def get_api(authtoken=None):
+    if authtoken is None:
+        if not connexion.request.headers.get('Authorization'):
+            raise MissingAuthorization()
+        authtoken = connexion.request.headers['Authorization']
+        if not authtoken.startswith("Bearer ") or authtoken.startswith("OAuth2 "):
+            raise ValueError("Authorization token must start with 'Bearer '")
         authtoken = authtoken[7:]
     return arvados.api_from_config(version="v1", apiconfig={
         "ARVADOS_API_HOST": os.environ["ARVADOS_API_HOST"],
@@ -55,6 +57,10 @@ def catch_exceptions_wrapper(self, *args, **kwargs):
             return {"msg": str(e), "status_code": 500}, 500
         except MissingAuthorization:
             return {"msg": "'Authorization' header is missing or empty, expecting Arvados API token", "status_code": 401}, 401
+        except ValueError as e:
+            return {"msg": str(e), "status_code": 400}, 400
+        except Exception as e:
+            return {"msg": str(e), "status_code": 500}, 500
 
     return catch_exceptions_wrapper
 
@@ -66,7 +72,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 +114,11 @@ 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, authtoken=None):
+        get_api(authtoken).logs().create(body={"log": {"object_uuid": run_id,
+                                                       "event_type": "stderr",
+                                                       "properties": {"text": message+"\n"}}}).execute()
+
     def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
                           env, project_uuid,
                           tempdir):
@@ -118,9 +129,18 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
         })
 
         try:
-            with tempfile.NamedTemporaryFile() as inputtemp:
+            with tempfile.NamedTemporaryFile(dir=tempdir, suffix=".json") as inputtemp:
                 json.dump(workflow_params, inputtemp)
                 inputtemp.flush()
+
+                msg = ""
+                for dirpath, dirs, files in os.walk(tempdir):
+                    for f in files:
+                        msg += "  " + dirpath + "/" + f + "\n"
+
+                self.log_for_run(cr_uuid, "Contents of %s:\n%s" % (tempdir, msg),
+                                 env['ARVADOS_API_TOKEN'])
+
                 # TODO: run submission process in a container to prevent
                 # a-c-r submission processes from seeing each other.
 
@@ -133,6 +153,8 @@ def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
                 cmd.append(workflow_url)
                 cmd.append(inputtemp.name)
 
+                self.log_for_run(cr_uuid, "Executing %s" % cmd, env['ARVADOS_API_TOKEN'])
+
                 proc = subprocess.Popen(cmd, env=env,
                                         cwd=tempdir,
                                         stdout=subprocess.PIPE,
@@ -141,9 +163,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, env['ARVADOS_API_TOKEN'])
+
                 if tempdir:
                     shutil.rmtree(tempdir)
 
@@ -153,8 +174,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,17 +197,25 @@ def RunWorkflow(self, **args):
                                                     "output_path": "n/a",
                                                     "priority": 500}}).execute()
 
-        workflow_url = body.get("workflow_url")
+        try:
+            tempdir, body = self.collect_attachments(cr["uuid"])
+
+            workflow_url = body.get("workflow_url")
 
-        project_uuid = body.get("workflow_engine_parameters", {}).get("project_uuid")
+            project_uuid = body.get("workflow_engine_parameters", {}).get("project_uuid")
 
-        threading.Thread(target=self.invoke_cwl_runner, args=(cr["uuid"],
-                                                              workflow_url,
-                                                              body["workflow_params"],
-                                                              env,
-                                                              project_uuid,
-                                                              tempdir)).start()
+            threading.Thread(target=self.invoke_cwl_runner, args=(cr["uuid"],
+                                                                  workflow_url,
+                                                                  body["workflow_params"],
+                                                                  env,
+                                                                  project_uuid,
+                                                                  tempdir)).start()
 
+        except Exception as e:
+            self.log_for_run(cr["uuid"], str(e))
+            cr = api.container_requests().update(uuid=cr["uuid"],
+                                                 body={"container_request":
+                                                       {"priority": 0}}).execute()
         return {"run_id": cr["uuid"]}
 
     @catch_exceptions
@@ -203,7 +230,11 @@ def GetRunLog(self, run_id):
             containers_map = {c["uuid"]: c for c in tasks}
             containers_map[container["uuid"]] = container
         else:
-            container = {"state": "Queued", "exit_code": None, "log": None}
+            container = {
+                "state": "Queued" if request["priority"] > 0 else "Cancelled",
+                "exit_code": None,
+                "log": None
+            }
             tasks = []
             containers_map = {}
             task_reqs = []
@@ -256,7 +287,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,21 +43,29 @@ 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())
+                    content = v.read()
+                    body[k] = json.loads(content)
                 else:
                     body[k] = v.read()
 
         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