Merge pull request #17 from common-workflow-language/arv

support http references

Author: tetron

Dockerfile

@@ -0,0 +1,49 @@
+FROM debian:9
+
+# Install passenger
+
+RUN apt-get update && \
+    apt-get install -y dirmngr gnupg && \
+    apt-key adv --keyserver hkp://keyserver.ubuntu.com:80 --recv-keys 561F9B9CAC40B2F7 && \
+    apt-get install -y apt-transport-https ca-certificates && \
+    sh -c 'echo deb https://oss-binaries.phusionpassenger.com/apt/passenger stretch main > /etc/apt/sources.list.d/passenger.list'
+
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends passenger python-setuptools build-essential python-dev python-pip git && \
+    pip install pip==9.0.3
+
+RUN apt-get install -y --no-install-recommends libcurl4-openssl-dev libssl1.0-dev
+
+RUN apt-key adv --keyserver hkp://pool.sks-keyservers.net:80 --recv-keys 58118E89F3A912897C070ADBF76221572C52609D || \
+    apt-key adv --keyserver hkp://pgp.mit.edu:80 --recv-keys 58118E89F3A912897C070ADBF76221572C52609D
+
+RUN mkdir -p /etc/apt/sources.list.d && \
+    echo deb https://apt.dockerproject.org/repo debian-stretch main > /etc/apt/sources.list.d/docker.list && \
+    apt-get update && \
+    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 .
+
+RUN cd /root && tar xzf wes-service-${version}.tar.gz && \
+    cd wes-service-${version} && \
+    pip install .[arvados]
+
+COPY passenger_wsgi.py /var/www/wes-server/passenger_wsgi.py
+
+EXPOSE 443
+
+WORKDIR /var/www/wes-server/
+RUN chown www-data:www-data -R /var/www && adduser www-data docker
+
+CMD ["passenger", "start", "--environment=production", "--user=www-data", "--port=443", "--ssl", \
+    "--ssl-certificate=/etc/ssl/certs/ssl-cert-wes.pem", \
+    "--ssl-certificate-key=/etc/ssl/private/ssl-cert-wes.key"]

MANIFEST.in

@@ -0,0 +1,2 @@
+include README.pypi.rst
+include wes_service/openapi/workflow_execution_service.swagger.yaml

passenger_wsgi.py

@@ -0,0 +1,3 @@
+import wes_service
+
+application = wes_service.setup()

setup.py

@@ -15,7 +15,7 @@
     long_description = readmeFile.read()
 
 setup(name='wes-service',
-      version='2.2',
+      version='2.3',
       description='GA4GH Workflow Execution Service reference implementation',
       long_description=long_description,
       author='GA4GH Containers and Workflows task team',
@@ -30,7 +30,8 @@
           'connexion',
           'bravado',
           'ruamel.yaml >= 0.12.4, < 0.15',
-          'cwl-runner'
+          'cwlref-runner',
+          'schema-salad'
         ],
       entry_points={
           'console_scripts': [ "wes-server=wes_service:main",
@@ -41,5 +42,5 @@
               "arvados-cwl-runner"
           ]
       },
-      zip_safe=True
+      zip_safe=False
 )

testdata/md5sum.cwl.json

@@ -1,10 +1,6 @@
 {
   "input_file": {
         "class": "File",
-        "path": "../../testdata/md5sum.input"
-    },
-    "output_file": {
-        "class": "File",
-        "path": "/tmp/md5sum.txt"
+        "path": "md5sum.input"
     }
-}
+}

wes-docker.sh

@@ -0,0 +1,10 @@
+#!/bin/sh
+set -e
+python setup.py sdist
+docker build --build-arg version=2.3 -t commonworkflowlanguage/workflow-service .
+docker run -ti \
+       -v$PWD/config.yml:/var/www/wes-server/config.yml \
+       -v/etc/ssl/certs/ssl-cert-snakeoil.pem:/etc/ssl/certs/ssl-cert-wes.pem \
+       -v/etc/ssl/private/ssl-cert-snakeoil.key:/etc/ssl/private/ssl-cert-wes.key \
+       -v/var/run/docker.sock:/var/run/docker.sock \
+       commonworkflowlanguage/workflow-service

wes_client/__init__.py

@@ -13,7 +13,7 @@
 from wes_service.util import visit
 import urllib
 import ruamel.yaml as yaml
-
+import schema_salad.ref_resolver
 
 def main(argv=sys.argv[1:]):
     parser = argparse.ArgumentParser(description='Workflow Execution Service')
@@ -31,6 +31,7 @@ def main(argv=sys.argv[1:]):
     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("--info", action="store_true", default=False)
     exgroup.add_argument("--version", action="store_true", default=False)
 
     exgroup = parser.add_mutually_exclusive_group()
@@ -75,17 +76,36 @@ def main(argv=sys.argv[1:]):
         json.dump(response.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)
+    if args.info:
+        response = client.WorkflowExecutionService.GetServiceInfo()
+        json.dump(response.result(), sys.stdout, indent=4)
+        return 0
+
+    loader = schema_salad.ref_resolver.Loader({
+        "location": {"@type": "@id"},
+        "path": {"@type": "@id"}
+    })
+    input, _ = loader.resolve_ref(args.job_order)
+
+    basedir = os.path.dirname(args.job_order)
 
-        def fixpaths(d):
-            if isinstance(d, dict) and "location" in d:
-                if ":" not in d["location"]:
+    def fixpaths(d):
+        if isinstance(d, dict):
+            if "path" in d:
+                if ":" not in d["path"]:
                     local_path = os.path.normpath(
-                        os.path.join(os.getcwd(), basedir, d["location"]))
+                        os.path.join(os.getcwd(), basedir, d["path"]))
                     d["location"] = urllib.pathname2url(local_path)
-        visit(input, fixpaths)
+                else:
+                    d["location"] = d["path"]
+                del d["path"]
+            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)
+
+    visit(input, fixpaths)
 
     workflow_url = args.workflow_url
     if not workflow_url.startswith("/") and ":" not in workflow_url:
@@ -125,7 +145,7 @@ def fixpaths(d):
         del s["outputs"]["fields"]
     json.dump(s["outputs"], sys.stdout, indent=4)
 
-    if r["state"] == "Complete":
+    if r["state"] == "COMPLETE":
         return 0
     else:
         return 1

wes_service/__init__.py

@@ -1,19 +1,31 @@
 import argparse
 import sys
+import ruamel.yaml
+import os
+import logging
+
+logging.basicConfig(level=logging.INFO)
 
 import connexion
 import connexion.utils as utils
 from connexion.resolver import Resolver
 
+def setup(args=None):
+    if args is None:
+        args = argparse.Namespace()
 
-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")
-    parser.add_argument("--debug", action="store_true", default=False)
-    args = parser.parse_args(argv)
+    configfile = "config.yml"
+    if os.path.isfile(configfile):
+        logging.info("Loading %s", configfile)
+        with open(configfile, "r") as f:
+            config = ruamel.yaml.safe_load(f)
+        for c in config:
+            setattr(args, c, config[c])
+
+
+    logging.info("Using config:")
+    for n in args.__dict__:
+        logging.info("  %s: %s", n, getattr(args, n))
 
     app = connexion.App(__name__)
     backend = utils.get_function_from_name(
@@ -26,6 +38,20 @@ def rs(x):
         'openapi/workflow_execution_service.swagger.yaml',
         resolver=Resolver(rs))
 
+    return app
+
+
+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")
+    parser.add_argument("--debug", action="store_true", default=False)
+    args = parser.parse_args(argv)
+
+    app = setup(args)
+
     app.run(port=args.port, debug=args.debug)
 
 

wes_service/__init__.pyc

@@ -1,4 +1,5 @@
 import arvados
+import arvados.util
 import arvados.collection
 import os
 import connexion
@@ -17,22 +18,22 @@ def get_api():
 
 
 statemap = {
-    "Queued": "Queued",
-    "Locked": "Initializing",
-    "Running": "Running",
-    "Complete": "Complete",
-    "Cancelled": "Canceled"
+    "Queued": "QUEUED",
+    "Locked": "INITIALIZING",
+    "Running": "RUNNING",
+    "Complete": "COMPLETE",
+    "Cancelled": "CANCELED"
 }
 
 
 class ArvadosBackend(WESBackend):
     def GetServiceInfo(self):
         return {
             "workflow_type_versions": {
-                "CWL": ["v1.0"]
+                "CWL": {"workflow_type_version": ["v1.0"]}
             },
-            "supported_wes_versions": "0.1.0",
-            "supported_filesystem_protocols": ["file"],
+            "supported_wes_versions": "0.2.1",
+            "supported_filesystem_protocols": ["file", "http", "https", "keep"],
             "engine_versions": "cwl-runner",
             "system_state_counts": {},
             "key_values": {}
@@ -41,19 +42,21 @@ def GetServiceInfo(self):
     def ListWorkflows(self):
         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"]]]],  # NOQA
-                                           select=["uuid", "state"]).execute()
+        requests = arvados.util.list_all(api.container_requests().list,
+                                         filters=[["requesting_container_uuid", "=", None],
+                                                  ["container_uuid", "!=", None]],
+                                         select=["uuid", "command", "container_uuid"])
+        containers = arvados.util.list_all(api.containers().list,
+                                           filters=[["uuid", "in", [w["container_uuid"] for w in requests]]],
+                                           select=["uuid", "state"])
 
-        uuidmap = {c["uuid"]: statemap[c["state"]] for c in containers["items"]}  # NOQA
+        uuidmap = {c["uuid"]: statemap[c["state"]] for c in containers}
 
         return {
             "workflows": [{"workflow_id": cr["uuid"],
-                           "state": uuidmap[cr["container_uuid"]]}
-                          for cr in requests["items"]
-                          if cr["command"][0] == "arvados-cwl-runner"],
+                           "state": uuidmap.get(cr["container_uuid"])}
+                          for cr in requests
+                          if cr["command"] and cr["command"][0] == "arvados-cwl-runner"],
             "next_page_token": ""
         }
 
@@ -82,7 +85,7 @@ def GetWorkflowLog(self, workflow_id):
 
         outputobj = {}
         if request["output_uuid"]:
-            c = arvados.collection.CollectionReader(request["output_uuid"])
+            c = arvados.collection.CollectionReader(request["output_uuid"], api_client=api)
             with c.open("cwl.output.json") as f:
                 outputobj = json.load(f)
 
@@ -94,7 +97,7 @@ def keepref(d):
 
         stderr = ""
         if request["log_uuid"]:
-            c = arvados.collection.CollectionReader(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()
@@ -114,7 +117,7 @@ def keepref(d):
             "outputs": outputobj
         }
         if container["exit_code"] is not None:
-            r["workflow_log"]["exitCode"] = container["exit_code"]
+            r["workflow_log"]["exit_code"] = container["exit_code"]
         return r
 
     def CancelJob(self, workflow_id):  # NOQA