                                                                                                    t)
                                                                              (list))))
                                                                        compute-clusters)))
-                          _ (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))))
+                          _ (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-future user->quota
+                                                                 mesos-run-as-user user->usage user->quota
                                                                  num-considerable offers
                                                                  rebalancer-reservation-atom pool-name compute-clusters
                                                                  job->acceptable-compute-clusters-fn)]

                                                      (max number-unmatched-jobs)) ; Autoscale at least the pods that failed to match.
                  ;; We need to filter pending jobs based on quota so that we don't
                  ;; trigger autoscaling beyond what users have quota to actually run
                  autoscalable-jobs (->> pool-name
                                         (get @pool-name->pending-jobs-atom)
                                         (tools/filter-pending-jobs-for-quota pool-name (atom {}) (atom {})
                                                                              user->quota user->usage (tools/global-pool-quota (config/pool-quotas) pool-name))
                                         (take max-jobs-for-autoscaling-scaled))
                  filtered-autoscalable-jobs (remove #(.getIfPresent caches/recent-synthetic-pod-job-uuids (:job/uuid %)) autoscalable-jobs)]
              ; When we have at least a minimum number of jobs being looked at, metric which fraction have matched.
              ; This lets us measure how well we're matching on existing resources.

(defn generate-user-usage-map
  "Returns a mapping from user to usage stats"
  [unfiltered-db pool-name]
  (tracing/with-span [s {:name "scheduler.generate-user-usage-map"
                         :tags {:pool pool-name :component tracing-component-tag}}]
    (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"
                         :tags {:pool pool-name :component tracing-component-tag :num-pending-jobs pending-jobs}}]
    (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
