(ns tigerbeetle.checker.perf
  "Supporting functions for performance analysis."
  (:require [clojure.stacktrace :as trace]
            [fipp.edn :refer [pprint]]
            [clojure.core.reducers :as r]
            [clojure.set :as set]
            [clojure.java.io :as io]
            [clojure.tools.logging :refer [info warn]]
            [jepsen.checker :as checker]
            [jepsen.util :as util]
            [jepsen.store :as store]
            [multiset.core :as multiset]
            [gnuplot.core :as g]
            [knossos.core :as knossos]
            [knossos.op :as op]
            [knossos.history :as history]
            [slingshot.slingshot :refer [try+ throw+]]))

(def default-nemesis-color "#cccccc")
(def nemesis-alpha 0.6)

(defn bucket-scale
  "Given a bucket size dt, and a bucket number (e.g. 0, 1, ...), returns the
  time at the midpoint of that bucket."
  [dt b]
  (-> b long (* dt) (+ (/ dt 2))))

(defn bucket-time
  "Given a bucket size dt and a time t, computes the time at the midpoint of
  the bucket this time falls into."
  [dt t]
  (bucket-scale dt (/ t dt)))

(defn buckets
  "Given a bucket size dt, emits a lazy sequence of times at the midpoints of
  each bucket."
  ([dt]
   (->> (iterate inc 0)
        (map (partial bucket-scale dt))))
  ([dt tmax]
   (take-while (partial >= tmax) (buckets dt))))

(defn bucket-points
  "Takes a time window dt and a sequence of [time, _] points, and emits a
  seq of [time, points-in-window] buckets, ordered by time. Time is at the
  midpoint of the window."
  [dt points]
  (->> points
       (group-by #(->> % first (bucket-time dt)))
       (into (sorted-map))))

(defn quantiles
  "Takes a sequence of quantiles from 0 to 1 and a sequence of values, and
  returns a map of quantiles to values at those quantiles."
  [qs points]
  (let [sorted (sort points)]
    (when-not (empty? sorted)
      (let [n (count sorted)
            extract (fn [q]
                      (let [idx (min (dec n) (long (Math/floor (* n q))))]
                        (nth sorted idx)))]
        (zipmap qs (map extract qs))))))

(defn latencies->quantiles
  "Takes a time window in seconds, a sequence of quantiles from 0 to 1, and a
  sequence of [time, latency] pairs. Groups pairs by their time window and
  emits a emits a map of quantiles to sequences of [time,
  latency-at-that-quantile] pairs, one per time window."
  [dt qs points]
  (assert (number? dt))
  (assert (every? number? qs))
  (assert (every? #(<= 0 % 1) qs))
  (let [buckets (->> points
                     (bucket-points dt)
                     (map (fn [[bucket-time points]]
                            [bucket-time (quantiles qs (map second points))])))]
    ; At this point we have a sequence of
  [ time , { q1 v1 , q2 ; v2 , ... } , ... ]
    ; pairs, and we want a map of
    ; {q1 -> [[time, v1], [time2, v2], ...], ...}
    (->> qs
         (map (fn [q]
                (map (fn [[t qs]]
                       [t (get qs q)])
                     buckets)))
         (zipmap qs))))

(defn first-time
  "Takes a history and returns the first :time in it, in seconds, as a double."
  [history]
  (when-let [t (->> history
                    (keep :time)
                    first)]
    (double (util/nanos->secs t))))

(defn invokes-by-type
  "Splits up a sequence of invocations into ok, failed, and crashed ops by
  looking at their corresponding completions."
  [ops]
  {:ok   (filter #(= :ok   (:type (:completion %))) ops)
   :fail (filter #(= :fail (:type (:completion %))) ops)
   :info (filter #(= :info (:type (:completion %))) ops)})

(defn invokes-by-f
  "Takes a history and returns a map of f -> ops, for all invocations."
  [history]
  (->> history
       (filter op/invoke?)
       (group-by :f)))

(defn invokes-by-f-type
  "Takes a history and returns a map of f -> type -> ops, for all invocations."
  [history]
  (->> history
       (filter op/invoke?)
       (group-by :f)
       (util/map-kv (fn [[f ops]] [f (invokes-by-type ops)]))))

(defn completions-by-f-type
  "Takes a history and returns a map of f -> type-> ops, for all completions in
  history."
  [history]
  (->> history
       (remove op/invoke?)
       (group-by :f)
       (util/map-kv (fn [[f ops]] [f (group-by :type ops)]))))

(defn rate
  "Map breaking down the mean rate of completions by f and type, plus totals at
  each level."
  [history]
  (->> history
       (r/remove op/invoke?)
       (reduce (fn [m op]
                 (let [f (:f op)
                       t (:type op)]
                   ; slow and bad
                   (-> m
                       (update-in [f t]         util/inc*)
                       (update-in [f ::all]     util/inc*)
                       (update-in [::all t]     util/inc*)
                       (update-in [::all ::all] util/inc*)))))))

(defn latency-point
  "Given an operation, returns a [time, latency] pair: times in seconds,
  latencies in ms."
  [op]
  (list (double (util/nanos->secs (:time op)))
        (double (util/nanos->ms   (:latency op)))))

(defn fs->points
  "Given a sequence of :f's, yields a map of f -> gnuplot-point-type, so we can
  render each function in a different style."
  [fs]
  (->> fs
       (map-indexed (fn [i f] [f (* 2 (+ 2 i))]))
       (into {})))

(defn qs->colors
  "Given a sequence of quantiles q, yields a map of q -> gnuplot-color, so we
  can render each latency curve in a different color."
  [qs]
  (-> qs
      sort
      reverse
      (zipmap (map vector
                   (repeat 'rgb)
                   (cycle ["red"
                           "orange"
                           "purple"
                           "blue"
                           "green"
                           "grey"])))))

(def types
  "What types are we rendering?"
  [:ok :info :fail])

(def type->color
  "Takes a type of operation (e.g. :ok) and returns a gnuplot color."
  {:ok   ['rgb "#81BFFC"]
   :info ['rgb "#FFA400"]
   :fail ['rgb "#FF1E90"]})

(defn nemesis-ops
  "Given a history and a nemeses specification, partitions the set of nemesis
  operations in the history into different nemeses, as per the spec. Returns
  the nemesis spec, restricted to just those nemeses taking part in this
  history, and with each spec augmented with an :ops key, which contains all
  operations that nemesis performed."
  [nemeses history]
  ; Build an index mapping :fs to nemeses.
  ; TODO: verify no nemesis fs overlap
  (assert (every? :name nemeses))
  (assert (= (distinct (map :name nemeses)) (map :name nemeses)))
  (let [index (->> nemeses
                   (map (fn [nemesis]
                          (zipmap (concat (:start nemesis   [:start])
                                          (:stop  nemesis   [:stop])
                                          (:fs    nemesis))
                                  (repeat (:name nemesis)))))
                   (reduce merge {}))
        ; Go through the history and build up a map of names to ops.
        ops-by-nemesis (->> history
                            (filter #(= :nemesis (:process %)))
                            (group-by (comp index :f)))
        ; And zip that together with the nemesis spec
        nemeses (keep (fn [n]
                        (when-let [ops (ops-by-nemesis (:name n))]
                          (assoc n :ops ops)))
                      nemeses)
        ; And add a default for any unmatched ops
        nemeses (if-let [ops (ops-by-nemesis nil)]
                  (conj nemeses {:name "nemesis"
                                 :ops  ops})
                  nemeses)]
    nemeses))

(defn nemesis-activity
  "Given a nemesis specification and a history, partitions the set of nemesis
  operations in the history into different nemeses, as per the spec. Returns
  the spec, restricted to just those nemeses taking part in this history, and
  with each spec augmented with two keys:

    :ops        All operations the nemeses performed
    :intervals  A set of [start end] paired ops."
  [nemeses history]
  (->> history
       (nemesis-ops nemeses)
       (map (fn [nemesis]
              (assoc nemesis :intervals
                     (util/nemesis-intervals (:ops nemesis) nemesis))))))

(defn interval->times
  "Given an interval of two operations [a b], returns the times [time-a time-b]
  covering the interval. If b is missing, yields [time-a nil]."
  [[a b]]
  [(double (util/nanos->secs (:time a)))
   (when b (double (util/nanos->secs (:time b))))])

(defn nemesis-regions
  "Given nemesis activity, emits a sequence of gnuplot commands rendering
  shaded regions where each nemesis was active. We can render a maximum of 12
  nemeses; this keeps size and spacing consistent."
  [plot nemeses]
  (->> nemeses
       (map-indexed
        (fn [i n]
          (let [color           (or (:fill-color n)
                                    (:color n)
                                    default-nemesis-color)
                transparency    (:transparency n nemesis-alpha)
                graph-top-edge  1
                 ; Divide our y-axis space into twelfths
                height          0.0834
                padding         0.00615
                bot             (- graph-top-edge
                                   (* height (inc i)))
                top             (+ bot height)]
            (->> (:intervals n)
                 (map interval->times)
                 (map (fn [[start stop]]
                        [:set :obj :rect
                         :from (g/list start [:graph (+ bot padding)])
                         :to   (g/list (or stop [:graph 1])
                                       [:graph (- top padding)])
                         :fillcolor :rgb color
                         :fillstyle :transparent :solid transparency
                         :noborder]))))))
       (reduce concat)))

(defn nemesis-lines
  "Given nemesis activity, emits a sequence of gnuplot commands rendering
  vertical lines where nemesis events occurred."
  [plot nemeses]
  (let [tfilter (if-let [[xmin xmax] (:xrange plot)]
                  (fn [t] (<= xmin t xmax))
                  identity)]
    (->> nemeses
         (mapcat (fn [n]
                   (let [color (or (:line-color n)
                                   (:color n)
                                   default-nemesis-color)
                         width (:line-width n "1")]
                     (->> (:ops n)
                          (map (comp double util/nanos->secs :time))
                          (filter tfilter)
                          (map (fn [t]
                                 [:set :arrow
                                  :from (g/list t [:graph 0])
                                  :to   (g/list t [:graph 1])
                                  :lc :rgb color
                                  :lw width
                                  :nohead])))))))))

(defn nemesis-series
  "Given nemesis activity, constructs the series required to show every present
  nemesis' activity in the legend. We do this by constructing dummy data, and a
  key that will match the way that nemesis's activity is rendered."
  [plot nemeses]
  (->> nemeses
       (map (fn [n]
              {:title     (str (:name n))
               :with      :lines
               :linecolor ['rgb (or (:fill-color n)
                                    (:color n)
                                    default-nemesis-color)]
               :linewidth 6
               :data      [[-1 -1]]}))))

(defn with-nemeses
  "Augments a plot map to render nemesis activity. Takes a nemesis
  specification: a collection of nemesis spec maps, each of which has keys:

    :name   A string uniquely naming this nemesis
    :color  What color to use for drawing this nemesis (e.g. \"#abcd01\")
    :start  A set of :f's which begin this nemesis' activity
    :stop   A set of :f's which end this nemesis' activity
    :fs     A set of :f's otherwise related to this nemesis"
  [plot history nemeses]
  (let [nemeses (nemesis-activity nemeses history)]
    (-> plot
        (update :series   concat (nemesis-series  plot nemeses))
        (update :preamble concat (nemesis-regions plot nemeses)
                (nemesis-lines   plot nemeses)))))

(defn preamble
  "Shared gnuplot preamble"
  [output-path]
  (concat [[:set :output output-path]
           [:set :term :png, :truecolor, :size (g/list 900 400)]]
          '[[set xlabel "Time (s)"]
            [set key outside top right]]))

(defn broaden-range
  "Given a [lower upper] range for a plot, returns [lower' upper'], which
  covers the original range, but slightly expanded, to fall nicely on integral
  boundaries."
  [[a b]]
  (if (= a b)
  If it 's a zero - width interval , give it exactly 1 on either side .
    [(dec a) (inc a)]
    (let [; How big is the range?
          size (Math/abs (- (double b) (double a)))
  Divide the range into tenths
          grid (/ size 10)
  What 's the nearest power of 10 ?
          scale (->> grid Math/log10 Math/round (Math/pow 10))
          ; Clamp
          a' (- a (mod a scale))
          b' (let [m (mod b scale)]
               ; If b is suuuuper close to the scale tick already...
               (if (< (/ m scale) 0.001)
                 b                                ; Just use b as is
                 (+ scale (- b (mod b scale)))))  ; Push b up to the next tick
          a' (min a a')
          b' (max b b')]
      [a' b'])))

(defn without-empty-series
  "Takes a plot, and strips out empty series objects."
  [plot]
  (update plot :series (partial filter (comp seq :data))))

(defn has-data?
  "Takes a plot and returns true iff it has at least one series with
  data points."
  [plot]
  (boolean (some (comp seq :data) (:series plot))))

(defn with-range
  "Takes a plot object. Where xrange or yrange are not provided, fills them in
  by iterating over each series :data."
  [plot]
  (let [data    (mapcat :data (:series plot))
        _       (when-not (seq data)
                  (throw+ {:type ::no-points
                           :plot plot}))
        [x0 y0] (first data)
        [xmin xmax ymin ymax] (reduce (fn [[xmin xmax ymin ymax] [x y :as pair]]
                                        [(min xmin x)
                                         (max xmax x)
                                         (min ymin y)
                                         (max ymax y)])
                                      [x0 x0 y0 y0]
                                      data)
        xrange (broaden-range [xmin xmax])
        yrange (if (= :y (:logscale plot))
                 ; Don't broaden logscale plots; we'll probably expand the
                 ; bounds to 0.
                 [ymin ymax]
                 (broaden-range [ymin ymax]))]
    (assoc plot
           :xrange (:xrange plot xrange)
           :yrange (:yrange plot yrange))))

(defn latency-preamble
  "Gnuplot commands for setting up a latency plot."
  [test output-path]
  (concat (preamble output-path)
          [[:set :title (str (:name test) " latency")]]
          '[[set ylabel "Latency (ms)"]
            [set logscale y]]))

(defn legend-part
  "Takes a series map and returns the list of gnuplot commands to render that
  series."
  [series]
  (remove nil?
          ["-"
           'with      (:with series)
           (when-let [t (:linetype series)]  ['linetype t])
           (when-let [c (:linecolor series)] ['linecolor c])
           (when-let [t (:pointtype series)] ['pointtype t])
           (when-let [w (:linewidth series)] ['linewidth w])
           (if-let [t (:title series)]       ['title t]       'notitle)]))

(defn plot!
  "Renders a gnuplot plot. Takes an option map:

    :preamble             Gnuplot commands to send first
    :series               A vector of series maps
    :draw-fewer-on-top?   If passed, renders series with fewer points on top
    :xrange               A pair [xmin xmax] which controls the xrange
    :yrange               Ditto, for the y axis
    :logscale             e.g. :y

  A series map is a map with:

    :data       A sequence of data points to render, e,g. [[0 0] [1 2] [2 4]]
    :with       How to draw this series, e.g. 'points
    :linetype   What kind of line to use
    :pointtype  What kind of point to use
    :title      A string, or nil, to label this series map
  "
  [opts]
  ( info : plotting ( with - out - str ( pprint opts ) ) )
  (assert (every? sequential? (map :data (:series opts)))
          (str "Series has no :data points\n"
               (with-out-str (pprint (remove (comp seq :data)
                                             (:series opts))))))
  (if (:draw-fewer-on-top? opts)
  We 're going to transform our series in two ways : one , in their normal
  order , but with only a single dummy point , and second , those with the
  most points first , but without titles , so they do n't appear in the
    ; legend.
    (let [series (:series opts)
          series (concat (->> series
                              (sort-by (comp count :data))
                              reverse
                              (map #(dissoc % :title)))
                         ; Then, the normal series, but with dummy points
                         (->> series
                              (map #(assoc % :data [[0 -1]]))))]
      ; OK, go ahead and render that
      (recur (assoc opts
                    :series series
                    :draw-fewer-on-top? false)))

    ; OK, normal rendering
    (let [series      (:series opts)
          preamble    (:preamble opts)
          xrange      (:xrange opts)
          yrange      (:yrange opts)
          logscale    (:logscale opts)
          ; The plot command
          plot        [['plot (apply g/list (map legend-part series))]]
          ; All commands
          commands    (cond-> []
                        preamble  (into preamble)
                        logscale  (conj [:set :logscale logscale])
                        xrange    (conj [:set :xrange (apply g/range xrange)])
                        yrange    (conj [:set :yrange (apply g/range yrange)])
                        true      (concat plot))
          ; Datasets
          data        (map :data series)]
      ; Go!
      ;(info (with-out-str
      ;        (pprint commands)
         ( pprint ( map ( partial take 2 ) data ) ) ) )
      (try (g/raw-plot! commands data)
           (catch java.io.IOException e
             (throw (IllegalStateException. "Error rendering plot, verify gnuplot is installed and reachable" e)))))))

(defn point-graph!
  "Writes a plot of raw latency data points."
  [test history {:keys [subdirectory nemeses] :as opts}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        history     (util/history->latencies history)
        datasets    (invokes-by-f-type history)
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        output-path (.getCanonicalPath (store/path! test
                                                    subdirectory
                                                    "latency-raw.png"))
        preamble    (latency-preamble test output-path)
        series      (->> (for [f fs, t types]
                           (when-let [data (seq (get-in datasets [f t]))]
                             {:title     (str (util/name+ f) " " (name t))
                              :with      'points
                              :linetype  (type->color t)
                              :pointtype (fs->points- f)
                              :data      (map latency-point data)}))
                         (remove nil?))]
    (-> {:preamble           preamble
         :draw-fewer-on-top? true
         :logscale           :y
         :series             series}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn quantiles-graph!
  "Writes a plot of latency quantiles, by f, over time."
  [test history {:keys [subdirectory nemeses]}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        history     (util/history->latencies history)
        dt          1
        qs          [0.5 0.95 0.99 1]
        datasets    (->> history
                         invokes-by-f
                         ;; For each f, emit a map of quantiles to points
                         (util/map-kv
                          (fn [[f ops]]
                            (->> ops
                                 (map latency-point)
                                 (latencies->quantiles dt qs)
                                 (vector f)))))
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        qs->colors- (qs->colors qs)
        output-path (.getCanonicalPath
                     (store/path! test
                                  subdirectory
                                  "latency-quantiles.png"))

        preamble    (latency-preamble test output-path)
        series      (for [f fs, q qs]
                      {:title     (str (util/name+ f) " " q)
                       :with      'linespoints
                       :linetype  (qs->colors- q)
                       :pointtype  (fs->points- f)
                       :data      (get-in datasets [f q])})]
    (-> {:preamble preamble
         :series   series
         :logscale :y}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn rate-preamble
  "Gnuplot commands for setting up a rate plot."
  [test output-path]
  (concat (preamble output-path)
          [[:set :title (str (:name test) " rate")]]
          '[[set ylabel "Throughput (hz)"]]))

(defn rate-graph!
  "Writes a plot of operation rate by their completion times."
  [test history {:keys [subdirectory nemeses]}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        dt          1
        td          (double (/ dt))
        t-max       (->> history (r/map :time) (reduce max 0) util/nanos->secs)
        datasets    (->> history
                         (r/remove op/invoke?)
                         ; Don't graph nemeses
                         (r/filter (comp integer? :process))
                         ; Compute rates
                         (reduce (fn [m op]
                                   (update-in m [(:f op)
                                                 (:type op)
                                                 (bucket-time dt
                                                              (util/nanos->secs
                                                               (:time op)))]
                                              #(+ td (or % 0))))
                                 {}))
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        output-path (.getCanonicalPath (store/path! test
                                                    subdirectory
                                                    "rate.png"))

        preamble (rate-preamble test output-path)
        series   (->> (for [f fs, t types]
                        (when-let [data (get-in datasets [f t])]
                          {:title     (str (util/name+ f) " " (name t))
                           :with      'linespoints
                           :linetype  (type->color t)
                           :pointtype (fs->points- f)
                           :data      (map (juxt identity #(get data % 0))
                                           (buckets dt t-max))}))
                      (remove nil?))]
    (-> {:preamble  preamble
         :series    series}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn open-ops-preamble
  "Gnuplot commands for setting up an open ops plot."
  [test output-path]
  (concat (preamble output-path)
          [[:set :title (str (:name test) " open ops")]]
          '[[set ylabel "Open Ops"]]))

(defn open-ops-graph!
  "Writes a plot of open operations by time."
  [test history {:keys [subdirectory nemeses]}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        dt          1
        td          (double (/ dt))
        t-max       (->> history (r/map :time) (reduce max 0) util/nanos->secs)
        history+     (->> history (history/pair-index+))
        [datasets _] (->> history
                          ; Don't graph nemeses
                          (r/filter (comp integer? :process))
                          ; Compute open/close ops
                          (reduce (fn [[m open-totals] {:keys [type f time] :as op}]
                                    (let [outcome    (->> op (history/completion history+) :type)
                                          open-total (get-in open-totals [f outcome] 0)
                                          open-total (case type
                                                       :invoke           (+ open-total 1)
                                                       (:ok :info :fail) (- open-total 1))]
                                      [(assoc-in m [f
                                                    outcome
                                                    (bucket-time dt
                                                                 (util/nanos->secs time))]
                                                 open-total)
                                       (assoc-in open-totals [f outcome] open-total)]))
                                  [{} {}]))
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        output-path (.getCanonicalPath (store/path! test
                                                    subdirectory
                                                    "open-ops.png"))

        preamble (open-ops-preamble test output-path)
        series   (->> (for [f fs t types]
                        (when-let [data (get-in datasets [f t])]
                          (let [[data _] (->> (buckets dt t-max)
                                              (reduce (fn [[acc curr] bt]
                                                        (let [curr (or (get data bt) curr)]
                                                          [(assoc acc bt curr) curr]))
                                                      [{} 0]))]
                            {:title     (str (util/name+ f) " " (name t))
                             :with      'linespoints
                             :linetype  (type->color t)
                             :pointtype (fs->points- f)
                             :data      (map (juxt identity #(get data % 0))
                                             (buckets dt t-max))})))
                      (remove nil?))]
    (-> {:preamble  preamble
         :series    series}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn latency-graph
  "Spits out graphs of latencies. Checker options take precedence over
  those passed in with this constructor."
  ([]
   (latency-graph {}))
  ([opts]
   (reify checker/Checker
     (check [_ test history c-opts]
       (let [o (merge opts c-opts)]
         (point-graph!     test history o)
         (quantiles-graph! test history o)
         {:valid? true})))))

(defn rate-graph
  "Spits out graphs of throughput over time. Checker options take precedence
  over those passed in with this constructor."
  ([]
   (rate-graph {}))
  ([opts]
   (reify checker/Checker
     (check [_ test history c-opts]
       (let [o (merge opts c-opts)]
         (rate-graph! test history o)
         {:valid? true})))))

(defn open-ops-graph
  "Spits out graphs of open ops over time. Checker options take precedence
  over those passed in with this constructor."
  ([]
   (open-ops-graph {}))
  ([opts]
   (reify checker/Checker
     (check [_ test history c-opts]
       (let [o (merge opts c-opts)]
         (open-ops-graph! test history o)
         {:valid? true})))))

(defn perf
  "Composes various performance statistics. Checker options take precedence over
  those passed in with this constructor."
  ([]
   (perf {}))
  ([opts]
   (checker/compose {:latency-graph  (latency-graph  opts)
                     :rate-graph     (rate-graph     opts)
                     :open-ops-graph (open-ops-graph opts)})))

 At this point we have a sequence of
 v2 , ... } , ... ]
 pairs, and we want a map of
 {q1 -> [[time, v1], [time2, v2], ...], ...}
 slow and bad
 Build an index mapping :fs to nemeses.
 TODO: verify no nemesis fs overlap
 Go through the history and build up a map of names to ops.
 And zip that together with the nemesis spec
 And add a default for any unmatched ops
 this keeps size and spacing consistent."
 Divide our y-axis space into twelfths
 How big is the range?
 Clamp
 If b is suuuuper close to the scale tick already...
 Just use b as is
 Push b up to the next tick
 Don't broaden logscale plots; we'll probably expand the
 bounds to 0.
 legend.
 Then, the normal series, but with dummy points
 OK, go ahead and render that
 OK, normal rendering
 The plot command
 All commands
 Datasets
 Go!
(info (with-out-str
        (pprint commands)
 For each f, emit a map of quantiles to points
 Don't graph nemeses
 Compute rates
 Don't graph nemeses
 Compute open/close ops
(ns tigerbeetle.checker.perf
  "Supporting functions for performance analysis."
  (:require [clojure.stacktrace :as trace]
            [fipp.edn :refer [pprint]]
            [clojure.core.reducers :as r]
            [clojure.set :as set]
            [clojure.java.io :as io]
            [clojure.tools.logging :refer [info warn]]
            [jepsen.checker :as checker]
            [jepsen.util :as util]
            [jepsen.store :as store]
            [multiset.core :as multiset]
            [gnuplot.core :as g]
            [knossos.core :as knossos]
            [knossos.op :as op]
            [knossos.history :as history]
            [slingshot.slingshot :refer [try+ throw+]]))

(def default-nemesis-color "#cccccc")
(def nemesis-alpha 0.6)

(defn bucket-scale
  "Given a bucket size dt, and a bucket number (e.g. 0, 1, ...), returns the
  time at the midpoint of that bucket."
  [dt b]
  (-> b long (* dt) (+ (/ dt 2))))

(defn bucket-time
  "Given a bucket size dt and a time t, computes the time at the midpoint of
  the bucket this time falls into."
  [dt t]
  (bucket-scale dt (/ t dt)))

(defn buckets
  "Given a bucket size dt, emits a lazy sequence of times at the midpoints of
  each bucket."
  ([dt]
   (->> (iterate inc 0)
        (map (partial bucket-scale dt))))
  ([dt tmax]
   (take-while (partial >= tmax) (buckets dt))))

(defn bucket-points
  "Takes a time window dt and a sequence of [time, _] points, and emits a
  seq of [time, points-in-window] buckets, ordered by time. Time is at the
  midpoint of the window."
  [dt points]
  (->> points
       (group-by #(->> % first (bucket-time dt)))
       (into (sorted-map))))

(defn quantiles
  "Takes a sequence of quantiles from 0 to 1 and a sequence of values, and
  returns a map of quantiles to values at those quantiles."
  [qs points]
  (let [sorted (sort points)]
    (when-not (empty? sorted)
      (let [n (count sorted)
            extract (fn [q]
                      (let [idx (min (dec n) (long (Math/floor (* n q))))]
                        (nth sorted idx)))]
        (zipmap qs (map extract qs))))))

(defn latencies->quantiles
  "Takes a time window in seconds, a sequence of quantiles from 0 to 1, and a
  sequence of [time, latency] pairs. Groups pairs by their time window and
  emits a emits a map of quantiles to sequences of [time,
  latency-at-that-quantile] pairs, one per time window."
  [dt qs points]
  (assert (number? dt))
  (assert (every? number? qs))
  (assert (every? #(<= 0 % 1) qs))
  (let [buckets (->> points
                     (bucket-points dt)
                     (map (fn [[bucket-time points]]
                            [bucket-time (quantiles qs (map second points))])))]
    (->> qs
         (map (fn [q]
                (map (fn [[t qs]]
                       [t (get qs q)])
                     buckets)))
         (zipmap qs))))

(defn first-time
  "Takes a history and returns the first :time in it, in seconds, as a double."
  [history]
  (when-let [t (->> history
                    (keep :time)
                    first)]
    (double (util/nanos->secs t))))

(defn invokes-by-type
  "Splits up a sequence of invocations into ok, failed, and crashed ops by
  looking at their corresponding completions."
  [ops]
  {:ok   (filter #(= :ok   (:type (:completion %))) ops)
   :fail (filter #(= :fail (:type (:completion %))) ops)
   :info (filter #(= :info (:type (:completion %))) ops)})

(defn invokes-by-f
  "Takes a history and returns a map of f -> ops, for all invocations."
  [history]
  (->> history
       (filter op/invoke?)
       (group-by :f)))

(defn invokes-by-f-type
  "Takes a history and returns a map of f -> type -> ops, for all invocations."
  [history]
  (->> history
       (filter op/invoke?)
       (group-by :f)
       (util/map-kv (fn [[f ops]] [f (invokes-by-type ops)]))))

(defn completions-by-f-type
  "Takes a history and returns a map of f -> type-> ops, for all completions in
  history."
  [history]
  (->> history
       (remove op/invoke?)
       (group-by :f)
       (util/map-kv (fn [[f ops]] [f (group-by :type ops)]))))

(defn rate
  "Map breaking down the mean rate of completions by f and type, plus totals at
  each level."
  [history]
  (->> history
       (r/remove op/invoke?)
       (reduce (fn [m op]
                 (let [f (:f op)
                       t (:type op)]
                   (-> m
                       (update-in [f t]         util/inc*)
                       (update-in [f ::all]     util/inc*)
                       (update-in [::all t]     util/inc*)
                       (update-in [::all ::all] util/inc*)))))))

(defn latency-point
  "Given an operation, returns a [time, latency] pair: times in seconds,
  latencies in ms."
  [op]
  (list (double (util/nanos->secs (:time op)))
        (double (util/nanos->ms   (:latency op)))))

(defn fs->points
  "Given a sequence of :f's, yields a map of f -> gnuplot-point-type, so we can
  render each function in a different style."
  [fs]
  (->> fs
       (map-indexed (fn [i f] [f (* 2 (+ 2 i))]))
       (into {})))

(defn qs->colors
  "Given a sequence of quantiles q, yields a map of q -> gnuplot-color, so we
  can render each latency curve in a different color."
  [qs]
  (-> qs
      sort
      reverse
      (zipmap (map vector
                   (repeat 'rgb)
                   (cycle ["red"
                           "orange"
                           "purple"
                           "blue"
                           "green"
                           "grey"])))))

(def types
  "What types are we rendering?"
  [:ok :info :fail])

(def type->color
  "Takes a type of operation (e.g. :ok) and returns a gnuplot color."
  {:ok   ['rgb "#81BFFC"]
   :info ['rgb "#FFA400"]
   :fail ['rgb "#FF1E90"]})

(defn nemesis-ops
  "Given a history and a nemeses specification, partitions the set of nemesis
  operations in the history into different nemeses, as per the spec. Returns
  the nemesis spec, restricted to just those nemeses taking part in this
  history, and with each spec augmented with an :ops key, which contains all
  operations that nemesis performed."
  [nemeses history]
  (assert (every? :name nemeses))
  (assert (= (distinct (map :name nemeses)) (map :name nemeses)))
  (let [index (->> nemeses
                   (map (fn [nemesis]
                          (zipmap (concat (:start nemesis   [:start])
                                          (:stop  nemesis   [:stop])
                                          (:fs    nemesis))
                                  (repeat (:name nemesis)))))
                   (reduce merge {}))
        ops-by-nemesis (->> history
                            (filter #(= :nemesis (:process %)))
                            (group-by (comp index :f)))
        nemeses (keep (fn [n]
                        (when-let [ops (ops-by-nemesis (:name n))]
                          (assoc n :ops ops)))
                      nemeses)
        nemeses (if-let [ops (ops-by-nemesis nil)]
                  (conj nemeses {:name "nemesis"
                                 :ops  ops})
                  nemeses)]
    nemeses))

(defn nemesis-activity
  "Given a nemesis specification and a history, partitions the set of nemesis
  operations in the history into different nemeses, as per the spec. Returns
  the spec, restricted to just those nemeses taking part in this history, and
  with each spec augmented with two keys:

    :ops        All operations the nemeses performed
    :intervals  A set of [start end] paired ops."
  [nemeses history]
  (->> history
       (nemesis-ops nemeses)
       (map (fn [nemesis]
              (assoc nemesis :intervals
                     (util/nemesis-intervals (:ops nemesis) nemesis))))))

(defn interval->times
  "Given an interval of two operations [a b], returns the times [time-a time-b]
  covering the interval. If b is missing, yields [time-a nil]."
  [[a b]]
  [(double (util/nanos->secs (:time a)))
   (when b (double (util/nanos->secs (:time b))))])

(defn nemesis-regions
  "Given nemesis activity, emits a sequence of gnuplot commands rendering
  shaded regions where each nemesis was active. We can render a maximum of 12
  [plot nemeses]
  (->> nemeses
       (map-indexed
        (fn [i n]
          (let [color           (or (:fill-color n)
                                    (:color n)
                                    default-nemesis-color)
                transparency    (:transparency n nemesis-alpha)
                graph-top-edge  1
                height          0.0834
                padding         0.00615
                bot             (- graph-top-edge
                                   (* height (inc i)))
                top             (+ bot height)]
            (->> (:intervals n)
                 (map interval->times)
                 (map (fn [[start stop]]
                        [:set :obj :rect
                         :from (g/list start [:graph (+ bot padding)])
                         :to   (g/list (or stop [:graph 1])
                                       [:graph (- top padding)])
                         :fillcolor :rgb color
                         :fillstyle :transparent :solid transparency
                         :noborder]))))))
       (reduce concat)))

(defn nemesis-lines
  "Given nemesis activity, emits a sequence of gnuplot commands rendering
  vertical lines where nemesis events occurred."
  [plot nemeses]
  (let [tfilter (if-let [[xmin xmax] (:xrange plot)]
                  (fn [t] (<= xmin t xmax))
                  identity)]
    (->> nemeses
         (mapcat (fn [n]
                   (let [color (or (:line-color n)
                                   (:color n)
                                   default-nemesis-color)
                         width (:line-width n "1")]
                     (->> (:ops n)
                          (map (comp double util/nanos->secs :time))
                          (filter tfilter)
                          (map (fn [t]
                                 [:set :arrow
                                  :from (g/list t [:graph 0])
                                  :to   (g/list t [:graph 1])
                                  :lc :rgb color
                                  :lw width
                                  :nohead])))))))))

(defn nemesis-series
  "Given nemesis activity, constructs the series required to show every present
  nemesis' activity in the legend. We do this by constructing dummy data, and a
  key that will match the way that nemesis's activity is rendered."
  [plot nemeses]
  (->> nemeses
       (map (fn [n]
              {:title     (str (:name n))
               :with      :lines
               :linecolor ['rgb (or (:fill-color n)
                                    (:color n)
                                    default-nemesis-color)]
               :linewidth 6
               :data      [[-1 -1]]}))))

(defn with-nemeses
  "Augments a plot map to render nemesis activity. Takes a nemesis
  specification: a collection of nemesis spec maps, each of which has keys:

    :name   A string uniquely naming this nemesis
    :color  What color to use for drawing this nemesis (e.g. \"#abcd01\")
    :start  A set of :f's which begin this nemesis' activity
    :stop   A set of :f's which end this nemesis' activity
    :fs     A set of :f's otherwise related to this nemesis"
  [plot history nemeses]
  (let [nemeses (nemesis-activity nemeses history)]
    (-> plot
        (update :series   concat (nemesis-series  plot nemeses))
        (update :preamble concat (nemesis-regions plot nemeses)
                (nemesis-lines   plot nemeses)))))

(defn preamble
  "Shared gnuplot preamble"
  [output-path]
  (concat [[:set :output output-path]
           [:set :term :png, :truecolor, :size (g/list 900 400)]]
          '[[set xlabel "Time (s)"]
            [set key outside top right]]))

(defn broaden-range
  "Given a [lower upper] range for a plot, returns [lower' upper'], which
  covers the original range, but slightly expanded, to fall nicely on integral
  boundaries."
  [[a b]]
  (if (= a b)
  If it 's a zero - width interval , give it exactly 1 on either side .
    [(dec a) (inc a)]
          size (Math/abs (- (double b) (double a)))
  Divide the range into tenths
          grid (/ size 10)
  What 's the nearest power of 10 ?
          scale (->> grid Math/log10 Math/round (Math/pow 10))
          a' (- a (mod a scale))
          b' (let [m (mod b scale)]
               (if (< (/ m scale) 0.001)
          a' (min a a')
          b' (max b b')]
      [a' b'])))

(defn without-empty-series
  "Takes a plot, and strips out empty series objects."
  [plot]
  (update plot :series (partial filter (comp seq :data))))

(defn has-data?
  "Takes a plot and returns true iff it has at least one series with
  data points."
  [plot]
  (boolean (some (comp seq :data) (:series plot))))

(defn with-range
  "Takes a plot object. Where xrange or yrange are not provided, fills them in
  by iterating over each series :data."
  [plot]
  (let [data    (mapcat :data (:series plot))
        _       (when-not (seq data)
                  (throw+ {:type ::no-points
                           :plot plot}))
        [x0 y0] (first data)
        [xmin xmax ymin ymax] (reduce (fn [[xmin xmax ymin ymax] [x y :as pair]]
                                        [(min xmin x)
                                         (max xmax x)
                                         (min ymin y)
                                         (max ymax y)])
                                      [x0 x0 y0 y0]
                                      data)
        xrange (broaden-range [xmin xmax])
        yrange (if (= :y (:logscale plot))
                 [ymin ymax]
                 (broaden-range [ymin ymax]))]
    (assoc plot
           :xrange (:xrange plot xrange)
           :yrange (:yrange plot yrange))))

(defn latency-preamble
  "Gnuplot commands for setting up a latency plot."
  [test output-path]
  (concat (preamble output-path)
          [[:set :title (str (:name test) " latency")]]
          '[[set ylabel "Latency (ms)"]
            [set logscale y]]))

(defn legend-part
  "Takes a series map and returns the list of gnuplot commands to render that
  series."
  [series]
  (remove nil?
          ["-"
           'with      (:with series)
           (when-let [t (:linetype series)]  ['linetype t])
           (when-let [c (:linecolor series)] ['linecolor c])
           (when-let [t (:pointtype series)] ['pointtype t])
           (when-let [w (:linewidth series)] ['linewidth w])
           (if-let [t (:title series)]       ['title t]       'notitle)]))

(defn plot!
  "Renders a gnuplot plot. Takes an option map:

    :preamble             Gnuplot commands to send first
    :series               A vector of series maps
    :draw-fewer-on-top?   If passed, renders series with fewer points on top
    :xrange               A pair [xmin xmax] which controls the xrange
    :yrange               Ditto, for the y axis
    :logscale             e.g. :y

  A series map is a map with:

    :data       A sequence of data points to render, e,g. [[0 0] [1 2] [2 4]]
    :with       How to draw this series, e.g. 'points
    :linetype   What kind of line to use
    :pointtype  What kind of point to use
    :title      A string, or nil, to label this series map
  "
  [opts]
  ( info : plotting ( with - out - str ( pprint opts ) ) )
  (assert (every? sequential? (map :data (:series opts)))
          (str "Series has no :data points\n"
               (with-out-str (pprint (remove (comp seq :data)
                                             (:series opts))))))
  (if (:draw-fewer-on-top? opts)
  We 're going to transform our series in two ways : one , in their normal
  order , but with only a single dummy point , and second , those with the
  most points first , but without titles , so they do n't appear in the
    (let [series (:series opts)
          series (concat (->> series
                              (sort-by (comp count :data))
                              reverse
                              (map #(dissoc % :title)))
                         (->> series
                              (map #(assoc % :data [[0 -1]]))))]
      (recur (assoc opts
                    :series series
                    :draw-fewer-on-top? false)))

    (let [series      (:series opts)
          preamble    (:preamble opts)
          xrange      (:xrange opts)
          yrange      (:yrange opts)
          logscale    (:logscale opts)
          plot        [['plot (apply g/list (map legend-part series))]]
          commands    (cond-> []
                        preamble  (into preamble)
                        logscale  (conj [:set :logscale logscale])
                        xrange    (conj [:set :xrange (apply g/range xrange)])
                        yrange    (conj [:set :yrange (apply g/range yrange)])
                        true      (concat plot))
          data        (map :data series)]
         ( pprint ( map ( partial take 2 ) data ) ) ) )
      (try (g/raw-plot! commands data)
           (catch java.io.IOException e
             (throw (IllegalStateException. "Error rendering plot, verify gnuplot is installed and reachable" e)))))))

(defn point-graph!
  "Writes a plot of raw latency data points."
  [test history {:keys [subdirectory nemeses] :as opts}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        history     (util/history->latencies history)
        datasets    (invokes-by-f-type history)
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        output-path (.getCanonicalPath (store/path! test
                                                    subdirectory
                                                    "latency-raw.png"))
        preamble    (latency-preamble test output-path)
        series      (->> (for [f fs, t types]
                           (when-let [data (seq (get-in datasets [f t]))]
                             {:title     (str (util/name+ f) " " (name t))
                              :with      'points
                              :linetype  (type->color t)
                              :pointtype (fs->points- f)
                              :data      (map latency-point data)}))
                         (remove nil?))]
    (-> {:preamble           preamble
         :draw-fewer-on-top? true
         :logscale           :y
         :series             series}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn quantiles-graph!
  "Writes a plot of latency quantiles, by f, over time."
  [test history {:keys [subdirectory nemeses]}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        history     (util/history->latencies history)
        dt          1
        qs          [0.5 0.95 0.99 1]
        datasets    (->> history
                         invokes-by-f
                         (util/map-kv
                          (fn [[f ops]]
                            (->> ops
                                 (map latency-point)
                                 (latencies->quantiles dt qs)
                                 (vector f)))))
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        qs->colors- (qs->colors qs)
        output-path (.getCanonicalPath
                     (store/path! test
                                  subdirectory
                                  "latency-quantiles.png"))

        preamble    (latency-preamble test output-path)
        series      (for [f fs, q qs]
                      {:title     (str (util/name+ f) " " q)
                       :with      'linespoints
                       :linetype  (qs->colors- q)
                       :pointtype  (fs->points- f)
                       :data      (get-in datasets [f q])})]
    (-> {:preamble preamble
         :series   series
         :logscale :y}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn rate-preamble
  "Gnuplot commands for setting up a rate plot."
  [test output-path]
  (concat (preamble output-path)
          [[:set :title (str (:name test) " rate")]]
          '[[set ylabel "Throughput (hz)"]]))

(defn rate-graph!
  "Writes a plot of operation rate by their completion times."
  [test history {:keys [subdirectory nemeses]}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        dt          1
        td          (double (/ dt))
        t-max       (->> history (r/map :time) (reduce max 0) util/nanos->secs)
        datasets    (->> history
                         (r/remove op/invoke?)
                         (r/filter (comp integer? :process))
                         (reduce (fn [m op]
                                   (update-in m [(:f op)
                                                 (:type op)
                                                 (bucket-time dt
                                                              (util/nanos->secs
                                                               (:time op)))]
                                              #(+ td (or % 0))))
                                 {}))
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        output-path (.getCanonicalPath (store/path! test
                                                    subdirectory
                                                    "rate.png"))

        preamble (rate-preamble test output-path)
        series   (->> (for [f fs, t types]
                        (when-let [data (get-in datasets [f t])]
                          {:title     (str (util/name+ f) " " (name t))
                           :with      'linespoints
                           :linetype  (type->color t)
                           :pointtype (fs->points- f)
                           :data      (map (juxt identity #(get data % 0))
                                           (buckets dt t-max))}))
                      (remove nil?))]
    (-> {:preamble  preamble
         :series    series}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn open-ops-preamble
  "Gnuplot commands for setting up an open ops plot."
  [test output-path]
  (concat (preamble output-path)
          [[:set :title (str (:name test) " open ops")]]
          '[[set ylabel "Open Ops"]]))

(defn open-ops-graph!
  "Writes a plot of open operations by time."
  [test history {:keys [subdirectory nemeses]}]
  (let [nemeses     (or nemeses (:nemeses (:plot test)))
        dt          1
        td          (double (/ dt))
        t-max       (->> history (r/map :time) (reduce max 0) util/nanos->secs)
        history+     (->> history (history/pair-index+))
        [datasets _] (->> history
                          (r/filter (comp integer? :process))
                          (reduce (fn [[m open-totals] {:keys [type f time] :as op}]
                                    (let [outcome    (->> op (history/completion history+) :type)
                                          open-total (get-in open-totals [f outcome] 0)
                                          open-total (case type
                                                       :invoke           (+ open-total 1)
                                                       (:ok :info :fail) (- open-total 1))]
                                      [(assoc-in m [f
                                                    outcome
                                                    (bucket-time dt
                                                                 (util/nanos->secs time))]
                                                 open-total)
                                       (assoc-in open-totals [f outcome] open-total)]))
                                  [{} {}]))
        fs          (util/polysort (keys datasets))
        fs->points- (fs->points fs)
        output-path (.getCanonicalPath (store/path! test
                                                    subdirectory
                                                    "open-ops.png"))

        preamble (open-ops-preamble test output-path)
        series   (->> (for [f fs t types]
                        (when-let [data (get-in datasets [f t])]
                          (let [[data _] (->> (buckets dt t-max)
                                              (reduce (fn [[acc curr] bt]
                                                        (let [curr (or (get data bt) curr)]
                                                          [(assoc acc bt curr) curr]))
                                                      [{} 0]))]
                            {:title     (str (util/name+ f) " " (name t))
                             :with      'linespoints
                             :linetype  (type->color t)
                             :pointtype (fs->points- f)
                             :data      (map (juxt identity #(get data % 0))
                                             (buckets dt t-max))})))
                      (remove nil?))]
    (-> {:preamble  preamble
         :series    series}
        (with-range)
        (with-nemeses history nemeses)
        plot!
        (try+ (catch [:type ::no-points] _ :no-points)))))

(defn latency-graph
  "Spits out graphs of latencies. Checker options take precedence over
  those passed in with this constructor."
  ([]
   (latency-graph {}))
  ([opts]
   (reify checker/Checker
     (check [_ test history c-opts]
       (let [o (merge opts c-opts)]
         (point-graph!     test history o)
         (quantiles-graph! test history o)
         {:valid? true})))))

(defn rate-graph
  "Spits out graphs of throughput over time. Checker options take precedence
  over those passed in with this constructor."
  ([]
   (rate-graph {}))
  ([opts]
   (reify checker/Checker
     (check [_ test history c-opts]
       (let [o (merge opts c-opts)]
         (rate-graph! test history o)
         {:valid? true})))))

(defn open-ops-graph
  "Spits out graphs of open ops over time. Checker options take precedence
  over those passed in with this constructor."
  ([]
   (open-ops-graph {}))
  ([opts]
   (reify checker/Checker
     (check [_ test history c-opts]
       (let [o (merge opts c-opts)]
         (open-ops-graph! test history o)
         {:valid? true})))))

(defn perf
  "Composes various performance statistics. Checker options take precedence over
  those passed in with this constructor."
  ([]
   (perf {}))
  ([opts]
   (checker/compose {:latency-graph  (latency-graph  opts)
                     :rate-graph     (rate-graph     opts)
                     :open-ops-graph (open-ops-graph opts)})))
