X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/b8de9b3e62e82b806576b237be5f317bf378169f..0bb435a47e427b12fa2351141a22a1ba1e28a49d:/sdk/cwl/arvados_cwl/fsaccess.py diff --git a/sdk/cwl/arvados_cwl/fsaccess.py b/sdk/cwl/arvados_cwl/fsaccess.py index a99b2a7414..5bdd8dd085 100644 --- a/sdk/cwl/arvados_cwl/fsaccess.py +++ b/sdk/cwl/arvados_cwl/fsaccess.py @@ -1,9 +1,14 @@ +# Copyright (C) The Arvados Authors. All rights reserved. +# +# SPDX-License-Identifier: Apache-2.0 + import fnmatch import os import errno import urlparse import re import logging +import threading import ruamel.yaml as yaml @@ -20,23 +25,35 @@ from schema_salad.ref_resolver import DefaultFetcher logger = logging.getLogger('arvados.cwl-runner') -class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): - """Implement the cwltool FsAccess interface for Arvados Collections.""" - - def __init__(self, basedir, api_client=None, keep_client=None): - super(CollectionFsAccess, self).__init__(basedir) +class CollectionCache(object): + def __init__(self, api_client, keep_client, num_retries): self.api_client = api_client self.keep_client = keep_client self.collections = {} + self.lock = threading.Lock() - def get_collection(self, path): - p = path.split("/") - if p[0].startswith("keep:") and arvados.util.keep_locator_pattern.match(p[0][5:]): - pdh = p[0][5:] + def get(self, pdh): + with self.lock: if pdh not in self.collections: + logger.debug("Creating collection reader for %s", pdh) self.collections[pdh] = arvados.collection.CollectionReader(pdh, api_client=self.api_client, keep_client=self.keep_client) - return (self.collections[pdh], "/".join(p[1:])) + return self.collections[pdh] + + +class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): + """Implement the cwltool FsAccess interface for Arvados Collections.""" + + def __init__(self, basedir, collection_cache=None): + super(CollectionFsAccess, self).__init__(basedir) + self.collection_cache = collection_cache + + def get_collection(self, path): + sp = path.split("/", 1) + p = sp[0] + if p.startswith("keep:") and arvados.util.keep_locator_pattern.match(p[5:]): + pdh = p[5:] + return (self.collection_cache.get(pdh), sp[1] if len(sp) == 2 else None) else: return (None, path) @@ -64,28 +81,31 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): def glob(self, pattern): collection, rest = self.get_collection(pattern) - if collection and not rest: + if collection is not None and not rest: return [pattern] patternsegments = rest.split("/") return self._match(collection, patternsegments, "keep:" + collection.manifest_locator()) def open(self, fn, mode): collection, rest = self.get_collection(fn) - if collection: + if collection is not None: return collection.open(rest, mode) else: return super(CollectionFsAccess, self).open(self._abs(fn), mode) def exists(self, fn): collection, rest = self.get_collection(fn) - if collection: - return collection.exists(rest) + if collection is not None: + if rest: + return collection.exists(rest) + else: + return True else: return super(CollectionFsAccess, self).exists(fn) def isfile(self, fn): # type: (unicode) -> bool collection, rest = self.get_collection(fn) - if collection: + if collection is not None: if rest: return isinstance(collection.find(rest), arvados.arvfile.ArvadosFile) else: @@ -95,7 +115,7 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): def isdir(self, fn): # type: (unicode) -> bool collection, rest = self.get_collection(fn) - if collection: + if collection is not None: if rest: return isinstance(collection.find(rest), arvados.collection.RichCollectionBase) else: @@ -105,7 +125,7 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): def listdir(self, fn): # type: (unicode) -> List[unicode] collection, rest = self.get_collection(fn) - if collection: + if collection is not None: if rest: dir = collection.find(rest) else: @@ -127,19 +147,22 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess): if path.startswith("$(task.tmpdir)") or path.startswith("$(task.outdir)"): return path collection, rest = self.get_collection(path) - if collection: + if collection is not None: return path else: return os.path.realpath(path) class CollectionFetcher(DefaultFetcher): - def __init__(self, cache, session, api_client=None, keep_client=None, num_retries=4): + def __init__(self, cache, session, api_client=None, fs_access=None, num_retries=4, overrides=None): super(CollectionFetcher, self).__init__(cache, session) self.api_client = api_client - self.fsaccess = CollectionFsAccess("", api_client=api_client, keep_client=keep_client) + self.fsaccess = fs_access self.num_retries = num_retries + self.overrides = overrides if overrides else {} def fetch_text(self, url): + if url in self.overrides: + return self.overrides[url] if url.startswith("keep:"): with self.fsaccess.open(url, "r") as f: return f.read() @@ -150,7 +173,11 @@ class CollectionFetcher(DefaultFetcher): return super(CollectionFetcher, self).fetch_text(url) def check_exists(self, url): + if url in self.overrides: + return True try: + if url.startswith("http://arvados.org/cwl"): + return True if url.startswith("keep:"): return self.fsaccess.exists(url) if url.startswith("arvwf:"): @@ -200,6 +227,9 @@ workflow_uuid_pattern = re.compile(r'[a-z0-9]{5}-7fd4e-[a-z0-9]{15}') pipeline_template_uuid_pattern = re.compile(r'[a-z0-9]{5}-p5p6p-[a-z0-9]{15}') def collectionResolver(api_client, document_loader, uri, num_retries=4): + if uri.startswith("keep:") or uri.startswith("arvwf:"): + return uri + if workflow_uuid_pattern.match(uri): return "arvwf:%s#main" % (uri)