串流組裝

Open-Falcon 的 judge 模組有下列的告警偵測語意：

1. max(#n)
   連續 n 個事件，取其值的最大值，來做條件判斷。

2. min(#n) 連續 n 個事件，取其值的最小值，來做條件判斷。

3. sum(#n) 連續 n 個事件，取其值的總和，來做條件判斷。

4. avg(#n) 連續 n 個事件，取其值的平均值，來做條件判斷。

5. all(#n) 連續 n 個事件，做條件判斷，且所有 (all) 事件都滿足判斷條件。比如cpu.idle all(#3) < 5，表示 cpu.idle 的值連續 3 次小於 5% 觸發告警。

6. diff(#n) 拿最新push上来的点（被减数），与历史最新的3个点（3个减数）相减，得到3个差，只要有一个差满足阈值条件则报警

7. pdiff(#n) 拿最新push上来的点，与历史最新的3个点相减，得到3个差，再将3个差值分别除以减数，得到3个商值，只要有一个商值满足阈值则报警

用 Riemann 來實現 Open-Falcon Judge 的事件偵測語意

原始檔案連結

max(#n)

定義 Riemann 的函數：

i. Riemann 內大多數的 streams 函數，都是只處理單一的 event ，而 Open-Falcon 的 judge 語意卻需要一次看 n 個 events 。所以要使用 moving-event-window 這個函數，來一次取得 n 個 events

ii. 由於 smap 是接在 moving-event-window 之後，所以這個範例中的 smap 它拿到的是一個 events 的 vector 。於是 smap 對於這個 events 的 vector 做了 folds/maximum 的操作，將 events vector 轉換成 event ，再往下一個「串流 (streams)」去傳遞。

(defn maximum
   "falcon max(#n)"
   [n & children]
      (moving-event-window n
         (apply smap folds/maximum children)))

接上串流 (streams) 時的使用方式：

(streams
   (f/maximum 3
      (where (> metric 90)
         #(info "maxmium alert event!" %))))

all(#n)

定義 Riemann 的函數：

i. Riemann 內大多數的 streams 函數，都是只處理單一的 event ，而 Open-Falcon 的 judge 語意卻需要一次看 n 個 events 。所以要使用 moving-event-window 這個函數，來一次取得 n 個 events

ii. 由於 smap 是接在 moving-event-window 之後，所以這個範例中的 smap 它拿到的是一個 events 的 vector 。於是 smap 對於這個 events 的 vector 做自定義的 all-fold 的操作，將 events vector 轉換成 event ，再往下一個「串流 (streams)」去傳遞。

iii. 在 all-fold 這個函數，它會接收一個函數 f 做為參數。對於它收到的 events vector ，它做的處理如下：

1. 移去空 (nil) 的 event 。
2. 移去「值」(metric)為空 (nil) 的 event 。
3. 對於所有的 event 做 f 函數的操作，即 (map f metrics)
4. 如果步驟 3 的結果，全部的值都為「真」，則傳遞最後一個 event 給下一個串流。

(defn- all-fold
  "Return an fn which first do (map f events)
   and will then return the newest event
   if every value in result vector is true.
   Ignore the nil event and nil metric."
   [f]
   (fn [events]
       (when-let [e (some identity events)]
          (let [metrics (folds/non-nil-metrics events)]
             (when (seq metrics)
                (when (every? true? (map f metrics))
                  (last events)))))))

(defn all
   "falcon all(#n)"
   [n f & children]
      (moving-event-window n
         (apply smap (all-fold f) children)))

接上串流 (streams) 時的使用方式：

(streams
   (f/all 3 #(> % 9)
       #(info "all alert event!!!" (show %))))