From 1253eff2f5737183dbd702382534e920b32b2270 Mon Sep 17 00:00:00 2001
From: Pablo Moreno <pablo.a.moreno@gmail.com>
Date: Wed, 4 Oct 2017 16:57:33 +0100
Subject: [PATCH 1/5] Adds k8s support for CPU and memory resource settings.

---
 config/job_conf.xml.sample_advanced   |  33 +++++++++
 lib/galaxy/jobs/runners/kubernetes.py | 102 ++++++++++++++++++++++++++
 2 files changed, 135 insertions(+)

diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced
index bdb4f8f0bbe..c4a4b04f7c3 100644
--- a/config/job_conf.xml.sample_advanced
+++ b/config/job_conf.xml.sample_advanced
@@ -248,6 +248,39 @@
                  a requirement of the other.
 
                  -->
+            <!-- <param id="k8s_default_requests_cpu">"500m"</param> -->
+            <!-- <param id="k8s_default_requests_memory">"500Mi"</param> -->
+            <!-- <param id="k8s_default_limits_cpu">"2"</param> -->
+            <!-- <param id="k8s_default_limits_memory">"2Gi"</param> -->
+            <!-- Kubernetes resource Requests and Limits
+                 Parameters above (k8s_default_requests_* and k8s_default_limits_*) set default minimal (requests) and
+                 maximal (limits) CPU and memory resources to be allocated to all containers in Kubernetes jobs by
+                 default.
+
+                 Limits and requests for CPU resources are measured in cpu units. Fractional requests are allowed. A
+                 Container with requests_cpu of 0.5 is guaranteed half as much CPU as one that asks for 1 CPU. The
+                 expression 0.1 is equivalent to the expression 100m, which can be read as “one hundred millicpu”. Some
+                 people say “one hundred millicores”, and this is understood to mean the same thing. A request with a
+                 decimal point, like 0.1, is converted to 100m by the API, and precision finer than 1m is not allowed.
+                 For this reason, the form 100m might be preferred.
+
+                 You can express memory as a plain integer or as a fixed-point integer using one of these suffixes: E,
+                 P, T, G, M, K. You can also use the power-of-two equivalents: Ei, Pi, Ti, Gi, Mi, Ki.
+                 For instance, "1Gi" stands for 1 Gigabyte, and "500Mi" stands for 500 megabytes. Using other formats
+                 will make the jobs fail as Kubernetes won't recognize the assignment.
+
+                 For more details on the Kubernetes resources management, see:
+                 https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/
+
+                 It is the responsability of the tool developer or administrator to override these limits, either at the
+                 destination or the tool itself. If not set, current defaults set on the code are:
+
+                 requests_cpu     = "500m"
+                 limits_cpu       = "1"
+                 requests_memory  = "500Mi"
+                 limits_memory    = "1Gi"
+                 -->
+
         </plugin>
         <plugin id="godocker" type="runner" load="galaxy.jobs.runners.godocker:GodockerJobRunner">
             <!-- Go-Docker is a batch computing/cluster management tool using Docker
diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py
index b6027437044..5cdbb25b6cb 100644
--- a/lib/galaxy/jobs/runners/kubernetes.py
+++ b/lib/galaxy/jobs/runners/kubernetes.py
@@ -52,6 +52,10 @@ class KubernetesJobRunner(AsynchronousJobRunner):
             k8s_supplemental_group_id=dict(map=str),
             k8s_pull_policy=dict(map=str, default="Default"),
             k8s_fs_group_id=dict(map=int),
+            k8s_default_requests_cpu=dict(map=str, default=None),
+            k8s_default_requests_memory=dict(map=str, default=None),
+            k8s_default_limits_cpu=dict(map=str, default=None),
+            k8s_default_limits_memory=dict(map=str, default=None),
             k8s_pod_retrials=dict(map=int, valid=lambda x: int > 0, default=3))
 
         if 'runner_param_specs' not in kwargs:
@@ -220,6 +224,10 @@ class KubernetesJobRunner(AsynchronousJobRunner):
             }]
         }
 
+        resources = self.__get_resources(job_wrapper)
+        if resources:
+            k8s_container['resources'] = resources
+
         if self._default_pull_policy:
             k8s_container["imagePullPolicy"] = self._default_pull_policy
         # if self.__requires_ports(job_wrapper):
@@ -232,6 +240,100 @@ class KubernetesJobRunner(AsynchronousJobRunner):
     #    for k,v self.runner_params:
     #        if k.startswith("container_port_"):
 
+    def __get_resources(self, job_wrapper):
+        mem_request = self.__get_memory_request(job_wrapper)
+        cpu_request = self.__get_cpu_request(job_wrapper)
+
+        mem_limit = self.__get_memory_limit(job_wrapper)
+        cpu_limit = self.__get_cpu_limit(job_wrapper)
+
+        requests = {}
+        limits = {}
+
+        if mem_request:
+            requests['memory'] = mem_request
+        if cpu_request:
+            requests['cpu'] = cpu_request
+
+        if mem_limit:
+            limits['memory'] = mem_limit
+        if cpu_limit:
+            limits['cpu'] = cpu_limit
+
+        resources = {}
+        if requests:
+            resources['requests'] = requests
+        if limits:
+            resources['limits'] = limits
+
+        return resources
+
+
+    def __get_memory_request(self, job_wrapper):
+        """Obtains memory requests for job, checking if available on the destination, otherwise using the default"""
+        job_destinantion = job_wrapper.job_destination
+
+        if 'requests_memory' in job_destinantion.params:
+            return self.__transform_memory_value(job_destinantion.params['requests_memory'])
+        return self.runner_params['k8s_default_requests_memory']
+
+    def __get_memory_limit(self, job_wrapper):
+        """Obtains memory limits for job, checking if available on the destination, otherwise using the default"""
+        job_destinantion = job_wrapper.job_destination
+
+        if 'limits_memory' in job_destinantion.params:
+            return self.__transform_memory_value(job_destinantion.params['limits_memory'])
+        return self.runner_params['k8s_default_limits_memory']
+
+    def __get_cpu_request(self, job_wrapper):
+        """Obtains cpu requests for job, checking if available on the destination, otherwise using the default"""
+        job_destinantion = job_wrapper.job_destination
+
+        if 'requests_cpu' in job_destinantion.params:
+            return self.__transform_cpu_value(job_destinantion.params['requests_cpu'])
+        return self.runner_params['k8s_default_requests_cpu']
+
+    def __get_cpu_limit(self, job_wrapper):
+        """Obtains cpu requests for job, checking if available on the destination, otherwise using the default"""
+        job_destinantion = job_wrapper.job_destination
+
+        if 'limits_cpu' in job_destinantion.params:
+            return self.__transform_cpu_value(job_destinantion.params['limits_cpu'])
+        return self.runner_params['k8s_default_limits_cpu']
+
+    def __transform_cpu_value(self, cpu_value):
+        """Transforms cpu value
+
+           If the value is 0 and not a string, then None is returned.
+           If the value is a float, then it is multiplied by 1000 and expressed as mili cpus.
+           If the value is an integer, then it is and expressed as CPUs (no unit).
+           If it is an already formatted string, it is returned as it was.
+        """
+        if not isinstance(cpu_value, str) and float(cpu_value) == 0:
+            return None
+        if isinstance(cpu_value, float):
+            return str(int(cpu_value*1000))+"m"
+        elif isinstance(cpu_value, int):
+            return str(cpu_value)
+        return cpu_value
+
+    def __transform_memory_value(self, mem_value):
+        """Transforms memory value
+
+           If the value is 0 and not a string, then None is returned.
+           If the value has a decimal part, then it is multiplied by 1000 and expressed as Megabytes.
+           If the value is an integer, then it is truncated and expressed as Gigabytes.
+           If it is an already formatted string, it is returned as it was.
+        """
+        if not isinstance(mem_value, str) and float(mem_value) == 0:
+            return None
+        if isinstance(mem_value, float):
+            return str(int(mem_value*1000))+"M"
+        elif isinstance(mem_value, int):
+            return str(mem_value)+"G"
+        return mem_value
+
+
     def __assemble_k8s_container_image_name(self, job_wrapper):
         """Assembles the container image name as repo/owner/image:tag, where repo, owner and tag are optional"""
         job_destination = job_wrapper.job_destination

From 3145509b7a870d3ff06a2ad6037d5c5ce20eb559 Mon Sep 17 00:00:00 2001
From: Pablo Moreno <pablo.a.moreno@gmail.com>
Date: Mon, 2 Oct 2017 14:27:37 +0100
Subject: [PATCH 2/5] Avoids cleaning up container name when destination id is
 not str, probably applicable for dynamic destinations.

---
 lib/galaxy/jobs/runners/kubernetes.py | 10 ++++++----
 1 file changed, 6 insertions(+), 4 deletions(-)

diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py
index 5cdbb25b6cb..8de3f9e252b 100644
--- a/lib/galaxy/jobs/runners/kubernetes.py
+++ b/lib/galaxy/jobs/runners/kubernetes.py
@@ -357,10 +357,12 @@ class KubernetesJobRunner(AsynchronousJobRunner):
     def __get_k8s_container_name(self, job_wrapper):
         # These must follow a specific regex for Kubernetes.
         raw_id = job_wrapper.job_destination.id
-        cleaned_id = re.sub("[^-a-z0-9]", "-", raw_id)
-        if cleaned_id.startswith("-") or cleaned_id.endswith("-"):
-            cleaned_id = "x%sx" % cleaned_id
-        return cleaned_id
+        if isinstance(raw_id, str):
+            cleaned_id = re.sub("[^-a-z0-9]", "-", raw_id)
+            if cleaned_id.startswith("-") or cleaned_id.endswith("-"):
+                cleaned_id = "x%sx" % cleaned_id
+            return cleaned_id
+        return "job-container"
 
     def check_watched_item(self, job_state):
         """Checks the state of a job already submitted on k8s. Job state is a AsynchronousJobState"""

From ca2e121fa67443797436189387e7ff6733cfc6b9 Mon Sep 17 00:00:00 2001
From: Pablo Moreno <pablo.a.moreno@gmail.com>
Date: Wed, 4 Oct 2017 21:13:49 +0100
Subject: [PATCH 3/5] Minor code formatting for tox tests

---
 lib/galaxy/jobs/runners/kubernetes.py | 8 +++-----
 1 file changed, 3 insertions(+), 5 deletions(-)

diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py
index 8de3f9e252b..421a1f5df2c 100644
--- a/lib/galaxy/jobs/runners/kubernetes.py
+++ b/lib/galaxy/jobs/runners/kubernetes.py
@@ -268,7 +268,6 @@ class KubernetesJobRunner(AsynchronousJobRunner):
 
         return resources
 
-
     def __get_memory_request(self, job_wrapper):
         """Obtains memory requests for job, checking if available on the destination, otherwise using the default"""
         job_destinantion = job_wrapper.job_destination
@@ -312,7 +311,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
         if not isinstance(cpu_value, str) and float(cpu_value) == 0:
             return None
         if isinstance(cpu_value, float):
-            return str(int(cpu_value*1000))+"m"
+            return str(int(cpu_value * 1000)) + "m"
         elif isinstance(cpu_value, int):
             return str(cpu_value)
         return cpu_value
@@ -328,12 +327,11 @@ class KubernetesJobRunner(AsynchronousJobRunner):
         if not isinstance(mem_value, str) and float(mem_value) == 0:
             return None
         if isinstance(mem_value, float):
-            return str(int(mem_value*1000))+"M"
+            return str(int(mem_value * 1000)) + "M"
         elif isinstance(mem_value, int):
-            return str(mem_value)+"G"
+            return str(mem_value) + "G"
         return mem_value
 
-
     def __assemble_k8s_container_image_name(self, job_wrapper):
         """Assembles the container image name as repo/owner/image:tag, where repo, owner and tag are optional"""
         job_destination = job_wrapper.job_destination

From 7654935f031afdd852df3a800f2f012a89bdb08f Mon Sep 17 00:00:00 2001
From: Pablo Moreno <pablo.a.moreno@gmail.com>
Date: Mon, 16 Oct 2017 14:26:14 +0100
Subject: [PATCH 4/5] k8s runner: Adds ability to detect OOMKilled jobs and
 signals it for resubmission.

---
 lib/galaxy/jobs/runners/kubernetes.py | 47 +++++++++++++++++++++------
 1 file changed, 37 insertions(+), 10 deletions(-)

diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py
index 421a1f5df2c..d8fe7221a39 100644
--- a/lib/galaxy/jobs/runners/kubernetes.py
+++ b/lib/galaxy/jobs/runners/kubernetes.py
@@ -11,7 +11,8 @@ from six import text_type
 from galaxy import model
 from galaxy.jobs.runners import (
     AsynchronousJobRunner,
-    AsynchronousJobState
+    AsynchronousJobState,
+    JobState
 )
 
 # pykube imports:
@@ -394,19 +395,13 @@ 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_retrials:
                 job_state.running = True
                 return job_state
             elif failed > max_pod_retrials:
-                self.__produce_log_file(job_state)
-                error_file = open(job_state.error_file, 'w')
-                error_file.write("Exceeded max number of Kubernetes pod retrials allowed for job\n")
-                error_file.close()
-                job_state.running = False
-                job_state.fail_message = "More pods failed than allowed. See stdout for pods details."
-                self.mark_as_failed(job_state)
-                job.scale(replicas=0)
-                return None
+                return self._handle_job_failure(job, job_state)
             # We should not get here
             log.debug(
                 "Reaching unexpected point for Kubernetes job, where it is not classified as succ., active nor failed.")
@@ -430,6 +425,38 @@ class KubernetesJobRunner(AsynchronousJobRunner):
             self.mark_as_failed(job_state)
             return job_state
 
+    def _handle_job_failure(self, job, job_state, reason=None):
+        self.__produce_log_file(job_state)
+        error_file = open(job_state.error_file, 'w')
+        if reason == "OOM":
+            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
+        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."
+        error_file.close()
+        job_state.running = False
+        self.mark_as_failed(job_state)
+        job.scale(replicas=0)
+        return None
+
+    def __job_failed_due_to_low_memory(self, job_state):
+        """
+        checks the state of the pod to see if it was killed
+        for being out of memory (pod status OOMKilled). If that is the case
+        marks the job for resubmission (resubmit logic is part of destinations).
+        """
+
+        pods = Pod.objects(self._pykube_api).filter(selector="app=" + job_state.job_id)
+        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
+
+        return False
+
     def fail_job(self, job_state):
         """
         Kubernetes runner overrides fail_job (called by mark_as_failed) to rescue the pod's log files which are left as

From ac477fc0216c9c22af659eb9f1de7622c9207779 Mon Sep 17 00:00:00 2001
From: John Chilton <jmchilton@gmail.com>
Date: Tue, 12 Dec 2017 11:08:38 -0500
Subject: [PATCH 5/5] Rework k8s resource request parameters.

xref https://github.com/galaxyproject/galaxy/pull/4751#issuecomment-341196780
---
 config/job_conf.xml.sample_advanced   | 22 +++++++---------------
 lib/galaxy/jobs/runners/kubernetes.py |  8 ++++----
 2 files changed, 11 insertions(+), 19 deletions(-)

diff --git a/config/job_conf.xml.sample_advanced b/config/job_conf.xml.sample_advanced
index c4a4b04f7c3..89ae94aba71 100644
--- a/config/job_conf.xml.sample_advanced
+++ b/config/job_conf.xml.sample_advanced
@@ -248,14 +248,14 @@
                  a requirement of the other.
 
                  -->
-            <!-- <param id="k8s_default_requests_cpu">"500m"</param> -->
-            <!-- <param id="k8s_default_requests_memory">"500Mi"</param> -->
-            <!-- <param id="k8s_default_limits_cpu">"2"</param> -->
-            <!-- <param id="k8s_default_limits_memory">"2Gi"</param> -->
+            <!-- <param id="requests_cpu">500m</param> -->
+            <!-- <param id="requests_memory">500Mi</param> -->
+            <!-- <param id="limits_cpu">2</param> -->
+            <!-- <param id="limits_memory">2Gi</param> -->
             <!-- Kubernetes resource Requests and Limits
-                 Parameters above (k8s_default_requests_* and k8s_default_limits_*) set default minimal (requests) and
-                 maximal (limits) CPU and memory resources to be allocated to all containers in Kubernetes jobs by
-                 default.
+                 Parameters above (requests_* and limits_*) set default minimal (requests) and
+                 maximal (limits) CPU and memory resources to be allocated to all containers in
+                 Kubernetes jobs by default.
 
                  Limits and requests for CPU resources are measured in cpu units. Fractional requests are allowed. A
                  Container with requests_cpu of 0.5 is guaranteed half as much CPU as one that asks for 1 CPU. The
@@ -271,14 +271,6 @@
 
                  For more details on the Kubernetes resources management, see:
                  https://kubernetes.io/docs/concepts/configuration/manage-compute-resources-container/
-
-                 It is the responsability of the tool developer or administrator to override these limits, either at the
-                 destination or the tool itself. If not set, current defaults set on the code are:
-
-                 requests_cpu     = "500m"
-                 limits_cpu       = "1"
-                 requests_memory  = "500Mi"
-                 limits_memory    = "1Gi"
                  -->
 
         </plugin>
diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py
index d8fe7221a39..72a0c9560e7 100644
--- a/lib/galaxy/jobs/runners/kubernetes.py
+++ b/lib/galaxy/jobs/runners/kubernetes.py
@@ -275,7 +275,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
 
         if 'requests_memory' in job_destinantion.params:
             return self.__transform_memory_value(job_destinantion.params['requests_memory'])
-        return self.runner_params['k8s_default_requests_memory']
+        return None
 
     def __get_memory_limit(self, job_wrapper):
         """Obtains memory limits for job, checking if available on the destination, otherwise using the default"""
@@ -283,7 +283,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
 
         if 'limits_memory' in job_destinantion.params:
             return self.__transform_memory_value(job_destinantion.params['limits_memory'])
-        return self.runner_params['k8s_default_limits_memory']
+        return None
 
     def __get_cpu_request(self, job_wrapper):
         """Obtains cpu requests for job, checking if available on the destination, otherwise using the default"""
@@ -291,7 +291,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
 
         if 'requests_cpu' in job_destinantion.params:
             return self.__transform_cpu_value(job_destinantion.params['requests_cpu'])
-        return self.runner_params['k8s_default_requests_cpu']
+        return None
 
     def __get_cpu_limit(self, job_wrapper):
         """Obtains cpu requests for job, checking if available on the destination, otherwise using the default"""
@@ -299,7 +299,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
 
         if 'limits_cpu' in job_destinantion.params:
             return self.__transform_cpu_value(job_destinantion.params['limits_cpu'])
-        return self.runner_params['k8s_default_limits_cpu']
+        return None
 
     def __transform_cpu_value(self, cpu_value):
         """Transforms cpu value