Rework plugin API to support simple option passing.

Author: Peter Amstutz

wes_service/__init__.py

@@ -18,11 +18,13 @@ 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 utils.get_function_from_name(args.backend + "." + x)
+        return getattr(backend, x)
 
     res = resource_stream(__name__, 'swagger/proto/workflow_execution.swagger.json')
     app.add_api(json.load(res), resolver=Resolver(rs))

wes_service/arvados_wes.py

@@ -5,6 +5,7 @@
 import json
 import subprocess
 import tempfile
+from wes_service.util import visit, WESBackend
 
 def get_api():
     return arvados.api_from_config(version="v1", apiconfig={
@@ -22,109 +23,112 @@ def get_api():
     "Cancelled": "Canceled"
 }
 
-
-def GetServiceInfo():
-    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(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(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(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(workflow_id):
-    api = get_api()
-    request = api.container_requests().update(body={"priority": 0}).execute()
-    return {"workflow_id": request["uuid"]}
-
-def GetWorkflowStatus(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"]]}
+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)

wes_service/cwl_runner.py

@@ -6,15 +6,15 @@
 import json
 import urllib
 import sys
-from wes_service.util import visit
+from wes_service.util import visit, WESBackend
 
 class Workflow(object):
     def __init__(self, workflow_id):
         super(Workflow, self).__init__()
         self.workflow_id = workflow_id
         self.workdir = os.path.join(os.getcwd(), "workflows", self.workflow_id)
 
-    def run(self, request):
+    def run(self, request, opts):
         os.makedirs(self.workdir)
         outdir = os.path.join(self.workdir, "outdir")
         os.mkdir(outdir)
@@ -35,7 +35,10 @@ def run(self, request):
         output = open(os.path.join(self.workdir, "cwl.output.json"), "w")
         stderr = open(os.path.join(self.workdir, "stderr"), "w")
 
-        proc = subprocess.Popen(["cwl-runner", workflow_url, inputtemp.name],
+        runner = opts.getopt("runner", "cwl-runner")
+        extra = opts.getoptlist("extra")
+
+        proc = subprocess.Popen([runner]+extra+[workflow_url, inputtemp.name],
                                 stdout=output,
                                 stderr=stderr,
                                 close_fds=True,
@@ -117,49 +120,54 @@ def getlog(self):
     def cancel(self):
         pass
 
-def GetServiceInfo():
-    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(body=None):
-    # body["page_size"]
-    # body["page_token"]
-    # body["key_value_search"]
-
-    wf = []
-    for l in os.listdir(os.path.join(os.getcwd(), "workflows")):
-        if os.path.isdir(os.path.join(os.getcwd(), "workflows", l)):
-            wf.append(Workflow(l))
-    return {
-        "workflows": [{"workflow_id": w.workflow_id, "state": w.getstate()[0]} for w in wf],
-        "next_page_token": ""
-    }
-
-def RunWorkflow(body):
-    if body["workflow_type"] != "CWL" or body["workflow_type_version"] != "v1.0":
-        return
-    workflow_id = uuid.uuid4().hex
-    job = Workflow(workflow_id)
-    job.run(body)
-    return {"workflow_id": workflow_id}
-
-def GetWorkflowLog(workflow_id):
-    job = Workflow(workflow_id)
-    return job.getlog()
-
-def CancelJob(workflow_id):
-    job = Workflow(workflow_id)
-    job.cancel()
-    return {"workflow_id": workflow_id}
-
-def GetWorkflowStatus(workflow_id):
-    job = Workflow(workflow_id)
-    return job.getstatus()
+
+class CWLRunnerBackend(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"]
+
+        wf = []
+        for l in os.listdir(os.path.join(os.getcwd(), "workflows")):
+            if os.path.isdir(os.path.join(os.getcwd(), "workflows", l)):
+                wf.append(Workflow(l))
+        return {
+            "workflows": [{"workflow_id": w.workflow_id, "state": w.getstate()[0]} for w in wf],
+            "next_page_token": ""
+        }
+
+    def RunWorkflow(self, body):
+        if body["workflow_type"] != "CWL" or body["workflow_type_version"] != "v1.0":
+            return
+        workflow_id = uuid.uuid4().hex
+        job = Workflow(workflow_id)
+        job.run(body, self)
+        return {"workflow_id": workflow_id}
+
+    def GetWorkflowLog(self, workflow_id):
+        job = Workflow(workflow_id)
+        return job.getlog()
+
+    def CancelJob(self, workflow_id):
+        job = Workflow(workflow_id)
+        job.cancel()
+        return {"workflow_id": workflow_id}
+
+    def GetWorkflowStatus(self, workflow_id):
+        job = Workflow(workflow_id)
+        return job.getstatus()
+
+def create_backend(opts):
+    return CWLRunnerBackend(opts)

wes_service/util.py

@@ -6,3 +6,23 @@ def visit(d, op):
     elif isinstance(d, dict):
         for i in d.itervalues():
             visit(i, op)
+
+class WESBackend(object):
+    def __init__(self, opts):
+        self.pairs = []
+        for o in opts:
+            k, v = o.split("=", 1)
+            self.pairs.append((k, v))
+
+    def getopt(self, p, default=None):
+        for k,v in self.pairs:
+            if k == p:
+                return v
+        return default
+
+    def getoptlist(self, p):
+        l = []
+        for k, v in self.pairs:
+            if k == p:
+                l.append(v)
+        return l