diff --git a/.ci/pep8_sources.txt b/.ci/pep8_sources.txt index 885d355e12a..0a42b946330 100644 --- a/.ci/pep8_sources.txt +++ b/.ci/pep8_sources.txt @@ -7,7 +7,7 @@ lib/galaxy/datatypes/{assembly,binary,checkers,chrominfo,coverage,data,proteomic lib/galaxy/datatypes/converters/{bedgraph_to_array_tree_converter,bgzip,wiggle_to_array_tree_converter}.py lib/galaxy/exceptions/ lib/galaxy/external_services/ -lib/galaxy/jobs/{__init__,error_level,manager,stock_rules}.py +lib/galaxy/jobs/ lib/galaxy/managers/ lib/galaxy/model/orm/{engine_factory,now}.py lib/galaxy/model/{base,custom_types,search,util}.py diff --git a/lib/galaxy/jobs/deferred/__init__.py b/lib/galaxy/jobs/deferred/__init__.py index 23380fac36c..0810ed76ba4 100644 --- a/lib/galaxy/jobs/deferred/__init__.py +++ b/lib/galaxy/jobs/deferred/__init__.py @@ -1,18 +1,22 @@ """ Queue for running deferred code via plugins. """ -import os, sys, logging, threading -from Queue import Queue, Empty +import logging +import os +import threading +from Queue import Queue from galaxy import model from galaxy.util.bunch import Bunch log = logging.getLogger( __name__ ) + class DeferredJobQueue( object ): - job_states = Bunch( READY = 'ready', - WAIT = 'wait', - INVALID = 'invalid' ) + job_states = Bunch( READY='ready', + WAIT='wait', + INVALID='invalid' ) + def __init__( self, app ): self.app = app self.sa_session = app.model.context.current @@ -26,6 +30,7 @@ class DeferredJobQueue( object ): self.monitor_thread = threading.Thread( target=self.__monitor ) self.monitor_thread.start() log.info( 'Deferred job queue started' ) + def _load_plugins( self ): for fname in os.listdir( os.path.dirname( __file__ ) ): if not fname.startswith( '_' ) and fname.endswith( '.py' ): @@ -52,6 +57,7 @@ class DeferredJobQueue( object ): self.plugins[obj] = plugin( self.app ) self.plugins[obj].job_states = self.job_states log.debug( 'Loaded deferred job plugin: %s' % display_name ) + def __check_jobs_at_startup( self ): waiting_jobs = self.sa_session.query( model.DeferredJob ) \ .filter( model.DeferredJob.state == model.DeferredJob.states.WAITING ).all() @@ -64,6 +70,7 @@ class DeferredJobQueue( object ): # Pass the job ID as opposed to the job, since the monitor thread # needs to load it in its own threadlocal scoped session. self.waiting_jobs.append( job.id ) + def __monitor( self ): while self.running: try: @@ -72,6 +79,7 @@ class DeferredJobQueue( object ): log.exception( 'Exception in monitor_step' ) self.sleeper.sleep( 1 ) log.info( 'job queue stopped' ) + def __monitor_step( self ): # TODO: Querying the database with this frequency is bad, we need message passing new_jobs = self.sa_session.query( model.DeferredJob ) \ @@ -117,6 +125,7 @@ class DeferredJobQueue( object ): else: new_waiting.append( job ) self.waiting_jobs = new_waiting + def __check_job_plugin( self, job ): if job.plugin not in self.plugins: log.error( 'Invalid deferred job plugin: %s' ) % job.plugin @@ -125,16 +134,20 @@ class DeferredJobQueue( object ): self.sa_session.flush() return False return True + def __check_if_ready_to_run( self, job ): return self.plugins[job.plugin].check_job( job ) + def __fail_job( self, job ): job.state = model.DeferredJob.states.ERROR self.sa_session.add( job ) self.sa_session.flush() + def shutdown( self ): self.running = False self.sleeper.wake() + class Sleeper( object ): """ Provides a 'sleep' method that sleeps for a number of seconds *unless* @@ -142,15 +155,18 @@ class Sleeper( object ): """ def __init__( self ): self.condition = threading.Condition() + def sleep( self, seconds ): self.condition.acquire() self.condition.wait( seconds ) self.condition.release() + def wake( self ): self.condition.acquire() self.condition.notify() self.condition.release() + class FakeTrans( object ): """A fake trans for calling the external set metadata tool""" def __init__( self, app, history=None, user=None): @@ -169,15 +185,19 @@ class FakeTrans( object ): else: self.user = user self.model = app.model + def get_galaxy_session( self ): return self.dummy + def log_event( self, message, tool_id=None ): pass + def get_current_user_roles( self ): if self.user: return self.user.all_roles() else: return [] + def db_dataset_for( self, dbkey ): if self.history is None: return None diff --git a/lib/galaxy/jobs/deferred/data_transfer.py b/lib/galaxy/jobs/deferred/data_transfer.py index 4589b2f7117..1b20b0064db 100644 --- a/lib/galaxy/jobs/deferred/data_transfer.py +++ b/lib/galaxy/jobs/deferred/data_transfer.py @@ -1,35 +1,41 @@ """ Module for managing data transfer jobs. """ -import logging, urllib2, re, shutil +import logging +import re +import shutil -from galaxy import eggs from sqlalchemy import and_ -from galaxy.util.odict import odict -from galaxy.workflow.modules import module_factory +from galaxy.datatypes import sniff from galaxy.jobs.actions.post import ActionBox from galaxy.jobs.deferred import FakeTrans - from galaxy.tools.parameters import visit_input_values from galaxy.tools.parameters.basic import DataToolParameter -from galaxy.datatypes import sniff +from galaxy.util.odict import odict +from galaxy.workflow.modules import module_factory + log = logging.getLogger( __name__ ) __all__ = [ 'DataTransfer' ] + class DataTransfer( object ): check_interval = 15 dataset_name_re = re.compile( '(dataset\d+)_(name)' ) dataset_datatype_re = re.compile( '(dataset\d+)_(datatype)' ) + def __init__( self, app ): self.app = app self.sa_session = app.model.context.current + def create_job( self, trans, **kwd ): raise Exception( "Unimplemented Method" ) + def check_job( self, job ): raise Exception( "Unimplemented Method" ) + def run_job( self, job ): if job.params[ 'type' ] == 'init_transfer': # TODO: don't create new downloads on restart. @@ -115,16 +121,16 @@ class DataTransfer( object ): self.sa_session.add( ld ) self.sa_session.flush() self.app.security_agent.copy_library_permissions( FakeTrans( self.app ), sample.folder, ld ) - ldda = self.app.model.LibraryDatasetDatasetAssociation( name = library_dataset_name, - extension = extension, - dbkey = '?', - library_dataset = ld, - create_dataset = True, - sa_session = self.sa_session ) + ldda = self.app.model.LibraryDatasetDatasetAssociation( name=library_dataset_name, + extension=extension, + dbkey='?', + library_dataset=ld, + create_dataset=True, + sa_session=self.sa_session ) ldda.message = 'Transferred by the Data Transfer Plugin' self.sa_session.add( ldda ) self.sa_session.flush() - ldda.state = ldda.states.QUEUED # flushed in the set property + ldda.state = ldda.states.QUEUED # flushed in the set property ld.library_dataset_dataset_association_id = ldda.id self.sa_session.add( ld ) self.sa_session.flush() @@ -140,13 +146,13 @@ class DataTransfer( object ): FakeTrans( self.app, history=sample.history, user=sample.request.user ), - incoming = { 'input1':ldda } ) + incoming={ 'input1': ldda } ) ldda.state = ldda.states.OK # TODO: not sure if this flush is necessary self.sa_session.add( ldda ) self.sa_session.flush() except Exception, e: - log.exception( 'Failure preparing library dataset for finished transfer job (id: %s) via deferred job (id: %s):' % \ + log.exception( 'Failure preparing library dataset for finished transfer job (id: %s) via deferred job (id: %s):' % ( str( job.transfer_job.id ), str( job.id ) ) ) ldda.state = ldda.states.ERROR if sample.workflow: @@ -157,7 +163,7 @@ class DataTransfer( object ): sub_done = True rep_done = False for k, v in sample.workflow[ 'mappings' ].iteritems(): - if not 'hda' in v and v[ 'ds_tag' ].startswith( 'hi|' ): + if 'hda' not in v and v[ 'ds_tag' ].startswith( 'hi|' ): sample.workflow[ 'mappings' ][ k ][ 'hda' ] = self.app.security.decode_id( v[ 'ds_tag' ][3:] ) for key, value in sample.workflow[ 'mappings' ].iteritems(): if 'url' in value and value[ 'url' ] == job.params[ 'result' ][ 'url' ]: @@ -166,7 +172,7 @@ class DataTransfer( object ): sample.workflow[ 'mappings' ][ key ][ 'ldda' ] = ldda.id rep_done = True # DBTODO replace the hi| mappings with the hda here. Just rip off the first three chars. - elif not 'ldda' in value and not 'hda' in value: + elif 'ldda' not in value and 'hda' not in value: # We're not done if some mappings still don't have ldda or hda mappings. sub_done = False if sub_done and rep_done: @@ -203,17 +209,18 @@ class DataTransfer( object ): self._update_sample_state( sample.id, new_state ) # Update the state of the request, if possible self._update_request_state( sample.request.id ) + def _missing_params( self, params, required_params ): missing_params = filter( lambda x: x not in params, required_params ) if missing_params: log.error( 'Job parameters missing required keys: %s' % ', '.join( missing_params ) ) return True return False + def _update_sample_dataset_status( self, protocol, sample_id, result_dict, new_status, error_msg=None ): # result_dict looks something like: # {'url': '127.0.0.1/data/filtered_subreads.fa', 'name': 'Filtered reads'} - # Check if the new status is a valid transfer status - valid_statuses = [ v[1] for v in self.app.model.SampleDataset.transfer_status.items() ] + # TODO: error checking on valid new_status value if protocol in [ 'http', 'https' ]: sample_dataset = self.sa_session.query( self.app.model.SampleDataset ) \ @@ -227,6 +234,7 @@ class DataTransfer( object ): sample_dataset.error_msg = error_msg self.sa_session.add( sample_dataset ) self.sa_session.flush() + def _update_sample_state( self, sample_id, new_state, comment=None ): sample = self.sa_session.query( self.app.model.Sample ).get( sample_id ) if comment is None: @@ -234,6 +242,7 @@ class DataTransfer( object ): event = self.app.model.SampleEvent( sample, new_state, comment ) self.sa_session.add( event ) self.sa_session.flush() + def _update_request_state( self, request_id ): request = self.sa_session.query( self.app.model.Request ).get( request_id ) # Make sure all the samples of the current request have the same state @@ -247,13 +256,11 @@ class DataTransfer( object ): self.sa_session.add( event ) self.sa_session.flush() else: - final_state = False request_type_state = request.type.final_sample_state if common_state.id == request_type_state.id: # Since all the samples are in the final state, change the request state to 'Complete' comment = "All samples of this sequencing request are in the final sample state (%s). " % request_type_state.name state = request.states.COMPLETE - final_state = True else: comment = "All samples of this sequencing request are in the (%s) sample state. " % common_state.name state = request.states.SUBMITTED @@ -261,6 +268,7 @@ class DataTransfer( object ): self.sa_session.add( event ) self.sa_session.flush() # TODO: handle email notification if it is configured to be sent when the samples are in this state. + def _execute_workflow( self, sample): for key, value in sample.workflow['mappings'].iteritems(): if 'hda' not in value and 'ldda' in value: @@ -280,7 +288,7 @@ class DataTransfer( object ): if not isinstance(key, int): new_wf_dict['mappings'][int(key)] = workflow_dict['mappings'][key] workflow_dict = new_wf_dict - fk_trans = FakeTrans(self.app, history = sample.history, user=sample.request.user) + fk_trans = FakeTrans(self.app, history=sample.history, user=sample.request.user) workflow = self.sa_session.query(self.app.model.Workflow).get(workflow_dict['id']) if not workflow: log.error("Workflow mapping failure.") @@ -296,8 +304,6 @@ class DataTransfer( object ): return # Build the state for each step errors = {} - has_upgrade_messages = False - has_errors = False # Build a fake dictionary prior to execution. # Prepare each step for step in workflow.steps: @@ -308,8 +314,6 @@ class DataTransfer( object ): step.module = module_factory.from_workflow_step( fk_trans, step ) # Fix any missing parameters step.upgrade_messages = step.module.check_and_update_state() - if step.upgrade_messages: - has_upgrade_messages = True # Any connected input needs to have value DummyDataset (these # are not persisted so we need to do it every time) step.module.add_dummy_datasets( connections=step.input_connections ) @@ -317,10 +321,9 @@ class DataTransfer( object ): step.state = step.module.state # Error dict if step.tool_errors: - has_errors = True errors[step.id] = step.tool_errors else: - ## Non-tool specific stuff? + # Non-tool specific stuff? step.module = module_factory.from_workflow_step( fk_trans, step ) step.state = step.module.get_runtime_state() # Connections by input name @@ -336,8 +339,6 @@ class DataTransfer( object ): module = module_factory.from_workflow_step( fk_trans, step ) # Fix any missing parameters step.upgrade_messages = module.check_and_update_state() - if step.upgrade_messages: - has_upgrade_messages = True # Any connected input needs to have value DummyDataset (these # are not persisted so we need to do it every time) module.add_dummy_datasets( connections=step.input_connections ) @@ -345,8 +346,6 @@ class DataTransfer( object ): tool = module.tool # Get the state step.state = state = module.state - # Get old errors - old_errors = state.inputs.pop( "__errors__", {} ) if step_errors: errors[step.id] = state.inputs["__errors__"] = step_errors # Run each step, connecting outputs to inputs @@ -357,6 +356,7 @@ class DataTransfer( object ): job = None if step.type == 'tool' or step.type is None: tool = self.app.toolbox.get_tool( step.tool_id ) + def callback( input, value, prefixed_name, prefixed_label ): if isinstance( input, DataToolParameter ): if prefixed_name in step.input_connections_by_name: diff --git a/lib/galaxy/jobs/deferred/manual_data_transfer.py b/lib/galaxy/jobs/deferred/manual_data_transfer.py index 4ce987c79bf..94812e191a6 100644 --- a/lib/galaxy/jobs/deferred/manual_data_transfer.py +++ b/lib/galaxy/jobs/deferred/manual_data_transfer.py @@ -2,16 +2,18 @@ Generic module for managing manual data transfer jobs using Galaxy's built-in file browser. This module can be used by various external services that are configured to transfer data manually. """ -import logging, urllib2, re, shutil -from data_transfer import * +import logging +from data_transfer import DataTransfer log = logging.getLogger( __name__ ) __all__ = [ 'ManualDataTransferPlugin' ] + class ManualDataTransferPlugin( DataTransfer ): def __init__( self, app ): super( ManualDataTransferPlugin, self ).__init__( app ) + def create_job( self, trans, **kwd ): if 'sample' in kwd and 'sample_datasets' in kwd and 'external_service' in kwd and 'external_service_type' in kwd: sample = kwd[ 'sample' ] @@ -29,7 +31,7 @@ class ManualDataTransferPlugin( DataTransfer ): # but without a way for the user to associate stuff it's useless. However, allowing the user this ability may # open a can of worms, so maybe we shouldn't do it??? # - #for run_result_file_name, run_result_file_datatype in external_service_type.run_details[ 'results' ].items(): + # for run_result_file_name, run_result_file_datatype in external_service_type.run_details[ 'results' ].items(): # # external_service_type.run_details[ 'results' ] looks something like: {'dataset1_name': 'dataset1_datatype'} # if run_result_file_datatype in external_service.form_values.content: # datatype = external_service.form_values.content[ run_result_file_datatype ] @@ -41,13 +43,13 @@ class ManualDataTransferPlugin( DataTransfer ): sample_datasets_dict = {} for sample_dataset in sample_datasets: sample_dataset_id = sample_dataset.id - sample_dataset_dict = dict( sample_id = sample_dataset.sample.id, - name = sample_dataset.name, - file_path = sample_dataset.file_path, - status = sample_dataset.status, - error_msg = sample_dataset.error_msg, - size = sample_dataset.size, - external_service_id = sample_dataset.external_service.id ) + sample_dataset_dict = dict( sample_id=sample_dataset.sample.id, + name=sample_dataset.name, + file_path=sample_dataset.file_path, + status=sample_dataset.status, + error_msg=sample_dataset.error_msg, + size=sample_dataset.size, + external_service_id=sample_dataset.external_service.id ) sample_datasets_dict[ sample_dataset_id ] = sample_dataset_dict params = { 'type' : 'init_transfer', 'sample_id' : sample.id, @@ -72,6 +74,7 @@ class ManualDataTransferPlugin( DataTransfer ): self.sa_session.flush() log.debug( 'Created a deferred job in the ManualDataTransferPlugin of type: %s' % params[ 'type' ] ) # TODO: error reporting to caller (if possible?) + def check_job( self, job ): if self._missing_params( job.params, [ 'type' ] ): return self.job_states.INVALID diff --git a/lib/galaxy/jobs/deferred/pacific_biosciences_smrt_portal.py b/lib/galaxy/jobs/deferred/pacific_biosciences_smrt_portal.py index ead0fb2575d..e2ed1b33694 100644 --- a/lib/galaxy/jobs/deferred/pacific_biosciences_smrt_portal.py +++ b/lib/galaxy/jobs/deferred/pacific_biosciences_smrt_portal.py @@ -2,20 +2,24 @@ Module for managing jobs in Pacific Bioscience's SMRT Portal and automatically transferring files produced by SMRT Portal. """ -import logging, urllib2, re, shutil +import logging +import urllib2 from string import Template from galaxy.util import json -from data_transfer import * +from data_transfer import DataTransfer log = logging.getLogger( __name__ ) __all__ = [ 'SMRTPortalPlugin' ] + class SMRTPortalPlugin( DataTransfer ): api_path = '/smrtportal/api' + def __init__( self, app ): super( SMRTPortalPlugin, self ).__init__( app ) + def create_job( self, trans, **kwd ): if 'secondary_analysis_job_id' in kwd: sample = kwd[ 'sample' ] @@ -42,7 +46,7 @@ class SMRTPortalPlugin( DataTransfer ): results[ id ] = { field : v } for id, attrs in results.items(): url_template = external_service_type.run_details[ 'results_urls' ].get( id + '_name' ) - url = Template( url_template ).substitute( host = smrt_host, secondary_analysis_job_id = kwd[ 'secondary_analysis_job_id' ] ) + url = Template( url_template ).substitute( host=smrt_host, secondary_analysis_job_id=kwd[ 'secondary_analysis_job_id' ] ) results[ id ][ 'url' ] = url if sample.workflow: # DBTODO Make sure all ds| mappings get the URL of the dataset, for linking to later. @@ -75,6 +79,7 @@ class SMRTPortalPlugin( DataTransfer ): self.sa_session.flush() log.debug( 'Created a deferred job in the SMRTPortalPlugin of type: %s' % params[ 'type' ] ) # TODO: error reporting to caller (if possible?) + def check_job( self, job ): if self._missing_params( job.params, [ 'type' ] ): return self.job_states.INVALID @@ -106,6 +111,7 @@ class SMRTPortalPlugin( DataTransfer ): else: log.error( 'Unknown job type for SMRTPortalPlugin: %s' % str( job.params[ 'type' ] ) ) return self.job_states.INVALID + def _associate_untransferred_datasets_with_sample( self, sample, external_service, results_dict ): # results_dict looks something like: # {'dataset2': {'datatype': 'fasta', 'url': '127.0.0.1:8080/data/filtered_subreads.fa', 'name': 'Filtered reads'} } diff --git a/lib/galaxy/jobs/handler.py b/lib/galaxy/jobs/handler.py index d366d5f1f7c..0d0afe2777d 100644 --- a/lib/galaxy/jobs/handler.py +++ b/lib/galaxy/jobs/handler.py @@ -109,14 +109,14 @@ class JobHandlerQueue( object ): model.Job.states.RUNNING ) if self.app.config.user_activation_on: jobs_at_startup = self.sa_session.query( model.Job ).enable_eagerloads( False ) \ - .outerjoin( model.User ) \ - .filter( model.Job.state.in_( in_list ) \ - & ( model.Job.handler == self.app.config.server_name ) \ - & or_( ( model.Job.user_id == None ), ( model.User.active == True ) ) ).all() + .outerjoin( model.User ) \ + .filter( model.Job.state.in_( in_list ) + & ( model.Job.handler == self.app.config.server_name ) + & or_( ( model.Job.user_id is None ), ( model.User.active is True ) ) ).all() else: jobs_at_startup = self.sa_session.query( model.Job ).enable_eagerloads( False ) \ - .filter( model.Job.state.in_( in_list ) \ - & ( model.Job.handler == self.app.config.server_name ) ).all() + .filter( model.Job.state.in_( in_list ) + & ( model.Job.handler == self.app.config.server_name ) ).all() for job in jobs_at_startup: if not self.app.toolbox.has_tool( job.tool_id, job.tool_version, exact=True ): @@ -199,32 +199,32 @@ class JobHandlerQueue( object ): self.sa_session.expunge_all() # Fetch all new jobs hda_not_ready = self.sa_session.query(model.Job.id).enable_eagerloads(False) \ - .join(model.JobToInputDatasetAssociation) \ - .join(model.HistoryDatasetAssociation) \ - .join(model.Dataset) \ - .filter(and_( (model.Job.state == model.Job.states.NEW ), - or_( ( model.HistoryDatasetAssociation._state == model.HistoryDatasetAssociation.states.FAILED_METADATA ), - ( model.HistoryDatasetAssociation.deleted == True ), - ( model.Dataset.state != model.Dataset.states.OK ), - ( model.Dataset.deleted == True) ) ) ).subquery() + .join(model.JobToInputDatasetAssociation) \ + .join(model.HistoryDatasetAssociation) \ + .join(model.Dataset) \ + .filter(and_( (model.Job.state == model.Job.states.NEW ), + or_( ( model.HistoryDatasetAssociation._state == model.HistoryDatasetAssociation.states.FAILED_METADATA ), + ( model.HistoryDatasetAssociation.deleted is True ), + ( model.Dataset.state != model.Dataset.states.OK ), + ( model.Dataset.deleted is True) ) ) ).subquery() ldda_not_ready = self.sa_session.query(model.Job.id).enable_eagerloads(False) \ - .join(model.JobToInputLibraryDatasetAssociation) \ - .join(model.LibraryDatasetDatasetAssociation) \ - .join(model.Dataset) \ - .filter(and_((model.Job.state == model.Job.states.NEW), - or_((model.LibraryDatasetDatasetAssociation._state != None), - (model.LibraryDatasetDatasetAssociation.deleted == True), - (model.Dataset.state != model.Dataset.states.OK), - (model.Dataset.deleted == True)))).subquery() + .join(model.JobToInputLibraryDatasetAssociation) \ + .join(model.LibraryDatasetDatasetAssociation) \ + .join(model.Dataset) \ + .filter(and_((model.Job.state == model.Job.states.NEW), + or_((model.LibraryDatasetDatasetAssociation._state is not None), + (model.LibraryDatasetDatasetAssociation.deleted is True), + (model.Dataset.state != model.Dataset.states.OK), + (model.Dataset.deleted is True)))).subquery() if self.app.config.user_activation_on: jobs_to_check = self.sa_session.query(model.Job).enable_eagerloads(False) \ - .outerjoin( model.User ) \ - .filter(and_((model.Job.state == model.Job.states.NEW), - or_((model.Job.user_id == None), (model.User.active == True)), - (model.Job.handler == self.app.config.server_name), - ~model.Job.table.c.id.in_(hda_not_ready), - ~model.Job.table.c.id.in_(ldda_not_ready))) \ - .order_by(model.Job.id).all() + .outerjoin( model.User ) \ + .filter(and_((model.Job.state == model.Job.states.NEW), + or_((model.Job.user_id is None), (model.User.active is True)), + (model.Job.handler == self.app.config.server_name), + ~model.Job.table.c.id.in_(hda_not_ready), + ~model.Job.table.c.id.in_(ldda_not_ready))) \ + .order_by(model.Job.id).all() else: jobs_to_check = self.sa_session.query(model.Job).enable_eagerloads(False) \ .filter(and_((model.Job.state == model.Job.states.NEW), @@ -234,9 +234,9 @@ class JobHandlerQueue( object ): .order_by(model.Job.id).all() # Fetch all "resubmit" jobs resubmit_jobs = self.sa_session.query(model.Job).enable_eagerloads(False) \ - .filter(and_((model.Job.state == model.Job.states.RESUBMITTED), - (model.Job.handler == self.app.config.server_name))) \ - .order_by(model.Job.id).all() + .filter(and_((model.Job.state == model.Job.states.RESUBMITTED), + (model.Job.handler == self.app.config.server_name))) \ + .order_by(model.Job.id).all() else: # Get job objects and append to watch queue for any which were # previously waiting @@ -428,7 +428,7 @@ class JobHandlerQueue( object ): # This could have been incremented by a previous job dispatched on this iteration, even if we're not caching rval = self.user_job_count.get(user_id, 0) if not self.app.config.cache_user_job_count: - result = self.sa_session.execute(select([func.count(model.Job.table.c.id)]) \ + result = self.sa_session.execute(select([func.count(model.Job.table.c.id)]) .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING, model.Job.states.RESUBMITTED)), @@ -442,12 +442,12 @@ class JobHandlerQueue( object ): # Cache the job count if necessary if self.user_job_count is None and self.app.config.cache_user_job_count: self.user_job_count = {} - query = self.sa_session.execute(select([model.Job.table.c.user_id, func.count(model.Job.table.c.user_id)]) \ + query = self.sa_session.execute(select([model.Job.table.c.user_id, func.count(model.Job.table.c.user_id)]) .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING, model.Job.states.RESUBMITTED)), - (model.Job.table.c.user_id is not None))) \ - .group_by(model.Job.table.c.user_id)) + (model.Job.table.c.user_id is not None))) + .group_by(model.Job.table.c.user_id)) for row in query: self.user_job_count[row[0]] = row[1] elif self.user_job_count is None: @@ -465,8 +465,8 @@ class JobHandlerQueue( object ): # queue. rval = {} rval.update(cached) - result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')]) \ - .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)), (model.Job.table.c.user_id == user_id))) \ + result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')]) + .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)), (model.Job.table.c.user_id == user_id))) .group_by(model.Job.table.c.destination_id)) for row in result: # Add the count from the database to the cached count @@ -477,8 +477,8 @@ class JobHandlerQueue( object ): # Cache the job count if necessary if self.user_job_count_per_destination is None and self.app.config.cache_user_job_count: self.user_job_count_per_destination = {} - result = self.sa_session.execute(select([model.Job.table.c.user_id, model.Job.table.c.destination_id, func.count(model.Job.table.c.user_id).label('job_count')]) \ - .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)))) \ + result = self.sa_session.execute(select([model.Job.table.c.user_id, model.Job.table.c.destination_id, func.count(model.Job.table.c.user_id).label('job_count')]) + .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)))) .group_by(model.Job.table.c.user_id, model.Job.table.c.destination_id)) for row in result: if row['user_id'] not in self.user_job_count_per_destination: @@ -552,8 +552,8 @@ class JobHandlerQueue( object ): # Cache the job count if necessary if self.total_job_count_per_destination is None: self.total_job_count_per_destination = {} - result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')]) \ - .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)))) \ + result = self.sa_session.execute(select([model.Job.table.c.destination_id, func.count(model.Job.table.c.destination_id).label('job_count')]) + .where(and_(model.Job.table.c.state.in_((model.Job.states.QUEUED, model.Job.states.RUNNING)))) .group_by(model.Job.table.c.destination_id)) for row in result: self.total_job_count_per_destination[row['destination_id']] = row['job_count'] @@ -663,7 +663,7 @@ class JobHandlerStopQueue( object ): self.sa_session.expunge_all() # Fetch all new jobs newly_deleted_jobs = self.sa_session.query( model.Job ).enable_eagerloads( False ) \ - .filter( ( model.Job.state == model.Job.states.DELETED_NEW ) \ + .filter( ( model.Job.state == model.Job.states.DELETED_NEW ) & ( model.Job.handler == self.app.config.server_name ) ).all() for job in newly_deleted_jobs: jobs_to_check.append( ( job, job.stderr ) ) @@ -750,7 +750,7 @@ class DefaultJobDispatcher( object ): runner_name = self.__get_runner_name( job_wrapper ) try: if isinstance(job_wrapper, TaskWrapper): - #DBTODO Refactor + # DBTODO Refactor log.debug( "(%s) Dispatching task %s to %s runner" % ( job_wrapper.job_id, job_wrapper.task_id, runner_name ) ) else: log.debug( "(%s) Dispatching to %s runner" % ( job_wrapper.job_id, runner_name ) ) diff --git a/lib/galaxy/jobs/mapper.py b/lib/galaxy/jobs/mapper.py index 6534d4b980c..9f6faaa9dde 100644 --- a/lib/galaxy/jobs/mapper.py +++ b/lib/galaxy/jobs/mapper.py @@ -57,8 +57,8 @@ class JobRunnerMapper( object ): def __get_rule_modules( self ): unsorted_module_names = self.__get_rule_module_names( ) - ## Load modules in reverse order to allow hierarchical overrides - ## i.e. 000_galaxy_rules.py, 100_site_rules.py, 200_instance_rules.py + # Load modules in reverse order to allow hierarchical overrides + # i.e. 000_galaxy_rules.py, 100_site_rules.py, 200_instance_rules.py module_names = sorted( unsorted_module_names, reverse=True ) modules = [] for rule_module_name in module_names: diff --git a/lib/galaxy/jobs/metrics/formatting.py b/lib/galaxy/jobs/metrics/formatting.py index 4057b2226e2..1147edd5e73 100644 --- a/lib/galaxy/jobs/metrics/formatting.py +++ b/lib/galaxy/jobs/metrics/formatting.py @@ -7,7 +7,7 @@ class JobMetricFormatter( object ): return ( str( key ), str( value ) ) -## Formatting utilities +# Formatting utilities def seconds_to_str( value ): if value < 60: diff --git a/lib/galaxy/jobs/metrics/instrumenters/collectl.py b/lib/galaxy/jobs/metrics/instrumenters/collectl.py index 05711806f22..7d69cb116cb 100644 --- a/lib/galaxy/jobs/metrics/instrumenters/collectl.py +++ b/lib/galaxy/jobs/metrics/instrumenters/collectl.py @@ -14,8 +14,7 @@ import logging log = logging.getLogger( __name__ ) DEFAULT_PROCFILT_ON = "username" # By default, only grab statistics for user - # processes (as identifiers by username). -DEFAULT_SUBSYSTEMS = "process" +DEFAULT_SUBSYSTEMS = "process" # processes (as identifiers by username). DEFAULT_FLUSH_INTERVAL = "0" # Set to zero to flush every collection. FORMATTED_RESOURCE_TITLES = { diff --git a/lib/galaxy/jobs/output_checker.py b/lib/galaxy/jobs/output_checker.py index 344f4e22cb4..90a0e1ea2a8 100644 --- a/lib/galaxy/jobs/output_checker.py +++ b/lib/galaxy/jobs/output_checker.py @@ -39,7 +39,7 @@ def check_output( tool, stdout, stderr, tool_exit_code, job ): # that range, then apply the error level and add a message. # If we've reached a fatal error rule, then stop. max_error_level = StdioErrorLevel.NO_ERROR - if tool_exit_code != None: + if tool_exit_code is not None: for stdio_exit_code in tool.stdio_exit_codes: if ( tool_exit_code >= stdio_exit_code.range_start and tool_exit_code <= stdio_exit_code.range_end ): diff --git a/lib/galaxy/jobs/rule_helper.py b/lib/galaxy/jobs/rule_helper.py index 83e9b34f026..872e2f66c1d 100644 --- a/lib/galaxy/jobs/rule_helper.py +++ b/lib/galaxy/jobs/rule_helper.py @@ -2,10 +2,7 @@ from datetime import datetime import hashlib import random -from sqlalchemy import ( - and_, - func -) +from sqlalchemy import func from galaxy import model from galaxy import util diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py index 486a6cf223c..4567ad6401e 100644 --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -251,15 +251,15 @@ class BaseJobRunner( object ): Set metadata externally. Used by the local and lwr job runners where this shouldn't be attached to command-line to execute. """ - #run the metadata setting script here - #this is terminate-able when output dataset/job is deleted - #so that long running set_meta()s can be canceled without having to reboot the server + # run the metadata setting script here + # this is terminate-able when output dataset/job is deleted + # so that long running set_meta()s can be canceled without having to reboot the server if job_wrapper.get_state() not in [ model.Job.states.ERROR, model.Job.states.DELETED ] and job_wrapper.output_paths: lib_adjust = GALAXY_LIB_ADJUST_TEMPLATE % job_wrapper.galaxy_lib_dir external_metadata_script = job_wrapper.setup_external_metadata( output_fnames=job_wrapper.get_output_fnames(), set_extension=True, tmp_dir=job_wrapper.working_directory, - #we don't want to overwrite metadata that was copied over in init_meta(), as per established behavior + # We don't want to overwrite metadata that was copied over in init_meta(), as per established behavior kwds={ 'overwrite' : False } ) external_metadata_script = "%s %s" % (lib_adjust, external_metadata_script) if resolve_requirements: @@ -304,8 +304,8 @@ class BaseJobRunner( object ): working_directory=os.path.abspath( job_wrapper.working_directory ), command=command_line, ) - ## Additional logging to enable if debugging from_work_dir handling, metadata - ## commands, etc... (or just peak in the job script.) + # 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( '(%s) command is: %s' % ( job_id, command_line ) ) options.update(**kwds) @@ -361,11 +361,12 @@ class JobState( object ): Encapsulate state of jobs. """ runner_states = Bunch( - WALLTIME_REACHED = 'walltime_reached', - MEMORY_LIMIT_REACHED = 'memory_limit_reached', - GLOBAL_WALLTIME_REACHED = 'global_walltime_reached', - OUTPUT_SIZE_LIMIT = 'output_size_limit' + WALLTIME_REACHED='walltime_reached', + MEMORY_LIMIT_REACHED='memory_limit_reached', + GLOBAL_WALLTIME_REACHED='global_walltime_reached', + OUTPUT_SIZE_LIMIT='output_size_limit' ) + def __init__( self ): self.runner_state_handled = False diff --git a/lib/galaxy/jobs/runners/cli.py b/lib/galaxy/jobs/runners/cli.py index 09b144986db..1bbc6abd01e 100644 --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -53,9 +53,6 @@ class ShellJobRunner( AsynchronousJobRunner ): 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 shell and job execution interface job_destination = job_wrapper.job_destination shell_params, job_params = self.parse_destination_params(job_destination.params) @@ -98,7 +95,7 @@ class ShellJobRunner( AsynchronousJobRunner ): log.error('(%s) submission failed (stderr): %s' % (galaxy_id_tag, cmd_out.stderr)) job_wrapper.fail("failure submitting job") return - # Some job runners return something like 'Submitted batch job XXXX' + # Some job runners return something like 'Submitted batch job XXXX' # Strip and split to get job ID. external_job_id = cmd_out.stdout.strip().split()[-1] if not external_job_id: diff --git a/lib/galaxy/jobs/runners/condor.py b/lib/galaxy/jobs/runners/condor.py index b2d031aec1b..5e072c809bd 100644 --- a/lib/galaxy/jobs/runners/condor.py +++ b/lib/galaxy/jobs/runners/condor.py @@ -49,9 +49,6 @@ class CondorJobRunner( AsynchronousJobRunner ): if not self.prepare_job( job_wrapper, include_metadata=include_metadata): 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 @@ -177,7 +174,7 @@ class CondorJobRunner( AsynchronousJobRunner ): if not job_running and cjs.running: log.debug( "(%s/%s) job has stopped running" % ( galaxy_id_tag, job_id ) ) # Will switching from RUNNING to QUEUED confuse Galaxy? - #cjs.job_wrapper.change_state( model.Job.states.QUEUED ) + # cjs.job_wrapper.change_state( model.Job.states.QUEUED ) if job_complete: if cjs.job_wrapper.get_state() != model.Job.states.DELETED: external_metadata = not asbool( cjs.job_wrapper.job_destination.params.get( "embed_metadata_in_job", True) ) diff --git a/lib/galaxy/jobs/runners/drmaa.py b/lib/galaxy/jobs/runners/drmaa.py index 488ef93d8ec..ce5743faa79 100644 --- a/lib/galaxy/jobs/runners/drmaa.py +++ b/lib/galaxy/jobs/runners/drmaa.py @@ -7,7 +7,6 @@ import logging import os import string import subprocess -import sys import time from galaxy import eggs @@ -41,11 +40,11 @@ class DRMAAJobRunner( AsynchronousJobRunner ): global drmaa runner_param_specs = dict( - drmaa_library_path = dict( map = str, default = os.environ.get( 'DRMAA_LIBRARY_PATH', None ) ), - invalidjobexception_state = dict( map = str, valid = lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default = model.Job.states.OK ), - invalidjobexception_retries = dict( map = int, valid = lambda x: int >= 0, default = 0 ), - internalexception_state = dict( map = str, valid = lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default = model.Job.states.OK ), - internalexception_retries = dict( map = int, valid = lambda x: int >= 0, default = 0 ) ) + drmaa_library_path=dict( map=str, default=os.environ.get( 'DRMAA_LIBRARY_PATH', None ) ), + invalidjobexception_state=dict( map=str, valid=lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default=model.Job.states.OK ), + invalidjobexception_retries=dict( map=int, valid=lambda x: int >= 0, default=0 ), + internalexception_state=dict( map=str, valid=lambda x: x in ( model.Job.states.OK, model.Job.states.ERROR ), default=model.Job.states.OK ), + internalexception_retries=dict( map=int, valid=lambda x: int >= 0, default=0 ) ) if 'runner_param_specs' not in kwargs: kwargs[ 'runner_param_specs' ] = dict() @@ -208,7 +207,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): pwent = job_wrapper.galaxy_system_pwent log.debug( '(%s) submitting with credentials: %s [uid: %s]' % ( galaxy_id_tag, pwent[0], pwent[2] ) ) filename = self.store_jobtemplate(job_wrapper, jt) - self.userid = pwent[2] + self.userid = pwent[2] external_job_id = self.external_runjob(filename, pwent[2]).strip() log.info( "(%s) queued as %s" % ( galaxy_id_tag, external_job_id ) ) @@ -296,7 +295,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): ajs.running = True ajs.job_wrapper.change_state( model.Job.states.RUNNING ) if state in ( drmaa.JobState.FAILED, drmaa.JobState.DONE ): - self._complete_terminal_job( ajs, drmaa_state = state ) + self._complete_terminal_job( ajs, drmaa_state=state ) continue if ajs.check_limits(): self.work_queue.put( ( self.fail_job, ajs ) ) @@ -357,7 +356,6 @@ class DRMAAJobRunner( AsynchronousJobRunner ): ajs.exit_code_file = "%s.drmec" % os.path.join(os.getcwd(), ajs.job_wrapper.working_directory, ajs.job_wrapper.get_id_tag()) ajs.job_file = job_file - def store_jobtemplate(self, job_wrapper, jt): """ Stores the content of a DRMAA JobTemplate object in a file as a JSON string. Path is hard-coded, but it's no worse than other path in this module. @@ -370,7 +368,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): except: pass s = json.dumps(data) - f = open(filename,'w+') + f = open(filename, 'w+') f.write(s) f.close() log.debug( '(%s) Job script for external submission is: %s' % ( job_wrapper.job_id, filename ) ) @@ -393,7 +391,7 @@ class DRMAAJobRunner( AsynchronousJobRunner ): shell=False, stdout=subprocess.PIPE, stderr=subprocess.PIPE) (stdoutdata, stderrdata) = p.communicate() exitcode = p.returncode - #os.unlink(jobtemplate_filename) + # os.unlink(jobtemplate_filename) if exitcode != 0: # There was an error in the child process raise RuntimeError("External_runjob failed (exit code %s)\nChild process reported error:\n%s" % (str(exitcode), stderrdata)) @@ -404,5 +402,3 @@ class DRMAAJobRunner( AsynchronousJobRunner ): # the DRMAA job-ID. If not the case, will throw an error. jobId = stdoutdata return jobId - - diff --git a/lib/galaxy/jobs/runners/local.py b/lib/galaxy/jobs/runners/local.py index 475ea703bab..ed4b4f628c6 100644 --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -35,10 +35,10 @@ class LocalJobRunner( BaseJobRunner ): def __init__( self, app, nworkers ): """Start the job runner """ - #create a local copy of os.environ to use as env for subprocess.Popen + # create a local copy of os.environ to use as env for subprocess.Popen self._environ = os.environ.copy() - #Set TEMP if a valid temp value is not already set + # 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()) @@ -50,8 +50,8 @@ class LocalJobRunner( BaseJobRunner ): """ command_line = job_wrapper.runner_command_line - ## 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 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 ) if slots: slots_statement = 'GALAXY_SLOTS="%d"; export GALAXY_SLOTS; GALAXY_SLOTS_CONFIGURED="1"; export GALAXY_SLOTS_CONFIGURED;' % ( int( slots ) ) @@ -132,7 +132,7 @@ class LocalJobRunner( BaseJobRunner ): job_wrapper.fail("Unable to finish job", exception=True) def stop_job( self, job ): - #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + # if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished job_ext_output_metadata = job.get_external_output_metadata() try: pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them diff --git a/lib/galaxy/jobs/runners/lwr.py b/lib/galaxy/jobs/runners/lwr.py index 9a84c13ee3f..7ad7273d087 100644 --- a/lib/galaxy/jobs/runners/lwr.py +++ b/lib/galaxy/jobs/runners/lwr.py @@ -320,7 +320,7 @@ class LwrJobRunner( AsynchronousJobRunner ): def get_client( self, job_destination_params, job_id, env=[] ): # Cannot use url_for outside of web thread. - #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) + # files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) encoded_job_id = self.app.security.encode_id(job_id) job_key = self.app.security.encode_id( job_id, kind="jobs_files" ) @@ -400,7 +400,7 @@ class LwrJobRunner( AsynchronousJobRunner ): return False def stop_job( self, job ): - #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + # if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished job_ext_output_metadata = job.get_external_output_metadata() if job_ext_output_metadata: pid = job_ext_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them diff --git a/lib/galaxy/jobs/runners/pbs.py b/lib/galaxy/jobs/runners/pbs.py index 651cdc11dc3..58743c333da 100644 --- a/lib/galaxy/jobs/runners/pbs.py +++ b/lib/galaxy/jobs/runners/pbs.py @@ -189,10 +189,10 @@ class PBSJobRunner( AsynchronousJobRunner ): if arg == 'l': resource_attrs = value.split(',') for j, ( res, val ) in enumerate( [ a.split('=', 1) for a in resource_attrs ] ): - rval.append( dict( name = pbs.ATTR_l, value = val, resource = res ) ) + rval.append( dict( name=pbs.ATTR_l, value=val, resource=res ) ) else: try: - rval.append( dict( name = getattr( pbs, 'ATTR_' + arg ), value = value ) ) + rval.append( dict( name=getattr( pbs, 'ATTR_' + arg ), value=value ) ) except AttributeError, e: raise Exception("Invalid parameter '%s': %s" % (arg, e)) return rval @@ -208,9 +208,6 @@ class PBSJobRunner( AsynchronousJobRunner ): if not self.prepare_job( job_wrapper, include_metadata=not( self.app.config.pbs_stage_path ) ): return - # command line has been added to the wrapper by prepare_job() - command_line = job_wrapper.runner_command_line - job_destination = job_wrapper.job_destination # Determine the job's PBS destination (server/queue) and options from the job destination definition @@ -260,20 +257,20 @@ class PBSJobRunner( AsynchronousJobRunner ): stagein = self.get_stage_in_out( job_wrapper.get_input_fnames() + output_files, symlink=True ) stageout = self.get_stage_in_out( output_files ) attrs = [ - dict( name = pbs.ATTR_o, value = pbs_ofile ), - dict( name = pbs.ATTR_e, value = pbs_efile ), - dict( name = pbs.ATTR_stagein, value = stagein ), - dict( name = pbs.ATTR_stageout, value = stageout ), + dict( name=pbs.ATTR_o, value=pbs_ofile ), + dict( name=pbs.ATTR_e, value=pbs_efile ), + dict( name=pbs.ATTR_stagein, value=stagein ), + dict( name=pbs.ATTR_stageout, value=stageout ), ] # If not, we're using NFS else: attrs = [ - dict( name = pbs.ATTR_o, value = ofile ), - dict( name = pbs.ATTR_e, value = efile ), + dict( name=pbs.ATTR_o, value=ofile ), + dict( name=pbs.ATTR_e, value=efile ), ] # define PBS job options - attrs.append( dict( name = pbs.ATTR_N, value = str( "%s_%s_%s" % ( job_wrapper.job_id, job_wrapper.tool.id, job_wrapper.user ) ) ) ) + attrs.append( dict( name=pbs.ATTR_N, value=str( "%s_%s_%s" % ( job_wrapper.job_id, job_wrapper.tool.id, job_wrapper.user ) ) ) ) job_attrs = pbs.new_attropl( len( attrs ) + len( pbs_options ) ) for i, attr in enumerate( attrs + pbs_options ): job_attrs[i].name = attr['name'] @@ -366,7 +363,6 @@ class PBSJobRunner( AsynchronousJobRunner ): ( failures, statuses ) = self.check_all_jobs() for pbs_job_state in self.watched: job_id = pbs_job_state.job_id - #galaxy_job_id = pbs_job_state.job_wrapper.job_id galaxy_job_id = pbs_job_state.job_wrapper.get_id_tag() old_state = pbs_job_state.old_state pbs_server_name = self.__get_pbs_server(pbs_job_state.job_destination.params) diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py index 8067b3528cb..0c2a81cbc2c 100644 --- a/lib/galaxy/jobs/runners/pulsar.py +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -301,7 +301,7 @@ class PulsarJobRunner( AsynchronousJobRunner ): if value is PARAMETER_SPECIFICATION_IGNORED: log.warn( "Pulsar runner in selected configuration ignores parameter %s" % key ) continue - #if self.runner_params.get( key, None ): + # if self.runner_params.get( key, None ): # # Let plugin define defaults for some parameters - # # for instance that way jobs_directory can be # # configured next to AMQP url (where it belongs). @@ -345,7 +345,7 @@ class PulsarJobRunner( AsynchronousJobRunner ): def get_client( self, job_destination_params, job_id, env=[] ): # Cannot use url_for outside of web thread. - #files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) + # files_endpoint = url_for( controller="job_files", job_id=encoded_job_id ) encoded_job_id = self.app.security.encode_id(job_id) job_key = self.app.security.encode_id( job_id, kind="jobs_files" ) @@ -429,7 +429,7 @@ class PulsarJobRunner( AsynchronousJobRunner ): return False def stop_job( self, job ): - #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + # if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished client = self.get_client( job.destination_params, job.job_runner_external_id ) job_ext_output_metadata = job.get_external_output_metadata() if not PulsarJobRunner.__remote_metadata( client ) and job_ext_output_metadata: diff --git a/lib/galaxy/jobs/runners/slurm.py b/lib/galaxy/jobs/runners/slurm.py index 8461cc0ac7b..ee9a4d51e7e 100644 --- a/lib/galaxy/jobs/runners/slurm.py +++ b/lib/galaxy/jobs/runners/slurm.py @@ -57,7 +57,7 @@ class SlurmJobRunner( DRMAAJobRunner ): ajs.runner_state = ajs.runner_states.WALLTIME_REACHED elif job_info['JobState'] == 'NODE_FAIL': log.warning( '(%s/%s) Job failed due to node failure, attempting resubmission', ajs.job_wrapper.get_id_tag(), ajs.job_id ) - ajs.job_wrapper.change_state( model.Job.states.QUEUED, info = 'Job was resubmitted due to node failure' ) + ajs.job_wrapper.change_state( model.Job.states.QUEUED, info='Job was resubmitted due to node failure' ) try: self.queue_job( ajs.job_wrapper ) return @@ -82,9 +82,9 @@ class SlurmJobRunner( DRMAAJobRunner ): return except Exception, e: log.exception( '(%s/%s) Unable to inspect failed slurm job using scontrol, job will be unconditionally failed: %s', ajs.job_wrapper.get_id_tag(), ajs.job_id, e ) - super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state = drmaa_state ) + super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state=drmaa_state ) # by default, finish as if the job was successful. - super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state = drmaa_state ) + super( SlurmJobRunner, self )._complete_terminal_job( ajs, drmaa_state=drmaa_state ) def __check_memory_limit( self, efile_path ): """ diff --git a/lib/galaxy/jobs/runners/state_handler_factory.py b/lib/galaxy/jobs/runners/state_handler_factory.py index 724de7d8c7e..6f1d24ddb72 100644 --- a/lib/galaxy/jobs/runners/state_handler_factory.py +++ b/lib/galaxy/jobs/runners/state_handler_factory.py @@ -1,4 +1,3 @@ -import os import logging from galaxy.util.submodules import submodules @@ -8,9 +7,11 @@ import galaxy.jobs.runners.state_handlers log = logging.getLogger(__name__) + def build_state_handlers(): return _get_state_handlers_dict() + def _get_state_handlers_dict(): state_handlers = {} for module in submodules(galaxy.jobs.runners.state_handlers): @@ -20,4 +21,3 @@ def _get_state_handlers_dict(): state_handlers[func].append(getattr(module, func)) log.debug("Loaded '%s' state handler from module %s", func, module.__name__) return state_handlers - diff --git a/lib/galaxy/jobs/runners/state_handlers/resubmit.py b/lib/galaxy/jobs/runners/state_handlers/resubmit.py index c62d24a9287..25320c51af5 100644 --- a/lib/galaxy/jobs/runners/state_handlers/resubmit.py +++ b/lib/galaxy/jobs/runners/state_handlers/resubmit.py @@ -1,15 +1,12 @@ import logging -from galaxy import model - - __all__ = ['failure'] log = logging.getLogger(__name__) MESSAGES = dict( - walltime_reached = 'it reached the walltime', - memory_limit_reached = 'it exceeded the amount of allocated memory' + walltime_reached='it reached the walltime', + memory_limit_reached='it exceeded the amount of allocated memory' ) diff --git a/lib/galaxy/jobs/runners/tasks.py b/lib/galaxy/jobs/runners/tasks.py index cab93909483..8a338e4c0dc 100644 --- a/lib/galaxy/jobs/runners/tasks.py +++ b/lib/galaxy/jobs/runners/tasks.py @@ -1,5 +1,4 @@ import logging -import subprocess from galaxy import model @@ -54,7 +53,7 @@ class TaskedJobRunner( BaseJobRunner ): # Split with the defined method. parallelism = job_wrapper.get_parallelism() try: - splitter = getattr(__import__('galaxy.jobs.splitters', globals(), locals(), [parallelism.method]), parallelism.method) + splitter = getattr(__import__('galaxy.jobs.splitters', globals(), locals(), [parallelism.method]), parallelism.method) except: job_wrapper.change_state( model.Job.states.ERROR ) job_wrapper.fail("Job Splitting Failed, no match for '%s'" % parallelism) @@ -78,8 +77,8 @@ class TaskedJobRunner( BaseJobRunner ): sleep_time = 1 # sleep/loop until no more progress can be made. That is when # all tasks are one of { OK, ERROR, DELETED }. If a task - completed_states = [ model.Task.states.OK, \ - model.Task.states.ERROR, \ + completed_states = [ model.Task.states.OK, + model.Task.states.ERROR, model.Task.states.DELETED ] # TODO: Should we report an error (and not merge outputs) if @@ -104,7 +103,7 @@ class TaskedJobRunner( BaseJobRunner ): self._cancel_job( job_wrapper, task_wrappers ) tasks_complete = True break - elif not task_state in completed_states: + elif task_state not in completed_states: tasks_complete = False else: job_exit_code = tw.get_exit_code() @@ -115,15 +114,15 @@ class TaskedJobRunner( BaseJobRunner ): sleep_time *= 2 job_wrapper.reclaim_ownership() # if running as the actual user, change ownership before merging. log.debug('execution finished - beginning merge: %s' % command_line) - stdout, stderr = splitter.do_merge(job_wrapper, task_wrappers) + stdout, stderr = splitter.do_merge(job_wrapper, task_wrappers) except Exception: job_wrapper.fail( "failure running job", exception=True ) log.exception("failure running job %d" % job_wrapper.job_id) return - #run the metadata setting script here - #this is terminate-able when output dataset/job is deleted - #so that long running set_meta()s can be canceled without having to reboot the server + # run the metadata setting script here + # this is terminate-able when output dataset/job is deleted + # so that long running set_meta()s can be canceled without having to reboot the server self._handle_metadata_externally(job_wrapper, resolve_requirements=True ) # Finish the job try: @@ -147,7 +146,7 @@ class TaskedJobRunner( BaseJobRunner ): # this if the tasks runner is used but the tool does not use # parallelism. else: - #if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished + # if our local job has JobExternalOutputMetadata associated, then our primary job has to have already finished if job.external_output_metadata: pid = job.external_output_metadata[0].job_runner_external_pid # every JobExternalOutputMetadata has a pid set, we just need to take from one of them else: diff --git a/lib/galaxy/jobs/splitters/basic.py b/lib/galaxy/jobs/splitters/basic.py index 50d7151691e..6ac56b6e850 100644 --- a/lib/galaxy/jobs/splitters/basic.py +++ b/lib/galaxy/jobs/splitters/basic.py @@ -3,22 +3,24 @@ import multi log = logging.getLogger( __name__ ) + def set_basic_defaults(job_wrapper): parent_job = job_wrapper.get_job() parallelism = job_wrapper.get_parallelism() parallelism.attributes['split_inputs'] = parent_job.input_datasets[0].name parallelism.attributes['merge_outputs'] = job_wrapper.get_output_hdas_and_fnames().keys()[0] -def do_split (job_wrapper): + +def do_split(job_wrapper): if len(job_wrapper.get_input_fnames()) > 1 or len(job_wrapper.get_output_fnames()) > 1: log.error("The basic splitter is not capable of handling jobs with multiple inputs or outputs.") - raise Exception, "Job Splitting Failed, the basic splitter only handles tools with one input and one output" + raise Exception( "Job Splitting Failed, the basic splitter only handles tools with one input and one output" ) # add in the missing information for splitting the one input and merging the one output set_basic_defaults(job_wrapper) return multi.do_split(job_wrapper) -def do_merge( job_wrapper, task_wrappers): + +def do_merge( job_wrapper, task_wrappers): # add in the missing information for splitting the one input and merging the one output set_basic_defaults(job_wrapper) - return multi.do_merge(job_wrapper, task_wrappers) - + return multi.do_merge(job_wrapper, task_wrappers) diff --git a/lib/galaxy/jobs/splitters/multi.py b/lib/galaxy/jobs/splitters/multi.py index dc8de457fc2..b4988ea4697 100644 --- a/lib/galaxy/jobs/splitters/multi.py +++ b/lib/galaxy/jobs/splitters/multi.py @@ -8,20 +8,21 @@ from galaxy import model, util log = logging.getLogger( __name__ ) -def do_split (job_wrapper): + +def do_split(job_wrapper): parent_job = job_wrapper.get_job() working_directory = os.path.abspath(job_wrapper.working_directory) parallel_settings = job_wrapper.get_parallelism().attributes # Syntax: split_inputs="input1,input2" shared_inputs="genome" # Designates inputs to be split or shared - split_inputs=parallel_settings.get("split_inputs") + split_inputs = parallel_settings.get("split_inputs") if split_inputs is None: split_inputs = [] else: split_inputs = [x.strip() for x in split_inputs.split(",")] - shared_inputs=parallel_settings.get("shared_inputs") + shared_inputs = parallel_settings.get("shared_inputs") auto_shared_inputs = False if shared_inputs is None: shared_inputs = [] @@ -32,10 +33,11 @@ def do_split (job_wrapper): if len(illegal_inputs) > 0: raise Exception("Inputs have conflicting parallelism attributes: %s" % str( illegal_inputs )) - subdir_index = [0] # use a list to get around Python 2.x lame closure support + subdir_index = [0] # use a list to get around Python 2.x lame closure support task_dirs = [] + def get_new_working_directory_name(): - dir=os.path.join(working_directory, 'task_%d' % subdir_index[0]) + dir = os.path.join(working_directory, 'task_%d' % subdir_index[0]) subdir_index[0] = subdir_index[0] + 1 if not os.path.exists(dir): os.makedirs(dir) @@ -55,9 +57,9 @@ def do_split (job_wrapper): pass else: if input.name in split_inputs: - type_to_input_map.setdefault(input.dataset.datatype, []).append(input.name) + type_to_input_map.setdefault(input.dataset.datatype, []).append(input.name) elif input.name in shared_inputs: - pass # pass original file name + pass # pass original file name elif auto_shared_inputs: shared_inputs.append(input.name) else: @@ -97,7 +99,7 @@ def do_split (job_wrapper): names = job_wrapper.get_input_dataset_fnames(input.dataset) for dir in task_dirs: for file in names: - os.symlink(file, os.path.join(dir, os.path.basename(file))) + os.symlink(file, os.path.join(dir, os.path.basename(file))) tasks = [] prepare_files = os.path.join(util.galaxy_directory(), 'extract_dataset_parts.sh') + ' %s' for dir in task_dirs: @@ -106,7 +108,7 @@ def do_split (job_wrapper): return tasks -def do_merge( job_wrapper, task_wrappers): +def do_merge( job_wrapper, task_wrappers): parallel_settings = job_wrapper.get_parallelism().attributes # Syntax: merge_outputs="export" pickone_outputs="genomesize" # Designates outputs to be merged, or selected from as a representative @@ -137,14 +139,14 @@ def do_merge( job_wrapper, task_wrappers): output_paths = job_wrapper.get_output_fnames() pickone_done = [] task_dirs = [os.path.join(working_directory, x) for x in os.listdir(working_directory) if x.startswith('task_')] - task_dirs.sort(key = lambda x: int(x.split('task_')[-1])) + task_dirs.sort(key=lambda x: int(x.split('task_')[-1])) for index, output in enumerate( outputs ): output_file_name = str( output_paths[ index ] ) # Use false_path if set, else real path. base_output_name = os.path.basename(output_file_name) if output in merge_outputs: output_dataset = outputs[output][0] output_type = output_dataset.datatype - output_files = [os.path.join(dir,base_output_name) for dir in task_dirs] + output_files = [os.path.join(dir, base_output_name) for dir in task_dirs] # Just include those files f in the output list for which the # file f exists; some files may not exist if a task fails. output_files = [ f for f in output_files if os.path.exists(f) ] @@ -170,14 +172,14 @@ def do_merge( job_wrapper, task_wrappers): # just pick one of them if output not in pickone_done: task_file_name = os.path.join(task_dirs[0], base_output_name) - shutil.move( task_file_name, output_file_name ) + shutil.move( task_file_name, output_file_name ) pickone_done.append(output) else: log_error = "The output '%s' does not define a method for implementing parallelism" % output log.exception(log_error) raise Exception(log_error) except Exception, e: - stdout = 'Error merging files'; + stdout = 'Error merging files' log.exception( stdout ) stderr = str(e) @@ -190,5 +192,4 @@ def do_merge( job_wrapper, task_wrappers): stdout += "\n" + tw.working_directory + ':\n' + out if len(err) > 0: stderr += "\n" + tw.working_directory + ':\n' + err - return (stdout, stderr) - + return (stdout, stderr) diff --git a/lib/galaxy/jobs/transfer_manager.py b/lib/galaxy/jobs/transfer_manager.py index 3e3a4af4b7c..fb332e53c76 100644 --- a/lib/galaxy/jobs/transfer_manager.py +++ b/lib/galaxy/jobs/transfer_manager.py @@ -2,13 +2,17 @@ Manage transfers from arbitrary URLs to temporary files. Socket interface for IPC with multiple process configurations. """ -import os, subprocess, socket, logging, threading +import logging +import os +import subprocess +import socket +import threading -from galaxy import eggs from galaxy.util import listify, json log = logging.getLogger( __name__ ) + class TransferManager( object ): """ Manage simple data transfers from URLs to temporary locations. @@ -23,6 +27,7 @@ class TransferManager( object ): self.sleeper = Sleeper() self.restarter = threading.Thread( target=self.__restarter ) self.restarter.start() + def new( self, path=None, **kwd ): if 'protocol' not in kwd: raise Exception( 'Missing required parameter "protocol".' ) @@ -40,6 +45,7 @@ class TransferManager( object ): self.sa_session.add( transfer_job ) self.sa_session.flush() return transfer_job + def run( self, transfer_jobs ): """ This method blocks, so if invoking the transfer manager ever starts @@ -56,8 +62,6 @@ class TransferManager( object ): self.sa_session.add_all( transfer_jobs ) self.sa_session.flush() for tj in transfer_jobs: - params_dict = tj.params - protocol = params_dict[ 'protocol' ] # The transfer script should daemonize fairly quickly - if this is # not the case, this process will need to be moved to a # non-blocking method. @@ -72,6 +76,7 @@ class TransferManager( object ): tj.info = 'Spawning transfer job failed: %s' % output.splitlines()[-1] self.sa_session.add( tj ) self.sa_session.flush() + def get_state( self, transfer_jobs, via_socket=False ): transfer_jobs = listify( transfer_jobs ) rval = [] @@ -116,11 +121,12 @@ class TransferManager( object ): if len( rval ) == 1: return rval[0] return rval + def __restarter( self ): log.info( 'Transfer job restarter starting up...' ) while self.running: dead = [] - self.sa_session.expunge_all() # our session is threadlocal so this is safe. + self.sa_session.expunge_all() # our session is threadlocal so this is safe. for tj in self.sa_session.query( self.app.model.TransferJob ) \ .filter( self.app.model.TransferJob.state == self.app.model.TransferJob.states.RUNNING ): if not tj.pid: @@ -145,10 +151,12 @@ class TransferManager( object ): self.run( dead ) self.sleeper.sleep( 30 ) log.info( 'Transfer job restarter shutting down...' ) + def shutdown( self ): self.running = False self.sleeper.wake() + class Sleeper( object ): """ Provides a 'sleep' method that sleeps for a number of seconds *unless* @@ -156,10 +164,12 @@ class Sleeper( object ): """ def __init__( self ): self.condition = threading.Condition() + def sleep( self, seconds ): self.condition.acquire() self.condition.wait( seconds ) self.condition.release() + def wake( self ): self.condition.acquire() self.condition.notify()