   compute-cluster->jobs map. That is the API any future
   improvements need to stick to."
   [autoscalable-jobs pool-name compute-clusters job->acceptable-compute-clusters-fn]
-  (let [compute-cluster->jobs
-        (group-by
-          (fn choose-compute-cluster-for-autoscaling
-            [{:keys [job/uuid] :as job}]
-            (let [preferred-compute-clusters
-                  (job->acceptable-compute-clusters-fn job compute-clusters)]
-              (if (empty? preferred-compute-clusters)
-                :no-acceptable-compute-cluster
-                (nth preferred-compute-clusters
-                     (-> uuid hash (mod (count preferred-compute-clusters)))))))
-          autoscalable-jobs)]
-    (when-let [jobs (:no-acceptable-compute-cluster compute-cluster->jobs)]
-      (log-structured/info "There are jobs with no acceptable compute cluster for autoscaling"
-                           {:pool pool-name
-                            :first-ten-jobs (print-str (->> jobs (take 10) (map :job/uuid) (map str)))}))
-    (dissoc compute-cluster->jobs :no-acceptable-compute-cluster)))
+  (tracing/with-span [s {:name "scheduler.distribute-jobs-to-compute-cluster"
+                         :tags {:pool pool-name :component tracing-component-tag}}]
+    (let [compute-cluster->jobs
+          (group-by
+            (fn choose-compute-cluster-for-autoscaling
+              [{:keys [job/uuid] :as job}]
+              (let [preferred-compute-clusters
+                    (job->acceptable-compute-clusters-fn job compute-clusters)]
+                (if (empty? preferred-compute-clusters)
+                  :no-acceptable-compute-cluster
+                  (nth preferred-compute-clusters
+                       (-> uuid hash (mod (count preferred-compute-clusters)))))))
+            autoscalable-jobs)]
+      (when-let [jobs (:no-acceptable-compute-cluster compute-cluster->jobs)]
+        (log-structured/info "There are jobs with no acceptable compute cluster for autoscaling"
+                             {:pool pool-name
+                              :first-ten-jobs (print-str (->> jobs (take 10) (map :job/uuid) (map str)))}))
+      (dissoc compute-cluster->jobs :no-acceptable-compute-cluster))))
 
 (defn trigger-autoscaling!
   "Autoscales the given pool to satisfy the given pending jobs, if:

(defn generate-user-usage-map
  "Returns a mapping from user to usage stats"
  [unfiltered-db pool-name]
  (timers/time!
    generate-user-usage-map-duration
    (->> (tools/get-running-task-ents unfiltered-db)
         (map :job/_instance)
         (remove #(not= pool-name (cached-queries/job->pool-name %)))
         (group-by :job/user)
         (pc/map-vals (fn [jobs]
                        (->> jobs
                             (map tools/job->usage)
                             (reduce (partial merge-with +))))))))

(defn pending-jobs->considerable-jobs
  "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" :num-pending-jobs pending-jobs
                         :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 (config/pool-quotas) 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})
    (tracing/set-tags {:num-considerable (count considerable-jobs)})
    considerable-jobs))


(defn matches->jobs

                          ;;  2. Once the above two items are addressed, user->usage should always correctly
                          ;;     reflect *Cook*'s understanding of the state of the world at this point.
                          ;;     When this happens, users should never exceed their quota
                          user->usage-future (future (tracing/with-span [s1 {:from s :finish? false}] ; NOTE: finish? is set to false to prevent early finishing of the span
                                                                        (generate-user-usage-map (d/db conn) pool-name)))
                          ;; Try to clear the channel
                          ;; Merge the pending offers from all compute clusters.
                          compute-clusters (vals @cook.compute-cluster/cluster-name->compute-cluster-atom)
