mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Move S3 Object Store to its own module (more to come, probably).
This commit is contained in:
Nate Coraor
@@ -5,14 +5,11 @@ tools
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import random
|
||||
import shutil
|
||||
import logging
|
||||
import threading
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
|
||||
from galaxy import util
|
||||
from galaxy.jobs import Sleeper
|
||||
@@ -21,15 +18,7 @@ from galaxy.exceptions import ObjectNotFound, ObjectInvalid
|
||||
|
||||
from sqlalchemy.orm import object_session
|
||||
|
||||
import multiprocessing
|
||||
from galaxy.objectstore.s3_multipart_upload import multipart_upload
|
||||
import boto
|
||||
from boto.s3.key import Key
|
||||
from boto.s3.connection import S3Connection
|
||||
from boto.exception import S3ResponseError
|
||||
|
||||
log = logging.getLogger( __name__ )
|
||||
logging.getLogger('boto').setLevel(logging.INFO) # Otherwise boto is quite noisy
|
||||
|
||||
|
||||
class ObjectStore(object):
|
||||
@@ -373,490 +362,6 @@ class CachingObjectStore(ObjectStore):
|
||||
super(CachingObjectStore, self).__init__(self, path, backend)
|
||||
|
||||
|
||||
class S3ObjectStore(ObjectStore):
|
||||
"""
|
||||
Object store that stores objects as items in an AWS S3 bucket. A local
|
||||
cache exists that is used as an intermediate location for files between
|
||||
Galaxy and S3.
|
||||
"""
|
||||
def __init__(self, config):
|
||||
super(S3ObjectStore, self).__init__()
|
||||
self.config = config
|
||||
self.staging_path = self.config.file_path
|
||||
self.s3_conn = get_OS_connection(self.config)
|
||||
self.bucket = self._get_bucket(self.config.os_bucket_name)
|
||||
self.use_rr = self.config.os_use_reduced_redundancy
|
||||
self.cache_size = self.config.object_store_cache_size
|
||||
self.transfer_progress = 0
|
||||
# Clean cache only if value is set in universe_wsgi.ini
|
||||
if self.cache_size != -1:
|
||||
# Convert GBs to bytes for comparison
|
||||
self.cache_size = self.cache_size * 1073741824
|
||||
# Helper for interruptable sleep
|
||||
self.sleeper = Sleeper()
|
||||
self.cache_monitor_thread = threading.Thread(target=self.__cache_monitor)
|
||||
self.cache_monitor_thread.start()
|
||||
log.info("Cache cleaner manager started")
|
||||
# Test if 'axel' is available for parallel download and pull the key into cache
|
||||
try:
|
||||
subprocess.call('axel')
|
||||
self.use_axel = True
|
||||
except OSError:
|
||||
self.use_axel = False
|
||||
|
||||
def __cache_monitor(self):
|
||||
time.sleep(2) # Wait for things to load before starting the monitor
|
||||
while self.running:
|
||||
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)
|
||||
total_size += file_size
|
||||
# Get the time given file was last accessed
|
||||
last_access_time = time.localtime(os.stat(fp)[7])
|
||||
# Compose a tuple of the access time and the file path
|
||||
file_tuple = last_access_time, fp, 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)))
|
||||
# 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 #%?
|
||||
# For now, delete enough to leave at least 10% of the total cache free
|
||||
delete_this_much = total_size - cache_limit
|
||||
self.__clean_cache(file_list, delete_this_much)
|
||||
self.sleeper.sleep(30) # Test cache size every 30 seconds?
|
||||
|
||||
def __clean_cache(self, file_list, delete_this_much):
|
||||
""" Keep deleting files from the file_list until the size of the deleted
|
||||
files is greater than the value in delete_this_much parameter.
|
||||
|
||||
:type file_list: list
|
||||
:param file_list: List of candidate files that can be deleted. This method
|
||||
will start deleting files from the beginning of the list so the list
|
||||
should be sorted accordingly. The list must contains 3-element tuples,
|
||||
positioned as follows: position 0 holds file last accessed timestamp
|
||||
(as time.struct_time), position 1 holds file path, and position 2 has
|
||||
file size (e.g., (<access time>, /mnt/data/dataset_1.dat), 472394)
|
||||
|
||||
:type delete_this_much: int
|
||||
:param delete_this_much: Total size of files, in bytes, that should be deleted.
|
||||
"""
|
||||
# Keep deleting datasets from file_list until deleted_amount does not
|
||||
# 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):
|
||||
if deleted_amount < delete_this_much:
|
||||
deleted_amount += f[2]
|
||||
os.remove(f[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])
|
||||
# log.debug("%s. %-25s %s, size %s (deleted %s/%s)" \
|
||||
# % (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))
|
||||
return
|
||||
|
||||
def _get_bucket(self, bucket_name):
|
||||
""" Sometimes a handle to a bucket is not established right away so try
|
||||
it a few times. Raise error is connection is not established. """
|
||||
for i in range(5):
|
||||
try:
|
||||
bucket = self.s3_conn.get_bucket(bucket_name)
|
||||
log.debug("Using cloud object store with bucket '%s'" % bucket.name)
|
||||
return bucket
|
||||
except S3ResponseError:
|
||||
log.debug("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
|
||||
raise S3ResponseError
|
||||
|
||||
def _fix_permissions(self, rel_path):
|
||||
""" Set permissions on rel_path"""
|
||||
for basedir, dirs, files in os.walk(rel_path):
|
||||
util.umask_fix_perms(basedir, self.config.umask, 0777, self.config.gid)
|
||||
for f in files:
|
||||
path = os.path.join(basedir, f)
|
||||
# Ignore symlinks
|
||||
if os.path.islink(path):
|
||||
continue
|
||||
util.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):
|
||||
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)
|
||||
# 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
|
||||
|
||||
def _get_cache_path(self, rel_path):
|
||||
return os.path.abspath(os.path.join(self.staging_path, rel_path))
|
||||
|
||||
def _get_transfer_progress(self):
|
||||
return self.transfer_progress
|
||||
|
||||
def _get_size_in_s3(self, rel_path):
|
||||
try:
|
||||
key = self.bucket.get_key(rel_path)
|
||||
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
|
||||
|
||||
def _key_exists(self, rel_path):
|
||||
exists = False
|
||||
try:
|
||||
# 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:
|
||||
exists = True
|
||||
else:
|
||||
exists = False
|
||||
else:
|
||||
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))
|
||||
return False
|
||||
#print "Checking if '%s' exists in S3: %s" % (rel_path, exists)
|
||||
if rel_path[0] == '/':
|
||||
raise
|
||||
return exists
|
||||
|
||||
def _in_cache(self, rel_path):
|
||||
""" Check if the given dataset is in the local cache and return True if so. """
|
||||
# log.debug("------ Checking cache for rel_path %s" % rel_path)
|
||||
cache_path = self._get_cache_path(rel_path)
|
||||
return os.path.exists(cache_path)
|
||||
# TODO: Part of checking if a file is in cache should be to ensure the
|
||||
# size of the cached file matches that on S3. Once the upload tool explicitly
|
||||
# creates, this check sould be implemented- in the mean time, it's not
|
||||
# looking likely to be implementable reliably.
|
||||
# if os.path.exists(cache_path):
|
||||
# # print "***1 %s exists" % cache_path
|
||||
# if self._key_exists(rel_path):
|
||||
# # print "***2 %s exists in S3" % rel_path
|
||||
# # Make sure the size in cache is available in its entirety
|
||||
# # print "File '%s' cache size: %s, S3 size: %s" % (cache_path, os.path.getsize(cache_path), self._get_size_in_s3(rel_path))
|
||||
# if os.path.getsize(cache_path) == self._get_size_in_s3(rel_path):
|
||||
# # print "***2.1 %s exists in S3 and the size is the same as in cache (in_cache=True)" % rel_path
|
||||
# exists = True
|
||||
# else:
|
||||
# # print "***2.2 %s exists but differs in size from cache (in_cache=False)" % cache_path
|
||||
# exists = False
|
||||
# else:
|
||||
# # Although not perfect decision making, this most likely means
|
||||
# # that the file is currently being uploaded
|
||||
# # print "***3 %s found in cache but not in S3 (in_cache=True)" % cache_path
|
||||
# exists = True
|
||||
# else:
|
||||
# return False
|
||||
|
||||
def _pull_into_cache(self, rel_path):
|
||||
# Ensure the cache directory structure exists (e.g., dataset_#_files/)
|
||||
rel_path_dir = os.path.dirname(rel_path)
|
||||
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)
|
||||
self._fix_permissions(self._get_cache_path(rel_path_dir))
|
||||
return 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)))
|
||||
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))
|
||||
return False
|
||||
if self.use_axel:
|
||||
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)))
|
||||
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))
|
||||
return False
|
||||
|
||||
def _push_to_os(self, rel_path, source_file=None, from_string=None):
|
||||
"""
|
||||
Push the file pointed to by ``rel_path`` to the object store naming the key
|
||||
``rel_path``. If ``source_file`` is provided, push that file instead while
|
||||
still using ``rel_path`` as the key name.
|
||||
If ``from_string`` is provided, set contents of the file to the value of
|
||||
the string.
|
||||
"""
|
||||
try:
|
||||
source_file = source_file if source_file else self._get_cache_path(rel_path)
|
||||
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))
|
||||
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))
|
||||
else:
|
||||
start_time = datetime.now()
|
||||
# print "Pushing cache file '%s' of size %s bytes to key '%s'" % (source_file, os.path.getsize(source_file), rel_path)
|
||||
# print "+ Push started at '%s'" % start_time
|
||||
mb_size = os.path.getsize(source_file) / 1e6
|
||||
if mb_size < 60 or self.config.object_store == 'swift':
|
||||
self.transfer_progress = 0 # Reset transfer progress counter
|
||||
key.set_contents_from_filename(source_file, reduced_redundancy=self.use_rr,
|
||||
cb=self._transfer_cb, num_cb=10)
|
||||
else:
|
||||
multipart_upload(self.bucket, key.name, source_file, mb_size, use_rr=self.use_rr)
|
||||
end_time = datetime.now()
|
||||
# print "+ Push ended at '%s'; %s bytes transfered in %ssec" % (end_time, 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))
|
||||
return False
|
||||
|
||||
def file_ready(self, obj, **kwargs):
|
||||
"""
|
||||
A helper method that checks if a file corresponding to a dataset is
|
||||
ready and available to be used. Return ``True`` if so, ``False`` otherwise.
|
||||
"""
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
# Make sure the size in cache is available in its entirety
|
||||
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)))
|
||||
return False
|
||||
|
||||
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
|
||||
# Check S3
|
||||
in_s3 = self._key_exists(rel_path)
|
||||
# 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)
|
||||
if dir_only:
|
||||
if in_cache or in_s3:
|
||||
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))
|
||||
return True
|
||||
elif in_s3:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def create(self, obj, **kwargs):
|
||||
if not self.exists(obj, **kwargs):
|
||||
#print "S3 OS creating a dataset with ID %s" % 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)
|
||||
# print "---- Processing: %s; %s" % (alt_name, locals())
|
||||
# Construct hashed path
|
||||
rel_path = os.path.join(*directory_hash_id(obj.id))
|
||||
|
||||
# Optionally append extra_dir
|
||||
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)
|
||||
# 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
|
||||
# s3_dir = '%s/' % rel_path
|
||||
# self._push_to_os(s3_dir, from_string='')
|
||||
# If instructed, create the dataset in cache & in S3
|
||||
if not dir_only:
|
||||
rel_path = os.path.join(rel_path, alt_name if alt_name else "dataset_%s.dat" % obj.id)
|
||||
open(os.path.join(self.staging_path, rel_path), 'w').close()
|
||||
self._push_to_os(rel_path, from_string='')
|
||||
|
||||
def empty(self, obj, **kwargs):
|
||||
if self.exists(obj, **kwargs):
|
||||
return bool(self.size(obj, **kwargs) > 0)
|
||||
else:
|
||||
raise ObjectNotFound( 'objectstore.empty, object does not exist: %s, kwargs: %s'
|
||||
%( str( obj ), str( kwargs ) ) )
|
||||
|
||||
def size(self, obj, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
if self._in_cache(rel_path):
|
||||
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))
|
||||
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)
|
||||
return 0
|
||||
|
||||
def delete(self, obj, entire_dir=False, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
extra_dir = kwargs.get('extra_dir', None)
|
||||
try:
|
||||
# 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,
|
||||
# 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)
|
||||
key.delete()
|
||||
return True
|
||||
else:
|
||||
# Delete from cache first
|
||||
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
|
||||
except S3ResponseError, 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))
|
||||
return False
|
||||
|
||||
def get_data(self, obj, start=0, count=-1, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
# Check cache first and get file if not there
|
||||
if not self._in_cache(rel_path):
|
||||
self._pull_into_cache(rel_path)
|
||||
#else:
|
||||
# print "(cccc) Getting '%s' from cache" % self._get_cache_path(rel_path)
|
||||
# Read the file content from cache
|
||||
data_file = open(self._get_cache_path(rel_path), 'r')
|
||||
data_file.seek(start)
|
||||
content = data_file.read(count)
|
||||
data_file.close()
|
||||
return content
|
||||
|
||||
def get_filename(self, obj, **kwargs):
|
||||
#print "S3 get_filename for dataset: %s" % dataset_id
|
||||
dir_only = kwargs.get('dir_only', False)
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
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
|
||||
# the expected cache path.
|
||||
# dir_only = kwargs.get('dir_only', False)
|
||||
# if dir_only:
|
||||
# if not os.path.exists(cache_path):
|
||||
# os.makedirs(cache_path)
|
||||
# return cache_path
|
||||
# Check if the file exists in the cache first
|
||||
if self._in_cache(rel_path):
|
||||
return cache_path
|
||||
# Check if the file exists in persistent storage and, if it does, pull it into cache
|
||||
elif self.exists(obj, **kwargs):
|
||||
if dir_only: # Directories do not get pulled into cache
|
||||
return cache_path
|
||||
else:
|
||||
if self._pull_into_cache(rel_path):
|
||||
return cache_path
|
||||
# For the case of retrieving a directory only, return the expected path
|
||||
# even if it does not exist.
|
||||
# if dir_only:
|
||||
# return cache_path
|
||||
raise ObjectNotFound( 'objectstore.get_filename, no cache_path: %s, kwargs: %s'
|
||||
%( str( obj ), str( kwargs ) ) )
|
||||
# return cache_path # Until the upload tool does not explicitly create the dataset, return expected path
|
||||
|
||||
def update_from_file(self, obj, file_name=None, create=False, **kwargs):
|
||||
if create:
|
||||
self.create(obj, **kwargs)
|
||||
if self.exists(obj, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
# Chose whether to use the dataset file itself or an alternate file
|
||||
if file_name:
|
||||
source_file = os.path.abspath(file_name)
|
||||
# Copy into cache
|
||||
cache_file = self._get_cache_path(rel_path)
|
||||
try:
|
||||
if source_file != cache_file:
|
||||
# FIXME? Should this be a `move`?
|
||||
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))
|
||||
else:
|
||||
source_file = self._get_cache_path(rel_path)
|
||||
# Update the file on S3
|
||||
self._push_to_os(rel_path, source_file)
|
||||
else:
|
||||
raise ObjectNotFound( 'objectstore.update_from_file, object does not exist: %s, kwargs: %s'
|
||||
%( str( obj ), str( kwargs ) ) )
|
||||
|
||||
def get_object_url(self, obj, **kwargs):
|
||||
if self.exists(obj, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
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))
|
||||
return None
|
||||
|
||||
def get_store_usage_percent(self):
|
||||
return 0.0
|
||||
|
||||
class DistributedObjectStore(ObjectStore):
|
||||
"""
|
||||
ObjectStore that defers to a list of backends, for getting objects the
|
||||
@@ -1011,6 +516,7 @@ class DistributedObjectStore(ObjectStore):
|
||||
return id
|
||||
return None
|
||||
|
||||
|
||||
class HierarchicalObjectStore(ObjectStore):
|
||||
"""
|
||||
ObjectStore that defers to a list of backends, for getting objects the
|
||||
@@ -1021,6 +527,7 @@ class HierarchicalObjectStore(ObjectStore):
|
||||
def __init__(self, backends=[]):
|
||||
super(HierarchicalObjectStore, self).__init__()
|
||||
|
||||
|
||||
def build_object_store_from_config(config, fsmon=False):
|
||||
""" Depending on the configuration setting, invoke the appropriate object store
|
||||
"""
|
||||
@@ -1028,6 +535,7 @@ def build_object_store_from_config(config, fsmon=False):
|
||||
if store == 'disk':
|
||||
return DiskObjectStore(config=config)
|
||||
elif store == 's3' or store == 'swift':
|
||||
from galaxy.objectstore.s3 import S3ObjectStore
|
||||
return S3ObjectStore(config=config)
|
||||
elif store == 'distributed':
|
||||
return DistributedObjectStore(config=config, fsmon=fsmon)
|
||||
|
||||
@@ -0,0 +1,535 @@
|
||||
"""
|
||||
Object Store plugin for the Amazon Simple Storage Service (S3)
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
import shutil
|
||||
import logging
|
||||
import threading
|
||||
import subprocess
|
||||
from datetime import datetime
|
||||
|
||||
from galaxy import util
|
||||
from galaxy.jobs import Sleeper
|
||||
from galaxy.model import directory_hash_id
|
||||
from galaxy.objectstore import ObjectStore, convert_bytes
|
||||
from galaxy.exceptions import ObjectNotFound, ObjectInvalid
|
||||
|
||||
import multiprocessing
|
||||
from galaxy.objectstore.s3_multipart_upload import multipart_upload
|
||||
import boto
|
||||
from boto.s3.key import Key
|
||||
from boto.s3.connection import S3Connection
|
||||
from boto.exception import S3ResponseError
|
||||
|
||||
log = logging.getLogger( __name__ )
|
||||
logging.getLogger('boto').setLevel(logging.INFO) # Otherwise boto is quite noisy
|
||||
|
||||
|
||||
class S3ObjectStore(ObjectStore):
|
||||
"""
|
||||
Object store that stores objects as items in an AWS S3 bucket. A local
|
||||
cache exists that is used as an intermediate location for files between
|
||||
Galaxy and S3.
|
||||
"""
|
||||
def __init__(self, config):
|
||||
super(S3ObjectStore, self).__init__()
|
||||
self.config = config
|
||||
self.staging_path = self.config.file_path
|
||||
self.s3_conn = get_OS_connection(self.config)
|
||||
self.bucket = self._get_bucket(self.config.os_bucket_name)
|
||||
self.use_rr = self.config.os_use_reduced_redundancy
|
||||
self.cache_size = self.config.object_store_cache_size
|
||||
self.transfer_progress = 0
|
||||
# Clean cache only if value is set in universe_wsgi.ini
|
||||
if self.cache_size != -1:
|
||||
# Convert GBs to bytes for comparison
|
||||
self.cache_size = self.cache_size * 1073741824
|
||||
# Helper for interruptable sleep
|
||||
self.sleeper = Sleeper()
|
||||
self.cache_monitor_thread = threading.Thread(target=self.__cache_monitor)
|
||||
self.cache_monitor_thread.start()
|
||||
log.info("Cache cleaner manager started")
|
||||
# Test if 'axel' is available for parallel download and pull the key into cache
|
||||
try:
|
||||
subprocess.call('axel')
|
||||
self.use_axel = True
|
||||
except OSError:
|
||||
self.use_axel = False
|
||||
|
||||
def __cache_monitor(self):
|
||||
time.sleep(2) # Wait for things to load before starting the monitor
|
||||
while self.running:
|
||||
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)
|
||||
total_size += file_size
|
||||
# Get the time given file was last accessed
|
||||
last_access_time = time.localtime(os.stat(fp)[7])
|
||||
# Compose a tuple of the access time and the file path
|
||||
file_tuple = last_access_time, fp, 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)))
|
||||
# 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 #%?
|
||||
# For now, delete enough to leave at least 10% of the total cache free
|
||||
delete_this_much = total_size - cache_limit
|
||||
self.__clean_cache(file_list, delete_this_much)
|
||||
self.sleeper.sleep(30) # Test cache size every 30 seconds?
|
||||
|
||||
def __clean_cache(self, file_list, delete_this_much):
|
||||
""" Keep deleting files from the file_list until the size of the deleted
|
||||
files is greater than the value in delete_this_much parameter.
|
||||
|
||||
:type file_list: list
|
||||
:param file_list: List of candidate files that can be deleted. This method
|
||||
will start deleting files from the beginning of the list so the list
|
||||
should be sorted accordingly. The list must contains 3-element tuples,
|
||||
positioned as follows: position 0 holds file last accessed timestamp
|
||||
(as time.struct_time), position 1 holds file path, and position 2 has
|
||||
file size (e.g., (<access time>, /mnt/data/dataset_1.dat), 472394)
|
||||
|
||||
:type delete_this_much: int
|
||||
:param delete_this_much: Total size of files, in bytes, that should be deleted.
|
||||
"""
|
||||
# Keep deleting datasets from file_list until deleted_amount does not
|
||||
# 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):
|
||||
if deleted_amount < delete_this_much:
|
||||
deleted_amount += f[2]
|
||||
os.remove(f[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])
|
||||
# log.debug("%s. %-25s %s, size %s (deleted %s/%s)" \
|
||||
# % (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))
|
||||
return
|
||||
|
||||
def _get_bucket(self, bucket_name):
|
||||
""" Sometimes a handle to a bucket is not established right away so try
|
||||
it a few times. Raise error is connection is not established. """
|
||||
for i in range(5):
|
||||
try:
|
||||
bucket = self.s3_conn.get_bucket(bucket_name)
|
||||
log.debug("Using cloud object store with bucket '%s'" % bucket.name)
|
||||
return bucket
|
||||
except S3ResponseError:
|
||||
log.debug("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
|
||||
raise S3ResponseError
|
||||
|
||||
def _fix_permissions(self, rel_path):
|
||||
""" Set permissions on rel_path"""
|
||||
for basedir, dirs, files in os.walk(rel_path):
|
||||
util.umask_fix_perms(basedir, self.config.umask, 0777, self.config.gid)
|
||||
for f in files:
|
||||
path = os.path.join(basedir, f)
|
||||
# Ignore symlinks
|
||||
if os.path.islink(path):
|
||||
continue
|
||||
util.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):
|
||||
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)
|
||||
# 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
|
||||
|
||||
def _get_cache_path(self, rel_path):
|
||||
return os.path.abspath(os.path.join(self.staging_path, rel_path))
|
||||
|
||||
def _get_transfer_progress(self):
|
||||
return self.transfer_progress
|
||||
|
||||
def _get_size_in_s3(self, rel_path):
|
||||
try:
|
||||
key = self.bucket.get_key(rel_path)
|
||||
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
|
||||
|
||||
def _key_exists(self, rel_path):
|
||||
exists = False
|
||||
try:
|
||||
# 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:
|
||||
exists = True
|
||||
else:
|
||||
exists = False
|
||||
else:
|
||||
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))
|
||||
return False
|
||||
#print "Checking if '%s' exists in S3: %s" % (rel_path, exists)
|
||||
if rel_path[0] == '/':
|
||||
raise
|
||||
return exists
|
||||
|
||||
def _in_cache(self, rel_path):
|
||||
""" Check if the given dataset is in the local cache and return True if so. """
|
||||
# log.debug("------ Checking cache for rel_path %s" % rel_path)
|
||||
cache_path = self._get_cache_path(rel_path)
|
||||
return os.path.exists(cache_path)
|
||||
# TODO: Part of checking if a file is in cache should be to ensure the
|
||||
# size of the cached file matches that on S3. Once the upload tool explicitly
|
||||
# creates, this check sould be implemented- in the mean time, it's not
|
||||
# looking likely to be implementable reliably.
|
||||
# if os.path.exists(cache_path):
|
||||
# # print "***1 %s exists" % cache_path
|
||||
# if self._key_exists(rel_path):
|
||||
# # print "***2 %s exists in S3" % rel_path
|
||||
# # Make sure the size in cache is available in its entirety
|
||||
# # print "File '%s' cache size: %s, S3 size: %s" % (cache_path, os.path.getsize(cache_path), self._get_size_in_s3(rel_path))
|
||||
# if os.path.getsize(cache_path) == self._get_size_in_s3(rel_path):
|
||||
# # print "***2.1 %s exists in S3 and the size is the same as in cache (in_cache=True)" % rel_path
|
||||
# exists = True
|
||||
# else:
|
||||
# # print "***2.2 %s exists but differs in size from cache (in_cache=False)" % cache_path
|
||||
# exists = False
|
||||
# else:
|
||||
# # Although not perfect decision making, this most likely means
|
||||
# # that the file is currently being uploaded
|
||||
# # print "***3 %s found in cache but not in S3 (in_cache=True)" % cache_path
|
||||
# exists = True
|
||||
# else:
|
||||
# return False
|
||||
|
||||
def _pull_into_cache(self, rel_path):
|
||||
# Ensure the cache directory structure exists (e.g., dataset_#_files/)
|
||||
rel_path_dir = os.path.dirname(rel_path)
|
||||
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)
|
||||
self._fix_permissions(self._get_cache_path(rel_path_dir))
|
||||
return 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)))
|
||||
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))
|
||||
return False
|
||||
if self.use_axel:
|
||||
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)))
|
||||
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))
|
||||
return False
|
||||
|
||||
def _push_to_os(self, rel_path, source_file=None, from_string=None):
|
||||
"""
|
||||
Push the file pointed to by ``rel_path`` to the object store naming the key
|
||||
``rel_path``. If ``source_file`` is provided, push that file instead while
|
||||
still using ``rel_path`` as the key name.
|
||||
If ``from_string`` is provided, set contents of the file to the value of
|
||||
the string.
|
||||
"""
|
||||
try:
|
||||
source_file = source_file if source_file else self._get_cache_path(rel_path)
|
||||
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))
|
||||
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))
|
||||
else:
|
||||
start_time = datetime.now()
|
||||
# print "Pushing cache file '%s' of size %s bytes to key '%s'" % (source_file, os.path.getsize(source_file), rel_path)
|
||||
# print "+ Push started at '%s'" % start_time
|
||||
mb_size = os.path.getsize(source_file) / 1e6
|
||||
if mb_size < 60 or self.config.object_store == 'swift':
|
||||
self.transfer_progress = 0 # Reset transfer progress counter
|
||||
key.set_contents_from_filename(source_file, reduced_redundancy=self.use_rr,
|
||||
cb=self._transfer_cb, num_cb=10)
|
||||
else:
|
||||
multipart_upload(self.bucket, key.name, source_file, mb_size, use_rr=self.use_rr)
|
||||
end_time = datetime.now()
|
||||
# print "+ Push ended at '%s'; %s bytes transfered in %ssec" % (end_time, 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))
|
||||
return False
|
||||
|
||||
def file_ready(self, obj, **kwargs):
|
||||
"""
|
||||
A helper method that checks if a file corresponding to a dataset is
|
||||
ready and available to be used. Return ``True`` if so, ``False`` otherwise.
|
||||
"""
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
# Make sure the size in cache is available in its entirety
|
||||
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)))
|
||||
return False
|
||||
|
||||
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
|
||||
# Check S3
|
||||
in_s3 = self._key_exists(rel_path)
|
||||
# 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)
|
||||
if dir_only:
|
||||
if in_cache or in_s3:
|
||||
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))
|
||||
return True
|
||||
elif in_s3:
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def create(self, obj, **kwargs):
|
||||
if not self.exists(obj, **kwargs):
|
||||
#print "S3 OS creating a dataset with ID %s" % 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)
|
||||
# print "---- Processing: %s; %s" % (alt_name, locals())
|
||||
# Construct hashed path
|
||||
rel_path = os.path.join(*directory_hash_id(obj.id))
|
||||
|
||||
# Optionally append extra_dir
|
||||
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)
|
||||
# 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
|
||||
# s3_dir = '%s/' % rel_path
|
||||
# self._push_to_os(s3_dir, from_string='')
|
||||
# If instructed, create the dataset in cache & in S3
|
||||
if not dir_only:
|
||||
rel_path = os.path.join(rel_path, alt_name if alt_name else "dataset_%s.dat" % obj.id)
|
||||
open(os.path.join(self.staging_path, rel_path), 'w').close()
|
||||
self._push_to_os(rel_path, from_string='')
|
||||
|
||||
def empty(self, obj, **kwargs):
|
||||
if self.exists(obj, **kwargs):
|
||||
return bool(self.size(obj, **kwargs) > 0)
|
||||
else:
|
||||
raise ObjectNotFound( 'objectstore.empty, object does not exist: %s, kwargs: %s'
|
||||
%( str( obj ), str( kwargs ) ) )
|
||||
|
||||
def size(self, obj, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
if self._in_cache(rel_path):
|
||||
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))
|
||||
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)
|
||||
return 0
|
||||
|
||||
def delete(self, obj, entire_dir=False, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
extra_dir = kwargs.get('extra_dir', None)
|
||||
try:
|
||||
# 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,
|
||||
# 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)
|
||||
key.delete()
|
||||
return True
|
||||
else:
|
||||
# Delete from cache first
|
||||
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
|
||||
except S3ResponseError, 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))
|
||||
return False
|
||||
|
||||
def get_data(self, obj, start=0, count=-1, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
# Check cache first and get file if not there
|
||||
if not self._in_cache(rel_path):
|
||||
self._pull_into_cache(rel_path)
|
||||
#else:
|
||||
# print "(cccc) Getting '%s' from cache" % self._get_cache_path(rel_path)
|
||||
# Read the file content from cache
|
||||
data_file = open(self._get_cache_path(rel_path), 'r')
|
||||
data_file.seek(start)
|
||||
content = data_file.read(count)
|
||||
data_file.close()
|
||||
return content
|
||||
|
||||
def get_filename(self, obj, **kwargs):
|
||||
#print "S3 get_filename for dataset: %s" % dataset_id
|
||||
dir_only = kwargs.get('dir_only', False)
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
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
|
||||
# the expected cache path.
|
||||
# dir_only = kwargs.get('dir_only', False)
|
||||
# if dir_only:
|
||||
# if not os.path.exists(cache_path):
|
||||
# os.makedirs(cache_path)
|
||||
# return cache_path
|
||||
# Check if the file exists in the cache first
|
||||
if self._in_cache(rel_path):
|
||||
return cache_path
|
||||
# Check if the file exists in persistent storage and, if it does, pull it into cache
|
||||
elif self.exists(obj, **kwargs):
|
||||
if dir_only: # Directories do not get pulled into cache
|
||||
return cache_path
|
||||
else:
|
||||
if self._pull_into_cache(rel_path):
|
||||
return cache_path
|
||||
# For the case of retrieving a directory only, return the expected path
|
||||
# even if it does not exist.
|
||||
# if dir_only:
|
||||
# return cache_path
|
||||
raise ObjectNotFound( 'objectstore.get_filename, no cache_path: %s, kwargs: %s'
|
||||
%( str( obj ), str( kwargs ) ) )
|
||||
# return cache_path # Until the upload tool does not explicitly create the dataset, return expected path
|
||||
|
||||
def update_from_file(self, obj, file_name=None, create=False, **kwargs):
|
||||
if create:
|
||||
self.create(obj, **kwargs)
|
||||
if self.exists(obj, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
# Chose whether to use the dataset file itself or an alternate file
|
||||
if file_name:
|
||||
source_file = os.path.abspath(file_name)
|
||||
# Copy into cache
|
||||
cache_file = self._get_cache_path(rel_path)
|
||||
try:
|
||||
if source_file != cache_file:
|
||||
# FIXME? Should this be a `move`?
|
||||
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))
|
||||
else:
|
||||
source_file = self._get_cache_path(rel_path)
|
||||
# Update the file on S3
|
||||
self._push_to_os(rel_path, source_file)
|
||||
else:
|
||||
raise ObjectNotFound( 'objectstore.update_from_file, object does not exist: %s, kwargs: %s'
|
||||
%( str( obj ), str( kwargs ) ) )
|
||||
|
||||
def get_object_url(self, obj, **kwargs):
|
||||
if self.exists(obj, **kwargs):
|
||||
rel_path = self._construct_path(obj, **kwargs)
|
||||
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))
|
||||
return None
|
||||
|
||||
def get_store_usage_percent(self):
|
||||
return 0.0
|
||||
|
||||
|
||||
def get_OS_connection(config):
|
||||
"""
|
||||
Get a connection object for a cloud Object Store specified in the config.
|
||||
Currently, this is a ``boto`` connection object.
|
||||
"""
|
||||
log.debug("Getting a connection object for '{0}' object store".format(config.object_store))
|
||||
a_key = config.os_access_key
|
||||
s_key = config.os_secret_key
|
||||
if config.object_store == 's3':
|
||||
return S3Connection(a_key, s_key)
|
||||
else:
|
||||
# Establish the connection now
|
||||
calling_format = boto.s3.connection.OrdinaryCallingFormat()
|
||||
s3_conn = boto.connect_s3(aws_access_key_id=a_key,
|
||||
aws_secret_access_key=s_key,
|
||||
is_secure=config.os_is_secure,
|
||||
host=config.os_host,
|
||||
port=int(config.os_port),
|
||||
calling_format=calling_format,
|
||||
path=config.os_conn_path)
|
||||
return s3_conn
|
||||
Reference in New Issue
Block a user