Merge pull request #354 from charz/fixed_object_store_path

Fixed object store path ('job_work')

config/object_store_conf.xml.sample

@@ -23,9 +23,11 @@
 
         <!--  Sample S3 Object Store
         <object_store type="s3">
-            <auth access_key="...." secret_key="....." />
-            <bucket name="unique_bucket_name_all_lowercase" use_reduced_redundancy="False" />
-            <cache path="database/files/" size="1000" />
+             <auth access_key="...." secret_key="....." />
+             <bucket name="unique_bucket_name_all_lowercase" use_reduced_redundancy="False" />
+             <cache path="database/object_store_cache" size="1000" />
+             <extra_dir type="job_work" path="database/job_working_directory_s3"/>
+             <extra_dir type="temp" path="database/tmp_s3"/>
         </object_store>
         -->
 
@@ -34,7 +36,9 @@
              <auth access_key="...." secret_key="....." />
              <bucket name="unique_bucket_name" use_reduced_redundancy="False" max_chunk_size="250"/>
              <connection host="" port="" is_secure="" conn_path="" multipart="True"/>
-             <cache path="database/files/" size="1000" />
+             <cache path="database/object_store_cache" size="1000" />
+             <extra_dir type="job_work" path="database/job_working_directory_swift"/>
+             <extra_dir type="temp" path="database/tmp_swift"/>
          </object_store>
          -->
 

lib/galaxy/objectstore/__init__.py

@@ -208,7 +208,7 @@ class DiskObjectStore(ObjectStore):
                 if e.tag == 'files_dir':
                     self.file_path = e.get('path')
                 else:
-                    self.extra_dirs[e.tag] = e.get('path')
+                    self.extra_dirs[e.get('type')] = e.get('path')
         if extra_dirs is not None:
             self.extra_dirs.update( extra_dirs )
 

lib/galaxy/objectstore/s3.py

@@ -92,6 +92,14 @@ class S3ObjectStore(ObjectStore):
             self.conn_path = cn_xml.get('conn_path', '/')
             c_xml = config_xml.findall('cache')[0]
             self.cache_size = float(c_xml.get('size', -1))
+            self.staging_path = c_xml.get('path', self.config.object_store_cache_path)
+
+            for d_xml in config_xml.findall('extra_dir'):
+                self.extra_dirs[d_xml.get('type')] = d_xml.get('path')
+
+            log.debug("Object cache dir:    %s", self.staging_path)
+            log.debug("       job work dir: %s", self.extra_dirs['job_work'])
+
             # for multipart upload
             self.s3server = {'access_key': self.access_key,
                              'secret_key': self.secret_key,
@@ -200,15 +208,24 @@ class S3ObjectStore(ObjectStore):
                     continue
                 umask_fix_perms( path, self.config.umask, 0666, self.config.gid )
 
-    def _construct_path(self, obj, dir_only=None, extra_dir=None, extra_dir_at_root=False, alt_name=None, **kwargs):
+    def _construct_path(self, obj, base_dir=None, dir_only=None, extra_dir=None, extra_dir_at_root=False, alt_name=None, obj_dir=False, **kwargs):
         rel_path = os.path.join(*directory_hash_id(obj.id))
         if extra_dir is not None:
             if extra_dir_at_root:
                 rel_path = os.path.join(extra_dir, rel_path)
             else:
                 rel_path = os.path.join(rel_path, extra_dir)
+
+        # for JOB_WORK directory
+        if obj_dir:
+            rel_path = os.path.join(rel_path, str(obj.id))
+        if base_dir:
+            base = self.extra_dirs.get(base_dir)
+            return os.path.join(base, rel_path)
+
         # S3 folders are marked by having trailing '/' so add it now
         rel_path = '%s/' % rel_path
+
         if not dir_only:
             rel_path = os.path.join(rel_path, alt_name if alt_name else "dataset_%s.dat" % obj.id)
         return rel_path
@@ -374,6 +391,7 @@ class S3ObjectStore(ObjectStore):
     def exists(self, obj, **kwargs):
         in_cache = in_s3 = False
         rel_path = self._construct_path(obj, **kwargs)
+
         # Check cache
         if self._in_cache(rel_path):
             in_cache = True
@@ -382,11 +400,18 @@ class S3ObjectStore(ObjectStore):
         # log.debug("~~~~~~ File '%s' exists in cache: %s; in s3: %s" % (rel_path, in_cache, in_s3))
         # dir_only does not get synced so shortcut the decision
         dir_only = kwargs.get('dir_only', False)
+        base_dir = kwargs.get('base_dir', None)
         if dir_only:
             if in_cache or in_s3:
                 return True
+            # for JOB_WORK directory
+            elif base_dir:
+                if not os.path.exists(rel_path):
+                    os.makedirs(rel_path)
+                return True
             else:
                 return False
+
         # TODO: Sync should probably not be done here. Add this to an async upload stack?
         if in_cache and not in_s3:
             self._push_to_os(rel_path, source_file=self._get_cache_path(rel_path))
@@ -398,11 +423,13 @@ class S3ObjectStore(ObjectStore):
 
     def create(self, obj, **kwargs):
         if not self.exists(obj, **kwargs):
+
             # Pull out locally used fields
             extra_dir = kwargs.get('extra_dir', None)
             extra_dir_at_root = kwargs.get('extra_dir_at_root', False)
             dir_only = kwargs.get('dir_only', False)
             alt_name = kwargs.get('alt_name', None)
+
             # Construct hashed path
             rel_path = os.path.join(*directory_hash_id(obj.id))
 
@@ -412,10 +439,12 @@ class S3ObjectStore(ObjectStore):
                     rel_path = os.path.join(extra_dir, rel_path)
                 else:
                     rel_path = os.path.join(rel_path, extra_dir)
+
             # Create given directory in cache
             cache_dir = os.path.join(self.staging_path, rel_path)
             if not os.path.exists(cache_dir):
                 os.makedirs(cache_dir)
+
             # Although not really necessary to create S3 folders (because S3 has
             # flat namespace), do so for consistency with the regular file system
             # S3 folders are marked by having trailing '/' so add it now
@@ -449,7 +478,15 @@ class S3ObjectStore(ObjectStore):
     def delete(self, obj, entire_dir=False, **kwargs):
         rel_path = self._construct_path(obj, **kwargs)
         extra_dir = kwargs.get('extra_dir', None)
+        base_dir = kwargs.get('base_dir', None)
+        dir_only = kwargs.get('dir_only', False)
+        obj_dir = kwargs.get('obj_dir', False)
         try:
+            # Remove temparory data in JOB_WORK directory
+            if base_dir and dir_only and obj_dir:
+                shutil.rmtree(os.path.abspath(rel_path))
+                return True
+
             # For the case of extra_files, because we don't have a reference to
             # individual files/keys we need to remove the entire directory structure
             # with all the files in it. This is easy for the local file system,
@@ -489,8 +526,15 @@ class S3ObjectStore(ObjectStore):
         return content
 
     def get_filename(self, obj, **kwargs):
+        base_dir = kwargs.get('base_dir', None)
         dir_only = kwargs.get('dir_only', False)
+        obj_dir = kwargs.get('obj_dir', False)
         rel_path = self._construct_path(obj, **kwargs)
+
+        # for JOB_WORK directory
+        if base_dir and dir_only and obj_dir:
+            return os.path.abspath(rel_path)
+
         cache_path = self._get_cache_path(rel_path)
         # S3 does not recognize directories as files so cannot check if those exist.
         # So, if checking dir only, ensure given dir exists in cache and return