Add support for k8s ttlSecondsAfterFinished

Author: nuwang

lib/galaxy/jobs/runners/kubernetes.py

@@ -52,6 +52,7 @@ def __init__(self, app, nworkers, **kwargs):
             k8s_galaxy_instance_id=dict(map=str),
             k8s_timeout_seconds_job_deletion=dict(map=int, valid=lambda x: int > 0, default=30),
             k8s_job_api_version=dict(map=str, default="batch/v1"),
+            k8s_job_ttl_secs_after_finished=dict(map=int, valid=lambda x: x is None or int(x) >= 0, default=300),
             k8s_supplemental_group_id=dict(map=str),
             k8s_pull_policy=dict(map=str, default="Default"),
             k8s_fs_group_id=dict(map=int),
@@ -221,6 +222,9 @@ def __produce_unique_k8s_job_name(self, galaxy_internal_job_id):
     def __get_k8s_job_spec(self, ajs):
         """Creates the k8s Job spec. For a Job spec, the only requirement is to have a .spec.template."""
         k8s_job_spec = {"template": self.__get_k8s_job_spec_template(ajs)}
+        job_ttl = self.runner_params["k8s_job_ttl_secs_after_finished"]
+        if self.runner_params["k8s_cleanup_job"] != "never" and job_ttl is not None:
+            k8s_job_spec["ttlSecondsAfterFinished"] = job_ttl
         return k8s_job_spec
 
     def __get_k8s_job_spec_template(self, ajs):
@@ -478,13 +482,18 @@ def __cleanup_k8s_job(self, job):
         job.scale(replicas=0)
         if (k8s_cleanup_job == "always" or
                 (k8s_cleanup_job == "onsuccess" and not job_failed)):
-            delete_options = {
-                "apiVersion": "v1",
-                "kind": "DeleteOptions",
-                "propagationPolicy": "Background"
-            }
-            r = job.api.delete(json=delete_options, **job.api_kwargs())
-            job.api.raise_for_status(r)
+            job_ttl = self.runner_params["k8s_job_ttl_secs_after_finished"]
+            if job_ttl is None:
+                delete_options = {
+                    "apiVersion": "v1",
+                    "kind": "DeleteOptions",
+                    "propagationPolicy": "Background"
+                }
+                r = job.api.delete(json=delete_options, **job.api_kwargs())
+                job.api.raise_for_status(r)
+            else:
+                # Let the k8s ttl take care of deletion
+                pass
 
     def __job_failed_due_to_low_memory(self, job_state):
         """