May 2014 - galaxy-commits - lists.galaxyproject.org

commit/galaxy-central: jmchilton: More performant database interaction during collection creation.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/1fae433a54ea/ Changeset: 1fae433a54ea User: jmchilton Date: 2014-05-19 19:15:48 Summary: More performant database interaction during collection creation. Affected #: 2 files diff -r 9452c068ec99c52c66dcb2bbb208787f8d379d22 -r 1fae433a54eaaab6392da9e9a8eec94a3b568c14 lib/galaxy/dataset_collections/__init__.py --- a/lib/galaxy/dataset_collections/__init__.py +++ b/lib/galaxy/dataset_collections/__init__.py @@ -180,7 +180,6 @@ return dataset_collection_instance def __recursively_create_collections( self, trans, element_identifiers ): - # TODO: Optimize - don't recheck parent, reload created model, just use as is. for index, element_identifier in enumerate( element_identifiers ): try: if not element_identifier[ "src" ] == "new_collection": @@ -197,9 +196,7 @@ collection_type=collection_type, element_identifiers=element_identifier[ "element_identifiers" ], ) - self.__persist( collection ) - element_identifier[ "src" ] = "dc" - element_identifier[ "id" ] = trans.security.encode_id( collection.id ) + element_identifier[ "__object__" ] = collection return element_identifiers @@ -215,7 +212,12 @@ # # consistent with other API methods though. # element_identifier = dict( src='hda', id=str( element_identifier ) ) - # dateset_identifier is dict {src=hda|ldda, id=<encoded_id>} + # Previously created collection already found in request, just pass + # through as is. + if "__object__" in element_identifier: + return element_identifier[ "__object__" ] + + # dateset_identifier is dict {src=hda|ldda|hdca|new_collection, id=<encoded_id>} try: src_type = element_identifier.get( 'src', 'hda' ) except AttributeError: @@ -233,8 +235,6 @@ # TODO: Option to copy? Force copy? Copy or allow if not owned? element = self.__get_history_collection_instance( trans, encoded_id ).collection # TODO: ldca. - elif src_type == "dc": - element = self.get_dataset_collection( trans, encoded_id ) else: raise RequestParameterInvalidException( "Unknown src_type parameter supplied '%s'." % src_type ) return element diff -r 9452c068ec99c52c66dcb2bbb208787f8d379d22 -r 1fae433a54eaaab6392da9e9a8eec94a3b568c14 lib/galaxy/dataset_collections/util.py --- a/lib/galaxy/dataset_collections/util.py +++ b/lib/galaxy/dataset_collections/util.py @@ -6,6 +6,7 @@ ERROR_MESSAGE_NO_NESTED_IDENTIFIERS = "Dataset source new_collection requires nested element_identifiers for new collection." ERROR_MESSAGE_NO_NAME = "Cannot load invalid dataset identifier - missing name - %s" ERROR_MESSAGE_NO_COLLECTION_TYPE = "No collection_type define for nested collection %s." +ERROR_MESSAGE_INVALID_PARAMETER_FOUND = "Found invalid parameter %s in element identifier description %s." def api_payload_to_create_params( payload ): @@ -32,6 +33,9 @@ and verify the structure is valid. """ for element_identifier in element_identifiers: + if "__object__" in element_identifier: + message = ERROR_MESSAGE_INVALID_PARAMETER_FOUND % ( "__model_object__", element_identifier ) + raise exceptions.RequestParameterInvalidException( message ) if "name" not in element_identifier: message = ERROR_MESSAGE_NO_NAME % element_identifier raise exceptions.RequestParameterInvalidException( message ) 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 0

commit/galaxy-central: jmchilton: Rework, improve validation of dataset creation requests.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/9452c068ec99/ Changeset: 9452c068ec99 User: jmchilton Date: 2014-05-19 18:51:07 Summary: Rework, improve validation of dataset creation requests. Being more thorough fixes some potential security problems and will allow some performance optimizations I'm planning. Affected #: 2 files diff -r c4404368f34df8089812f14dff298092e3ba7a81 -r 9452c068ec99c52c66dcb2bbb208787f8d379d22 lib/galaxy/dataset_collections/__init__.py --- a/lib/galaxy/dataset_collections/__init__.py +++ b/lib/galaxy/dataset_collections/__init__.py @@ -1,6 +1,8 @@ from .registry import DatasetCollectionTypesRegistry from .matching import MatchingCollections from .type_description import CollectionTypeDescriptionFactory +from .util import validate_input_element_identifiers + from galaxy import model from galaxy.exceptions import MessageException @@ -53,6 +55,8 @@ ): """ """ + if element_identifiers: + validate_input_element_identifiers( element_identifiers ) dataset_collection = self.__create_dataset_collection( trans=trans, collection_type=collection_type, @@ -99,6 +103,8 @@ if not collection_type: raise RequestParameterInvalidException( ERROR_NO_COLLECTION_TYPE ) collection_type_description = self.collection_type_descriptions.for_collection_type( collection_type ) + # If we have elements, this is an internal request, don't need to load + # objects from identifiers. if elements is None: if collection_type_description.has_subcollections( ): # Nested collection - recursively create collections and update identifiers. @@ -186,8 +192,6 @@ # element identifier is a dict with src new_collection... collection_type = element_identifier.get( "collection_type", None ) - if not collection_type: - raise RequestParameterInvalidException( "No collection_type define for nested collection." ) collection = self.__create_dataset_collection( trans=trans, collection_type=collection_type, @@ -202,10 +206,6 @@ def __load_elements( self, trans, element_identifiers ): elements = odict.odict() for element_identifier in element_identifiers: - if "name" not in element_identifier: - raise RequestParameterInvalidException( - "Cannot load invalid dataset identifier - missing name - %s" % element_identifier - ) elements[ element_identifier[ "name" ] ] = self.__load_element( trans, element_identifier ) return elements @@ -234,8 +234,6 @@ element = self.__get_history_collection_instance( trans, encoded_id ).collection # TODO: ldca. elif src_type == "dc": - # TODO: Force only used internally during nested creation so no - # need to recheck security. element = self.get_dataset_collection( trans, encoded_id ) else: raise RequestParameterInvalidException( "Unknown src_type parameter supplied '%s'." % src_type ) diff -r c4404368f34df8089812f14dff298092e3ba7a81 -r 9452c068ec99c52c66dcb2bbb208787f8d379d22 lib/galaxy/dataset_collections/util.py --- a/lib/galaxy/dataset_collections/util.py +++ b/lib/galaxy/dataset_collections/util.py @@ -2,6 +2,11 @@ from galaxy import web from galaxy import model +ERROR_MESSAGE_UNKNOWN_SRC = "Unknown dataset source (src) %s." +ERROR_MESSAGE_NO_NESTED_IDENTIFIERS = "Dataset source new_collection requires nested element_identifiers for new collection." +ERROR_MESSAGE_NO_NAME = "Cannot load invalid dataset identifier - missing name - %s" +ERROR_MESSAGE_NO_COLLECTION_TYPE = "No collection_type define for nested collection %s." + def api_payload_to_create_params( payload ): """ @@ -22,6 +27,28 @@ return params +def validate_input_element_identifiers( element_identifiers ): + """ Scan through the list of element identifiers supplied by the API consumer + and verify the structure is valid. + """ + for element_identifier in element_identifiers: + if "name" not in element_identifier: + message = ERROR_MESSAGE_NO_NAME % element_identifier + raise exceptions.RequestParameterInvalidException( message ) + src = element_identifier.get( "src", "hda" ) + if src not in [ "hda", "hdca", "ldda", "new_collection" ]: + message = ERROR_MESSAGE_UNKNOWN_SRC % src + raise exceptions.RequestParameterInvalidException( message ) + if src == "new_collection": + if "element_identifiers" not in element_identifier: + message = ERROR_MESSAGE_NO_NESTED_IDENTIFIERS + raise exceptions.RequestParameterInvalidException( ERROR_MESSAGE_NO_NESTED_IDENTIFIERS ) + if "collection_type" not in element_identifier: + message = ERROR_MESSAGE_NO_COLLECTION_TYPE % element_identifier + raise exceptions.RequestParameterInvalidException( message ) + validate_input_element_identifiers( element_identifier[ "element_identifiers" ] ) + + def dictify_dataset_collection_instance( dataset_colleciton_instance, parent, security, view="element" ): dict_value = dataset_colleciton_instance.to_dict( view=view ) encoded_id = security.encode_id( dataset_colleciton_instance.id ) 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 0

commit/galaxy-central: dannon: Followup from commentary on PR394, cleaned up incorrect log.exception usage in base controller.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/c4404368f34d/ Changeset: c4404368f34d User: dannon Date: 2014-05-19 18:47:10 Summary: Followup from commentary on PR394, cleaned up incorrect log.exception usage in base controller. Affected #: 1 file diff -r 9afc3e4a3561166beb69d344178338fa954ad5ee -r c4404368f34df8089812f14dff298092e3ba7a81 lib/galaxy/web/base/controller.py --- a/lib/galaxy/web/base/controller.py +++ b/lib/galaxy/web/base/controller.py @@ -125,8 +125,8 @@ assert item_class is not None item = trans.sa_session.query( item_class ).get( decoded_id ) assert item is not None - except Exception, exc: - log.exception( "Invalid %s id ( %s ) specified: %s" % ( class_name, id, str( exc ) ) ) + except Exception: + log.exception( "Invalid %s id ( %s ) specified." % ( class_name, id ) ) raise MessageException( "Invalid %s id ( %s ) specified" % ( class_name, id ), type="error" ) if check_ownership or check_accessible: @@ -204,7 +204,7 @@ except MessageException, e: raise HTTPBadRequest( detail=e.err_msg ) except Exception, e: - log.exception( "Exception in get_object check for %s %s: %s" % ( class_name, str( id ), str( e ) ) ) + log.exception( "Exception in get_object check for %s %s." % ( class_name, str( id ) ) ) raise HTTPInternalServerError( comment=str( e ) ) def validate_in_users_and_groups( self, trans, payload ): @@ -796,8 +796,8 @@ if expose_dataset_path: try: hda_dict[ 'file_name' ] = hda.file_name - except objectstore.ObjectNotFound, onf: - log.exception( 'objectstore.ObjectNotFound, HDA %s: %s', hda.id, onf ) + except objectstore.ObjectNotFound: + log.exception( 'objectstore.ObjectNotFound, HDA %s.', hda.id ) hda_dict[ 'download_url' ] = url_for( 'history_contents_display', history_id = trans.security.encode_id( hda.history.id ), 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 0

commit/galaxy-central: dannon: Merge PR394 from stable.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/9afc3e4a3561/ Changeset: 9afc3e4a3561 User: dannon Date: 2014-05-19 18:39:59 Summary: Merge PR394 from stable. Affected #: 2 files diff -r 40c02d5cfc7da40d6b844a8120a019d611462e02 -r 9afc3e4a3561166beb69d344178338fa954ad5ee lib/galaxy/web/base/controller.py --- a/lib/galaxy/web/base/controller.py +++ b/lib/galaxy/web/base/controller.py @@ -188,7 +188,7 @@ except MessageException: raise # handled in the caller except: - log.exception( "Execption in get_object check for %s %s:" % ( class_name, str( id ) ) ) + log.exception( "Exception in get_object check for %s %s:" % ( class_name, str( id ) ) ) raise Exception( 'Server error retrieving %s id ( %s ).' % ( class_name, str( id ) ) ) @@ -204,7 +204,7 @@ except MessageException, e: raise HTTPBadRequest( detail=e.err_msg ) except Exception, e: - log.exception( "Execption in get_object check for %s %s: %s" % ( class_name, str( id ), str( e ) ) ) + log.exception( "Exception in get_object check for %s %s: %s" % ( class_name, str( id ), str( e ) ) ) raise HTTPInternalServerError( comment=str( e ) ) def validate_in_users_and_groups( self, trans, payload ): @@ -2771,7 +2771,7 @@ return self.get_object( trans, id, 'Quota', check_ownership=False, check_accessible=False, deleted=deleted ) -class UsesTagsMixin( object ): +class UsesTagsMixin( SharableItemSecurityMixin ): def get_tag_handler( self, trans ): return trans.app.tag_handler diff -r 40c02d5cfc7da40d6b844a8120a019d611462e02 -r 9afc3e4a3561166beb69d344178338fa954ad5ee lib/galaxy/webapps/galaxy/api/item_tags.py --- a/lib/galaxy/webapps/galaxy/api/item_tags.py +++ b/lib/galaxy/webapps/galaxy/api/item_tags.py @@ -4,7 +4,7 @@ import logging from galaxy import web from galaxy.web.base.controller import BaseAPIController, UsesTagsMixin -from paste.httpexceptions import HTTPNotImplemented, HTTPBadRequest +from paste.httpexceptions import HTTPBadRequest log = logging.getLogger( __name__ ) 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 0

commit/galaxy-central: nsoranzo: Fix for "'HistoryTagsController' object has no attribute 'security_check'".
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/68a8b0397947/ Changeset: 68a8b0397947 Branch: stable User: nsoranzo Date: 2014-05-19 18:03:21 Summary: Fix for "'HistoryTagsController' object has no attribute 'security_check'". Trying to access /api/histories/$history_id/tags gives this error: Execption in get_object check for History 1723db999e478a37: 'HistoryTagsController' object has no attribute 'security_check' Traceback (most recent call last): File "/opt/galaxy/lib/galaxy/web/base/controller.py", line 233, in get_object check_ownership=check_ownership, check_accessible=check_accessible, deleted=deleted ) File "/opt/galaxy/lib/galaxy/web/base/controller.py", line 131, in get_object self.security_check( trans, item, check_ownership, check_accessible ) AttributeError: 'HistoryTagsController' object has no attribute 'security_check' The same happens for workflow tags. Affected #: 2 files diff -r cc0fe62e54dc59607c26cd9152ef1ea7c85af625 -r 68a8b0397947c732b28207d465d3f3c4e2a7a8a0 lib/galaxy/web/base/controller.py --- a/lib/galaxy/web/base/controller.py +++ b/lib/galaxy/web/base/controller.py @@ -221,7 +221,7 @@ except MessageException: raise # handled in the caller except: - log.exception( "Execption in get_object check for %s %s:" % ( class_name, str( id ) ) ) + log.exception( "Exception in get_object check for %s %s:" % ( class_name, str( id ) ) ) raise Exception( 'Server error retrieving %s id ( %s ).' % ( class_name, str( id ) ) ) @@ -237,7 +237,7 @@ except MessageException, e: raise HTTPBadRequest( detail=e.err_msg ) except Exception, e: - log.exception( "Execption in get_object check for %s %s: %s" % ( class_name, str( id ), str( e ) ) ) + log.exception( "Exception in get_object check for %s %s: %s" % ( class_name, str( id ), str( e ) ) ) raise HTTPInternalServerError( comment=str( e ) ) def validate_in_users_and_groups( self, trans, payload ): @@ -2779,7 +2779,7 @@ return self.get_object( trans, id, 'Quota', check_ownership=False, check_accessible=False, deleted=deleted ) -class UsesTagsMixin( object ): +class UsesTagsMixin( SharableItemSecurityMixin ): def get_tag_handler( self, trans ): return trans.app.tag_handler diff -r cc0fe62e54dc59607c26cd9152ef1ea7c85af625 -r 68a8b0397947c732b28207d465d3f3c4e2a7a8a0 lib/galaxy/webapps/galaxy/api/item_tags.py --- a/lib/galaxy/webapps/galaxy/api/item_tags.py +++ b/lib/galaxy/webapps/galaxy/api/item_tags.py @@ -4,7 +4,7 @@ import logging from galaxy import web from galaxy.web.base.controller import BaseAPIController, UsesTagsMixin -from paste.httpexceptions import HTTPNotImplemented, HTTPBadRequest +from paste.httpexceptions import HTTPBadRequest log = logging.getLogger( __name__ ) 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 0

commit/galaxy-central: jmchilton: More reworking of job logging.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/40c02d5cfc7d/ Changeset: 40c02d5cfc7d User: jmchilton Date: 2014-05-19 18:02:59 Summary: More reworking of job logging. Drop now duplicated log message in drmaa, condor, cli, etc.... Have local job runner report which job script it is running - but log it as a job script. For each job runner log the full command-line which includes dependency resolution and not the base version produced directly by the tool and use the old format from the external runners. Affected #: 7 files diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/command_factory.py --- a/lib/galaxy/jobs/command_factory.py +++ b/lib/galaxy/jobs/command_factory.py @@ -19,8 +19,8 @@ - commands to set metadata (if include_metadata is True) """ base_command_line = job_wrapper.get_command_line() - job_id = job_wrapper.job_id - log.debug( 'Tool evaluation for job (%s) produced command-line: %s' % ( job_id, base_command_line ) ) + # job_id = job_wrapper.job_id + # log.debug( 'Tool evaluation for job (%s) produced command-line: %s' % ( job_id, base_command_line ) ) commands_builder = CommandsBuilder(base_command_line) # All job runners currently handle this case which should never occur diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/runners/__init__.py --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -268,8 +268,8 @@ ) ## Additional logging to enable if debugging from_work_dir handling, metadata ## commands, etc... (or just peak in the job script.) - # job_id = job_wrapper.job_id - # log.debug( 'Extended command-line for job (%s) is: %s' % ( job_id, command_line ) ) + job_id = job_wrapper.job_id + log.debug( '(%s) command is: %s' % ( job_id, command_line ) ) options.update(**kwds) return job_script(**options) diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/runners/cli.py --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -91,7 +91,6 @@ return log.debug( "(%s) submitting file: %s" % ( galaxy_id_tag, ajs.job_file ) ) - log.debug( "(%s) command is: %s" % ( galaxy_id_tag, command_line ) ) cmd_out = shell.execute(job_interface.submit(ajs.job_file)) if cmd_out.returncode != 0: diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/runners/condor.py --- a/lib/galaxy/jobs/runners/condor.py +++ b/lib/galaxy/jobs/runners/condor.py @@ -120,7 +120,6 @@ return log.debug( "(%s) submitting file %s" % ( galaxy_id_tag, executable ) ) - log.debug( "(%s) command is: %s" % ( galaxy_id_tag, command_line ) ) external_job_id, message = condor_submit(submit_file) if external_job_id is None: diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/runners/drmaa.py --- a/lib/galaxy/jobs/runners/drmaa.py +++ b/lib/galaxy/jobs/runners/drmaa.py @@ -114,9 +114,6 @@ if not self.prepare_job( job_wrapper, include_metadata=True ): return - # command line has been added to the wrapper by prepare_job() - command_line = job_wrapper.runner_command_line - # get configured job destination job_destination = job_wrapper.job_destination @@ -164,7 +161,6 @@ return log.debug( "(%s) submitting file %s", galaxy_id_tag, ajs.job_file ) - log.debug( "(%s) command is: %s", galaxy_id_tag, command_line ) if native_spec: log.debug( "(%s) native specification is: %s", galaxy_id_tag, native_spec ) diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -88,6 +88,7 @@ try: stdout_file = tempfile.NamedTemporaryFile( suffix='_stdout', dir=job_wrapper.working_directory ) stderr_file = tempfile.NamedTemporaryFile( suffix='_stderr', dir=job_wrapper.working_directory ) + log.debug( '(%s) executing job script: %s' % ( job_id, command_line ) ) proc = subprocess.Popen( args=command_line, shell=True, cwd=job_wrapper.working_directory, diff -r f73317a85e5cc38492e0338538ada18484701030 -r 40c02d5cfc7da40d6b844a8120a019d611462e02 lib/galaxy/jobs/runners/pbs.py --- a/lib/galaxy/jobs/runners/pbs.py +++ b/lib/galaxy/jobs/runners/pbs.py @@ -317,7 +317,6 @@ # (if a TaskWrapper was passed in): galaxy_job_id = job_wrapper.get_id_tag() log.debug("(%s) submitting file %s" % ( galaxy_job_id, job_file ) ) - log.debug("(%s) command is: %s" % ( galaxy_job_id, command_line ) ) tries = 0 while tries < 5: 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 0

commit/galaxy-central: jmchilton: Improved job logging.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/f73317a85e5c/ Changeset: f73317a85e5c User: jmchilton Date: 2014-05-19 17:16:12 Summary: Improved job logging. Remove now less useful local runner command-line log statement (still contend it will be valuable for debugging purposes to be able to tweak and rerun the exact script - but point taken :)) . Remove stray statement added in ff4b5fe. Add new logging statement that logs just the base command-line produced by tools for all job runners. Add little note to runners/__init__.py about how to produce an extended command-line logging (with metadata handling, dependencies, and from_work_dir handling). Affected #: 4 files diff -r 42ced7ee518c0035dba9514b255ca1dbb8dd4c1a -r f73317a85e5cc38492e0338538ada18484701030 lib/galaxy/datatypes/metadata.py --- a/lib/galaxy/datatypes/metadata.py +++ b/lib/galaxy/datatypes/metadata.py @@ -658,7 +658,6 @@ __get_filename_override(), metadata_path_on_compute(metadata_files.filename_override_metadata), ) - log.info(line) return line if not isinstance( datasets, list ): datasets = [ datasets ] diff -r 42ced7ee518c0035dba9514b255ca1dbb8dd4c1a -r f73317a85e5cc38492e0338538ada18484701030 lib/galaxy/jobs/command_factory.py --- a/lib/galaxy/jobs/command_factory.py +++ b/lib/galaxy/jobs/command_factory.py @@ -4,6 +4,9 @@ CAPTURE_RETURN_CODE = "return_code=$?" YIELD_CAPTURED_CODE = 'sh -c "exit $return_code"' +from logging import getLogger +log = getLogger( __name__ ) + def build_command( runner, job_wrapper, include_metadata=False, include_work_dir_outputs=True, remote_command_params={} ): """ @@ -15,8 +18,10 @@ - command line taken from job wrapper - commands to set metadata (if include_metadata is True) """ - - commands_builder = CommandsBuilder(job_wrapper.get_command_line()) + base_command_line = job_wrapper.get_command_line() + job_id = job_wrapper.job_id + log.debug( 'Tool evaluation for job (%s) produced command-line: %s' % ( job_id, base_command_line ) ) + commands_builder = CommandsBuilder(base_command_line) # All job runners currently handle this case which should never occur if not commands_builder.commands: diff -r 42ced7ee518c0035dba9514b255ca1dbb8dd4c1a -r f73317a85e5cc38492e0338538ada18484701030 lib/galaxy/jobs/runners/__init__.py --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -151,7 +151,11 @@ # Prepare the job try: job_wrapper.prepare() - job_wrapper.runner_command_line = self.build_command_line( job_wrapper, include_metadata=include_metadata, include_work_dir_outputs=include_work_dir_outputs ) + job_wrapper.runner_command_line = self.build_command_line( + job_wrapper, + include_metadata=include_metadata, + include_work_dir_outputs=include_work_dir_outputs + ) except: log.exception("(%s) Failure preparing job" % job_id) job_wrapper.fail( "failure preparing job", exception=True ) @@ -254,13 +258,18 @@ envs = destination.get( "env", [] ) for env in envs: env_setup_commands.append( env_to_statement( env ) ) + command_line = job_wrapper.runner_command_line options = dict( job_instrumenter=job_instrumenter, galaxy_lib=job_wrapper.galaxy_lib_dir, env_setup_commands=env_setup_commands, working_directory=os.path.abspath( job_wrapper.working_directory ), - command=job_wrapper.runner_command_line, + command=command_line, ) + ## Additional logging to enable if debugging from_work_dir handling, metadata + ## commands, etc... (or just peak in the job script.) + # job_id = job_wrapper.job_id + # log.debug( 'Extended command-line for job (%s) is: %s' % ( job_id, command_line ) ) options.update(**kwds) return job_script(**options) diff -r 42ced7ee518c0035dba9514b255ca1dbb8dd4c1a -r f73317a85e5cc38492e0338538ada18484701030 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -86,7 +86,6 @@ job_id = job_wrapper.get_id_tag() try: - log.debug( '(%s) executing: %s' % ( job_id, command_line ) ) stdout_file = tempfile.NamedTemporaryFile( suffix='_stdout', dir=job_wrapper.working_directory ) stderr_file = tempfile.NamedTemporaryFile( suffix='_stderr', dir=job_wrapper.working_directory ) proc = subprocess.Popen( args=command_line, 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 0

commit/galaxy-central: dan: Fixes for genomespace exporter recieving UnvalidatedValue objects.
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/42ced7ee518c/ Changeset: 42ced7ee518c User: dan Date: 2014-05-19 16:17:36 Summary: Fixes for genomespace exporter recieving UnvalidatedValue objects. Affected #: 1 file diff -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 -r 42ced7ee518c0035dba9514b255ca1dbb8dd4c1a tools/genomespace/genomespace_exporter.py --- a/tools/genomespace/genomespace_exporter.py +++ b/tools/genomespace/genomespace_exporter.py @@ -120,7 +120,13 @@ def galaxy_code_get_genomespace_folders( genomespace_site='prod', trans=None, value=None, base_url=None, **kwd ): if value: - value = value[0]#single select, only 1 value + if isinstance( value, list ): + value = value[0] #single select, only 1 value + elif not isinstance( value, basestring ): + #unvalidated value + value = value.value + if isinstance( value, list ): + value = value[0] #single select, only 1 value def recurse_directory_dict( url_opener, cur_options, url ): cur_directory = urllib2.Request( url, headers = { 'Content-Type': 'application/json', 'Accept': 'application/json, text/plain' } ) cur_directory.get_method = lambda: 'GET' 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 0

commit/galaxy-central: jmchilton: Merged in jmchilton/galaxy-central-fork-1 (pull request #384)
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

1 new commit in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/5bf39d99f789/ Changeset: 5bf39d99f789 User: jmchilton Date: 2014-05-19 04:12:36 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #384) Local Job Runner Enhancements Affected #: 5 files diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1163,6 +1163,11 @@ return 'Job ran longer than the maximum allowed execution time (%s), please try different inputs or parameters' % self.app.job_config.limits.walltime return None + def has_limits( self ): + has_output_limit = self.app.job_config.limits.output_size > 0 + has_walltime_limit = self.app.job_config.limits.walltime_delta is not None + return has_output_limit or has_walltime_limit + def get_command_line( self ): return self.command_line diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 lib/galaxy/jobs/runners/__init__.py --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -196,7 +196,7 @@ output_pairs = [] # Walk job's output associations to find and use from_work_dir attributes. job = job_wrapper.get_job() - job_tool = self.app.toolbox.tools_by_id.get( job.tool_id, None ) + job_tool = job_wrapper.tool for dataset_assoc in job.output_datasets + job.output_library_datasets: for dataset in dataset_assoc.dataset.dataset.history_associations + dataset_assoc.dataset.dataset.library_associations: if isinstance( dataset, self.app.model.HistoryDatasetAssociation ): @@ -269,7 +269,36 @@ self.work_queue.put( ( self.finish_job, ajs ) ) -class AsynchronousJobState( object ): +class JobState( object ): + """ + Encapsulate state of jobs. + """ + + def set_defaults( self, files_dir ): + if self.job_wrapper is not None: + id_tag = self.job_wrapper.get_id_tag() + if files_dir is not None: + self.job_file = JobState.default_job_file( files_dir, id_tag ) + self.output_file = os.path.join( files_dir, 'galaxy_%s.o' % id_tag ) + self.error_file = os.path.join( files_dir, 'galaxy_%s.e' % id_tag ) + self.exit_code_file = os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) + job_name = 'g%s' % id_tag + if self.job_wrapper.tool.old_id: + job_name += '_%s' % self.job_wrapper.tool.old_id + if self.job_wrapper.user: + job_name += '_%s' % self.job_wrapper.user + self.job_name = ''.join( map( lambda x: x if x in ( string.letters + string.digits + '_' ) else '_', job_name ) ) + + @staticmethod + def default_job_file( files_dir, id_tag ): + return os.path.join( files_dir, 'galaxy_%s.sh' % id_tag ) + + @staticmethod + def default_exit_code_file( files_dir, id_tag ): + return os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) + + +class AsynchronousJobState( JobState ): """ Encapsulate the state of an asynchronous job, this should be subclassed as needed for various job runners to capture additional information needed @@ -296,21 +325,6 @@ self.cleanup_file_attributes = [ 'job_file', 'output_file', 'error_file', 'exit_code_file' ] - def set_defaults( self, files_dir ): - if self.job_wrapper is not None: - id_tag = self.job_wrapper.get_id_tag() - if files_dir is not None: - self.job_file = os.path.join( files_dir, 'galaxy_%s.sh' % id_tag ) - self.output_file = os.path.join( files_dir, 'galaxy_%s.o' % id_tag ) - self.error_file = os.path.join( files_dir, 'galaxy_%s.e' % id_tag ) - self.exit_code_file = os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) - job_name = 'g%s' % id_tag - if self.job_wrapper.tool.old_id: - job_name += '_%s' % self.job_wrapper.tool.old_id - if self.job_wrapper.user: - job_name += '_%s' % self.job_wrapper.user - self.job_name = ''.join( map( lambda x: x if x in ( string.letters + string.digits + '_' ) else '_', job_name ) ) - def cleanup( self ): for file in [ getattr( self, a ) for a in self.cleanup_file_attributes if hasattr( self, a ) ]: try: diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -11,13 +11,20 @@ from time import sleep from galaxy import model -from galaxy.jobs.runners import BaseJobRunner +from ..runners import BaseJobRunner +from ..runners import JobState from galaxy.util import DATABASE_MAX_STRING_SIZE, shrink_stream_by_size +from galaxy.util import asbool log = logging.getLogger( __name__ ) __all__ = [ 'LocalJobRunner' ] +DEFAULT_POOL_SLEEP_TIME = 1 +# TODO: Set to false and just get rid of this option. It would simplify this +# class nicely. -John +DEFAULT_EMBED_METADATA_IN_JOB = False + class LocalJobRunner( BaseJobRunner ): """ @@ -31,12 +38,6 @@ #create a local copy of os.environ to use as env for subprocess.Popen self._environ = os.environ.copy() - # put lib into the PYTHONPATH for subprocesses - if 'PYTHONPATH' in self._environ: - self._environ['PYTHONPATH'] = '%s:%s' % ( self._environ['PYTHONPATH'], os.path.abspath( 'lib' ) ) - else: - self._environ['PYTHONPATH'] = os.path.abspath( 'lib' ) - #Set TEMP if a valid temp value is not already set if not ( 'TMPDIR' in self._environ or 'TEMP' in self._environ or 'TMP' in self._environ ): self._environ[ 'TEMP' ] = os.path.abspath(tempfile.gettempdir()) @@ -52,24 +53,36 @@ ## slots would be cleaner name, but don't want deployers to see examples and think it ## is going to work with other job runners. slots = job_wrapper.job_destination.params.get( "local_slots", None ) - command_line = command_line.lstrip( " ;" ) if slots: - command_line = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED; %s' % ( int( slots ), command_line ) + slots_statement = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED;' % ( int( slots ) ) else: - command_line = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS; %s' % command_line - return command_line + slots_statement = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS;' + + job_id = job_wrapper.get_id_tag() + job_file = JobState.default_job_file( job_wrapper.working_directory, job_id ) + exit_code_path = JobState.default_exit_code_file( job_wrapper.working_directory, job_id ) + job_script_props = { + 'slots_statement': slots_statement, + 'command': command_line, + 'exit_code_path': exit_code_path, + 'working_directory': job_wrapper.working_directory, + } + job_file_contents = self.get_job_file( job_wrapper, **job_script_props ) + open( job_file, 'w' ).write( job_file_contents ) + os.chmod( job_file, 0755 ) + return job_file, exit_code_path def queue_job( self, job_wrapper ): # prepare the job - if not self.prepare_job( job_wrapper ): + include_metadata = asbool( job_wrapper.job_destination.params.get( "embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB ) ) + if not self.prepare_job( job_wrapper, include_metadata=include_metadata ): return stderr = stdout = '' exit_code = 0 # command line has been added to the wrapper by prepare_job() - command_line = self.__command_line( job_wrapper ) - + command_line, exit_code_path = self.__command_line( job_wrapper ) job_id = job_wrapper.get_id_tag() try: @@ -85,22 +98,18 @@ preexec_fn=os.setpgrp ) job_wrapper.set_job_destination(job_wrapper.job_destination, proc.pid) job_wrapper.change_state( model.Job.states.RUNNING ) - job_start = datetime.datetime.now() - i = 0 - # Iterate until the process exits, periodically checking its limits - while proc.poll() is None: - i += 1 - if (i % 20) == 0: - msg = job_wrapper.check_limits(runtime=datetime.datetime.now() - job_start) - if msg is not None: - job_wrapper.fail(msg) - log.debug('(%s) Terminating process group' % job_id) - self._terminate(proc) - return - else: - sleep(1) + + terminated = self.__poll_if_needed( proc, job_wrapper, job_id ) + if terminated: + return + # Reap the process and get the exit code. exit_code = proc.wait() + try: + exit_code = int( open( exit_code_path, 'r' ).read() ) + except Exception: + log.warn( "Failed to read exit code from path %s" % exit_code_path ) + pass stdout_file.seek( 0 ) stderr_file.seek( 0 ) stdout = shrink_stream_by_size( stdout_file, DATABASE_MAX_STRING_SIZE, join_by="\n..\n", left_larger=True, beginning_on_size_error=True ) @@ -112,7 +121,9 @@ log.exception("failure running job %d" % job_wrapper.job_id) job_wrapper.fail( "failure running job", exception=True ) return - self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + external_metadata = not asbool( job_wrapper.job_destination.params.get( "embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB ) ) + if external_metadata: + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) # Finish the job! try: job_wrapper.finish( stdout, stderr, exit_code ) @@ -168,3 +179,23 @@ if proc.poll() is None: os.killpg( proc.pid, 9 ) return proc.wait() # reap + + def __poll_if_needed( self, proc, job_wrapper, job_id ): + # Only poll if needed (i.e. job limits are set) + if not job_wrapper.has_limits(): + return + + job_start = datetime.datetime.now() + i = 0 + # Iterate until the process exits, periodically checking its limits + while proc.poll() is None: + i += 1 + if (i % 20) == 0: + msg = job_wrapper.check_limits(runtime=datetime.datetime.now() - job_start) + if msg is not None: + job_wrapper.fail(msg) + log.debug('(%s) Terminating process group' % job_id) + self._terminate(proc) + return True + else: + sleep( DEFAULT_POOL_SLEEP_TIME ) diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 test/unit/jobs/test_runner_local.py --- /dev/null +++ b/test/unit/jobs/test_runner_local.py @@ -0,0 +1,166 @@ +import os +import threading +import time +from unittest import TestCase + +from galaxy.util import bunch +from galaxy.jobs.runners import local +from galaxy.jobs import metrics +from galaxy import model + +from tools_support import ( + UsesApp, + UsesTools +) + + +class TestLocalJobRunner( TestCase, UsesApp, UsesTools ): + + def setUp( self ): + self.setup_app() + self._init_tool() + self.app.job_metrics = metrics.JobMetrics() + self.job_wrapper = MockJobWrapper( self.app, self.test_directory, self.tool ) + + def tearDown( self ): + self.tear_down_app() + + def test_run( self ): + self.job_wrapper.command_line = "echo HelloWorld" + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "HelloWorld" + + def test_galaxy_lib_on_path( self ): + self.job_wrapper.command_line = '''python -c "import galaxy.util"''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.exit_code == 0 + + def test_default_slots( self ): + self.job_wrapper.command_line = '''echo $GALAXY_SLOTS''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "1" + + def test_slots_override( self ): + # Set local_slots in job destination to specify slots for + # local job runner. + self.job_wrapper.job_destination.params[ "local_slots" ] = 3 + self.job_wrapper.command_line = '''echo $GALAXY_SLOTS''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "3" + + def test_exit_code( self ): + self.job_wrapper.command_line = '''sh -c "exit 4"''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.exit_code == 4 + + def test_metadata_gets_set( self ): + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert os.path.exists( self.job_wrapper.mock_metadata_path ) + + def test_metadata_gets_set_if_embedded( self ): + self.job_wrapper.job_destination.params[ "embed_metadata_in_job" ] = "True" + + # Kill off cruft for _handle_metadata_externally and make sure job stil works... + self.job_wrapper.external_output_metadata = None + self.app.datatypes_registry.set_external_metadata_tool = None + + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert os.path.exists( self.job_wrapper.mock_metadata_path ) + + def test_stopping_job( self ): + self.job_wrapper.command_line = '''python -c "import time; time.sleep(15)"''' + runner = local.LocalJobRunner( self.app, 1 ) + + def queue(): + runner.queue_job( self.job_wrapper ) + + t = threading.Thread(target=queue) + t.start() + while True: + if self.job_wrapper.external_id: + break + time.sleep( .01 ) + external_id = self.job_wrapper.external_id + mock_job = bunch.Bunch( + get_external_output_metadata=lambda: None, + get_job_runner_external_id=lambda: str(external_id), + get_id=lambda: 1 + ) + runner.stop_job( mock_job ) + t.join(1) + + +class MockJobWrapper( object ): + + def __init__( self, app, test_directory, tool ): + working_directory = os.path.join( test_directory, "workdir" ) + os.makedirs( working_directory ) + self.app = app + self.tool = tool + self.state = model.Job.states.QUEUED + self.command_line = "echo HelloWorld" + self.prepare_called = False + self.write_version_cmd = None + self.dependency_shell_commands = None + self.working_directory = working_directory + self.requires_setting_metadata = True + self.job_destination = bunch.Bunch( id="default", params={} ) + self.galaxy_lib_dir = os.path.abspath( "lib" ) + self.job_id = 1 + self.external_id = None + self.output_paths = [ '/tmp/output1.dat' ] + self.mock_metadata_path = os.path.abspath( os.path.join( test_directory, "METADATA_SET" ) ) + self.metadata_command = "touch %s" % self.mock_metadata_path + + # Cruft for setting metadata externally, axe at some point. + self.external_output_metadata = bunch.Bunch( + set_job_runner_external_pid=lambda pid, session: None + ) + self.app.datatypes_registry.set_external_metadata_tool = bunch.Bunch( + build_dependency_shell_commands=lambda: [] + ) + + def prepare( self ): + self.prepare_called = True + + def set_job_destination( self, job_destination, external_id ): + self.external_id = external_id + + def get_command_line( self ): + return self.command_line + + def get_id_tag( self ): + return "1" + + def get_state( self ): + return self.state + + def change_state( self, state ): + self.state = state + + def get_output_fnames( self ): + return [] + + def get_job( self ): + return model.Job() + + def setup_external_metadata( self, **kwds ): + return self.metadata_command + + def get_env_setup_clause( self ): + return "" + + def has_limits( self ): + return False + + def finish( self, stdout, stderr, exit_code ): + self.stdout = stdout + self.stderr = stderr + self.exit_code = exit_code diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 test/unit/tools_support.py --- a/test/unit/tools_support.py +++ b/test/unit/tools_support.py @@ -58,7 +58,7 @@ class UsesTools( object ): - def _init_tool( self, tool_contents ): + def _init_tool( self, tool_contents=SIMPLE_TOOL_CONTENTS ): self.tool_file = os.path.join( self.test_directory, "tool.xml" ) self.app.config.drmaa_external_runjob_script = "" self.app.config.tool_secret = "testsecret" 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 0

commit/galaxy-central: 7 new changesets
by commits-noreply＠bitbucket.org 19 May '14

19 May '14

7 new commits in galaxy-central: https://bitbucket.org/galaxy/galaxy-central/commits/bed23d528080/ Changeset: bed23d528080 User: jmchilton Date: 2014-05-05 19:35:09 Summary: Simplify command for grabbing tool in get_work_dir_outputs. Subtlely reduces code duplication and makes runners slightly easier to unit test (fewer dependencies). Affected #: 1 file diff -r b94b4da49bfdf4f204d0ad2dce878a37e6e4f3d3 -r bed23d5280808dbc2d6b56ddcfe572e9bedda97a lib/galaxy/jobs/runners/__init__.py --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -196,7 +196,7 @@ output_pairs = [] # Walk job's output associations to find and use from_work_dir attributes. job = job_wrapper.get_job() - job_tool = self.app.toolbox.tools_by_id.get( job.tool_id, None ) + job_tool = job_wrapper.tool for dataset_assoc in job.output_datasets + job.output_library_datasets: for dataset in dataset_assoc.dataset.dataset.history_associations + dataset_assoc.dataset.dataset.library_associations: if isinstance( dataset, self.app.model.HistoryDatasetAssociation ): https://bitbucket.org/galaxy/galaxy-central/commits/51c248b80e72/ Changeset: 51c248b80e72 User: jmchilton Date: 2014-05-05 19:35:09 Summary: Unit tests for local job runner. Test running, capturing standard output and exit code, killing job, setting metadata, adding Galaxy to jobs PYTHONPATH, GALAXY_SLOTS logic (default and overidding). Affected #: 2 files diff -r bed23d5280808dbc2d6b56ddcfe572e9bedda97a -r 51c248b80e729ec6a3ef39e851ebdf79ef888476 test/unit/jobs/test_runner_local.py --- /dev/null +++ b/test/unit/jobs/test_runner_local.py @@ -0,0 +1,155 @@ +import os +import threading +import time +from unittest import TestCase + +from galaxy.util import bunch +from galaxy.jobs.runners import local +from galaxy.jobs import metrics +from galaxy import model + +from tools_support import ( + UsesApp, + UsesTools +) + + +class TestLocalJobRunner( TestCase, UsesApp, UsesTools ): + + def setUp( self ): + self.setup_app() + self._init_tool() + self.app.job_metrics = metrics.JobMetrics() + self.job_wrapper = MockJobWrapper( self.app, self.test_directory, self.tool ) + + def tearDown( self ): + self.tear_down_app() + + def test_run( self ): + self.job_wrapper.command_line = "echo HelloWorld" + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "HelloWorld" + + def test_galaxy_lib_on_path( self ): + self.job_wrapper.command_line = '''python -c "import galaxy.util"''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.exit_code == 0 + + def test_default_slots( self ): + self.job_wrapper.command_line = '''echo $GALAXY_SLOTS''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "1" + + def test_slots_override( self ): + # Set local_slots in job destination to specify slots for + # local job runner. + self.job_wrapper.job_destination.params[ "local_slots" ] = 3 + self.job_wrapper.command_line = '''echo $GALAXY_SLOTS''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "3" + + def test_exit_code( self ): + self.job_wrapper.command_line = '''sh -c "exit 4"''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.exit_code == 4 + + def test_metadata_gets_set( self ): + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert os.path.exists( self.job_wrapper.mock_metadata_path ) + + def test_stopping_job( self ): + self.job_wrapper.command_line = '''python -c "import time; time.sleep(15)"''' + runner = local.LocalJobRunner( self.app, 1 ) + + def queue(): + runner.queue_job( self.job_wrapper ) + + t = threading.Thread(target=queue) + t.start() + while True: + if self.job_wrapper.external_id: + break + time.sleep( .01 ) + external_id = self.job_wrapper.external_id + mock_job = bunch.Bunch( + get_external_output_metadata=lambda: None, + get_job_runner_external_id=lambda: str(external_id), + get_id=lambda: 1 + ) + runner.stop_job( mock_job ) + t.join(1) + + +class MockJobWrapper( object ): + + def __init__( self, app, test_directory, tool ): + working_directory = os.path.join( test_directory, "workdir" ) + os.makedirs( working_directory ) + self.app = app + self.tool = tool + self.state = model.Job.states.QUEUED + self.command_line = "echo HelloWorld" + self.prepare_called = False + self.write_version_cmd = None + self.dependency_shell_commands = None + self.working_directory = working_directory + self.requires_setting_metadata = True + self.job_destination = bunch.Bunch( id="default", params={} ) + self.galaxy_lib_dir = os.path.abspath( "lib" ) + self.job_id = 1 + self.external_id = None + self.output_paths = [ '/tmp/output1.dat' ] + self.mock_metadata_path = os.path.abspath( os.path.join( test_directory, "METADATA_SET" ) ) + self.metadata_command = "touch %s" % self.mock_metadata_path + + # Cruft for setting metadata externally, axe at some point. + self.external_output_metadata = bunch.Bunch( + set_job_runner_external_pid=lambda pid, session: None + ) + self.app.datatypes_registry.set_external_metadata_tool = bunch.Bunch( + build_dependency_shell_commands=lambda: [] + ) + + def prepare( self ): + self.prepare_called = True + + def set_job_destination( self, job_destination, external_id ): + self.external_id = external_id + + def get_command_line( self ): + return self.command_line + + def get_id_tag( self ): + return "1" + + def get_state( self ): + return self.state + + def change_state( self, state ): + self.state = state + + def get_output_fnames( self ): + return [] + + def get_job( self ): + return model.Job() + + def setup_external_metadata( self, **kwds ): + return self.metadata_command + + def get_env_setup_clause( self ): + return "" + + def has_limits( self ): + return False + + def finish( self, stdout, stderr, exit_code ): + self.stdout = stdout + self.stderr = stderr + self.exit_code = exit_code diff -r bed23d5280808dbc2d6b56ddcfe572e9bedda97a -r 51c248b80e729ec6a3ef39e851ebdf79ef888476 test/unit/tools_support.py --- a/test/unit/tools_support.py +++ b/test/unit/tools_support.py @@ -58,7 +58,7 @@ class UsesTools( object ): - def _init_tool( self, tool_contents ): + def _init_tool( self, tool_contents=SIMPLE_TOOL_CONTENTS ): self.tool_file = os.path.join( self.test_directory, "tool.xml" ) self.app.config.drmaa_external_runjob_script = "" self.app.config.tool_secret = "testsecret" https://bitbucket.org/galaxy/galaxy-central/commits/ff666103459d/ Changeset: ff666103459d User: jmchilton Date: 2014-05-05 19:35:09 Summary: Refactor small __poll helper method out of local job runner's queue. Affected #: 1 file diff -r 51c248b80e729ec6a3ef39e851ebdf79ef888476 -r ff666103459de3c87595a1d25783d7a3283b4951 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -18,6 +18,8 @@ __all__ = [ 'LocalJobRunner' ] +DEFAULT_POOL_SLEEP_TIME = 1 + class LocalJobRunner( BaseJobRunner ): """ @@ -85,20 +87,11 @@ preexec_fn=os.setpgrp ) job_wrapper.set_job_destination(job_wrapper.job_destination, proc.pid) job_wrapper.change_state( model.Job.states.RUNNING ) - job_start = datetime.datetime.now() - i = 0 - # Iterate until the process exits, periodically checking its limits - while proc.poll() is None: - i += 1 - if (i % 20) == 0: - msg = job_wrapper.check_limits(runtime=datetime.datetime.now() - job_start) - if msg is not None: - job_wrapper.fail(msg) - log.debug('(%s) Terminating process group' % job_id) - self._terminate(proc) - return - else: - sleep(1) + + terminated = self.__poll( proc, job_wrapper, job_id ) + if terminated: + return + # Reap the process and get the exit code. exit_code = proc.wait() stdout_file.seek( 0 ) @@ -168,3 +161,19 @@ if proc.poll() is None: os.killpg( proc.pid, 9 ) return proc.wait() # reap + + def __poll( self, proc, job_wrapper, job_id ): + job_start = datetime.datetime.now() + i = 0 + # Iterate until the process exits, periodically checking its limits + while proc.poll() is None: + i += 1 + if (i % 20) == 0: + msg = job_wrapper.check_limits(runtime=datetime.datetime.now() - job_start) + if msg is not None: + job_wrapper.fail(msg) + log.debug('(%s) Terminating process group' % job_id) + self._terminate(proc) + return True + else: + sleep( DEFAULT_POOL_SLEEP_TIME ) https://bitbucket.org/galaxy/galaxy-central/commits/50de2360e451/ Changeset: 50de2360e451 User: jmchilton Date: 2014-05-05 19:35:09 Summary: Local job runner - only poll if limits set (so can be set). Small optimization aimed at speeding up tests - unit and functional. Affected #: 2 files diff -r ff666103459de3c87595a1d25783d7a3283b4951 -r 50de2360e4511b91f22ed437c1d1d62826136342 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1161,6 +1161,11 @@ return 'Job ran longer than the maximum allowed execution time (%s), please try different inputs or parameters' % self.app.job_config.limits.walltime return None + def has_limits( self ): + has_output_limit = self.app.job_config.limits.output_size > 0 + has_walltime_limit = self.app.job_config.limits.walltime_delta is not None + return has_output_limit or has_walltime_limit + def get_command_line( self ): return self.command_line diff -r ff666103459de3c87595a1d25783d7a3283b4951 -r 50de2360e4511b91f22ed437c1d1d62826136342 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -88,7 +88,7 @@ job_wrapper.set_job_destination(job_wrapper.job_destination, proc.pid) job_wrapper.change_state( model.Job.states.RUNNING ) - terminated = self.__poll( proc, job_wrapper, job_id ) + terminated = self.__poll_if_needed( proc, job_wrapper, job_id ) if terminated: return @@ -162,7 +162,11 @@ os.killpg( proc.pid, 9 ) return proc.wait() # reap - def __poll( self, proc, job_wrapper, job_id ): + def __poll_if_needed( self, proc, job_wrapper, job_id ): + # Only poll if needed (i.e. job limits are set) + if not job_wrapper.has_limits(): + return + job_start = datetime.datetime.now() i = 0 # Iterate until the process exits, periodically checking its limits https://bitbucket.org/galaxy/galaxy-central/commits/03d912919356/ Changeset: 03d912919356 User: jmchilton Date: 2014-05-05 19:35:09 Summary: Allow embedding setting of metadata as part of local jobs. (Like it works for all other runners except the LWR). This is essentially scaffolding toward elimination of extra metadata step in local runner. Got Dan's blessing to kill this off a while ago but I still don't quite understand why it is separated out - so doing this in two parts. This changeset changes nothing by default but adds an option to local runner destinations (embed_metadata_in_job) to allow them operate in the new fashion (embed the metadata setting step in the job itself). At some later point this option should just be removed and all local jobs should operate this way. Use this new option to speed up the local job runner or to test out how this runner will work in the future. Affected #: 2 files diff -r 50de2360e4511b91f22ed437c1d1d62826136342 -r 03d912919356db7ff34d289985c8ee3ac2a67dd1 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -13,12 +13,16 @@ from galaxy import model from galaxy.jobs.runners import BaseJobRunner from galaxy.util import DATABASE_MAX_STRING_SIZE, shrink_stream_by_size +from galaxy.util import asbool log = logging.getLogger( __name__ ) __all__ = [ 'LocalJobRunner' ] DEFAULT_POOL_SLEEP_TIME = 1 +# TODO: Set to false and just get rid of this option. It would simplify this +# class nicely. -John +DEFAULT_EMBED_METADATA_IN_JOB = False class LocalJobRunner( BaseJobRunner ): @@ -63,7 +67,8 @@ def queue_job( self, job_wrapper ): # prepare the job - if not self.prepare_job( job_wrapper ): + include_metadata = asbool( job_wrapper.job_destination.params.get( "embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB ) ) + if not self.prepare_job( job_wrapper, include_metadata=include_metadata ): return stderr = stdout = '' @@ -105,7 +110,9 @@ log.exception("failure running job %d" % job_wrapper.job_id) job_wrapper.fail( "failure running job", exception=True ) return - self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + external_metadata = not asbool( job_wrapper.job_destination.params.get( "embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB ) ) + if external_metadata: + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) # Finish the job! try: job_wrapper.finish( stdout, stderr, exit_code ) diff -r 50de2360e4511b91f22ed437c1d1d62826136342 -r 03d912919356db7ff34d289985c8ee3ac2a67dd1 test/unit/jobs/test_runner_local.py --- a/test/unit/jobs/test_runner_local.py +++ b/test/unit/jobs/test_runner_local.py @@ -63,6 +63,17 @@ runner.queue_job( self.job_wrapper ) assert os.path.exists( self.job_wrapper.mock_metadata_path ) + def test_metadata_gets_set_if_embedded( self ): + self.job_wrapper.job_destination.params[ "embed_metadata_in_job" ] = "True" + + # Kill off cruft for _handle_metadata_externally and make sure job stil works... + self.job_wrapper.external_output_metadata = None + self.app.datatypes_registry.set_external_metadata_tool = None + + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert os.path.exists( self.job_wrapper.mock_metadata_path ) + def test_stopping_job( self ): self.job_wrapper.command_line = '''python -c "import time; time.sleep(15)"''' runner = local.LocalJobRunner( self.app, 1 ) https://bitbucket.org/galaxy/galaxy-central/commits/e1db77201a5d/ Changeset: e1db77201a5d User: jmchilton Date: 2014-05-05 19:35:09 Summary: Rework local job runner to create job script. Affected #: 2 files diff -r 03d912919356db7ff34d289985c8ee3ac2a67dd1 -r e1db77201a5d9e93cf7729c2695c34e98abf5252 lib/galaxy/jobs/runners/__init__.py --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -270,7 +270,36 @@ self.work_queue.put( ( self.finish_job, ajs ) ) -class AsynchronousJobState( object ): +class JobState( object ): + """ + Encapsulate state of jobs. + """ + + def set_defaults( self, files_dir ): + if self.job_wrapper is not None: + id_tag = self.job_wrapper.get_id_tag() + if files_dir is not None: + self.job_file = JobState.default_job_file( files_dir, id_tag ) + self.output_file = os.path.join( files_dir, 'galaxy_%s.o' % id_tag ) + self.error_file = os.path.join( files_dir, 'galaxy_%s.e' % id_tag ) + self.exit_code_file = os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) + job_name = 'g%s' % id_tag + if self.job_wrapper.tool.old_id: + job_name += '_%s' % self.job_wrapper.tool.old_id + if self.job_wrapper.user: + job_name += '_%s' % self.job_wrapper.user + self.job_name = ''.join( map( lambda x: x if x in ( string.letters + string.digits + '_' ) else '_', job_name ) ) + + @staticmethod + def default_job_file( files_dir, id_tag ): + return os.path.join( files_dir, 'galaxy_%s.sh' % id_tag ) + + @staticmethod + def default_exit_code_file( files_dir, id_tag ): + return os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) + + +class AsynchronousJobState( JobState ): """ Encapsulate the state of an asynchronous job, this should be subclassed as needed for various job runners to capture additional information needed @@ -297,21 +326,6 @@ self.cleanup_file_attributes = [ 'job_file', 'output_file', 'error_file', 'exit_code_file' ] - def set_defaults( self, files_dir ): - if self.job_wrapper is not None: - id_tag = self.job_wrapper.get_id_tag() - if files_dir is not None: - self.job_file = os.path.join( files_dir, 'galaxy_%s.sh' % id_tag ) - self.output_file = os.path.join( files_dir, 'galaxy_%s.o' % id_tag ) - self.error_file = os.path.join( files_dir, 'galaxy_%s.e' % id_tag ) - self.exit_code_file = os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) - job_name = 'g%s' % id_tag - if self.job_wrapper.tool.old_id: - job_name += '_%s' % self.job_wrapper.tool.old_id - if self.job_wrapper.user: - job_name += '_%s' % self.job_wrapper.user - self.job_name = ''.join( map( lambda x: x if x in ( string.letters + string.digits + '_' ) else '_', job_name ) ) - def cleanup( self ): for file in [ getattr( self, a ) for a in self.cleanup_file_attributes if hasattr( self, a ) ]: try: diff -r 03d912919356db7ff34d289985c8ee3ac2a67dd1 -r e1db77201a5d9e93cf7729c2695c34e98abf5252 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -11,7 +11,8 @@ from time import sleep from galaxy import model -from galaxy.jobs.runners import BaseJobRunner +from ..runners import BaseJobRunner +from ..runners import JobState from galaxy.util import DATABASE_MAX_STRING_SIZE, shrink_stream_by_size from galaxy.util import asbool @@ -37,12 +38,6 @@ #create a local copy of os.environ to use as env for subprocess.Popen self._environ = os.environ.copy() - # put lib into the PYTHONPATH for subprocesses - if 'PYTHONPATH' in self._environ: - self._environ['PYTHONPATH'] = '%s:%s' % ( self._environ['PYTHONPATH'], os.path.abspath( 'lib' ) ) - else: - self._environ['PYTHONPATH'] = os.path.abspath( 'lib' ) - #Set TEMP if a valid temp value is not already set if not ( 'TMPDIR' in self._environ or 'TEMP' in self._environ or 'TMP' in self._environ ): self._environ[ 'TEMP' ] = tempfile.gettempdir() @@ -58,12 +53,24 @@ ## slots would be cleaner name, but don't want deployers to see examples and think it ## is going to work with other job runners. slots = job_wrapper.job_destination.params.get( "local_slots", None ) - command_line = command_line.lstrip( " ;" ) if slots: - command_line = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED; %s' % ( int( slots ), command_line ) + slots_statement = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED;' % ( int( slots ) ) else: - command_line = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS; %s' % command_line - return command_line + slots_statement = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS;' + + job_id = job_wrapper.get_id_tag() + job_file = JobState.default_job_file( job_wrapper.working_directory, job_id ) + exit_code_path = JobState.default_exit_code_file( job_wrapper.working_directory, job_id ) + job_script_props = { + 'slots_statement': slots_statement, + 'command': command_line, + 'exit_code_path': exit_code_path, + 'working_directory': job_wrapper.working_directory, + } + job_file_contents = self.get_job_file( job_wrapper, **job_script_props ) + open( job_file, 'w' ).write( job_file_contents ) + os.chmod( job_file, 0755 ) + return job_file, exit_code_path def queue_job( self, job_wrapper ): # prepare the job @@ -75,8 +82,7 @@ exit_code = 0 # command line has been added to the wrapper by prepare_job() - command_line = self.__command_line( job_wrapper ) - + command_line, exit_code_path = self.__command_line( job_wrapper ) job_id = job_wrapper.get_id_tag() try: @@ -99,6 +105,11 @@ # Reap the process and get the exit code. exit_code = proc.wait() + try: + exit_code = int( open( exit_code_path, 'r' ).read() ) + except Exception: + log.warn( "Failed to read exit code from path %s" % exit_code_path ) + pass stdout_file.seek( 0 ) stderr_file.seek( 0 ) stdout = shrink_stream_by_size( stdout_file, DATABASE_MAX_STRING_SIZE, join_by="\n..\n", left_larger=True, beginning_on_size_error=True ) https://bitbucket.org/galaxy/galaxy-central/commits/5bf39d99f789/ Changeset: 5bf39d99f789 User: jmchilton Date: 2014-05-19 04:12:36 Summary: Merged in jmchilton/galaxy-central-fork-1 (pull request #384) Local Job Runner Enhancements Affected #: 5 files diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 lib/galaxy/jobs/__init__.py --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1163,6 +1163,11 @@ return 'Job ran longer than the maximum allowed execution time (%s), please try different inputs or parameters' % self.app.job_config.limits.walltime return None + def has_limits( self ): + has_output_limit = self.app.job_config.limits.output_size > 0 + has_walltime_limit = self.app.job_config.limits.walltime_delta is not None + return has_output_limit or has_walltime_limit + def get_command_line( self ): return self.command_line diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 lib/galaxy/jobs/runners/__init__.py --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -196,7 +196,7 @@ output_pairs = [] # Walk job's output associations to find and use from_work_dir attributes. job = job_wrapper.get_job() - job_tool = self.app.toolbox.tools_by_id.get( job.tool_id, None ) + job_tool = job_wrapper.tool for dataset_assoc in job.output_datasets + job.output_library_datasets: for dataset in dataset_assoc.dataset.dataset.history_associations + dataset_assoc.dataset.dataset.library_associations: if isinstance( dataset, self.app.model.HistoryDatasetAssociation ): @@ -269,7 +269,36 @@ self.work_queue.put( ( self.finish_job, ajs ) ) -class AsynchronousJobState( object ): +class JobState( object ): + """ + Encapsulate state of jobs. + """ + + def set_defaults( self, files_dir ): + if self.job_wrapper is not None: + id_tag = self.job_wrapper.get_id_tag() + if files_dir is not None: + self.job_file = JobState.default_job_file( files_dir, id_tag ) + self.output_file = os.path.join( files_dir, 'galaxy_%s.o' % id_tag ) + self.error_file = os.path.join( files_dir, 'galaxy_%s.e' % id_tag ) + self.exit_code_file = os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) + job_name = 'g%s' % id_tag + if self.job_wrapper.tool.old_id: + job_name += '_%s' % self.job_wrapper.tool.old_id + if self.job_wrapper.user: + job_name += '_%s' % self.job_wrapper.user + self.job_name = ''.join( map( lambda x: x if x in ( string.letters + string.digits + '_' ) else '_', job_name ) ) + + @staticmethod + def default_job_file( files_dir, id_tag ): + return os.path.join( files_dir, 'galaxy_%s.sh' % id_tag ) + + @staticmethod + def default_exit_code_file( files_dir, id_tag ): + return os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) + + +class AsynchronousJobState( JobState ): """ Encapsulate the state of an asynchronous job, this should be subclassed as needed for various job runners to capture additional information needed @@ -296,21 +325,6 @@ self.cleanup_file_attributes = [ 'job_file', 'output_file', 'error_file', 'exit_code_file' ] - def set_defaults( self, files_dir ): - if self.job_wrapper is not None: - id_tag = self.job_wrapper.get_id_tag() - if files_dir is not None: - self.job_file = os.path.join( files_dir, 'galaxy_%s.sh' % id_tag ) - self.output_file = os.path.join( files_dir, 'galaxy_%s.o' % id_tag ) - self.error_file = os.path.join( files_dir, 'galaxy_%s.e' % id_tag ) - self.exit_code_file = os.path.join( files_dir, 'galaxy_%s.ec' % id_tag ) - job_name = 'g%s' % id_tag - if self.job_wrapper.tool.old_id: - job_name += '_%s' % self.job_wrapper.tool.old_id - if self.job_wrapper.user: - job_name += '_%s' % self.job_wrapper.user - self.job_name = ''.join( map( lambda x: x if x in ( string.letters + string.digits + '_' ) else '_', job_name ) ) - def cleanup( self ): for file in [ getattr( self, a ) for a in self.cleanup_file_attributes if hasattr( self, a ) ]: try: diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 lib/galaxy/jobs/runners/local.py --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -11,13 +11,20 @@ from time import sleep from galaxy import model -from galaxy.jobs.runners import BaseJobRunner +from ..runners import BaseJobRunner +from ..runners import JobState from galaxy.util import DATABASE_MAX_STRING_SIZE, shrink_stream_by_size +from galaxy.util import asbool log = logging.getLogger( __name__ ) __all__ = [ 'LocalJobRunner' ] +DEFAULT_POOL_SLEEP_TIME = 1 +# TODO: Set to false and just get rid of this option. It would simplify this +# class nicely. -John +DEFAULT_EMBED_METADATA_IN_JOB = False + class LocalJobRunner( BaseJobRunner ): """ @@ -31,12 +38,6 @@ #create a local copy of os.environ to use as env for subprocess.Popen self._environ = os.environ.copy() - # put lib into the PYTHONPATH for subprocesses - if 'PYTHONPATH' in self._environ: - self._environ['PYTHONPATH'] = '%s:%s' % ( self._environ['PYTHONPATH'], os.path.abspath( 'lib' ) ) - else: - self._environ['PYTHONPATH'] = os.path.abspath( 'lib' ) - #Set TEMP if a valid temp value is not already set if not ( 'TMPDIR' in self._environ or 'TEMP' in self._environ or 'TMP' in self._environ ): self._environ[ 'TEMP' ] = os.path.abspath(tempfile.gettempdir()) @@ -52,24 +53,36 @@ ## slots would be cleaner name, but don't want deployers to see examples and think it ## is going to work with other job runners. slots = job_wrapper.job_destination.params.get( "local_slots", None ) - command_line = command_line.lstrip( " ;" ) if slots: - command_line = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED; %s' % ( int( slots ), command_line ) + slots_statement = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED;' % ( int( slots ) ) else: - command_line = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS; %s' % command_line - return command_line + slots_statement = 'GALAXY_SLOTS="1"; export GALAXY_SLOTS;' + + job_id = job_wrapper.get_id_tag() + job_file = JobState.default_job_file( job_wrapper.working_directory, job_id ) + exit_code_path = JobState.default_exit_code_file( job_wrapper.working_directory, job_id ) + job_script_props = { + 'slots_statement': slots_statement, + 'command': command_line, + 'exit_code_path': exit_code_path, + 'working_directory': job_wrapper.working_directory, + } + job_file_contents = self.get_job_file( job_wrapper, **job_script_props ) + open( job_file, 'w' ).write( job_file_contents ) + os.chmod( job_file, 0755 ) + return job_file, exit_code_path def queue_job( self, job_wrapper ): # prepare the job - if not self.prepare_job( job_wrapper ): + include_metadata = asbool( job_wrapper.job_destination.params.get( "embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB ) ) + if not self.prepare_job( job_wrapper, include_metadata=include_metadata ): return stderr = stdout = '' exit_code = 0 # command line has been added to the wrapper by prepare_job() - command_line = self.__command_line( job_wrapper ) - + command_line, exit_code_path = self.__command_line( job_wrapper ) job_id = job_wrapper.get_id_tag() try: @@ -85,22 +98,18 @@ preexec_fn=os.setpgrp ) job_wrapper.set_job_destination(job_wrapper.job_destination, proc.pid) job_wrapper.change_state( model.Job.states.RUNNING ) - job_start = datetime.datetime.now() - i = 0 - # Iterate until the process exits, periodically checking its limits - while proc.poll() is None: - i += 1 - if (i % 20) == 0: - msg = job_wrapper.check_limits(runtime=datetime.datetime.now() - job_start) - if msg is not None: - job_wrapper.fail(msg) - log.debug('(%s) Terminating process group' % job_id) - self._terminate(proc) - return - else: - sleep(1) + + terminated = self.__poll_if_needed( proc, job_wrapper, job_id ) + if terminated: + return + # Reap the process and get the exit code. exit_code = proc.wait() + try: + exit_code = int( open( exit_code_path, 'r' ).read() ) + except Exception: + log.warn( "Failed to read exit code from path %s" % exit_code_path ) + pass stdout_file.seek( 0 ) stderr_file.seek( 0 ) stdout = shrink_stream_by_size( stdout_file, DATABASE_MAX_STRING_SIZE, join_by="\n..\n", left_larger=True, beginning_on_size_error=True ) @@ -112,7 +121,9 @@ log.exception("failure running job %d" % job_wrapper.job_id) job_wrapper.fail( "failure running job", exception=True ) return - self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) + external_metadata = not asbool( job_wrapper.job_destination.params.get( "embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB ) ) + if external_metadata: + self._handle_metadata_externally( job_wrapper, resolve_requirements=True ) # Finish the job! try: job_wrapper.finish( stdout, stderr, exit_code ) @@ -168,3 +179,23 @@ if proc.poll() is None: os.killpg( proc.pid, 9 ) return proc.wait() # reap + + def __poll_if_needed( self, proc, job_wrapper, job_id ): + # Only poll if needed (i.e. job limits are set) + if not job_wrapper.has_limits(): + return + + job_start = datetime.datetime.now() + i = 0 + # Iterate until the process exits, periodically checking its limits + while proc.poll() is None: + i += 1 + if (i % 20) == 0: + msg = job_wrapper.check_limits(runtime=datetime.datetime.now() - job_start) + if msg is not None: + job_wrapper.fail(msg) + log.debug('(%s) Terminating process group' % job_id) + self._terminate(proc) + return True + else: + sleep( DEFAULT_POOL_SLEEP_TIME ) diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 test/unit/jobs/test_runner_local.py --- /dev/null +++ b/test/unit/jobs/test_runner_local.py @@ -0,0 +1,166 @@ +import os +import threading +import time +from unittest import TestCase + +from galaxy.util import bunch +from galaxy.jobs.runners import local +from galaxy.jobs import metrics +from galaxy import model + +from tools_support import ( + UsesApp, + UsesTools +) + + +class TestLocalJobRunner( TestCase, UsesApp, UsesTools ): + + def setUp( self ): + self.setup_app() + self._init_tool() + self.app.job_metrics = metrics.JobMetrics() + self.job_wrapper = MockJobWrapper( self.app, self.test_directory, self.tool ) + + def tearDown( self ): + self.tear_down_app() + + def test_run( self ): + self.job_wrapper.command_line = "echo HelloWorld" + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "HelloWorld" + + def test_galaxy_lib_on_path( self ): + self.job_wrapper.command_line = '''python -c "import galaxy.util"''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.exit_code == 0 + + def test_default_slots( self ): + self.job_wrapper.command_line = '''echo $GALAXY_SLOTS''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "1" + + def test_slots_override( self ): + # Set local_slots in job destination to specify slots for + # local job runner. + self.job_wrapper.job_destination.params[ "local_slots" ] = 3 + self.job_wrapper.command_line = '''echo $GALAXY_SLOTS''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.stdout.strip() == "3" + + def test_exit_code( self ): + self.job_wrapper.command_line = '''sh -c "exit 4"''' + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert self.job_wrapper.exit_code == 4 + + def test_metadata_gets_set( self ): + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert os.path.exists( self.job_wrapper.mock_metadata_path ) + + def test_metadata_gets_set_if_embedded( self ): + self.job_wrapper.job_destination.params[ "embed_metadata_in_job" ] = "True" + + # Kill off cruft for _handle_metadata_externally and make sure job stil works... + self.job_wrapper.external_output_metadata = None + self.app.datatypes_registry.set_external_metadata_tool = None + + runner = local.LocalJobRunner( self.app, 1 ) + runner.queue_job( self.job_wrapper ) + assert os.path.exists( self.job_wrapper.mock_metadata_path ) + + def test_stopping_job( self ): + self.job_wrapper.command_line = '''python -c "import time; time.sleep(15)"''' + runner = local.LocalJobRunner( self.app, 1 ) + + def queue(): + runner.queue_job( self.job_wrapper ) + + t = threading.Thread(target=queue) + t.start() + while True: + if self.job_wrapper.external_id: + break + time.sleep( .01 ) + external_id = self.job_wrapper.external_id + mock_job = bunch.Bunch( + get_external_output_metadata=lambda: None, + get_job_runner_external_id=lambda: str(external_id), + get_id=lambda: 1 + ) + runner.stop_job( mock_job ) + t.join(1) + + +class MockJobWrapper( object ): + + def __init__( self, app, test_directory, tool ): + working_directory = os.path.join( test_directory, "workdir" ) + os.makedirs( working_directory ) + self.app = app + self.tool = tool + self.state = model.Job.states.QUEUED + self.command_line = "echo HelloWorld" + self.prepare_called = False + self.write_version_cmd = None + self.dependency_shell_commands = None + self.working_directory = working_directory + self.requires_setting_metadata = True + self.job_destination = bunch.Bunch( id="default", params={} ) + self.galaxy_lib_dir = os.path.abspath( "lib" ) + self.job_id = 1 + self.external_id = None + self.output_paths = [ '/tmp/output1.dat' ] + self.mock_metadata_path = os.path.abspath( os.path.join( test_directory, "METADATA_SET" ) ) + self.metadata_command = "touch %s" % self.mock_metadata_path + + # Cruft for setting metadata externally, axe at some point. + self.external_output_metadata = bunch.Bunch( + set_job_runner_external_pid=lambda pid, session: None + ) + self.app.datatypes_registry.set_external_metadata_tool = bunch.Bunch( + build_dependency_shell_commands=lambda: [] + ) + + def prepare( self ): + self.prepare_called = True + + def set_job_destination( self, job_destination, external_id ): + self.external_id = external_id + + def get_command_line( self ): + return self.command_line + + def get_id_tag( self ): + return "1" + + def get_state( self ): + return self.state + + def change_state( self, state ): + self.state = state + + def get_output_fnames( self ): + return [] + + def get_job( self ): + return model.Job() + + def setup_external_metadata( self, **kwds ): + return self.metadata_command + + def get_env_setup_clause( self ): + return "" + + def has_limits( self ): + return False + + def finish( self, stdout, stderr, exit_code ): + self.stdout = stdout + self.stderr = stderr + self.exit_code = exit_code diff -r 632087d479859d0a8152267f6c21a623722bc875 -r 5bf39d99f7892bd8d9733f5efd68a8bb356c7bd9 test/unit/tools_support.py --- a/test/unit/tools_support.py +++ b/test/unit/tools_support.py @@ -58,7 +58,7 @@ class UsesTools( object ): - def _init_tool( self, tool_contents ): + def _init_tool( self, tool_contents=SIMPLE_TOOL_CONTENTS ): self.tool_file = os.path.join( self.test_directory, "tool.xml" ) self.app.config.drmaa_external_runjob_script = "" self.app.config.tool_secret = "testsecret" 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 0