From 4a132fb52c8a066ad6e4dd08360e04bb8af7ba67 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Tue, 26 Mar 2013 15:40:36 -0400 Subject: [PATCH] Fix setting of external metadata broken by refactoring job preparation. --- lib/galaxy/jobs/runners/__init__.py | 22 +++++++++++++++------- lib/galaxy/jobs/runners/cli.py | 7 +++---- lib/galaxy/jobs/runners/condor.py | 7 +++---- lib/galaxy/jobs/runners/drmaa.py | 9 ++++----- lib/galaxy/jobs/runners/local.py | 7 +++---- lib/galaxy/jobs/runners/pbs.py | 7 +++---- lib/galaxy/jobs/runners/tasks.py | 8 +++++--- 7 files changed, 36 insertions(+), 31 deletions(-) diff --git a/lib/galaxy/jobs/runners/__init__.py b/lib/galaxy/jobs/runners/__init__.py index a71f47b2941..f3118b137f6 100644 --- a/lib/galaxy/jobs/runners/__init__.py +++ b/lib/galaxy/jobs/runners/__init__.py @@ -95,35 +95,43 @@ class BaseJobRunner( object ): """ raise NotImplementedError() - # Runners must override the job handling methods - def queue_job(self, job_wrapper): + def prepare_job(self, job_wrapper, include_metadata=False, include_work_dir_outputs=True): """Some sanity checks that all runners' queue_job() methods are likely to want to do """ job_id = job_wrapper.get_id_tag() job_state = job_wrapper.get_state() job_wrapper.is_ready = False + job_wrapper.runner_command_line = None # Make sure the job hasn't been deleted if job_state == model.Job.states.DELETED: log.debug( "(%s) Job deleted by user before it entered the %s queue" % ( job_id, self.runner_name ) ) if self.app.config.cleanup_job in ( "always", "onsuccess" ): job_wrapper.cleanup() - return + return False elif job_state != model.Job.states.QUEUED: log.info( "(%d) Job is in state %s, skipping execution" % ( job_id, job_state ) ) # cleanup may not be safe in all states - return + return False # Prepare the job try: job_wrapper.prepare() - job_wrapper.runner_command_line = self.build_command_line( job_wrapper ) + 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 ) - return + return False - job_wrapper.is_ready = True + if not job_wrapper.runner_command_line: + job_wrapper.finish( '', '' ) + return False + + return True + + # Runners must override the job handling methods + def queue_job(self, job_wrapper): + raise NotImplementedError() def stop_job(self, job): raise NotImplementedError() diff --git a/lib/galaxy/jobs/runners/cli.py b/lib/galaxy/jobs/runners/cli.py index 41f78082d44..f59648eeab6 100644 --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -75,12 +75,11 @@ class ShellJobRunner( AsynchronousJobRunner ): def queue_job( self, job_wrapper ): """Create job script and submit it to the DRM""" - # Superclass method has some basic sanity checks - super( ShellJobRunner, self ).queue_job( job_wrapper ) - if not job_wrapper.is_ready: + # prepare the job + if not self.prepare_job( job_wrapper, include_metadata=True ): return - # command line has been added to the wrapper by the superclass queue_job() + # command line has been added to the wrapper by prepare_job() command_line = job_wrapper.runner_command_line # Get shell and job execution interface diff --git a/lib/galaxy/jobs/runners/condor.py b/lib/galaxy/jobs/runners/condor.py index 8aefcb41c55..9378a8fafac 100644 --- a/lib/galaxy/jobs/runners/condor.py +++ b/lib/galaxy/jobs/runners/condor.py @@ -58,12 +58,11 @@ class CondorJobRunner( AsynchronousJobRunner ): def queue_job( self, job_wrapper ): """Create job script and submit it to the DRM""" - # Superclass method has some basic sanity checks - super( CondorJobRunner, self ).queue_job( job_wrapper ) - if not job_wrapper.is_ready: + # prepare the job + if not self.prepare_job( job_wrapper, include_metadata=True ): return - # command line has been added to the wrapper by the superclass queue_job() + # command line has been added to the wrapper by prepare_job() command_line = job_wrapper.runner_command_line # get configured job destination diff --git a/lib/galaxy/jobs/runners/drmaa.py b/lib/galaxy/jobs/runners/drmaa.py index e3c0881be57..fca44fd15f9 100644 --- a/lib/galaxy/jobs/runners/drmaa.py +++ b/lib/galaxy/jobs/runners/drmaa.py @@ -105,14 +105,13 @@ class DRMAAJobRunner( AsynchronousJobRunner ): def queue_job( self, job_wrapper ): """Create job script and submit it to the DRM""" - # Superclass method has some basic sanity checks - super( DRMAAJobRunner, self ).queue_job( job_wrapper ) - if not job_wrapper.is_ready: + # prepare the job + if not self.prepare_job( job_wrapper, include_metadata=True ): return - # command line has been added to the wrapper by the superclass queue_job() + # 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 diff --git a/lib/galaxy/jobs/runners/local.py b/lib/galaxy/jobs/runners/local.py index bdc870a2bec..dfab9c8c881 100644 --- a/lib/galaxy/jobs/runners/local.py +++ b/lib/galaxy/jobs/runners/local.py @@ -35,15 +35,14 @@ class LocalJobRunner( BaseJobRunner ): self._init_worker_threads() def queue_job( self, job_wrapper ): - # Superclass method has some basic sanity checks - super( LocalJobRunner, self ).queue_job( job_wrapper ) - if not job_wrapper.is_ready: + # prepare the job + if not self.prepare_job( job_wrapper ): return stderr = stdout = '' exit_code = 0 - # command line has been added to the wrapper by the superclass queue_job() + # command line has been added to the wrapper by prepare_job() command_line = job_wrapper.runner_command_line job_id = job_wrapper.get_id_tag() diff --git a/lib/galaxy/jobs/runners/pbs.py b/lib/galaxy/jobs/runners/pbs.py index 6a18fab38e6..a9685107fc3 100644 --- a/lib/galaxy/jobs/runners/pbs.py +++ b/lib/galaxy/jobs/runners/pbs.py @@ -222,12 +222,11 @@ class PBSJobRunner( AsynchronousJobRunner ): def queue_job( self, job_wrapper ): """Create PBS script for a job and submit it to the PBS queue""" - # Superclass method has some basic sanity checks - super( PBSJobRunner, self ).queue_job( job_wrapper ) - if not job_wrapper.is_ready: + # prepare the job + 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 the superclass queue_job() + # command line has been added to the wrapper by prepare_job() command_line = job_wrapper.runner_command_line job_destination = job_wrapper.job_destination diff --git a/lib/galaxy/jobs/runners/tasks.py b/lib/galaxy/jobs/runners/tasks.py index 5fc23f5efec..7245cad29cf 100644 --- a/lib/galaxy/jobs/runners/tasks.py +++ b/lib/galaxy/jobs/runners/tasks.py @@ -26,13 +26,15 @@ class TaskedJobRunner( BaseJobRunner ): self._init_worker_threads() def queue_job( self, job_wrapper ): - super( TaskedJobRunner, self ).queue_job( job_wrapper ) - if not job_wrapper.is_ready: + # prepare the job + if not self.prepare_job( job_wrapper ): return - stderr = stdout = '' + # command line has been added to the wrapper by prepare_job() command_line = job_wrapper.runner_command_line + stderr = stdout = '' + # Persist the destination job_wrapper.set_job_destination(job_wrapper.job_destination)