Repository: twosigma/Cook (language: python)

Issue:
Make Fenzo config pool specific and setup for other schedulers
## Changes proposed in this PR

- Add pool-specific scheduler config. This allows you to configure Fenzo settings per pool, instead of globally. Mainly, this change was introduced to allow us to configure a different scheduler from Fenzo (e.g. Kubernetes) for pools in the future.
- The new pool-specific scheduler config is routed into the new pool handler factory, although it currently only supports the existing Fenzo scheduler.
- Modify main function and unit tests to handle the new config.

## Why are we making these changes?





Discussion / hints:

Fix patch:
diff --git a/scheduler/.gitignore b/scheduler/.gitignore
index eb94687e47..62d947d223 100644
--- a/scheduler/.gitignore
+++ b/scheduler/.gitignore
@@ -1,2 +1,3 @@
 .pytest_cache
 gclog*
+.calva/
diff --git a/scheduler/src/cook/config.clj b/scheduler/src/cook/config.clj
index d14c21071a..8a5b926059 100644
--- a/scheduler/src/cook/config.clj
+++ b/scheduler/src/cook/config.clj
@@ -106,6 +106,7 @@
 
 (def default-authorization {:authorization-fn 'cook.rest.authorization/open-auth})
 (def default-fitness-calculator "com.netflix.fenzo.plugins.BinPackingFitnessCalculators/cpuMemBinPacker")
+
 (def default-fenzo-scheduler-config {:scheduler "fenzo"
                                      :good-enough-fitness 0.8
                                      :fenzo-fitness-calculator default-fitness-calculator
@@ -113,6 +114,10 @@
                                      :fenzo-scaleback 0.95
                                      :fenzo-floor-iterations-before-warn 10
                                      :fenzo-floor-iterations-before-reset 1000})
+(def default-kubernetes-scheduler-config {:scheduler "kubernetes"
+                                          :max-jobs-considered 1000})
+(def default-schedulers-config [{:pool-regex ".*"
+                                 :scheduler-config default-fenzo-scheduler-config}])
 
 (defrecord UserRateLimit [id quota auth-bypass-quota ttl]
   RateLimit
@@ -205,6 +210,22 @@
       (when-not (contains? valid-types default-type)
         (throw (ex-info (str "Default disk type for pool-regex " pool-regex " is not listed as a valid disk type") entry))))))
 
+(defn guard-invalid-schedulers-config
+  "When scheduler config for pools is provided, ensure each is associated with a pool and is using a valid scheduler."
+  [schedulers-config]
+  (when schedulers-config
+    (doseq [{pool-regex :pool-regex 
+             {scheduler :scheduler :as scheduler-config} :scheduler-config 
+             :as entry} schedulers-config] 
+      (when-not pool-regex
+        (throw (ex-info (str "pool-regex key is missing from config") entry)))
+      (when-not scheduler-config
+        (throw (ex-info (str "scheduler-config key is missing from config") entry)))
+      (when-not scheduler
+        (throw (ex-info (str "scheduler key is missing from scheduler-config") entry)))
+      (when-not (or (= scheduler "fenzo") (= scheduler "kubernetes"))
+        (throw (ex-info (str "scheduler must be fenzo or kubernetes") entry))))))
+
 (def config-settings
   "Parses the settings out of a config file"
   (graph/eager-compile
@@ -448,24 +469,35 @@
                          (throw (ex-info "You enabled nrepl but didn't configure a port. Please configure a port in your config file." {})))
                        ((util/lazy-load-var 'clojure.tools.nrepl.server/start-server) :port port)))
      :pools (fnk [[:config {pools nil}]]
-              (guard-invalid-gpu-config (:valid-gpu-models pools))
-              (guard-invalid-disk-config (:disk pools))
-              (cond-> pools
-                (:job-resource-adjustment pools)
-                (update :job-resource-adjustment
-                        #(-> %
-                           (update :pool-regex re-pattern)))
-                (not (:default-containers pools))
-                (assoc :default-containers [])
-                (not (:default-env pools))
-                (assoc :default-env [])
-                (not (:quotas pools))
-                (assoc :quotas [])
-                (not (:schedulers pools))
-                (assoc :schedulers [{:pool-regex ".*"
-                                     :scheduler-config default-fenzo-scheduler-config}])))
+                 (guard-invalid-gpu-config (:valid-gpu-models pools))
+                 (guard-invalid-disk-config (:disk pools))
+                 (guard-invalid-schedulers-config (:schedulers pools))
+                 (cond-> pools
+                   (:job-resource-adjustment pools)
+                   (update :job-resource-adjustment
+                           #(-> %
+                                (update :pool-regex re-pattern)))
+                   (not (:default-containers pools))
+                   (assoc :default-containers [])
+                   (not (:default-env pools))
+                   (assoc :default-env [])
+                   (not (:quotas pools))
+                   (assoc :quotas [])
+
+                   ; If valid configs up-to "scheduler" are provided, merge defaults
+                   (seq (:schedulers pools))
+                   (assoc :schedulers
+                          (mapv (fn [{:keys [pool-regex] {:keys [scheduler] :as scheduler-config} :scheduler-config}]
+                                 {:pool-regex pool-regex
+                                  :scheduler-config (cond
+                                                      (= scheduler "fenzo") (merge default-fenzo-scheduler-config scheduler-config)
+                                                      (= scheduler "kubernetes") (merge default-kubernetes-scheduler-config scheduler-config)
+                                                      :else (throw (ex-info (str "Unexpected scheduler configured " scheduler) {})))}) 
+                               (:schedulers pools)))
+                   (empty? (:schedulers pools))
+                   (assoc :schedulers default-schedulers-config)))
      :rank (fnk [[:config {rank {:number-to-force 1000}}]]
-             rank)
+                rank)
      :api-only? (fnk [[:config {api-only? false}]]
                   api-only?)
      :cache-working-set-size (fnk [[:config {cache-working-set-size 1000000}]]
diff --git a/scheduler/src/cook/kubernetes/api.clj b/scheduler/src/cook/kubernetes/api.clj
index 6da525b75f..b63ec319b6 100644
--- a/scheduler/src/cook/kubernetes/api.clj
+++ b/scheduler/src/cook/kubernetes/api.clj
@@ -78,6 +78,11 @@
   [pod-name]
   (str/starts-with? pod-name cook-synthetic-pod-name-prefix))
 
+(defn kubernetes-scheduler-pod?
+  "Given a pod, returns true if its scheduling is handled by Kubernetes."
+  [^V1Pod pod]
+  (= "kubernetes" (some-> pod .getMetadata .getLabels (.get "twosigma.com/cook-scheduler"))))
+
 (defn pod-name->job-uuid
   "If a pod is synthetic, return the uuid of the job it was created for"
   [pod-name]
@@ -1816,26 +1821,28 @@
 (defn pod-unschedulable?
   "Returns true if the given pod status has a PodScheduled
   condition with status False and reason Unschedulable"
-  [pod-name ^V1PodStatus pod-status]
-  (let [{:keys [pod-condition-unschedulable-seconds
+  [^V1Pod pod]
+  (let [pod-status (.getStatus pod)
+        {:keys [pod-condition-unschedulable-seconds
                 synthetic-pod-condition-unschedulable-seconds]}
         (config/kubernetes)
         unschedulable-seconds
-        (if (some-> pod-name synthetic-pod?)
+        (if (or (some-> (V1Pod->name pod) synthetic-pod?)
+                (some-> pod kubernetes-scheduler-pod?))
           synthetic-pod-condition-unschedulable-seconds
           pod-condition-unschedulable-seconds)]
     (some->> pod-status
              .getConditions
              (some
-               (fn pod-condition-unschedulable?
-                 [^V1PodCondition condition]
-                 (and (-> condition .getType (= "PodScheduled"))
-                      (-> condition .getStatus (= "False"))
-                      (-> condition .getReason (= "Unschedulable"))
-                      (-> condition
-                          .getLastTransitionTime
-                          (t/plus (t/seconds unschedulable-seconds))
-                          (t/before? (t/now)))))))))
+              (fn pod-condition-unschedulable?
+                [^V1PodCondition condition]
+                (and (-> condition .getType (= "PodScheduled"))
+                     (-> condition .getStatus (= "False"))
+                     (-> condition .getReason (= "Unschedulable"))
+                     (-> condition
+                         .getLastTransitionTime
+                         (t/plus (t/seconds unschedulable-seconds))
+                         (t/before? (t/now)))))))))
 
 (defn pod-has-stuck-condition?
   "Returns true if the given pod status has a pod condition that is deemed stuck,
@@ -1996,7 +2003,7 @@
                 ; if the ToBeDeletedByClusterAutoscaler taint gets added between when we
                 ; saw available capacity on a node and when we submitted the pod to that
                 ; node, then the pod will never get scheduled.
-                (pod-unschedulable? pod-name pod-status)
+                (pod-unschedulable? pod)
                 (let [synthesized-state {:state :pod/failed
                                          :reason "Unschedulable"}]
                   (log-structured/info "Encountered unschedulable pod"
diff --git a/scheduler/src/cook/kubernetes/controller.clj b/scheduler/src/cook/kubernetes/controller.clj
index 0d75a59f3c..faca25a828 100644
--- a/scheduler/src/cook/kubernetes/controller.clj
+++ b/scheduler/src/cook/kubernetes/controller.clj
@@ -6,6 +6,7 @@
             [cook.datomic :as datomic]
             [cook.kubernetes.api :as api]
             [cook.kubernetes.metrics :as metrics]
+            [cook.log-structured :as log-structured]
             [cook.mesos.sandbox :as sandbox]
             [cook.passport :as passport]
             [cook.prometheus-metrics :as prom]
@@ -250,9 +251,10 @@
 
 (defn container-status->failure-reason
   "Maps kubernetes failure reasons to cook failure reasons"
-  [{:keys [name]} instance-id ^V1PodStatus pod-status ^V1ContainerStatus container-status]
+  [{:keys [name]} ^V1PodStatus pod-status ^V1Pod pod ^V1ContainerStatus container-status]
   ; TODO map additional kubernetes failure reasons
-  (let [container-terminated-reason (some-> container-status .getState .getTerminated .getReason)
+  (let [pod-name (api/V1Pod->name pod)
+        container-terminated-reason (some-> container-status .getState .getTerminated .getReason)
         pod-status-reason (.getReason pod-status)
         pod-status-message (.getMessage pod-status)]
     (cond
@@ -263,7 +265,7 @@
       ; Check the node pod for NodeAffinity.
       (= "NodeAffinity" pod-status-reason)
       (do
-        (log/info "In compute cluster" name ", encountered NodeAffinity pod status reason for " instance-id)
+        (log/info "In compute cluster" name ", encountered NodeAffinity pod status reason for " pod-name)
         :reason-slave-removed)
 
       (and pod-status-message (or
@@ -277,27 +279,30 @@
       ; then the node didn't have enough CPUs to start the container
       (and (nil? container-status) (= "OutOfcpu" pod-status-reason))
       (do
-        (log/info "In compute cluster" name ", encountered OutOfcpu pod status reason for" instance-id)
+        (log/info "In compute cluster" name ", encountered OutOfcpu pod status reason for" pod-name)
         :reason-invalid-offers)
 
-      (api/pod-unschedulable? instance-id pod-status)
+      ; NOTE: the timeout used for determining whether a pod is unschedulable
+      ; depends on if it was submitted via Fenzo or Kubernetes Schedulers.
+      ; TODO: simplify this logic once we are no longer using both schedulers.
+      (api/pod-unschedulable? pod)
       (do
-        (log/info "In compute cluster" name ", encountered unschedulable pod" instance-id)
+        (log/info "In compute cluster" name ", encountered unschedulable pod" pod-name)
         :reason-scheduling-failed-on-host)
 
-      (api/pod-containers-not-initialized? instance-id pod-status)
+      (api/pod-containers-not-initialized? pod-name pod-status)
       (do
-        (log/info "In compute cluster" name ", encountered containers not initialized in pod" instance-id)
+        (log/info "In compute cluster" name ", encountered containers not initialized in pod" pod-name)
         :reason-container-initialization-timed-out)
 
-      (api/pod-containers-not-ready? instance-id pod-status)
+      (api/pod-containers-not-ready? pod-name pod-status)
       (do
-        (log/info "In compute cluster" name ", encountered containers not ready in pod" instance-id)
+        (log/info "In compute cluster" name ", encountered containers not ready in pod" pod-name)
         :reason-container-readiness-timed-out)
 
       :default
       (do
-        (log/warn "In compute cluster" name ", unable to determine failure reason for" instance-id
+        (log/warn "In compute cluster" name ", unable to determine failure reason for" pod-name
                   {:pod-status pod-status :container-status container-status})
         :unknown))))
 
@@ -337,8 +342,8 @@
                            :task-failed)
               reason (or reason
                          (when succeeded? :reason-normal-exit)
-                         (container-status->failure-reason compute-cluster pod-name
-                                                           pod-status job-container-status))
+                         (container-status->failure-reason compute-cluster pod-status
+                                                           pod job-container-status))
               status {:task-id {:value pod-name}
                       :state task-state
                       :reason reason}
@@ -420,6 +425,18 @@
                           nil))]
       (when sandbox-url (scheduler/write-sandbox-url-to-datomic datomic/conn task-id sandbox-url)))))
 
+(defn record-hostname
+  "Record the hostname in Datomic for the instance. This is necessary for pods
+   submitted directly to Kubernetes (i.e. when we let Kubernetes do bin packing
+   instead of Fenzo). In this case, the instance has no hostname and needs to be 
+   updated once the pod is scheduled on a node by Kubernetes.
+   
+   If the hostname is null, we do not write to Datomic."
+  [^V1Pod pod]
+  (let [pod-name (-> pod .getMetadata .getName)
+        hostname (api/pod->node-name pod)]
+    (when hostname (scheduler/write-hostname-to-datomic datomic/conn pod-name hostname))))
+
 (defn handle-pod-killed
   "A pod was killed. So now we need to update the status in datomic and store the exit code."
   [compute-cluster pod-name]
@@ -711,6 +728,11 @@
            old-file-server-state (:sandbox-file-server-container-state old-state)]
        (when (and (= new-file-server-state :running) (not= old-file-server-state :running))
          (record-sandbox-url pod-name new-state)))
+     ; Hostname will change for pods scheduled by Kubernetes.
+     (when (api/kubernetes-scheduler-pod? pod)
+       (when-not (= (api/pod->node-name (:pod new-state))
+                    (api/pod->node-name (:pod old-state)))
+         (record-hostname pod)))
      (when (or force-process?
                (not (k8s-actual-state-equivalent? old-state new-state)))
        (when-not force-process?
diff --git a/scheduler/src/cook/mesos.clj b/scheduler/src/cook/mesos.clj
index e507f2db5a..85c959ba75 100644
--- a/scheduler/src/cook/mesos.clj
+++ b/scheduler/src/cook/mesos.clj
@@ -33,7 +33,6 @@
             [datomic.api :as d :refer [q]]
             [mesomatic.scheduler]
             [mesomatic.types]
-            [metatransaction.core :refer [db]]
             [metatransaction.utils :as dutils]
             [metrics.counters :as counters]
             [plumbing.core :refer [map-from-keys]]
diff --git a/scheduler/src/cook/prometheus_metrics.clj b/scheduler/src/cook/prometheus_metrics.clj
index b09a7f3313..ecd075baf7 100644
--- a/scheduler/src/cook/prometheus_metrics.clj
+++ b/scheduler/src/cook/prometheus_metrics.clj
@@ -31,7 +31,7 @@
 (def scheduler-match-cycle-duration :cook/scheduler-match-cycle-duration-seconds)
 (def scheduler-generate-user-usage-map-duration :cook/scheduler-generate-user-usage-map-duration-seconds)
 (def scheduler-handle-resource-offers-total-duration :cook/scheduler-handle-resource-offers-total-duration-seconds)
-(def scheduler-handle-resource-offers-pending-to-considerable-duration :cook/scheduler-handle-resource-offers-pending-to-considerable-duration-seconds)
+(def scheduler-pool-handler-pending-to-considerable-duration :cook/scheduler-pool-handler-pending-to-considerable-duration)
 (def scheduler-fenzo-schedule-once-duration :cook/scheduler-fenzo-schedule-once-duration-seconds)
 (def scheduler-handle-resource-offers-match-duration :cook/scheduler-handle-resource-offers-match-duration-seconds)
 (def scheduler-handle-resource-offers-matches-to-job-uuids-duration :cook/scheduler-handle-resource-offers-matches-to-job-uuids-duration-seconds)
@@ -39,6 +39,7 @@
 (def scheduler-launch-all-matched-tasks-transact-duration :cook/scheduler-launch-all-matched-tasks-transact-duration-seconds)
 (def scheduler-launch-all-matched-tasks-submit-duration :cook/scheduler-launch-all-matched-tasks-submit-duration-seconds)
 (def scheduler-trigger-autoscaling-duration :cook/scheduler-trigger-autoscaling-duration-seconds)
+(def scheduler-distribute-jobs-to-kubernetes-duration :cook/scheduler-distribute-jobs-to-kubernetes-duration-seconds)
 (def scheduler-kill-cancelled-tasks-duration :cook/scheduler-kill-cancelled-tasks-duration-seconds)
 (def scheduler-sort-jobs-hierarchy-duration :cook/scheduler-sort-jobs-hierarchy-duration-seconds)
 (def scheduler-filter-offensive-jobs-duration :cook/scheduler-filter-offensive-jobs-duration-seconds)
@@ -139,7 +140,7 @@
                           {:description "Distribution of total handle-resource-offers! duration"
                            :labels [:pool]
                            :quantiles default-summary-quantiles})
-      (prometheus/summary scheduler-handle-resource-offers-pending-to-considerable-duration
+      (prometheus/summary scheduler-pool-handler-pending-to-considerable-duration
                           {:description "Distribution of filtering pending to considerable jobs duration"
                            :labels [:pool]
                            :quantiles default-summary-quantiles})
@@ -171,6 +172,10 @@
                           {:description "Distribution of trigger autoscaling latency"
                            :labels [:pool]
                            :quantiles default-summary-quantiles})
+      (prometheus/summary scheduler-distribute-jobs-to-kubernetes-duration
+                          {:description "Distribution of distributing jobs to Kubernetes latency"
+                           :labels [:pool]
+                           :quantiles default-summary-quantiles})
       (prometheus/summary scheduler-kill-cancelled-tasks-duration
                           {:description "Distribution of kill cancelled tasks latency"
                            :labels [:pool]
diff --git a/scheduler/src/cook/rest/api.clj b/scheduler/src/cook/rest/api.clj
index 9dac750c3e..5775d9ec89 100644
--- a/scheduler/src/cook/rest/api.clj
+++ b/scheduler/src/cook/rest/api.clj
@@ -223,13 +223,13 @@
   {:status s/Str
    :task_id s/Uuid
    :executor_id s/Uuid
-   :agent_id s/Str
-   :slave_id s/Str
-   :hostname s/Str
    :preempted s/Bool
    :backfilled s/Bool
    :ports [s/Int]
    :compute-cluster ComputeCluster
+   (s/optional-key :agent_id) s/Str
+   (s/optional-key :slave_id) s/Str
+   (s/optional-key :hostname) s/Str
    (s/optional-key :start_time) s/Int
    (s/optional-key :mesos_start_time) s/Int
    (s/optional-key :end_time) s/Int
@@ -1173,6 +1173,7 @@
     (timers/time!
       (timers/timer ["cook-mesos" "internal" "fetch-instance-map"])
       (let [hostname (:instance/hostname instance)
+            slave-id (:instance/slave-id instance)
             task-id (:instance/task-id instance)
             executor (:instance/executor instance)
             sandbox-directory (:instance/sandbox-directory instance)
@@ -1190,11 +1191,18 @@
         (cond-> {:backfilled false ;; Backfill has been deprecated
                  :compute-cluster (fetch-compute-cluster-map db (:instance/compute-cluster instance))
                  :executor_id (:instance/executor-id instance)
-                 :hostname hostname
+                 ; NOTE: the CLI requires hostname, slave-id, and agent-id to be non-nil. An instance's
+                 ; pod which is scheduled by Kubernetes instead of Fenzo, does not initially have a
+                 ; hostname, for example. We avoid setting fake data in this case on the instance in
+                 ; Datomic and choose to return it at the API layer.
+                 ;
+                 ; TODO: stop this workaround once the CLI supports these as optional.
+                 :hostname (or hostname "Unknown")
+                 :slave_id (or slave-id "Unknown")
+                 :agent_id (or slave-id "Unknown")
+
                  :ports (vec (sort (:instance/ports instance)))
                  :preempted (:instance/preempted? instance false)
-                 :agent_id (:instance/slave-id instance)
-                 :slave_id (:instance/slave-id instance)
                  :status (name (:instance/status instance))
                  :task_id task-id}
           executor (assoc :executor (name executor))
diff --git a/scheduler/src/cook/scheduler/scheduler.clj b/scheduler/src/cook/scheduler/scheduler.clj
index 930bdae7e7..4c09d8a0a4 100644
--- a/scheduler/src/cook/scheduler/scheduler.clj
+++ b/scheduler/src/cook/scheduler/scheduler.clj
@@ -328,6 +328,27 @@
     (catch Exception e
       (log/error e "Sandbox file server URL update error"))))
 
+(defn write-hostname-to-datomic
+  "Takes a hostname for an instance and saves it to datomic."
+  [conn task-id hostname] 
+  (log-structured/debug "Writing hostname" {:task-id task-id :hostname hostname})
+  (if task-id
+    (try
+      (let [db (db conn)
+            [instance] (first (q '[:find ?i
+                                   :in $ ?task-id
+                                   :where
+                                   [?i :instance/task-id ?task-id]]
+                                 db task-id))]
+        (if (nil? instance)
+          (log-structured/error "Failed to write hostname: no instance for task-id."
+                                {:hostname hostname :task-id task-id})
+          @(d/transact conn [[:db/add instance :instance/hostname hostname]])))
+      (catch Exception e
+        (log/error e "Failed to write hostname: Datomic error")))
+    (log-structured/error "Failed to write hostname: task-id is nil."
+                          {:hostname hostname})))
+
 (defn- task-id->instance-id
   "Retrieves the instance-id given a task-id"
   [db task-id]
@@ -713,33 +734,36 @@
   "Limit the pending jobs to considerable jobs based on usage and quota.
    Further limit the considerable jobs to a maximum of num-considerable jobs."
   [db pending-jobs user->quota user->usage num-considerable pool-name]
-  (tracing/with-span [s {:name "scheduler.pending-jobs-to-considerable"
-                         :tags {:pool pool-name :component tracing-component-tag}}]
-    (log-structured/debug (print-str "There are pending jobs:" pending-jobs)
-                          {:pool pool-name :num-pending-jobs (count pending-jobs)})
-    (let [enforcing-job-launch-rate-limit? (ratelimit/enforce? quota/per-user-per-pool-launch-rate-limiter)
-          user->rate-limit-count (atom {})
-          user->passed-count (atom {})
-          considerable-jobs
-          (->> pending-jobs
-               (tools/filter-pending-jobs-for-quota pool-name user->rate-limit-count user->passed-count
-                                                    user->quota user->usage
-                                                    (tools/global-pool-quota pool-name))
-               (filter (fn [job] (tools/job-allowed-to-start? db job)))
-               (filter launch-plugin/filter-job-launches)
-               (take num-considerable)
+  (prom/with-duration
+    prom/scheduler-pool-handler-pending-to-considerable-duration {:pool pool-name}
+    (timers/time!
+     (timers/timer (metric-title "pool-handler-considerable-jobs-duration" pool-name))
+     (tracing/with-span [s {:name "scheduler.pending-jobs-to-considerable"
+                            :tags {:pool pool-name :component tracing-component-tag}}]
+       (log-structured/debug (print-str "There are pending jobs:" pending-jobs)
+                             {:pool pool-name :num-pending-jobs (count pending-jobs)})
+       (let [enforcing-job-launch-rate-limit? (ratelimit/enforce? quota/per-user-per-pool-launch-rate-limiter)
+             user->rate-limit-count (atom {})
+             user->passed-count (atom {})
+             considerable-jobs
+             (->> pending-jobs
+                  (tools/filter-pending-jobs-for-quota pool-name user->rate-limit-count user->passed-count
+                                                       user->quota user->usage
+                                                       (tools/global-pool-quota pool-name))
+                  (filter (fn [job] (tools/job-allowed-to-start? db job)))
+                  (filter launch-plugin/filter-job-launches)
+                  (take num-considerable)
                ; Force this to be taken eagerly so that the log line is accurate.
-               (doall))]
-      (swap! tools/pool->user->num-rate-limited-jobs update pool-name (constantly @user->rate-limit-count))
-      (log-structured/info "Job launch rate-limiting"
-                {:enforcing-job-launch-rate-limit? enforcing-job-launch-rate-limit?
-                 :total-rate-limit-count (->> @user->rate-limit-count vals (reduce +))
-                 :user->rate-limit-count @user->rate-limit-count
-                 :total-passed-count (->> @user->passed-count vals (reduce +))
-                 :user->passed-count @user->passed-count
-                 :pool pool-name})
-      considerable-jobs)))
-
+                  (doall))]
+         (swap! tools/pool->user->num-rate-limited-jobs update pool-name (constantly @user->rate-limit-count))
+         (log-structured/info "Job launch rate-limiting"
+                              {:enforcing-job-launch-rate-limit? enforcing-job-launch-rate-limit?
+                               :total-rate-limit-count (->> @user->rate-limit-count vals (reduce +))
+                               :user->rate-limit-count @user->rate-limit-count
+                               :total-passed-count (->> @user->passed-count vals (reduce +))
+                               :user->passed-count @user->passed-count
+                               :pool pool-name})
+         considerable-jobs)))))
 
 (defn matches->jobs
   "Given a collection of matches, returns the matched jobs"
@@ -1062,6 +1086,28 @@
                               :first-ten-jobs (print-str (->> jobs (take 10) (map :job/uuid) (map str)))}))
       (dissoc compute-cluster->jobs :no-acceptable-compute-cluster))))
 
+(defn distribute-and-launch-jobs
+  "Assign each job to a compute cluster (using configuration such as pool name and cluster
+   capability) and launch them in parallel.
+
+   A function `launch-distributed-job-fn` is used to actually do the launching. This is 
+   needed to implement the different launch logic for launching synthetic pods for 
+   autoscaling or for using the Kubernetes Scheduler directly."
+  [jobs pool-name compute-clusters job->acceptable-compute-clusters-fn launch-distributed-job-fn]
+  (try
+    (when (seq jobs)
+      (log-structured/info "Preparing for job distribution" {:pool pool-name})
+      (let [compute-cluster->jobs (distribute-jobs-to-compute-clusters jobs pool-name compute-clusters
+                                                                       job->acceptable-compute-clusters-fn)]
+        (log-structured/info "Starting job distribution" {:pool pool-name})
+        (->> compute-cluster->jobs
+             (map #(future (launch-distributed-job-fn %)))
+             doall
+             (run! deref)))
+      (log-structured/info "Done distributing jobs" {:pool pool-name}))
+    (catch Throwable e
+      (log-structured/error "Encountered error while distributing jobs" {:pool pool-name} e))))
+
 (defn trigger-autoscaling!
   "Autoscales the given pool to satisfy the given pending jobs, if:
   - There is at least one pending job
@@ -1072,29 +1118,19 @@
     (prom/with-duration
       prom/scheduler-trigger-autoscaling-duration {:pool pool-name}
       (timers/time!
-        (timers/timer (metric-title "trigger-autoscaling!-duration" pool-name))
-        (try
-          (let [autoscaling-compute-clusters (filter #(cc/autoscaling? % pool-name) compute-clusters)
-                num-autoscaling-compute-clusters (count autoscaling-compute-clusters)]
-            (if (and (pos? num-autoscaling-compute-clusters) (seq pending-jobs-for-autoscaling))
-              (do
-                (log-structured/info "Preparing for autoscaling" {:pool pool-name})
-                (let [compute-cluster->jobs (distribute-jobs-to-compute-clusters
-                                              pending-jobs-for-autoscaling pool-name autoscaling-compute-clusters
-                                              job->acceptable-compute-clusters-fn)]
-                  (log-structured/info "Starting autoscaling" {:pool pool-name})
-                  (->> compute-cluster->jobs
-                       (map
-                         (fn [[compute-cluster jobs-for-cluster]]
-                           (future (cc/autoscale! compute-cluster pool-name jobs-for-cluster adjust-job-resources-for-pool-fn))))
-                       doall
-                       (run! deref)))
-                (log-structured/info "Done autoscaling" {:pool pool-name}))
-              ; Update the synthetic pod counters even if we're not autoscaling, so we have accurate metrics
-              (doseq [compute-cluster autoscaling-compute-clusters]
-                (cc/set-synthetic-pods-counters compute-cluster pool-name))))
-          (catch Throwable e
-            (log-structured/error "Encountered error while triggering autoscaling" {:pool pool-name} e)))))))
+       (timers/timer (metric-title "trigger-autoscaling!-duration" pool-name))
+       (let [autoscaling-compute-clusters (filter #(cc/autoscaling? % pool-name) compute-clusters)
+             num-autoscaling-compute-clusters (count autoscaling-compute-clusters)
+             launch-distributed-job-fn
+             (fn [[compute-cluster jobs-for-cluster]]
+               (cc/autoscale! compute-cluster pool-name jobs-for-cluster adjust-job-resources-for-pool-fn))]
+
+         (if (and (pos? num-autoscaling-compute-clusters) (seq pending-jobs-for-autoscaling))
+           (distribute-and-launch-jobs pending-jobs-for-autoscaling pool-name autoscaling-compute-clusters
+                                       job->acceptable-compute-clusters-fn launch-distributed-job-fn)
+           ; Update the synthetic pod counters even if we're not autoscaling, so we have accurate metrics 
+           (doseq [compute-cluster autoscaling-compute-clusters]
+             (cc/set-synthetic-pods-counters compute-cluster pool-name))))))))
 
 (def pool-name->unmatched-job-uuid->unmatched-cycles-atom (atom {}))
 
@@ -1252,12 +1288,8 @@
           (try
             (let [db (d/db conn)
                   pending-jobs (get @pool-name->pending-jobs-atom pool-name)
-                  considerable-jobs (prom/with-duration
-                                      prom/scheduler-handle-resource-offers-pending-to-considerable-duration {:pool pool-name}
-                                      (timers/time!
-                                        (timers/timer (metric-title "handle-resource-offer!-considerable-jobs-duration" pool-name))
-                                        (pending-jobs->considerable-jobs
-                                          db pending-jobs user->quota user->usage num-considerable pool-name)))
+                  considerable-jobs (pending-jobs->considerable-jobs
+                                     db pending-jobs user->quota user->usage num-considerable pool-name)
                   ; matches is a vector of maps of {:hostname .. :leases .. :tasks}
                   {:keys [matches failures]} (prom/with-duration
                                                prom/scheduler-handle-resource-offers-match-duration {:pool pool-name}
@@ -1539,6 +1571,141 @@
                    max-considerable)
                  next-considerable))))))
 
+(defn job->task-txn
+  "Generate task transaction for a job handled by Kubernetes Scheduler."
+  [job metadata compute-cluster]
+  (let [{:keys [job/last-waiting-start-time job/uuid]} job
+        {:keys [executor task-id]} metadata
+        job-ref [:job/uuid uuid]
+        instance-start-time (now)]
+    [[:job/allowed-to-start? job-ref]
+     ; The job will remain waiting until the instance is running.
+     [:db/add job-ref :job/state :job.state/waiting]
+     (cond->
+      {:db/id (d/tempid :db.part/user)
+       :job/_instance job-ref
+       :instance/compute-cluster (cc/db-id compute-cluster)
+       :instance/executor executor
+       :instance/executor-id task-id ; NB command executor uses the task-id as the executor-id
+       :instance/ports []
+       :instance/preempted? false
+       :instance/progress 0
+       :instance/start-time instance-start-time
+       :instance/status :instance.status/unknown
+       :instance/task-id task-id}
+       last-waiting-start-time
+       (assoc :instance/queue-time
+              (- (.getTime instance-start-time)
+                 (.getTime ^Date last-waiting-start-time))))]))
+
+(defn jobs->kubernetes-task-metadata
+  "Generate sequence of task-metadata for jobs in a compute-cluster, suitable for Kubernetes Scheduler."
+  [jobs pool-name mesos-run-as-user compute-cluster]
+  (let [jobs->task-id (plumbing.core/map-from-keys (fn [_] (str (d/squuid))) jobs)]
+    (for [job jobs]
+      (let [pool-specific-resources
+            ((adjust-job-resources-for-pool-fn pool-name) job (tools/job-ent->resources job))]
+        (merge
+         (task/job->task-metadata compute-cluster mesos-run-as-user
+                                  job (jobs->task-id job))
+         {; Job constraints need to be expressed on a real job
+          ; pod so that if it triggers the cluster autoscaler,
+          ; CA will spin up nodes that will end up satisfying it. 
+          :pod-constraints (constraints/job->constraints job)
+
+          ; We need to label the pods handled completely by Kubernetes
+          ; so that we can treat them differently throughout the scheduler.
+          :pod-labels {"twosigma.com/cook-scheduler" "kubernetes"}
+
+          ; Cook has a "novel host constraint", which disallows a job from
+          ; running on the same host twice. So, we need to avoid running a
+          ; real job pod on any of the hosts that it won't be able to run on.
+          :pod-hostnames-to-avoid (constraints/job->previous-hosts-to-avoid job)
+          :task-request {:scalar-requests (walk/stringify-keys pool-specific-resources)
+                         :job (merge job {:job/pool {:pool/name pool-name}})
+                         ; Need to pass in resources to task-metadata->pod for gpu count
+                         :resources pool-specific-resources}})))))
+
+(defn distribute-jobs-to-kubernetes
+  "Distributes considerable jobs across the compute clusters and launches in Kubernetes."
+  [conn considerable-jobs pool-name compute-clusters mesos-run-as-user job->acceptable-compute-clusters-fn]
+  (tracing/with-span
+    [s {:name "scheduler.distribute-jobs-to-kubernetes" :tags {:component tracing-component-tag}}]
+    (prom/with-duration
+      prom/scheduler-distribute-jobs-to-kubernetes-duration {:pool pool-name}
+      (timers/time!
+       (timers/timer (metric-title "distribute-jobs-to-kubernetes-duration" pool-name))
+       (let [; NOTE: this is required to properly filter compute clusters to
+             ; those configured for this pool. This is done via the synthetic-pods
+             ; config, accessed via the autoscaling? method. In the future, this list
+             ; of pools should be slightly restructured to be listed one level higher,
+             ; on the compute-cluster-template config.
+             compute-clusters-for-pool (filter #(cc/autoscaling? % pool-name) compute-clusters)
+             launch-distributed-job-fn
+             (fn [[compute-cluster jobs]]
+               (let [kill-lock-object (cc/kill-lock-object compute-cluster)
+                     task-metadata-seq (jobs->kubernetes-task-metadata jobs pool-name mesos-run-as-user compute-cluster)
+                     task-txns (map (fn [job metadata] (job->task-txn job metadata compute-cluster)) jobs task-metadata-seq)]
+                 (try
+                   (log-structured/debug "Acquiring lock to commit tasks and and launch to Kubernetes."
+                                         {:pool pool-name :compute-cluster compute-cluster})
+                   (.. kill-lock-object readLock lock)
+                   (timers/time!
+                    (timers/timer (metric-title "distribute-jobs-to-kubernetes-transact-task-duration" pool-name))
+                    (datomic/transact
+                     conn
+                     (reduce into [] task-txns)
+                     (fn [e]
+                       (log-structured/warn "Transaction timed out, so these tasks might be present"
+                                            "in Datomic without actually having been launched in compute cluster"
+                                            {:compute-cluster compute-cluster :pool pool-name} e)
+                       (throw e))))
+                   (timers/time!
+                    (timers/timer (metric-title "distribute-jobs-to-kubernetes-launch-duration" pool-name))
+                    (cc/launch-tasks
+                     compute-cluster pool-name [{:task-metadata-seq task-metadata-seq}]
+                     (fn process-task-post-launch!
+                       [{:keys [_ task-request]}]
+                       (let [user (get-in task-request [:job :job/user])
+                             compute-cluster-launch-rate-limiter (cc/launch-rate-limiter compute-cluster)
+                             token-key (quota/pool+user->token-key pool-name user)]
+                         (ratelimit/spend! quota/per-user-per-pool-launch-rate-limiter token-key 1)
+                         (ratelimit/spend! compute-cluster-launch-rate-limiter ratelimit/compute-cluster-launch-rate-limiter-key 1))
+                       (prom/inc prom/scheduler-jobs-launched {:pool pool-name :compute-cluster (cc/compute-cluster-name compute-cluster)}))))
+                   (finally
+                     (.. kill-lock-object readLock unlock)))))]
+         (log-structured/info "Launching jobs in Kubernetes" {:pool pool-name})
+         (distribute-and-launch-jobs considerable-jobs pool-name compute-clusters-for-pool
+                                     job->acceptable-compute-clusters-fn launch-distributed-job-fn)
+         (log-structured/info "Launched jobs in Kubernetes" {:pool pool-name :number-launched-jobs (count considerable-jobs)}))))))
+
+(defn handle-kubernetes-scheduler-pool
+  "Handle scheduling pending jobs using the Kubernetes Scheduler."
+  [conn pool-name->pending-jobs-atom pool-name scheduler-config compute-clusters
+   job->acceptable-compute-clusters-fn user->quota user->usage-future mesos-run-as-user]
+  (log-structured/info "Running handler for Kubernetes Scheduler pool" {:pool pool-name})
+  (try
+    (let [user->usage (tracing/with-span [s {:name "scheduler.kubernetes-handler.resolve-user-to-usage-future"
+                                             :tags {:pool pool-name :component tracing-component-tag}}]
+                        @user->usage-future)
+          {max-considerable :max-jobs-considered} scheduler-config
+          db (db conn)
+          pending-jobs (get @pool-name->pending-jobs-atom pool-name)
+          considerable-jobs (pending-jobs->considerable-jobs db pending-jobs user->quota user->usage max-considerable pool-name)
+          considerable-job-uuids (set (map :job/uuid considerable-jobs))]
+      (log-structured/info "Considering jobs" {:pool pool-name :number-considered-jobs (count considerable-job-uuids)})
+      (if (seq considerable-jobs)
+        (do
+          (swap! pool-name->pending-jobs-atom
+                 remove-matched-jobs-from-pending-jobs
+                 considerable-job-uuids pool-name)
+          (log-structured/debug (print-str "Updated pool-name->pending-jobs-atom:" (get @pool-name->pending-jobs-atom pool-name))
+                                {:pool pool-name})
+          (distribute-jobs-to-kubernetes conn considerable-jobs pool-name compute-clusters mesos-run-as-user job->acceptable-compute-clusters-fn))
+        (log-structured/info "No considerable jobs launched this cycle" {:pool pool-name})))
+    (catch Exception e
+      (log-structured/error "Kubernetes handler encountered exception; continuing" {:pool pool-name} e))))
+
 (defn reconcile-jobs
   "Ensure all jobs saw their final state change"
   [conn]
@@ -2130,11 +2297,6 @@
                 "schedule each pool less often than the desired setting of every " target-per-pool-match-interval-millis " ms."))
     (async/pipe (chime-ch (util/time-seq (time/now) (time/millis match-interval-millis))) match-trigger-chan)))
 
-(defn kubernetes-scheduler?
-  "Return true if the scheduler is Kubernetes Scheduler."
-  [scheduler-name]
-  (= "kubernetes" scheduler-name))
-
 (defn pool-scheduler-config
   "Get the scheduler config for a pool."
   [schedulers pool-name]
@@ -2152,7 +2314,7 @@
         resources-atom (atom (view-incubating-offers fenzo))
         using-pools? (not (nil? (config/default-pool)))
         scheduler-name (:scheduler scheduler-config)
-        using-kubernetes-scheduler? (kubernetes-scheduler? scheduler-name)]
+        using-kubernetes-scheduler? (= "kubernetes" scheduler-name)]
     (when-not using-kubernetes-scheduler?
       (reset! fenzo-num-considerable-atom (:fenzo-max-jobs-considered scheduler-config)))
     (tools/chime-at-ch
@@ -2184,9 +2346,12 @@
                                                 (generate-user-usage-map (d/db conn) pool-name)))
                    compute-clusters (vals @cook.compute-cluster/cluster-name->compute-cluster-atom)
                    user->quota (quota/create-user->quota-fn (d/db conn) (if using-pools? pool-name nil))]
-               (handle-fenzo-pool conn fenzo fenzo-state resources-atom pool-name->pending-jobs-atom agent-attributes-cache 
-                                  scheduler-config rebalancer-reservation-atom mesos-run-as-user pool-name compute-clusters
-                                  job->acceptable-compute-clusters-fn user->quota user->usage-future))
+               (if using-kubernetes-scheduler?
+                 (handle-kubernetes-scheduler-pool conn pool-name->pending-jobs-atom pool-name scheduler-config compute-clusters
+                                                   job->acceptable-compute-clusters-fn user->quota user->usage-future mesos-run-as-user)
+                 (handle-fenzo-pool conn fenzo fenzo-state resources-atom pool-name->pending-jobs-atom agent-attributes-cache
+                                    scheduler-config rebalancer-reservation-atom mesos-run-as-user pool-name compute-clusters
+                                    job->acceptable-compute-clusters-fn user->quota user->usage-future)))
              (catch Exception e
                (log-structured/error "Pool handler encountered exception; continuing" {:pool pool-name} e)))))
        (log-structured/info "Done with pool handler" {:pool pool-name}))
@@ -2194,9 +2359,9 @@
     resources-atom))
 
 (defn create-datomic-scheduler
-  [{:keys [conn cluster-name->compute-cluster-atom mea-culpa-failure-limit 
+  [{:keys [conn cluster-name->compute-cluster-atom mea-culpa-failure-limit
            mesos-run-as-user agent-attributes-cache offer-incubate-time-ms
-           pool-name->pending-jobs-atom rebalancer-reservation-atom 
+           pool-name->pending-jobs-atom rebalancer-reservation-atom
            task-constraints trigger-chans]}]
 
   (persist-mea-culpa-failure-limit! conn mea-culpa-failure-limit)


Test patch:
diff --git a/scheduler/test/cook/test/config.clj b/scheduler/test/cook/test/config.clj
index bcad9c7c4d..fd23c0a60d 100644
--- a/scheduler/test/cook/test/config.clj
+++ b/scheduler/test/cook/test/config.clj
@@ -16,7 +16,7 @@
 (ns cook.test.config
   (:require [clojure.test :refer :all]
             [congestion.limits :as limits]
-            [cook.config :refer [config config-settings config-string->fitness-calculator default-pool env read-edn-config]
+            [cook.config :refer [config config-settings config-string->fitness-calculator default-pool env read-edn-config default-schedulers-config]
              :as config]
             [cook.test.rest.api :as api]
             [cook.test.testutil :refer [setup]])
@@ -49,7 +49,6 @@
   (with-redefs [config {:settings {}}]
     (is (nil? (default-pool)))))
 
-
 (def dummy-fitness-calculator
   "This calculator simply returns 0.0 for every Fenzo fitness calculation."
   (reify VMTaskFitnessCalculator
@@ -97,7 +96,77 @@
     (let [bad-config (assoc-in (api/minimal-config)
                                [:config :kubernetes :controller-lock-num-shards]
                                32778)]
-      (is (thrown? ExceptionInfo (config-settings bad-config))))))
+      (is (thrown? ExceptionInfo (config-settings bad-config)))))
+
+  (testing "default pool schedulers config applies"
+    (let [valid-config (dissoc (api/minimal-config) :pools)
+          applied-config (config-settings valid-config)
+          actual (get-in applied-config [:pools :schedulers])]
+      (is applied-config)
+      (is (= actual
+             default-schedulers-config))))
+
+  (testing "default fenzo config merges with partial but valid scheduler-config"
+    (let [valid-config (assoc-in (api/minimal-config) [:config :pools :schedulers] 
+                                 [{:pool-regex "test-pool"
+                                   :scheduler-config {:scheduler "fenzo"
+                                                      :fenzo-fitness-calculator "foo"}}])
+          applied-config (config-settings valid-config)
+          actual (get-in applied-config [:pools :schedulers])]
+      (is applied-config)
+      (is (= 1 (count actual)))
+      (is (= actual
+             [{:pool-regex "test-pool"
+               :scheduler-config {:scheduler "fenzo"
+                                  :good-enough-fitness 0.8
+                                  :fenzo-fitness-calculator "foo"
+                                  :fenzo-max-jobs-considered 1000
+                                  :fenzo-scaleback 0.95
+                                  :fenzo-floor-iterations-before-warn 10
+                                  :fenzo-floor-iterations-before-reset 1000}}]))))
+  
+  (testing "default kubernetes config merges with partial but valid scheduler-config"
+    (let [valid-config (assoc-in (api/minimal-config) [:config :pools :schedulers]
+                                 [{:pool-regex "test-pool"
+                                   :scheduler-config {:scheduler "kubernetes"}}])
+          applied-config (config-settings valid-config)
+          actual (get-in applied-config [:pools :schedulers])]
+      (is applied-config)
+      (is (= 1 (count actual)))
+      (is (= actual
+             [{:pool-regex "test-pool"
+               :scheduler-config {:scheduler "kubernetes"
+                                  :max-jobs-considered 1000}}])))))
+
+(deftest test-valid-schedulers-config
+  (testing "empty valid-schedulers-config"
+    (is (nil? (config/guard-invalid-schedulers-config []))))
+
+  (testing "valid config"
+    (is (nil? (config/guard-invalid-schedulers-config [{:pool-regex "test-pool"
+                                                        :scheduler-config {:scheduler "fenzo"}}]))))
+
+  (testing "no valid pool regex"
+    (is (thrown-with-msg? ExceptionInfo
+                          #"pool-regex key is missing from config"
+                          (config/guard-invalid-schedulers-config [{}]))))
+
+  (testing "no scheduler config"
+    (is (thrown-with-msg? ExceptionInfo
+                          #"scheduler-config key is missing from config"
+                          (config/guard-invalid-schedulers-config [{:pool-regex "test-pool"}]))))
+
+  (testing "no scheduler"
+    (is (thrown-with-msg? ExceptionInfo
+                          #"scheduler key is missing from scheduler-config"
+                          (config/guard-invalid-schedulers-config [{:pool-regex "test-pool"
+                                                                    :scheduler-config {}}]))))
+
+  (testing "invalid scheduler"
+    (is (thrown-with-msg? ExceptionInfo
+                          #"scheduler must be fenzo or kubernetes"
+                          (config/guard-invalid-schedulers-config [{:pool-regex "test-pool"
+                                                                    :scheduler-config {:scheduler ""}}])))))
 
 (deftest test-valid-gpu-models-config-settings
   (testing "empty valid-gpu-models"
diff --git a/scheduler/test/cook/test/kubernetes/api.clj b/scheduler/test/cook/test/kubernetes/api.clj
index b8a7a96101..7afd7d7380 100644
--- a/scheduler/test/cook/test/kubernetes/api.clj
+++ b/scheduler/test/cook/test/kubernetes/api.clj
@@ -677,167 +677,167 @@
 
 
 (deftest test-pod->synthesized-pod-state
-  (testing "returns nil for empty pod"
-    (is (nil? (api/pod->synthesized-pod-state nil nil))))
+  (with-redefs [api/kubernetes-scheduler-pod? (constantly false)]
+    (testing "returns nil for empty pod"
+      (is (nil? (api/pod->synthesized-pod-state nil nil))))
 
-  (testing "no container status -> waiting"
-    (let [pod (V1Pod.)
-          pod-status (V1PodStatus.)
-          pod-metadata (V1ObjectMeta.)
-          pod-name "test-pod"]
-      (.setStatus pod pod-status)
-      (.setName pod-metadata pod-name)
-      (.setMetadata pod pod-metadata)
-      (is (= {:state :pod/waiting
-              :reason "Pending"}
-             (api/pod->synthesized-pod-state pod-name pod)))))
-
-  (testing "waiting"
-    (let [pod (V1Pod.)
-          pod-status (V1PodStatus.)
-          pod-metadata (V1ObjectMeta.)
-          container-status (V1ContainerStatus.)
-          container-state (V1ContainerState.)
-          waiting (V1ContainerStateWaiting.)]
-      (.setReason waiting "waiting")
-      (.setWaiting container-state waiting)
-      (.setState container-status container-state)
-      (.setName container-status "required-cook-job-container")
-      (.setContainerStatuses pod-status [container-status])
-      (.setStatus pod pod-status)
-      (.setMetadata pod pod-metadata)
-      (is (= {:state :pod/waiting
-              :reason "waiting"}
-             (api/pod->synthesized-pod-state "test-pod" pod)))))
-
-  (testing "pod failed phase"
-    (let [pod (V1Pod.)
-          pod-status (V1PodStatus.)]
-      (.setPhase pod-status "Failed")
-      (.setReason pod-status "SomeSillyReason")
-      (.setStatus pod pod-status)
-      (is (= {:state :pod/failed
-              :reason "SomeSillyReason"}
-             (api/pod->synthesized-pod-state "test-pod" pod)))))
+    (testing "no container status -> waiting"
+      (let [pod (V1Pod.)
+            pod-status (V1PodStatus.)
+            pod-metadata (V1ObjectMeta.)
+            pod-name "test-pod"]
+        (.setStatus pod pod-status)
+        (.setName pod-metadata pod-name)
+        (.setMetadata pod pod-metadata)
+        (is (= {:state :pod/waiting
+                :reason "Pending"}
+               (api/pod->synthesized-pod-state pod-name pod)))))
 
-  (let [make-pod-with-condition-fn
-        (fn [pod-name pod-condition-type pod-condition-status
-             pod-condition-reason pod-condition-last-transition-time]
-          (let [pod (V1Pod.)
-                pod-status (V1PodStatus.)
-                pod-metadata (V1ObjectMeta.)
-                pod-condition (V1PodCondition.)]
-            (.setType pod-condition pod-condition-type)
-            (.setStatus pod-condition pod-condition-status)
-            (.setReason pod-condition pod-condition-reason)
-            (.setLastTransitionTime pod-condition pod-condition-last-transition-time)
-            (.addConditionsItem pod-status pod-condition)
-            (.setStatus pod pod-status)
-            (.setName pod-metadata pod-name)
-            (.setMetadata pod pod-metadata)
-            pod))
-
-        make-unschedulable-pod-fn
-        (fn [pod-name pod-condition-last-transition-time]
-          (make-pod-with-condition-fn pod-name "PodScheduled" "False" "Unschedulable"
-                                      pod-condition-last-transition-time))]
-
-    (testing "unschedulable pod"
-      (let [pod-name "test-pod"
-            pod (make-unschedulable-pod-fn pod-name (t/epoch))]
-        (with-redefs [config/kubernetes
-                      (constantly {:pod-condition-unschedulable-seconds 60})]
-          (is (= {:state :pod/failed
-                  :reason "Unschedulable"}
-                 (api/pod->synthesized-pod-state pod-name pod))))))
-
-    (testing "briefly unschedulable pod"
-      (let [pod-name "test-pod"
-            pod (make-unschedulable-pod-fn pod-name (t/now))]
-        (with-redefs [config/kubernetes
-                      (constantly {:pod-condition-unschedulable-seconds 60})]
-          (is (= {:state :pod/waiting
-                  :reason "Pending"}
-                 (api/pod->synthesized-pod-state pod-name pod))))))
-
-    (testing "unschedulable synthetic pod"
-      (let [pod-name api/cook-synthetic-pod-name-prefix
-            pod (make-unschedulable-pod-fn pod-name (t/minus (t/now) (t/minutes 16)))]
-        (with-redefs [config/kubernetes
-                      (constantly {:synthetic-pod-condition-unschedulable-seconds 900})]
-          (is (= {:state :pod/failed
-                  :reason "Unschedulable"}
-                 (api/pod->synthesized-pod-state pod-name pod))))))
-
-    (testing "briefly unschedulable synthetic pod"
-      (let [pod-name api/cook-synthetic-pod-name-prefix
-            pod (make-unschedulable-pod-fn pod-name (t/minus (t/now) (t/minutes 14)))]
-        (with-redefs [config/kubernetes
-                      (constantly {:synthetic-pod-condition-unschedulable-seconds 900})]
-          (is (= {:state :pod/waiting
-                  :reason "Pending"}
-                 (api/pod->synthesized-pod-state pod-name pod))))))
-
-    (testing "containers not initialized"
-      (let [pod-name "test-pod"
-            pod (make-pod-with-condition-fn pod-name "Initialized" "False" "ContainersNotInitialized" (t/epoch))
-            container-status (V1ContainerStatus.)
-            container-state (V1ContainerState.)
-            waiting (V1ContainerStateWaiting.)
-            pod-status (.getStatus pod)]
-        (.setReason waiting "waiting")
-        (.setWaiting container-state waiting)
-        (.setState container-status container-state)
-        (.setName container-status "required-cook-job-container")
-        (.setContainerStatuses pod-status [container-status])
-        (with-redefs [config/kubernetes
-                      (constantly {:pod-condition-containers-not-initialized-seconds 60})]
-          (is (= {:state :pod/failed
-                  :reason "ContainersNotInitialized"}
-                 (api/pod->synthesized-pod-state pod-name pod))))))
-
-    (testing "containers not ready"
-      (let [pod-name "test-pod"
-            pod (make-pod-with-condition-fn pod-name "ContainersReady" "False" "ContainersNotReady" (t/epoch))
+    (testing "waiting"
+      (let [pod (V1Pod.)
+            pod-status (V1PodStatus.)
+            pod-metadata (V1ObjectMeta.)
             container-status (V1ContainerStatus.)
             container-state (V1ContainerState.)
-            waiting (V1ContainerStateWaiting.)
-            pod-status (.getStatus pod)]
+            waiting (V1ContainerStateWaiting.)]
         (.setReason waiting "waiting")
         (.setWaiting container-state waiting)
         (.setState container-status container-state)
         (.setName container-status "required-cook-job-container")
         (.setContainerStatuses pod-status [container-status])
-        (with-redefs [config/kubernetes
-                      (constantly {:pod-condition-containers-not-ready-seconds 60})]
-          (is (= {:state :pod/failed
-                  :reason "ContainersNotReady"}
-                 (api/pod->synthesized-pod-state pod-name pod))))))
+        (.setStatus pod pod-status)
+        (.setMetadata pod pod-metadata)
+        (is (= {:state :pod/waiting
+                :reason "waiting"}
+               (api/pod->synthesized-pod-state "test-pod" pod)))))
 
-    (testing "node preempted default label"
+    (testing "pod failed phase"
       (let [pod (V1Pod.)
-            pod-metadata (V1ObjectMeta.)
             pod-status (V1PodStatus.)]
+        (.setPhase pod-status "Failed")
+        (.setReason pod-status "SomeSillyReason")
         (.setStatus pod pod-status)
-        (.setMetadata pod pod-metadata)
-        (.setLabels pod-metadata {"node-preempted" 1589084484537})
-        (is (= {:reason "Pending"
-                :state :pod/waiting
-                :pod-preempted-timestamp 1589084484537}
+        (is (= {:state :pod/failed
+                :reason "SomeSillyReason"}
                (api/pod->synthesized-pod-state "test-pod" pod)))))
+    (let [make-pod-with-condition-fn
+          (fn [pod-name pod-condition-type pod-condition-status
+               pod-condition-reason pod-condition-last-transition-time]
+            (let [pod (V1Pod.)
+                  pod-status (V1PodStatus.)
+                  pod-metadata (V1ObjectMeta.)
+                  pod-condition (V1PodCondition.)]
+              (.setType pod-condition pod-condition-type)
+              (.setStatus pod-condition pod-condition-status)
+              (.setReason pod-condition pod-condition-reason)
+              (.setLastTransitionTime pod-condition pod-condition-last-transition-time)
+              (.addConditionsItem pod-status pod-condition)
+              (.setStatus pod pod-status)
+              (.setName pod-metadata pod-name)
+              (.setMetadata pod pod-metadata)
+              pod))
+
+          make-unschedulable-pod-fn
+          (fn [pod-name pod-condition-last-transition-time]
+            (make-pod-with-condition-fn pod-name "PodScheduled" "False" "Unschedulable"
+                                        pod-condition-last-transition-time))]
+
+      (testing "unschedulable pod"
+        (let [pod-name "test-pod"
+              pod (make-unschedulable-pod-fn pod-name (t/epoch))]
+          (with-redefs [config/kubernetes
+                        (constantly {:pod-condition-unschedulable-seconds 60})]
+            (is (= {:state :pod/failed
+                    :reason "Unschedulable"}
+                   (api/pod->synthesized-pod-state pod-name pod))))))
+
+      (testing "briefly unschedulable pod"
+        (let [pod-name "test-pod"
+              pod (make-unschedulable-pod-fn pod-name (t/now))]
+          (with-redefs [config/kubernetes
+                        (constantly {:pod-condition-unschedulable-seconds 60})]
+            (is (= {:state :pod/waiting
+                    :reason "Pending"}
+                   (api/pod->synthesized-pod-state pod-name pod))))))
+
+      (testing "unschedulable synthetic pod"
+        (let [pod-name api/cook-synthetic-pod-name-prefix
+              pod (make-unschedulable-pod-fn pod-name (t/minus (t/now) (t/minutes 16)))]
+          (with-redefs [config/kubernetes
+                        (constantly {:synthetic-pod-condition-unschedulable-seconds 900})]
+            (is (= {:state :pod/failed
+                    :reason "Unschedulable"}
+                   (api/pod->synthesized-pod-state pod-name pod))))))
+
+      (testing "briefly unschedulable synthetic pod"
+        (let [pod-name api/cook-synthetic-pod-name-prefix
+              pod (make-unschedulable-pod-fn pod-name (t/minus (t/now) (t/minutes 14)))]
+          (with-redefs [config/kubernetes
+                        (constantly {:synthetic-pod-condition-unschedulable-seconds 900})]
+            (is (= {:state :pod/waiting
+                    :reason "Pending"}
+                   (api/pod->synthesized-pod-state pod-name pod))))))
+
+      (testing "containers not initialized"
+        (let [pod-name "test-pod"
+              pod (make-pod-with-condition-fn pod-name "Initialized" "False" "ContainersNotInitialized" (t/epoch))
+              container-status (V1ContainerStatus.)
+              container-state (V1ContainerState.)
+              waiting (V1ContainerStateWaiting.)
+              pod-status (.getStatus pod)]
+          (.setReason waiting "waiting")
+          (.setWaiting container-state waiting)
+          (.setState container-status container-state)
+          (.setName container-status "required-cook-job-container")
+          (.setContainerStatuses pod-status [container-status])
+          (with-redefs [config/kubernetes
+                        (constantly {:pod-condition-containers-not-initialized-seconds 60})]
+            (is (= {:state :pod/failed
+                    :reason "ContainersNotInitialized"}
+                   (api/pod->synthesized-pod-state pod-name pod))))))
+
+      (testing "containers not ready"
+        (let [pod-name "test-pod"
+              pod (make-pod-with-condition-fn pod-name "ContainersReady" "False" "ContainersNotReady" (t/epoch))
+              container-status (V1ContainerStatus.)
+              container-state (V1ContainerState.)
+              waiting (V1ContainerStateWaiting.)
+              pod-status (.getStatus pod)]
+          (.setReason waiting "waiting")
+          (.setWaiting container-state waiting)
+          (.setState container-status container-state)
+          (.setName container-status "required-cook-job-container")
+          (.setContainerStatuses pod-status [container-status])
+          (with-redefs [config/kubernetes
+                        (constantly {:pod-condition-containers-not-ready-seconds 60})]
+            (is (= {:state :pod/failed
+                    :reason "ContainersNotReady"}
+                   (api/pod->synthesized-pod-state pod-name pod))))))
 
-    (testing "node preempted custom label"
-      (with-redefs [config/kubernetes (fn [] {:node-preempted-label "custom-node-preempted"})]
+      (testing "node preempted default label"
         (let [pod (V1Pod.)
               pod-metadata (V1ObjectMeta.)
               pod-status (V1PodStatus.)]
           (.setStatus pod pod-status)
           (.setMetadata pod pod-metadata)
-          (.setLabels pod-metadata {"custom-node-preempted" 1589084484537})
+          (.setLabels pod-metadata {"node-preempted" 1589084484537})
           (is (= {:reason "Pending"
                   :state :pod/waiting
                   :pod-preempted-timestamp 1589084484537}
-                 (api/pod->synthesized-pod-state "test-pod" pod))))))))
+                 (api/pod->synthesized-pod-state "test-pod" pod)))))
+
+      (testing "node preempted custom label"
+        (with-redefs [config/kubernetes (fn [] {:node-preempted-label "custom-node-preempted"})]
+          (let [pod (V1Pod.)
+                pod-metadata (V1ObjectMeta.)
+                pod-status (V1PodStatus.)]
+            (.setStatus pod pod-status)
+            (.setMetadata pod pod-metadata)
+            (.setLabels pod-metadata {"custom-node-preempted" 1589084484537})
+            (is (= {:reason "Pending"
+                    :state :pod/waiting
+                    :pod-preempted-timestamp 1589084484537}
+                   (api/pod->synthesized-pod-state "test-pod" pod)))))))))
 
 (deftest test-node-schedulable
   ;; TODO: Need the 'stuck pod scanner' to detect stuck states and move them into killed.
diff --git a/scheduler/test/cook/test/kubernetes/controller.clj b/scheduler/test/cook/test/kubernetes/controller.clj
index 922ab267ff..61f29c3725 100644
--- a/scheduler/test/cook/test/kubernetes/controller.clj
+++ b/scheduler/test/cook/test/kubernetes/controller.clj
@@ -24,11 +24,12 @@
 (deftest test-container-status->failure-reason
   (testing "no container status + out-of-cpu pod status"
     (let [pod-status (V1PodStatus.)
-          container-status nil]
+          container-status nil
+          pod (tu/pod-helper "12345" "hostA" {:cpus 1.0 :mem 100.0})]
       (.setReason pod-status "OutOfcpu")
       (is (= :reason-invalid-offers
-             (controller/container-status->failure-reason {:name "test-cluster"} "12345"
-                                                          pod-status container-status))))))
+             (controller/container-status->failure-reason {:name "test-cluster"} pod-status
+                                                          pod container-status))))))
 (deftest test-process
   (tu/setup)
   (with-redefs [cached-queries/instance-uuid->job-uuid-datomic-query (constantly (java.util.UUID/randomUUID))
@@ -293,7 +294,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))
@@ -305,7 +306,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)))))
diff --git a/scheduler/test/cook/test/scheduler/scheduler.clj b/scheduler/test/cook/test/scheduler/scheduler.clj
index 9d8c866135..56969573c0 100644
--- a/scheduler/test/cook/test/scheduler/scheduler.clj
+++ b/scheduler/test/cook/test/scheduler/scheduler.clj
@@ -2523,15 +2523,15 @@
       compute-cluster-3 {:cluster-definition
                          {:config {:location location-a}}
                          :name compute-cluster-3-name}
-      instance-first {:instance/compute-cluster
-                      {:compute-cluster/cluster-name
-                       compute-cluster-2-name}
-                      :instance/start-time 1}
-      instance-second {:instance/compute-cluster
-                       {:compute-cluster/cluster-name
-                        compute-cluster-1-name}
-                       :instance/start-time 2}
-      instances [instance-second instance-first]]
+      instance-1 {:instance/compute-cluster
+                  {:compute-cluster/cluster-name
+                   compute-cluster-2-name}
+                  :instance/start-time 1}
+      instance-2 {:instance/compute-cluster
+                  {:compute-cluster/cluster-name
+                   compute-cluster-1-name}
+                  :instance/start-time 2}
+      instances [instance-1 instance-2]]
 
   (deftest test-job->preferred-compute-clusters
     (reset! cook.compute-cluster/cluster-name->compute-cluster-atom
@@ -2549,7 +2549,7 @@
     (is (= [compute-cluster-2]
            (sched/job->acceptable-compute-clusters
             {:job/checkpoint true
-              :job/instance [instance-first]}
+             :job/instance [instance-1]}
             [compute-cluster-1
              compute-cluster-2
              compute-cluster-3])))
@@ -2606,4 +2606,43 @@
                 [(assoc job :job/uuid (UUID/randomUUID))]
                 "test-pool"
                 [compute-cluster-2]
-                 sched/job->acceptable-compute-clusters)))))))
\ No newline at end of file
+                sched/job->acceptable-compute-clusters))))))
+
+  (deftest test-jobs-kubernetes-task-metadata
+    (testutil/setup)
+    (reset! cook.compute-cluster/cluster-name->compute-cluster-atom
+            {compute-cluster-1-name compute-cluster-1
+             compute-cluster-2-name compute-cluster-2
+             compute-cluster-3-name compute-cluster-3})
+    (with-redefs [cc/use-cook-executor? (fn [_] false)]
+      (let [base-job {:job/checkpoint true :job/user "wrinkle" :job/environment "env"}]
+        (let [job1 (assoc base-job :job/uuid (UUID/randomUUID) :job/name "job1")
+              job2 (assoc base-job :job/uuid (UUID/randomUUID) :job/name "job2")
+              job3 (assoc base-job :job/uuid (UUID/randomUUID) :job/name "job3")
+              jobs [job1 job2 job3]
+              compute-cluster->jobs (sched/distribute-jobs-to-compute-clusters
+                                     jobs
+                                     "test-pool"
+                                     [compute-cluster-1
+                                      compute-cluster-2
+                                      compute-cluster-3]
+                                     sched/job->acceptable-compute-clusters)
+              mesos-run-as-user "wrinkle"]
+          (doseq [[compute-cluster jobs] compute-cluster->jobs
+                  :let [task-metadata-seq (sched/jobs->kubernetes-task-metadata jobs "test-pool" mesos-run-as-user compute-cluster)]]
+            ; sanity check
+            (is (= (count task-metadata-seq)
+                   (count jobs)))
+
+            (doseq [[job metadata] (mapv vector jobs task-metadata-seq)
+                    :let [{:keys [job/name job/user job/uuid job/environment]} job
+                          metadata-job (get-in metadata [:task-request :job])]]
+              ; zipped in order of jobs (expected behavior in `handle-kubernetes-scheduler-pool` for generating task-txns)
+              (is (= (metadata-job :job/uuid)
+                     uuid))
+              (is (= (metadata-job :job/name)
+                     name))
+              (is (= (metadata-job :job/user)
+                     user))
+              (is (= (metadata-job :job/environment)
+                     environment)))))))))