Merge branch '13399-lock-starvation'

[arvados.git] / sdk / cwl / arvados_cwl / pathmapper.py

diff --git a/sdk/cwl/arvados_cwl/pathmapper.py b/sdk/cwl/arvados_cwl/pathmapper.py
index a749b51625b3af008e18ac4a727ca461c78a8712..d083b78f5a061906164a5978530af9230e767473 100644 (file)
--- a/sdk/cwl/arvados_cwl/pathmapper.py
+++ b/sdk/cwl/arvados_cwl/pathmapper.py
@@ -7,8 +7,8 @@ import logging
 import uuid
 import os
 import urllib
-import datetime
 
+from arvados_cwl.util import get_current_container, get_intermediate_collection_info
 import arvados.commands.run
 import arvados.collection
 
@@ -44,7 +44,7 @@ class ArvPathMapper(PathMapper):
     pdh_dirpath = re.compile(r'^keep:[0-9a-f]{32}\+\d+(/.*)?$')
 
     def __init__(self, arvrunner, referenced_files, input_basedir,
-                 collection_pattern, file_pattern, name=None, single_collection=False, **kwargs):
+                 collection_pattern, file_pattern, name=None, single_collection=False):
         self.arvrunner = arvrunner
         self.input_basedir = input_basedir
         self.collection_pattern = collection_pattern
@@ -155,13 +155,14 @@ class ArvPathMapper(PathMapper):
                 for l in srcobj.get("listing", []):
                     self.addentry(l, c, ".", remap)
 
-                trash_time, props = self.__get_collection_attributes()
+                container = get_current_container(self.arvrunner.api, self.arvrunner.num_retries, logger)
+                info = get_intermediate_collection_info(None, container, self.arvrunner.intermediate_output_ttl)
 
-                c.save_new(name="Intermediate collection", 
-                           owner_uuid=self.arvrunner.project_uuid, 
-                           ensure_unique_name=True, 
-                           trash_at=trash_time, 
-                           properties=props)
+                c.save_new(name=info["name"],
+                           owner_uuid=self.arvrunner.project_uuid,
+                           ensure_unique_name=True,
+                           trash_at=info["trash_at"],
+                           properties=info["properties"])
 
                 ab = self.collection_pattern % c.portable_data_hash()
                 self._pathmap[srcobj["location"]] = MapperEnt("keep:"+c.portable_data_hash(), ab, "Directory", True)
@@ -173,13 +174,14 @@ class ArvPathMapper(PathMapper):
                                                   num_retries=self.arvrunner.num_retries                                                  )
                 self.addentry(srcobj, c, ".", remap)
 
-                trash_time, props = self.__get_collection_attributes()
+                container = get_current_container(self.arvrunner.api, self.arvrunner.num_retries, logger)
+                info = get_intermediate_collection_info(None, container, self.arvrunner.intermediate_output_ttl)
 
-                c.save_new(name="Intermediate collection", 
-                           owner_uuid=self.arvrunner.project_uuid, 
-                           ensure_unique_name=True, 
-                           trash_at=trash_time, 
-                           properties=props)
+                c.save_new(name=info["name"],
+                           owner_uuid=self.arvrunner.project_uuid,
+                           ensure_unique_name=True,
+                           trash_at=info["trash_at"],
+                           properties=info["properties"])
 
                 ab = self.file_pattern % (c.portable_data_hash(), srcobj["basename"])
                 self._pathmap[srcobj["location"]] = MapperEnt("keep:%s/%s" % (c.portable_data_hash(), srcobj["basename"]),
@@ -212,24 +214,6 @@ class ArvPathMapper(PathMapper):
         else:
             return None
 
-    def __get_collection_attributes(self):
-            trash_time = None 
-            if self.arvrunner.intermediate_output_ttl > 0: 
-                trash_time = datetime.datetime.now() + datetime.timedelta(seconds=self.arvrunner.intermediate_output_ttl) 
-
-            current_container_uuid = None 
-            try: 
-                current_container = self.arvrunner.api.containers().current().execute(num_retries=self.arvrunner.num_retries) 
-                current_container_uuid = current_container['uuid'] 
-            except ApiError as e: 
-                # Status code 404 just means we're not running in a container. 
-                if e.resp.status != 404: 
-                    logger.info("Getting current container: %s", e)
-            properties = {"type": "Intermediate", 
-                          "container": current_container_uuid}
-
-            return (trash_time, properties)
-
 
 class StagingPathMapper(PathMapper):
     _follow_dirs = True