1 new commit in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/changeset/663e03e40c86/
changeset: 663e03e40c86
user: dannon
date: 2012-06-20 09:39:05
summary: Remove debugging exception from 06249bb8e6c9.
affected #: 1 file
diff -r 06249bb8e6c9be1e2f94fbd860f123f4ae51a356 -r 663e03e40c86ffa0766dd6638ea7c0886632117d lib/galaxy/datatypes/data.py
--- a/lib/galaxy/datatypes/data.py
+++ b/lib/galaxy/datatypes/data.py
@@ -284,7 +284,6 @@
return open( dataset.file_name )
def display_data(self, trans, data, preview=False, filename=None, to_ext=None, size=None, offset=None, **kwd):
- raise Exception
""" Old display method, for transition """
#Relocate all composite datatype display to a common location.
composite_extensions = trans.app.datatypes_registry.get_composite_extensions( )
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
1 new commit in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/changeset/06249bb8e6c9/
changeset: 06249bb8e6c9
user: Scott McManus
date: 2012-06-20 08:14:55
summary: Adding in exit code management for local changes; I'm holding off on the PBS and generic DRMAA runners until they're tested.
affected #: 4 files
diff -r f1c5d6639f2efcf0eec530d1901c9816d65b32b0 -r 06249bb8e6c9be1e2f94fbd860f123f4ae51a356 lib/galaxy/datatypes/data.py
--- a/lib/galaxy/datatypes/data.py
+++ b/lib/galaxy/datatypes/data.py
@@ -284,6 +284,7 @@
return open( dataset.file_name )
def display_data(self, trans, data, preview=False, filename=None, to_ext=None, size=None, offset=None, **kwd):
+ raise Exception
""" Old display method, for transition """
#Relocate all composite datatype display to a common location.
composite_extensions = trans.app.datatypes_registry.get_composite_extensions( )
diff -r f1c5d6639f2efcf0eec530d1901c9816d65b32b0 -r 06249bb8e6c9be1e2f94fbd860f123f4ae51a356 lib/galaxy/jobs/__init__.py
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -285,12 +285,14 @@
self.sa_session.add( job )
self.sa_session.flush()
- def finish( self, stdout, stderr ):
+ def finish( self, stdout, stderr, tool_exit_code=0 ):
"""
Called to indicate that the associated command has been run. Updates
the output datasets based on stderr and stdout from the command, and
the contents of the output files.
"""
+ # TODO: Eliminate debugging code after testing all runners
+ log.debug( "JobWrapper.finish: exit code:" + str(tool_exit_code) )
# default post job setup
self.sa_session.expunge_all()
job = self.get_job()
@@ -317,17 +319,15 @@
# that range, then apply the error level and add in a message.
# If we've reached a fatal error rule, then stop.
max_error_level = galaxy.tools.StdioErrorLevel.NO_ERROR
- for exit_code in self.tool.stdio_exit_codes:
- # TODO: Fetch the exit code from the .rc file:
- tool_exit_code = 0
- if ( tool_exit_code >= exit_code.range_start and
- tool_exit_code <= exit_code.range_end ):
- if None != exit_code.desc:
- err_msg += exit_code.desc
+ for stdio_exit_code in self.tool.stdio_exit_codes:
+ if ( tool_exit_code >= stdio_exit_code.range_start and
+ tool_exit_code <= stdio_exit_code.range_end ):
+ if None != stdio_exit_code.desc:
+ err_msg += stdio_exit_code.desc
# TODO: Find somewhere to stick the err_msg - possibly to
# the source (stderr/stdout), possibly in a new db column.
max_error_level = max( max_error_level,
- exit_code.error_level )
+ stdio_exit_code.error_level )
if max_error_level >= galaxy.tools.StdioErrorLevel.FATAL:
break
# If there is a regular expression for scanning stdout/stderr,
diff -r f1c5d6639f2efcf0eec530d1901c9816d65b32b0 -r 06249bb8e6c9be1e2f94fbd860f123f4ae51a356 lib/galaxy/jobs/handler.py
--- a/lib/galaxy/jobs/handler.py
+++ b/lib/galaxy/jobs/handler.py
@@ -368,6 +368,7 @@
start_job_runners.append("tasks")
for name in start_job_runners:
self._load_plugin( name )
+ log.debug( "Job runners: " + ':'.join( start_job_runners ) )
def _load_plugin( self, name ):
module_name = 'galaxy.jobs.runners.' + name
@@ -397,6 +398,7 @@
def put( self, job_wrapper ):
try:
runner_name = self.__get_runner_name( job_wrapper )
+ log.debug( "Runner_name: " + runner_name )
if self.app.config.use_tasked_jobs and job_wrapper.tool.parallelism is not None and isinstance(job_wrapper, TaskWrapper):
#DBTODO Refactor
log.debug( "dispatching task %s, of job %d, to %s runner" %( job_wrapper.task_id, job_wrapper.job_id, runner_name ) )
diff -r f1c5d6639f2efcf0eec530d1901c9816d65b32b0 -r 06249bb8e6c9be1e2f94fbd860f123f4ae51a356 lib/galaxy/jobs/runners/local.py
--- a/lib/galaxy/jobs/runners/local.py
+++ b/lib/galaxy/jobs/runners/local.py
@@ -54,6 +54,7 @@
def run_job( self, job_wrapper ):
job_wrapper.set_runner( 'local:///', None )
stderr = stdout = command_line = ''
+ exit_code = 0
# Prepare the job to run
try:
job_wrapper.prepare()
@@ -99,7 +100,11 @@
if sleep_time < 8:
# So we don't stat every second
sleep_time *= 2
- proc.wait() # reap
+ # Reap the process and get the exit code. The exit code should
+ # only be None if the process isn't finished, but check anyway.
+ exit_code = proc.wait() # reap
+ if None == exit_code:
+ exit_code = 0
stdout_file.seek( 0 )
stderr_file.seek( 0 )
stdout = stdout_file.read( 32768 )
@@ -128,9 +133,9 @@
external_metadata_proc.wait()
log.debug( 'execution of external set_meta for job %d finished' % job_wrapper.job_id )
- # Finish the job
+ # Finish the job!
try:
- job_wrapper.finish( stdout, stderr )
+ job_wrapper.finish( stdout, stderr, exit_code )
except:
log.exception("Job wrapper finish method failed")
job_wrapper.fail("Unable to finish job", exception=True)
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
1 new commit in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/changeset/8d6138fa2251/
changeset: 8d6138fa2251
user: dannon
date: 2012-06-19 22:42:25
summary: S3 object store now functional.
Whitespace cleanup.
affected #: 1 file
diff -r 958d1f8b2caf9595d79e66d1eb1705110a1a6784 -r 8d6138fa2251566b5d17d64e9f56f129d9dc624a lib/galaxy/objectstore/__init__.py
--- a/lib/galaxy/objectstore/__init__.py
+++ b/lib/galaxy/objectstore/__init__.py
@@ -40,20 +40,20 @@
def __init__(self):
self.running = True
self.extra_dirs = {}
-
+
def shutdown(self):
self.running = False
-
+
def exists(self, obj, base_dir=None, dir_only=False, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""
Returns True if the object identified by `obj` exists in this file
store, False otherwise.
-
+
FIELD DESCRIPTIONS (these apply to all the methods in this class):
:type obj: object
:param obj: A Galaxy object with an assigned database ID accessible via
the .id attribute.
-
+
:type base_dir: string
:param base_dir: A key in self.extra_dirs corresponding to the base
directory in which this object should be created, or
@@ -64,19 +64,19 @@
identified by `obj` should be located, not the dataset
itself. This option applies to `extra_dir` argument as
well.
-
+
:type extra_dir: string
:param extra_dir: Append `extra_dir` to the directory structure where
the dataset identified by `obj` should be located.
(e.g., 000/extra_dir/obj.id)
-
+
:type extra_dir_at_root: bool
:param extra_dir_at_root: Applicable only if `extra_dir` is set.
If True, the `extra_dir` argument is placed at
root of the created directory structure rather
than at the end (e.g., extra_dir/000/obj.id
vs. 000/extra_dir/obj.id)
-
+
:type alt_name: string
:param alt_name: Use this name as the alternative name for the created
dataset rather than the default.
@@ -84,10 +84,10 @@
raise NotImplementedError()
def file_ready(self, obj, base_dir=None, dir_only=False, extra_dir=None, extra_dir_at_root=False, alt_name=None):
- """ A helper method that checks if a file corresponding to a dataset
+ """ 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."""
return True
-
+
def create(self, obj, base_dir=None, dir_only=False, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""
Mark the object identified by `obj` as existing in the store, but with
@@ -104,7 +104,7 @@
See `exists` method for the description of the fields.
"""
raise NotImplementedError()
-
+
def size(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""
Return size of the object identified by `obj`.
@@ -112,13 +112,13 @@
See `exists` method for the description of the fields.
"""
raise NotImplementedError()
-
+
def delete(self, obj, entire_dir=False, base_dir=None, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""
Deletes the object identified by `obj`.
See `exists` method for the description of other fields.
:type entire_dir: bool
- :param entire_dir: If True, delete the entire directory pointed to by
+ :param entire_dir: If True, delete the entire directory pointed to by
extra_dir. For safety reasons, this option applies
only for and in conjunction with the extra_dir option.
"""
@@ -130,15 +130,15 @@
object identified uniquely by `obj`.
If the object does not exist raises `ObjectNotFound`.
See `exists` method for the description of other fields.
-
+
:type start: int
:param start: Set the position to start reading the dataset file
-
+
:type count: int
:param count: Read at most `count` bytes from the dataset
"""
raise NotImplementedError()
-
+
def get_filename(self, obj, base_dir=None, dir_only=False, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""
Get the expected filename (including the absolute path) which can be used
@@ -146,7 +146,7 @@
See `exists` method for the description of the fields.
"""
raise NotImplementedError()
-
+
def update_from_file(self, obj, base_dir=None, extra_dir=None, extra_dir_at_root=False, alt_name=None, file_name=None, create=False):
"""
Inform the store that the file associated with the object has been
@@ -154,16 +154,16 @@
of the default.
If the object does not exist raises `ObjectNotFound`.
See `exists` method for the description of other fields.
-
+
:type file_name: string
- :param file_name: Use file pointed to by `file_name` as the source for
+ :param file_name: Use file pointed to by `file_name` as the source for
updating the dataset identified by `obj`
-
+
:type create: bool
:param create: If True and the default dataset does not exist, create it first.
"""
raise NotImplementedError()
-
+
def get_object_url(self, obj, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""
If the store supports direct URL access, return a URL. Otherwise return
@@ -178,7 +178,7 @@
Return the percentage indicating how full the store is
"""
raise NotImplementedError()
-
+
## def get_staging_command( id ):
## """
## Return a shell command that can be prepended to the job script to stage the
@@ -212,21 +212,21 @@
self.extra_dirs['temp'] = config.new_file_path
if extra_dirs is not None:
self.extra_dirs.update( extra_dirs )
-
+
def _get_filename(self, obj, base_dir=None, dir_only=False, extra_dir=None, extra_dir_at_root=False, alt_name=None):
"""Class method that returns the absolute path for the file corresponding
- to the `obj`.id regardless of whether the file exists.
+ to the `obj`.id regardless of whether the file exists.
"""
path = self._construct_path(obj, base_dir=base_dir, dir_only=dir_only, extra_dir=extra_dir, extra_dir_at_root=extra_dir_at_root, alt_name=alt_name, old_style=True)
- # For backward compatibility, check the old style root path first; otherwise,
+ # For backward compatibility, check the old style root path first; otherwise,
# construct hashed path
if not os.path.exists(path):
return self._construct_path(obj, base_dir=base_dir, dir_only=dir_only, extra_dir=extra_dir, extra_dir_at_root=extra_dir_at_root, alt_name=alt_name)
-
+
def _construct_path(self, obj, old_style=False, base_dir=None, dir_only=False, extra_dir=None, extra_dir_at_root=False, alt_name=None, **kwargs):
""" Construct the expected absolute path for accessing the object
identified by `obj`.id.
-
+
:type base_dir: string
:param base_dir: A key in self.extra_dirs corresponding to the base
directory in which this object should be created, or
@@ -237,16 +237,16 @@
identified by `obj` should be located, not the
dataset itself. This option applies to `extra_dir`
argument as well.
-
+
:type extra_dir: string
:param extra_dir: Append the value of this parameter to the expected path
used to access the object identified by `obj`
(e.g., /files/000/<extra_dir>/dataset_10.dat).
-
+
:type alt_name: string
:param alt_name: Use this name as the alternative name for the returned
dataset rather than the default.
-
+
:type old_style: bool
param old_style: This option is used for backward compatibility. If True
the composed directory structure does not include a hash id
@@ -274,7 +274,7 @@
def exists(self, obj, **kwargs):
path = self._construct_path(obj, old_style=True, **kwargs)
- # For backward compatibility, check root path first; otherwise, construct
+ # For backward compatibility, check root path first; otherwise, construct
# and check hashed path
if os.path.exists(path):
return True
@@ -292,12 +292,12 @@
os.makedirs(dir)
# Create the file if it does not exist
if not dir_only:
- open(path, 'w').close()
+ open(path, 'w').close()
util.umask_fix_perms(path, self.config.umask, 0666)
def empty(self, obj, **kwargs):
return os.path.getsize(self.get_filename(obj, **kwargs)) > 0
-
+
def size(self, obj, **kwargs):
if self.exists(obj, **kwargs):
try:
@@ -306,7 +306,7 @@
return 0
else:
return 0
-
+
def delete(self, obj, entire_dir=False, **kwargs):
path = self.get_filename(obj, **kwargs)
extra_dir = kwargs.get('extra_dir', None)
@@ -327,16 +327,16 @@
content = data_file.read(count)
data_file.close()
return content
-
+
def get_filename(self, obj, **kwargs):
path = self._construct_path(obj, old_style=True, **kwargs)
- # For backward compatibility, check root path first; otherwise, construct
+ # For backward compatibility, check root path first; otherwise, construct
# and return hashed path
if os.path.exists(path):
return path
else:
return self._construct_path(obj, **kwargs)
-
+
def update_from_file(self, obj, file_name=None, create=False, **kwargs):
""" `create` parameter is not used in this implementation """
if create:
@@ -345,12 +345,12 @@
try:
shutil.copy(file_name, self.get_filename(obj, **kwargs))
except IOError, ex:
- log.critical('Error copying %s to %s: %s' % (file_name,
+ log.critical('Error copying %s to %s: %s' % (file_name,
self._get_filename(obj, **kwargs), ex))
-
+
def get_object_url(self, obj, **kwargs):
return None
-
+
def get_store_usage_percent(self):
st = os.statvfs(self.file_path)
return (float(st.f_blocks - st.f_bavail)/st.f_blocks) * 100
@@ -361,10 +361,9 @@
Object store that uses a directory for caching files, but defers and writes
back to another object store.
"""
-
+
def __init__(self, path, backend):
super(CachingObjectStore, self).__init__(self, path, backend)
-
class S3ObjectStore(ObjectStore):
@@ -381,16 +380,24 @@
self.s3_conn = S3Connection()
self.bucket = self._get_bucket(self.config.s3_bucket)
self.use_rr = self.config.use_reduced_redundancy
- self.cache_size = self.config.object_store_cache_size * 1073741824 # Convert GBs to bytes
+ 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:
@@ -421,19 +428,19 @@
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
+ (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.
"""
@@ -454,7 +461,7 @@
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. """
@@ -463,13 +470,13 @@
bucket = self.s3_conn.get_bucket(bucket_name)
log.debug("Using S3 object store; got bucket '%s'" % bucket.name)
return bucket
- except S3ResponseError:
+ 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):
@@ -478,10 +485,10 @@
path = os.path.join(basedir, f)
# Ignore symlinks
if os.path.islink(path):
- continue
+ 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):
+
+ 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:
@@ -496,10 +503,10 @@
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)
@@ -510,7 +517,7 @@
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:
@@ -532,15 +539,13 @@
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)
- exists = os.path.exists(cache_path)
- # print "Checking chache for %s; returning %s" % (cache_path, exists)
- return exists
- # EATODO: Part of checking if a file is in cache should be to ensure the
+ 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.
@@ -562,11 +567,7 @@
# # print "***3 %s found in cache but not in S3 (in_cache=True)" % cache_path
# exists = True
# else:
- # # print "***4 %s does not exist (in_cache=False)" % cache_path
- # exists = False
- # # print "Checking cache for %s; returning %s" % (cache_path, exists)
- # return exists
- # # return False
+ # return False
def _pull_into_cache(self, rel_path):
# Ensure the cache directory structure exists (e.g., dataset_#_files/)
@@ -577,45 +578,39 @@
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
- # print "Dataset transfer progress: %s" % self.transfer_progress
-
+
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 is cache is large enough to hold the new file
- if key.size > self.cache_size:
+ # 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
- # Test if 'axel' is available for parallel download and pull the key into cache
- try:
- ret_code = subprocess.call('axel')
- except OSError:
- ret_code = 127
- if ret_code == 127:
- 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)
- #print "(ssss1) Pulled key '%s' into cache to %s" % (rel_path, self._get_cache_path(rel_path))
- return True
- else:
+ 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:
- #print "(ssss2) Parallel pulled key '%s' into cache to %s" % (rel_path, self._get_cache_path(rel_path))
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_s3(self, rel_path, source_file=None, from_string=None):
- """
- Push the file pointed to by `rel_path` to S3 naming the key `rel_path`.
- If `source_file` is provided, push that file instead while still using
+ """
+ Push the file pointed to by `rel_path` to S3 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
@@ -651,16 +646,16 @@
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
+ """ 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) and os.path.getsize(self._get_cache_path(rel_path)) == self._get_size_in_s3(rel_path):
return True
return False
-
+
def exists(self, obj, **kwargs):
in_cache = in_s3 = False
rel_path = self._construct_path(obj, **kwargs)
@@ -685,10 +680,10 @@
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" % dataset_id
+ #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)
@@ -696,7 +691,8 @@
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))
+ 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:
@@ -717,13 +713,13 @@
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_s3(rel_path, from_string='')
-
+
def empty(self, obj, **kwargs):
if self.exists(obj, **kwargs):
return bool(self.size(obj, **kwargs) > 0)
else:
raise ObjectNotFound()
-
+
def size(self, obj, **kwargs):
rel_path = self._construct_path(obj, **kwargs)
if self._in_cache(rel_path):
@@ -735,7 +731,7 @@
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)
@@ -765,7 +761,7 @@
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
@@ -779,7 +775,7 @@
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)
@@ -809,8 +805,8 @@
# return cache_path
raise ObjectNotFound()
# 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):
+
+ def update_from_file(self, obj, file_name=None, create=False, **kwargs):
if create:
self.create(obj, **kwargs)
if self.exists(obj, **kwargs):
@@ -833,7 +829,7 @@
self._push_to_s3(rel_path, source_file)
else:
raise ObjectNotFound()
-
+
def get_object_url(self, obj, **kwargs):
if self.exists(obj, **kwargs):
rel_path = self._construct_path(obj, **kwargs)
@@ -854,7 +850,7 @@
first store where the object exists is used, objects are created in a
store selected randomly, but with weighting.
"""
-
+
def __init__(self, config):
super(DistributedObjectStore, self).__init__()
self.distributed_config = config.distributed_object_store_config_file
@@ -1003,7 +999,7 @@
first store where the object exists is used, objects are always created
in the first store.
"""
-
+
def __init__(self, backends=[]):
super(HierarchicalObjectStore, self).__init__()
@@ -1027,7 +1023,7 @@
if bytes is None:
bytes = 0
bytes = float(bytes)
-
+
if bytes >= 1099511627776:
terabytes = bytes / 1099511627776
size = '%.2fTB' % terabytes
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
1 new commit in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/changeset/22eea899376f/
changeset: 22eea899376f
user: Scott McManus
date: 2012-06-19 18:18:11
summary: Fixing vimdiff errors.
affected #: 1 file
diff -r 3b2dc0a51d147c3fc8f0560d8fd3ebefc0792e99 -r 22eea899376f142134c30abafb6e156a514574dd lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -2361,12 +2361,12 @@
installed_tool_dependencies = self.tool_shed_repository.tool_dependencies
else:
installed_tool_dependencies = None
- for requirement in self.requirements:
- # TODO: currently only supporting requirements of type package,
- # need to implement some mechanism for mapping other types
- # back to packages
+ for requirement in self.requirements:
+ # TODO: currently only supporting requirements of type package,
+ # need to implement some mechanism for mapping other types
+ # back to packages
log.debug( "Building dependency shell command for dependency '%s'", requirement.name )
- if requirement.type == 'package':
+ if requirement.type == 'package':
script_file, base_path, version = self.app.toolbox.dependency_manager.find_dep( name=requirement.name,
version=requirement.version,
type=requirement.type,
@@ -2656,7 +2656,7 @@
param_dict.update( { 'type' : 'number', 'init_value' : input.value,
'html' : urllib.quote( input.get_html( trans ) ),
'min': input.min,
- 'max': input.max
+ 'max': input.max,
'value': input.value
} )
else:
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
1 new commit in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/changeset/3b2dc0a51d14/
changeset: 3b2dc0a51d14
user: Scott McManus
date: 2012-06-19 17:35:36
summary: Added application of regular expressions and exit code. Pulling exit
codes from the runners will be part of a separate submission.
affected #: 3 files
diff -r 7c495f835a1d436ad33dff6107784f106cc24980 -r 3b2dc0a51d147c3fc8f0560d8fd3ebefc0792e99 lib/galaxy/jobs/__init__.py
--- a/lib/galaxy/jobs/__init__.py
+++ b/lib/galaxy/jobs/__init__.py
@@ -305,10 +305,81 @@
if job.state == job.states.DELETED or job.state == job.states.ERROR:
#ERROR at this point means the job was deleted by an administrator.
return self.fail( job.info )
- if stderr:
- job.state = job.states.ERROR
+
+ err_msg = ""
+ # Check exit codes and match regular expressions against stdout and
+ # stderr if this tool was configured to do so.
+ if ( len( self.tool.stdio_regexes ) > 0 or
+ len( self.tool.exit_codes ) > 0 ):
+ # We will check the exit code ranges in the order in which
+ # they were specified. Each exit_code is a ToolStdioExitCode
+ # that includes an applicable range. If the exit code was in
+ # that range, then apply the error level and add in a message.
+ # If we've reached a fatal error rule, then stop.
+ max_error_level = galaxy.tools.StdioErrorLevel.NO_ERROR
+ for exit_code in self.tool.stdio_exit_codes:
+ # TODO: Fetch the exit code from the .rc file:
+ tool_exit_code = 0
+ if ( tool_exit_code >= exit_code.range_start and
+ tool_exit_code <= exit_code.range_end ):
+ if None != exit_code.desc:
+ err_msg += exit_code.desc
+ # TODO: Find somewhere to stick the err_msg - possibly to
+ # the source (stderr/stdout), possibly in a new db column.
+ max_error_level = max( max_error_level,
+ exit_code.error_level )
+ if max_error_level >= galaxy.tools.StdioErrorLevel.FATAL:
+ break
+ # If there is a regular expression for scanning stdout/stderr,
+ # then we assume that the tool writer overwrote the default
+ # behavior of just setting an error if there is *anything* on
+ # stderr.
+ if max_error_level < galaxy.tools.StdioErrorLevel.FATAL:
+ # We'll examine every regex. Each regex specifies whether
+ # it is to be run on stdout, stderr, or both. (It is
+ # possible for neither stdout nor stderr to be scanned,
+ # but those won't be scanned.) We record the highest
+ # error level, which are currently "warning" and "fatal".
+ # If fatal, then we set the job's state to ERROR.
+ # If warning, then we still set the job's state to OK
+ # but include a message. We'll do this if we haven't seen
+ # a fatal error yet
+ for regex in self.tool.stdio_regexes:
+ # If ( this regex should be matched against stdout )
+ # - Run the regex's match pattern against stdout
+ # - If it matched, then determine the error level.
+ # o If it was fatal, then we're done - break.
+ # Repeat the stdout stuff for stderr.
+ # TODO: Collapse this into a single function.
+ if ( regex.stdout_match ):
+ regex_match = re.search( regex.match, stdout )
+ if ( regex_match ):
+ err_msg += self.regex_err_msg( regex_match, regex )
+ max_error_level = max( max_error_level, regex.error_level )
+ if max_error_level >= galaxy.tools.StdioErrorLevel.FATAL:
+ break
+ if ( regex.stderr_match ):
+ regex_match = re.search( regex.match, stderr )
+ if ( regex_match ):
+ err_msg += self.regex_err_msg( regex_match, regex )
+ max_error_level = max( max_error_level,
+ regex.error_level )
+ if max_error_level >= galaxy.tools.StdioErrorLevel.FATAL:
+ break
+ # If we encountered a fatal error, then we'll need to set the
+ # job state accordingly. Otherwise the job is ok:
+ if max_error_level >= galaxy.tools.StdioErrorLevel.FATAL:
+ job.state = job.states.ERROR
+ else:
+ job.state = job.states.OK
+ # When there are no regular expressions and no exit codes to check,
+ # default to the previous behavior: when there's anything on stderr
+ # the job has an error, and the job is ok otherwise.
else:
- job.state = job.states.OK
+ if stderr:
+ job.state = job.states.ERROR
+ else:
+ job.state = job.states.OK
if self.version_string_cmd:
version_filename = self.get_version_string_path()
if os.path.exists(version_filename):
@@ -330,6 +401,7 @@
return self.fail( "Job %s's output dataset(s) could not be read" % job.id )
job_context = ExpressionContext( dict( stdout = stdout, stderr = stderr ) )
job_tool = self.app.toolbox.tools_by_id.get( job.tool_id, None )
+
def in_directory( file, directory ):
# Make both absolute.
directory = os.path.abspath( directory )
@@ -370,7 +442,11 @@
# Update (non-library) job output datasets through the object store
if dataset not in job.output_library_datasets:
self.app.object_store.update_from_file(dataset.dataset, create=True)
- if context['stderr']:
+ # TODO: The context['stderr'] holds stderr's contents. An error
+ # only really occurs if the job also has an error. So check the
+ # job's state:
+ #if context['stderr']:
+ if job.states.ERROR == job.state:
dataset.blurb = "error"
elif dataset.has_data():
# If the tool was expected to set the extension, attempt to retrieve it
@@ -385,7 +461,14 @@
( not self.external_output_metadata.external_metadata_set_successfully( dataset, self.sa_session ) \
and self.app.config.retry_metadata_internally ):
dataset.set_meta( overwrite = False )
- elif not self.external_output_metadata.external_metadata_set_successfully( dataset, self.sa_session ) and not context['stderr']:
+ # TODO: The context['stderr'] used to indicate that there
+ # was an error. Now we must rely on the job's state instead;
+ # that indicates whether the tool relied on stderr to indicate
+ # the state or whether the tool used exit codes and regular
+ # expressions to do so. So we use
+ # job.state == job.states.ERROR to replace this same test.
+ #elif not self.external_output_metadata.external_metadata_set_successfully( dataset, self.sa_session ) and not context['stderr']:
+ elif not self.external_output_metadata.external_metadata_set_successfully( dataset, self.sa_session ) and job.states.ERROR != job.state:
dataset._state = model.Dataset.states.FAILED_METADATA
else:
#load metadata from file
@@ -415,7 +498,12 @@
if dataset.ext == 'auto':
dataset.extension = 'txt'
self.sa_session.add( dataset )
- if context['stderr']:
+ # TODO: job.states.ERROR == job.state now replaces checking
+ # stderr for a problem:
+ #if context['stderr']:
+ if job.states.ERROR == job.state:
+ log.debug( "setting dataset state to ERROR" )
+ # TODO: This is where the state is being set to error. Change it!
dataset_assoc.dataset.dataset.state = model.Dataset.states.ERROR
else:
dataset_assoc.dataset.dataset.state = model.Dataset.states.OK
@@ -480,6 +568,29 @@
if self.app.config.cleanup_job == 'always' or ( not stderr and self.app.config.cleanup_job == 'onsuccess' ):
self.cleanup()
+ def regex_err_msg( self, match, regex ):
+ """
+ Return a message about the match on tool output using the given
+ ToolStdioRegex regex object. The regex_match is a MatchObject
+ that will contain the string matched on.
+ """
+ # Get the description for the error level:
+ err_msg = galaxy.tools.StdioErrorLevel.desc( regex.error_level ) + ": "
+ # If there's a description for the regular expression, then use it.
+ # Otherwise, we'll take the first 256 characters of the match.
+ if None != regex.desc:
+ err_msg += regex.desc
+ else:
+ mstart = match.start()
+ mend = match.end()
+ err_msg += "Matched on "
+ # TODO: Move the constant 256 somewhere else besides here.
+ if mend - mstart > 256:
+ err_msg += match.string[ mstart : mstart+256 ] + "..."
+ else:
+ err_msg += match.string[ mstart: mend ]
+ return err_msg
+
def cleanup( self ):
# remove temporary files
try:
diff -r 7c495f835a1d436ad33dff6107784f106cc24980 -r 3b2dc0a51d147c3fc8f0560d8fd3ebefc0792e99 lib/galaxy/tools/__init__.py
--- a/lib/galaxy/tools/__init__.py
+++ b/lib/galaxy/tools/__init__.py
@@ -37,6 +37,23 @@
log = logging.getLogger( __name__ )
+# These determine stdio-based error levels from matching on regular expressions
+# and exit codes. They are meant to be used comparatively, such as showing
+# that warning < fatal. This is really meant to just be an enum.
+class StdioErrorLevel( object ):
+ NO_ERROR = 0
+ WARNING = 1
+ FATAL = 2
+ MAX = 2
+ descs = {NO_ERROR : 'No error', WARNING : 'Warning', FATAL : 'Fatal error'}
+ @staticmethod
+ def desc( error_level ):
+ err_msg = "Unknown error"
+ if ( error_level > 0 and
+ error_level <= StdioErrorLevel.MAX ):
+ err_msg = StdioErrorLevel.descs[ error_level ]
+ return err_msg
+
class ToolNotFoundException( Exception ):
pass
@@ -1140,6 +1157,12 @@
# a warning and skip to the next.
for exit_code_elem in ( stdio_elem.findall( "exit_code" ) ):
exit_code = ToolStdioExitCode()
+ # Each exit code has an optional description that can be
+ # part of the "desc" or "description" attributes:
+ exit_code.desc = exit_code_elem.get( "desc" )
+ if None == exit_code.desc:
+ exit_code.desc = exit_code_elem.get( "description" )
+ # Parse the error level:
exit_code.error_level = (
self.parse_error_level( exit_code_elem.get( "level" )))
code_range = exit_code_elem.get( "range", "" )
@@ -1155,11 +1178,9 @@
# X:Y - Split on the colon. We do not allow a colon
# without a beginning or end, though we could.
# Also note that whitespace is eliminated.
- # TODO: Turn this into a single match - it will be
- # more efficient
- string.strip( code_range )
+ # TODO: Turn this into a single match - it should be
+ # more efficient.
code_range = re.sub( "\s", "", code_range )
- log.debug( "Code range after sub: %s" % code_range )
code_ranges = re.split( ":", code_range )
if ( len( code_ranges ) == 2 ):
if ( None == code_ranges[0] or '' == code_ranges[0] ):
@@ -1216,6 +1237,12 @@
for regex_elem in ( stdio_elem.findall( "regex" ) ):
# TODO: Fill in ToolStdioRegex
regex = ToolStdioRegex()
+ # Each regex has an optional description that can be
+ # part of the "desc" or "description" attributes:
+ regex.desc = regex_elem.get( "desc" )
+ if None == regex.desc:
+ regex.desc = regex_elem.get( "description" )
+ # Parse the error level
regex.error_level = (
self.parse_error_level( regex_elem.get( "level" ) ) )
regex.match = regex_elem.get( "match", "" )
@@ -1243,9 +1270,9 @@
# and anything to do with "err". If neither stdout nor
# stderr were specified, then raise a warning and scan both.
for src in src_list:
- if re.match( "out", src, re.IGNORECASE ):
+ if re.search( "out", src, re.IGNORECASE ):
regex.stdout_match = True
- if re.match( "err", src, re.IGNORECASE ):
+ if re.search( "err", src, re.IGNORECASE ):
regex.stderr_match = True
if (not regex.stdout_match and not regex.stderr_match):
log.warning( "Unable to determine if tool stream "
@@ -1262,24 +1289,25 @@
trace_msg = repr( traceback.format_tb( trace ) )
log.error( "Traceback: %s" % trace_msg )
+ # TODO: This method doesn't have to be part of the Tool class.
def parse_error_level( self, err_level ):
"""
Return fatal or warning depending on what's in the error level.
This will assume that the error level fatal is returned if it's
- unparsable. (This doesn't have to be part of the Tool class.)
+ unparsable.
"""
# What should the default be? I'm claiming it should be fatal:
# if you went to the trouble to write the rule, then it's
# probably a problem. I think there are easily three substantial
# camps: make it fatal, make it a warning, or, if it's missing,
- # just throw an exception and ignore it.
- return_level = "fatal"
+ # just throw an exception and ignore the exit_code element.
+ return_level = StdioErrorLevel.FATAL
try:
if ( None != err_level ):
if ( re.search( "warning", err_level, re.IGNORECASE ) ):
- return_level = "warning"
+ return_level = StdioErrorLevel.WARNING
elif ( re.search( "fatal", err_level, re.IGNORECASE ) ):
- return_level = "fatal"
+ return_level = StdioErrorLevel.FATAL
except Exception, e:
log.error( "Exception in parse_error_level "
+ str(sys.exc_info() ) )
@@ -2333,16 +2361,18 @@
installed_tool_dependencies = self.tool_shed_repository.tool_dependencies
else:
installed_tool_dependencies = None
- for requirement in self.requirements:
- # TODO: currently only supporting requirements of type package,
- # need to implement some mechanism for mapping other types
- # back to packages
+ for requirement in self.requirements:
+ # TODO: currently only supporting requirements of type package,
+ # need to implement some mechanism for mapping other types
+ # back to packages
log.debug( "Building dependency shell command for dependency '%s'", requirement.name )
- if requirement.type == 'package':
+ if requirement.type == 'package':
script_file, base_path, version = self.app.toolbox.dependency_manager.find_dep( name=requirement.name,
version=requirement.version,
type=requirement.type,
installed_tool_dependencies=installed_tool_dependencies )
+ if requirement.type == 'package':
+ script_file, base_path, version = self.app.toolbox.dependency_manager.find_dep( requirement.name, requirement.version )
if script_file is None and base_path is None:
log.warn( "Failed to resolve dependency on '%s', ignoring", requirement.name )
elif script_file is None:
@@ -2617,7 +2647,7 @@
elif isinstance( input, SelectToolParameter ):
param_dict.update( { 'type' : 'select',
'html' : urllib.quote( input.get_html( trans ) ),
- 'options': input.static_options
+ 'options': input.static_options
} )
elif isinstance( input, Conditional ):
# TODO.
@@ -2626,7 +2656,7 @@
param_dict.update( { 'type' : 'number', 'init_value' : input.value,
'html' : urllib.quote( input.get_html( trans ) ),
'min': input.min,
- 'max': input.max,
+ 'max': input.max
'value': input.value
} )
else:
@@ -2798,6 +2828,7 @@
self.stderr_match = False
# TODO: Define a common class or constant for error level:
self.error_level = "fatal"
+ self.desc = ""
class ToolStdioExitCode( object ):
"""
@@ -2809,6 +2840,7 @@
self.range_end = float( "inf" )
# TODO: Define a common class or constant for error level:
self.error_level = "fatal"
+ self.desc = ""
class ToolParameterValueWrapper( object ):
"""
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.
1 new commit in galaxy-central:
https://bitbucket.org/galaxy/galaxy-central/changeset/87be4c8d7f7f/
changeset: 87be4c8d7f7f
user: afgane
date: 2012-06-18 06:53:02
summary: Fix API history purge; a few documentation corrections
affected #: 3 files
diff -r aaf5c82a55794c97125696aecb27056b5087fdb7 -r 87be4c8d7f7fa32f6023f0e251e3c9af8447f5cd lib/galaxy/web/api/histories.py
--- a/lib/galaxy/web/api/histories.py
+++ b/lib/galaxy/web/api/histories.py
@@ -130,6 +130,7 @@
history.deleted = True
if purge and trans.app.config.allow_user_dataset_purge:
+ # First purge all the datasets
for hda in history.datasets:
if hda.purged:
continue
@@ -143,6 +144,9 @@
except:
pass
trans.sa_session.flush()
+ # Now mark the history as purged
+ history.purged = True
+ self.sa_session.add( history )
trans.sa_session.flush()
return 'OK'
@@ -150,8 +154,8 @@
@web.expose_api
def undelete( self, trans, id, **kwd ):
"""
- POST /api/histories/deleted/{encoded_quota_id}/undelete
- Undeletes a quota
+ POST /api/histories/deleted/{encoded_history_id}/undelete
+ Undeletes a history
"""
history_id = id
history = self.get_history( trans, history_id, check_ownership=True, check_accessible=False, deleted=True )
diff -r aaf5c82a55794c97125696aecb27056b5087fdb7 -r 87be4c8d7f7fa32f6023f0e251e3c9af8447f5cd lib/galaxy/web/api/users.py
--- a/lib/galaxy/web/api/users.py
+++ b/lib/galaxy/web/api/users.py
@@ -65,7 +65,7 @@
@web.expose_api
def create( self, trans, payload, **kwd ):
"""
- /api/users
+ POST /api/users
Creates a new Galaxy user.
"""
if not trans.app.config.allow_user_creation:
diff -r aaf5c82a55794c97125696aecb27056b5087fdb7 -r 87be4c8d7f7fa32f6023f0e251e3c9af8447f5cd scripts/api/library_upload_from_import_dir.py
--- a/scripts/api/library_upload_from_import_dir.py
+++ b/scripts/api/library_upload_from_import_dir.py
@@ -1,5 +1,8 @@
#!/usr/bin/env python
-
+"""
+Example usage:
+./library_upload_from_import_dir.py <key> http://127.0.0.1:8080/api/libraries/dda47097d9189f15/contents Fdda47097d9189f15 auto /Users/EnisAfgan/projects/pprojects/galaxy/lib_upload_dir ?
+"""
import os, sys
sys.path.insert( 0, os.path.dirname( __file__ ) )
from common import submit
Repository URL: https://bitbucket.org/galaxy/galaxy-central/
--
This is a commit notification from bitbucket.org. You are receiving
this because you have the service enabled, addressing the recipient of
this email.