opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-10-25 00:12:08 +08:00
Code Issues Packages Projects Releases Wiki Activity

Style standardization in s3objectstore, a couple fixes.

Browse Source
This commit is contained in:
Dannon Baker
2015-02-10 14:38:01 -05:00
parent 59d73090fe
commit 1880f94c26
1 changed files with 60 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/objectstore/s3.py
+60 -65
View File
@@ -13,8 +13,7 @@ import time
from datetime import datetime
from galaxy.exceptions import ObjectNotFound
from galaxy.util import umask_fix_perms, string_as_bool
from galaxy.util import string_as_bool
from galaxy.util import string_as_bool, umask_fix_perms
from galaxy.util.directory_hash import directory_hash_id
from galaxy.util.sleeper import Sleeper
from .s3_multipart_upload import multipart_upload
@@ -28,7 +27,8 @@ try:
except ImportError:
boto = None
NO_BOTO_ERROR_MESSAGE = "S3/Swift object store configured, but no boto dependency available. Please install and properly configure boto or modify object store configuration."
NO_BOTO_ERROR_MESSAGE = ("S3/Swift object store configured, but no boto dependency available."
"Please install and properly configure boto or modify object store configuration.")
log = logging.getLogger( __name__ )
logging.getLogger('boto').setLevel(logging.INFO) # Otherwise boto is quite noisy
@@ -91,8 +91,6 @@ 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.cache_path = c_xml.get('path')
# for multipart upload
self.s3server = {'access_key': self.access_key,
'secret_key': self.secret_key,
@@ -102,8 +100,6 @@ class S3ObjectStore(ObjectStore):
'port': self.port,
'use_rr': self.use_rr,
'conn_path': self.conn_path}
except Exception:
# Toss it back up after logging, we can't continue loading at this point.
log.exception("Malformed ObjectStore Configuration XML -- unable to continue")
@@ -115,23 +111,23 @@ class S3ObjectStore(ObjectStore):
total_size = 0
# Is this going to be too expensive of an operation to be done frequently?
file_list = []
for dirpath, dirnames, filenames in os.walk(self.staging_path):
for f in filenames:
fp = os.path.join(dirpath, f)
file_size = os.path.getsize(fp)
for dirpath, _, filenames in os.walk(self.staging_path):
for filename in filenames:
filepath = os.path.join(dirpath, filename)
file_size = os.path.getsize(filepath)
total_size += file_size
# Get the time given file was last accessed
last_access_time = time.localtime(os.stat(fp)[7])
last_access_time = time.localtime(os.stat(filepath)[7])
# Compose a tuple of the access time and the file path
file_tuple = last_access_time, fp, file_size
file_tuple = last_access_time, filepath, file_size
file_list.append(file_tuple)
# Sort the file list (based on access time)
file_list.sort()
# Initiate cleaning once within 10% of the defined cache size?
cache_limit = self.cache_size * 0.9
if total_size > cache_limit:
log.info("Initiating cache cleaning: current cache size: %s; clean until smaller than: %s"
% (convert_bytes(total_size), convert_bytes(cache_limit)))
log.info("Initiating cache cleaning: current cache size: %s; clean until smaller than: %s",
convert_bytes(total_size), convert_bytes(cache_limit))
# How much to delete? If simply deleting up to the cache-10% limit,
# is likely to be deleting frequently and may run the risk of hitting
# the limit - maybe delete additional #%?
@@ -159,10 +155,10 @@ class S3ObjectStore(ObjectStore):
# exceed delete_this_much; start deleting from the front of the file list,
# which assumes the oldest files come first on the list.
deleted_amount = 0
for i, f in enumerate(file_list):
for entry in enumerate(file_list):
if deleted_amount < delete_this_much:
deleted_amount += f[2]
os.remove(f[1])
deleted_amount += entry[2]
os.remove(entry[1])
# Debugging code for printing deleted files' stats
# folder, file_name = os.path.split(f[1])
# file_date = time.strftime("%m/%d/%y %H:%M:%S", f[0])
@@ -170,7 +166,7 @@ class S3ObjectStore(ObjectStore):
# % (i, file_name, convert_bytes(f[2]), file_date, \
# convert_bytes(deleted_amount), convert_bytes(delete_this_much)))
else:
log.debug("Cache cleaning done. Total space freed: %s" % convert_bytes(deleted_amount))
log.debug("Cache cleaning done. Total space freed: %s", convert_bytes(deleted_amount))
return
def _get_bucket(self, bucket_name):
@@ -179,14 +175,14 @@ class S3ObjectStore(ObjectStore):
for i in range(5):
try:
bucket = self.conn.get_bucket(bucket_name)
log.debug("Using cloud object store with bucket '%s'" % bucket.name)
log.debug("Using cloud object store with bucket '%s'", bucket.name)
return bucket
except S3ResponseError:
try:
log.debug("Bucket not found, creating s3 bucket with handle '%s'" % bucket_name)
log.debug("Bucket not found, creating s3 bucket with handle '%s'", bucket_name)
self.conn.create_bucket(bucket_name)
except S3ResponseError:
log.exception("Could not get bucket '%s', attempt %s/5" % (bucket_name, i + 1))
log.exception("Could not get bucket '%s', attempt %s/5", bucket_name, i + 1)
time.sleep(2)
# All the attempts have been exhausted and connection was not established,
# raise error
@@ -194,10 +190,10 @@ class S3ObjectStore(ObjectStore):
def _fix_permissions(self, rel_path):
""" Set permissions on rel_path"""
for basedir, dirs, files in os.walk(rel_path):
for basedir, _, files in os.walk(rel_path):
umask_fix_perms(basedir, self.config.umask, 0777, self.config.gid)
for f in files:
path = os.path.join(basedir, f)
for filename in files:
path = os.path.join(basedir, filename)
# Ignore symlinks
if os.path.islink(path):
continue
@@ -228,10 +224,8 @@ class S3ObjectStore(ObjectStore):
if key:
return key.size
except S3ResponseError, ex:
log.error("Could not get size of key '%s' from S3: %s" % (rel_path, ex))
except Exception, ex:
log.error("Could not get reference to the key object '%s'; returning -1 for key size: %s" % (rel_path, ex))
return -1
log.error("Could not get size of key '%s' from S3: %s", rel_path, ex)
return -1
def _key_exists(self, rel_path):
exists = False
@@ -239,8 +233,8 @@ class S3ObjectStore(ObjectStore):
# A hackish way of testing if the rel_path is a folder vs a file
is_dir = rel_path[-1] == '/'
if is_dir:
rs = self.bucket.get_all_keys(prefix=rel_path)
if len(rs) > 0:
keyresult = self.bucket.get_all_keys(prefix=rel_path)
if len(keyresult) > 0:
exists = True
else:
exists = False
@@ -248,7 +242,7 @@ class S3ObjectStore(ObjectStore):
key = Key(self.bucket, rel_path)
exists = key.exists()
except S3ResponseError, ex:
log.error("Trouble checking existence of S3 key '%s': %s" % (rel_path, ex))
log.error("Trouble checking existence of S3 key '%s': %s", rel_path, ex)
return False
if rel_path[0] == '/':
raise
@@ -289,36 +283,36 @@ class S3ObjectStore(ObjectStore):
if not os.path.exists(self._get_cache_path(rel_path_dir)):
os.makedirs(self._get_cache_path(rel_path_dir))
# Now pull in the file
ok = self._download(rel_path)
file_ok = self._download(rel_path)
self._fix_permissions(self._get_cache_path(rel_path_dir))
return ok
return file_ok
def _transfer_cb(self, complete, total):
self.transfer_progress += 10
def _download(self, rel_path):
try:
log.debug("Pulling key '%s' into cache to %s" % (rel_path, self._get_cache_path(rel_path)))
log.debug("Pulling key '%s' into cache to %s", rel_path, self._get_cache_path(rel_path))
key = self.bucket.get_key(rel_path)
# Test if cache is large enough to hold the new file
if self.cache_size > 0 and key.size > self.cache_size:
log.critical("File %s is larger (%s) than the cache size (%s). Cannot download."
% (rel_path, key.size, self.cache_size))
log.critical("File %s is larger (%s) than the cache size (%s). Cannot download.",
rel_path, key.size, self.cache_size)
return False
if self.use_axel:
log.debug("Parallel pulled key '%s' into cache to %s" % (rel_path, self._get_cache_path(rel_path)))
log.debug("Parallel pulled key '%s' into cache to %s", rel_path, self._get_cache_path(rel_path))
ncores = multiprocessing.cpu_count()
url = key.generate_url(7200)
ret_code = subprocess.call("axel -a -n %s '%s'" % (ncores, url))
if ret_code == 0:
return True
else:
log.debug("Pulled key '%s' into cache to %s" % (rel_path, self._get_cache_path(rel_path)))
log.debug("Pulled key '%s' into cache to %s", rel_path, self._get_cache_path(rel_path))
self.transfer_progress = 0 # Reset transfer progress counter
key.get_contents_to_filename(self._get_cache_path(rel_path), cb=self._transfer_cb, num_cb=10)
return True
except S3ResponseError, ex:
log.error("Problem downloading key '%s' from S3 bucket '%s': %s" % (rel_path, self.bucket.name, ex))
except S3ResponseError:
log.exception("Problem downloading key '%s' from S3 bucket '%s'", rel_path, self.bucket.name)
return False
def _push_to_os(self, rel_path, source_file=None, from_string=None):
@@ -334,14 +328,14 @@ class S3ObjectStore(ObjectStore):
if os.path.exists(source_file):
key = Key(self.bucket, rel_path)
if os.path.getsize(source_file) == 0 and key.exists():
log.debug("Wanted to push file '%s' to S3 key '%s' but its size is 0; skipping." % (source_file, rel_path))
log.debug("Wanted to push file '%s' to S3 key '%s' but its size is 0; skipping.", source_file, rel_path)
return True
if from_string:
key.set_contents_from_string(from_string, reduced_redundancy=self.use_rr)
log.debug("Pushed data from string '%s' to key '%s'" % (from_string, rel_path))
log.debug("Pushed data from string '%s' to key '%s'", from_string, rel_path)
else:
start_time = datetime.now()
log.debug("Pushing cache file '%s' of size %s bytes to key '%s'" % (source_file, os.path.getsize(source_file), rel_path))
log.debug("Pushing cache file '%s' of size %s bytes to key '%s'", source_file, os.path.getsize(source_file), rel_path)
mb_size = os.path.getsize(source_file) / 1e6
if mb_size < 10 or (not self.multipart):
self.transfer_progress = 0 # Reset transfer progress counter
@@ -352,13 +346,14 @@ class S3ObjectStore(ObjectStore):
else:
multipart_upload(self.s3server, self.bucket, key.name, source_file, mb_size)
end_time = datetime.now()
log.debug("Pushed cache file '%s' to key '%s' (%s bytes transfered in %s sec)" % (source_file, rel_path, os.path.getsize(source_file), end_time - start_time))
log.debug("Pushed cache file '%s' to key '%s' (%s bytes transfered in %s sec)",
source_file, rel_path, os.path.getsize(source_file), end_time - start_time)
return True
else:
log.error("Tried updating key '%s' from source file '%s', but source file does not exist."
% (rel_path, source_file))
except S3ResponseError, ex:
log.error("Trouble pushing S3 key '%s' from file '%s': %s" % (rel_path, source_file, ex))
log.error("Tried updating key '%s' from source file '%s', but source file does not exist.",
rel_path, source_file)
except S3ResponseError:
log.exception("Trouble pushing S3 key '%s' from file '%s'", rel_path, source_file)
return False
def file_ready(self, obj, **kwargs):
@@ -371,8 +366,8 @@ class S3ObjectStore(ObjectStore):
if self._in_cache(rel_path):
if os.path.getsize(self._get_cache_path(rel_path)) == self._get_size_in_s3(rel_path):
return True
log.debug("Waiting for dataset {0} to transfer from OS: {1}/{2}".format(rel_path,
os.path.getsize(self._get_cache_path(rel_path)), self._get_size_in_s3(rel_path)))
log.debug("Waiting for dataset %s to transfer from OS: %s/%s", rel_path,
os.path.getsize(self._get_cache_path(rel_path)), self._get_size_in_s3(rel_path))
return False
def exists(self, obj, **kwargs):
@@ -444,10 +439,10 @@ class S3ObjectStore(ObjectStore):
try:
return os.path.getsize(self._get_cache_path(rel_path))
except OSError, ex:
log.info("Could not get size of file '%s' in local cache, will try S3. Error: %s" % (rel_path, ex))
log.info("Could not get size of file '%s' in local cache, will try S3. Error: %s", rel_path, ex)
elif self.exists(obj, **kwargs):
return self._get_size_in_s3(rel_path)
log.warning("Did not find dataset '%s', returning 0 for size" % rel_path)
log.warning("Did not find dataset '%s', returning 0 for size", rel_path)
return 0
def delete(self, obj, entire_dir=False, **kwargs):
@@ -460,9 +455,9 @@ class S3ObjectStore(ObjectStore):
# but requires iterating through each individual key in S3 and deleing it.
if entire_dir and extra_dir:
shutil.rmtree(self._get_cache_path(rel_path))
rs = self.bucket.get_all_keys(prefix=rel_path)
for key in rs:
log.debug("Deleting key %s" % key.name)
results = self.bucket.get_all_keys(prefix=rel_path)
for key in results:
log.debug("Deleting key %s", key.name)
key.delete()
return True
else:
@@ -470,14 +465,14 @@ class S3ObjectStore(ObjectStore):
os.unlink(self._get_cache_path(rel_path))
# Delete from S3 as well
if self._key_exists(rel_path):
key = Key(self.bucket, rel_path)
log.debug("Deleting key %s" % key.name)
key.delete()
return True
key = Key(self.bucket, rel_path)
log.debug("Deleting key %s", key.name)
key.delete()
return True
except S3ResponseError, ex:
log.error("Could not delete key '%s' from S3: %s" % (rel_path, ex))
log.error("Could not delete key '%s' from S3: %s", rel_path, ex)
except OSError, ex:
log.error('%s delete error %s' % (self._get_filename(obj, **kwargs), ex))
log.error('%s delete error %s', self._get_filename(obj, **kwargs), ex)
return False
def get_data(self, obj, start=0, count=-1, **kwargs):
@@ -538,7 +533,7 @@ class S3ObjectStore(ObjectStore):
shutil.copy2(source_file, cache_file)
self._fix_permissions(cache_file)
except OSError, ex:
log.error("Trouble copying source file '%s' to cache '%s': %s" % (source_file, cache_file, ex))
log.error("Trouble copying source file '%s' to cache '%s': %s", source_file, cache_file, ex)
else:
source_file = self._get_cache_path(rel_path)
# Update the file on S3
@@ -553,8 +548,8 @@ class S3ObjectStore(ObjectStore):
try:
key = Key(self.bucket, rel_path)
return key.generate_url(expires_in=86400) # 24hrs
except S3ResponseError, ex:
log.warning("Trouble generating URL for dataset '%s': %s" % (rel_path, ex))
except S3ResponseError:
log.exception("Trouble generating URL for dataset '%s'", rel_path)
return None
def get_store_usage_percent(self):