Fix setting of external metadata broken by refactoring job preparation.

This commit is contained in:
Nate Coraor
2013-03-26 15:40:36 -04:00
parent 1f7504620a
commit 4a132fb52c
7 changed files with 36 additions and 31 deletions
+15 -7
View File
@@ -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()
+3 -4
View File
@@ -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
+3 -4
View File
@@ -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
+4 -5
View File
@@ -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
+3 -4
View File
@@ -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()
+3 -4
View File
@@ -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
+5 -3
View File
@@ -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)