riemann.config

(use '[cemerick.pomegranate :only (add-dependencies)])

(add-dependencies :coordinates '[[org.clojure/test.check "0.9.0"]
                                 [cheshire "5.6.3"]
                                 [clj-time "0.13.0"]
                                 [org.clojure/tools.trace "0.7.9"]]
                  :repositories (merge cemerick.pomegranate.aether/maven-central
                                       {"clojars" "http://clojars.org/repo"}))

(require '[cheshire.core :refer [parse-stream]]
         '[clojure.java.io :refer [reader file]]
         '[robinhood.notifiers.log :refer [log-info log-warning log-debug]]
         '[robinhood.notifiers.elasticsearch :refer [push-to-es]]
         '[riemann.folds :as folds]
         '[riemann.test :as test])

(let [config-dir (.getParent (file *config-file*))
      separator (java.io.File/separator)]

  (def rh_config (parse-stream
                  (reader (str config-dir separator "config.json"))))

  (logging/init {:file (str (rh_config "logfile"))})
  (logging/set-level ch.qos.logback.classic.Level/INFO))

(defn opentsdb-parser [event]
  (assoc event
    :tags (clojure.string/split (:roles event) #",")
    :metric-name (:description event)))

; Listen on the all interfaces over TCP (5555), UDP (5555), websockets
; (5556) and opentsdb (4242)
(let [all-interfaces "0.0.0.0"
      option-map {:host all-interfaces}]
  (tcp-server      option-map)
  (udp-server      option-map)
  (ws-server       option-map)
  (repl-server     option-map)
  (opentsdb-server (assoc option-map :parser-fn opentsdb-parser)))

(def index (riemann.config/index))
(def watcher-streams (atom []))
(def keep-list [:host
                :metric-name
                :roles
                :rule-id
                :service
                :service-status
                :service-description])

(require '[robinhood.consumers.all :as consumers]
         'robinhood.tests.all)

(let [consumer-streams (flatten (doall consumers/all-streams))
      combined-stream (apply sdo consumer-streams)
      interesting-metrics (set (flatten consumers/interesting-metrics))
      log-extraneous-event-count (fixed-time-window
                                  60
                                  (smap folds/count
                                        #(log-warning
                                          (str "Got " (:metric %)
                                               " extraneous events in the last 60 seconds."))))
      log-expired-count (fixed-time-window
                         60
                         (smap folds/count
                               #(log-warning
                                 (str "Got " (:metric %)
                                      " expired events in the last 60 seconds."))))]

  (log-info (str "Filtering events through " (count consumer-streams) " streams."))
  (log-info (str "We are only interested in " (count interesting-metrics)
                 " metrics types, the rest are extraneous."))

  ; Expire old events from the index every 5 seconds.
  (periodically-expire 30 {:keep-keys keep-list})

  (streams
   (default :ttl (rh_config "default_ttl")
     (where* expired?
             (where* #(not= "expired" (:state %))
                     log-expired-count
                     (else
                      (apply sdo @watcher-streams)))
             (else
              (with :es_log_probability (rh_config "es_log_probability" 0.01)
                    (test/io push-to-es))
              (where* #(contains? interesting-metrics (:metric-name %))
                      combined-stream
                      (else log-extraneous-event-count)))))))