sdk/cwl/arvados_cwl/__init__.py

   1 #!/usr/bin/env python
   2
   3 # Implement cwl-runner interface for submitting and running work on Arvados, using
   4 # either the Crunch jobs API or Crunch containers API.
   5
   6 import argparse
   7 import logging
   8 import os
   9 import sys
  10 import threading
  11 import hashlib
  12 import copy
  13 import json
  14 import re
  15 from functools import partial
  16 import pkg_resources  # part of setuptools
  17
  18 from cwltool.errors import WorkflowException
  19 import cwltool.main
  20 import cwltool.workflow
  21 import cwltool.process
  22 import schema_salad
  23 from schema_salad.sourceline import SourceLine
  24
  25 import arvados
  26 import arvados.config
  27 from arvados.keep import KeepClient
  28 from arvados.errors import ApiError
  29
  30 from .arvcontainer import ArvadosContainer, RunnerContainer
  31 from .arvjob import ArvadosJob, RunnerJob, RunnerTemplate
  32 from. runner import Runner, upload_docker, upload_job_order, upload_workflow_deps, upload_dependencies
  33 from .arvtool import ArvadosCommandTool
  34 from .arvworkflow import ArvadosWorkflow, upload_workflow
  35 from .fsaccess import CollectionFsAccess, CollectionFetcher, collectionResolver
  36 from .perf import Perf
  37 from .pathmapper import NoFollowPathMapper
  38 from ._version import __version__
  39
  40 from cwltool.pack import pack
  41 from cwltool.process import shortname, UnsupportedRequirement, getListing
  42 from cwltool.pathmapper import adjustFileObjs, adjustDirObjs
  43 from cwltool.draft2tool import compute_checksums
  44 from arvados.api import OrderedJsonModel
  45
  46 logger = logging.getLogger('arvados.cwl-runner')
  47 metrics = logging.getLogger('arvados.cwl-runner.metrics')
  48 logger.setLevel(logging.INFO)
  49
  50 arvados.log_handler.setFormatter(logging.Formatter(
  51         '%(asctime)s %(name)s %(levelname)s: %(message)s',
  52         '%Y-%m-%d %H:%M:%S'))
  53
  54 class ArvCwlRunner(object):
  55     """Execute a CWL tool or workflow, submit work (using either jobs or
  56     containers API), wait for them to complete, and report output.
  57
  58     """
  59
  60     def __init__(self, api_client, work_api=None, keep_client=None, output_name=None, output_tags=None, num_retries=4):
  61         self.api = api_client
  62         self.processes = {}
  63         self.lock = threading.Lock()
  64         self.cond = threading.Condition(self.lock)
  65         self.final_output = None
  66         self.final_status = None
  67         self.uploaded = {}
  68         self.num_retries = num_retries
  69         self.uuid = None
  70         self.stop_polling = threading.Event()
  71         self.poll_api = None
  72         self.pipeline = None
  73         self.final_output_collection = None
  74         self.output_name = output_name
  75         self.output_tags = output_tags
  76         self.project_uuid = None
  77
  78         if keep_client is not None:
  79             self.keep_client = keep_client
  80         else:
  81             self.keep_client = arvados.keep.KeepClient(api_client=self.api, num_retries=self.num_retries)
  82
  83         self.work_api = None
  84         expected_api = ["jobs", "containers"]
  85         for api in expected_api:
  86             try:
  87                 methods = self.api._rootDesc.get('resources')[api]['methods']
  88                 if ('httpMethod' in methods['create'] and
  89                     (work_api == api or work_api is None)):
  90                     self.work_api = api
  91                     break
  92             except KeyError:
  93                 pass
  94
  95         if not self.work_api:
  96             if work_api is None:
  97                 raise Exception("No supported APIs")
  98             else:
  99                 raise Exception("Unsupported API '%s', expected one of %s" % (work_api, expected_api))
 100
 101     def arv_make_tool(self, toolpath_object, **kwargs):
 102         kwargs["work_api"] = self.work_api
 103         kwargs["fetcher_constructor"] = partial(CollectionFetcher,
 104                                                 api_client=self.api,
 105                                                 keep_client=self.keep_client)
 106         if "class" in toolpath_object and toolpath_object["class"] == "CommandLineTool":
 107             return ArvadosCommandTool(self, toolpath_object, **kwargs)
 108         elif "class" in toolpath_object and toolpath_object["class"] == "Workflow":
 109             return ArvadosWorkflow(self, toolpath_object, **kwargs)
 110         else:
 111             return cwltool.workflow.defaultMakeTool(toolpath_object, **kwargs)
 112
 113     def output_callback(self, out, processStatus):
 114         if processStatus == "success":
 115             logger.info("Overall process status is %s", processStatus)
 116             if self.pipeline:
 117                 self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
 118                                                      body={"state": "Complete"}).execute(num_retries=self.num_retries)
 119         else:
 120             logger.warn("Overall process status is %s", processStatus)
 121             if self.pipeline:
 122                 self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
 123                                                      body={"state": "Failed"}).execute(num_retries=self.num_retries)
 124         self.final_status = processStatus
 125         self.final_output = out
 126
 127     def on_message(self, event):
 128         if "object_uuid" in event:
 129             if event["object_uuid"] in self.processes and event["event_type"] == "update":
 130                 if event["properties"]["new_attributes"]["state"] == "Running" and self.processes[event["object_uuid"]].running is False:
 131                     uuid = event["object_uuid"]
 132                     with self.lock:
 133                         j = self.processes[uuid]
 134                         logger.info("%s %s is Running", self.label(j), uuid)
 135                         j.running = True
 136                         j.update_pipeline_component(event["properties"]["new_attributes"])
 137                 elif event["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled", "Final"):
 138                     uuid = event["object_uuid"]
 139                     try:
 140                         self.cond.acquire()
 141                         j = self.processes[uuid]
 142                         logger.info("%s %s is %s", self.label(j), uuid, event["properties"]["new_attributes"]["state"])
 143                         with Perf(metrics, "done %s" % j.name):
 144                             j.done(event["properties"]["new_attributes"])
 145                         self.cond.notify()
 146                     finally:
 147                         self.cond.release()
 148
 149     def label(self, obj):
 150         return "[%s %s]" % (self.work_api[0:-1], obj.name)
 151
 152     def poll_states(self):
 153         """Poll status of jobs or containers listed in the processes dict.
 154
 155         Runs in a separate thread.
 156         """
 157
 158         try:
 159             while True:
 160                 self.stop_polling.wait(15)
 161                 if self.stop_polling.is_set():
 162                     break
 163                 with self.lock:
 164                     keys = self.processes.keys()
 165                 if not keys:
 166                     continue
 167
 168                 if self.work_api == "containers":
 169                     table = self.poll_api.container_requests()
 170                 elif self.work_api == "jobs":
 171                     table = self.poll_api.jobs()
 172
 173                 try:
 174                     proc_states = table.list(filters=[["uuid", "in", keys]]).execute(num_retries=self.num_retries)
 175                 except Exception as e:
 176                     logger.warn("Error checking states on API server: %s", e)
 177                     continue
 178
 179                 for p in proc_states["items"]:
 180                     self.on_message({
 181                         "object_uuid": p["uuid"],
 182                         "event_type": "update",
 183                         "properties": {
 184                             "new_attributes": p
 185                         }
 186                     })
 187         except:
 188             logger.error("Fatal error in state polling thread.", exc_info=(sys.exc_info()[1] if self.debug else False))
 189             self.cond.acquire()
 190             self.processes.clear()
 191             self.cond.notify()
 192             self.cond.release()
 193         finally:
 194             self.stop_polling.set()
 195
 196     def get_uploaded(self):
 197         return self.uploaded.copy()
 198
 199     def add_uploaded(self, src, pair):
 200         self.uploaded[src] = pair
 201
 202     def check_features(self, obj):
 203         if isinstance(obj, dict):
 204             if obj.get("writable"):
 205                 raise SourceLine(obj, "writable", UnsupportedRequirement).makeError("InitialWorkDir feature 'writable: true' not supported")
 206             if obj.get("class") == "DockerRequirement":
 207                 if obj.get("dockerOutputDirectory"):
 208                     # TODO: can be supported by containers API, but not jobs API.
 209                     raise SourceLine(obj, "dockerOutputDirectory", UnsupportedRequirement).makeError(
 210                         "Option 'dockerOutputDirectory' of DockerRequirement not supported.")
 211             for v in obj.itervalues():
 212                 self.check_features(v)
 213         elif isinstance(obj, list):
 214             for i,v in enumerate(obj):
 215                 with SourceLine(obj, i, UnsupportedRequirement):
 216                     self.check_features(v)
 217
 218     def make_output_collection(self, name, tagsString, outputObj):
 219         outputObj = copy.deepcopy(outputObj)
 220
 221         files = []
 222         def capture(fileobj):
 223             files.append(fileobj)
 224
 225         adjustDirObjs(outputObj, capture)
 226         adjustFileObjs(outputObj, capture)
 227
 228         generatemapper = NoFollowPathMapper(files, "", "", separateDirs=False)
 229
 230         final = arvados.collection.Collection(api_client=self.api,
 231                                               keep_client=self.keep_client,
 232                                               num_retries=self.num_retries)
 233
 234         srccollections = {}
 235         for k,v in generatemapper.items():
 236             if k.startswith("_:"):
 237                 if v.type == "Directory":
 238                     continue
 239                 if v.type == "CreateFile":
 240                     with final.open(v.target, "wb") as f:
 241                         f.write(v.resolved.encode("utf-8"))
 242                     continue
 243
 244             if not k.startswith("keep:"):
 245                 raise Exception("Output source is not in keep or a literal")
 246             sp = k.split("/")
 247             srccollection = sp[0][5:]
 248             if srccollection not in srccollections:
 249                 try:
 250                     srccollections[srccollection] = arvados.collection.CollectionReader(
 251                         srccollection,
 252                         api_client=self.api,
 253                         keep_client=self.keep_client,
 254                         num_retries=self.num_retries)
 255                 except arvados.errors.ArgumentError as e:
 256                     logger.error("Creating CollectionReader for '%s' '%s': %s", k, v, e)
 257                     raise
 258             reader = srccollections[srccollection]
 259             try:
 260                 srcpath = "/".join(sp[1:]) if len(sp) > 1 else "."
 261                 final.copy(srcpath, v.target, source_collection=reader, overwrite=False)
 262             except IOError as e:
 263                 logger.warn("While preparing output collection: %s", e)
 264
 265         def rewrite(fileobj):
 266             fileobj["location"] = generatemapper.mapper(fileobj["location"]).target
 267             for k in ("basename", "listing", "contents"):
 268                 if k in fileobj:
 269                     del fileobj[k]
 270
 271         adjustDirObjs(outputObj, rewrite)
 272         adjustFileObjs(outputObj, rewrite)
 273
 274         with final.open("cwl.output.json", "w") as f:
 275             json.dump(outputObj, f, sort_keys=True, indent=4, separators=(',',': '))
 276
 277         final.save_new(name=name, owner_uuid=self.project_uuid, ensure_unique_name=True)
 278
 279         logger.info("Final output collection %s \"%s\" (%s)", final.portable_data_hash(),
 280                     final.api_response()["name"],
 281                     final.manifest_locator())
 282
 283         final_uuid = final.manifest_locator()
 284         tags = tagsString.split(',')
 285         for tag in tags:
 286              self.api.links().create(body={
 287                 "head_uuid": final_uuid, "link_class": "tag", "name": tag
 288                 }).execute(num_retries=self.num_retries)
 289
 290         def finalcollection(fileobj):
 291             fileobj["location"] = "keep:%s/%s" % (final.portable_data_hash(), fileobj["location"])
 292
 293         adjustDirObjs(outputObj, finalcollection)
 294         adjustFileObjs(outputObj, finalcollection)
 295
 296         return (outputObj, final)
 297
 298     def set_crunch_output(self):
 299         if self.work_api == "containers":
 300             try:
 301                 current = self.api.containers().current().execute(num_retries=self.num_retries)
 302             except ApiError as e:
 303                 # Status code 404 just means we're not running in a container.
 304                 if e.resp.status != 404:
 305                     logger.info("Getting current container: %s", e)
 306                 return
 307             try:
 308                 self.api.containers().update(uuid=current['uuid'],
 309                                              body={
 310                                                  'output': self.final_output_collection.portable_data_hash(),
 311                                              }).execute(num_retries=self.num_retries)
 312                 self.api.collections().update(uuid=self.final_output_collection.manifest_locator(),
 313                                               body={
 314                                                   'is_trashed': True
 315                                               }).execute(num_retries=self.num_retries)
 316             except Exception as e:
 317                 logger.info("Setting container output: %s", e)
 318         elif self.work_api == "jobs" and "TASK_UUID" in os.environ:
 319             self.api.job_tasks().update(uuid=os.environ["TASK_UUID"],
 320                                    body={
 321                                        'output': self.final_output_collection.portable_data_hash(),
 322                                        'success': self.final_status == "success",
 323                                        'progress':1.0
 324                                    }).execute(num_retries=self.num_retries)
 325
 326     def arv_executor(self, tool, job_order, **kwargs):
 327         self.debug = kwargs.get("debug")
 328
 329         tool.visit(self.check_features)
 330
 331         self.project_uuid = kwargs.get("project_uuid")
 332         self.pipeline = None
 333         make_fs_access = kwargs.get("make_fs_access") or partial(CollectionFsAccess,
 334                                                                  api_client=self.api,
 335                                                                  keep_client=self.keep_client)
 336         self.fs_access = make_fs_access(kwargs["basedir"])
 337
 338         if not kwargs.get("name"):
 339             kwargs["name"] = self.name = tool.tool.get("label") or tool.metadata.get("label") or os.path.basename(tool.tool["id"])
 340
 341         # Upload direct dependencies of workflow steps, get back mapping of files to keep references.
 342         # Also uploads docker images.
 343         upload_workflow_deps(self, tool)
 344
 345         # Reload tool object which may have been updated by
 346         # upload_workflow_deps
 347         tool = self.arv_make_tool(tool.doc_loader.idx[tool.tool["id"]],
 348                                   makeTool=self.arv_make_tool,
 349                                   loader=tool.doc_loader,
 350                                   avsc_names=tool.doc_schema,
 351                                   metadata=tool.metadata)
 352
 353         # Upload local file references in the job order.
 354         job_order = upload_job_order(self, "%s input" % kwargs["name"],
 355                                      tool, job_order)
 356
 357         existing_uuid = kwargs.get("update_workflow")
 358         if existing_uuid or kwargs.get("create_workflow"):
 359             # Create a pipeline template or workflow record and exit.
 360             if self.work_api == "jobs":
 361                 tmpl = RunnerTemplate(self, tool, job_order,
 362                                       kwargs.get("enable_reuse"),
 363                                       uuid=existing_uuid,
 364                                       submit_runner_ram=kwargs.get("submit_runner_ram"),
 365                                       name=kwargs["name"])
 366                 tmpl.save()
 367                 # cwltool.main will write our return value to stdout.
 368                 return (tmpl.uuid, "success")
 369             elif self.work_api == "containers":
 370                 return (upload_workflow(self, tool, job_order,
 371                                         self.project_uuid,
 372                                         uuid=existing_uuid,
 373                                         submit_runner_ram=kwargs.get("submit_runner_ram"),
 374                                         name=kwargs["name"]),
 375                         "success")
 376
 377         self.ignore_docker_for_reuse = kwargs.get("ignore_docker_for_reuse")
 378
 379         kwargs["make_fs_access"] = make_fs_access
 380         kwargs["enable_reuse"] = kwargs.get("enable_reuse")
 381         kwargs["use_container"] = True
 382         kwargs["tmpdir_prefix"] = "tmp"
 383         kwargs["compute_checksum"] = kwargs.get("compute_checksum")
 384
 385         if self.work_api == "containers":
 386             kwargs["outdir"] = "/var/spool/cwl"
 387             kwargs["docker_outdir"] = "/var/spool/cwl"
 388             kwargs["tmpdir"] = "/tmp"
 389             kwargs["docker_tmpdir"] = "/tmp"
 390         elif self.work_api == "jobs":
 391             kwargs["outdir"] = "$(task.outdir)"
 392             kwargs["docker_outdir"] = "$(task.outdir)"
 393             kwargs["tmpdir"] = "$(task.tmpdir)"
 394
 395         runnerjob = None
 396         if kwargs.get("submit"):
 397             # Submit a runner job to run the workflow for us.
 398             if self.work_api == "containers":
 399                 if tool.tool["class"] == "CommandLineTool":
 400                     kwargs["runnerjob"] = tool.tool["id"]
 401                     upload_dependencies(self,
 402                                         kwargs["name"],
 403                                         tool.doc_loader,
 404                                         tool.tool,
 405                                         tool.tool["id"],
 406                                         False)
 407                     runnerjob = tool.job(job_order,
 408                                          self.output_callback,
 409                                          **kwargs).next()
 410                 else:
 411                     runnerjob = RunnerContainer(self, tool, job_order, kwargs.get("enable_reuse"),
 412                                                 self.output_name,
 413                                                 self.output_tags,
 414                                                 submit_runner_ram=kwargs.get("submit_runner_ram"),
 415                                                 name=kwargs.get("name"),
 416                                                 on_error=kwargs.get("on_error"),
 417                                                 submit_runner_image=kwargs.get("submit_runner_image"))
 418             elif self.work_api == "jobs":
 419                 runnerjob = RunnerJob(self, tool, job_order, kwargs.get("enable_reuse"),
 420                                       self.output_name,
 421                                       self.output_tags,
 422                                       submit_runner_ram=kwargs.get("submit_runner_ram"),
 423                                       name=kwargs.get("name"),
 424                                       on_error=kwargs.get("on_error"),
 425                                       submit_runner_image=kwargs.get("submit_runner_image"))
 426
 427         if not kwargs.get("submit") and "cwl_runner_job" not in kwargs and self.work_api == "jobs":
 428             # Create pipeline for local run
 429             self.pipeline = self.api.pipeline_instances().create(
 430                 body={
 431                     "owner_uuid": self.project_uuid,
 432                     "name": kwargs["name"] if kwargs.get("name") else shortname(tool.tool["id"]),
 433                     "components": {},
 434                     "state": "RunningOnClient"}).execute(num_retries=self.num_retries)
 435             logger.info("Pipeline instance %s", self.pipeline["uuid"])
 436
 437         if runnerjob and not kwargs.get("wait"):
 438             runnerjob.run(wait=kwargs.get("wait"))
 439             return (runnerjob.uuid, "success")
 440
 441         self.poll_api = arvados.api('v1')
 442         self.polling_thread = threading.Thread(target=self.poll_states)
 443         self.polling_thread.start()
 444
 445         if runnerjob:
 446             jobiter = iter((runnerjob,))
 447         else:
 448             if "cwl_runner_job" in kwargs:
 449                 self.uuid = kwargs.get("cwl_runner_job").get('uuid')
 450             jobiter = tool.job(job_order,
 451                                self.output_callback,
 452                                **kwargs)
 453
 454         try:
 455             self.cond.acquire()
 456             # Will continue to hold the lock for the duration of this code
 457             # except when in cond.wait(), at which point on_message can update
 458             # job state and process output callbacks.
 459
 460             loopperf = Perf(metrics, "jobiter")
 461             loopperf.__enter__()
 462             for runnable in jobiter:
 463                 loopperf.__exit__()
 464
 465                 if self.stop_polling.is_set():
 466                     break
 467
 468                 if runnable:
 469                     with Perf(metrics, "run"):
 470                         runnable.run(**kwargs)
 471                 else:
 472                     if self.processes:
 473                         self.cond.wait(1)
 474                     else:
 475                         logger.error("Workflow is deadlocked, no runnable jobs and not waiting on any pending jobs.")
 476                         break
 477                 loopperf.__enter__()
 478             loopperf.__exit__()
 479
 480             while self.processes:
 481                 self.cond.wait(1)
 482
 483         except UnsupportedRequirement:
 484             raise
 485         except:
 486             if sys.exc_info()[0] is KeyboardInterrupt:
 487                 logger.error("Interrupted, marking pipeline as failed")
 488             else:
 489                 logger.error("Execution failed: %s", sys.exc_info()[1], exc_info=(sys.exc_info()[1] if self.debug else False))
 490             if self.pipeline:
 491                 self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
 492                                                      body={"state": "Failed"}).execute(num_retries=self.num_retries)
 493             if runnerjob and runnerjob.uuid and self.work_api == "containers":
 494                 self.api.container_requests().update(uuid=runnerjob.uuid,
 495                                                      body={"priority": "0"}).execute(num_retries=self.num_retries)
 496         finally:
 497             self.cond.release()
 498             self.stop_polling.set()
 499             self.polling_thread.join()
 500
 501         if self.final_status == "UnsupportedRequirement":
 502             raise UnsupportedRequirement("Check log for details.")
 503
 504         if self.final_output is None:
 505             raise WorkflowException("Workflow did not return a result.")
 506
 507         if kwargs.get("submit") and isinstance(runnerjob, Runner):
 508             logger.info("Final output collection %s", runnerjob.final_output)
 509         else:
 510             if self.output_name is None:
 511                 self.output_name = "Output of %s" % (shortname(tool.tool["id"]))
 512             if self.output_tags is None:
 513                 self.output_tags = ""
 514             self.final_output, self.final_output_collection = self.make_output_collection(self.output_name, self.output_tags, self.final_output)
 515             self.set_crunch_output()
 516
 517         if kwargs.get("compute_checksum"):
 518             adjustDirObjs(self.final_output, partial(getListing, self.fs_access))
 519             adjustFileObjs(self.final_output, partial(compute_checksums, self.fs_access))
 520
 521         return (self.final_output, self.final_status)
 522
 523
 524 def versionstring():
 525     """Print version string of key packages for provenance and debugging."""
 526
 527     arvcwlpkg = pkg_resources.require("arvados-cwl-runner")
 528     arvpkg = pkg_resources.require("arvados-python-client")
 529     cwlpkg = pkg_resources.require("cwltool")
 530
 531     return "%s %s %s, %s %s, %s %s" % (sys.argv[0], __version__, arvcwlpkg[0].version,
 532                                     "arvados-python-client", arvpkg[0].version,
 533                                     "cwltool", cwlpkg[0].version)
 534
 535
 536 def arg_parser():  # type: () -> argparse.ArgumentParser
 537     parser = argparse.ArgumentParser(description='Arvados executor for Common Workflow Language')
 538
 539     parser.add_argument("--basedir", type=str,
 540                         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).")
 541     parser.add_argument("--outdir", type=str, default=os.path.abspath('.'),
 542                         help="Output directory, default current directory")
 543
 544     parser.add_argument("--eval-timeout",
 545                         help="Time to wait for a Javascript expression to evaluate before giving an error, default 20s.",
 546                         type=float,
 547                         default=20)
 548     parser.add_argument("--version", action="store_true", help="Print version and exit")
 549
 550     exgroup = parser.add_mutually_exclusive_group()
 551     exgroup.add_argument("--verbose", action="store_true", help="Default logging")
 552     exgroup.add_argument("--quiet", action="store_true", help="Only print warnings and errors.")
 553     exgroup.add_argument("--debug", action="store_true", help="Print even more logging")
 554
 555     parser.add_argument("--metrics", action="store_true", help="Print timing metrics")
 556
 557     parser.add_argument("--tool-help", action="store_true", help="Print command line help for tool")
 558
 559     exgroup = parser.add_mutually_exclusive_group()
 560     exgroup.add_argument("--enable-reuse", action="store_true",
 561                         default=True, dest="enable_reuse",
 562                         help="")
 563     exgroup.add_argument("--disable-reuse", action="store_false",
 564                         default=True, dest="enable_reuse",
 565                         help="")
 566
 567     parser.add_argument("--project-uuid", type=str, metavar="UUID", help="Project that will own the workflow jobs, if not provided, will go to home project.")
 568     parser.add_argument("--output-name", type=str, help="Name to use for collection that stores the final output.", default=None)
 569     parser.add_argument("--output-tags", type=str, help="Tags for the final output collection separated by commas, e.g., '--output-tags tag0,tag1,tag2'.", default=None)
 570     parser.add_argument("--ignore-docker-for-reuse", action="store_true",
 571                         help="Ignore Docker image version when deciding whether to reuse past jobs.",
 572                         default=False)
 573
 574     exgroup = parser.add_mutually_exclusive_group()
 575     exgroup.add_argument("--submit", action="store_true", help="Submit workflow to run on Arvados.",
 576                         default=True, dest="submit")
 577     exgroup.add_argument("--local", action="store_false", help="Run workflow on local host (submits jobs to Arvados).",
 578                         default=True, dest="submit")
 579     exgroup.add_argument("--create-template", action="store_true", help="(Deprecated) synonym for --create-workflow.",
 580                          dest="create_workflow")
 581     exgroup.add_argument("--create-workflow", action="store_true", help="Create an Arvados workflow (if using the 'containers' API) or pipeline template (if using the 'jobs' API). See --api.")
 582     exgroup.add_argument("--update-workflow", type=str, metavar="UUID", help="Update an existing Arvados workflow or pipeline template with the given UUID.")
 583
 584     exgroup = parser.add_mutually_exclusive_group()
 585     exgroup.add_argument("--wait", action="store_true", help="After submitting workflow runner job, wait for completion.",
 586                         default=True, dest="wait")
 587     exgroup.add_argument("--no-wait", action="store_false", help="Submit workflow runner job and exit.",
 588                         default=True, dest="wait")
 589
 590     exgroup = parser.add_mutually_exclusive_group()
 591     exgroup.add_argument("--log-timestamps", action="store_true", help="Prefix logging lines with timestamp",
 592                         default=True, dest="log_timestamps")
 593     exgroup.add_argument("--no-log-timestamps", action="store_false", help="No timestamp on logging lines",
 594                         default=True, dest="log_timestamps")
 595
 596     parser.add_argument("--api", type=str,
 597                         default=None, dest="work_api",
 598                         help="Select work submission API, one of 'jobs' or 'containers'. Default is 'jobs' if that API is available, otherwise 'containers'.")
 599
 600     parser.add_argument("--compute-checksum", action="store_true", default=False,
 601                         help="Compute checksum of contents while collecting outputs",
 602                         dest="compute_checksum")
 603
 604     parser.add_argument("--submit-runner-ram", type=int,
 605                         help="RAM (in MiB) required for the workflow runner job (default 1024)",
 606                         default=1024)
 607
 608     parser.add_argument("--submit-runner-image", type=str,
 609                         help="Docker image for workflow runner job, default arvados/jobs:%s" % __version__,
 610                         default=None)
 611
 612     parser.add_argument("--name", type=str,
 613                         help="Name to use for workflow execution instance.",
 614                         default=None)
 615
 616     parser.add_argument("--on-error", type=str,
 617                         help="Desired workflow behavior when a step fails.  One of 'stop' or 'continue'. "
 618                         "Default is 'continue'.", default="continue", choices=("stop", "continue"))
 619
 620     parser.add_argument("workflow", type=str, nargs="?", default=None, help="The workflow to execute")
 621     parser.add_argument("job_order", nargs=argparse.REMAINDER, help="The input object to the workflow.")
 622
 623     return parser
 624
 625 def add_arv_hints():
 626     cache = {}
 627     cwltool.draft2tool.ACCEPTLIST_RE = re.compile(r".*")
 628     cwltool.draft2tool.ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")
 629     res = pkg_resources.resource_stream(__name__, 'arv-cwl-schema.yml')
 630     cache["http://arvados.org/cwl"] = res.read()
 631     res.close()
 632     document_loader, cwlnames, _, _ = cwltool.process.get_schema("v1.0")
 633     _, extnames, _, _ = schema_salad.schema.load_schema("http://arvados.org/cwl", cache=cache)
 634     for n in extnames.names:
 635         if not cwlnames.has_name("http://arvados.org/cwl#"+n, ""):
 636             cwlnames.add_name("http://arvados.org/cwl#"+n, "", extnames.get_name(n, ""))
 637         document_loader.idx["http://arvados.org/cwl#"+n] = {}
 638
 639 def main(args, stdout, stderr, api_client=None, keep_client=None):
 640     parser = arg_parser()
 641
 642     job_order_object = None
 643     arvargs = parser.parse_args(args)
 644
 645     if arvargs.version:
 646         print versionstring()
 647         return
 648
 649     if arvargs.update_workflow:
 650         if arvargs.update_workflow.find('-7fd4e-') == 5:
 651             want_api = 'containers'
 652         elif arvargs.update_workflow.find('-p5p6p-') == 5:
 653             want_api = 'jobs'
 654         else:
 655             want_api = None
 656         if want_api and arvargs.work_api and want_api != arvargs.work_api:
 657             logger.error('--update-workflow arg {!r} uses {!r} API, but --api={!r} specified'.format(
 658                 arvargs.update_workflow, want_api, arvargs.work_api))
 659             return 1
 660         arvargs.work_api = want_api
 661
 662     if (arvargs.create_workflow or arvargs.update_workflow) and not arvargs.job_order:
 663         job_order_object = ({}, "")
 664
 665     add_arv_hints()
 666
 667     try:
 668         if api_client is None:
 669             api_client=arvados.api('v1', model=OrderedJsonModel())
 670         if keep_client is None:
 671             keep_client = arvados.keep.KeepClient(api_client=api_client, num_retries=4)
 672         runner = ArvCwlRunner(api_client, work_api=arvargs.work_api, keep_client=keep_client,
 673                               num_retries=4, output_name=arvargs.output_name,
 674                               output_tags=arvargs.output_tags)
 675     except Exception as e:
 676         logger.error(e)
 677         return 1
 678
 679     if arvargs.debug:
 680         logger.setLevel(logging.DEBUG)
 681         logging.getLogger('arvados').setLevel(logging.DEBUG)
 682
 683     if arvargs.quiet:
 684         logger.setLevel(logging.WARN)
 685         logging.getLogger('arvados').setLevel(logging.WARN)
 686         logging.getLogger('arvados.arv-run').setLevel(logging.WARN)
 687
 688     if arvargs.metrics:
 689         metrics.setLevel(logging.DEBUG)
 690         logging.getLogger("cwltool.metrics").setLevel(logging.DEBUG)
 691
 692     if arvargs.log_timestamps:
 693         arvados.log_handler.setFormatter(logging.Formatter(
 694             '%(asctime)s %(name)s %(levelname)s: %(message)s',
 695             '%Y-%m-%d %H:%M:%S'))
 696     else:
 697         arvados.log_handler.setFormatter(logging.Formatter('%(name)s %(levelname)s: %(message)s'))
 698
 699     arvargs.conformance_test = None
 700     arvargs.use_container = True
 701     arvargs.relax_path_checks = True
 702     arvargs.validate = None
 703
 704     return cwltool.main.main(args=arvargs,
 705                              stdout=stdout,
 706                              stderr=stderr,
 707                              executor=runner.arv_executor,
 708                              makeTool=runner.arv_make_tool,
 709                              versionfunc=versionstring,
 710                              job_order_object=job_order_object,
 711                              make_fs_access=partial(CollectionFsAccess,
 712                                                     api_client=api_client,
 713                                                     keep_client=keep_client),
 714                              fetcher_constructor=partial(CollectionFetcher,
 715                                                          api_client=api_client,
 716                                                          keep_client=keep_client,
 717                                                          num_retries=runner.num_retries),
 718                              resolver=partial(collectionResolver, api_client, num_retries=runner.num_retries),
 719                              logger_handler=arvados.log_handler)