Merge pull request #1 from common-workflow-language/ga4gh-wes

GA4GH workflow execution service

Author: tetron

setup.py

@@ -8,27 +8,33 @@
 from setuptools import setup, find_packages
 
 SETUP_DIR = os.path.dirname(__file__)
-README = os.path.join(SETUP_DIR, 'README')
+README = os.path.join(SETUP_DIR, 'README.md')
 
-setup(name='cwltool_service',
+setup(name='wes_service',
       version='2.0',
-      description='Common workflow language runner service',
+      description='GA4GH Workflow Execution Service reference implementation',
       long_description=open(README).read(),
-      author='Common workflow language working group',
+      author='GA4GH Containers and Workflows task team',
       author_email='[email protected]',
       url="https://github.com/common-workflow-language/cwltool-service",
       download_url="https://github.com/common-workflow-language/cwltool-service",
       license='Apache 2.0',
-      py_modules=["cwltool_stream", "cwl_flask", "cwltool_client"],
+      packages=["wes_service", "wes_client"],
+      package_data={'wes_service': ['swagger/proto/workflow_execution.swagger.json']},
+      include_package_data=True,
       install_requires=[
-          'Flask',
-          'requests',
-          'PyYAML'
+          'connexion',
+          'bravado',
+          'ruamel.yaml >= 0.12.4, < 0.15',
         ],
       entry_points={
-          'console_scripts': [ "cwltool-stream=cwltool_stream:main",
-                               "cwl-server=cwl_flask:main",
-                               "cwl-client=cwl_client:main"]
+          'console_scripts': [ "wes-server=wes_service:main",
+                               "wes-client=wes_client:main"]
+      },
+      extras_require={
+          "arvados": [
+              "arvados-cwl-runner"
+          ]
       },
       zip_safe=True
 )

wes_client/__init__.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python
+
+from bravado.client import SwaggerClient
+from bravado.requests_client import RequestsClient
+import json
+import time
+import pprint
+import sys
+import os
+import argparse
+import logging
+import urlparse
+import pkg_resources  # part of setuptools
+from wes_service.util import visit
+import urllib
+import ruamel.yaml as yaml
+
+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"))
+    parser.add_argument("--auth", type=str, default=os.environ.get("WES_API_AUTH"))
+    parser.add_argument("--proto", type=str, default=os.environ.get("WES_API_PROTO", "https"))
+    parser.add_argument("--quiet", action="store_true", default=False)
+    parser.add_argument("--outdir", type=str)
+
+    exgroup = parser.add_mutually_exclusive_group()
+    exgroup.add_argument("--run", action="store_true", default=False)
+    exgroup.add_argument("--get", type=str, default=None)
+    exgroup.add_argument("--log", type=str, default=None)
+    exgroup.add_argument("--list", action="store_true", default=False)
+    exgroup.add_argument("--version", action="store_true", default=False)
+
+    parser.add_argument("workflow_url", type=str, nargs="?", default=None)
+    parser.add_argument("job_order", type=str, nargs="?", default=None)
+    args = parser.parse_args(argv)
+
+    if args.version:
+        pkg = pkg_resources.require("wes_service")
+        print u"%s %s" % (sys.argv[0], pkg[0].version)
+        exit(0)
+
+    http_client = RequestsClient()
+    split = urlparse.urlsplit("%s://%s/" % (args.proto, args.host))
+
+    http_client.set_api_key(
+        split.hostname, args.auth,
+        param_name='Authorization', param_in='header')
+    client = SwaggerClient.from_url("%s://%s/swagger.json" % (args.proto, args.host),
+                                    http_client=http_client, config={'use_models': False})
+
+    if args.list:
+        l = client.WorkflowExecutionService.ListWorkflows()
+        json.dump(l.result(), sys.stdout, indent=4)
+        return 0
+
+    if args.log:
+        l = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=args.log)
+        sys.stdout.write(l.result()["workflow_log"]["stderr"])
+        return 0
+
+    if args.get:
+        l = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=args.get)
+        json.dump(l.result(), sys.stdout, indent=4)
+        return 0
+
+    with open(args.job_order) as f:
+        input = yaml.safe_load(f)
+        basedir = os.path.dirname(args.job_order)
+        def fixpaths(d):
+            if isinstance(d, dict) and "location" in d:
+                if not ":" in d["location"]:
+                    d["location"] = urllib.pathname2url(os.path.normpath(os.path.join(os.getcwd(), basedir, d["location"])))
+        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)
+
+    if args.quiet:
+        logging.basicConfig(level=logging.WARNING)
+    else:
+        logging.basicConfig(level=logging.INFO)
+
+    r = client.WorkflowExecutionService.RunWorkflow(body={
+        "workflow_url": workflow_url,
+        "workflow_params": input,
+        "workflow_type": "CWL",
+        "workflow_type_version": "v1.0"}).result()
+
+    logging.info("Workflow id is %s", r["workflow_id"])
+
+    r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r["workflow_id"]).result()
+    while r["state"] in ("Queued", "Initializing", "Running"):
+        time.sleep(1)
+        r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r["workflow_id"]).result()
+
+    logging.info("State is %s", r["state"])
+
+    s = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=r["workflow_id"]).result()
+    logging.info(s["workflow_log"]["stderr"])
+
+    json.dump(s["outputs"], sys.stdout, indent=4)
+
+    if r["state"] == "Complete":
+        return 0
+    else:
+        return 1
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

wes_service/__init__.py

@@ -0,0 +1,35 @@
+import connexion
+from connexion.resolver import Resolver
+import connexion.utils as utils
+
+import threading
+import tempfile
+import subprocess
+import uuid
+import os
+import json
+import urllib
+import argparse
+import sys
+
+from pkg_resources import resource_stream
+
+def main(argv=sys.argv[1:]):
+    parser = argparse.ArgumentParser(description='Workflow Execution Service')
+    parser.add_argument("--backend", type=str, default="wes_service.cwl_runner")
+    parser.add_argument("--port", type=int, default=8080)
+    parser.add_argument("--opt", type=str, action="append")
+    args = parser.parse_args(argv)
+
+    app = connexion.App(__name__)
+    backend = utils.get_function_from_name(args.backend + ".create_backend")(args.opt)
+    def rs(x):
+        return getattr(backend, x)
+
+    res = resource_stream(__name__, 'swagger/proto/workflow_execution.swagger.json')
+    app.add_api(json.load(res), resolver=Resolver(rs))
+
+    app.run(port=args.port)
+
+if __name__ == "__main__":
+    main(sys.argv[1:])

wes_service/arvados_wes.py

@@ -0,0 +1,134 @@
+import arvados
+import arvados.collection
+import os
+import connexion
+import json
+import subprocess
+import tempfile
+from wes_service.util import visit, WESBackend
+
+def get_api():
+    return arvados.api_from_config(version="v1", apiconfig={
+        "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"),
+    })
+
+
+statemap = {
+    "Queued": "Queued",
+    "Locked": "Initializing",
+    "Running": "Running",
+    "Complete": "Complete",
+    "Cancelled": "Canceled"
+}
+
+class ArvadosBackend(WESBackend):
+    def GetServiceInfo(self):
+        return {
+            "workflow_type_versions": {
+                "CWL": ["v1.0"]
+            },
+            "supported_wes_versions": "0.1.0",
+            "supported_filesystem_protocols": ["file"],
+            "engine_versions": "cwl-runner",
+            "system_state_counts": {},
+            "key_values": {}
+        }
+
+    def ListWorkflows(self, body=None):
+        # body["page_size"]
+        # body["page_token"]
+        # body["key_value_search"]
+
+        api = get_api()
+
+        requests = api.container_requests().list(filters=[["requesting_container_uuid", "=", None]],
+                                                 select=["uuid", "command", "container_uuid"]).execute()
+        containers = api.containers().list(filters=[["uuid", "in", [w["container_uuid"] for w in requests["items"]]]],
+                                           select=["uuid", "state"]).execute()
+
+        uuidmap = {c["uuid"]: statemap[c["state"]] for c in containers["items"]}
+
+        return {
+            "workflows": [{"workflow_id": cr["uuid"],
+                           "state": uuidmap[cr["container_uuid"]]}
+                          for cr in requests["items"]
+                          if cr["command"][0] == "arvados-cwl-runner"],
+            "next_page_token": ""
+        }
+
+    def RunWorkflow(self, body):
+        if body["workflow_type"] != "CWL" or body["workflow_type_version"] != "v1.0":
+            return
+
+        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")
+        }
+        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",
+                                                   body.get("workflow_url"), inputtemp.name], env=env).strip()
+        return {"workflow_id": workflow_id}
+
+
+    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()
+
+        outputobj = {}
+        if request["output_uuid"]:
+            c = arvados.collection.CollectionReader(request["output_uuid"])
+            with c.open("cwl.output.json") as f:
+                outputobj = json.load(f)
+                def keepref(d):
+                    if isinstance(d, dict) and "location" in d:
+                        d["location"] = "keep:%s/%s" % (c.portable_data_hash(), d["location"])
+                visit(outputobj, keepref)
+
+        stderr = ""
+        if request["log_uuid"]:
+            c = arvados.collection.CollectionReader(request["log_uuid"])
+            if "stderr.txt" in c:
+                with c.open("stderr.txt") as f:
+                    stderr = f.read()
+
+        r = {
+            "workflow_id": request["uuid"],
+            "request": {},
+            "state": statemap[container["state"]],
+            "workflow_log": {
+                "cmd": [""],
+                "startTime": "",
+                "endTime": "",
+                "stdout": "",
+                "stderr": stderr
+            },
+            "task_logs": [],
+            "outputs": outputobj
+        }
+        if container["exit_code"] is not None:
+            r["workflow_log"]["exitCode"] = container["exit_code"]
+        return r
+
+
+    def CancelJob(self, workflow_id):
+        api = get_api()
+        request = api.container_requests().update(body={"priority": 0}).execute()
+        return {"workflow_id": request["uuid"]}
+
+    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()
+        return {"workflow_id": request["uuid"],
+                "state": statemap[container["state"]]}
+
+def create_backend(opts):
+    return ArvadosBackend(optdict)