Repository: twosigma/Cook (language: python)

Issue:
Initial implementation for submitting jobs directly to Kubernetes Scheduler
## Kubernetes Bin Packing
Configure pools to submit real job pods directly to Kubernetes, which handles the scheduling and cluster autoscaling. Pools that use the Kubernetes Scheduler will not use the "match-offer" loop, Fenzo bin packing algorithm, nor synthetic pods.

This is a proof of concept that will require a few iterations to get to a production state. However, my internal tests shows this feature behaves well. The following is my current understanding of the gaps between this changeset and a version we can test with real jobs. I'm sure there are more, please let me know and I can document it here.

### Configuration
#2139 introduced configuring schedulers per pool. Here is an example of running the Kubernetes Scheduler feature in one pool and Fenzo in all others.

```
:pools {
  ...
  :schedulers [{:pool-regex "^k8s-exp$"
                        :scheduler-config {:scheduler "kubernetes"
                                           :max-jobs-considered 1000}}
                       {:pool-regex ".*"
                        :scheduler-config {:scheduler "fenzo"
                                           :good-enough-fitness 1.0
                                           :fenzo-fitness-calculator "com.netflix.fenzo.plugins.BinPackingFitnessCalculators/cpuMemBinPacker"
                                           :fenzo-max-jobs-considered 400
                                           :fenzo-scaleback 1
                                           :fenzo-floor-iterations-before-warn 10
                                           :fenzo-floor-iterations-before-reset 1000}}]
}
```

### Remaining Work
* Real job pods now trigger autoscaling. In both Fenzo and Kubernetes Scheduler pools, the instance runtime begins when the pod is submitted to Kubernetes. This is usually fine in the Fenzo case, because a pod is only launched when there is available space for it on a node. In this case, runtime is pretty close to when the pod is actually in the Running state (although it does count initialization). However, for the Kubernetes Scheduler pools, a pod could trigger autoscaling, leading to 5 to 10 minutes of inaccurate runtime. We should consider starting instance runtime in both cases when the pod is actually running on the node.
* A backpressure mechanism remains, in order to moderate the amount of pending pods in each GKE cluster.
* Proper metrics accounting in the new Kubernetes handler and code. This functionality has been a bit of a moving target because of Sam's great work on it over the past quarter. I envision there will be a round of review and edits for this specifically at a later date.
* Audit values of the instance saved to Datomic. For example, do we need to set `ports`, `slave-id`, etc?


Discussion / hints:

Fix patch:
diff --git a/scheduler/src/cook/kubernetes/controller.clj b/scheduler/src/cook/kubernetes/controller.clj
index faca25a828..8a7a43fc64 100644
--- a/scheduler/src/cook/kubernetes/controller.clj
+++ b/scheduler/src/cook/kubernetes/controller.clj
@@ -249,11 +249,23 @@
     (swap! map-atom dissoc key)
     (swap! map-atom assoc key value)))
 
+(defn- get-job-container-status
+  "Extract the container status for the main cook job container (defined in api/cook-container-name-for-job).
+  We use this because we need the actual state object to determine the failure reason for failed jobs."
+  [^V1PodStatus pod-status]
+  (->> pod-status
+       .getContainerStatuses
+       (filter (fn [^V1ContainerStatus container-status] (= api/cook-container-name-for-job
+                                                            (.getName container-status))))
+       first))
+
 (defn container-status->failure-reason
   "Maps kubernetes failure reasons to cook failure reasons"
-  [{:keys [name]} ^V1PodStatus pod-status ^V1Pod pod ^V1ContainerStatus container-status]
+  [{:keys [name]} ^V1Pod pod]
   ; TODO map additional kubernetes failure reasons
   (let [pod-name (api/V1Pod->name pod)
+        pod-status (.getStatus pod)
+        ^V1ContainerStatus container-status (get-job-container-status pod-status)
         container-terminated-reason (some-> container-status .getState .getTerminated .getReason)
         pod-status-reason (.getReason pod-status)
         pod-status-message (.getMessage pod-status)]
@@ -269,10 +281,10 @@
         :reason-slave-removed)
 
       (and pod-status-message (or
-                                ; this message is given when pod exceeds pod ephemeral-storage limit
-                                (str/includes? pod-status-message "ephemeral local storage usage exceeds")
-                                ; this message is given when pod exceeds ephemeral-storage request and available disk space on node is low
-                                (str/includes? pod-status-message "low on resource: ephemeral-storage")))
+                               ; this message is given when pod exceeds pod ephemeral-storage limit
+                               (str/includes? pod-status-message "ephemeral local storage usage exceeds")
+                               ; this message is given when pod exceeds ephemeral-storage request and available disk space on node is low
+                               (str/includes? pod-status-message "low on resource: ephemeral-storage")))
       :reason-container-limitation-disk
 
       ; If there is no container status and the pod status reason is Outofcpu,
@@ -306,16 +318,6 @@
                   {:pod-status pod-status :container-status container-status})
         :unknown))))
 
-(defn- get-job-container-status
-  "Extract the container status for the main cook job container (defined in api/cook-container-name-for-job).
-  We use this because we need the actual state object to determine the failure reason for failed jobs."
-  [^V1PodStatus pod-status]
-  (->> pod-status
-       .getContainerStatuses
-       (filter (fn [^V1ContainerStatus container-status] (= api/cook-container-name-for-job
-                                                            (.getName container-status))))
-       first))
-
 (defn make-failed-task-result
   "Make a pod status result corresponding to a failed pod"
   [instance-id reason]
@@ -342,8 +344,7 @@
                            :task-failed)
               reason (or reason
                          (when succeeded? :reason-normal-exit)
-                         (container-status->failure-reason compute-cluster pod-status
-                                                           pod job-container-status))
+                         (container-status->failure-reason compute-cluster pod))
               status {:task-id {:value pod-name}
                       :state task-state
                       :reason reason}


Test patch:
diff --git a/scheduler/test/cook/test/kubernetes/controller.clj b/scheduler/test/cook/test/kubernetes/controller.clj
index 61f29c3725..d125eef1c8 100644
--- a/scheduler/test/cook/test/kubernetes/controller.clj
+++ b/scheduler/test/cook/test/kubernetes/controller.clj
@@ -24,12 +24,11 @@
 (deftest test-container-status->failure-reason
   (testing "no container status + out-of-cpu pod status"
     (let [pod-status (V1PodStatus.)
-          container-status nil
           pod (tu/pod-helper "12345" "hostA" {:cpus 1.0 :mem 100.0})]
       (.setReason pod-status "OutOfcpu")
+      (.setStatus pod pod-status)
       (is (= :reason-invalid-offers
-             (controller/container-status->failure-reason {:name "test-cluster"} pod-status
-                                                          pod container-status))))))
+             (controller/container-status->failure-reason {:name "test-cluster"} pod))))))
 (deftest test-process
   (tu/setup)
   (with-redefs [cached-queries/instance-uuid->job-uuid-datomic-query (constantly (java.util.UUID/randomUUID))
@@ -294,7 +293,7 @@
             reason (atom nil)]
         (.setStatus pod (V1PodStatus.))
         (with-redefs [controller/write-status-to-datomic (fn [_ status] (reset! reason (:reason status)))
-                      controller/container-status->failure-reason (fn [_ _ _ _ _]
+                      controller/container-status->failure-reason (fn [_ _]
                                                                     (throw (ex-info "Shouldn't get called" {})))]
           (controller/handle-pod-completed nil "podA" {:pod pod :synthesized-state {:state :pod/failed}}
                                            :reason :reason-task-invalid))
@@ -306,7 +305,7 @@
         (.setStatus pod (V1PodStatus.))
         (with-redefs [cc/compute-cluster-name (constantly "compute-cluster-name")
                       controller/write-status-to-datomic (fn [_ status] (reset! reason (:reason status)))
-                      controller/container-status->failure-reason (fn [_ _ _ _ _]
+                      controller/container-status->failure-reason (fn [_ _]
                                                                     (throw (ex-info "Got Exception" {})))]
           (controller/handle-pod-completed nil "podA" {:pod pod :synthesized-state {:state :pod/failed}})
           (is (= :reason-task-unknown @reason)))))