from cwltool.errors import WorkflowException
import cwltool.main
import cwltool.workflow
+import cwltool.process
import schema_salad
+from schema_salad.sourceline import SourceLine
import arvados
import arvados.config
metrics = logging.getLogger('arvados.cwl-runner.metrics')
logger.setLevel(logging.INFO)
+arvados.log_handler.setFormatter(logging.Formatter(
+ '%(asctime)s %(name)s %(levelname)s: %(message)s',
+ '%Y-%m-%d %H:%M:%S'))
class ArvCwlRunner(object):
"""Execute a CWL tool or workflow, submit work (using either jobs or
uuid = event["object_uuid"]
with self.lock:
j = self.processes[uuid]
- logger.info("Job %s (%s) is Running", j.name, uuid)
+ logger.info("%s %s is Running", self.label(j), uuid)
j.running = True
j.update_pipeline_component(event["properties"]["new_attributes"])
elif event["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled", "Final"):
try:
self.cond.acquire()
j = self.processes[uuid]
- txt = self.work_api[0].upper() + self.work_api[1:-1]
- logger.info("%s %s (%s) is %s", txt, j.name, uuid, event["properties"]["new_attributes"]["state"])
+ logger.info("%s %s is %s", self.label(j), uuid, event["properties"]["new_attributes"]["state"])
with Perf(metrics, "done %s" % j.name):
j.done(event["properties"]["new_attributes"])
self.cond.notify()
finally:
self.cond.release()
+ def label(self, obj):
+ return "[%s %s]" % (self.work_api[0:-1], obj.name)
+
def poll_states(self):
"""Poll status of jobs or containers listed in the processes dict.
def add_uploaded(self, src, pair):
self.uploaded[src] = pair
- def check_writable(self, obj):
+ def check_features(self, obj):
if isinstance(obj, dict):
+ if obj.get("class") == "InitialWorkDirRequirement":
+ if self.work_api == "containers":
+ raise UnsupportedRequirement("InitialWorkDirRequirement not supported with --api=containers")
if obj.get("writable"):
- raise UnsupportedRequirement("InitialWorkDir feature 'writable: true' not supported")
+ raise SourceLine(obj, "writable", UnsupportedRequirement).makeError("InitialWorkDir feature 'writable: true' not supported")
+ if obj.get("class") == "CommandLineTool":
+ if self.work_api == "containers":
+ if obj.get("stdin"):
+ raise SourceLine(obj, "stdin", UnsupportedRequirement).makeError("Stdin redirection currently not suppported with --api=containers")
+ if obj.get("stderr"):
+ raise SourceLine(obj, "stderr", UnsupportedRequirement).makeError("Stderr redirection currently not suppported with --api=containers")
for v in obj.itervalues():
- self.check_writable(v)
- if isinstance(obj, list):
- for v in obj:
- self.check_writable(v)
+ self.check_features(v)
+ elif isinstance(obj, list):
+ for i,v in enumerate(obj):
+ with SourceLine(obj, i, UnsupportedRequirement):
+ self.check_features(v)
def make_output_collection(self, name, tagsString, outputObj):
outputObj = copy.deepcopy(outputObj)
def arv_executor(self, tool, job_order, **kwargs):
self.debug = kwargs.get("debug")
- tool.visit(self.check_writable)
+ tool.visit(self.check_features)
self.project_uuid = kwargs.get("project_uuid")
self.pipeline = None
name=kwargs.get("name"))
tmpl.save()
# cwltool.main will write our return value to stdout.
- return tmpl.uuid
+ return (tmpl.uuid, "success")
else:
- return upload_workflow(self, tool, job_order,
+ return (upload_workflow(self, tool, job_order,
self.project_uuid,
uuid=existing_uuid,
submit_runner_ram=kwargs.get("submit_runner_ram"),
- name=kwargs.get("name"))
+ name=kwargs.get("name")), "success")
self.ignore_docker_for_reuse = kwargs.get("ignore_docker_for_reuse")
kwargs["enable_reuse"] = kwargs.get("enable_reuse")
kwargs["use_container"] = True
kwargs["tmpdir_prefix"] = "tmp"
- kwargs["on_error"] = "continue"
kwargs["compute_checksum"] = kwargs.get("compute_checksum")
if not kwargs["name"]:
else:
runnerjob = RunnerContainer(self, tool, job_order, kwargs.get("enable_reuse"), self.output_name,
self.output_tags, submit_runner_ram=kwargs.get("submit_runner_ram"),
- name=kwargs.get("name"))
+ name=kwargs.get("name"), on_error=kwargs.get("on_error"))
else:
runnerjob = RunnerJob(self, tool, job_order, kwargs.get("enable_reuse"), self.output_name,
self.output_tags, submit_runner_ram=kwargs.get("submit_runner_ram"),
- name=kwargs.get("name"))
+ name=kwargs.get("name"), on_error=kwargs.get("on_error"))
if not kwargs.get("submit") and "cwl_runner_job" not in kwargs and not self.work_api == "containers":
# Create pipeline for local run
if runnerjob and not kwargs.get("wait"):
runnerjob.run(wait=kwargs.get("wait"))
- return runnerjob.uuid
+ return (runnerjob.uuid, "success")
self.poll_api = arvados.api('v1')
self.polling_thread = threading.Thread(target=self.poll_states)
self.final_output, self.final_output_collection = self.make_output_collection(self.output_name, self.output_tags, self.final_output)
self.set_crunch_output()
- if self.final_status != "success":
- raise WorkflowException("Workflow failed.")
-
if kwargs.get("compute_checksum"):
adjustDirObjs(self.final_output, partial(getListing, self.fs_access))
adjustFileObjs(self.final_output, partial(compute_checksums, self.fs_access))
- return self.final_output
+ return (self.final_output, self.final_status)
def versionstring():
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("--log-timestamps", action="store_true", help="Prefix logging lines with timestamp",
+ default=True, dest="log_timestamps")
+ exgroup.add_argument("--no-log-timestamps", action="store_false", help="No timestamp on logging lines",
+ default=True, dest="log_timestamps")
+
parser.add_argument("--api", type=str,
default=None, dest="work_api",
help="Select work submission API, one of 'jobs' or 'containers'. Default is 'jobs' if that API is available, otherwise 'containers'.")
help="Name to use for workflow execution instance.",
default=None)
+ parser.add_argument("--on-error", type=str,
+ help="Desired workflow behavior when a step fails. One of 'stop' or 'continue'. "
+ "Default is 'continue'.", default="continue", choices=("stop", "continue"))
+
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.")
metrics.setLevel(logging.DEBUG)
logging.getLogger("cwltool.metrics").setLevel(logging.DEBUG)
+ if arvargs.log_timestamps:
+ arvados.log_handler.setFormatter(logging.Formatter(
+ '%(asctime)s %(name)s %(levelname)s: %(message)s',
+ '%Y-%m-%d %H:%M:%S'))
+ else:
+ arvados.log_handler.setFormatter(logging.Formatter('%(name)s %(levelname)s: %(message)s'))
+
arvargs.conformance_test = None
arvargs.use_container = True
arvargs.relax_path_checks = True
+ arvargs.validate = None
return cwltool.main.main(args=arvargs,
stdout=stdout,
fetcher_constructor=partial(CollectionFetcher,
api_client=api_client,
keep_client=keep_client),
- resolver=partial(collectionResolver, api_client))
+ resolver=partial(collectionResolver, api_client),
+ logger_handler=arvados.log_handler)