index-utils.clj

(require '[clojure.java.io :as io]
         '[clojure.tools.logging :refer [infof]]
         '[riemann.common :refer [encode decode-inputstream]]
         '[riemann.config :refer [service!]]
         '[riemann.service :refer [thread-service]]
         '[riemann.time :refer [unix-time]])
(import '[java.io.File FileOutputStream])
  
(defn index-save
  [file index]
  (let [file       (.getAbsoluteFile (io/file file))
        dir        (.getParentFile file)
        events     (seq index)
        tmp-suffix (str (long (* 1000 (unix-time))))
        tmp-file   (File/createTempFile (.getName file) tmp-suffix dir)]
    (infof "Writing %d events to state-file %s" (count events) tmp-file)
    (with-open [w (new FileOutputStream tmp-file)]
      (.deleteOnExit tmp-file)
      (.write w (encode {:events events}))
      (.flush w)
      (-> w (.getChannel) (.force true)))
    (infof "Swapping state-file: %s => %s" tmp-file file)
    (.renameTo tmp-file file)))

(defn index-save-service
  [interval file]
  (thread-service
    ::index-save [file]
    (bound-fn worker [core]
      (Thread/sleep (* 1000 interval))

      (index-save file (:index core)))))

(defn periodically-save-index
  [interval file]
  (service! (index-save-service interval file)))

(defn index-restore
  ([file]
   (let [index (or (:index @riemann.config/next-core) (:index @riemann.config/core))]
     (index-restore file index)))
  ([file index]
   (let [file (.getAbsoluteFile (io/file file))]
     (if (.isFile file)
       (with-open [r (io/input-stream file)]
         (let [msg    (decode-inputstream r)
               events (:events msg)]
           (infof "Restoring %d events from state-file %s" (count events) file)
           (doseq [event events]
             (index event))))))))