19464: Record the file path in the git repo as well
[arvados.git] / sdk / cwl / arvados_cwl / executor.py
1 # Copyright (C) The Arvados Authors. All rights reserved.
2 #
3 # SPDX-License-Identifier: Apache-2.0
4
5 from __future__ import division
6 from builtins import next
7 from builtins import object
8 from builtins import str
9 from future.utils import viewvalues, viewitems
10
11 import argparse
12 import logging
13 import os
14 import sys
15 import threading
16 import copy
17 import json
18 import re
19 from functools import partial
20 import subprocess
21 import time
22 import urllib
23
24 from cwltool.errors import WorkflowException
25 import cwltool.workflow
26 from schema_salad.sourceline import SourceLine
27 import schema_salad.validate as validate
28 from schema_salad.ref_resolver import file_uri, uri_file_path
29
30 import arvados
31 import arvados.config
32 from arvados.keep import KeepClient
33 from arvados.errors import ApiError
34
35 import arvados_cwl.util
36 from .arvcontainer import RunnerContainer, cleanup_name_for_collection
37 from .runner import Runner, upload_docker, upload_job_order, upload_workflow_deps, make_builder
38 from .arvtool import ArvadosCommandTool, validate_cluster_target, ArvadosExpressionTool
39 from .arvworkflow import ArvadosWorkflow, upload_workflow
40 from .fsaccess import CollectionFsAccess, CollectionFetcher, collectionResolver, CollectionCache, pdh_size
41 from .perf import Perf
42 from .pathmapper import NoFollowPathMapper
43 from cwltool.task_queue import TaskQueue
44 from .context import ArvLoadingContext, ArvRuntimeContext
45 from ._version import __version__
46
47 from cwltool.process import shortname, UnsupportedRequirement, use_custom_schema
48 from cwltool.utils import adjustFileObjs, adjustDirObjs, get_listing, visit_class, aslist
49 from cwltool.command_line_tool import compute_checksums
50 from cwltool.load_tool import load_tool
51
52 logger = logging.getLogger('arvados.cwl-runner')
53 metrics = logging.getLogger('arvados.cwl-runner.metrics')
54
55 DEFAULT_PRIORITY = 500
56
57 class RuntimeStatusLoggingHandler(logging.Handler):
58     """
59     Intercepts logging calls and report them as runtime statuses on runner
60     containers.
61     """
62     def __init__(self, runtime_status_update_func):
63         super(RuntimeStatusLoggingHandler, self).__init__()
64         self.runtime_status_update = runtime_status_update_func
65         self.updatingRuntimeStatus = False
66
67     def emit(self, record):
68         kind = None
69         if record.levelno >= logging.ERROR:
70             kind = 'error'
71         elif record.levelno >= logging.WARNING:
72             kind = 'warning'
73         if kind is not None and self.updatingRuntimeStatus is not True:
74             self.updatingRuntimeStatus = True
75             try:
76                 log_msg = record.getMessage()
77                 if '\n' in log_msg:
78                     # If the logged message is multi-line, use its first line as status
79                     # and the rest as detail.
80                     status, detail = log_msg.split('\n', 1)
81                     self.runtime_status_update(
82                         kind,
83                         "%s: %s" % (record.name, status),
84                         detail
85                     )
86                 else:
87                     self.runtime_status_update(
88                         kind,
89                         "%s: %s" % (record.name, record.getMessage())
90                     )
91             finally:
92                 self.updatingRuntimeStatus = False
93
94
95 class ArvCwlExecutor(object):
96     """Execute a CWL tool or workflow, submit work (using containers API),
97     wait for them to complete, and report output.
98
99     """
100
101     def __init__(self, api_client,
102                  arvargs=None,
103                  keep_client=None,
104                  num_retries=4,
105                  thread_count=4,
106                  stdout=sys.stdout):
107
108         if arvargs is None:
109             arvargs = argparse.Namespace()
110             arvargs.work_api = None
111             arvargs.output_name = None
112             arvargs.output_tags = None
113             arvargs.thread_count = 1
114             arvargs.collection_cache_size = None
115
116         self.api = api_client
117         self.processes = {}
118         self.workflow_eval_lock = threading.Condition(threading.RLock())
119         self.final_output = None
120         self.final_status = None
121         self.num_retries = num_retries
122         self.uuid = None
123         self.stop_polling = threading.Event()
124         self.poll_api = None
125         self.pipeline = None
126         self.final_output_collection = None
127         self.output_name = arvargs.output_name
128         self.output_tags = arvargs.output_tags
129         self.project_uuid = None
130         self.intermediate_output_ttl = 0
131         self.intermediate_output_collections = []
132         self.trash_intermediate = False
133         self.thread_count = arvargs.thread_count
134         self.poll_interval = 12
135         self.loadingContext = None
136         self.should_estimate_cache_size = True
137         self.fs_access = None
138         self.secret_store = None
139         self.stdout = stdout
140
141         if keep_client is not None:
142             self.keep_client = keep_client
143         else:
144             self.keep_client = arvados.keep.KeepClient(api_client=self.api, num_retries=self.num_retries)
145
146         if arvargs.collection_cache_size:
147             collection_cache_size = arvargs.collection_cache_size*1024*1024
148             self.should_estimate_cache_size = False
149         else:
150             collection_cache_size = 256*1024*1024
151
152         self.collection_cache = CollectionCache(self.api, self.keep_client, self.num_retries,
153                                                 cap=collection_cache_size)
154
155         self.fetcher_constructor = partial(CollectionFetcher,
156                                            api_client=self.api,
157                                            fs_access=CollectionFsAccess("", collection_cache=self.collection_cache),
158                                            num_retries=self.num_retries)
159
160         self.work_api = None
161         expected_api = ["containers"]
162         for api in expected_api:
163             try:
164                 methods = self.api._rootDesc.get('resources')[api]['methods']
165                 if ('httpMethod' in methods['create'] and
166                     (arvargs.work_api == api or arvargs.work_api is None)):
167                     self.work_api = api
168                     break
169             except KeyError:
170                 pass
171
172         if not self.work_api:
173             if arvargs.work_api is None:
174                 raise Exception("No supported APIs")
175             else:
176                 raise Exception("Unsupported API '%s', expected one of %s" % (arvargs.work_api, expected_api))
177
178         if self.work_api == "jobs":
179             logger.error("""
180 *******************************
181 The 'jobs' API is no longer supported.
182 *******************************""")
183             exit(1)
184
185         self.loadingContext = ArvLoadingContext(vars(arvargs))
186         self.loadingContext.fetcher_constructor = self.fetcher_constructor
187         self.loadingContext.resolver = partial(collectionResolver, self.api, num_retries=self.num_retries)
188         self.loadingContext.construct_tool_object = self.arv_make_tool
189
190         # Add a custom logging handler to the root logger for runtime status reporting
191         # if running inside a container
192         if arvados_cwl.util.get_current_container(self.api, self.num_retries, logger):
193             root_logger = logging.getLogger('')
194
195             # Remove existing RuntimeStatusLoggingHandlers if they exist
196             handlers = [h for h in root_logger.handlers if not isinstance(h, RuntimeStatusLoggingHandler)]
197             root_logger.handlers = handlers
198
199             handler = RuntimeStatusLoggingHandler(self.runtime_status_update)
200             root_logger.addHandler(handler)
201
202         self.toplevel_runtimeContext = ArvRuntimeContext(vars(arvargs))
203         self.toplevel_runtimeContext.make_fs_access = partial(CollectionFsAccess,
204                                                      collection_cache=self.collection_cache)
205
206         validate_cluster_target(self, self.toplevel_runtimeContext)
207
208
209     def arv_make_tool(self, toolpath_object, loadingContext):
210         if "class" in toolpath_object and toolpath_object["class"] == "CommandLineTool":
211             return ArvadosCommandTool(self, toolpath_object, loadingContext)
212         elif "class" in toolpath_object and toolpath_object["class"] == "Workflow":
213             return ArvadosWorkflow(self, toolpath_object, loadingContext)
214         elif "class" in toolpath_object and toolpath_object["class"] == "ExpressionTool":
215             return ArvadosExpressionTool(self, toolpath_object, loadingContext)
216         else:
217             raise Exception("Unknown tool %s" % toolpath_object.get("class"))
218
219     def output_callback(self, out, processStatus):
220         with self.workflow_eval_lock:
221             if processStatus == "success":
222                 logger.info("Overall process status is %s", processStatus)
223                 state = "Complete"
224             else:
225                 logger.error("Overall process status is %s", processStatus)
226                 state = "Failed"
227             if self.pipeline:
228                 self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
229                                                         body={"state": state}).execute(num_retries=self.num_retries)
230             self.final_status = processStatus
231             self.final_output = out
232             self.workflow_eval_lock.notifyAll()
233
234
235     def start_run(self, runnable, runtimeContext):
236         self.task_queue.add(partial(runnable.run, runtimeContext),
237                             self.workflow_eval_lock, self.stop_polling)
238
239     def process_submitted(self, container):
240         with self.workflow_eval_lock:
241             self.processes[container.uuid] = container
242
243     def process_done(self, uuid, record):
244         with self.workflow_eval_lock:
245             j = self.processes[uuid]
246             logger.info("%s %s is %s", self.label(j), uuid, record["state"])
247             self.task_queue.add(partial(j.done, record),
248                                 self.workflow_eval_lock, self.stop_polling)
249             del self.processes[uuid]
250
251     def runtime_status_update(self, kind, message, detail=None):
252         """
253         Updates the runtime_status field on the runner container.
254         Called when there's a need to report errors, warnings or just
255         activity statuses, for example in the RuntimeStatusLoggingHandler.
256         """
257
258         if kind not in ('error', 'warning'):
259             # Ignore any other status kind
260             return
261
262         with self.workflow_eval_lock:
263             current = None
264             try:
265                 current = arvados_cwl.util.get_current_container(self.api, self.num_retries, logger)
266             except Exception as e:
267                 logger.info("Couldn't get current container: %s", e)
268             if current is None:
269                 return
270             runtime_status = current.get('runtime_status', {})
271
272             original_updatemessage = updatemessage = runtime_status.get(kind, "")
273             if not updatemessage:
274                 updatemessage = message
275
276             # Subsequent messages tacked on in detail
277             original_updatedetail = updatedetail = runtime_status.get(kind+'Detail', "")
278             maxlines = 40
279             if updatedetail.count("\n") < maxlines:
280                 if updatedetail:
281                     updatedetail += "\n"
282                 updatedetail += message + "\n"
283
284                 if detail:
285                     updatedetail += detail + "\n"
286
287                 if updatedetail.count("\n") >= maxlines:
288                     updatedetail += "\nSome messages may have been omitted.  Check the full log."
289
290             if updatemessage == original_updatemessage and updatedetail == original_updatedetail:
291                 # don't waste time doing an update if nothing changed
292                 # (usually because we exceeded the max lines)
293                 return
294
295             runtime_status.update({
296                 kind: updatemessage,
297                 kind+'Detail': updatedetail,
298             })
299
300             try:
301                 self.api.containers().update(uuid=current['uuid'],
302                                             body={
303                                                 'runtime_status': runtime_status,
304                                             }).execute(num_retries=self.num_retries)
305             except Exception as e:
306                 logger.info("Couldn't update runtime_status: %s", e)
307
308     def wrapped_callback(self, cb, obj, st):
309         with self.workflow_eval_lock:
310             cb(obj, st)
311             self.workflow_eval_lock.notifyAll()
312
313     def get_wrapped_callback(self, cb):
314         return partial(self.wrapped_callback, cb)
315
316     def on_message(self, event):
317         if event.get("object_uuid") in self.processes and event["event_type"] == "update":
318             uuid = event["object_uuid"]
319             if event["properties"]["new_attributes"]["state"] == "Running":
320                 with self.workflow_eval_lock:
321                     j = self.processes[uuid]
322                     if j.running is False:
323                         j.running = True
324                         j.update_pipeline_component(event["properties"]["new_attributes"])
325                         logger.info("%s %s is Running", self.label(j), uuid)
326             elif event["properties"]["new_attributes"]["state"] in ("Complete", "Failed", "Cancelled", "Final"):
327                 self.process_done(uuid, event["properties"]["new_attributes"])
328
329     def label(self, obj):
330         return "[%s %s]" % (self.work_api[0:-1], obj.name)
331
332     def poll_states(self):
333         """Poll status of containers listed in the processes dict.
334
335         Runs in a separate thread.
336         """
337
338         try:
339             remain_wait = self.poll_interval
340             while True:
341                 if remain_wait > 0:
342                     self.stop_polling.wait(remain_wait)
343                 if self.stop_polling.is_set():
344                     break
345                 with self.workflow_eval_lock:
346                     keys = list(self.processes)
347                 if not keys:
348                     remain_wait = self.poll_interval
349                     continue
350
351                 begin_poll = time.time()
352                 if self.work_api == "containers":
353                     table = self.poll_api.container_requests()
354
355                 pageSize = self.poll_api._rootDesc.get('maxItemsPerResponse', 1000)
356
357                 while keys:
358                     page = keys[:pageSize]
359                     try:
360                         proc_states = table.list(filters=[["uuid", "in", page]]).execute(num_retries=self.num_retries)
361                     except Exception:
362                         logger.exception("Error checking states on API server: %s")
363                         remain_wait = self.poll_interval
364                         continue
365
366                     for p in proc_states["items"]:
367                         self.on_message({
368                             "object_uuid": p["uuid"],
369                             "event_type": "update",
370                             "properties": {
371                                 "new_attributes": p
372                             }
373                         })
374                     keys = keys[pageSize:]
375
376                 finish_poll = time.time()
377                 remain_wait = self.poll_interval - (finish_poll - begin_poll)
378         except:
379             logger.exception("Fatal error in state polling thread.")
380             with self.workflow_eval_lock:
381                 self.processes.clear()
382                 self.workflow_eval_lock.notifyAll()
383         finally:
384             self.stop_polling.set()
385
386     def add_intermediate_output(self, uuid):
387         if uuid:
388             self.intermediate_output_collections.append(uuid)
389
390     def trash_intermediate_output(self):
391         logger.info("Cleaning up intermediate output collections")
392         for i in self.intermediate_output_collections:
393             try:
394                 self.api.collections().delete(uuid=i).execute(num_retries=self.num_retries)
395             except Exception:
396                 logger.warning("Failed to delete intermediate output: %s", sys.exc_info()[1], exc_info=(sys.exc_info()[1] if self.debug else False))
397             except (KeyboardInterrupt, SystemExit):
398                 break
399
400     def check_features(self, obj, parentfield=""):
401         if isinstance(obj, dict):
402             if obj.get("class") == "DockerRequirement":
403                 if obj.get("dockerOutputDirectory"):
404                     if not obj.get("dockerOutputDirectory").startswith('/'):
405                         raise SourceLine(obj, "dockerOutputDirectory", validate.ValidationException).makeError(
406                             "Option 'dockerOutputDirectory' must be an absolute path.")
407             if obj.get("class") == "InplaceUpdateRequirement":
408                 if obj["inplaceUpdate"] and parentfield == "requirements":
409                     raise SourceLine(obj, "class", UnsupportedRequirement).makeError("InplaceUpdateRequirement not supported for keep collections.")
410             for k,v in viewitems(obj):
411                 self.check_features(v, parentfield=k)
412         elif isinstance(obj, list):
413             for i,v in enumerate(obj):
414                 with SourceLine(obj, i, UnsupportedRequirement, logger.isEnabledFor(logging.DEBUG)):
415                     self.check_features(v, parentfield=parentfield)
416
417     def make_output_collection(self, name, storage_classes, tagsString, output_properties, outputObj):
418         outputObj = copy.deepcopy(outputObj)
419
420         files = []
421         def capture(fileobj):
422             files.append(fileobj)
423
424         adjustDirObjs(outputObj, capture)
425         adjustFileObjs(outputObj, capture)
426
427         generatemapper = NoFollowPathMapper(files, "", "", separateDirs=False)
428
429         final = arvados.collection.Collection(api_client=self.api,
430                                               keep_client=self.keep_client,
431                                               num_retries=self.num_retries)
432
433         for k,v in generatemapper.items():
434             if v.type == "Directory" and v.resolved.startswith("_:"):
435                     continue
436             if v.type == "CreateFile" and (k.startswith("_:") or v.resolved.startswith("_:")):
437                 with final.open(v.target, "wb") as f:
438                     f.write(v.resolved.encode("utf-8"))
439                     continue
440
441             if not v.resolved.startswith("keep:"):
442                 raise Exception("Output source is not in keep or a literal")
443             sp = v.resolved.split("/")
444             srccollection = sp[0][5:]
445             try:
446                 reader = self.collection_cache.get(srccollection)
447                 srcpath = urllib.parse.unquote("/".join(sp[1:]) if len(sp) > 1 else ".")
448                 final.copy(srcpath, v.target, source_collection=reader, overwrite=False)
449             except arvados.errors.ArgumentError as e:
450                 logger.error("Creating CollectionReader for '%s' '%s': %s", k, v, e)
451                 raise
452             except IOError as e:
453                 logger.error("While preparing output collection: %s", e)
454                 raise
455
456         def rewrite(fileobj):
457             fileobj["location"] = generatemapper.mapper(fileobj["location"]).target
458             for k in ("listing", "contents", "nameext", "nameroot", "dirname"):
459                 if k in fileobj:
460                     del fileobj[k]
461
462         adjustDirObjs(outputObj, rewrite)
463         adjustFileObjs(outputObj, rewrite)
464
465         with final.open("cwl.output.json", "w") as f:
466             res = str(json.dumps(outputObj, sort_keys=True, indent=4, separators=(',',': '), ensure_ascii=False))
467             f.write(res)
468
469
470         final.save_new(name=name, owner_uuid=self.project_uuid, storage_classes=storage_classes,
471                        ensure_unique_name=True, properties=output_properties)
472
473         logger.info("Final output collection %s \"%s\" (%s)", final.portable_data_hash(),
474                     final.api_response()["name"],
475                     final.manifest_locator())
476
477         final_uuid = final.manifest_locator()
478         tags = tagsString.split(',')
479         for tag in tags:
480              self.api.links().create(body={
481                 "head_uuid": final_uuid, "link_class": "tag", "name": tag
482                 }).execute(num_retries=self.num_retries)
483
484         def finalcollection(fileobj):
485             fileobj["location"] = "keep:%s/%s" % (final.portable_data_hash(), fileobj["location"])
486
487         adjustDirObjs(outputObj, finalcollection)
488         adjustFileObjs(outputObj, finalcollection)
489
490         return (outputObj, final)
491
492     def set_crunch_output(self):
493         if self.work_api == "containers":
494             current = arvados_cwl.util.get_current_container(self.api, self.num_retries, logger)
495             if current is None:
496                 return
497             try:
498                 self.api.containers().update(uuid=current['uuid'],
499                                              body={
500                                                  'output': self.final_output_collection.portable_data_hash(),
501                                                  'output_properties': self.final_output_collection.get_properties(),
502                                              }).execute(num_retries=self.num_retries)
503                 self.api.collections().update(uuid=self.final_output_collection.manifest_locator(),
504                                               body={
505                                                   'is_trashed': True
506                                               }).execute(num_retries=self.num_retries)
507             except Exception:
508                 logger.exception("Setting container output")
509                 raise
510
511     def apply_reqs(self, job_order_object, tool):
512         if "https://w3id.org/cwl/cwl#requirements" in job_order_object:
513             if tool.metadata.get("http://commonwl.org/cwltool#original_cwlVersion") == 'v1.0':
514                 raise WorkflowException(
515                     "`cwl:requirements` in the input object is not part of CWL "
516                     "v1.0. You can adjust to use `cwltool:overrides` instead; or you "
517                     "can set the cwlVersion to v1.1 or greater and re-run with "
518                     "--enable-dev.")
519             job_reqs = job_order_object["https://w3id.org/cwl/cwl#requirements"]
520             for req in job_reqs:
521                 tool.requirements.append(req)
522
523     def get_git_info(self, tool):
524         in_a_git_repo = False
525         cwd = None
526         filepath = None
527
528         if tool.tool["id"].startswith("file://"):
529             # check if git is installed
530             try:
531                 filepath = uri_file_path(tool.tool["id"])
532                 cwd = os.path.dirname(filepath)
533                 subprocess.run(["git", "log", "--format=%H", "-n1", "HEAD"], cwd=cwd, check=True, capture_output=True, text=True)
534                 in_a_git_repo = True
535             except Exception as e:
536                 pass
537
538         gitproperties = {}
539
540         if in_a_git_repo:
541             git_commit = subprocess.run(["git", "log", "--format=%H", "-n1", "HEAD"], cwd=cwd, capture_output=True, text=True).stdout
542             git_date = subprocess.run(["git", "log", "--format=%cD", "-n1", "HEAD"], cwd=cwd, capture_output=True, text=True).stdout
543             git_committer = subprocess.run(["git", "log", "--format=%cn <%ce>", "-n1", "HEAD"], cwd=cwd, capture_output=True, text=True).stdout
544             git_branch = subprocess.run(["git", "branch", "--show-current"], cwd=cwd, capture_output=True, text=True).stdout
545             git_origin = subprocess.run(["git", "remote", "get-url", "origin"], cwd=cwd, capture_output=True, text=True).stdout
546             git_status = subprocess.run(["git", "status", "--untracked-files=no", "--porcelain"], cwd=cwd, capture_output=True, text=True).stdout
547             git_describe = subprocess.run(["git", "describe", "--always"], cwd=cwd, capture_output=True, text=True).stdout
548             git_toplevel = subprocess.run(["git", "rev-parse", "--show-toplevel"], cwd=cwd, capture_output=True, text=True).stdout
549             git_path = filepath[len(git_toplevel):]
550
551             gitproperties = {
552                 "http://arvados.org/cwl#gitCommit": git_commit.strip(),
553                 "http://arvados.org/cwl#gitDate": git_date.strip(),
554                 "http://arvados.org/cwl#gitCommitter": git_committer.strip(),
555                 "http://arvados.org/cwl#gitBranch": git_branch.strip(),
556                 "http://arvados.org/cwl#gitOrigin": git_origin.strip(),
557                 "http://arvados.org/cwl#gitStatus": git_status.strip(),
558                 "http://arvados.org/cwl#gitDescribe": git_describe.strip(),
559                 "http://arvados.org/cwl#gitPath": git_path.strip(),
560             }
561         else:
562             for g in ("http://arvados.org/cwl#gitCommit",
563                       "http://arvados.org/cwl#gitDate",
564                       "http://arvados.org/cwl#gitCommitter",
565                       "http://arvados.org/cwl#gitBranch",
566                       "http://arvados.org/cwl#gitOrigin",
567                       "http://arvados.org/cwl#gitStatus",
568                       "http://arvados.org/cwl#gitDescribe",
569                       "http://arvados.org/cwl#gitPath"):
570                 if g in tool.metadata:
571                     gitproperties[g] = tool.metadata[g]
572
573         return gitproperties
574
575     def arv_executor(self, updated_tool, job_order, runtimeContext, logger=None):
576         self.debug = runtimeContext.debug
577
578         git_info = self.get_git_info(updated_tool)
579         if git_info:
580             logger.info("Git provenance of %s", updated_tool.tool["id"])
581             for g in git_info:
582                 if git_info[g]:
583                     logger.info("  %s: %s", g.split("#", 1)[1], git_info[g])
584
585         workbench1 = self.api.config()["Services"]["Workbench1"]["ExternalURL"]
586         workbench2 = self.api.config()["Services"]["Workbench2"]["ExternalURL"]
587         controller = self.api.config()["Services"]["Controller"]["ExternalURL"]
588         logger.info("Using cluster %s (%s)", self.api.config()["ClusterID"], workbench2 or workbench1 or controller)
589
590         updated_tool.visit(self.check_features)
591
592         self.pipeline = None
593         self.fs_access = runtimeContext.make_fs_access(runtimeContext.basedir)
594         self.secret_store = runtimeContext.secret_store
595
596         self.trash_intermediate = runtimeContext.trash_intermediate
597         if self.trash_intermediate and self.work_api != "containers":
598             raise Exception("--trash-intermediate is only supported with --api=containers.")
599
600         self.intermediate_output_ttl = runtimeContext.intermediate_output_ttl
601         if self.intermediate_output_ttl and self.work_api != "containers":
602             raise Exception("--intermediate-output-ttl is only supported with --api=containers.")
603         if self.intermediate_output_ttl < 0:
604             raise Exception("Invalid value %d for --intermediate-output-ttl, cannot be less than zero" % self.intermediate_output_ttl)
605
606         if runtimeContext.submit_request_uuid and self.work_api != "containers":
607             raise Exception("--submit-request-uuid requires containers API, but using '{}' api".format(self.work_api))
608
609         runtimeContext = runtimeContext.copy()
610
611         default_storage_classes = ",".join([k for k,v in self.api.config().get("StorageClasses", {"default": {"Default": True}}).items() if v.get("Default") is True])
612         if runtimeContext.storage_classes == "default":
613             runtimeContext.storage_classes = default_storage_classes
614         if runtimeContext.intermediate_storage_classes == "default":
615             runtimeContext.intermediate_storage_classes = default_storage_classes
616
617         if not runtimeContext.name:
618             self.name = updated_tool.tool.get("label") or updated_tool.metadata.get("label") or os.path.basename(updated_tool.tool["id"])
619             if git_info.get("http://arvados.org/cwl#gitDescribe"):
620                 self.name = "%s (%s)" % (self.name, git_info.get("http://arvados.org/cwl#gitDescribe"))
621             runtimeContext.name = self.name
622
623         if runtimeContext.copy_deps is None and (runtimeContext.create_workflow or runtimeContext.update_workflow):
624             # When creating or updating workflow record, by default
625             # always copy dependencies and ensure Docker images are up
626             # to date.
627             runtimeContext.copy_deps = True
628             runtimeContext.match_local_docker = True
629
630         if runtimeContext.update_workflow and self.project_uuid is None:
631             # If we are updating a workflow, make sure anything that
632             # gets uploaded goes into the same parent project, unless
633             # an alternate --project-uuid was provided.
634             existing_wf = self.api.workflows().get(uuid=runtimeContext.update_workflow).execute()
635             runtimeContext.project_uuid = existing_wf["owner_uuid"]
636
637         self.project_uuid = runtimeContext.project_uuid
638
639         # Upload local file references in the job order.
640         with Perf(metrics, "upload_job_order"):
641             job_order = upload_job_order(self, "%s input" % runtimeContext.name,
642                                          updated_tool, job_order, runtimeContext)
643
644         # the last clause means: if it is a command line tool, and we
645         # are going to wait for the result, and always_submit_runner
646         # is false, then we don't submit a runner process.
647
648         submitting = (runtimeContext.update_workflow or
649                       runtimeContext.create_workflow or
650                       (runtimeContext.submit and not
651                        (updated_tool.tool["class"] == "CommandLineTool" and
652                         runtimeContext.wait and
653                         not runtimeContext.always_submit_runner)))
654
655         loadingContext = self.loadingContext.copy()
656         loadingContext.do_validate = False
657         loadingContext.disable_js_validation = True
658         if submitting:
659             loadingContext.do_update = False
660             # Document may have been auto-updated. Reload the original
661             # document with updating disabled because we want to
662             # submit the document with its original CWL version, not
663             # the auto-updated one.
664             with Perf(metrics, "load_tool original"):
665                 tool = load_tool(updated_tool.tool["id"], loadingContext)
666         else:
667             tool = updated_tool
668
669         # Upload direct dependencies of workflow steps, get back mapping of files to keep references.
670         # Also uploads docker images.
671         logger.info("Uploading workflow dependencies")
672         with Perf(metrics, "upload_workflow_deps"):
673             merged_map = upload_workflow_deps(self, tool, runtimeContext)
674
675         # Recreate process object (ArvadosWorkflow or
676         # ArvadosCommandTool) because tool document may have been
677         # updated by upload_workflow_deps in ways that modify
678         # inheritance of hints or requirements.
679         loadingContext.loader = tool.doc_loader
680         loadingContext.avsc_names = tool.doc_schema
681         loadingContext.metadata = tool.metadata
682         with Perf(metrics, "load_tool"):
683             tool = load_tool(tool.tool, loadingContext)
684
685         if runtimeContext.update_workflow or runtimeContext.create_workflow:
686             # Create a pipeline template or workflow record and exit.
687             if self.work_api == "containers":
688                 uuid = upload_workflow(self, tool, job_order,
689                                        runtimeContext.project_uuid,
690                                        runtimeContext,
691                                        uuid=runtimeContext.update_workflow,
692                                        submit_runner_ram=runtimeContext.submit_runner_ram,
693                                        name=runtimeContext.name,
694                                        merged_map=merged_map,
695                                        submit_runner_image=runtimeContext.submit_runner_image,
696                                        git_info=git_info)
697                 self.stdout.write(uuid + "\n")
698                 return (None, "success")
699
700         self.apply_reqs(job_order, tool)
701
702         self.ignore_docker_for_reuse = runtimeContext.ignore_docker_for_reuse
703         self.eval_timeout = runtimeContext.eval_timeout
704
705         runtimeContext.use_container = True
706         runtimeContext.tmpdir_prefix = "tmp"
707         runtimeContext.work_api = self.work_api
708
709         if not self.output_name:
710              self.output_name = "Output from workflow %s" % runtimeContext.name
711
712         self.output_name  = cleanup_name_for_collection(self.output_name)
713
714         if self.work_api == "containers":
715             if self.ignore_docker_for_reuse:
716                 raise Exception("--ignore-docker-for-reuse not supported with containers API.")
717             runtimeContext.outdir = "/var/spool/cwl"
718             runtimeContext.docker_outdir = "/var/spool/cwl"
719             runtimeContext.tmpdir = "/tmp"
720             runtimeContext.docker_tmpdir = "/tmp"
721
722         if runtimeContext.priority < 1 or runtimeContext.priority > 1000:
723             raise Exception("--priority must be in the range 1..1000.")
724
725         if self.should_estimate_cache_size:
726             visited = set()
727             estimated_size = [0]
728             def estimate_collection_cache(obj):
729                 if obj.get("location", "").startswith("keep:"):
730                     m = pdh_size.match(obj["location"][5:])
731                     if m and m.group(1) not in visited:
732                         visited.add(m.group(1))
733                         estimated_size[0] += int(m.group(2))
734             visit_class(job_order, ("File", "Directory"), estimate_collection_cache)
735             runtimeContext.collection_cache_size = max(((estimated_size[0]*192) // (1024*1024))+1, 256)
736             self.collection_cache.set_cap(runtimeContext.collection_cache_size*1024*1024)
737
738         logger.info("Using collection cache size %s MiB", runtimeContext.collection_cache_size)
739
740         runnerjob = None
741         if runtimeContext.submit:
742             # Submit a runner job to run the workflow for us.
743             if self.work_api == "containers":
744                 if submitting:
745                     tool = RunnerContainer(self, updated_tool,
746                                            tool, loadingContext, runtimeContext.enable_reuse,
747                                            self.output_name,
748                                            self.output_tags,
749                                            submit_runner_ram=runtimeContext.submit_runner_ram,
750                                            name=runtimeContext.name,
751                                            on_error=runtimeContext.on_error,
752                                            submit_runner_image=runtimeContext.submit_runner_image,
753                                            intermediate_output_ttl=runtimeContext.intermediate_output_ttl,
754                                            merged_map=merged_map,
755                                            priority=runtimeContext.priority,
756                                            secret_store=self.secret_store,
757                                            collection_cache_size=runtimeContext.collection_cache_size,
758                                            collection_cache_is_default=self.should_estimate_cache_size,
759                                            git_info=git_info)
760                 else:
761                     runtimeContext.runnerjob = tool.tool["id"]
762
763         if runtimeContext.cwl_runner_job is not None:
764             self.uuid = runtimeContext.cwl_runner_job.get('uuid')
765
766         jobiter = tool.job(job_order,
767                            self.output_callback,
768                            runtimeContext)
769
770         if runtimeContext.submit and not runtimeContext.wait:
771             runnerjob = next(jobiter)
772             runnerjob.run(runtimeContext)
773             self.stdout.write(runnerjob.uuid+"\n")
774             return (None, "success")
775
776         current_container = arvados_cwl.util.get_current_container(self.api, self.num_retries, logger)
777         if current_container:
778             logger.info("Running inside container %s", current_container.get("uuid"))
779
780         self.poll_api = arvados.api('v1', timeout=runtimeContext.http_timeout)
781         self.polling_thread = threading.Thread(target=self.poll_states)
782         self.polling_thread.start()
783
784         self.task_queue = TaskQueue(self.workflow_eval_lock, self.thread_count)
785
786         try:
787             self.workflow_eval_lock.acquire()
788
789             # Holds the lock while this code runs and releases it when
790             # it is safe to do so in self.workflow_eval_lock.wait(),
791             # at which point on_message can update job state and
792             # process output callbacks.
793
794             loopperf = Perf(metrics, "jobiter")
795             loopperf.__enter__()
796             for runnable in jobiter:
797                 loopperf.__exit__()
798
799                 if self.stop_polling.is_set():
800                     break
801
802                 if self.task_queue.error is not None:
803                     raise self.task_queue.error
804
805                 if runnable:
806                     with Perf(metrics, "run"):
807                         self.start_run(runnable, runtimeContext)
808                 else:
809                     if (self.task_queue.in_flight + len(self.processes)) > 0:
810                         self.workflow_eval_lock.wait(3)
811                     else:
812                         logger.error("Workflow is deadlocked, no runnable processes and not waiting on any pending processes.")
813                         break
814
815                 if self.stop_polling.is_set():
816                     break
817
818                 loopperf.__enter__()
819             loopperf.__exit__()
820
821             while (self.task_queue.in_flight + len(self.processes)) > 0:
822                 if self.task_queue.error is not None:
823                     raise self.task_queue.error
824                 self.workflow_eval_lock.wait(3)
825
826         except UnsupportedRequirement:
827             raise
828         except:
829             if sys.exc_info()[0] is KeyboardInterrupt or sys.exc_info()[0] is SystemExit:
830                 logger.error("Interrupted, workflow will be cancelled")
831             elif isinstance(sys.exc_info()[1], WorkflowException):
832                 logger.error("Workflow execution failed:\n%s", sys.exc_info()[1], exc_info=(sys.exc_info()[1] if self.debug else False))
833             else:
834                 logger.exception("Workflow execution failed")
835
836             if self.pipeline:
837                 self.api.pipeline_instances().update(uuid=self.pipeline["uuid"],
838                                                      body={"state": "Failed"}).execute(num_retries=self.num_retries)
839
840             if self.work_api == "containers" and not current_container:
841                 # Not running in a crunch container, so cancel any outstanding processes.
842                 for p in self.processes:
843                     try:
844                         self.api.container_requests().update(uuid=p,
845                                                              body={"priority": "0"}
846                         ).execute(num_retries=self.num_retries)
847                     except Exception:
848                         pass
849         finally:
850             self.workflow_eval_lock.release()
851             self.task_queue.drain()
852             self.stop_polling.set()
853             self.polling_thread.join()
854             self.task_queue.join()
855
856         if self.final_status == "UnsupportedRequirement":
857             raise UnsupportedRequirement("Check log for details.")
858
859         if self.final_output is None:
860             raise WorkflowException("Workflow did not return a result.")
861
862         if runtimeContext.submit and isinstance(tool, Runner):
863             logger.info("Final output collection %s", tool.final_output)
864             if workbench2 or workbench1:
865                 logger.info("Output at %scollections/%s", workbench2 or workbench1, tool.final_output)
866         else:
867             if self.output_tags is None:
868                 self.output_tags = ""
869
870             storage_classes = ""
871             storage_class_req, _ = tool.get_requirement("http://arvados.org/cwl#OutputStorageClass")
872             if storage_class_req and storage_class_req.get("finalStorageClass"):
873                 storage_classes = aslist(storage_class_req["finalStorageClass"])
874             else:
875                 storage_classes = runtimeContext.storage_classes.strip().split(",")
876
877             output_properties = {}
878             output_properties_req, _ = tool.get_requirement("http://arvados.org/cwl#OutputCollectionProperties")
879             if output_properties_req:
880                 builder = make_builder(job_order, tool.hints, tool.requirements, runtimeContext, tool.metadata)
881                 for pr in output_properties_req["outputProperties"]:
882                     output_properties[pr["propertyName"]] = builder.do_eval(pr["propertyValue"])
883
884             self.final_output, self.final_output_collection = self.make_output_collection(self.output_name, storage_classes,
885                                                                                           self.output_tags, output_properties,
886                                                                                           self.final_output)
887             self.set_crunch_output()
888
889         if runtimeContext.compute_checksum:
890             adjustDirObjs(self.final_output, partial(get_listing, self.fs_access))
891             adjustFileObjs(self.final_output, partial(compute_checksums, self.fs_access))
892
893         if self.trash_intermediate and self.final_status == "success":
894             self.trash_intermediate_output()
895
896         return (self.final_output, self.final_status)