;; Licensed to the Apache Software Foundation (ASF) under one
;; or more contributor license agreements.  See the NOTICE file
;; distributed with this work for additional information
;; regarding copyright ownership.  The ASF licenses this file
;; to you under the Apache License, Version 2.0 (the
;; "License"); you may not use this file except in compliance
;; with the License.  You may obtain a copy of the License at
;;
;; http://www.apache.org/licenses/LICENSE-2.0
;;
;; Unless required by applicable law or agreed to in writing, software
;; distributed under the License is distributed on an "AS IS" BASIS,
;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;; See the License for the specific language governing permissions and
;; limitations under the License.

(ns backtype.storm.stats
  (:import [backtype.storm.generated Nimbus Nimbus$Processor Nimbus$Iface StormTopology ShellComponent
            NotAliveException AlreadyAliveException InvalidTopologyException GlobalStreamId
            ClusterSummary TopologyInfo TopologySummary ExecutorInfo ExecutorSummary ExecutorStats
            ExecutorSpecificStats SpoutStats BoltStats ErrorInfo
            SupervisorSummary CommonAggregateStats ComponentAggregateStats
            ComponentPageInfo ComponentType BoltAggregateStats
            ExecutorAggregateStats SpecificAggregateStats
            SpoutAggregateStats TopologyPageInfo TopologyStats])
  (:import [backtype.storm.utils Utils])
  (:import [backtype.storm.metric.internal MultiCountStatAndMetric MultiLatencyStatAndMetric])
  (:use [backtype.storm log util])
  (:use [clojure.math.numeric-tower :only [ceil]]))

(def TEN-MIN-IN-SECONDS (* 10 60))

(def COMMON-FIELDS [:emitted :transferred])
(defrecord CommonStats [^MultiCountStatAndMetric emitted
                        ^MultiCountStatAndMetric transferred
                        rate])

(def BOLT-FIELDS [:acked :failed :process-latencies :executed :execute-latencies])
;;acked and failed count individual tuples
(defrecord BoltExecutorStats [^CommonStats common
                              ^MultiCountStatAndMetric acked
                              ^MultiCountStatAndMetric failed
                              ^MultiLatencyStatAndMetric process-latencies
                              ^MultiCountStatAndMetric executed
                              ^MultiLatencyStatAndMetric execute-latencies])

(def SPOUT-FIELDS [:acked :failed :complete-latencies])
;;acked and failed count tuple completion
(defrecord SpoutExecutorStats [^CommonStats common
                               ^MultiCountStatAndMetric acked
                               ^MultiCountStatAndMetric failed
                               ^MultiLatencyStatAndMetric complete-latencies])

(def NUM-STAT-BUCKETS 20)

(defn- mk-common-stats
  [rate]
  (CommonStats.
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    rate))

(defn mk-bolt-stats
  [rate]
  (BoltExecutorStats.
    (mk-common-stats rate)
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    (MultiLatencyStatAndMetric. NUM-STAT-BUCKETS)
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    (MultiLatencyStatAndMetric. NUM-STAT-BUCKETS)))

(defn mk-spout-stats
  [rate]
  (SpoutExecutorStats.
    (mk-common-stats rate)
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    (MultiCountStatAndMetric. NUM-STAT-BUCKETS)
    (MultiLatencyStatAndMetric. NUM-STAT-BUCKETS)))

(defmacro stats-rate
  [stats]
  `(-> ~stats :common :rate))

(defmacro stats-emitted
  [stats]
  `(-> ~stats :common :emitted))

(defmacro stats-transferred
  [stats]
  `(-> ~stats :common :transferred))

(defmacro stats-executed
  [stats]
  `(:executed ~stats))

(defmacro stats-acked
  [stats]
  `(:acked ~stats))

(defmacro stats-failed
  [stats]
  `(:failed ~stats))

(defmacro stats-execute-latencies
  [stats]
  `(:execute-latencies ~stats))

(defmacro stats-process-latencies
  [stats]
  `(:process-latencies ~stats))

(defmacro stats-complete-latencies
  [stats]
  `(:complete-latencies ~stats))

(defn emitted-tuple!
  [stats stream]
  (.incBy ^MultiCountStatAndMetric (stats-emitted stats) ^Object stream ^long (stats-rate stats)))

(defn transferred-tuples!
  [stats stream amt]
  (.incBy ^MultiCountStatAndMetric (stats-transferred stats) ^Object stream ^long (* (stats-rate stats) amt)))

(defn bolt-execute-tuple!
  [^BoltExecutorStats stats component stream latency-ms]
  (let [key [component stream]
        ^MultiCountStatAndMetric executed (stats-executed stats)
        ^MultiLatencyStatAndMetric exec-lat (stats-execute-latencies stats)]
    (.incBy executed key (stats-rate stats))
    (.record exec-lat key latency-ms)))

(defn bolt-acked-tuple!
  [^BoltExecutorStats stats component stream latency-ms]
  (let [key [component stream]
        ^MultiCountStatAndMetric acked (stats-acked stats)
        ^MultiLatencyStatAndMetric process-lat (stats-process-latencies stats)]
    (.incBy acked key (stats-rate stats))
    (.record process-lat key latency-ms)))

(defn bolt-failed-tuple!
  [^BoltExecutorStats stats component stream latency-ms]
  (let [key [component stream]
        ^MultiCountStatAndMetric failed (stats-failed stats)]
    (.incBy failed key (stats-rate stats))))

(defn spout-acked-tuple!
  [^SpoutExecutorStats stats stream latency-ms]
  (.incBy ^MultiCountStatAndMetric (stats-acked stats) stream (stats-rate stats))
  (.record ^MultiLatencyStatAndMetric (stats-complete-latencies stats) stream latency-ms))

(defn spout-failed-tuple!
  [^SpoutExecutorStats stats stream latency-ms]
  (.incBy ^MultiCountStatAndMetric (stats-failed stats) stream (stats-rate stats)))

(defn- cleanup-stat! [stat]
  (.close stat))

(defn- cleanup-common-stats!
  [^CommonStats stats]
  (doseq [f COMMON-FIELDS]
    (cleanup-stat! (f stats))))

(defn cleanup-bolt-stats!
  [^BoltExecutorStats stats]
  (cleanup-common-stats! (:common stats))
  (doseq [f BOLT-FIELDS]
    (cleanup-stat! (f stats))))

(defn cleanup-spout-stats!
  [^SpoutExecutorStats stats]
  (cleanup-common-stats! (:common stats))
  (doseq [f SPOUT-FIELDS]
    (cleanup-stat! (f stats))))

(defn- value-stats
  [stats fields]
  (into {} (dofor [f fields]
                  [f (if (instance? MultiCountStatAndMetric (f stats))
                         (.getTimeCounts ^MultiCountStatAndMetric (f stats))
                         (.getTimeLatAvg ^MultiLatencyStatAndMetric (f stats)))])))

(defn- value-common-stats
  [^CommonStats stats]
  (merge
    (value-stats stats COMMON-FIELDS)
    {:rate (:rate stats)}))

(defn value-bolt-stats!
  [^BoltExecutorStats stats]
  (cleanup-bolt-stats! stats)
  (merge (value-common-stats (:common stats))
         (value-stats stats BOLT-FIELDS)
         {:type :bolt}))

(defn value-spout-stats!
  [^SpoutExecutorStats stats]
  (cleanup-spout-stats! stats)
  (merge (value-common-stats (:common stats))
         (value-stats stats SPOUT-FIELDS)
         {:type :spout}))

(defmulti render-stats! class-selector)

(defmethod render-stats! SpoutExecutorStats
  [stats]
  (value-spout-stats! stats))

(defmethod render-stats! BoltExecutorStats
  [stats]
  (value-bolt-stats! stats))

(defmulti thriftify-specific-stats :type)
(defmulti clojurify-specific-stats class-selector)

(defn window-set-converter
  ([stats key-fn first-key-fun]
    (into {}
      (for [[k v] stats]
        ;apply the first-key-fun only to first key.
        [(first-key-fun k)
         (into {} (for [[k2 v2] v]
                    [(key-fn k2) v2]))])))
  ([stats first-key-fun]
    (window-set-converter stats identity first-key-fun)))

(defn to-global-stream-id
  [[component stream]]
  (GlobalStreamId. component stream))

(defn from-global-stream-id [global-stream-id]
  [(.get_componentId global-stream-id) (.get_streamId global-stream-id)])

(defmethod clojurify-specific-stats BoltStats [^BoltStats stats]
  [(window-set-converter (.get_acked stats) from-global-stream-id identity)
   (window-set-converter (.get_failed stats) from-global-stream-id identity)
   (window-set-converter (.get_process_ms_avg stats) from-global-stream-id identity)
   (window-set-converter (.get_executed stats) from-global-stream-id identity)
   (window-set-converter (.get_execute_ms_avg stats) from-global-stream-id identity)])

(defmethod clojurify-specific-stats SpoutStats [^SpoutStats stats]
  [(.get_acked stats)
   (.get_failed stats)
   (.get_complete_ms_avg stats)])


(defn clojurify-executor-stats
  [^ExecutorStats stats]
  (let [ specific-stats (.get_specific stats)
         is_bolt? (.is_set_bolt specific-stats)
         specific-stats (if is_bolt? (.get_bolt specific-stats) (.get_spout specific-stats))
         specific-stats (clojurify-specific-stats specific-stats)
         common-stats (CommonStats. (.get_emitted stats)
                                    (.get_transferred stats)
                                    (.get_rate stats))]
    (if is_bolt?
      ; worker heart beat does not store the BoltExecutorStats or SpoutExecutorStats , instead it stores the result returned by render-stats!
      ; which flattens the BoltExecutorStats/SpoutExecutorStats by extracting values from all atoms and merging all values inside :common to top
      ;level map we are pretty much doing the same here.
      (dissoc (merge common-stats {:type :bolt}  (apply ->BoltExecutorStats (into [nil] specific-stats))) :common)
      (dissoc (merge common-stats {:type :spout} (apply ->SpoutExecutorStats (into [nil] specific-stats))) :common)
      )))

(defmethod thriftify-specific-stats :bolt
  [stats]
  (ExecutorSpecificStats/bolt
    (BoltStats.
      (window-set-converter (:acked stats) to-global-stream-id str)
      (window-set-converter (:failed stats) to-global-stream-id str)
      (window-set-converter (:process-latencies stats) to-global-stream-id str)
      (window-set-converter (:executed stats) to-global-stream-id str)
      (window-set-converter (:execute-latencies stats) to-global-stream-id str))))

(defmethod thriftify-specific-stats :spout
  [stats]
  (ExecutorSpecificStats/spout
    (SpoutStats. (window-set-converter (:acked stats) str)
      (window-set-converter (:failed stats) str)
      (window-set-converter (:complete-latencies stats) str))))

(defn thriftify-executor-stats
  [stats]
  (let [specific-stats (thriftify-specific-stats stats)
        rate (:rate stats)]
    (ExecutorStats. (window-set-converter (:emitted stats) str)
      (window-set-converter (:transferred stats) str)
      specific-stats
      rate)))

(defn valid-number?
  "Returns true if x is a number that is not NaN or Infinity, false otherwise"
  [x]
  (and (number? x)
       (not (Double/isNaN x))
       (not (Double/isInfinite x))))

(defn apply-default
  [f defaulting-fn & args]
  (apply f (map defaulting-fn args)))

(defn apply-or-0
  [f & args]
  (apply apply-default
         f
         #(if (valid-number? %) % 0)
         args))

(defn sum-or-0
  [& args]
  (apply apply-or-0 + args))

(defn product-or-0
  [& args]
  (apply apply-or-0 * args))

(defn max-or-0
  [& args]
  (apply apply-or-0 max args))

(defn- agg-bolt-lat-and-count
  "Aggregates number executed, process latency, and execute latency across all
  streams."
  [idk->exec-avg idk->proc-avg idk->num-executed]
  (letfn [(weight-avg [[id avg]]
            (let [num-e (get idk->num-executed id)]
              (product-or-0 avg num-e)))]
    {:executeLatencyTotal (sum (map weight-avg idk->exec-avg))
     :processLatencyTotal (sum (map weight-avg idk->proc-avg))
     :executed (sum (vals idk->num-executed))}))

(defn- agg-spout-lat-and-count
  "Aggregates number acked and complete latencies across all streams."
  [sid->comp-avg sid->num-acked]
  (letfn [(weight-avg [[id avg]]
            (product-or-0 avg (get sid->num-acked id)))]
    {:completeLatencyTotal (sum (map weight-avg sid->comp-avg))
     :acked (sum (vals sid->num-acked))}))

(defn add-pairs
  ([] [0 0])
  ([[a1 a2] [b1 b2]]
   [(+ a1 b1) (+ a2 b2)]))

(defn mk-include-sys-fn
  [include-sys?]
  (if include-sys?
    (fn [_] true)
    (fn [stream] (and (string? stream) (not (Utils/isSystemId stream))))))

(defn mk-include-sys-filter
  "Returns a function that includes or excludes map entries whose keys are
  system ids."
  [include-sys?]
  (if include-sys?
    identity
    (partial filter-key (mk-include-sys-fn false))))

(defn- agg-bolt-streams-lat-and-count
  "Aggregates number executed and process & execute latencies."
  [idk->exec-avg idk->proc-avg idk->executed]
  (letfn [(weight-avg [id avg]
            (let [num-e (idk->executed id)]
              (product-or-0 avg num-e)))]
    (into {}
      (for [k (keys idk->exec-avg)]
        [k {:executeLatencyTotal (weight-avg k (get idk->exec-avg k))
            :processLatencyTotal (weight-avg k (get idk->proc-avg k))
            :executed (idk->executed k)}]))))

(defn- agg-spout-streams-lat-and-count
  "Aggregates number acked and complete latencies."
  [idk->comp-avg idk->acked]
  (letfn [(weight-avg [id avg]
            (let [num-e (get idk->acked id)]
              (product-or-0 avg num-e)))]
    (into {}
      (for [k (keys idk->comp-avg)]
        [k {:completeLatencyTotal (weight-avg k (get idk->comp-avg k))
            :acked (get idk->acked k)}]))))

(defn swap-map-order
  "For a nested map, rearrange data such that the top-level keys become the
  nested map's keys and vice versa.
  Example:
  {:a {:X :banana, :Y :pear}, :b {:X :apple, :Y :orange}}
  -> {:Y {:a :pear, :b :orange}, :X {:a :banana, :b :apple}}"
  [m]
  (apply merge-with
         merge
         (map (fn [[k v]]
                (into {}
                      (for [[k2 v2] v]
                        [k2 {k v2}])))
              m)))

(defn- compute-agg-capacity
  "Computes the capacity metric for one executor given its heartbeat data and
  uptime."
  [m uptime]
  (when uptime
    (->>
      ;; For each stream, create weighted averages and counts.
      (merge-with (fn weighted-avg+count-fn
                    [avg cnt]
                    [(* avg cnt) cnt])
                  (get (:execute-latencies m) (str TEN-MIN-IN-SECONDS))
                  (get (:executed m) (str TEN-MIN-IN-SECONDS)))
      vals ;; Ignore the stream ids.
      (reduce add-pairs
              [0. 0]) ;; Combine weighted averages and counts.
      ((fn [[weighted-avg cnt]]
        (div weighted-avg (* 1000 (min uptime TEN-MIN-IN-SECONDS))))))))

(defn agg-pre-merge-comp-page-bolt
  [{exec-id :exec-id
    host :host
    port :port
    uptime :uptime
    comp-id :comp-id
    num-tasks :num-tasks
    statk->w->sid->num :stats}
   window
   include-sys?]
  (let [str-key (partial map-key str)
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    {:executor-id exec-id,
     :host host,
     :port port,
     :uptime uptime,
     :num-executors 1,
     :num-tasks num-tasks,
     :capacity (compute-agg-capacity statk->w->sid->num uptime)
     :cid+sid->input-stats
     (merge-with
       merge
       (swap-map-order
         {:acked (-> statk->w->sid->num
                     :acked
                     str-key
                     (get window))
          :failed (-> statk->w->sid->num
                      :failed
                      str-key
                      (get window))})
       (agg-bolt-streams-lat-and-count (-> statk->w->sid->num
                                           :execute-latencies
                                           str-key
                                           (get window))
                                       (-> statk->w->sid->num
                                           :process-latencies
                                           str-key
                                           (get window))
                                       (-> statk->w->sid->num
                                           :executed
                                           str-key
                                           (get window)))),
     :sid->output-stats
     (swap-map-order
       {:emitted (-> statk->w->sid->num
                     :emitted
                     str-key
                     (get window)
                     handle-sys-components-fn)
        :transferred (-> statk->w->sid->num
                         :transferred
                         str-key
                         (get window)
                         handle-sys-components-fn)})}))

(defn agg-pre-merge-comp-page-spout
  [{exec-id :exec-id
    host :host
    port :port
    uptime :uptime
    comp-id :comp-id
    num-tasks :num-tasks
    statk->w->sid->num :stats}
   window
   include-sys?]
  (let [str-key (partial map-key str)
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    {:executor-id exec-id,
     :host host,
     :port port,
     :uptime uptime,
     :num-executors 1,
     :num-tasks num-tasks,
     :sid->output-stats
     (merge-with
       merge
       (agg-spout-streams-lat-and-count (-> statk->w->sid->num
                                            :complete-latencies
                                            str-key
                                            (get window))
                                        (-> statk->w->sid->num
                                            :acked
                                            str-key
                                            (get window)))
       (swap-map-order
         {:acked (-> statk->w->sid->num
                     :acked
                     str-key
                     (get window))
          :failed (-> statk->w->sid->num
                      :failed
                      str-key
                      (get window))
          :emitted (-> statk->w->sid->num
                       :emitted
                       str-key
                       (get window)
                       handle-sys-components-fn)
          :transferred (-> statk->w->sid->num
                           :transferred
                           str-key
                           (get window)
                           handle-sys-components-fn)}))}))

(defn agg-pre-merge-topo-page-bolt
  [{comp-id :comp-id
    num-tasks :num-tasks
    statk->w->sid->num :stats
    uptime :uptime}
   window
   include-sys?]
  (let [str-key (partial map-key str)
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    {comp-id
     (merge
       (agg-bolt-lat-and-count (-> statk->w->sid->num
                                   :execute-latencies
                                   str-key
                                   (get window))
                               (-> statk->w->sid->num
                                   :process-latencies
                                   str-key
                                   (get window))
                               (-> statk->w->sid->num
                                   :executed
                                   str-key
                                   (get window)))
       {:num-executors 1
        :num-tasks num-tasks
        :emitted (-> statk->w->sid->num
                     :emitted
                     str-key
                     (get window)
                     handle-sys-components-fn
                     vals
                     sum)
        :transferred (-> statk->w->sid->num
                         :transferred
                         str-key
                         (get window)
                         handle-sys-components-fn
                         vals
                         sum)
        :capacity (compute-agg-capacity statk->w->sid->num uptime)
        :acked (-> statk->w->sid->num
                   :acked
                   str-key
                   (get window)
                   vals
                   sum)
        :failed (-> statk->w->sid->num
                    :failed
                    str-key
                    (get window)
                    vals
                    sum)})}))

(defn agg-pre-merge-topo-page-spout
  [{comp-id :comp-id
    num-tasks :num-tasks
    statk->w->sid->num :stats}
   window
   include-sys?]
  (let [str-key (partial map-key str)
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    {comp-id
     (merge
       (agg-spout-lat-and-count (-> statk->w->sid->num
                                    :complete-latencies
                                    str-key
                                    (get window))
                                (-> statk->w->sid->num
                                    :acked
                                    str-key
                                    (get window)))
       {:num-executors 1
        :num-tasks num-tasks
        :emitted (-> statk->w->sid->num
                     :emitted
                     str-key
                     (get window)
                     handle-sys-components-fn
                     vals
                     sum)
        :transferred (-> statk->w->sid->num
                         :transferred
                         str-key
                         (get window)
                         handle-sys-components-fn
                         vals
                         sum)
        :failed (-> statk->w->sid->num
                    :failed
                    str-key
                    (get window)
                    vals
                    sum)})}))

(defn merge-agg-comp-stats-comp-page-bolt
  [{acc-in :cid+sid->input-stats
    acc-out :sid->output-stats
    :as acc-bolt-stats}
   {bolt-in :cid+sid->input-stats
    bolt-out :sid->output-stats
    :as bolt-stats}]
  {:num-executors (inc (or (:num-executors acc-bolt-stats) 0)),
   :num-tasks (sum-or-0 (:num-tasks acc-bolt-stats) (:num-tasks bolt-stats)),
   :sid->output-stats (merge-with (partial merge-with sum-or-0)
                                  acc-out
                                  bolt-out),
   :cid+sid->input-stats (merge-with (partial merge-with sum-or-0)
                                     acc-in
                                     bolt-in),
   :executor-stats
   (let [sum-streams (fn [m k] (->> m vals (map k) (apply sum-or-0)))
         executed (sum-streams bolt-in :executed)]
     (conj (:executor-stats acc-bolt-stats)
           (merge
             (select-keys bolt-stats
                          [:executor-id :uptime :host :port :capacity])
             {:emitted (sum-streams bolt-out :emitted)
              :transferred (sum-streams bolt-out :transferred)
              :acked (sum-streams bolt-in :acked)
              :failed (sum-streams bolt-in :failed)
              :executed executed}
             (->>
               (if (and executed (pos? executed))
                 [(div (sum-streams bolt-in :executeLatencyTotal) executed)
                  (div (sum-streams bolt-in :processLatencyTotal) executed)]
                 [nil nil])
               (mapcat vector [:execute-latency :process-latency])
               (apply assoc {})))))})

(defn merge-agg-comp-stats-comp-page-spout
  [{acc-out :sid->output-stats
    :as acc-spout-stats}
   {spout-out :sid->output-stats
    :as spout-stats}]
  {:num-executors (inc (or (:num-executors acc-spout-stats) 0)),
   :num-tasks (sum-or-0 (:num-tasks acc-spout-stats) (:num-tasks spout-stats)),
   :sid->output-stats (merge-with (partial merge-with sum-or-0)
                                  acc-out
                                  spout-out),
   :executor-stats
   (let [sum-streams (fn [m k] (->> m vals (map k) (apply sum-or-0)))
         acked (sum-streams spout-out :acked)]
     (conj (:executor-stats acc-spout-stats)
           (merge
             (select-keys spout-stats [:executor-id :uptime :host :port])
             {:emitted (sum-streams spout-out :emitted)
              :transferred (sum-streams spout-out :transferred)
              :acked acked
              :failed (sum-streams spout-out :failed)}
             {:complete-latency (if (and acked (pos? acked))
                                  (div (sum-streams spout-out
                                                    :completeLatencyTotal)
                                       acked)
                                  nil)})))})

(defn merge-agg-comp-stats-topo-page-bolt
  [acc-bolt-stats bolt-stats]
  {:num-executors (inc (or (:num-executors acc-bolt-stats) 0))
   :num-tasks (sum-or-0 (:num-tasks acc-bolt-stats) (:num-tasks bolt-stats))
   :emitted (sum-or-0 (:emitted acc-bolt-stats) (:emitted bolt-stats))
   :transferred (sum-or-0 (:transferred acc-bolt-stats)
                          (:transferred bolt-stats))
   :capacity (max-or-0 (:capacity acc-bolt-stats) (:capacity bolt-stats))
   ;; We sum average latency totals here to avoid dividing at each step.
   ;; Compute the average latencies by dividing the total by the count.
   :executeLatencyTotal (sum-or-0 (:executeLatencyTotal acc-bolt-stats)
                                  (:executeLatencyTotal bolt-stats))
   :processLatencyTotal (sum-or-0 (:processLatencyTotal acc-bolt-stats)
                                  (:processLatencyTotal bolt-stats))
   :executed (sum-or-0 (:executed acc-bolt-stats) (:executed bolt-stats))
   :acked (sum-or-0 (:acked acc-bolt-stats) (:acked bolt-stats))
   :failed (sum-or-0 (:failed acc-bolt-stats) (:failed bolt-stats))})

(defn merge-agg-comp-stats-topo-page-spout
  [acc-spout-stats spout-stats]
  {:num-executors (inc (or (:num-executors acc-spout-stats) 0))
   :num-tasks (sum-or-0 (:num-tasks acc-spout-stats) (:num-tasks spout-stats))
   :emitted (sum-or-0 (:emitted acc-spout-stats) (:emitted spout-stats))
   :transferred (sum-or-0 (:transferred acc-spout-stats) (:transferred spout-stats))
   ;; We sum average latency totals here to avoid dividing at each step.
   ;; Compute the average latencies by dividing the total by the count.
   :completeLatencyTotal (sum-or-0 (:completeLatencyTotal acc-spout-stats)
                            (:completeLatencyTotal spout-stats))
   :acked (sum-or-0 (:acked acc-spout-stats) (:acked spout-stats))
   :failed (sum-or-0 (:failed acc-spout-stats) (:failed spout-stats))})

(defn aggregate-count-streams
  [stats]
  (->> stats
       (map-val #(reduce + (vals %)))))

(defn- agg-topo-exec-stats*
  "A helper function that does the common work to aggregate stats of one
  executor with the given map for the topology page."
  [window
   include-sys?
   {:keys [workers-set
           bolt-id->stats
           spout-id->stats
           window->emitted
           window->transferred
           window->comp-lat-wgt-avg
           window->acked
           window->failed] :as acc-stats}
   {:keys [stats] :as new-data}
   pre-merge-fn
   merge-fn
   comp-key]
  (let [cid->statk->num (pre-merge-fn new-data window include-sys?)
        {w->compLatWgtAvg :completeLatencyTotal
         w->acked :acked}
          (if (:complete-latencies stats)
            (swap-map-order
              (into {}
                    (for [w (keys (:acked stats))]
                         [w (agg-spout-lat-and-count
                              (get (:complete-latencies stats) w)
                              (get (:acked stats) w))])))
            {:completeLatencyTotal nil
             :acks (aggregate-count-streams (:acked stats))})
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    (assoc {:workers-set (conj workers-set
                               [(:host new-data) (:port new-data)])
            :bolt-id->stats bolt-id->stats
            :spout-id->stats spout-id->stats
            :window->emitted (->> (:emitted stats)
                                  (map-val handle-sys-components-fn)
                                  aggregate-count-streams
                                  (merge-with + window->emitted))
            :window->transferred (->> (:transferred stats)
                                      (map-val handle-sys-components-fn)
                                      aggregate-count-streams
                                      (merge-with + window->transferred))
            :window->comp-lat-wgt-avg (merge-with +
                                                  window->comp-lat-wgt-avg
                                                  w->compLatWgtAvg)
            :window->acked (if (= :spout (:type stats))
                             (merge-with + window->acked w->acked)
                             window->acked)
            :window->failed (if (= :spout (:type stats))
                              (->> (:failed stats)
                                   aggregate-count-streams
                                   (merge-with + window->failed))
                              window->failed)}
           comp-key (merge-with merge-fn
                                (acc-stats comp-key)
                                cid->statk->num)
           :type (:type stats))))

(defmulti agg-topo-exec-stats
  "Combines the aggregate stats of one executor with the given map, selecting
  the appropriate window and including system components as specified."
  (fn dispatch-fn [& args] (:type (last args))))

(defmethod agg-topo-exec-stats :bolt
  [window include-sys? acc-stats new-data]
  (agg-topo-exec-stats* window
                        include-sys?
                        acc-stats
                        new-data
                        agg-pre-merge-topo-page-bolt
                        merge-agg-comp-stats-topo-page-bolt
                        :bolt-id->stats))

(defmethod agg-topo-exec-stats :spout
  [window include-sys? acc-stats new-data]
  (agg-topo-exec-stats* window
                        include-sys?
                        acc-stats
                        new-data
                        agg-pre-merge-topo-page-spout
                        merge-agg-comp-stats-topo-page-spout
                        :spout-id->stats))

(defmethod agg-topo-exec-stats :default [_ _ acc-stats _] acc-stats)

(defn get-last-error
  [storm-cluster-state storm-id component-id]
  (if-let [e (.last-error storm-cluster-state storm-id component-id)]
    (ErrorInfo. (:error e) (:time-secs e))))

(defn component-type
  "Returns the component type (either :bolt or :spout) for a given
  topology and component id. Returns nil if not found."
  [^StormTopology topology id]
  (let [bolts (.get_bolts topology)
        spouts (.get_spouts topology)]
    (cond
      (Utils/isSystemId id) :bolt
      (.containsKey bolts id) :bolt
      (.containsKey spouts id) :spout)))

(defn extract-nodeinfos-from-hb-for-comp
  ([exec->host+port task->component include-sys? comp-id]
   (distinct (for [[[start end :as executor] [host port]] exec->host+port
         :let [id (task->component start)]
         :when (and (or (nil? comp-id) (= comp-id id))
                 (or include-sys? (not (Utils/isSystemId id))))]
     {:host host
      :port port}))))

(defn extract-data-from-hb
  ([exec->host+port task->component beats include-sys? topology comp-id]
   (for [[[start end :as executor] [host port]] exec->host+port
         :let [beat (beats executor)
               id (task->component start)]
         :when (and (or (nil? comp-id) (= comp-id id))
                    (or include-sys? (not (Utils/isSystemId id))))]
     {:exec-id executor
      :comp-id id
      :num-tasks (count (range start (inc end)))
      :host host
      :port port
      :uptime (:uptime beat)
      :stats (:stats beat)
      :type (or (:type (:stats beat))
                (component-type topology id))}))
  ([exec->host+port task->component beats include-sys? topology]
    (extract-data-from-hb exec->host+port
                          task->component
                          beats
                          include-sys?
                          topology
                          nil)))

(defn aggregate-topo-stats
  [window include-sys? data]
  (let [init-val {:workers-set #{}
                  :bolt-id->stats {}
                  :spout-id->stats {}
                  :window->emitted {}
                  :window->transferred {}
                  :window->comp-lat-wgt-avg {}
                  :window->acked {}
                  :window->failed {}}
        reducer-fn (partial agg-topo-exec-stats
                            window
                            include-sys?)]
    (reduce reducer-fn init-val data)))

(defn- compute-weighted-averages-per-window
  [acc-data wgt-avg-key divisor-key]
  (into {} (for [[window wgt-avg] (wgt-avg-key acc-data)
                 :let [divisor ((divisor-key acc-data) window)]
                 :when (and divisor (pos? divisor))]
             [(str window) (div wgt-avg divisor)])))

(defn- post-aggregate-topo-stats
  [task->component exec->node+port last-err-fn acc-data]
  {:num-tasks (count task->component)
   :num-workers (count (:workers-set acc-data))
   :num-executors (count exec->node+port)
   :bolt-id->stats
     (into {} (for [[id m] (:bolt-id->stats acc-data)
                    :let [executed (:executed m)]]
                     [id (-> m
                             (assoc :execute-latency
                                    (if (and executed (pos? executed))
                                      (div (or (:executeLatencyTotal m) 0)
                                           executed)
                                      0)
                                    :process-latency
                                    (if (and executed (pos? executed))
                                      (div (or (:processLatencyTotal m) 0)
                                           executed)
                                      0))
                             (dissoc :executeLatencyTotal
                                     :processLatencyTotal)
                             (assoc :lastError (last-err-fn id)))]))
   :spout-id->stats
     (into {} (for [[id m] (:spout-id->stats acc-data)
                    :let [acked (:acked m)]]
                    [id (-> m
                            (assoc :complete-latency
                                   (if (and acked (pos? acked))
                                     (div (:completeLatencyTotal m)
                                          (:acked m))
                                     0))
                            (dissoc :completeLatencyTotal)
                            (assoc :lastError (last-err-fn id)))]))
   :window->emitted (map-key str (:window->emitted acc-data))
   :window->transferred (map-key str (:window->transferred acc-data))
   :window->complete-latency
     (compute-weighted-averages-per-window acc-data
                                           :window->comp-lat-wgt-avg
                                           :window->acked)
   :window->acked (map-key str (:window->acked acc-data))
   :window->failed (map-key str (:window->failed acc-data))})

(defn- thriftify-common-agg-stats
  [^ComponentAggregateStats s
   {:keys [num-tasks
           emitted
           transferred
           acked
           failed
           num-executors] :as statk->num}]
  (let [cas (CommonAggregateStats.)]
    (and num-executors (.set_num_executors cas num-executors))
    (and num-tasks (.set_num_tasks cas num-tasks))
    (and emitted (.set_emitted cas emitted))
    (and transferred (.set_transferred cas transferred))
    (and acked (.set_acked cas acked))
    (and failed (.set_failed cas failed))
    (.set_common_stats s cas)))

(defn thriftify-bolt-agg-stats
  [statk->num]
  (let [{:keys [lastError
                execute-latency
                process-latency
                executed
                capacity]} statk->num
        s (ComponentAggregateStats.)]
    (.set_type s ComponentType/BOLT)
    (and lastError (.set_last_error s lastError))
    (thriftify-common-agg-stats s statk->num)
    (.set_specific_stats s
      (SpecificAggregateStats/bolt
        (let [bas (BoltAggregateStats.)]
          (and execute-latency (.set_execute_latency_ms bas execute-latency))
          (and process-latency (.set_process_latency_ms bas process-latency))
          (and executed (.set_executed bas executed))
          (and capacity (.set_capacity bas capacity))
          bas)))
    s))

(defn thriftify-spout-agg-stats
  [statk->num]
  (let [{:keys [lastError
                complete-latency]} statk->num
        s (ComponentAggregateStats.)]
    (.set_type s ComponentType/SPOUT)
    (and lastError (.set_last_error s lastError))
    (thriftify-common-agg-stats s statk->num)
    (.set_specific_stats s
      (SpecificAggregateStats/spout
        (let [sas (SpoutAggregateStats.)]
          (and complete-latency (.set_complete_latency_ms sas complete-latency))
          sas)))
    s))

(defn thriftify-topo-page-data
  [topology-id data]
  (let [{:keys [num-tasks
                num-workers
                num-executors
                spout-id->stats
                bolt-id->stats
                window->emitted
                window->transferred
                window->complete-latency
                window->acked
                window->failed]} data
        spout-agg-stats (into {}
                              (for [[id m] spout-id->stats
                                    :let [m (assoc m :type :spout)]]
                                [id
                                 (thriftify-spout-agg-stats m)]))
        bolt-agg-stats (into {}
                             (for [[id m] bolt-id->stats
                                   :let [m (assoc m :type :bolt)]]
                              [id
                               (thriftify-bolt-agg-stats m)]))
        topology-stats (doto (TopologyStats.)
                         (.set_window_to_emitted window->emitted)
                         (.set_window_to_transferred window->transferred)
                         (.set_window_to_complete_latencies_ms
                           window->complete-latency)
                         (.set_window_to_acked window->acked)
                         (.set_window_to_failed window->failed))
      topo-page-info (doto (TopologyPageInfo. topology-id)
                       (.set_num_tasks num-tasks)
                       (.set_num_workers num-workers)
                       (.set_num_executors num-executors)
                       (.set_id_to_spout_agg_stats spout-agg-stats)
                       (.set_id_to_bolt_agg_stats bolt-agg-stats)
                       (.set_topology_stats topology-stats))]
    topo-page-info))

(defn agg-topo-execs-stats
  "Aggregate various executor statistics for a topology from the given
  heartbeats."
  [topology-id
   exec->node+port
   task->component
   beats
   topology
   window
   include-sys?
   last-err-fn]
  (->> ;; This iterates over each executor one time, because of lazy evaluation.
    (extract-data-from-hb exec->node+port
                          task->component
                          beats
                          include-sys?
                          topology)
    (aggregate-topo-stats window include-sys?)
    (post-aggregate-topo-stats task->component exec->node+port last-err-fn)
    (thriftify-topo-page-data topology-id)))

(defn- agg-bolt-exec-win-stats
  "A helper function that aggregates windowed stats from one bolt executor."
  [acc-stats new-stats include-sys?]
  (let [{w->execLatWgtAvg :executeLatencyTotal
         w->procLatWgtAvg :processLatencyTotal
         w->executed :executed}
          (swap-map-order
            (into {} (for [w (keys (:executed new-stats))]
                       [w (agg-bolt-lat-and-count
                            (get (:execute-latencies new-stats) w)
                            (get (:process-latencies new-stats) w)
                            (get (:executed new-stats) w))])))
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    {:window->emitted (->> (:emitted new-stats)
                           (map-val handle-sys-components-fn)
                           aggregate-count-streams
                           (merge-with + (:window->emitted acc-stats)))
     :window->transferred (->> (:transferred new-stats)
                               (map-val handle-sys-components-fn)
                               aggregate-count-streams
                               (merge-with + (:window->transferred acc-stats)))
     :window->exec-lat-wgt-avg (merge-with +
                                           (:window->exec-lat-wgt-avg acc-stats)
                                           w->execLatWgtAvg)
     :window->proc-lat-wgt-avg (merge-with +
                                           (:window->proc-lat-wgt-avg acc-stats)
                                           w->procLatWgtAvg)
     :window->executed (merge-with + (:window->executed acc-stats) w->executed)
     :window->acked (->> (:acked new-stats)
                         aggregate-count-streams
                         (merge-with + (:window->acked acc-stats)))
     :window->failed (->> (:failed new-stats)
                          aggregate-count-streams
                          (merge-with + (:window->failed acc-stats)))}))

(defn- agg-spout-exec-win-stats
  "A helper function that aggregates windowed stats from one spout executor."
  [acc-stats new-stats include-sys?]
  (let [{w->compLatWgtAvg :completeLatencyTotal
         w->acked :acked}
          (swap-map-order
            (into {} (for [w (keys (:acked new-stats))]
                       [w (agg-spout-lat-and-count
                            (get (:complete-latencies new-stats) w)
                            (get (:acked new-stats) w))])))
        handle-sys-components-fn (mk-include-sys-filter include-sys?)]
    {:window->emitted (->> (:emitted new-stats)
                           (map-val handle-sys-components-fn)
                           aggregate-count-streams
                           (merge-with + (:window->emitted acc-stats)))
     :window->transferred (->> (:transferred new-stats)
                               (map-val handle-sys-components-fn)
                               aggregate-count-streams
                               (merge-with + (:window->transferred acc-stats)))
     :window->comp-lat-wgt-avg (merge-with +
                                           (:window->comp-lat-wgt-avg acc-stats)
                                           w->compLatWgtAvg)
     :window->acked (->> (:acked new-stats)
                         aggregate-count-streams
                         (merge-with + (:window->acked acc-stats)))
     :window->failed (->> (:failed new-stats)
                          aggregate-count-streams
                          (merge-with + (:window->failed acc-stats)))}))

(defmulti agg-comp-exec-stats
  "Combines the aggregate stats of one executor with the given map, selecting
  the appropriate window and including system components as specified."
  (fn dispatch-fn [_ _ init-val _] (:type init-val)))

(defmethod agg-comp-exec-stats :bolt
  [window include-sys? acc-stats new-data]
  (assoc (agg-bolt-exec-win-stats acc-stats (:stats new-data) include-sys?)
         :stats (merge-agg-comp-stats-comp-page-bolt
                  (:stats acc-stats)
                  (agg-pre-merge-comp-page-bolt new-data window include-sys?))
         :type :bolt))

(defmethod agg-comp-exec-stats :spout
  [window include-sys? acc-stats new-data]
  (assoc (agg-spout-exec-win-stats acc-stats (:stats new-data) include-sys?)
         :stats (merge-agg-comp-stats-comp-page-spout
                  (:stats acc-stats)
                  (agg-pre-merge-comp-page-spout new-data window include-sys?))
         :type :spout))

(defn- aggregate-comp-stats*
  [window include-sys? data init-val]
  (-> (partial agg-comp-exec-stats
               window
               include-sys?)
      (reduce init-val data)))

(defmulti aggregate-comp-stats
  (fn dispatch-fn [& args] (-> args last first :type)))

(defmethod aggregate-comp-stats :bolt
  [& args]
  (let [init-val {:type :bolt
                  :cid+sid->input-stats {}
                  :sid->output-stats {}
                  :executor-stats []
                  :window->emitted {}
                  :window->transferred {}
                  :window->exec-lat-wgt-avg {}
                  :window->executed {}
                  :window->proc-lat-wgt-avg {}
                  :window->acked {}
                  :window->failed {}}]
    (apply aggregate-comp-stats* (concat args (list init-val)))))

(defmethod aggregate-comp-stats :spout
  [& args]
  (let [init-val {:type :spout
                  :sid->output-stats {}
                  :executor-stats []
                  :window->emitted {}
                  :window->transferred {}
                  :window->comp-lat-wgt-avg {}
                  :window->acked {}
                  :window->failed {}}]
    (apply aggregate-comp-stats* (concat args (list init-val)))))

(defmethod aggregate-comp-stats :default [& _] {})

(defmulti post-aggregate-comp-stats
  (fn [_ _ data] (:type data)))

(defmethod post-aggregate-comp-stats :bolt
  [task->component
   exec->host+port
   {{i-stats :cid+sid->input-stats
     o-stats :sid->output-stats
     num-tasks :num-tasks
     num-executors :num-executors} :stats
    comp-type :type :as acc-data}]
  {:type comp-type
   :num-tasks num-tasks
   :num-executors num-executors
   :cid+sid->input-stats
   (->> i-stats
        (map-val (fn [m]
                     (let [executed (:executed m)
                           lats (if (and executed (pos? executed))
                                  {:execute-latency
                                   (div (or (:executeLatencyTotal m) 0)
                                        executed)
                                   :process-latency
                                   (div (or (:processLatencyTotal m) 0)
                                        executed)}
                                  {:execute-latency 0
                                   :process-latency 0})]
                       (-> m (merge lats) (dissoc :executeLatencyTotal
                                                  :processLatencyTotal))))))
   :sid->output-stats o-stats
   :executor-stats (:executor-stats (:stats acc-data))
   :window->emitted (map-key str (:window->emitted acc-data))
   :window->transferred (map-key str (:window->transferred acc-data))
   :window->execute-latency
     (compute-weighted-averages-per-window acc-data
                                           :window->exec-lat-wgt-avg
                                           :window->executed)
   :window->executed (map-key str (:window->executed acc-data))
   :window->process-latency
     (compute-weighted-averages-per-window acc-data
                                           :window->proc-lat-wgt-avg
                                           :window->executed)
   :window->acked (map-key str (:window->acked acc-data))
   :window->failed (map-key str (:window->failed acc-data))})

(defmethod post-aggregate-comp-stats :spout
  [task->component
   exec->host+port
   {{o-stats :sid->output-stats
     num-tasks :num-tasks
     num-executors :num-executors} :stats
    comp-type :type :as acc-data}]
  {:type comp-type
   :num-tasks num-tasks
   :num-executors num-executors
   :sid->output-stats
   (->> o-stats
        (map-val (fn [m]
                     (let [acked (:acked m)
                           lat (if (and acked (pos? acked))
                                 {:complete-latency
                                  (div (or (:completeLatencyTotal m) 0) acked)}
                                 {:complete-latency 0})]
                       (-> m (merge lat) (dissoc :completeLatencyTotal))))))
   :executor-stats (:executor-stats (:stats acc-data))
   :window->emitted (map-key str (:window->emitted acc-data))
   :window->transferred (map-key str (:window->transferred acc-data))
   :window->complete-latency
     (compute-weighted-averages-per-window acc-data
                                           :window->comp-lat-wgt-avg
                                           :window->acked)
   :window->acked (map-key str (:window->acked acc-data))
   :window->failed (map-key str (:window->failed acc-data))})

(defmethod post-aggregate-comp-stats :default [& _] {})

(defn thriftify-exec-agg-stats
  [comp-id comp-type {:keys [executor-id host port uptime] :as stats}]
  (doto (ExecutorAggregateStats.)
    (.set_exec_summary (ExecutorSummary. (apply #(ExecutorInfo. %1 %2)
                                                executor-id)
                                         comp-id
                                         host
                                         port
                                         (or uptime 0)))
    (.set_stats ((condp = comp-type
                   :bolt thriftify-bolt-agg-stats
                   :spout thriftify-spout-agg-stats) stats))))

(defn- thriftify-bolt-input-stats
  [cid+sid->input-stats]
  (into {} (for [[cid+sid input-stats] cid+sid->input-stats]
             [(to-global-stream-id cid+sid)
              (thriftify-bolt-agg-stats input-stats)])))

(defn- thriftify-bolt-output-stats
  [sid->output-stats]
  (map-val thriftify-bolt-agg-stats sid->output-stats))

(defn- thriftify-spout-output-stats
  [sid->output-stats]
  (map-val thriftify-spout-agg-stats sid->output-stats))

(defn thriftify-comp-page-data
  [topo-id topology comp-id data]
  (let [w->stats (swap-map-order
                   (merge
                     {:emitted (:window->emitted data)
                      :transferred (:window->transferred data)
                      :acked (:window->acked data)
                      :failed (:window->failed data)}
                     (condp = (:type data)
                       :bolt {:execute-latency (:window->execute-latency data)
                              :process-latency (:window->process-latency data)
                              :executed (:window->executed data)}
                       :spout {:complete-latency
                               (:window->complete-latency data)}
                       {}))) ; default
        [compType exec-stats w->stats gsid->input-stats sid->output-stats]
          (condp = (component-type topology comp-id)
            :bolt [ComponentType/BOLT
                   (->
                     (partial thriftify-exec-agg-stats comp-id :bolt)
                     (map (:executor-stats data)))
                   (map-val thriftify-bolt-agg-stats w->stats)
                   (thriftify-bolt-input-stats (:cid+sid->input-stats data))
                   (thriftify-bolt-output-stats (:sid->output-stats data))]
            :spout [ComponentType/SPOUT
                    (->
                      (partial thriftify-exec-agg-stats comp-id :spout)
                      (map (:executor-stats data)))
                    (map-val thriftify-spout-agg-stats w->stats)
                    nil ;; spouts do not have input stats
                    (thriftify-spout-output-stats (:sid->output-stats data))]),
        num-executors (:num-executors data)
        num-tasks (:num-tasks data)
        ret (doto (ComponentPageInfo. comp-id compType)
              (.set_topology_id topo-id)
              (.set_topology_name nil)
              (.set_window_to_stats w->stats)
              (.set_sid_to_output_stats sid->output-stats)
              (.set_exec_stats exec-stats))]
    (and num-executors (.set_num_executors ret num-executors))
    (and num-tasks (.set_num_tasks ret num-tasks))
    (and gsid->input-stats
         (.set_gsid_to_input_stats ret gsid->input-stats))
    ret))

(defn agg-comp-execs-stats
  "Aggregate various executor statistics for a component from the given
  heartbeats."
  [exec->host+port
   task->component
   beats
   window
   include-sys?
   topology-id
   topology
   component-id]
  (->> ;; This iterates over each executor one time, because of lazy evaluation.
    (extract-data-from-hb exec->host+port
                          task->component
                          beats
                          include-sys?
                          topology
                          component-id)
    (aggregate-comp-stats window include-sys?)
    (post-aggregate-comp-stats task->component exec->host+port)
    (thriftify-comp-page-data topology-id topology component-id)))

(defn expand-averages
  [avg counts]
  (let [avg (clojurify-structure avg)
        counts (clojurify-structure counts)]
    (into {}
          (for [[slice streams] counts]
            [slice
             (into {}
                   (for [[stream c] streams]
                     [stream
                      [(* c (get-in avg [slice stream]))
                       c]]
                     ))]))))

(defn expand-averages-seq
  [average-seq counts-seq]
  (->> (map vector average-seq counts-seq)
       (map #(apply expand-averages %))
       (apply merge-with (fn [s1 s2] (merge-with add-pairs s1 s2)))))

(defn- val-avg
  [[t c]]
  (if (= t 0) 0
    (double (/ t c))))

(defn aggregate-averages
  [average-seq counts-seq]
  (->> (expand-averages-seq average-seq counts-seq)
       (map-val
         (fn [s]
           (map-val val-avg s)))))

(defn aggregate-avg-streams
  [avg counts]
  (let [expanded (expand-averages avg counts)]
    (->> expanded
         (map-val #(reduce add-pairs (vals %)))
         (map-val val-avg))))

(defn pre-process
  [stream-summary include-sys?]
  (let [filter-fn (mk-include-sys-fn include-sys?)
        emitted (:emitted stream-summary)
        emitted (into {} (for [[window stat] emitted]
                           {window (filter-key filter-fn stat)}))
        transferred (:transferred stream-summary)
        transferred (into {} (for [[window stat] transferred]
                               {window (filter-key filter-fn stat)}))
        stream-summary (-> stream-summary (dissoc :emitted) (assoc :emitted emitted))
        stream-summary (-> stream-summary (dissoc :transferred) (assoc :transferred transferred))]
    stream-summary))

(defn aggregate-counts
  [counts-seq]
  (->> counts-seq
       (map clojurify-structure)
       (apply merge-with
              (fn [s1 s2]
                (merge-with + s1 s2)))))

(defn aggregate-common-stats
  [stats-seq]
  {:emitted (aggregate-counts (map #(.get_emitted ^ExecutorStats %) stats-seq))
   :transferred (aggregate-counts (map #(.get_transferred ^ExecutorStats %) stats-seq))})

(defn aggregate-bolt-stats
  [stats-seq include-sys?]
  (let [stats-seq (collectify stats-seq)]
    (merge (pre-process (aggregate-common-stats stats-seq) include-sys?)
           {:acked
            (aggregate-counts (map #(.. ^ExecutorStats % get_specific get_bolt get_acked)
                                   stats-seq))
            :failed
            (aggregate-counts (map #(.. ^ExecutorStats % get_specific get_bolt get_failed)
                                   stats-seq))
            :executed
            (aggregate-counts (map #(.. ^ExecutorStats % get_specific get_bolt get_executed)
                                   stats-seq))
            :process-latencies
            (aggregate-averages (map #(.. ^ExecutorStats % get_specific get_bolt get_process_ms_avg)
                                     stats-seq)
                                (map #(.. ^ExecutorStats % get_specific get_bolt get_acked)
                                     stats-seq))
            :execute-latencies
            (aggregate-averages (map #(.. ^ExecutorStats % get_specific get_bolt get_execute_ms_avg)
                                     stats-seq)
                                (map #(.. ^ExecutorStats % get_specific get_bolt get_executed)
                                     stats-seq))})))

(defn aggregate-spout-stats
  [stats-seq include-sys?]
  (let [stats-seq (collectify stats-seq)]
    (merge (pre-process (aggregate-common-stats stats-seq) include-sys?)
           {:acked
            (aggregate-counts (map #(.. ^ExecutorStats % get_specific get_spout get_acked)
                                   stats-seq))
            :failed
            (aggregate-counts (map #(.. ^ExecutorStats % get_specific get_spout get_failed)
                                   stats-seq))
            :complete-latencies
            (aggregate-averages (map #(.. ^ExecutorStats % get_specific get_spout get_complete_ms_avg)
                                     stats-seq)
                                (map #(.. ^ExecutorStats % get_specific get_spout get_acked)
                                     stats-seq))})))

(defn get-filled-stats
  [summs]
  (->> summs
       (map #(.get_stats ^ExecutorSummary %))
       (filter not-nil?)))

(defn aggregate-spout-streams
  [stats]
  {:acked (aggregate-count-streams (:acked stats))
   :failed (aggregate-count-streams (:failed stats))
   :emitted (aggregate-count-streams (:emitted stats))
   :transferred (aggregate-count-streams (:transferred stats))
   :complete-latencies (aggregate-avg-streams (:complete-latencies stats)
                                              (:acked stats))})

(defn spout-streams-stats
  [summs include-sys?]
  (let [stats-seq (get-filled-stats summs)]
    (aggregate-spout-streams
      (aggregate-spout-stats
        stats-seq include-sys?))))

(defn aggregate-bolt-streams
  [stats]
  {:acked (aggregate-count-streams (:acked stats))
   :failed (aggregate-count-streams (:failed stats))
   :emitted (aggregate-count-streams (:emitted stats))
   :transferred (aggregate-count-streams (:transferred stats))
   :process-latencies (aggregate-avg-streams (:process-latencies stats)
                                             (:acked stats))
   :executed (aggregate-count-streams (:executed stats))
   :execute-latencies (aggregate-avg-streams (:execute-latencies stats)
                                             (:executed stats))})

(defn compute-executor-capacity
  [^ExecutorSummary e]
  (let [stats (.get_stats e)
        stats (if stats
                (-> stats
                    (aggregate-bolt-stats true)
                    (aggregate-bolt-streams)
                    swap-map-order
                    (get (str TEN-MIN-IN-SECONDS))))
        uptime (nil-to-zero (.get_uptime_secs e))
        window (if (< uptime TEN-MIN-IN-SECONDS) uptime TEN-MIN-IN-SECONDS)
        executed (-> stats :executed nil-to-zero)
        latency (-> stats :execute-latencies nil-to-zero)]
    (if (> window 0)
      (div (* executed latency) (* 1000 window)))))

(defn bolt-streams-stats
  [summs include-sys?]
  (let [stats-seq (get-filled-stats summs)]
    (aggregate-bolt-streams
      (aggregate-bolt-stats
        stats-seq include-sys?))))

(defn total-aggregate-stats
  [spout-summs bolt-summs include-sys?]
  (let [spout-stats (get-filled-stats spout-summs)
        bolt-stats (get-filled-stats bolt-summs)
        agg-spout-stats (-> spout-stats
                            (aggregate-spout-stats include-sys?)
                            aggregate-spout-streams)
        agg-bolt-stats (-> bolt-stats
                           (aggregate-bolt-stats include-sys?)
                           aggregate-bolt-streams)]
    (merge-with
      (fn [s1 s2]
        (merge-with + s1 s2))
      (select-keys
        agg-bolt-stats
        ;; Include only keys that will be used.  We want to count acked and
        ;; failed only for the "tuple trees," so we do not include those keys
        ;; from the bolt executors.
        [:emitted :transferred])
      agg-spout-stats)))

(defn error-subset
  [error-str]
  (apply str (take 200 error-str)))

(defn most-recent-error
  [errors-list]
  (let [error (->> errors-list
                   (sort-by #(.get_error_time_secs ^ErrorInfo %))
                   reverse
                   first)]
    (if error
      (error-subset (.get_error ^ErrorInfo error))
      "")))

(defn float-str [n]
  (if n
    (format "%.3f" (float n))
    "0"))

(defn compute-bolt-capacity
  [executors]
  (->> executors
       (map compute-executor-capacity)
       (map nil-to-zero)
       (apply max)))