Merge pull request #19 from common-workflow-language/arv-support

Arv support

Author: tetron

Dockerfile

@@ -23,16 +23,14 @@ RUN mkdir -p /etc/apt/sources.list.d && \
     apt-get -yq --no-install-recommends install docker-engine=17.05.0~ce-0~debian-stretch && \
     apt-get clean
 
-ARG version
 ARG arvversion
-
-COPY dist/wes-service-${version}.tar.gz /root
 COPY dist/arvados-cwl-runner-${arvversion}.tar.gz /root
-
 RUN cd /root && tar xzf arvados-cwl-runner-${arvversion}.tar.gz && \
     cd arvados-cwl-runner-${arvversion} && \
     pip install .
 
+ARG version
+COPY dist/wes-service-${version}.tar.gz /root
 RUN cd /root && tar xzf wes-service-${version}.tar.gz && \
     cd wes-service-${version} && \
     pip install .[arvados]

wes_client/__init__.py

@@ -99,28 +99,40 @@ def fixpaths(d):
                 else:
                     d["location"] = d["path"]
                 del d["path"]
+            loc = d.get("location", "")
             if d.get("class") == "Directory":
-                loc = d.get("location", "")
                 if loc.startswith("http:") or loc.startswith("https:"):
                     logging.error("Directory inputs not supported with http references")
                     exit(33)
+            if not (loc.startswith("http:") or loc.startswith("https:")
+                    or args.job_order.startswith("http:") or args.job_order.startswith("https:")):
+                logging.error("Upload local files not supported, must use http: or https: references.")
+                exit(33)
 
     visit(input, fixpaths)
 
     workflow_url = args.workflow_url
     if not workflow_url.startswith("/") and ":" not in workflow_url:
-        workflow_url = os.path.abspath(workflow_url)
+        workflow_url = "file://" + os.path.abspath(workflow_url)
 
     if args.quiet:
         logging.basicConfig(level=logging.WARNING)
     else:
         logging.basicConfig(level=logging.INFO)
 
-    r = client.WorkflowExecutionService.RunWorkflow(body={
-        "workflow_url": workflow_url,
+    body = {
         "workflow_params": input,
         "workflow_type": "CWL",
-        "workflow_type_version": "v1.0"}).result()
+        "workflow_type_version": "v1.0"
+    }
+
+    if workflow_url.startswith("file://"):
+        with open(workflow_url[7:], "r") as f:
+            body["workflow_descriptor"] = f.read()
+    else:
+        body["workflow_url"] = workflow_url
+
+    r = client.WorkflowExecutionService.RunWorkflow(body=body).result()
 
     if args.wait:
         logging.info("Workflow id is %s", r["workflow_id"])
@@ -139,7 +151,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

@@ -1,15 +1,24 @@
 import arvados
 import arvados.util
 import arvados.collection
+import arvados.errors
 import os
 import connexion
 import json
 import subprocess
 import tempfile
+import functools
+import threading
+import logging
+
 from wes_service.util import visit, WESBackend
 
+class MissingAuthorization(Exception):
+    pass
 
 def get_api():
+    if not connexion.request.headers.get('Authorization'):
+        raise MissingAuthorization()
     return arvados.api_from_config(version="v1", apiconfig={
         "ARVADOS_API_HOST": os.environ["ARVADOS_API_HOST"],
         "ARVADOS_API_TOKEN": connexion.request.headers['Authorization'],
@@ -26,6 +35,23 @@ def get_api():
 }
 
 
+def catch_exceptions(orig_func):
+    """Catch uncaught exceptions and turn them into http errors"""
+
+    @functools.wraps(orig_func)
+    def catch_exceptions_wrapper(self, *args, **kwargs):
+        try:
+            return orig_func(self, *args, **kwargs)
+        except arvados.errors.ApiError as e:
+            logging.exception("Failure")
+            return {"msg": e._get_reason(), "status_code": e.resp.status}, int(e.resp.status)
+        except subprocess.CalledProcessError as e:
+            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
+
+    return catch_exceptions_wrapper
+
 class ArvadosBackend(WESBackend):
     def GetServiceInfo(self):
         return {
@@ -39,6 +65,7 @@ def GetServiceInfo(self):
             "key_values": {}
         }
 
+    @catch_exceptions
     def ListWorkflows(self):
         api = get_api()
 
@@ -60,28 +87,87 @@ def ListWorkflows(self):
             "next_page_token": ""
         }
 
+    def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params, env, workflow_descriptor_file):
+        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()
+                # TODO: run submission process in a container to prevent
+                # a-c-r submission processes from seeing each other.
+                proc = subprocess.Popen(["arvados-cwl-runner", "--submit-request-uuid="+cr_uuid, # NOQA
+                                         "--submit", "--no-wait", "--api=containers",     # 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.container_requests().update(uuid=cr_uuid, body={"priority": 0,
+                                                                "properties": {"arvados-cwl-runner-log": str(e)}}).execute()
+        finally:
+            if workflow_descriptor_file is not None:
+                workflow_descriptor_file.close()
+
+    @catch_exceptions
     def RunWorkflow(self, body):
         if body["workflow_type"] != "CWL" or body["workflow_type_version"] != "v1.0":  # NOQA
             return
 
+        if not connexion.request.headers.get('Authorization'):
+            raise MissingAuthorization()
+
         env = {
             "PATH": os.environ["PATH"],
             "ARVADOS_API_HOST": os.environ["ARVADOS_API_HOST"],
             "ARVADOS_API_TOKEN": connexion.request.headers['Authorization'],
             "ARVADOS_API_HOST_INSECURE": os.environ.get("ARVADOS_API_HOST_INSECURE", "false")  # NOQA
         }
-        with tempfile.NamedTemporaryFile() as inputtemp:
-            json.dump(body["workflow_params"], inputtemp)
-            inputtemp.flush()
-            workflow_id = subprocess.check_output(["arvados-cwl-runner", "--submit", "--no-wait", "--api=containers",  # NOQA
-                                                   body.get("workflow_url"), inputtemp.name], env=env).strip()  # NOQA
-        return {"workflow_id": workflow_id}
 
+        api = get_api()
+
+        cr = api.container_requests().create(body={"container_request":
+                                                   {"command": [""],
+                                                    "container_image": "n/a",
+                                                    "state": "Uncommitted",
+                                                    "output_path": "n/a",
+                                                    "priority": 500}}).execute()
+
+        workflow_url = body.get("workflow_url")
+        workflow_descriptor_file = None
+        if body.get("workflow_descriptor"):
+            workflow_descriptor_file = tempfile.NamedTemporaryFile()
+            workflow_descriptor_file.write(body.get('workflow_descriptor'))
+            workflow_descriptor_file.flush()
+            workflow_url = workflow_descriptor_file.name
+
+        threading.Thread(target=self.invoke_cwl_runner, args=(cr["uuid"],
+                                                              workflow_url,
+                                                              body["workflow_params"],
+                                                              env,
+                                                              workflow_descriptor_file)).start()
+
+        return {"workflow_id": cr["uuid"]}
+
+    @catch_exceptions
     def GetWorkflowLog(self, workflow_id):
         api = get_api()
 
         request = api.container_requests().get(uuid=workflow_id).execute()
-        container = api.containers().get(uuid=request["container_uuid"]).execute()  # NOQA
+        if request["container_uuid"]:
+            container = api.containers().get(uuid=request["container_uuid"]).execute()  # NOQA
+        else:
+            container = {"state": "Queued", "exit_code": None}
+
+        stderr = request["properties"].get("arvados-cwl-runner-log", "")
 
         outputobj = {}
         if request["output_uuid"]:
@@ -91,16 +177,15 @@ def GetWorkflowLog(self, workflow_id):
 
                 def keepref(d):
                     if isinstance(d, dict) and "location" in d:
-                        d["location"] = "keep:%s/%s" % (c.portable_data_hash(), d["location"])  # NOQA
+                        d["location"] = "%sc=%s/_/%s" % (api._resourceDesc["keepWebServiceUrl"], c.portable_data_hash(), d["location"])  # NOQA
 
                 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"],
@@ -120,15 +205,22 @@ def keepref(d):
             r["workflow_log"]["exit_code"] = container["exit_code"]
         return r
 
+    @catch_exceptions
     def CancelJob(self, workflow_id):  # NOQA
         api = get_api()
-        request = api.container_requests().update(body={"priority": 0}).execute()  # NOQA
+        request = api.container_requests().update(uuid=workflow_id, body={"priority": 0}).execute()  # NOQA
         return {"workflow_id": request["uuid"]}
 
+    @catch_exceptions
     def GetWorkflowStatus(self, workflow_id):
         api = get_api()
         request = api.container_requests().get(uuid=workflow_id).execute()
-        container = api.containers().get(uuid=request["container_uuid"]).execute()  # NOQA
+        if request["container_uuid"]:
+            container = api.containers().get(uuid=request["container_uuid"]).execute()  # NOQA
+        elif request["priority"] == 0:
+            container = {"state": "Cancelled"}
+        else:
+            container = {"state": "Queued"}
         return {"workflow_id": request["uuid"],
                 "state": statemap[container["state"]]}