   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:

                          ;;  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 (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)

                                                                                                   t)
                                                                             (list))))
                                                                       compute-clusters)))
                          _ (tracing/with-span [s {:name "scheduler.offer-handler.cache-offers-for-rebalancer"
                                                   :tags {:pool pool-name :component tracing-component-tag}}]
                                               (doseq [offer offers
                                                       :let [slave-id (-> offer :slave-id :value)]]
                                                 ; Cache offers for rebalancer so it can use job constraints when doing preemption decisions.
                                                 ; Computing get-offer-attr-map is pretty expensive because it includes calculating
                                                 ; currently running pods, so we have to union the set of pods k8s says are there and
                                                 ; the set of pods we're trying to put on the node. Even though it's not used by
                                                 ; rebalancer (and not needed). So it's OK if it's stale, so we do not need to refresh
                                                 ; and only store if it is a new node.
                                                 (when-not (ccache/get-if-present agent-attributes-cache identity slave-id)
                                                   (ccache/put-cache! agent-attributes-cache identity slave-id (offer/get-offer-attr-map offer)))))
                          using-pools? (not (nil? (config/default-pool)))
                          user->quota (quota/create-user->quota-fn (d/db conn) (if using-pools? pool-name nil))
                          user->usage (tracing/with-span [s {:name "scheduler.offer-handler.resolve-user-to-usage-future"
                                                             :tags {:pool pool-name :component tracing-component-tag}}]
                                                         @user->usage-future)
                          matched-head? (handle-resource-offers! conn fenzo-state pool-name->pending-jobs-atom
                                                                 mesos-run-as-user user->usage user->quota
                                                                 num-considerable offers
                                                                 rebalancer-reservation-atom pool-name compute-clusters
                                                                 job->acceptable-compute-clusters-fn)]
