Update to latest WES (wip)

Author: Peter Amstutz

wes_client/wes_client_main.py

@@ -62,17 +62,17 @@ def main(argv=sys.argv[1:]):
         http_client=http_client, config={"use_models": False})
 
     if args.list:
-        response = client.WorkflowExecutionService.ListWorkflows(page_token=args.page, page_size=args.page_size)
+        response = client.WorkflowExecutionService.ListRuns(page_token=args.page, page_size=args.page_size)
         json.dump(response.result(), sys.stdout, indent=4)
         return 0
 
     if args.log:
-        response = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=args.log)
+        response = client.WorkflowExecutionService.GetRunLog(workflow_id=args.log)
         sys.stdout.write(response.result()["workflow_log"]["stderr"])
         return 0
 
     if args.get:
-        response = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=args.get)
+        response = client.WorkflowExecutionService.GetRunLog(workflow_id=args.get)
         json.dump(response.result(), sys.stdout, indent=4)
         return 0
 
@@ -147,14 +147,14 @@ def fixpaths(d):
         sys.stdout.write(r["workflow_id"] + "\n")
         exit(0)
 
-    r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r["workflow_id"]).result()
+    r = client.WorkflowExecutionService.GetRunStatus(workflow_id=r["workflow_id"]).result()
     while r["state"] in ("QUEUED", "INITIALIZING", "RUNNING"):
         time.sleep(8)
-        r = client.WorkflowExecutionService.GetWorkflowStatus(workflow_id=r["workflow_id"]).result()
+        r = client.WorkflowExecutionService.GetRunStatus(workflow_id=r["workflow_id"]).result()
 
     logging.info("State is %s", r["state"])
 
-    s = client.WorkflowExecutionService.GetWorkflowLog(workflow_id=r["workflow_id"]).result()
+    s = client.WorkflowExecutionService.GetRunLog(workflow_id=r["workflow_id"]).result()
 
     try:
         # TODO: Only works with Arvados atm

wes_service/arvados_wes.py

@@ -81,7 +81,7 @@ def GetServiceInfo(self):
         }
 
     @catch_exceptions
-    def ListWorkflows(self, page_size=None, page_token=None, tag_search=None, state_search=None):
+    def ListRuns(self, page_size=None, page_token=None, state_search=None):
         api = get_api()
 
         paging = []
@@ -100,13 +100,13 @@ def ListWorkflows(self, page_size=None, page_token=None, tag_search=None, state_
 
         uuidmap = {c["uuid"]: statemap[c["state"]] for c in containers}
 
-        workflow_list = [{"workflow_id": cr["uuid"],
+        workflow_list = [{"run_id": cr["uuid"],
                           "state": uuidmap.get(cr["container_uuid"])}
                          for cr in requests
                          if cr["command"] and cr["command"][0] == "arvados-cwl-runner"]
         return {
             "workflows": workflow_list,
-            "next_page_token": workflow_list[-1]["workflow_id"] if workflow_list else ""
+            "next_page_token": workflow_list[-1]["run_id"] if workflow_list else ""
         }
 
     def invoke_cwl_runner(self, cr_uuid, workflow_url, workflow_params,
@@ -215,13 +215,13 @@ def RunWorkflow(self, workflow_params, workflow_type, workflow_type_version,
                                                               project_uuid,
                                                               tempdir)).start()
 
-        return {"workflow_id": cr["uuid"]}
+        return {"run_id": cr["uuid"]}
 
     @catch_exceptions
-    def GetWorkflowLog(self, workflow_id):
+    def GetRunLog(self, run_id):
         api = get_api()
 
-        request = api.container_requests().get(uuid=workflow_id).execute()
+        request = api.container_requests().get(uuid=run_id).execute()
         if request["container_uuid"]:
             container = api.containers().get(uuid=request["container_uuid"]).execute()  # NOQA
             task_reqs = arvados.util.list_all(api.container_requests().list, filters=[["requesting_container_uuid", "=", container["uuid"]]])
@@ -273,7 +273,7 @@ def log_object(cr):
             return r
 
         r = {
-            "workflow_id": request["uuid"],
+            "run_id": request["uuid"],
             "request": {
                 "workflow_url": "",
                 "workflow_params": request["mounts"].get("/var/lib/cwl/cwl.input.json", {}).get("content", {})
@@ -287,30 +287,30 @@ def log_object(cr):
         return r
 
     @catch_exceptions
-    def CancelJob(self, workflow_id):  # NOQA
+    def CancelRun(self, run_id):  # NOQA
         api = get_api()
-        request = api.container_requests().update(uuid=workflow_id, body={"priority": 0}).execute()  # NOQA
-        return {"workflow_id": request["uuid"]}
+        request = api.container_requests().update(uuid=run_id, body={"priority": 0}).execute()  # NOQA
+        return {"run_id": request["uuid"]}
 
     @catch_exceptions
-    def GetWorkflowStatus(self, workflow_id):
+    def GetRunStatus(self, run_id):
         api = get_api()
-        request = api.container_requests().get(uuid=workflow_id).execute()
+        request = api.container_requests().get(uuid=run_id).execute()
         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"],
+        return {"run_id": request["uuid"],
                 "state": statemap[container["state"]]}
 
 
-def dynamic_logs(workflow_id, logstream):
+def dynamic_logs(run_id, logstream):
     api = get_api()
-    cr = api.container_requests().get(uuid=workflow_id).execute()
+    cr = api.container_requests().get(uuid=run_id).execute()
     l1 = [t["properties"]["text"]
-          for t in api.logs().list(filters=[["object_uuid", "=", workflow_id],
+          for t in api.logs().list(filters=[["object_uuid", "=", run_id],
                                             ["event_type", "=", logstream]],
                                    order="created_at desc",
                                    limit=100).execute()["items"]]
@@ -327,5 +327,5 @@ def dynamic_logs(workflow_id, logstream):
 
 def create_backend(app, opts):
     ab = ArvadosBackend(opts)
-    app.app.route('/ga4gh/wes/v1/workflows/<workflow_id>/x-dynamic-logs/<logstream>')(dynamic_logs)
+    app.app.route('/ga4gh/wes/v1/runs/<run_id>/x-dynamic-logs/<logstream>')(dynamic_logs)
     return ab

wes_service/cwl_runner.py

@@ -13,10 +13,10 @@
 
 
 class Workflow(object):
-    def __init__(self, workflow_id):
+    def __init__(self, run_id):
         super(Workflow, self).__init__()
-        self.workflow_id = workflow_id
-        self.workdir = os.path.join(os.getcwd(), "workflows", self.workflow_id)
+        self.run_id = run_id
+        self.workdir = os.path.join(os.getcwd(), "workflows", self.run_id)
 
     def run(self, request, opts):
         """
@@ -37,7 +37,7 @@ def run(self, request, opts):
         :param dict request: A dictionary containing the cwl/json information.
         :param wes_service.util.WESBackend opts: contains the user's arguments;
                                                  specifically the runner and runner options
-        :return: {"workflow_id": self.workflow_id, "state": state}
+        :return: {"run_id": self.run_id, "state": state}
         """
         os.makedirs(self.workdir)
         outdir = os.path.join(self.workdir, "outdir")
@@ -117,7 +117,7 @@ def getstatus(self):
         state, exit_code = self.getstate()
 
         return {
-            "workflow_id": self.workflow_id,
+            "run_id": self.run_id,
             "state": state
         }
 
@@ -137,7 +137,7 @@ def getlog(self):
                 outputobj = json.load(outputtemp)
 
         return {
-            "workflow_id": self.workflow_id,
+            "run_id": self.run_id,
             "request": request,
             "state": state,
             "workflow_log": {
@@ -169,14 +169,14 @@ def GetServiceInfo(self):
             "key_values": {}
         }
 
-    def ListWorkflows(self):
+    def ListRuns(self, page_size=None, page_token=None, state_search=None):
         # FIXME #15 results don't page
         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))
 
-        workflows = [{"workflow_id": w.workflow_id, "state": w.getstate()[0]} for w in wf]  # NOQA
+        workflows = [{"run_id": w.run_id, "state": w.getstate()[0]} for w in wf]  # NOQA
         return {
             "workflows": workflows,
             "next_page_token": ""
@@ -204,23 +204,23 @@ def RunWorkflow(self):
         if index > 0:
             body["workflow_url"] = body["workflow_url"][index:]
 
-        workflow_id = uuid.uuid4().hex
-        job = Workflow(workflow_id)
+        run_id = uuid.uuid4().hex
+        job = Workflow(run_id)
 
         job.run(body, self)
-        return {"workflow_id": workflow_id}
+        return {"run_id": run_id}
 
-    def GetWorkflowLog(self, workflow_id):
-        job = Workflow(workflow_id)
+    def GetRunLog(self, run_id):
+        job = Workflow(run_id)
         return job.getlog()
 
-    def CancelJob(self, workflow_id):
-        job = Workflow(workflow_id)
+    def CancelRun(self, run_id):
+        job = Workflow(run_id)
         job.cancel()
-        return {"workflow_id": workflow_id}
+        return {"run_id": run_id}
 
-    def GetWorkflowStatus(self, workflow_id):
-        job = Workflow(workflow_id)
+    def GetRunStatus(self, run_id):
+        job = Workflow(run_id)
         return job.getstatus()