1 # Copyright (C) The Arvados Authors. All rights reserved.
3 # SPDX-License-Identifier: Apache-2.0
5 from future import standard_library
6 standard_library.install_aliases()
7 from builtins import str
8 from past.builtins import basestring
9 from future.utils import viewitems
15 import urllib.request, urllib.parse, urllib.error
17 import arvados_cwl.util
18 import arvados.commands.run
19 import arvados.collection
21 from schema_salad.sourceline import SourceLine
23 from arvados.errors import ApiError
24 from cwltool.pathmapper import PathMapper, MapperEnt
25 from cwltool.utils import adjustFileObjs, adjustDirObjs
26 from cwltool.stdfsaccess import abspath
27 from cwltool.workflow import WorkflowException
29 from .http import http_to_keep
31 logger = logging.getLogger('arvados.cwl-runner')
33 def trim_listing(obj):
34 """Remove 'listing' field from Directory objects that are keep references.
36 When Directory objects represent Keep references, it is redundant and
37 potentially very expensive to pass fully enumerated Directory objects
38 between instances of cwl-runner (e.g. a submitting a job, or using the
39 RunInSingleContainer feature), so delete the 'listing' field when it is
44 if obj.get("location", "").startswith("keep:") and "listing" in obj:
47 collection_pdh_path = re.compile(r'^keep:[0-9a-f]{32}\+\d+/.+$')
48 collection_pdh_pattern = re.compile(r'^keep:([0-9a-f]{32}\+\d+)(/.*)?')
49 collection_uuid_pattern = re.compile(r'^keep:([a-z0-9]{5}-4zz18-[a-z0-9]{15})(/.*)?$')
51 class ArvPathMapper(PathMapper):
52 """Convert container-local paths to and from Keep collection ids."""
54 def __init__(self, arvrunner, referenced_files, input_basedir,
55 collection_pattern, file_pattern, name=None, single_collection=False,
57 self.arvrunner = arvrunner
58 self.input_basedir = input_basedir
59 self.collection_pattern = collection_pattern
60 self.file_pattern = file_pattern
62 self.referenced_files = [r["location"] for r in referenced_files]
63 self.single_collection = single_collection
65 self.optional_deps = optional_deps or []
66 super(ArvPathMapper, self).__init__(referenced_files, input_basedir, None)
68 def visit(self, srcobj, uploadfiles):
69 src = srcobj["location"]
71 src = src[:src.index("#")]
73 debug = logger.isEnabledFor(logging.DEBUG)
75 if isinstance(src, basestring) and src.startswith("keep:"):
76 if collection_pdh_pattern.match(src):
77 self._pathmap[src] = MapperEnt(src, self.collection_pattern % urllib.parse.unquote(src[5:]), srcobj["class"], True)
79 if arvados_cwl.util.collectionUUID in srcobj:
80 self.pdh_to_uuid[src.split("/", 1)[0][5:]] = srcobj[arvados_cwl.util.collectionUUID]
81 elif not collection_uuid_pattern.match(src):
82 with SourceLine(srcobj, "location", WorkflowException, debug):
83 raise WorkflowException("Invalid keep reference '%s'" % src)
85 if src not in self._pathmap:
86 if src.startswith("file:"):
87 # Local FS ref, may need to be uploaded or may be on keep
89 ab = abspath(src, self.input_basedir)
90 st = arvados.commands.run.statfile("", ab,
91 fnPattern="keep:%s/%s",
92 dirPattern="keep:%s/%s",
94 with SourceLine(srcobj, "location", WorkflowException, debug):
95 if isinstance(st, arvados.commands.run.UploadFile):
96 uploadfiles.add((src, ab, st))
97 elif isinstance(st, arvados.commands.run.ArvFile):
98 self._pathmap[src] = MapperEnt(st.fn, self.collection_pattern % urllib.parse.unquote(st.fn[5:]), "File", True)
100 raise WorkflowException("Input file path '%s' is invalid" % st)
101 elif src.startswith("_:"):
102 if srcobj["class"] == "File" and "contents" not in srcobj:
103 raise WorkflowException("File literal '%s' is missing `contents`" % src)
104 if srcobj["class"] == "Directory" and "listing" not in srcobj:
105 raise WorkflowException("Directory literal '%s' is missing `listing`" % src)
106 elif src.startswith("http:") or src.startswith("https:"):
108 if self.arvrunner.defer_downloads:
109 # passthrough, we'll download it later.
110 self._pathmap[src] = MapperEnt(src, src, srcobj["class"], True)
112 keepref = http_to_keep(self.arvrunner.api, self.arvrunner.project_uuid, src)
113 logger.info("%s is %s", src, keepref)
114 self._pathmap[src] = MapperEnt(keepref, keepref, srcobj["class"], True)
115 except Exception as e:
116 logger.warning(str(e))
118 self._pathmap[src] = MapperEnt(src, src, srcobj["class"], True)
120 with SourceLine(srcobj, "secondaryFiles", WorkflowException, debug):
121 for l in srcobj.get("secondaryFiles", []):
122 self.visit(l, uploadfiles)
123 with SourceLine(srcobj, "listing", WorkflowException, debug):
124 for l in srcobj.get("listing", []):
125 self.visit(l, uploadfiles)
127 def addentry(self, obj, c, path, remap):
128 if obj["location"] in self._pathmap:
129 src, srcpath = self.arvrunner.fs_access.get_collection(self._pathmap[obj["location"]].resolved)
132 c.copy(srcpath, path + "/" + obj["basename"], source_collection=src, overwrite=True)
133 remap.append((obj["location"], path + "/" + obj["basename"]))
134 for l in obj.get("secondaryFiles", []):
135 self.addentry(l, c, path, remap)
136 elif obj["class"] == "Directory":
137 for l in obj.get("listing", []):
138 self.addentry(l, c, path + "/" + obj["basename"], remap)
139 remap.append((obj["location"], path + "/" + obj["basename"]))
140 elif obj["location"].startswith("_:") and "contents" in obj:
141 with c.open(path + "/" + obj["basename"], "w") as f:
142 f.write(obj["contents"])
143 remap.append((obj["location"], path + "/" + obj["basename"]))
145 for opt in self.optional_deps:
146 if obj["location"] == opt["location"]:
148 raise SourceLine(obj, "location", WorkflowException).makeError("Don't know what to do with '%s'" % obj["location"])
150 def needs_new_collection(self, srcobj, prefix=""):
151 """Check if files need to be staged into a new collection.
153 If all the files are in the same collection and in the same
154 paths they would be staged to, return False. Otherwise, a new
155 collection is needed with files copied/created in the
159 loc = srcobj["location"]
160 if loc.startswith("_:"):
165 loc_prefix = loc[:i+1]
168 # quote/unquote to ensure consistent quoting
169 suffix = urllib.parse.quote(urllib.parse.unquote(loc[i+1:]), "/+@")
176 if prefix != loc_prefix:
179 if "basename" in srcobj and suffix != urllib.parse.quote(srcobj["basename"], "/+@"):
182 if srcobj["class"] == "File" and loc not in self._pathmap:
184 for s in srcobj.get("secondaryFiles", []):
185 if self.needs_new_collection(s, prefix):
187 if srcobj.get("listing"):
188 prefix = "%s%s/" % (prefix, urllib.parse.quote(srcobj.get("basename", suffix), "/+@"))
189 for l in srcobj["listing"]:
190 if self.needs_new_collection(l, prefix):
194 def setup(self, referenced_files, basedir):
195 # type: (List[Any], unicode) -> None
199 if self.single_collection:
200 collection = arvados.collection.Collection(api_client=self.arvrunner.api,
201 keep_client=self.arvrunner.keep_client,
202 num_retries=self.arvrunner.num_retries)
204 for srcobj in referenced_files:
205 self.visit(srcobj, uploadfiles)
207 arvados.commands.run.uploadfiles([u[2] for u in uploadfiles],
210 num_retries=self.arvrunner.num_retries,
211 fnPattern="keep:%s/%s",
213 project=self.arvrunner.project_uuid,
214 collection=collection,
217 for src, ab, st in uploadfiles:
218 self._pathmap[src] = MapperEnt(urllib.parse.quote(st.fn, "/:+@"), urllib.parse.quote(self.collection_pattern % st.fn[5:], "/:+@"),
219 "Directory" if os.path.isdir(ab) else "File", True)
221 for srcobj in referenced_files:
223 if srcobj["class"] == "Directory" and srcobj["location"] not in self._pathmap:
224 c = arvados.collection.Collection(api_client=self.arvrunner.api,
225 keep_client=self.arvrunner.keep_client,
226 num_retries=self.arvrunner.num_retries)
227 for l in srcobj.get("listing", []):
228 self.addentry(l, c, ".", remap)
230 container = arvados_cwl.util.get_current_container(self.arvrunner.api, self.arvrunner.num_retries, logger)
231 info = arvados_cwl.util.get_intermediate_collection_info(None, container, self.arvrunner.intermediate_output_ttl)
233 c.save_new(name=info["name"],
234 owner_uuid=self.arvrunner.project_uuid,
235 ensure_unique_name=True,
236 trash_at=info["trash_at"],
237 properties=info["properties"])
239 ab = self.collection_pattern % c.portable_data_hash()
240 self._pathmap[srcobj["location"]] = MapperEnt("keep:"+c.portable_data_hash(), ab, "Directory", True)
241 elif srcobj["class"] == "File" and self.needs_new_collection(srcobj):
242 c = arvados.collection.Collection(api_client=self.arvrunner.api,
243 keep_client=self.arvrunner.keep_client,
244 num_retries=self.arvrunner.num_retries)
245 self.addentry(srcobj, c, ".", remap)
247 container = arvados_cwl.util.get_current_container(self.arvrunner.api, self.arvrunner.num_retries, logger)
248 info = arvados_cwl.util.get_intermediate_collection_info(None, container, self.arvrunner.intermediate_output_ttl)
250 c.save_new(name=info["name"],
251 owner_uuid=self.arvrunner.project_uuid,
252 ensure_unique_name=True,
253 trash_at=info["trash_at"],
254 properties=info["properties"])
256 ab = self.file_pattern % (c.portable_data_hash(), srcobj["basename"])
257 self._pathmap[srcobj["location"]] = MapperEnt("keep:%s/%s" % (c.portable_data_hash(), srcobj["basename"]),
259 if srcobj.get("secondaryFiles"):
260 ab = self.collection_pattern % c.portable_data_hash()
261 self._pathmap["_:" + str(uuid.uuid4())] = MapperEnt("keep:"+c.portable_data_hash(), ab, "Directory", True)
264 for loc, sub in remap:
265 # subdirs start with "./", strip it off
266 if sub.startswith("./"):
267 ab = self.file_pattern % (c.portable_data_hash(), sub[2:])
269 ab = self.file_pattern % (c.portable_data_hash(), sub)
270 self._pathmap[loc] = MapperEnt("keep:%s/%s" % (c.portable_data_hash(), sub[2:]),
271 ab, "Directory", True)
275 def reversemap(self, target):
276 p = super(ArvPathMapper, self).reversemap(target)
279 elif target.startswith("keep:"):
280 return (target, target)
281 elif self.keepdir and target.startswith(self.keepdir):
282 kp = "keep:" + target[len(self.keepdir)+1:]
288 class StagingPathMapper(PathMapper):
289 # Note that StagingPathMapper internally maps files from target to source.
290 # Specifically, the 'self._pathmap' dict keys are the target location and the
291 # values are 'MapperEnt' named tuples from which we use the 'resolved' attribute
292 # as the file identifier. This makes it possible to map an input file to multiple
293 # target directories. The exception is for file literals, which store the contents of
294 # the file in 'MapperEnt.resolved' and are therefore still mapped from source to target.
298 def __init__(self, referenced_files, basedir, stagedir, separateDirs=True):
300 super(StagingPathMapper, self).__init__(referenced_files, basedir, stagedir, separateDirs)
302 def visit(self, obj, stagedir, basedir, copy=False, staged=False):
303 # type: (Dict[unicode, Any], unicode, unicode, bool) -> None
304 loc = obj["location"]
305 stagedir = obj.get("dirname") or stagedir
306 tgt = os.path.join(stagedir, obj["basename"])
307 basetgt, baseext = os.path.splitext(tgt)
310 return tgt in self.targets and ("contents" not in obj) and (self._pathmap[tgt].resolved != loc)
311 def literalTargetExists():
312 return tgt in self.targets and "contents" in obj
315 if targetExists() or literalTargetExists():
316 while tgt in self.targets:
318 tgt = "%s_%i%s" % (basetgt, n, baseext)
319 self.targets.add(tgt)
320 if obj["class"] == "Directory":
321 if obj.get("writable"):
322 self._pathmap[tgt] = MapperEnt(loc, tgt, "WritableDirectory", staged)
324 self._pathmap[tgt] = MapperEnt(loc, tgt, "Directory", staged)
325 if loc.startswith("_:") or self._follow_dirs:
326 self.visitlisting(obj.get("listing", []), tgt, basedir)
327 elif obj["class"] == "File":
328 if tgt in self._pathmap:
330 if "contents" in obj and loc.startswith("_:"):
331 self._pathmap[loc] = MapperEnt(obj["contents"], tgt, "CreateFile", staged)
333 if copy or obj.get("writable"):
334 self._pathmap[tgt] = MapperEnt(loc, tgt, "WritableFile", staged)
336 self._pathmap[tgt] = MapperEnt(loc, tgt, "File", staged)
337 self.visitlisting(obj.get("secondaryFiles", []), stagedir, basedir)
339 def mapper(self, src): # type: (Text) -> MapperEnt.
340 # Overridden to maintain the use case of mapping by source (identifier) to
341 # target regardless of how the map is structured interally.
342 def getMapperEnt(src):
343 for k,v in viewitems(self._pathmap):
344 if (v.type != "CreateFile" and v.resolved == src) or (v.type == "CreateFile" and k == src):
349 v = getMapperEnt(src[i:])
350 return MapperEnt(v.resolved, v.target + src[i:], v.type, v.staged)
351 return getMapperEnt(src)
354 class VwdPathMapper(StagingPathMapper):
355 def setup(self, referenced_files, basedir):
356 # type: (List[Any], unicode) -> None
358 # Go through each file and set the target to its own directory along
359 # with any secondary files.
360 self.visitlisting(referenced_files, self.stagedir, basedir)
362 for path, (ab, tgt, type, staged) in viewitems(self._pathmap):
363 if type in ("File", "Directory") and ab.startswith("keep:"):
364 self._pathmap[path] = MapperEnt("$(task.keep)/%s" % ab[5:], tgt, type, staged)
367 class NoFollowPathMapper(StagingPathMapper):
369 def setup(self, referenced_files, basedir):
370 # type: (List[Any], unicode) -> None
371 self.visitlisting(referenced_files, self.stagedir, basedir)