Add filter-based rotation of aliases

.github/workflows/ci.yaml

@@ -0,0 +1,22 @@
+---
+name: CI
+
+on:
+  push:
+    branches:
+      - master
+  pull_request:
+    branches:
+      - master
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v2
+      - name: Setup Clojure
+        run: |
+          wget --quiet https://raw.githubusercontent.com/technomancy/leiningen/stable/bin/lein
+          chmod +x lein
+      - name: Run the test suite
+        run: ./lein test

project.clj

@@ -9,7 +9,9 @@
               [[cheshire "5.7.0"]
                [org.clojure/clojure "1.8.0"]
                [riemann "0.3.8"]
+               [com.fasterxml.jackson.core/jackson-core "2.10.0"]
                [clj-time "0.13.0"]
+               [cc.qbits/commons "0.5.1"]
                [org.clojure/tools.logging "0.3.1"]]}}
   :plugins [[lein-rpm "0.0.5"
              :exclusions [org.apache.maven/maven-plugin-api

src/riemann/plugin/samplerr.clj

@@ -485,3 +485,131 @@
   (let [service (apply rotation-service opts)]
     (swap! riemann.config/next-core core/conj-service service :force)))
 
+(defn index-parts
+  "Match the year, month and day out of an index name"
+  [index]
+  (let [index-name (:index index)]
+    (rest (re-find #"(\d{4})(?:\.(\d{2})(?:\.(\d{2}))?)?$" index-name))))
+
+(defn index-year
+  "Return the year of an index as an Integer"
+  [index]
+  (Integer. (first (index-parts index))))
+
+(defn index-month
+  "Return the month of an index as an Integer"
+  [index]
+  (Integer. (or (second (index-parts index)) "1")))
+
+(defn index-day
+  "Return the day of an index as an Integer"
+  [index]
+  (Integer. (or (nth (index-parts index) 2) "1")))
+
+(defn yearly-index?
+  "Return true if an index has yearly data"
+  [index]
+  (= nil (second (index-parts index))))
+
+(defn monthly-index?
+  "Return true if an index has monthly data"
+  [index]
+  (and
+    (not (yearly-index? index))
+    (= nil (nth (index-parts index) 2))))
+
+(defn daily-index?
+  "Return true if an index has daily data"
+  [index]
+  (and
+    (not (yearly-index? index))
+    (not (monthly-index? index))))
+
+(defn index-start-timestamp
+  "Return an ISO timestamp of the time the index starts at"
+  [index]
+  (format "%02d-%02d-%02dT00:00:00Z" (index-year index) (index-month index) (index-day index)))
+
+(defn compact
+  "Remove nil items from a collection"
+  [coll]
+  (remove nil? coll))
+
+(defn remove-existing-aliases-query
+  "Return a query to remove all the passed aliases"
+  [aliases]
+  {:url "_aliases" :method :post :body {:actions (map #(if true {:remove {:index (:index %) :alias (:alias %)}}) aliases)}})
+
+(defn create-aliases-query
+  "Return a query to create the aliases to query all passed indices"
+  [indices index-prefix alias-prefix]
+  (let [yearly-indices      (filter #(re-find #"\d{4}$" (:index %)) indices)
+        monthly-indices     (filter #(re-find #"\d{4}\.\d{2}$" (:index %)) indices)
+        daily-indices       (filter #(re-find #"\d{4}\.\d{2}\.\d{2}$" (:index %)) indices)
+        first-monthly-index (first monthly-indices)
+        first-daily-index   (first daily-indices)
+        yearly-aliases      (compact
+                              (map #(if (< (index-year %) (index-year first-monthly-index))
+                                      {:add {:index (:index %) :alias (clojure.string/replace (:index %) index-prefix alias-prefix)}}
+                                      (if (= (index-year %) (index-year first-monthly-index))
+                                        {:add {:index (:index %) :alias (clojure.string/replace (:index %) index-prefix alias-prefix) :filter {:range {"@timestamp" {:lt (index-start-timestamp first-monthly-index)}}}}})
+                                      ) yearly-indices))
+        monthly-aliases     (compact
+                              (map #(if (or (< (index-year %) (index-year first-daily-index))
+                                            (and (= (index-year %) (index-year first-daily-index))
+                                                 (< (index-month %) (index-month first-daily-index))))
+                                      {:add {:index (:index %) :alias (clojure.string/replace (:index %) index-prefix alias-prefix)}}
+                                      (if (and (= (index-year %) (index-year first-daily-index))
+                                               (= (index-month %) (index-month first-daily-index)))
+                                        {:add {:index (:index %) :alias (clojure.string/replace (:index %) index-prefix alias-prefix) :filter {:range {"@timestamp" {:lt (index-start-timestamp first-daily-index)}}}}})
+                                      ) monthly-indices))
+        daily-aliases     (map #(if true {:add {:index (:index %)  :alias (clojure.string/replace (:index %) index-prefix alias-prefix)}}) daily-indices)]
+    {:url "_aliases" :method :post :body {:actions (concat yearly-aliases monthly-aliases daily-aliases)}}))
+
+(defn update-aliases
+  "Maintain samplerr aliases"
+  [{:keys [conn index-prefix alias-prefix] :or {alias-prefix "samplerr-" index-prefix ".samplerr-"}}]
+  (let [response (es/request conn {:url (str "_cat/aliases/" alias-prefix "*?format=json&s=alias") :method :get})
+        aliases  (:body response)]
+    (if (seq aliases)
+      (do
+        (info "Removing samplerr aliases")
+        (es/request conn (remove-existing-aliases-query aliases)))))
+  (let [response            (es/request conn {:url (str "_cat/indices/" index-prefix "*?format=json&s=index") :method :get})
+        indices             (:body response)]
+    (if (seq indices)
+      (do
+        (info "Creating samplerr aliases")
+        (es/request conn (create-aliases-query indices index-prefix alias-prefix))))))
+
+(defn maintain
+  "Perform samplerr maintenance"
+  [{:keys [conn alias-prefix index-prefix archives purge? update-aliases?]
+    :or {alias-prefix "samplerr-"
+         index-prefix ".samplerr-"
+         purge? false
+         update-aliases? true}}]
+  (if purge?
+    (purge {:conn conn :index-prefix index-prefix :archives archives}))
+  (if update-aliases?
+    (update-aliases {:conn conn :index-prefix index-prefix :alias-prefix alias-prefix})))
+
+(defn periodically-maintain
+  "Periodically perform samplerr maintenance"
+  [{:keys [conn archives alias-prefix index-prefix purge? update-aliases?]
+    :or {alias-prefix "samplerr-"
+         index-prefix ".samplerr-"
+         purge? false
+         update-aliases? true}}]
+  (let [rotation-interval 86400
+        rotation-delay 60
+        beginning-of-last-rotation-interval (* (quot (unix-time) rotation-interval) rotation-interval)
+        beginning-of-next-rotation-interval (riemann.time/next-tick beginning-of-last-rotation-interval rotation-interval)
+        daily-rotation-delay (+ beginning-of-next-rotation-interval rotation-delay)
+        do-maintain (fn [] (maintain {:conn conn :alias-prefix alias-prefix :index-prefix index-prefix :archives archives :purge? purge? :update-aliases? update-aliases?}))]
+    (info (str "Scheduling startup samplerr maintenance in " rotation-delay " seconds"))
+    (riemann.time/after! rotation-delay do-maintain)
+    (info (str "Scheduling daily samplerr maintenance at " (.format (java.text.SimpleDateFormat. "yyyy-MM-dd'T'HH:mm:ssXXX") (* 1000 daily-rotation-delay))))
+    (riemann.time/once!
+      daily-rotation-delay
+      (fn [] (riemann.time/every! rotation-interval do-maintain)))))

test/riemann/plugin/samplerr_test.clj

@@ -0,0 +1,91 @@
+(ns riemann.plugin.samplerr-test
+  (:require [clojure.test :refer :all]
+            [riemann.plugin.samplerr :refer :all]
+            ))
+
+(deftest identify-index-date
+  (let [index {:index "samplerr-2020"}]
+    (is (= true                   (yearly-index? index)))
+    (is (= false                  (monthly-index? index)))
+    (is (= false                  (daily-index? index)))
+    (is (= '("2020" nil nil)      (index-parts index)))
+    (is (= 1                      (index-day index)))
+    (is (= 1                      (index-month index)))
+    (is (= 2020                   (index-year index)))
+    (is (= "2020-01-01T00:00:00Z" (index-start-timestamp index))))
+  (let [index {:index "samplerr-2020.04"}]
+    (is (= false                  (yearly-index? index)))
+    (is (= true                   (monthly-index? index)))
+    (is (= false                  (daily-index? index)))
+    (is (= '("2020" "04" nil)     (index-parts index)))
+    (is (= 1                      (index-day index)))
+    (is (= 4                      (index-month index)))
+    (is (= 2020                   (index-year index)))
+    (is (= "2020-04-01T00:00:00Z" (index-start-timestamp index))))
+  (let [index {:index "samplerr-2020.04.22"}]
+    (is (= false                  (yearly-index? index)))
+    (is (= false                  (monthly-index? index)))
+    (is (= true                   (daily-index? index)))
+    (is (= '("2020" "04" "22")    (index-parts index)))
+    (is (= 22                     (index-day index)))
+    (is (= 4                      (index-month index)))
+    (is (= 2020                   (index-year index)))
+    (is (= "2020-04-22T00:00:00Z" (index-start-timestamp index)))))
+
+(deftest test-remove-existing-aliases-query
+  (let [aliases '({:index ".samplerr-2021"       :alias "samplerr-2021"}
+                  {:index ".samplerr-2022"       :alias "samplerr-2022"}
+                  {:index ".samplerr-2022.12"    :alias "samplerr-2022.12"}
+                  {:index ".samplerr-2022.12.04" :alias "samplerr-2022.12.04"})]
+    (is (= {:url "_aliases"
+            :method :post
+            :body {:actions '({:remove {:index ".samplerr-2021"       :alias "samplerr-2021"}}
+                              {:remove {:index ".samplerr-2022"       :alias "samplerr-2022"}}
+                              {:remove {:index ".samplerr-2022.12"    :alias "samplerr-2022.12"}}
+                              {:remove {:index ".samplerr-2022.12.04" :alias "samplerr-2022.12.04"}})}}
+           (remove-existing-aliases-query aliases)))))
+(deftest test-create-aliases-query
+  (let [indices '({:index ".samplerr-2021"}
+                  {:index ".samplerr-2022"}
+                  {:index ".samplerr-2022.10"}
+                  {:index ".samplerr-2022.11"}
+                  {:index ".samplerr-2022.12"}
+                  {:index ".samplerr-2023"}
+                  {:index ".samplerr-2023.01"}
+                  {:index ".samplerr-2023.01.23"}
+                  {:index ".samplerr-2023.01.24"}
+                  {:index ".samplerr-2023.01.25"}
+                  {:index ".samplerr-2023.01.26"}
+                  {:index ".samplerr-2023.01.27"}
+                  {:index ".samplerr-2023.01.28"}
+                  {:index ".samplerr-2023.01.29"}
+                  {:index ".samplerr-2023.01.30"}
+                  {:index ".samplerr-2023.01.31"}
+                  {:index ".samplerr-2023.02"}
+                  {:index ".samplerr-2023.02.01"}
+                  {:index ".samplerr-2023.02.02"}
+                  {:index ".samplerr-2023.02.03"}
+                  {:index ".samplerr-2023.02.04"})]
+    (is (= {:url "_aliases"
+            :method :post
+            :body {:actions '({:add {:index ".samplerr-2021"       :alias "samplerr-2021"}}
+                              {:add {:index ".samplerr-2022"       :alias "samplerr-2022"       :filter {:range {"@timestamp" {:lt "2022-10-01T00:00:00Z"}}}}}
+                              {:add {:index ".samplerr-2022.10"    :alias "samplerr-2022.10"}}
+                              {:add {:index ".samplerr-2022.11"    :alias "samplerr-2022.11"}}
+                              {:add {:index ".samplerr-2022.12"    :alias "samplerr-2022.12"}}
+                              {:add {:index ".samplerr-2023.01"    :alias "samplerr-2023.01"    :filter {:range {"@timestamp" {:lt "2023-01-23T00:00:00Z"}}}}}
+                              {:add {:index ".samplerr-2023.01.23" :alias "samplerr-2023.01.23"}}
+                              {:add {:index ".samplerr-2023.01.24" :alias "samplerr-2023.01.24"}}
+                              {:add {:index ".samplerr-2023.01.25" :alias "samplerr-2023.01.25"}}
+                              {:add {:index ".samplerr-2023.01.26" :alias "samplerr-2023.01.26"}}
+                              {:add {:index ".samplerr-2023.01.27" :alias "samplerr-2023.01.27"}}
+                              {:add {:index ".samplerr-2023.01.28" :alias "samplerr-2023.01.28"}}
+                              {:add {:index ".samplerr-2023.01.29" :alias "samplerr-2023.01.29"}}
+                              {:add {:index ".samplerr-2023.01.30" :alias "samplerr-2023.01.30"}}
+                              {:add {:index ".samplerr-2023.01.31" :alias "samplerr-2023.01.31"}}
+                              {:add {:index ".samplerr-2023.02.01" :alias "samplerr-2023.02.01"}}
+                              {:add {:index ".samplerr-2023.02.02" :alias "samplerr-2023.02.02"}}
+                              {:add {:index ".samplerr-2023.02.03" :alias "samplerr-2023.02.03"}}
+                              {:add {:index ".samplerr-2023.02.04" :alias "samplerr-2023.02.04"}}
+                              )}}
+           (create-aliases-query indices ".samplerr-" "samplerr-")))))