Added Druid plugin (http://druid.io)

project.clj

@@ -80,6 +80,7 @@
                    :bench :bench
                    :focus :focus
                    :slack :slack
+                   :druid :druid
                    :cloudwatch :cloudwatch
                    :datadog :datadog
                    :stackdriver :stackdriver

src/riemann/config.clj

@@ -11,6 +11,7 @@
                      [common      :as common :refer [event]]
                      [core        :as core]
                      [datadog     :refer [datadog]]
+                     [druid       :refer [druid]]
                      [email       :refer [mailer]]
                      [folds       :as folds]
                      [graphite    :as graphite-client :refer [graphite]]

src/riemann/druid.clj

@@ -0,0 +1,64 @@
+(ns riemann.druid
+  "Forwards events to Druid"
+  (:require
+    [clj-http.client :as http]
+    [cheshire.core :refer [generate-string]]
+    [riemann.common :refer [unix-to-iso8601]]))
+
+(defn post-datapoint
+  "Post the riemann metrics as datapoints."
+  [host port dataset json-data]
+  (let [p (str port)
+        scheme "http://"
+        endpoint "/v1/post/"
+        url (str scheme host ":" p endpoint dataset)
+        http-options {:body json-data
+                      :content-type :json
+                      :conn-timeout 5000
+                      :socket-timeout 5000
+                      :throw-entire-message? true}]
+    (http/post url http-options)))
+
+(defn generate-event [event]
+  {:host        (:host event)
+   :service     (:service event)
+   :state       (:state event)
+   :timestamp   (unix-to-iso8601 (:time event))
+   :tags        (:tags event)
+   :description (:description event)
+   :value       (:metric event)})
+
+(defn druid
+  "Returns a function which accepts single events or batches of
+   events in a vectorand send them to Druid Tranquility Server.
+
+   Usage:
+   (druid {:host \"druid.example.com\"})
+
+   Options:
+   `:host`     Hostname of Druid Tranquility server. (default: `\"localhost\"`)
+   `:port`     Port at which Druid Tranquility is listening (default: `8200`)
+   `:dataset`  Dataset name to be given (default: `\"riemann\"`)
+
+   Example:
+   (def druid-async
+   (batch 100 1/10
+     (async-queue!
+       :druid-async          ; A name for the forwarder
+       {:queue-size     1e4  ; 10,000 events max
+        :core-pool-size 5    ; Minimum 5 threads
+        :max-pools-size 100} ; Maxium 100 threads
+        (druid {:host \"localhost\"}))))
+  "
+  [opts]
+  (let [opts (merge {:host     "localhost"
+                     :port     8200
+                     :dataset  "riemann"}
+                    opts)]
+    (fn [event]
+      (let [events (if (sequential? event)
+                     event
+                     [event])
+            post-data (mapv generate-event events)
+            json-data (generate-string post-data)]
+        (post-datapoint (:host opts) (:port opts) (:dataset opts) json-data)))))

test/riemann/druid_test.clj

@@ -0,0 +1,27 @@
+(ns riemann.druid-test
+  (:require [clj-http.client :as client]
+            [clojure.test :refer :all]
+            [riemann.druid :as druid]
+            [riemann.logging :as logging]
+            [riemann.test-utils :refer [with-mock]]))
+
+(logging/init)
+
+(deftest ^:druid druid-test
+  (with-mock [calls client/post]
+    (let [d (druid/druid {:host "localhost"})]
+
+      (testing "an event with metric")
+      (d {:host    "testhost"
+          :service "testservice"
+          :metric  42
+          :time    123456789
+          :state   "ok"})
+      (is (= 1 (count @calls)))
+      (is (= (vec (last @calls))
+             ["http://localhost:8200/v1/post/riemann"
+              {:body "[{\"host\":\"testhost\",\"service\":\"testservice\",\"state\":\"ok\",\"timestamp\":\"1973-11-29T21:33:09.000Z\",\"tags\":null,\"description\":null,\"value\":42}]"
+               :socket-timeout 5000
+               :conn-timeout 5000
+               :content-type :json
+               :throw-entire-message? true}])))))