X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/3da0fb0eef35b665c9c427fa9acf9b932dd45062..e59c1d365d9b6e1eff9b5cb030a8b1a3aaf14353:/sdk/cwl/arvados_cwl/fsaccess.py

diff --git a/sdk/cwl/arvados_cwl/fsaccess.py b/sdk/cwl/arvados_cwl/fsaccess.py
index 500ea0f420..7736cab4d8 100644
--- a/sdk/cwl/arvados_cwl/fsaccess.py
+++ b/sdk/cwl/arvados_cwl/fsaccess.py
@@ -3,6 +3,8 @@ import os
 import errno
 import urlparse
 import re
+import logging
+import threading
 
 import ruamel.yaml as yaml
 
@@ -13,26 +15,41 @@ import cwltool.resolver
 import arvados.util
 import arvados.collection
 import arvados.arvfile
+import arvados.errors
 
 from schema_salad.ref_resolver import DefaultFetcher
 
-class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
-    """Implement the cwltool FsAccess interface for Arvados Collections."""
+logger = logging.getLogger('arvados.cwl-runner')
 
-    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)
 
@@ -75,7 +92,10 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
     def exists(self, fn):
         collection, rest = self.get_collection(fn)
         if collection:
-            return collection.exists(rest)
+            if rest:
+                return collection.exists(rest)
+            else:
+                return True
         else:
             return super(CollectionFsAccess, self).exists(fn)
 
@@ -129,25 +149,41 @@ class CollectionFsAccess(cwltool.stdfsaccess.StdFsAccess):
             return os.path.realpath(path)
 
 class CollectionFetcher(DefaultFetcher):
-    def __init__(self, cache, session, api_client=None, keep_client=None):
+    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()
         if url.startswith("arvwf:"):
-            return self.api_client.workflows().get(uuid=url[6:]).execute()["definition"]
+            record = self.api_client.workflows().get(uuid=url[6:]).execute(num_retries=self.num_retries)
+            definition = record["definition"] + ('\nlabel: "%s"\n' % record["name"].replace('"', '\\"'))
+            return definition
         return super(CollectionFetcher, self).fetch_text(url)
 
     def check_exists(self, url):
-        if url.startswith("keep:"):
-            return self.fsaccess.exists(url)
-        if url.startswith("arvwf:"):
-            if self.fetch_text(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:"):
+                if self.fetch_text(url):
+                    return True
+        except arvados.errors.NotFoundError:
+            return False
+        except:
+            logger.exception("Got unexpected exception checking if file exists:")
+            return False
         return super(CollectionFetcher, self).check_exists(url)
 
     def urljoin(self, base_url, url):
@@ -186,12 +222,12 @@ class CollectionFetcher(DefaultFetcher):
 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):
+def collectionResolver(api_client, document_loader, uri, num_retries=4):
     if workflow_uuid_pattern.match(uri):
         return "arvwf:%s#main" % (uri)
 
     if pipeline_template_uuid_pattern.match(uri):
-        pt = api_client.pipeline_templates().get(uuid=uri).execute()
+        pt = api_client.pipeline_templates().get(uuid=uri).execute(num_retries=num_retries)
         return "keep:" + pt["components"].values()[0]["script_parameters"]["cwl:tool"]
 
     p = uri.split("/")