19126: Merge branch 'main' into 19126-nginx-proxy-settings-change
[arvados.git] / sdk / cwl / arvados_cwl / __init__.py
1 #!/usr/bin/env python3
2 # Copyright (C) The Arvados Authors. All rights reserved.
3 #
4 # SPDX-License-Identifier: Apache-2.0
5
6 # Implement cwl-runner interface for submitting and running work on Arvados, using
7 # the Crunch containers API.
8
9 from future.utils import viewitems
10 from builtins import str
11
12 import argparse
13 import logging
14 import os
15 import sys
16 import re
17 import pkg_resources  # part of setuptools
18
19 from schema_salad.sourceline import SourceLine
20 import schema_salad.validate as validate
21 import cwltool.main
22 import cwltool.workflow
23 import cwltool.process
24 import cwltool.argparser
25 from cwltool.errors import WorkflowException
26 from cwltool.process import shortname, UnsupportedRequirement, use_custom_schema
27 from cwltool.utils import adjustFileObjs, adjustDirObjs, get_listing
28
29 import arvados
30 import arvados.config
31 from arvados.keep import KeepClient
32 from arvados.errors import ApiError
33 import arvados.commands._util as arv_cmd
34 from arvados.api import OrderedJsonModel
35
36 from .perf import Perf
37 from ._version import __version__
38 from .executor import ArvCwlExecutor
39
40 # These aren't used directly in this file but
41 # other code expects to import them from here
42 from .arvcontainer import ArvadosContainer
43 from .arvtool import ArvadosCommandTool
44 from .fsaccess import CollectionFsAccess, CollectionCache, CollectionFetcher
45 from .util import get_current_container
46 from .executor import RuntimeStatusLoggingHandler, DEFAULT_PRIORITY
47 from .arvworkflow import ArvadosWorkflow
48
49 logger = logging.getLogger('arvados.cwl-runner')
50 metrics = logging.getLogger('arvados.cwl-runner.metrics')
51 logger.setLevel(logging.INFO)
52
53 arvados.log_handler.setFormatter(logging.Formatter(
54         '%(asctime)s %(name)s %(levelname)s: %(message)s',
55         '%Y-%m-%d %H:%M:%S'))
56
57 def versionstring():
58     """Print version string of key packages for provenance and debugging."""
59
60     arvcwlpkg = pkg_resources.require("arvados-cwl-runner")
61     arvpkg = pkg_resources.require("arvados-python-client")
62     cwlpkg = pkg_resources.require("cwltool")
63
64     return "%s %s, %s %s, %s %s" % (sys.argv[0], arvcwlpkg[0].version,
65                                     "arvados-python-client", arvpkg[0].version,
66                                     "cwltool", cwlpkg[0].version)
67
68
69 def arg_parser():  # type: () -> argparse.ArgumentParser
70     parser = argparse.ArgumentParser(description='Arvados executor for Common Workflow Language')
71
72     parser.add_argument("--basedir",
73                         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).")
74     parser.add_argument("--outdir", default=os.path.abspath('.'),
75                         help="Output directory, default current directory")
76
77     parser.add_argument("--eval-timeout",
78                         help="Time to wait for a Javascript expression to evaluate before giving an error, default 20s.",
79                         type=float,
80                         default=20)
81
82     exgroup = parser.add_mutually_exclusive_group()
83     exgroup.add_argument("--print-dot", action="store_true",
84                          help="Print workflow visualization in graphviz format and exit")
85     exgroup.add_argument("--version", action="version", help="Print version and exit", version=versionstring())
86     exgroup.add_argument("--validate", action="store_true", help="Validate CWL document only.")
87
88     exgroup = parser.add_mutually_exclusive_group()
89     exgroup.add_argument("--verbose", action="store_true", help="Default logging")
90     exgroup.add_argument("--quiet", action="store_true", help="Only print warnings and errors.")
91     exgroup.add_argument("--debug", action="store_true", help="Print even more logging")
92
93     parser.add_argument("--metrics", action="store_true", help="Print timing metrics")
94
95     parser.add_argument("--tool-help", action="store_true", help="Print command line help for tool")
96
97     exgroup = parser.add_mutually_exclusive_group()
98     exgroup.add_argument("--enable-reuse", action="store_true",
99                         default=True, dest="enable_reuse",
100                         help="Enable container reuse (default)")
101     exgroup.add_argument("--disable-reuse", action="store_false",
102                         default=True, dest="enable_reuse",
103                         help="Disable container reuse")
104
105     parser.add_argument("--project-uuid", metavar="UUID", help="Project that will own the workflow containers, if not provided, will go to home project.")
106     parser.add_argument("--output-name", help="Name to use for collection that stores the final output.", default=None)
107     parser.add_argument("--output-tags", help="Tags for the final output collection separated by commas, e.g., '--output-tags tag0,tag1,tag2'.", default=None)
108     parser.add_argument("--ignore-docker-for-reuse", action="store_true",
109                         help="Ignore Docker image version when deciding whether to reuse past containers.",
110                         default=False)
111
112     exgroup = parser.add_mutually_exclusive_group()
113     exgroup.add_argument("--submit", action="store_true", help="Submit workflow to run on Arvados.",
114                         default=True, dest="submit")
115     exgroup.add_argument("--local", action="store_false", help="Run workflow on local host (submits containers to Arvados).",
116                         default=True, dest="submit")
117     exgroup.add_argument("--create-template", action="store_true", help="(Deprecated) synonym for --create-workflow.",
118                          dest="create_workflow")
119     exgroup.add_argument("--create-workflow", action="store_true", help="Register an Arvados workflow that can be run from Workbench")
120     exgroup.add_argument("--update-workflow", metavar="UUID", help="Update an existing Arvados workflow with the given UUID.")
121
122     exgroup = parser.add_mutually_exclusive_group()
123     exgroup.add_argument("--wait", action="store_true", help="After submitting workflow runner, wait for completion.",
124                         default=True, dest="wait")
125     exgroup.add_argument("--no-wait", action="store_false", help="Submit workflow runner and exit.",
126                         default=True, dest="wait")
127
128     exgroup = parser.add_mutually_exclusive_group()
129     exgroup.add_argument("--log-timestamps", action="store_true", help="Prefix logging lines with timestamp",
130                         default=True, dest="log_timestamps")
131     exgroup.add_argument("--no-log-timestamps", action="store_false", help="No timestamp on logging lines",
132                         default=True, dest="log_timestamps")
133
134     parser.add_argument("--api",
135                         default=None, dest="work_api",
136                         choices=("containers",),
137                         help="Select work submission API.  Only supports 'containers'")
138
139     parser.add_argument("--compute-checksum", action="store_true", default=False,
140                         help="Compute checksum of contents while collecting outputs",
141                         dest="compute_checksum")
142
143     parser.add_argument("--submit-runner-ram", type=int,
144                         help="RAM (in MiB) required for the workflow runner job (default 1024)",
145                         default=None)
146
147     parser.add_argument("--submit-runner-image",
148                         help="Docker image for workflow runner job, default arvados/jobs:%s" % __version__,
149                         default=None)
150
151     parser.add_argument("--always-submit-runner", action="store_true",
152                         help="When invoked with --submit --wait, always submit a runner to manage the workflow, even when only running a single CommandLineTool",
153                         default=False)
154
155     parser.add_argument("--match-submitter-images", action="store_true",
156                         default=False, dest="match_local_docker",
157                         help="Where Arvados has more than one Docker image of the same name, use image from the Docker instance on the submitting node.")
158
159     exgroup = parser.add_mutually_exclusive_group()
160     exgroup.add_argument("--submit-request-uuid",
161                          default=None,
162                          help="Update and commit to supplied container request instead of creating a new one.",
163                          metavar="UUID")
164     exgroup.add_argument("--submit-runner-cluster",
165                          help="Submit workflow runner to a remote cluster",
166                          default=None,
167                          metavar="CLUSTER_ID")
168
169     parser.add_argument("--collection-cache-size", type=int,
170                         default=None,
171                         help="Collection cache size (in MiB, default 256).")
172
173     parser.add_argument("--name",
174                         help="Name to use for workflow execution instance.",
175                         default=None)
176
177     parser.add_argument("--on-error",
178                         help="Desired workflow behavior when a step fails.  One of 'stop' (do not submit any more steps) or "
179                         "'continue' (may submit other steps that are not downstream from the error). Default is 'continue'.",
180                         default="continue", choices=("stop", "continue"))
181
182     parser.add_argument("--enable-dev", action="store_true",
183                         help="Enable loading and running development versions "
184                              "of the CWL standards.", default=False)
185     parser.add_argument('--storage-classes', default="default",
186                         help="Specify comma separated list of storage classes to be used when saving final workflow output to Keep.")
187     parser.add_argument('--intermediate-storage-classes', default="default",
188                         help="Specify comma separated list of storage classes to be used when saving intermediate workflow output to Keep.")
189
190     parser.add_argument("--intermediate-output-ttl", type=int, metavar="N",
191                         help="If N > 0, intermediate output collections will be trashed N seconds after creation.  Default is 0 (don't trash).",
192                         default=0)
193
194     parser.add_argument("--priority", type=int,
195                         help="Workflow priority (range 1..1000, higher has precedence over lower)",
196                         default=DEFAULT_PRIORITY)
197
198     parser.add_argument("--disable-validate", dest="do_validate",
199                         action="store_false", default=True,
200                         help=argparse.SUPPRESS)
201
202     parser.add_argument("--disable-color", dest="enable_color",
203                         action="store_false", default=True,
204                         help=argparse.SUPPRESS)
205
206     parser.add_argument("--disable-js-validation",
207                         action="store_true", default=False,
208                         help=argparse.SUPPRESS)
209
210     parser.add_argument("--thread-count", type=int,
211                         default=0, help="Number of threads to use for job submit and output collection.")
212
213     parser.add_argument("--http-timeout", type=int,
214                         default=5*60, dest="http_timeout", help="API request timeout in seconds. Default is 300 seconds (5 minutes).")
215
216     exgroup = parser.add_mutually_exclusive_group()
217     exgroup.add_argument("--enable-preemptible", dest="enable_preemptible", default=None, action="store_true", help="Use preemptible instances. Control individual steps with arv:UsePreemptible hint.")
218     exgroup.add_argument("--disable-preemptible", dest="enable_preemptible", default=None, action="store_false", help="Don't use preemptible instances.")
219
220     exgroup = parser.add_mutually_exclusive_group()
221     exgroup.add_argument("--copy-deps", dest="copy_deps", default=None, action="store_true", help="Copy dependencies into the destination project.")
222     exgroup.add_argument("--no-copy-deps", dest="copy_deps", default=None, action="store_false", help="Leave dependencies where they are.")
223
224     parser.add_argument(
225         "--skip-schemas",
226         action="store_true",
227         help="Skip loading of schemas",
228         default=False,
229         dest="skip_schemas",
230     )
231
232     exgroup = parser.add_mutually_exclusive_group()
233     exgroup.add_argument("--trash-intermediate", action="store_true",
234                         default=False, dest="trash_intermediate",
235                          help="Immediately trash intermediate outputs on workflow success.")
236     exgroup.add_argument("--no-trash-intermediate", action="store_false",
237                         default=False, dest="trash_intermediate",
238                         help="Do not trash intermediate outputs (default).")
239
240     parser.add_argument("workflow", default=None, help="The workflow to execute")
241     parser.add_argument("job_order", nargs=argparse.REMAINDER, help="The input object to the workflow.")
242
243     return parser
244
245 def add_arv_hints():
246     cwltool.command_line_tool.ACCEPTLIST_EN_RELAXED_RE = re.compile(r".*")
247     cwltool.command_line_tool.ACCEPTLIST_RE = cwltool.command_line_tool.ACCEPTLIST_EN_RELAXED_RE
248     supported_versions = ["v1.0", "v1.1", "v1.2"]
249     for s in supported_versions:
250         res = pkg_resources.resource_stream(__name__, 'arv-cwl-schema-%s.yml' % s)
251         customschema = res.read().decode('utf-8')
252         use_custom_schema(s, "http://arvados.org/cwl", customschema)
253         res.close()
254     cwltool.process.supportedProcessRequirements.extend([
255         "http://arvados.org/cwl#RunInSingleContainer",
256         "http://arvados.org/cwl#OutputDirType",
257         "http://arvados.org/cwl#RuntimeConstraints",
258         "http://arvados.org/cwl#PartitionRequirement",
259         "http://arvados.org/cwl#APIRequirement",
260         "http://commonwl.org/cwltool#LoadListingRequirement",
261         "http://arvados.org/cwl#IntermediateOutput",
262         "http://arvados.org/cwl#ReuseRequirement",
263         "http://arvados.org/cwl#ClusterTarget",
264         "http://arvados.org/cwl#OutputStorageClass",
265         "http://arvados.org/cwl#ProcessProperties",
266         "http://commonwl.org/cwltool#CUDARequirement",
267         "http://arvados.org/cwl#UsePreemptible",
268     ])
269
270 def exit_signal_handler(sigcode, frame):
271     logger.error(str(u"Caught signal {}, exiting.").format(sigcode))
272     sys.exit(-sigcode)
273
274 def main(args=sys.argv[1:],
275          stdout=sys.stdout,
276          stderr=sys.stderr,
277          api_client=None,
278          keep_client=None,
279          install_sig_handlers=True):
280     parser = arg_parser()
281
282     job_order_object = None
283     arvargs = parser.parse_args(args)
284
285     arvargs.use_container = True
286     arvargs.relax_path_checks = True
287     arvargs.print_supported_versions = False
288
289     if install_sig_handlers:
290         arv_cmd.install_signal_handlers()
291
292     if arvargs.update_workflow:
293         if arvargs.update_workflow.find('-7fd4e-') == 5:
294             want_api = 'containers'
295         else:
296             want_api = None
297         if want_api and arvargs.work_api and want_api != arvargs.work_api:
298             logger.error(str(u'--update-workflow arg {!r} uses {!r} API, but --api={!r} specified').format(
299                 arvargs.update_workflow, want_api, arvargs.work_api))
300             return 1
301         arvargs.work_api = want_api
302
303     if (arvargs.create_workflow or arvargs.update_workflow) and not arvargs.job_order:
304         job_order_object = ({}, "")
305
306     add_arv_hints()
307
308     for key, val in viewitems(cwltool.argparser.get_default_args()):
309         if not hasattr(arvargs, key):
310             setattr(arvargs, key, val)
311
312     try:
313         if api_client is None:
314             api_client = arvados.safeapi.ThreadSafeApiCache(
315                 api_params={"model": OrderedJsonModel(), "timeout": arvargs.http_timeout},
316                 keep_params={"num_retries": 4})
317             keep_client = api_client.keep
318             # Make an API object now so errors are reported early.
319             api_client.users().current().execute()
320         if keep_client is None:
321             keep_client = arvados.keep.KeepClient(api_client=api_client, num_retries=4)
322         executor = ArvCwlExecutor(api_client, arvargs, keep_client=keep_client, num_retries=4, stdout=stdout)
323     except WorkflowException as e:
324         logger.error(e, exc_info=(sys.exc_info()[1] if arvargs.debug else False))
325         return 1
326     except Exception:
327         logger.exception("Error creating the Arvados CWL Executor")
328         return 1
329
330     # Note that unless in debug mode, some stack traces related to user
331     # workflow errors may be suppressed.
332     if arvargs.debug:
333         logger.setLevel(logging.DEBUG)
334         logging.getLogger('arvados').setLevel(logging.DEBUG)
335
336     if arvargs.quiet:
337         logger.setLevel(logging.WARN)
338         logging.getLogger('arvados').setLevel(logging.WARN)
339         logging.getLogger('arvados.arv-run').setLevel(logging.WARN)
340
341     if arvargs.metrics:
342         metrics.setLevel(logging.DEBUG)
343         logging.getLogger("cwltool.metrics").setLevel(logging.DEBUG)
344
345     if arvargs.log_timestamps:
346         arvados.log_handler.setFormatter(logging.Formatter(
347             '%(asctime)s %(name)s %(levelname)s: %(message)s',
348             '%Y-%m-%d %H:%M:%S'))
349     else:
350         arvados.log_handler.setFormatter(logging.Formatter('%(name)s %(levelname)s: %(message)s'))
351
352     if stdout is sys.stdout:
353         # cwltool.main has code to work around encoding issues with
354         # sys.stdout and unix pipes (they default to ASCII encoding,
355         # we want utf-8), so when stdout is sys.stdout set it to None
356         # to take advantage of that.  Don't override it for all cases
357         # since we still want to be able to capture stdout for the
358         # unit tests.
359         stdout = None
360
361     return cwltool.main.main(args=arvargs,
362                              stdout=stdout,
363                              stderr=stderr,
364                              executor=executor.arv_executor,
365                              versionfunc=versionstring,
366                              job_order_object=job_order_object,
367                              logger_handler=arvados.log_handler,
368                              custom_schema_callback=add_arv_hints,
369                              loadingContext=executor.loadingContext,
370                              runtimeContext=executor.toplevel_runtimeContext,
371                              input_required=not (arvargs.create_workflow or arvargs.update_workflow))