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

Consolidate IT cleanup

Browse Source
This commit is contained in:
almahmoud
2021-05-06 11:50:52 -04:00
parent e1eef06243
commit 6254961cfd
2 changed files with 20 additions and 69 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lib/galaxy/jobs/runners/kubernetes.py
+20 -39
View File
@@ -661,6 +661,8 @@ class KubernetesJobRunner(AsynchronousJobRunner):
job_state.running = False
self.mark_as_failed(job_state)
try:
if job_state.job_wrapper.guest_ports:
self.__cleanup_k8s_interactivetools(job_state.job_wrapper, job)
self.__cleanup_k8s_job(job)
except Exception:
log.exception("Could not clean up k8s batch job. Ignoring...")
@@ -684,19 +686,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
self.mark_as_failed(job_state)
try:
if job_state.job_wrapper.guest_ports:
k8s_job_prefix = self.__produce_k8s_job_prefix()
k8s_job_name = self.__get_k8s_job_name(k8s_job_prefix, job_state.job_wrapper)
job_failed = (job.obj['status']['failed'] > 0
if 'failed' in job.obj['status'] else False)
log.debug(f'Deleting service/ingress for job with ID {job_state.job_wrapper.get_id_tag()}')
ingress_to_delete = find_ingress_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if ingress_to_delete and len(ingress_to_delete.response['items']) > 0:
k8s_ingress = Ingress(self._pykube_api, ingress_to_delete.response['items'][0])
self.__cleanup_k8s_ingress(k8s_ingress, job_failed)
service_to_delete = find_service_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if service_to_delete and len(service_to_delete.response['items']) > 0:
k8s_service = Service(self._pykube_api, service_to_delete.response['items'][0])
self.__cleanup_k8s_service(k8s_service, job_failed)
self.__cleanup_k8s_interactivetools(job_state.job_wrapper, job)
self.__cleanup_k8s_job(job)
except Exception:
log.exception("Could not clean up k8s batch job. Ignoring...")
@@ -755,6 +745,21 @@ class KubernetesJobRunner(AsynchronousJobRunner):
pod = Pod(self._pykube_api, pods.response['items'][0])
return is_pod_unschedulable(self._pykube_api, pod, self.runner_params['k8s_namespace'])
def __cleanup_k8s_interactivetools(self, job_wrapper, k8s_job):
k8s_job_prefix = self.__produce_k8s_job_prefix()
k8s_job_name = "{}-{}".format(k8s_job_prefix, self.__force_label_conformity(job_wrapper.get_id_tag()))
log.debug(f'Deleting service/ingress for job with ID {job_wrapper.get_id_tag()}')
job_failed = (k8s_job.obj['status']['failed'] > 0
if 'failed' in k8s_job.obj['status'] else False)
ingress_to_delete = find_ingress_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if ingress_to_delete and len(ingress_to_delete.response['items']) > 0:
k8s_ingress = Ingress(self._pykube_api, ingress_to_delete.response['items'][0])
self.__cleanup_k8s_ingress(k8s_ingress, job_failed)
service_to_delete = find_service_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if service_to_delete and len(service_to_delete.response['items']) > 0:
k8s_service = Service(self._pykube_api, service_to_delete.response['items'][0])
self.__cleanup_k8s_service(k8s_service, job_failed)
def stop_job(self, job_wrapper):
"""Attempts to delete a dispatched job to the k8s cluster"""
job = job_wrapper.get_job()
@@ -763,19 +768,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
if job_to_delete and len(job_to_delete.response['items']) > 0:
k8s_job = Job(self._pykube_api, job_to_delete.response['items'][0])
if job_wrapper.guest_ports:
k8s_job_prefix = self.__produce_k8s_job_prefix()
k8s_job_name = "{}-{}".format(k8s_job_prefix, self.__force_label_conformity(job_wrapper.get_id_tag()))
log.debug(f'Deleting service/ingress for job with ID {job_wrapper.get_id_tag()}')
job_failed = (k8s_job.obj['status']['failed'] > 0
if 'failed' in k8s_job.obj['status'] else False)
ingress_to_delete = find_ingress_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if ingress_to_delete and len(ingress_to_delete.response['items']) > 0:
k8s_ingress = Ingress(self._pykube_api, ingress_to_delete.response['items'][0])
self.__cleanup_k8s_ingress(k8s_ingress, job_failed)
service_to_delete = find_service_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if service_to_delete and len(service_to_delete.response['items']) > 0:
k8s_service = Service(self._pykube_api, service_to_delete.response['items'][0])
self.__cleanup_k8s_service(k8s_service, job_failed)
self.__cleanup_k8s_interactivetools(job_wrapper, k8s_job)
self.__cleanup_k8s_job(k8s_job)
# TODO assert whether job parallelism == 0
# assert not job_to_delete.exists(), "Could not delete job,"+job.job_runner_external_id+" it still exists"
@@ -818,17 +811,5 @@ class KubernetesJobRunner(AsynchronousJobRunner):
" in job id '%s'", job_state.job_id)
job = Job(self._pykube_api, jobs.response['items'][0])
if job_state.job_wrapper.guest_ports:
k8s_job_prefix = self.__produce_k8s_job_prefix()
k8s_job_name = self.__get_k8s_job_name(k8s_job_prefix, job_state.job_wrapper)
job_failed = (job.obj['status']['failed'] > 0
if 'failed' in job.obj['status'] else False)
log.debug(f'Deleting service/ingress for job with ID {job_state.job_wrapper.get_id_tag()}')
ingress_to_delete = find_ingress_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if ingress_to_delete and len(ingress_to_delete.response['items']) > 0:
k8s_ingress = Ingress(self._pykube_api, ingress_to_delete.response['items'][0])
self.__cleanup_k8s_ingress(k8s_ingress, job_failed)
service_to_delete = find_service_object_by_name(self._pykube_api, k8s_job_name, self.runner_params['k8s_namespace'])
if service_to_delete and len(service_to_delete.response['items']) > 0:
k8s_service = Service(self._pykube_api, service_to_delete.response['items'][0])
self.__cleanup_k8s_service(k8s_service, job_failed)
self.__cleanup_k8s_interactivetools(job_state.job_wrapper, job)
self.__cleanup_k8s_job(job)
tools/interactive/interactivetool_terminal.xml
-30
View File
@@ -1,30 +0,0 @@
<tool id="interactive_tool_terminal" tool_type="interactive" name="Terminal" version="0.1">
<requirements>
<container type="docker">cloudve/ttyd:latest</container>
</requirements>
<entry_points>
<entry_point name="Terminal" requires_domain="True">
<port>7681</port>
<url>terminal</url>
</entry_point>
</entry_points>
<environment_variables>
<environment_variable name="HISTORY_ID" strip="True">${__app__.security.encode_id($terminal.history_id)}</environment_variable>
<environment_variable name="REMOTE_HOST">${__app__.config.galaxy_infrastructure_url}</environment_variable>
<environment_variable name="GALAXY_WEB_PORT">80</environment_variable>
<environment_variable name="GALAXY_URL">$__galaxy_url__</environment_variable>
<environment_variable name="DEBUG">true</environment_variable>
<environment_variable name="DISABLE_AUTH">true</environment_variable>
<environment_variable name="API_KEY" inject="api_key" />
</environment_variables>
<command detect_errors="aggressive"><![CDATA[
ttyd --base-path "/terminal" bash
]]>
</command>
<outputs>
<data name="terminal" format="" label="Interactive Terminal"></data>
</outputs>
<inputs>
<!--<param name="input" type="data" optional="true" label="Include data into the environment"/>-->
</inputs>
</tool>