From da100d3b274ed077eb40af70d0ff410dbad2ffb2 Mon Sep 17 00:00:00 2001 From: Nate Coraor Date: Wed, 21 Oct 2020 16:07:28 -0400 Subject: [PATCH] Update job runners for STOPPED job state --- lib/galaxy/jobs/runners/chronos.py | 5 +++-- lib/galaxy/jobs/runners/cli.py | 13 +++++++------ lib/galaxy/jobs/runners/condor.py | 13 +++++++------ lib/galaxy/jobs/runners/godocker.py | 9 +++++---- lib/galaxy/jobs/runners/kubernetes.py | 12 +++++++----- lib/galaxy/jobs/runners/pbs.py | 10 +++++----- lib/galaxy/jobs/runners/pulsar.py | 6 +++--- 7 files changed, 37 insertions(+), 31 deletions(-) diff --git a/lib/galaxy/jobs/runners/chronos.py b/lib/galaxy/jobs/runners/chronos.py index 5f7aa3dfee5..5e5c241025b 100644 --- a/lib/galaxy/jobs/runners/chronos.py +++ b/lib/galaxy/jobs/runners/chronos.py @@ -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'] diff --git a/lib/galaxy/jobs/runners/cli.py b/lib/galaxy/jobs/runners/cli.py index 9345953a3f4..8ddfe710ddb 100644 --- a/lib/galaxy/jobs/runners/cli.py +++ b/lib/galaxy/jobs/runners/cli.py @@ -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) diff --git a/lib/galaxy/jobs/runners/condor.py b/lib/galaxy/jobs/runners/condor.py index 41146d6fb64..a422ac257ad 100644 --- a/lib/galaxy/jobs/runners/condor.py +++ b/lib/galaxy/jobs/runners/condor.py @@ -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: diff --git a/lib/galaxy/jobs/runners/godocker.py b/lib/galaxy/jobs/runners/godocker.py index bfe2d4b52d7..9a997e19ae7 100644 --- a/lib/galaxy/jobs/runners/godocker.py +++ b/lib/galaxy/jobs/runners/godocker.py @@ -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) diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py index 5e390cd36e4..b8bf3232322 100644 --- a/lib/galaxy/jobs/runners/kubernetes.py +++ b/lib/galaxy/jobs/runners/kubernetes.py @@ -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) diff --git a/lib/galaxy/jobs/runners/pbs.py b/lib/galaxy/jobs/runners/pbs.py index c6aa74595a9..3074bac6b5d 100644 --- a/lib/galaxy/jobs/runners/pbs.py +++ b/lib/galaxy/jobs/runners/pbs.py @@ -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) diff --git a/lib/galaxy/jobs/runners/pulsar.py b/lib/galaxy/jobs/runners/pulsar.py index 2e8434ef30c..faf51fdd996 100644 --- a/lib/galaxy/jobs/runners/pulsar.py +++ b/lib/galaxy/jobs/runners/pulsar.py @@ -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)