opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-11-03 04:43:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update job runners for STOPPED job state

Browse Source
This commit is contained in:
Nate Coraor
2020-12-11 17:17:03 +01:00
committed by mvdbeek
parent b47da22cae
commit da100d3b27
7 changed files with 37 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/runners/chronos.py
+3 -2
View File
@@ -165,10 +165,10 @@ class ChronosJobRunner(AsynchronousJobRunner):
ajs.command_line = job.command_line
ajs.job_wrapper = job_wrapper
ajs.job_destination = job_wrapper.job_destination
if job.state == model.Job.states.RUNNING:
if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED):
LOGGER.debug(msg.format(
name=job.id, runner=job.job_runner_external_id,
state='running'))
state=job.state))
ajs.old_state = model.Job.states.RUNNING
ajs.running = True
self.monitor_queue.put(ajs)
@@ -184,6 +184,7 @@ class ChronosJobRunner(AsynchronousJobRunner):
def check_watched_item(self, job_state):
job_name = job_state.job_id
job = self._retrieve_job(job_name)
# TODO: how can stopped GxIT jobs be handled here?
if job:
succeeded = job['successCount']
errors = job['errorCount']
lib/galaxy/jobs/runners/cli.py
+7 -6
View File
@@ -88,8 +88,8 @@ class ShellJobRunner(AsynchronousJobRunner):
return
# job was deleted while we were preparing it
if job_wrapper.get_state() == model.Job.states.DELETED:
log.info("(%s) Job deleted by user before it entered the queue" % galaxy_id_tag)
if job_wrapper.get_state() in (model.Job.states.DELETED, model.Job.states.STOPPED):
log.debug("(%s) Job deleted/stopped by user before it entered the queue", galaxy_id_tag)
if job_wrapper.cleanup_job in ("always", "onsuccess"):
job_wrapper.cleanup()
return
@@ -168,12 +168,13 @@ class ShellJobRunner(AsynchronousJobRunner):
state = job_interface.parse_single_status(cmd_out.stdout, external_job_id)
if not state == model.Job.states.OK:
log.warning(f'({id_tag}/{external_job_id}) job not found in batch state check, but found in individual state check')
job_state = ajs.job_wrapper.get_state()
if state != old_state:
log.debug(f"({id_tag}/{external_job_id}) state change: from {old_state} to {state}")
if not state == model.Job.states.OK:
# No need to change_state when the state is OK, this will be handled by `self.finish_job`
ajs.job_wrapper.change_state(state)
if state == model.Job.states.ERROR:
if state == model.Job.states.ERROR and job_state != model.Job.states.STOPPED:
# Try to find out the reason for exiting
self.__handle_out_of_memory(ajs, external_job_id)
self.work_queue.put((self.mark_as_failed, ajs))
@@ -182,7 +183,7 @@ class ShellJobRunner(AsynchronousJobRunner):
if state == model.Job.states.RUNNING and not ajs.running:
ajs.running = True
ajs.old_state = state
if state == model.Job.states.OK:
if state == model.Job.states.OK or job_state == model.Job.states.STOPPED:
external_metadata = not asbool(ajs.job_wrapper.job_destination.params.get("embed_metadata_in_job", DEFAULT_EMBED_METADATA_IN_JOB))
if external_metadata:
self.work_queue.put((self.handle_metadata_externally, ajs))
@@ -249,8 +250,8 @@ class ShellJobRunner(AsynchronousJobRunner):
ajs.command_line = job.command_line
ajs.job_wrapper = job_wrapper
ajs.job_destination = job_wrapper.job_destination
if job.state == model.Job.states.RUNNING:
log.debug(f"({job.id}/{job.job_runner_external_id}) is still in running state, adding to the runner monitor queue")
if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED):
log.debug(f"({job.id}/{job.job_runner_external_id}) is still in {job.state} state, adding to the runner monitor queue")
ajs.old_state = model.Job.states.RUNNING
ajs.running = True
self.monitor_queue.put(ajs)
lib/galaxy/jobs/runners/condor.py
+7 -6
View File
@@ -147,8 +147,8 @@ class CondorJobRunner(AsynchronousJobRunner):
return
# job was deleted while we were preparing it
if job_wrapper.get_state() == model.Job.states.DELETED:
log.debug("Job %s deleted by user before it entered the queue" % galaxy_id_tag)
if job_wrapper.get_state() in (model.Job.states.DELETED, model.Job.states.STOPPED):
log.debug("(%s) Job deleted/stopped by user before it entered the queue", galaxy_id_tag)
if cleanup_job in ("always", "onsuccess"):
os.unlink(submit_file)
cjs.cleanup()
@@ -218,8 +218,9 @@ class CondorJobRunner(AsynchronousJobRunner):
log.debug(f"({galaxy_id_tag}/{job_id}) job has stopped running")
# Will switching from RUNNING to QUEUED confuse Galaxy?
# cjs.job_wrapper.change_state( model.Job.states.QUEUED )
if job_complete:
if cjs.job_wrapper.get_state() != model.Job.states.DELETED:
job_state = cjs.job_wrapper.get_state()
if job_complete or job_state == model.Job.states.STOPPED:
if job_state != model.Job.states.DELETED:
external_metadata = not asbool(cjs.job_wrapper.job_destination.params.get("embed_metadata_in_job", True))
if external_metadata:
self._handle_metadata_externally(cjs.job_wrapper, resolve_requirements=True)
@@ -292,8 +293,8 @@ class CondorJobRunner(AsynchronousJobRunner):
cjs.user_log = os.path.join(job_wrapper.working_directory, 'galaxy_%s.condor.log' % galaxy_id_tag)
cjs.register_cleanup_file_attribute('user_log')
self.__old_state_paths(cjs) # remove in 21.01
if job.state == model.Job.states.RUNNING:
log.debug(f"({job.id}/{job.job_runner_external_id}) is still in running state, adding to the DRM queue")
if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED):
log.debug(f"({job.id}/{job.get_job_runner_external_id()}) is still in {job.state} state, adding to the DRM queue")
cjs.running = True
self.monitor_queue.put(cjs)
elif job.state == model.Job.states.QUEUED:
lib/galaxy/jobs/runners/godocker.py
+5 -4
View File
@@ -178,10 +178,11 @@ class GodockerJobRunner(AsynchronousJobRunner):
AsynchronousJobState object with updated running status.
'''
""" Get task from GoDocker """
job_persisted_state = job_state.job_wrapper.get_state()
job_status_god = self.get_task(job_state.job_id)
log.debug("Job ID: " + str(job_state.job_id) + " Job Status: " + str(job_status_god['status']['primary']))
if job_status_god['status']['primary'] == "over":
if job_status_god['status']['primary'] == "over" or job_persisted_state == model.Job.states.STOPPED:
job_state.running = False
job_state.job_wrapper.change_state(model.Job.states.OK)
if self.create_log_file(job_state, job_status_god):
@@ -213,7 +214,7 @@ class GodockerJobRunner(AsynchronousJobRunner):
elif job_status_god['status']['primary'] == "pending":
return job_state
elif job_status_god['status']['exitcode'] not in [None, 0]:
elif job_status_god['status']['exitcode'] not in [None, 0] and job_persisted_state != model.Job.states.STOPPED:
job_state.running = False
job_state.job_wrapper.change_state(model.Job.states.ERROR)
self.create_log_file(job_state, job_status_god)
@@ -252,8 +253,8 @@ class GodockerJobRunner(AsynchronousJobRunner):
ajs.job_destination = job_wrapper.job_destination
job_wrapper.command_line = job.command_line
ajs.job_wrapper = job_wrapper
if job.state == model.Job.states.RUNNING:
log.debug(f"({job.id}/{job.get_job_runner_external_id()}) is still in running state, adding to the god queue")
if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED):
log.debug(f"({job.id}/{job.get_job_runner_external_id()}) is still in {job.state} state, adding to the god queue")
ajs.old_state = 'R'
ajs.running = True
self.monitor_queue.put(ajs)
lib/galaxy/jobs/runners/kubernetes.py
+7 -5
View File
@@ -425,8 +425,10 @@ class KubernetesJobRunner(AsynchronousJobRunner):
if 'failed' in job.obj['status']:
failed = job.obj['status']['failed']
job_persisted_state = job_state.job_wrapper.get_state()
# This assumes jobs dependent on a single pod, single container
if succeeded > 0:
if succeeded > 0 or job_state == model.Job.states.STOPPED:
job_state.running = False
self.mark_as_finished(job_state)
return None
@@ -435,7 +437,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
job_state.running = True
job_state.job_wrapper.change_state(model.Job.states.RUNNING)
return job_state
elif job_state.job_wrapper.get_job().state == model.Job.states.DELETED:
elif job_persisted_state == model.Job.states.DELETED:
# Job has been deleted via stop_job and job has not been deleted,
# remove from watched_jobs by returning `None`
if job_state.job_wrapper.cleanup_job in ("always", "onsuccess"):
@@ -538,9 +540,9 @@ class KubernetesJobRunner(AsynchronousJobRunner):
ajs.command_line = job.command_line
ajs.job_wrapper = job_wrapper
ajs.job_destination = job_wrapper.job_destination
if job.state == model.Job.states.RUNNING:
log.debug("({}/{}) is still in running state, adding to the runner monitor queue".format(
job.id, job.job_runner_external_id))
if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED):
log.debug("({}/{}) is still in {} state, adding to the runner monitor queue".format(
job.id, job.job_runner_external_id, job.state))
ajs.old_state = model.Job.states.RUNNING
ajs.running = True
self.monitor_queue.put(ajs)
lib/galaxy/jobs/runners/pbs.py
+5 -5
View File
@@ -290,8 +290,8 @@ class PBSJobRunner(AsynchronousJobRunner):
job_file = f"{self.app.config.cluster_files_directory}/{job_wrapper.job_id}.sh"
self.write_executable_script(job_file, script)
# job was deleted while we were preparing it
if job_wrapper.get_state() == model.Job.states.DELETED:
log.debug("Job %s deleted by user before it entered the PBS queue" % job_wrapper.job_id)
if job_wrapper.get_state() in (model.Job.states.DELETED, model.Job.states.STOPPED):
log.debug("Job %s deleted/stopped by user before it entered the PBS queue" % job_wrapper.job_id)
pbs.pbs_disconnect(c)
if job_wrapper.cleanup_job in ("always", "onsuccess"):
self.cleanup((ofile, efile, ecfile, job_file))
@@ -395,7 +395,7 @@ class PBSJobRunner(AsynchronousJobRunner):
elif status.job_state == "C":
# "keep_completed" is enabled in PBS, so try to check exit status
try:
assert int(status.exit_status) == 0
assert int(status.exit_status) == 0 or pbs_job_state.job_wrapper.get_state() == model.Job.states.STOPPED
log.debug(f"({galaxy_job_id}/{job_id}) PBS job has completed successfully")
except AssertionError:
exit_status = int(status.exit_status)
@@ -545,8 +545,8 @@ class PBSJobRunner(AsynchronousJobRunner):
pbs_job_state.job_destination = job_wrapper.job_destination
job_wrapper.command_line = job.command_line
pbs_job_state.job_wrapper = job_wrapper
if job.state == model.Job.states.RUNNING:
log.debug(f"({job.id}/{job.get_job_runner_external_id()}) is still in running state, adding to the PBS queue")
if job.state in (model.Job.states.RUNNING, model.Job.states.STOPPED):
log.debug(f"({job.id}/{job.get_job_runner_external_id()}) is still in {job.state} state, adding to the PBS queue")
pbs_job_state.old_state = 'R'
pbs_job_state.running = True
self.monitor_queue.put(pbs_job_state)
lib/galaxy/jobs/runners/pulsar.py
+3 -3
View File
@@ -305,7 +305,7 @@ class PulsarJobRunner(AsynchronousJobRunner):
def _update_job_state_for_status(self, job_state, pulsar_status, full_status=None):
log.debug('(%s) Received status update: %s %s', job_state.job_id, type(pulsar_status), pulsar_status)
if pulsar_status == "complete":
if pulsar_status == "complete" or job_state.job_wrapper.get_state() == model.Job.states.STOPPED:
self.mark_as_finished(job_state)
return None
if pulsar_status in ["failed", "lost"]:
@@ -705,8 +705,8 @@ class PulsarJobRunner(AsynchronousJobRunner):
job_state = self._job_state(job, job_wrapper)
job_wrapper.command_line = job.get_command_line()
state = job.get_state()
if state in [model.Job.states.RUNNING, model.Job.states.QUEUED]:
log.debug("(Pulsar/%s) is still in running state, adding to the Pulsar queue" % (job.id))
if state in [model.Job.states.RUNNING, model.Job.states.QUEUED, model.Job.states.STOPPED]:
log.debug(f"(Pulsar/{job.id}) is still in {state} state, adding to the Pulsar queue")
job_state.old_state = True
job_state.running = state == model.Job.states.RUNNING
self.monitor_queue.put(job_state)