# Implement cwl-runner interface for submitting and running work on Arvados, using
# the Crunch containers API.
-from future.utils import viewitems
-from builtins import str
-
import argparse
+import importlib.metadata
+import importlib.resources
import logging
import os
import sys
import re
-import pkg_resources # part of setuptools
from schema_salad.sourceline import SourceLine
import schema_salad.validate as validate
import arvados
import arvados.config
+import arvados.logging
from arvados.keep import KeepClient
from arvados.errors import ApiError
import arvados.commands._util as arv_cmd
-from arvados.api import OrderedJsonModel
from .perf import Perf
from ._version import __version__
def versionstring():
"""Print version string of key packages for provenance and debugging."""
-
- arvcwlpkg = pkg_resources.require("arvados-cwl-runner")
- arvpkg = pkg_resources.require("arvados-python-client")
- cwlpkg = pkg_resources.require("cwltool")
-
- return "%s %s, %s %s, %s %s" % (sys.argv[0], arvcwlpkg[0].version,
- "arvados-python-client", arvpkg[0].version,
- "cwltool", cwlpkg[0].version)
-
+ return "{} {}, arvados-python-client {}, cwltool {}".format(
+ sys.argv[0],
+ importlib.metadata.version('arvados-cwl-runner'),
+ importlib.metadata.version('arvados-python-client'),
+ importlib.metadata.version('cwltool'),
+ )
def arg_parser(): # type: () -> argparse.ArgumentParser
parser = argparse.ArgumentParser(
exgroup.add_argument("--create-workflow", action="store_true", help="Register an Arvados workflow that can be run from Workbench")
exgroup.add_argument("--update-workflow", metavar="UUID", help="Update an existing Arvados workflow with the given UUID.")
+ exgroup.add_argument("--print-keep-deps", action="store_true", help="To assist copying, print a list of Keep collections that this workflow depends on.")
+
exgroup = parser.add_mutually_exclusive_group()
exgroup.add_argument("--wait", action="store_true", help="After submitting workflow runner, wait for completion.",
default=True, dest="wait")
default=False, dest="trash_intermediate",
help="Do not trash intermediate outputs (default).")
+ exgroup = parser.add_mutually_exclusive_group()
+ exgroup.add_argument("--enable-usage-report", dest="enable_usage_report", default=None, action="store_true", help="Create usage_report.html with a summary of each step's resource usage.")
+ exgroup.add_argument("--disable-usage-report", dest="enable_usage_report", default=None, action="store_false", help="Disable usage report.")
+
parser.add_argument("workflow", default=None, help="The workflow to execute")
parser.add_argument("job_order", nargs=argparse.REMAINDER, help="The input object to the workflow.")
cwltool.command_line_tool.ACCEPTLIST_RE = cwltool.command_line_tool.ACCEPTLIST_EN_RELAXED_RE
supported_versions = ["v1.0", "v1.1", "v1.2"]
for s in supported_versions:
- res = pkg_resources.resource_stream(__name__, 'arv-cwl-schema-%s.yml' % s)
- customschema = res.read().decode('utf-8')
+ customschema = importlib.resources.read_text(__name__, f'arv-cwl-schema-{s}.yml', 'utf-8')
use_custom_schema(s, "http://arvados.org/cwl", customschema)
- res.close()
cwltool.process.supportedProcessRequirements.extend([
"http://arvados.org/cwl#RunInSingleContainer",
"http://arvados.org/cwl#OutputDirType",
return 1
arvargs.work_api = want_api
- if (arvargs.create_workflow or arvargs.update_workflow) and not arvargs.job_order:
+ workflow_op = arvargs.create_workflow or arvargs.update_workflow or arvargs.print_keep_deps
+
+ if workflow_op and not arvargs.job_order:
job_order_object = ({}, "")
add_arv_hints()
- for key, val in viewitems(cwltool.argparser.get_default_args()):
+ for key, val in cwltool.argparser.get_default_args().items():
if not hasattr(arvargs, key):
setattr(arvargs, key, val)
if api_client is None:
api_client = arvados.safeapi.ThreadSafeApiCache(
api_params={
- 'model': OrderedJsonModel(),
'num_retries': arvargs.retries,
'timeout': arvargs.http_timeout,
},
logger.setLevel(logging.INFO)
logging.getLogger('arvados').setLevel(logging.INFO)
logging.getLogger('arvados.keep').setLevel(logging.WARNING)
+ # API retries are filtered to the INFO level and can be noisy, but as long as
+ # they succeed we don't need to see warnings about it.
+ googleapiclient_http_logger = logging.getLogger('googleapiclient.http')
+ googleapiclient_http_logger.addFilter(arvados.logging.GoogleHTTPClientFilter())
+ googleapiclient_http_logger.setLevel(logging.WARNING)
if arvargs.debug:
logger.setLevel(logging.DEBUG)
logging.getLogger('arvados').setLevel(logging.DEBUG)
+ # In debug mode show logs about retries, but we arn't
+ # debugging the google client so we don't need to see
+ # everything.
+ googleapiclient_http_logger.setLevel(logging.NOTSET)
+ logging.getLogger('googleapiclient').setLevel(logging.INFO)
if arvargs.quiet:
logger.setLevel(logging.WARN)
# unit tests.
stdout = None
+ executor.loadingContext.default_docker_image = arvargs.submit_runner_image or "arvados/jobs:"+__version__
+
if arvargs.workflow.startswith("arvwf:") or workflow_uuid_pattern.match(arvargs.workflow) or arvargs.workflow.startswith("keep:"):
executor.loadingContext.do_validate = False
- if arvargs.submit:
+ if arvargs.submit and not workflow_op:
executor.fast_submit = True
return cwltool.main.main(args=arvargs,
custom_schema_callback=add_arv_hints,
loadingContext=executor.loadingContext,
runtimeContext=executor.toplevel_runtimeContext,
- input_required=not (arvargs.create_workflow or arvargs.update_workflow))
+ input_required=not workflow_op)
projects / arvados.git / blobdiff