{"action":"created","comment":{"url":"https://api.github.com/repos/apache/storm/pulls/comments/31266476","id":31266476,"diff_hunk":"@@ -373,4 +362,1222 @@\n     (ExecutorStats. (window-set-converter (:emitted stats) str)\n       (window-set-converter (:transferred stats) str)\n       specific-stats\n-      rate)))\n\\ No newline at end of file\n+      rate)))\n+\n+(defn- agg-bolt-lat-and-count\n+  \"Aggregates number executed, process latency, and execute latency across all\n+  streams.\"\n+  [idk->exec-avg idk->proc-avg idk->num-executed]\n+  {:pre (apply = (map #(set (keys %))\n+                      [idk->exec-avg\n+                       idk->proc-avg\n+                       idk->num-executed]))}\n+  (letfn [(weight-avg [[id avg]] (let [num-e (get idk->num-executed id)]\n+                                   (if (and avg num-e)\n+                                     (* avg num-e)\n+                                     0)))]\n+    {:executeLatencyTotal (sum (map weight-avg idk->exec-avg))\n+     :processLatencyTotal (sum (map weight-avg idk->proc-avg))\n+     :executed (sum (vals idk->num-executed))}))\n+\n+(defn- agg-spout-lat-and-count\n+  \"Aggregates number acked and complete latencies across all streams.\"\n+  [sid->comp-avg sid->num-acked]\n+  {:pre (apply = (map #(set (keys %))\n+                      [sid->comp-avg\n+                       sid->num-acked]))}\n+  (letfn [(weight-avg [[id avg]] (* avg (get sid->num-acked id)))]\n+    {:completeLatencyTotal (sum (map weight-avg sid->comp-avg))\n+     :acked (sum (vals sid->num-acked))}))\n+\n+(defn add-pairs\n+  ([] [0 0])\n+  ([[a1 a2] [b1 b2]]\n+   [(+ a1 b1) (+ a2 b2)]))\n+\n+(defn mk-include-sys-fn\n+  [include-sys?]\n+  (if include-sys?\n+    (fn [_] true)\n+    (fn [stream] (and (string? stream) (not (system-id? stream))))))\n+\n+(defn mk-include-sys-filter\n+  \"Returns a function that includes or excludes map entries whose keys are\n+  system ids.\"\n+  [include-sys?]\n+  (if include-sys?\n+    identity\n+    (partial filter-key (mk-include-sys-fn false))))\n+\n+(defn- agg-bolt-streams-lat-and-count\n+  \"Aggregates number executed and process & execute latencies.\"\n+  [idk->exec-avg idk->proc-avg idk->executed]\n+  {:pre (apply = (map #(set (keys %))\n+                      [idk->exec-avg\n+                       idk->proc-avg\n+                       idk->executed]))}\n+  (letfn [(weight-avg [id avg] (let [num-e (idk->executed id)]\n+                                   (if (and avg num-e)\n+                                     (* avg num-e)\n+                                     0)))]\n+    (into {}\n+      (for [k (keys idk->exec-avg)]\n+        [k {:executeLatencyTotal (weight-avg k (idk->exec-avg k))\n+            :processLatencyTotal (weight-avg k (idk->proc-avg k))\n+            :executed (idk->executed k)}]))))\n+\n+(defn- agg-spout-streams-lat-and-count\n+  \"Aggregates number acked and complete latencies.\"\n+  [idk->comp-avg idk->acked]\n+  {:pre (apply = (map #(set (keys %))\n+                      [idk->comp-avg\n+                       idk->acked]))}\n+  (letfn [(weight-avg [id avg] (let [num-e (get idk->acked id)]\n+                                   (if (and avg num-e)\n+                                     (* avg num-e)\n+                                     0)))]\n+    (into {}\n+      (for [k (keys idk->comp-avg)]\n+        [k {:completeLatencyTotal (weight-avg k (get idk->comp-avg k))\n+            :acked (get idk->acked k)}]))))\n+\n+(defn swap-map-order\n+  \"For a nested map, rearrange data such that the top-level keys become the\n+  nested map's keys and vice versa.\n+  Example:\n+  {:a {:X :banana, :Y :pear}, :b {:X :apple, :Y :orange}}\n+  -> {:Y {:a :pear, :b :orange}, :X {:a :banana, :b :apple}}\"\n+  [m]\n+  (apply merge-with\n+         merge\n+         (map (fn [[k v]]\n+                (into {}\n+                      (for [[k2 v2] v]\n+                        [k2 {k v2}])))\n+              m)))\n+\n+(defn- compute-agg-capacity\n+  \"Computes the capacity metric for one executor given its heartbeat data and\n+  uptime.\"\n+  [m uptime]\n+  (when uptime\n+    (->>\n+      ;; For each stream, create weighted averages and counts.\n+      (merge-with (fn weighted-avg+count-fn\n+                    [avg cnt]\n+                    [(* avg cnt) cnt])\n+                  (get (:execute-latencies m) (str TEN-MIN-IN-SECONDS))\n+                  (get (:executed m) (str TEN-MIN-IN-SECONDS)))\n+      vals ;; Ignore the stream ids.\n+      (reduce add-pairs\n+              [0. 0]) ;; Combine weighted averages and counts.\n+      ((fn [[weighted-avg cnt]]\n+        (div weighted-avg (* 1000 (min uptime TEN-MIN-IN-SECONDS))))))))\n+\n+(defn agg-pre-merge-comp-page-bolt\n+  [{exec-id :exec-id\n+    host :host\n+    port :port\n+    uptime :uptime\n+    comp-id :comp-id\n+    num-tasks :num-tasks\n+    statk->w->sid->num :stats}\n+   window\n+   include-sys?]\n+  (let [str-key (partial map-key str)\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    {:executor-id exec-id,\n+     :host host,\n+     :port port,\n+     :uptime uptime,\n+     :num-executors 1,\n+     :num-tasks num-tasks,\n+     :capacity (compute-agg-capacity statk->w->sid->num uptime)\n+     :cid+sid->input-stats\n+     (merge-with\n+       merge\n+       (swap-map-order\n+         {:acked (-> statk->w->sid->num\n+                     :acked\n+                     str-key\n+                     (get window))\n+          :failed (-> statk->w->sid->num\n+                      :failed\n+                      str-key\n+                      (get window))})\n+       (agg-bolt-streams-lat-and-count (-> statk->w->sid->num\n+                                           :execute-latencies\n+                                           str-key\n+                                           (get window))\n+                                       (-> statk->w->sid->num\n+                                           :process-latencies\n+                                           str-key\n+                                           (get window))\n+                                       (-> statk->w->sid->num\n+                                           :executed\n+                                           str-key\n+                                           (get window)))),\n+     :sid->output-stats\n+     (swap-map-order\n+       {:emitted (-> statk->w->sid->num\n+                     :emitted\n+                     str-key\n+                     (get window)\n+                     handle-sys-components-fn)\n+        :transferred (-> statk->w->sid->num\n+                         :transferred\n+                         str-key\n+                         (get window)\n+                         handle-sys-components-fn)})}))\n+\n+(defn agg-pre-merge-comp-page-spout\n+  [{exec-id :exec-id\n+    host :host\n+    port :port\n+    uptime :uptime\n+    comp-id :comp-id\n+    num-tasks :num-tasks\n+    statk->w->sid->num :stats}\n+   window\n+   include-sys?]\n+  (let [str-key (partial map-key str)\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    {:executor-id exec-id,\n+     :host host,\n+     :port port,\n+     :uptime uptime,\n+     :num-executors 1,\n+     :num-tasks num-tasks,\n+     :sid->output-stats\n+     (merge-with\n+       merge\n+       (agg-spout-streams-lat-and-count (-> statk->w->sid->num\n+                                            :complete-latencies\n+                                            str-key\n+                                            (get window))\n+                                        (-> statk->w->sid->num\n+                                            :acked\n+                                            str-key\n+                                            (get window)))\n+       (swap-map-order\n+         {:acked (-> statk->w->sid->num\n+                     :acked\n+                     str-key\n+                     (get window))\n+          :failed (-> statk->w->sid->num\n+                      :failed\n+                      str-key\n+                      (get window))\n+          :emitted (-> statk->w->sid->num\n+                       :emitted\n+                       str-key\n+                       (get window)\n+                       handle-sys-components-fn)\n+          :transferred (-> statk->w->sid->num\n+                           :transferred\n+                           str-key\n+                           (get window)\n+                           handle-sys-components-fn)}))}))\n+\n+(defn agg-pre-merge-topo-page-bolt\n+  [{comp-id :comp-id\n+    num-tasks :num-tasks\n+    statk->w->sid->num :stats\n+    uptime :uptime}\n+   window\n+   include-sys?]\n+  (let [str-key (partial map-key str)\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    {comp-id\n+     (merge\n+       (agg-bolt-lat-and-count (-> statk->w->sid->num\n+                                   :execute-latencies\n+                                   str-key\n+                                   (get window))\n+                               (-> statk->w->sid->num\n+                                   :process-latencies\n+                                   str-key\n+                                   (get window))\n+                               (-> statk->w->sid->num\n+                                   :executed\n+                                   str-key\n+                                   (get window)))\n+       {:num-executors 1\n+        :num-tasks num-tasks\n+        :emitted (-> statk->w->sid->num\n+                     :emitted\n+                     str-key\n+                     (get window)\n+                     handle-sys-components-fn\n+                     vals\n+                     sum)\n+        :transferred (-> statk->w->sid->num\n+                         :transferred\n+                         str-key\n+                         (get window)\n+                         handle-sys-components-fn\n+                         vals\n+                         sum)\n+        :capacity (compute-agg-capacity statk->w->sid->num uptime)\n+        :acked (-> statk->w->sid->num\n+                   :acked\n+                   str-key\n+                   (get window)\n+                   vals\n+                   sum)\n+        :failed (-> statk->w->sid->num\n+                    :failed\n+                    str-key\n+                    (get window)\n+                    vals\n+                    sum)})}))\n+\n+(defn agg-pre-merge-topo-page-spout\n+  [{comp-id :comp-id\n+    num-tasks :num-tasks\n+    statk->w->sid->num :stats}\n+   window\n+   include-sys?]\n+  (let [str-key (partial map-key str)\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    {comp-id\n+     (merge\n+       (agg-spout-lat-and-count (-> statk->w->sid->num\n+                                    :complete-latencies\n+                                    str-key\n+                                    (get window))\n+                                (-> statk->w->sid->num\n+                                    :acked\n+                                    str-key\n+                                    (get window)))\n+       {:num-executors 1\n+        :num-tasks num-tasks\n+        :emitted (-> statk->w->sid->num\n+                     :emitted\n+                     str-key\n+                     (get window)\n+                     handle-sys-components-fn\n+                     vals\n+                     sum)\n+        :transferred (-> statk->w->sid->num\n+                         :transferred\n+                         str-key\n+                         (get window)\n+                         handle-sys-components-fn\n+                         vals\n+                         sum)\n+        :failed (-> statk->w->sid->num\n+                    :failed\n+                    str-key\n+                    (get window)\n+                    vals\n+                    sum)})}))\n+\n+(defn apply-default\n+  [f defaulting-fn & args]\n+  (apply f (map defaulting-fn args)))\n+\n+(defn apply-or-0\n+  [f & args]\n+  (apply apply-default f #(or % 0) args))\n+\n+(defn sum-or-0\n+  [& args]\n+  (apply apply-or-0 + args))\n+\n+(defn max-or-0\n+  [& args]\n+  (apply apply-or-0 max args))\n+\n+(defn merge-agg-comp-stats-comp-page-bolt\n+  [{acc-in :cid+sid->input-stats\n+    acc-out :sid->output-stats\n+    :as acc-bolt-stats}\n+   {bolt-in :cid+sid->input-stats\n+    bolt-out :sid->output-stats\n+    :as bolt-stats}]\n+  {:num-executors (inc (or (:num-executors acc-bolt-stats) 0)),\n+   :num-tasks (sum-or-0 (:num-tasks acc-bolt-stats) (:num-tasks bolt-stats)),\n+   :sid->output-stats (merge-with (partial merge-with sum-or-0)\n+                                  acc-out\n+                                  bolt-out),\n+   :cid+sid->input-stats (merge-with (partial merge-with sum-or-0)\n+                                     acc-in\n+                                     bolt-in),\n+   :executor-stats\n+   (let [sum-streams (fn [m k] (->> m vals (map k) (apply sum-or-0)))\n+         executed (sum-streams bolt-in :executed)]\n+     (conj (:executor-stats acc-bolt-stats)\n+           (merge\n+             (select-keys bolt-stats\n+                          [:executor-id :uptime :host :port :capacity])\n+             {:emitted (sum-streams bolt-out :emitted)\n+              :transferred (sum-streams bolt-out :transferred)\n+              :acked (sum-streams bolt-in :acked)\n+              :failed (sum-streams bolt-in :failed)\n+              :executed executed}\n+             (->>\n+               (if (and executed (pos? executed))\n+                 [(div (sum-streams bolt-in :executeLatencyTotal) executed)\n+                  (div (sum-streams bolt-in :processLatencyTotal) executed)]\n+                 [nil nil])\n+               (mapcat vector [:execute-latency :process-latency])\n+               (apply assoc {})))))})\n+\n+(defn merge-agg-comp-stats-comp-page-spout\n+  [{acc-out :sid->output-stats\n+    :as acc-spout-stats}\n+   {spout-out :sid->output-stats\n+    :as spout-stats}]\n+  {:num-executors (inc (or (:num-executors acc-spout-stats) 0)),\n+   :num-tasks (sum-or-0 (:num-tasks acc-spout-stats) (:num-tasks spout-stats)),\n+   :sid->output-stats (merge-with (partial merge-with sum-or-0)\n+                                  acc-out\n+                                  spout-out),\n+   :executor-stats\n+   (let [sum-streams (fn [m k] (->> m vals (map k) (apply sum-or-0)))\n+         acked (sum-streams spout-out :acked)]\n+     (conj (:executor-stats acc-spout-stats)\n+           (merge\n+             (select-keys spout-stats [:executor-id :uptime :host :port])\n+             {:emitted (sum-streams spout-out :emitted)\n+              :transferred (sum-streams spout-out :transferred)\n+              :acked acked\n+              :failed (sum-streams spout-out :failed)}\n+             {:complete-latency (if (and acked (pos? acked))\n+                                  (div (sum-streams spout-out\n+                                                    :completeLatencyTotal)\n+                                       acked)\n+                                  nil)})))})\n+\n+(defn merge-agg-comp-stats-topo-page-bolt\n+  [acc-bolt-stats bolt-stats]\n+  {:num-executors (inc (or (:num-executors acc-bolt-stats) 0))\n+   :num-tasks (sum-or-0 (:num-tasks acc-bolt-stats) (:num-tasks bolt-stats))\n+   :emitted (sum-or-0 (:emitted acc-bolt-stats) (:emitted bolt-stats))\n+   :transferred (sum-or-0 (:transferred acc-bolt-stats)\n+                          (:transferred bolt-stats))\n+   :capacity (max-or-0 (:capacity acc-bolt-stats) (:capacity bolt-stats))\n+   ;; We sum average latency totals here to avoid dividing at each step.\n+   ;; Compute the average latencies by dividing the total by the count.\n+   :executeLatencyTotal (sum-or-0 (:executeLatencyTotal acc-bolt-stats)\n+                                  (:executeLatencyTotal bolt-stats))\n+   :processLatencyTotal (sum-or-0 (:processLatencyTotal acc-bolt-stats)\n+                                  (:processLatencyTotal bolt-stats))\n+   :executed (sum-or-0 (:executed acc-bolt-stats) (:executed bolt-stats))\n+   :acked (sum-or-0 (:acked acc-bolt-stats) (:acked bolt-stats))\n+   :failed (sum-or-0 (:failed acc-bolt-stats) (:failed bolt-stats))})\n+\n+(defn merge-agg-comp-stats-topo-page-spout\n+  [acc-spout-stats spout-stats]\n+  {:num-executors (inc (or (:num-executors acc-spout-stats) 0))\n+   :num-tasks (sum-or-0 (:num-tasks acc-spout-stats) (:num-tasks spout-stats))\n+   :emitted (sum-or-0 (:emitted acc-spout-stats) (:emitted spout-stats))\n+   :transferred (sum-or-0 (:transferred acc-spout-stats) (:transferred spout-stats))\n+   ;; We sum average latency totals here to avoid dividing at each step.\n+   ;; Compute the average latencies by dividing the total by the count.\n+   :completeLatencyTotal (sum-or-0 (:completeLatencyTotal acc-spout-stats)\n+                            (:completeLatencyTotal spout-stats))\n+   :acked (sum-or-0 (:acked acc-spout-stats) (:acked spout-stats))\n+   :failed (sum-or-0 (:failed acc-spout-stats) (:failed spout-stats))})\n+\n+(defn aggregate-count-streams\n+  [stats]\n+  (->> stats\n+       (map-val #(reduce + (vals %)))))\n+\n+(defn- agg-topo-exec-stats*\n+  \"A helper function that does the common work to aggregate stats of one\n+  executor with the given map for the topology page.\"\n+  [window\n+   include-sys?\n+   {:keys [workers-set\n+           bolt-id->stats\n+           spout-id->stats\n+           window->emitted\n+           window->transferred\n+           window->comp-lat-wgt-avg\n+           window->acked\n+           window->failed] :as acc-stats}\n+   {:keys [stats] :as new-data}\n+   pre-merge-fn\n+   merge-fn\n+   comp-key]\n+  (let [cid->statk->num (pre-merge-fn new-data window include-sys?)\n+        {w->compLatWgtAvg :completeLatencyTotal\n+         w->acked :acked}\n+          (if (:complete-latencies stats)\n+            (swap-map-order\n+              (into {}\n+                    (for [w (keys (:acked stats))]\n+                         [w (agg-spout-lat-and-count\n+                              (get (:complete-latencies stats) w)\n+                              (get (:acked stats) w))])))\n+            {:completeLatencyTotal nil\n+             :acks (aggregate-count-streams (:acked stats))})\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    (assoc {:workers-set (conj workers-set\n+                               [(:host new-data) (:port new-data)])\n+            :bolt-id->stats bolt-id->stats\n+            :spout-id->stats spout-id->stats\n+            :window->emitted (->> (:emitted stats)\n+                                  (map-val handle-sys-components-fn)\n+                                  aggregate-count-streams\n+                                  (merge-with + window->emitted))\n+            :window->transferred (->> (:transferred stats)\n+                                      (map-val handle-sys-components-fn)\n+                                      aggregate-count-streams\n+                                      (merge-with + window->transferred))\n+            :window->comp-lat-wgt-avg (merge-with +\n+                                                  window->comp-lat-wgt-avg\n+                                                  w->compLatWgtAvg)\n+            :window->acked (if (= :spout (:type stats))\n+                             (merge-with + window->acked w->acked)\n+                             window->acked)\n+            :window->failed (if (= :spout (:type stats))\n+                              (->> (:failed stats)\n+                                   aggregate-count-streams\n+                                   (merge-with + window->failed))\n+                              window->failed)}\n+           comp-key (merge-with merge-fn\n+                                (acc-stats comp-key)\n+                                cid->statk->num)\n+           :type (:type stats))))\n+\n+(defmulti agg-topo-exec-stats\n+  \"Combines the aggregate stats of one executor with the given map, selecting\n+  the appropriate window and including system components as specified.\"\n+  (fn dispatch-fn [& args] (:type (last args))))\n+\n+(defmethod agg-topo-exec-stats :bolt\n+  [window include-sys? acc-stats new-data]\n+  (agg-topo-exec-stats* window\n+                        include-sys?\n+                        acc-stats\n+                        new-data\n+                        agg-pre-merge-topo-page-bolt\n+                        merge-agg-comp-stats-topo-page-bolt\n+                        :bolt-id->stats))\n+\n+(defmethod agg-topo-exec-stats :spout\n+  [window include-sys? acc-stats new-data]\n+  (agg-topo-exec-stats* window\n+                        include-sys?\n+                        acc-stats\n+                        new-data\n+                        agg-pre-merge-topo-page-spout\n+                        merge-agg-comp-stats-topo-page-spout\n+                        :spout-id->stats))\n+\n+(defmethod agg-topo-exec-stats :default [_ _ acc-stats _] acc-stats)\n+\n+(defn get-last-error\n+  [storm-cluster-state storm-id component-id]\n+  (if-let [e (.last-error storm-cluster-state storm-id component-id)]\n+    (ErrorInfo. (:error e) (:time-secs e))))\n+\n+(defn component-type\n+  \"Returns the component type (either :bolt or :spout) for a given\n+  topology and component id. Returns nil if not found.\"\n+  [^StormTopology topology id]\n+  (let [bolts (.get_bolts topology)\n+        spouts (.get_spouts topology)]\n+    (cond\n+      (.containsKey bolts id) :bolt\n+      (.containsKey spouts id) :spout)))\n+\n+(defn extract-data-from-hb\n+  ([exec->host+port task->component beats include-sys? topology comp-id]\n+   (for [[[start end :as executor] [host port]] exec->host+port\n+         :let [beat (beats executor)\n+               id (task->component start)]\n+         :when (and (or (nil? comp-id) (= comp-id id))\n+                    (or include-sys? (not (system-id? id))))]\n+     {:exec-id executor\n+      :comp-id id\n+      :num-tasks (count (range start (inc end)))\n+      :host host\n+      :port port\n+      :uptime (:uptime beat)\n+      :stats (:stats beat)\n+      :type (or (:type (:stats beat))\n+                (component-type topology id))}))\n+  ([exec->host+port task->component beats include-sys? topology]\n+    (extract-data-from-hb exec->host+port\n+                          task->component\n+                          beats\n+                          include-sys?\n+                          topology\n+                          nil)))\n+\n+(defn aggregate-topo-stats\n+  [window include-sys? data]\n+  (let [init-val {:workers-set #{}\n+                  :bolt-id->stats {}\n+                  :spout-id->stats {}\n+                  :window->emitted {}\n+                  :window->transferred {}\n+                  :window->comp-lat-wgt-avg {}\n+                  :window->acked {}\n+                  :window->failed {}}\n+        reducer-fn (partial agg-topo-exec-stats\n+                            window\n+                            include-sys?)]\n+    (reduce reducer-fn init-val data)))\n+\n+(defn- compute-weighted-averages-per-window\n+  [acc-data wgt-avg-key divisor-key]\n+  (into {} (for [[window wgt-avg] (wgt-avg-key acc-data)\n+                 :let [divisor ((divisor-key acc-data) window)]\n+                 :when (and divisor (pos? divisor))]\n+             [(str window) (div wgt-avg divisor)])))\n+\n+(defn- post-aggregate-topo-stats\n+  [task->component exec->node+port last-err-fn acc-data]\n+  {:num-tasks (count task->component)\n+   :num-workers (count (:workers-set acc-data))\n+   :num-executors (count exec->node+port)\n+   :bolt-id->stats\n+     (into {} (for [[id m] (:bolt-id->stats acc-data)\n+                    :let [executed (:executed m)]]\n+                     [id (-> m\n+                             (assoc :execute-latency\n+                                    (if (and executed (pos? executed))\n+                                      (div (or (:executeLatencyTotal m) 0)\n+                                           executed)\n+                                      0)\n+                                    :process-latency\n+                                    (if (and executed (pos? executed))\n+                                      (div (or (:processLatencyTotal m) 0)\n+                                           executed)\n+                                      0))\n+                             (dissoc :executeLatencyTotal\n+                                     :processLatencyTotal)\n+                             (assoc :lastError (last-err-fn id)))]))\n+   :spout-id->stats\n+     (into {} (for [[id m] (:spout-id->stats acc-data)\n+                    :let [acked (:acked m)]]\n+                    [id (-> m\n+                            (assoc :completeLatency\n+                                   (if (and acked (pos? acked))\n+                                     (div (:completeLatencyTotal m)\n+                                          (:acked m))\n+                                     0))\n+                            (dissoc :completeLatencyTotal)\n+                            (assoc :lastError (last-err-fn id)))]))\n+   :window->emitted (map-key str (:window->emitted acc-data))\n+   :window->transferred (map-key str (:window->transferred acc-data))\n+   :window->complete-latency\n+     (compute-weighted-averages-per-window acc-data\n+                                           :window->comp-lat-wgt-avg\n+                                           :window->acked)\n+   :window->acked (map-key str (:window->acked acc-data))\n+   :window->failed (map-key str (:window->failed acc-data))})\n+\n+(defn- thriftify-common-agg-stats\n+  [^ComponentAggregateStats s\n+   {:keys [num-tasks\n+           emitted\n+           transferred\n+           acked\n+           failed\n+           num-executors] :as statk->num}]\n+  (let [cas (CommonAggregateStats.)]\n+    (and num-executors (.set_num_executors cas num-executors))\n+    (and num-tasks (.set_num_tasks cas num-tasks))\n+    (and emitted (.set_emitted cas emitted))\n+    (and transferred (.set_transferred cas transferred))\n+    (and acked (.set_acked cas acked))\n+    (and failed (.set_failed cas failed))\n+    (.set_common_stats s cas)))\n+\n+(defn thriftify-bolt-agg-stats\n+  [statk->num]\n+  (let [{:keys [lastError\n+                execute-latency\n+                process-latency\n+                executed\n+                capacity]} statk->num\n+        s (ComponentAggregateStats.)]\n+    (.set_type s ComponentType/BOLT)\n+    (and lastError (.set_last_error s lastError))\n+    (thriftify-common-agg-stats s statk->num)\n+    (.set_specific_stats s\n+      (SpecificAggregateStats/bolt\n+        (let [bas (BoltAggregateStats.)]\n+          (and execute-latency (.set_execute_latency_ms bas execute-latency))\n+          (and process-latency (.set_process_latency_ms bas process-latency))\n+          (and executed (.set_executed bas executed))\n+          (and capacity (.set_capacity bas capacity))\n+          bas)))\n+    s))\n+\n+(defn thriftify-spout-agg-stats\n+  [statk->num]\n+  (let [{:keys [lastError\n+                complete-latency]} statk->num\n+        s (ComponentAggregateStats.)]\n+    (.set_type s ComponentType/SPOUT)\n+    (and lastError (.set_last_error s lastError))\n+    (thriftify-common-agg-stats s statk->num)\n+    (.set_specific_stats s\n+      (SpecificAggregateStats/spout\n+        (let [sas (SpoutAggregateStats.)]\n+          (and complete-latency (.set_complete_latency_ms sas complete-latency))\n+          sas)))\n+    s))\n+\n+(defn thriftify-topo-page-data\n+  [topology-id data]\n+  (let [{:keys [num-tasks\n+                num-workers\n+                num-executors\n+                spout-id->stats\n+                bolt-id->stats\n+                window->emitted\n+                window->transferred\n+                window->complete-latency\n+                window->acked\n+                window->failed]} data\n+        spout-agg-stats (into {}\n+                              (for [[id m] spout-id->stats\n+                                    :let [m (assoc m :type :spout)]]\n+                                [id\n+                                 (thriftify-spout-agg-stats m)]))\n+        bolt-agg-stats (into {}\n+                             (for [[id m] bolt-id->stats\n+                                   :let [m (assoc m :type :bolt)]]\n+                              [id\n+                               (thriftify-bolt-agg-stats m)]))\n+        topology-stats (doto (TopologyStats.)\n+                         (.set_window_to_emitted window->emitted)\n+                         (.set_window_to_transferred window->transferred)\n+                         (.set_window_to_complete_latencies_ms\n+                           window->complete-latency)\n+                         (.set_window_to_acked window->acked)\n+                         (.set_window_to_failed window->failed))\n+      topo-page-info (doto (TopologyPageInfo. topology-id)\n+                       (.set_num_tasks num-tasks)\n+                       (.set_num_workers num-workers)\n+                       (.set_num_executors num-executors)\n+                       (.set_id_to_spout_agg_stats spout-agg-stats)\n+                       (.set_id_to_bolt_agg_stats bolt-agg-stats)\n+                       (.set_topology_stats topology-stats))]\n+    topo-page-info))\n+\n+(defn agg-topo-execs-stats\n+  \"Aggregate various executor statistics for a topology from the given\n+  heartbeats.\"\n+  [topology-id\n+   exec->node+port\n+   task->component\n+   beats\n+   topology\n+   window\n+   include-sys?\n+   last-err-fn]\n+  (->> ;; This iterates over each executor one time, because of lazy evaluation.\n+    (extract-data-from-hb exec->node+port\n+                          task->component\n+                          beats\n+                          include-sys?\n+                          topology)\n+    (aggregate-topo-stats window include-sys?)\n+    (post-aggregate-topo-stats task->component exec->node+port last-err-fn)\n+    (thriftify-topo-page-data topology-id)))\n+\n+(defn- agg-bolt-exec-win-stats\n+  \"A helper function that aggregates windowed stats from one bolt executor.\"\n+  [acc-stats new-stats include-sys?]\n+  (let [{w->execLatWgtAvg :executeLatencyTotal\n+         w->procLatWgtAvg :processLatencyTotal\n+         w->executed :executed}\n+          (swap-map-order\n+            (into {} (for [w (keys (:executed new-stats))]\n+                       [w (agg-bolt-lat-and-count\n+                            (get (:execute-latencies new-stats) w)\n+                            (get (:process-latencies new-stats) w)\n+                            (get (:executed new-stats) w))])))\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    {:window->emitted (->> (:emitted new-stats)\n+                           (map-val handle-sys-components-fn)\n+                           aggregate-count-streams\n+                           (merge-with + (:window->emitted acc-stats)))\n+     :window->transferred (->> (:transferred new-stats)\n+                               (map-val handle-sys-components-fn)\n+                               aggregate-count-streams\n+                               (merge-with + (:window->transferred acc-stats)))\n+     :window->exec-lat-wgt-avg (merge-with +\n+                                           (:window->exec-lat-wgt-avg acc-stats)\n+                                           w->execLatWgtAvg)\n+     :window->proc-lat-wgt-avg (merge-with +\n+                                           (:window->proc-lat-wgt-avg acc-stats)\n+                                           w->procLatWgtAvg)\n+     :window->executed (merge-with + (:window->executed acc-stats) w->executed)\n+     :window->acked (->> (:acked new-stats)\n+                         aggregate-count-streams\n+                         (merge-with + (:window->acked acc-stats)))\n+     :window->failed (->> (:failed new-stats)\n+                          aggregate-count-streams\n+                          (merge-with + (:window->failed acc-stats)))}))\n+\n+(defn- agg-spout-exec-win-stats\n+  \"A helper function that aggregates windowed stats from one spout executor.\"\n+  [acc-stats new-stats include-sys?]\n+  (let [{w->compLatWgtAvg :completeLatencyTotal\n+         w->acked :acked}\n+          (swap-map-order\n+            (into {} (for [w (keys (:acked new-stats))]\n+                       [w (agg-spout-lat-and-count\n+                            (get (:complete-latencies new-stats) w)\n+                            (get (:acked new-stats) w))])))\n+        handle-sys-components-fn (mk-include-sys-filter include-sys?)]\n+    {:window->emitted (->> (:emitted new-stats)\n+                           (map-val handle-sys-components-fn)\n+                           aggregate-count-streams\n+                           (merge-with + (:window->emitted acc-stats)))\n+     :window->transferred (->> (:transferred new-stats)\n+                               (map-val handle-sys-components-fn)\n+                               aggregate-count-streams\n+                               (merge-with + (:window->transferred acc-stats)))\n+     :window->comp-lat-wgt-avg (merge-with +\n+                                           (:window->comp-lat-wgt-avg acc-stats)\n+                                           w->compLatWgtAvg)\n+     :window->acked (->> (:acked new-stats)\n+                         aggregate-count-streams\n+                         (merge-with + (:window->acked acc-stats)))\n+     :window->failed (->> (:failed new-stats)\n+                          aggregate-count-streams\n+                          (merge-with + (:window->failed acc-stats)))}))\n+\n+(defmulti agg-comp-exec-stats\n+  \"Combines the aggregate stats of one executor with the given map, selecting\n+  the appropriate window and including system components as specified.\"\n+  (fn dispatch-fn [_ _ init-val _] (:type init-val)))\n+\n+(defmethod agg-comp-exec-stats :bolt\n+  [window include-sys? acc-stats new-data]\n+  (assoc (agg-bolt-exec-win-stats acc-stats (:stats new-data) include-sys?)\n+         :stats (merge-agg-comp-stats-comp-page-bolt\n+                  (:stats acc-stats)\n+                  (agg-pre-merge-comp-page-bolt new-data window include-sys?))\n+         :type :bolt))\n+\n+(defmethod agg-comp-exec-stats :spout\n+  [window include-sys? acc-stats new-data]\n+  (assoc (agg-spout-exec-win-stats acc-stats (:stats new-data) include-sys?)\n+         :stats (merge-agg-comp-stats-comp-page-spout\n+                  (:stats acc-stats)\n+                  (agg-pre-merge-comp-page-spout new-data window include-sys?))\n+         :type :spout))\n+\n+(defn- aggregate-comp-stats*\n+  [window include-sys? data init-val]\n+  (-> (partial agg-comp-exec-stats\n+               window\n+               include-sys?)\n+      (reduce init-val data)))\n+\n+(defmulti aggregate-comp-stats\n+  (fn dispatch-fn [& args] (-> args last first :stats :type)))","path":"storm-core/src/clj/backtype/storm/stats.clj","position":911,"original_position":911,"commit_id":"4b11f71b7e2fe45a62ad5a26d9d6bc32ebdec6df","original_commit_id":"4b11f71b7e2fe45a62ad5a26d9d6bc32ebdec6df","user":{"login":"d2r","id":905298,"avatar_url":"https://avatars.githubusercontent.com/u/905298?v=3","gravatar_id":"","url":"https://api.github.com/users/d2r","html_url":"https://github.com/d2r","followers_url":"https://api.github.com/users/d2r/followers","following_url":"https://api.github.com/users/d2r/following{/other_user}","gists_url":"https://api.github.com/users/d2r/gists{/gist_id}","starred_url":"https://api.github.com/users/d2r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/d2r/subscriptions","organizations_url":"https://api.github.com/users/d2r/orgs","repos_url":"https://api.github.com/users/d2r/repos","events_url":"https://api.github.com/users/d2r/events{/privacy}","received_events_url":"https://api.github.com/users/d2r/received_events","type":"User","site_admin":false},"body":"This could result in a null dispatch value if the component's heartbeat does not contain metrics.  We should be using the `:type` directly instead of accessing it via `:stats`. ","created_at":"2015-05-28T19:01:30Z","updated_at":"2015-05-28T19:01:30Z","html_url":"https://github.com/apache/storm/pull/554#discussion_r31266476","pull_request_url":"https://api.github.com/repos/apache/storm/pulls/554","_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/comments/31266476"},"html":{"href":"https://github.com/apache/storm/pull/554#discussion_r31266476"},"pull_request":{"href":"https://api.github.com/repos/apache/storm/pulls/554"}}},"pull_request":{"url":"https://api.github.com/repos/apache/storm/pulls/554","id":35575738,"html_url":"https://github.com/apache/storm/pull/554","diff_url":"https://github.com/apache/storm/pull/554.diff","patch_url":"https://github.com/apache/storm/pull/554.patch","issue_url":"https://api.github.com/repos/apache/storm/issues/554","number":554,"state":"open","locked":false,"title":"[STORM-820] Aggregate topo stats on nimbus, not ui","user":{"login":"d2r","id":905298,"avatar_url":"https://avatars.githubusercontent.com/u/905298?v=3","gravatar_id":"","url":"https://api.github.com/users/d2r","html_url":"https://github.com/d2r","followers_url":"https://api.github.com/users/d2r/followers","following_url":"https://api.github.com/users/d2r/following{/other_user}","gists_url":"https://api.github.com/users/d2r/gists{/gist_id}","starred_url":"https://api.github.com/users/d2r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/d2r/subscriptions","organizations_url":"https://api.github.com/users/d2r/orgs","repos_url":"https://api.github.com/users/d2r/repos","events_url":"https://api.github.com/users/d2r/events{/privacy}","received_events_url":"https://api.github.com/users/d2r/received_events","type":"User","site_admin":false},"body":"","created_at":"2015-05-16T19:12:12Z","updated_at":"2015-05-28T19:01:30Z","closed_at":null,"merged_at":null,"merge_commit_sha":"51a6d1c52a03a287e90330835e6c2af6ef5dc4f9","assignee":null,"milestone":null,"commits_url":"https://api.github.com/repos/apache/storm/pulls/554/commits","review_comments_url":"https://api.github.com/repos/apache/storm/pulls/554/comments","review_comment_url":"https://api.github.com/repos/apache/storm/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/storm/issues/554/comments","statuses_url":"https://api.github.com/repos/apache/storm/statuses/4b11f71b7e2fe45a62ad5a26d9d6bc32ebdec6df","head":{"label":"d2r:storm-820-agg-stats-on-nimbus","ref":"storm-820-agg-stats-on-nimbus","sha":"4b11f71b7e2fe45a62ad5a26d9d6bc32ebdec6df","user":{"login":"d2r","id":905298,"avatar_url":"https://avatars.githubusercontent.com/u/905298?v=3","gravatar_id":"","url":"https://api.github.com/users/d2r","html_url":"https://github.com/d2r","followers_url":"https://api.github.com/users/d2r/followers","following_url":"https://api.github.com/users/d2r/following{/other_user}","gists_url":"https://api.github.com/users/d2r/gists{/gist_id}","starred_url":"https://api.github.com/users/d2r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/d2r/subscriptions","organizations_url":"https://api.github.com/users/d2r/orgs","repos_url":"https://api.github.com/users/d2r/repos","events_url":"https://api.github.com/users/d2r/events{/privacy}","received_events_url":"https://api.github.com/users/d2r/received_events","type":"User","site_admin":false},"repo":{"id":15347626,"name":"storm","full_name":"d2r/storm","owner":{"login":"d2r","id":905298,"avatar_url":"https://avatars.githubusercontent.com/u/905298?v=3","gravatar_id":"","url":"https://api.github.com/users/d2r","html_url":"https://github.com/d2r","followers_url":"https://api.github.com/users/d2r/followers","following_url":"https://api.github.com/users/d2r/following{/other_user}","gists_url":"https://api.github.com/users/d2r/gists{/gist_id}","starred_url":"https://api.github.com/users/d2r/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/d2r/subscriptions","organizations_url":"https://api.github.com/users/d2r/orgs","repos_url":"https://api.github.com/users/d2r/repos","events_url":"https://api.github.com/users/d2r/events{/privacy}","received_events_url":"https://api.github.com/users/d2r/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/d2r/storm","description":"Mirror of Apache Storm","fork":true,"url":"https://api.github.com/repos/d2r/storm","forks_url":"https://api.github.com/repos/d2r/storm/forks","keys_url":"https://api.github.com/repos/d2r/storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/d2r/storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/d2r/storm/teams","hooks_url":"https://api.github.com/repos/d2r/storm/hooks","issue_events_url":"https://api.github.com/repos/d2r/storm/issues/events{/number}","events_url":"https://api.github.com/repos/d2r/storm/events","assignees_url":"https://api.github.com/repos/d2r/storm/assignees{/user}","branches_url":"https://api.github.com/repos/d2r/storm/branches{/branch}","tags_url":"https://api.github.com/repos/d2r/storm/tags","blobs_url":"https://api.github.com/repos/d2r/storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/d2r/storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/d2r/storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/d2r/storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/d2r/storm/statuses/{sha}","languages_url":"https://api.github.com/repos/d2r/storm/languages","stargazers_url":"https://api.github.com/repos/d2r/storm/stargazers","contributors_url":"https://api.github.com/repos/d2r/storm/contributors","subscribers_url":"https://api.github.com/repos/d2r/storm/subscribers","subscription_url":"https://api.github.com/repos/d2r/storm/subscription","commits_url":"https://api.github.com/repos/d2r/storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/d2r/storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/d2r/storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/d2r/storm/issues/comments{/number}","contents_url":"https://api.github.com/repos/d2r/storm/contents/{+path}","compare_url":"https://api.github.com/repos/d2r/storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/d2r/storm/merges","archive_url":"https://api.github.com/repos/d2r/storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/d2r/storm/downloads","issues_url":"https://api.github.com/repos/d2r/storm/issues{/number}","pulls_url":"https://api.github.com/repos/d2r/storm/pulls{/number}","milestones_url":"https://api.github.com/repos/d2r/storm/milestones{/number}","notifications_url":"https://api.github.com/repos/d2r/storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/d2r/storm/labels{/name}","releases_url":"https://api.github.com/repos/d2r/storm/releases{/id}","created_at":"2013-12-20T20:35:31Z","updated_at":"2015-03-26T21:29:37Z","pushed_at":"2015-05-22T21:51:29Z","git_url":"git://github.com/d2r/storm.git","ssh_url":"git@github.com:d2r/storm.git","clone_url":"https://github.com/d2r/storm.git","svn_url":"https://github.com/d2r/storm","homepage":null,"size":19991,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"512d3def033e2a66ce8f2c8a401c7cef99cbedea","user":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":14135470,"name":"storm","full_name":"apache/storm","owner":{"login":"apache","id":47359,"avatar_url":"https://avatars.githubusercontent.com/u/47359?v=3","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/storm","description":"Mirror of Apache Storm","fork":false,"url":"https://api.github.com/repos/apache/storm","forks_url":"https://api.github.com/repos/apache/storm/forks","keys_url":"https://api.github.com/repos/apache/storm/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/storm/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/storm/teams","hooks_url":"https://api.github.com/repos/apache/storm/hooks","issue_events_url":"https://api.github.com/repos/apache/storm/issues/events{/number}","events_url":"https://api.github.com/repos/apache/storm/events","assignees_url":"https://api.github.com/repos/apache/storm/assignees{/user}","branches_url":"https://api.github.com/repos/apache/storm/branches{/branch}","tags_url":"https://api.github.com/repos/apache/storm/tags","blobs_url":"https://api.github.com/repos/apache/storm/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/storm/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/storm/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/storm/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/storm/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/storm/languages","stargazers_url":"https://api.github.com/repos/apache/storm/stargazers","contributors_url":"https://api.github.com/repos/apache/storm/contributors","subscribers_url":"https://api.github.com/repos/apache/storm/subscribers","subscription_url":"https://api.github.com/repos/apache/storm/subscription","commits_url":"https://api.github.com/repos/apache/storm/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/storm/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/storm/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/storm/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/storm/contents/{+path}","compare_url":"https://api.github.com/repos/apache/storm/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/storm/merges","archive_url":"https://api.github.com/repos/apache/storm/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/storm/downloads","issues_url":"https://api.github.com/repos/apache/storm/issues{/number}","pulls_url":"https://api.github.com/repos/apache/storm/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/storm/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/storm/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/storm/labels{/name}","releases_url":"https://api.github.com/repos/apache/storm/releases{/id}","created_at":"2013-11-05T08:00:14Z","updated_at":"2015-05-28T18:26:29Z","pushed_at":"2015-05-27T22:09:32Z","git_url":"git://github.com/apache/storm.git","ssh_url":"git@github.com:apache/storm.git","clone_url":"https://github.com/apache/storm.git","svn_url":"https://github.com/apache/storm","homepage":null,"size":945151,"stargazers_count":1965,"watchers_count":1965,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":true,"forks_count":1311,"mirror_url":"git://git.apache.org/incubator-storm.git","open_issues_count":72,"forks":1311,"open_issues":72,"watchers":1965,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/storm/pulls/554"},"html":{"href":"https://github.com/apache/storm/pull/554"},"issue":{"href":"https://api.github.com/repos/apache/storm/issues/554"},"comments":{"href":"https://api.github.com/repos/apache/storm/issues/554/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/storm/pulls/554/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/storm/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/storm/pulls/554/commits"},"statuses":{"href":"https://api.github.com/repos/apache/storm/statuses/4b11f71b7e2fe45a62ad5a26d9d6bc32ebdec6df"}}}}