From 8668135855b400e7f6047ecacd9cfa27fed723f7 Mon Sep 17 00:00:00 2001 From: radhika Date: Wed, 16 Nov 2016 15:25:01 -0500 Subject: [PATCH] 10293: update cwl runner --- sdk/cwl/arvados_cwl/__init__.py | 4 +-- sdk/cwl/arvados_cwl/arvcontainer.py | 35 ++++++++-------------- sdk/cwl/arvados_cwl/done.py | 3 ++ sdk/cwl/tests/test_container.py | 45 ++++++++--------------------- 4 files changed, 29 insertions(+), 58 deletions(-) diff --git a/sdk/cwl/arvados_cwl/__init__.py b/sdk/cwl/arvados_cwl/__init__.py index 92be92d6e0..f3e1a2622d 100644 --- a/sdk/cwl/arvados_cwl/__init__.py +++ b/sdk/cwl/arvados_cwl/__init__.py @@ -119,7 +119,7 @@ class ArvCwlRunner(object): logger.info("Job %s (%s) is Running", j.name, uuid) j.running = True j.update_pipeline_component(event["properties"]["new_attributes"]) - elif event["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled"): + elif event["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled", "Final"): uuid = event["object_uuid"] try: self.cond.acquire() @@ -148,7 +148,7 @@ class ArvCwlRunner(object): continue if self.work_api == "containers": - table = self.poll_api.containers() + table = self.poll_api.container_requests() elif self.work_api == "jobs": table = self.poll_api.jobs() diff --git a/sdk/cwl/arvados_cwl/arvcontainer.py b/sdk/cwl/arvados_cwl/arvcontainer.py index aa088c5e8a..e9e6b3220b 100644 --- a/sdk/cwl/arvados_cwl/arvcontainer.py +++ b/sdk/cwl/arvados_cwl/arvcontainer.py @@ -115,24 +115,23 @@ class ArvadosContainer(object): body=container_request ).execute(num_retries=self.arvrunner.num_retries) - self.arvrunner.processes[response["container_uuid"]] = self + self.arvrunner.processes[response["uuid"]] = response["uuid"] - container = self.arvrunner.api.containers().get( - uuid=response["container_uuid"] - ).execute(num_retries=self.arvrunner.num_retries) + logger.info("Container request %s (%s) state is %s with container %s %s", self.name, response["uuid"], response["state"]) - logger.info("Container request %s (%s) state is %s with container %s %s", self.name, response["uuid"], response["state"], container["uuid"], container["state"]) - - if container["state"] in ("Complete", "Cancelled"): - self.done(container) + if response["state"] == "Final": + self.done(response) except Exception as e: logger.error("Got error %s" % str(e)) self.output_callback({}, "permanentFail") def done(self, record): try: - if record["state"] == "Complete": - rcode = record["exit_code"] + container = self.arvrunner.api.containers().get( + uuid=record["container_uuid"] + ).execute(num_retries=self.arvrunner.num_retries) + if container["state"] == "Complete": + rcode = container["exit_code"] if self.successCodes and rcode in self.successCodes: processStatus = "success" elif self.temporaryFailCodes and rcode in self.temporaryFailCodes: @@ -146,17 +145,7 @@ class ArvadosContainer(object): else: processStatus = "permanentFail" - try: - outputs = {} - if record["output"]: - outputs = done.done(self, record, "/tmp", self.outdir, "/keep") - except WorkflowException as e: - logger.error("Error while collecting container outputs:\n%s", e, exc_info=(e if self.arvrunner.debug else False)) - processStatus = "permanentFail" - except Exception as e: - logger.exception("Got unknown exception while collecting job outputs:") - processStatus = "permanentFail" - + outputs = done.done_outputs(self, container, "/tmp", self.outdir, "/keep") self.output_callback(outputs, processStatus) finally: del self.arvrunner.processes[record["uuid"]] @@ -242,9 +231,9 @@ class RunnerContainer(Runner): ).execute(num_retries=self.arvrunner.num_retries) self.uuid = response["uuid"] - self.arvrunner.processes[response["container_uuid"]] = self + self.arvrunner.processes[response["uuid"]] = response["uuid"] logger.info("Submitted container %s", response["uuid"]) - if response["state"] in ("Complete", "Failed", "Cancelled"): + if response["state"] == "Final": self.done(response) diff --git a/sdk/cwl/arvados_cwl/done.py b/sdk/cwl/arvados_cwl/done.py index 31f353e5d8..7bdae5a1bd 100644 --- a/sdk/cwl/arvados_cwl/done.py +++ b/sdk/cwl/arvados_cwl/done.py @@ -35,6 +35,9 @@ def done(self, record, tmpdir, outdir, keepdir): }, ensure_unique_name=True).execute( num_retries=self.arvrunner.num_retries) + done_outputs(self, record, tmpdir, outdir, keepdir) + +def done_outputs(self, record, tmpdir, outdir, keepdir): self.builder.outdir = outdir self.builder.pathmapper.keepdir = keepdir return self.collect_outputs("keep:" + record["output"]) diff --git a/sdk/cwl/tests/test_container.py b/sdk/cwl/tests/test_container.py index bb4bac31dd..b13bcdecb3 100644 --- a/sdk/cwl/tests/test_container.py +++ b/sdk/cwl/tests/test_container.py @@ -172,29 +172,14 @@ class TestContainer(unittest.TestCase): arvjob.outdir = "/var/spool/cwl" arvjob.done({ - "state": "Complete", - "output": "99999999999999999999999999999993+99", - "log": "99999999999999999999999999999994+99", - "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz", - "exit_code": 0 + "state": "Final", + "log_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz1", + "output_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2", + "uuid": "zzzzz-xvhdp-zzzzzzzzzzzzzzz", + "container_uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz" }) - api.collections().list.assert_has_calls([ - mock.call(), - mock.call(filters=[['owner_uuid', '=', 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'], - ['portable_data_hash', '=', '99999999999999999999999999999993+99'], - ['name', '=', 'Output 9999999 of testjob']]), - mock.call().execute(num_retries=0), - mock.call(limit=1, filters=[['portable_data_hash', '=', '99999999999999999999999999999993+99']], - select=['manifest_text']), - mock.call().execute(num_retries=0)]) - - api.collections().create.assert_called_with( - ensure_unique_name=True, - body={'portable_data_hash': '99999999999999999999999999999993+99', - 'manifest_text': 'XYZ', - 'owner_uuid': 'zzzzz-8i9sb-zzzzzzzzzzzzzzz', - 'name': 'Output 9999999 of testjob'}) + self.assertFalse(api.collections().create.called) @mock.patch("arvados.collection.Collection") def test_done_use_existing_collection(self, col): @@ -217,18 +202,12 @@ class TestContainer(unittest.TestCase): arvjob.outdir = "/var/spool/cwl" arvjob.done({ - "state": "Complete", - "output": "99999999999999999999999999999993+99", - "log": "99999999999999999999999999999994+99", - "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz", - "exit_code": 0 + "state": "Final", + "log_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz1", + "output_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2", + "log_uuid": "zzzzz-4zz18-zzzzzzzzzzzzzz2", + "uuid": "zzzzz-xvhdp-zzzzzzzzzzzzzzz", + "container_uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz" }) - api.collections().list.assert_has_calls([ - mock.call(), - mock.call(filters=[['owner_uuid', '=', 'zzzzz-8i9sb-zzzzzzzzzzzzzzz'], - ['portable_data_hash', '=', '99999999999999999999999999999993+99'], - ['name', '=', 'Output 9999999 of testjob']]), - mock.call().execute(num_retries=0)]) - self.assertFalse(api.collections().create.called) -- 2.30.2