[ARVADOS] updated: 7cd03709ba594237b9cf88afac76d39f28011b35
Git user
git at public.curoverse.com
Tue Jun 21 10:02:20 EDT 2016
Summary of changes:
sdk/cwl/arvados_cwl/__init__.py | 73 ++++++++++++++++++++-----------------
sdk/cwl/arvados_cwl/arvcontainer.py | 8 ++--
sdk/cwl/arvados_cwl/arvjob.py | 6 +--
sdk/cwl/arvados_cwl/arvtool.py | 4 +-
sdk/cwl/arvados_cwl/runner.py | 2 +-
sdk/cwl/tests/test_container.py | 12 ++++--
sdk/cwl/tests/test_job.py | 4 +-
sdk/cwl/tests/test_submit.py | 19 ++++++----
8 files changed, 71 insertions(+), 57 deletions(-)
via 7cd03709ba594237b9cf88afac76d39f28011b35 (commit)
via bb48c3155b90e4bf1994a66b8fab858f12a1f4f4 (commit)
from 2435a72a0a0c0058d2825ea0840836e95ebe5f25 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit 7cd03709ba594237b9cf88afac76d39f28011b35
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Tue Jun 21 08:22:54 2016 -0400
8442: Fix tests. Update comments. Rename jobs -> processes since it doesn't hold just "jobs" any more.
diff --git a/sdk/cwl/arvados_cwl/__init__.py b/sdk/cwl/arvados_cwl/__init__.py
index 9b92e6f..69c933c 100644
--- a/sdk/cwl/arvados_cwl/__init__.py
+++ b/sdk/cwl/arvados_cwl/__init__.py
@@ -1,6 +1,7 @@
#!/usr/bin/env python
-# Implement cwl-runner interface for submitting and running jobs on Arvados.
+# Implement cwl-runner interface for submitting and running work on Arvados, using
+# either the Crunch jobs API or Crunch containers API.
import argparse
import logging
@@ -28,12 +29,14 @@ logger = logging.getLogger('arvados.cwl-runner')
logger.setLevel(logging.INFO)
class ArvCwlRunner(object):
- """Execute a CWL tool or workflow, submit crunch jobs, wait for them to
- complete, and report output."""
+ """Execute a CWL tool or workflow, submit work (using either jobs or
+ containers API), wait for them to complete, and report output.
- def __init__(self, api_client, crunch2=False):
+ """
+
+ def __init__(self, api_client, work_api):
self.api = api_client
- self.jobs = {}
+ self.processes = {}
self.lock = threading.Lock()
self.cond = threading.Condition(self.lock)
self.final_output = None
@@ -41,22 +44,25 @@ class ArvCwlRunner(object):
self.uploaded = {}
self.num_retries = 4
self.uuid = None
- self.crunch2 = crunch2
+ self.work_api = work_api
+
+ if self.work_api not in ("containers", "jobs"):
+ raise Exception("Unsupported API '%s'" % self.work_api)
def arvMakeTool(self, toolpath_object, **kwargs):
if "class" in toolpath_object and toolpath_object["class"] == "CommandLineTool":
- return ArvadosCommandTool(self, toolpath_object, crunch2=self.crunch2, **kwargs)
+ return ArvadosCommandTool(self, toolpath_object, work_api=self.work_api, **kwargs)
else:
return cwltool.workflow.defaultMakeTool(toolpath_object, **kwargs)
def output_callback(self, out, processStatus):
if processStatus == "success":
- logger.info("Overall job status is %s", processStatus)
+ logger.info("Overall process status is %s", processStatus)
if self.pipeline:
self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
body={"state": "Complete"}).execute(num_retries=self.num_retries)
else:
- logger.warn("Overall job status is %s", processStatus)
+ logger.warn("Overall process status is %s", processStatus)
if self.pipeline:
self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
body={"state": "Failed"}).execute(num_retries=self.num_retries)
@@ -65,11 +71,11 @@ class ArvCwlRunner(object):
def on_message(self, event):
if "object_uuid" in event:
- if event["object_uuid"] in self.jobs and event["event_type"] == "update":
- if event["properties"]["new_attributes"]["state"] == "Running" and self.jobs[event["object_uuid"]].running is False:
+ if event["object_uuid"] in self.processes and event["event_type"] == "update":
+ if event["properties"]["new_attributes"]["state"] == "Running" and self.processes[event["object_uuid"]].running is False:
uuid = event["object_uuid"]
with self.lock:
- j = self.jobs[uuid]
+ j = self.processes[uuid]
logger.info("Job %s (%s) is Running", j.name, uuid)
j.running = True
j.update_pipeline_component(event["properties"]["new_attributes"])
@@ -77,7 +83,7 @@ class ArvCwlRunner(object):
uuid = event["object_uuid"]
try:
self.cond.acquire()
- j = self.jobs[uuid]
+ j = self.processes[uuid]
logger.info("Job %s (%s) is %s", j.name, uuid, event["properties"]["new_attributes"]["state"])
j.done(event["properties"]["new_attributes"])
self.cond.notify()
@@ -114,16 +120,16 @@ class ArvCwlRunner(object):
kwargs["fs_access"] = self.fs_access
kwargs["enable_reuse"] = kwargs.get("enable_reuse")
- if self.crunch2:
+ if self.work_api == "containers":
kwargs["outdir"] = "/var/spool/cwl"
kwargs["tmpdir"] = "/tmp"
- else:
+ elif self.work_api == "jobs":
kwargs["outdir"] = "$(task.outdir)"
kwargs["tmpdir"] = "$(task.tmpdir)"
runnerjob = None
if kwargs.get("submit"):
- if self.crunch2:
+ if self.work_api == "containers":
if tool.tool["class"] == "CommandLineTool":
runnerjob = tool.job(job_order,
self.output_callback,
@@ -133,7 +139,7 @@ class ArvCwlRunner(object):
else:
runnerjob = RunnerJob(self, tool, job_order, kwargs.get("enable_reuse"))
- if not kwargs.get("submit") and "cwl_runner_job" not in kwargs and not self.crunch2:
+ if not kwargs.get("submit") and "cwl_runner_job" not in kwargs and not self.work_api == "containers":
# Create pipeline for local run
self.pipeline = self.api.pipeline_instances().create(
body={
@@ -147,9 +153,9 @@ class ArvCwlRunner(object):
runnerjob.run()
return runnerjob.uuid
- if self.crunch2:
+ if self.work_api == "containers":
events = arvados.events.subscribe(arvados.api('v1'), [["object_uuid", "is_a", "arvados#container"]], self.on_message)
- else:
+ if self.work_api == "jobs":
events = arvados.events.subscribe(arvados.api('v1'), [["object_uuid", "is_a", "arvados#job"]], self.on_message)
if runnerjob:
@@ -172,13 +178,13 @@ class ArvCwlRunner(object):
if runnable:
runnable.run(**kwargs)
else:
- if self.jobs:
+ if self.processes:
self.cond.wait(1)
else:
logger.error("Workflow is deadlocked, no runnable jobs and not waiting on any pending jobs.")
break
- while self.jobs:
+ while self.processes:
self.cond.wait(1)
events.close()
@@ -192,7 +198,7 @@ class ArvCwlRunner(object):
if self.pipeline:
self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
body={"state": "Failed"}).execute(num_retries=self.num_retries)
- if runnerjob and runnerjob.uuid and self.crunch2:
+ if runnerjob and runnerjob.uuid and self.work_api == "containers":
self.api.container_requests().update(uuid=runnerjob.uuid,
body={"priority": "0"}).execute(num_retries=self.num_retries)
finally:
@@ -222,7 +228,6 @@ def versionstring():
def arg_parser(): # type: () -> argparse.ArgumentParser
parser = argparse.ArgumentParser(description='Arvados executor for Common Workflow Language')
- parser.add_argument("--conformance-test", action="store_true")
parser.add_argument("--basedir", type=str,
help="Base directory used to resolve relative references in the input, default to directory of input object file or current directory (if inputs piped/provided on command line).")
parser.add_argument("--outdir", type=str, default=os.path.abspath('.'),
@@ -268,11 +273,11 @@ def arg_parser(): # type: () -> argparse.ArgumentParser
default=True, dest="wait")
parser.add_argument("--api", type=str,
- default=None,
+ default=None, dest="work_api",
help="Select work submission API, one of 'jobs' or 'containers'.")
- parser.add_argument("workflow", type=str, nargs="?", default=None)
- parser.add_argument("job_order", nargs=argparse.REMAINDER)
+ parser.add_argument("workflow", type=str, nargs="?", default=None, help="The workflow to execute")
+ parser.add_argument("job_order", nargs=argparse.REMAINDER, help="The input object to the workflow.")
return parser
@@ -285,14 +290,19 @@ def main(args, stdout, stderr, api_client=None):
if arvargs.create_template and not arvargs.job_order:
job_order_object = ({}, "")
+ if arvargs.work_api is None:
+ arvargs.work_api = "jobs"
+
try:
if api_client is None:
api_client=arvados.api('v1', model=OrderedJsonModel())
- runner = ArvCwlRunner(api_client, crunch2=(arvargs.api == "containers"))
+ runner = ArvCwlRunner(api_client, work_api=arvargs.work_api)
except Exception as e:
logger.error(e)
return 1
+ arvargs.conformance_test = None
+
return cwltool.main.main(args=arvargs,
stdout=stdout,
stderr=stderr,
diff --git a/sdk/cwl/arvados_cwl/arvcontainer.py b/sdk/cwl/arvados_cwl/arvcontainer.py
index 8b5ac5a..384b03b 100644
--- a/sdk/cwl/arvados_cwl/arvcontainer.py
+++ b/sdk/cwl/arvados_cwl/arvcontainer.py
@@ -98,7 +98,7 @@ class ArvadosContainer(object):
body=container_request
).execute(num_retries=self.arvrunner.num_retries)
- self.arvrunner.jobs[response["container_uuid"]] = self
+ self.arvrunner.processes[response["container_uuid"]] = self
logger.info("Container %s (%s) request state is %s", self.name, response["container_uuid"], response["state"])
@@ -138,7 +138,7 @@ class ArvadosContainer(object):
self.output_callback(outputs, processStatus)
finally:
- del self.arvrunner.jobs[record["uuid"]]
+ del self.arvrunner.processes[record["uuid"]]
class RunnerContainer(Runner):
@@ -171,7 +171,7 @@ class RunnerContainer(Runner):
self.arvrunner.project_uuid)
return {
- "command": ["arvados-cwl-runner", "--local", "--crunch2", workflowpath, jobpath],
+ "command": ["arvados-cwl-runner", "--local", "--api=containers", workflowpath, jobpath],
"owner_uuid": self.arvrunner.project_uuid,
"name": self.name,
"output_path": "/var/spool/cwl",
@@ -214,7 +214,7 @@ class RunnerContainer(Runner):
).execute(num_retries=self.arvrunner.num_retries)
self.uuid = response["uuid"]
- self.arvrunner.jobs[response["container_uuid"]] = self
+ self.arvrunner.processes[response["container_uuid"]] = self
logger.info("Submitted container %s", response["uuid"])
diff --git a/sdk/cwl/arvados_cwl/arvjob.py b/sdk/cwl/arvados_cwl/arvjob.py
index a816441..693bbeb 100644
--- a/sdk/cwl/arvados_cwl/arvjob.py
+++ b/sdk/cwl/arvados_cwl/arvjob.py
@@ -90,7 +90,7 @@ class ArvadosJob(object):
find_or_create=kwargs.get("enable_reuse", True)
).execute(num_retries=self.arvrunner.num_retries)
- self.arvrunner.jobs[response["uuid"]] = self
+ self.arvrunner.processes[response["uuid"]] = self
self.update_pipeline_component(response)
@@ -171,7 +171,7 @@ class ArvadosJob(object):
self.output_callback(outputs, processStatus)
finally:
- del self.arvrunner.jobs[record["uuid"]]
+ del self.arvrunner.processes[record["uuid"]]
class RunnerJob(Runner):
@@ -208,7 +208,7 @@ class RunnerJob(Runner):
).execute(num_retries=self.arvrunner.num_retries)
self.uuid = response["uuid"]
- self.arvrunner.jobs[self.uuid] = self
+ self.arvrunner.processes[self.uuid] = self
logger.info("Submitted job %s", response["uuid"])
diff --git a/sdk/cwl/arvados_cwl/arvtool.py b/sdk/cwl/arvados_cwl/arvtool.py
index a2c5c9e..0601fa6 100644
--- a/sdk/cwl/arvados_cwl/arvtool.py
+++ b/sdk/cwl/arvados_cwl/arvtool.py
@@ -6,10 +6,10 @@ from .pathmapper import ArvPathMapper
class ArvadosCommandTool(CommandLineTool):
"""Wrap cwltool CommandLineTool to override selected methods."""
- def __init__(self, arvrunner, toolpath_object, crunch2, **kwargs):
+ def __init__(self, arvrunner, toolpath_object, **kwargs):
super(ArvadosCommandTool, self).__init__(toolpath_object, **kwargs)
self.arvrunner = arvrunner
- self.crunch2 = crunch2
+ self.crunch2 = (kwargs["work_api"] == "containers")
def makeJobRunner(self):
if self.crunch2:
diff --git a/sdk/cwl/arvados_cwl/runner.py b/sdk/cwl/arvados_cwl/runner.py
index 639426a..629b104 100644
--- a/sdk/cwl/arvados_cwl/runner.py
+++ b/sdk/cwl/arvados_cwl/runner.py
@@ -119,4 +119,4 @@ class Runner(object):
logger.error("While getting final output object: %s", e)
self.arvrunner.output_callback(outputs, processStatus)
finally:
- del self.arvrunner.jobs[record["uuid"]]
+ del self.arvrunner.processes[record["uuid"]]
diff --git a/sdk/cwl/tests/test_container.py b/sdk/cwl/tests/test_container.py
index 0237e80..df7d142 100644
--- a/sdk/cwl/tests/test_container.py
+++ b/sdk/cwl/tests/test_container.py
@@ -31,7 +31,7 @@ class TestContainer(unittest.TestCase):
"outputs": [],
"baseCommand": "ls"
}
- arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, True, avsc_names=avsc_names, basedir="")
+ arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="containers", avsc_names=avsc_names, basedir="")
arvtool.formatgraph = None
for j in arvtool.job({}, mock.MagicMock(), basedir="", name="test_run"):
j.run()
@@ -80,7 +80,7 @@ class TestContainer(unittest.TestCase):
}],
"baseCommand": "ls"
}
- arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, True, avsc_names=avsc_names)
+ arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="containers", avsc_names=avsc_names)
arvtool.formatgraph = None
for j in arvtool.job({}, mock.MagicMock(), basedir="", name="test_resource_requirements"):
j.run()
@@ -125,12 +125,14 @@ class TestContainer(unittest.TestCase):
arvjob.builder = mock.MagicMock()
arvjob.output_callback = mock.MagicMock()
arvjob.collect_outputs = mock.MagicMock()
+ arvjob.successCodes = [0]
arvjob.done({
"state": "Complete",
"output": "99999999999999999999999999999993+99",
"log": "99999999999999999999999999999994+99",
- "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
+ "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz",
+ "exit_code": 0
})
api.collections().list.assert_has_calls([
@@ -167,12 +169,14 @@ class TestContainer(unittest.TestCase):
arvjob.builder = mock.MagicMock()
arvjob.output_callback = mock.MagicMock()
arvjob.collect_outputs = mock.MagicMock()
+ arvjob.successCodes = [0]
arvjob.done({
"state": "Complete",
"output": "99999999999999999999999999999993+99",
"log": "99999999999999999999999999999994+99",
- "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz"
+ "uuid": "zzzzz-8i9sb-zzzzzzzzzzzzzzz",
+ "exit_code": 0
})
api.collections().list.assert_has_calls([
diff --git a/sdk/cwl/tests/test_job.py b/sdk/cwl/tests/test_job.py
index 701afcb..abaf161 100644
--- a/sdk/cwl/tests/test_job.py
+++ b/sdk/cwl/tests/test_job.py
@@ -25,7 +25,7 @@ class TestJob(unittest.TestCase):
"outputs": [],
"baseCommand": "ls"
}
- arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, False, avsc_names=avsc_names, basedir="")
+ arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="jobs", avsc_names=avsc_names, basedir="")
arvtool.formatgraph = None
for j in arvtool.job({}, mock.MagicMock(), basedir=""):
j.run()
@@ -76,7 +76,7 @@ class TestJob(unittest.TestCase):
}],
"baseCommand": "ls"
}
- arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, False, avsc_names=avsc_names)
+ arvtool = arvados_cwl.ArvadosCommandTool(runner, tool, work_api="jobs", avsc_names=avsc_names)
arvtool.formatgraph = None
for j in arvtool.job({}, mock.MagicMock(), basedir=""):
j.run()
diff --git a/sdk/cwl/tests/test_submit.py b/sdk/cwl/tests/test_submit.py
index 48e6ed2..6f056ab 100644
--- a/sdk/cwl/tests/test_submit.py
+++ b/sdk/cwl/tests/test_submit.py
@@ -90,27 +90,32 @@ def stubs(func):
stubs.expect_container_spec = {
'priority': 1,
'mounts': {
+ '/var/spool/cwl': {
+ 'writable': True,
+ 'kind': 'collection'
+ },
+ '/var/lib/cwl/workflow': {
+ 'portable_data_hash': '99999999999999999999999999999991+99',
+ 'kind': 'collection'
+ },
'stdout': {
'path': '/var/spool/cwl/cwl.output.json',
'kind': 'file'
},
- '/var/lib/cwl/workflow/submit_wf.cwl': {
- 'portable_data_hash': '999999999999999999999999991+99',
- 'kind': 'collection'
- },
'/var/lib/cwl/job/cwl.input.json': {
- 'portable_data_hash': '102435082199e5229f99b01165b67096+60/cwl.input.json',
+ 'portable_data_hash': '33be5c865fe12e1e4788d2f1bc627f7a+60/cwl.input.json',
'kind': 'collection'
}
},
'state': 'Committed',
'owner_uuid': 'zzzzz-tpzed-zzzzzzzzzzzzzzz',
- 'command': ['arvados-cwl-runner', '--local', '--crunch2', '/var/lib/cwl/workflow/submit_wf.cwl', '/var/lib/cwl/job/cwl.input.json'],
+ 'command': ['arvados-cwl-runner', '--local', '--api=containers', '/var/lib/cwl/workflow/submit_wf.cwl', '/var/lib/cwl/job/cwl.input.json'],
'name': 'submit_wf.cwl',
'container_image': '99999999999999999999999999999993+99',
'output_path': '/var/spool/cwl',
'cwd': '/var/spool/cwl',
'runtime_constraints': {
+ 'API': True,
'vcpus': 1,
'ram': 268435456
}
@@ -177,7 +182,7 @@ class TestSubmit(unittest.TestCase):
def test_submit_container(self, stubs):
capture_stdout = cStringIO.StringIO()
exited = arvados_cwl.main(
- ["--submit", "--no-wait", "--crunch2", "--debug",
+ ["--submit", "--no-wait", "--api=containers", "--debug",
"tests/wf/submit_wf.cwl", "tests/submit_test_job.json"],
capture_stdout, sys.stderr, api_client=stubs.api)
self.assertEqual(exited, 0)
commit bb48c3155b90e4bf1994a66b8fab858f12a1f4f4
Author: Peter Amstutz <peter.amstutz at curoverse.com>
Date: Mon Jun 20 17:34:07 2016 -0400
8442: Rename --crunch1 and --crunch2 options to --api=jobs and --api=containers
diff --git a/sdk/cwl/arvados_cwl/__init__.py b/sdk/cwl/arvados_cwl/__init__.py
index e14eb4b..9b92e6f 100644
--- a/sdk/cwl/arvados_cwl/__init__.py
+++ b/sdk/cwl/arvados_cwl/__init__.py
@@ -267,14 +267,9 @@ def arg_parser(): # type: () -> argparse.ArgumentParser
exgroup.add_argument("--no-wait", action="store_false", help="Submit workflow runner job and exit.",
default=True, dest="wait")
- exgroup = parser.add_mutually_exclusive_group()
- exgroup.add_argument("--crunch1", action="store_false",
- default=False, dest="crunch2",
- help="Use Crunch v1 Jobs API")
-
- exgroup.add_argument("--crunch2", action="store_true",
- default=False, dest="crunch2",
- help="Use Crunch v2 Containers API")
+ parser.add_argument("--api", type=str,
+ default=None,
+ help="Select work submission API, one of 'jobs' or 'containers'.")
parser.add_argument("workflow", type=str, nargs="?", default=None)
parser.add_argument("job_order", nargs=argparse.REMAINDER)
@@ -293,7 +288,7 @@ def main(args, stdout, stderr, api_client=None):
try:
if api_client is None:
api_client=arvados.api('v1', model=OrderedJsonModel())
- runner = ArvCwlRunner(api_client, crunch2=arvargs.crunch2)
+ runner = ArvCwlRunner(api_client, crunch2=(arvargs.api == "containers"))
except Exception as e:
logger.error(e)
return 1
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list