opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix and simplify k8s error handling

Browse Source
This commit is contained in:
mvdbeek
2019-07-10 17:16:37 +02:00
parent ff0d1f142e
commit 6fc3d09b60
1 changed files with 15 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/runners/kubernetes.py
+15 -9
View File
@@ -430,23 +430,18 @@ class KubernetesJobRunner(AsynchronousJobRunner):
job_state.running = False
self.mark_as_finished(job_state)
return None
elif failed > 0 and self.__job_failed_due_to_low_memory(job_state):
return self._handle_job_failure(job, job_state, reason="OOM")
elif active > 0 and failed <= max_pod_retries:
if not job_state.running:
job_state.running = True
job_state.job_wrapper.change_state(model.Job.states.RUNNING)
return job_state
elif failed > max_pod_retries:
return self._handle_job_failure(job, job_state)
elif job_state.job_wrapper.get_job().state == model.Job.states.DELETED:
# Job has been deleted via stop_job, cleanup and remove from watched_jobs by returning `None`
if job_state.job_wrapper.cleanup_job in ("always", "onsuccess"):
job_state.job_wrapper.cleanup()
return None
else:
# We really shouldn't reach this point, but we might if the job has been killed by the kubernetes admin
log.info("Kubernetes job '%s' not classified as succ., active or failed. Full Job object: \n%s", job.name, job.obj)
return self._handle_job_failure(job, job_state)
elif len(jobs.response['items']) == 0:
# there is no job responding to this job_id, it is either lost or something happened.
@@ -463,12 +458,17 @@ class KubernetesJobRunner(AsynchronousJobRunner):
self.mark_as_failed(job_state)
return job_state
def _handle_job_failure(self, job, job_state, reason=None):
def _handle_job_failure(self, job, job_state):
# Figure out why job has failed
with open(job_state.error_file, 'a') as error_file:
if reason == "OOM":
if self.__job_failed_due_to_low_memory(job_state):
error_file.write("Job killed after running out of memory. Try with more memory.\n")
job_state.fail_message = "Tool failed due to insufficient memory. Try with more memory."
job_state.runner_state = JobState.runner_states.MEMORY_LIMIT_REACHED
elif self.__job_failed_due_to_walltime_limit(job):
error_file.write("DeadlineExceeded")
job_state.fail_message = "Job was active longer than specified deadline"
job_state.runner_state = JobState.runner_states.WALLTIME_REACHED
else:
error_file.write("Exceeded max number of Kubernetes pod retrials allowed for job\n")
job_state.fail_message = "More pods failed than allowed. See stdout for pods details."
@@ -477,6 +477,10 @@ class KubernetesJobRunner(AsynchronousJobRunner):
job.scale(replicas=0)
return None
def __job_failed_due_to_walltime_limit(self, job):
conditions = job.obj['status']['conditions']
return any(True for c in conditions if c['type'] == 'Failed' and c['reason'] == 'DeadlineExceeded')
def __job_failed_due_to_low_memory(self, job_state):
"""
checks the state of the pod to see if it was killed
@@ -486,8 +490,10 @@ class KubernetesJobRunner(AsynchronousJobRunner):
pods = Pod.objects(self._pykube_api).filter(selector="app=%s" % job_state.job_id,
namespace=self.runner_params['k8s_namespace'])
pod = Pod(self._pykube_api, pods.response['items'][0])
if not pods.response['items']:
return False
pod = Pod(self._pykube_api, pods.response['items'][0])
if pod.obj['status']['phase'] == "Failed" and \
pod.obj['status']['containerStatuses'][0]['state']['terminated']['reason'] == "OOMKilled":
return True