Sending of Riemann events

example-config.clj

@@ -76,5 +76,7 @@
 
 
              }
-
+ ;; options for riemann client
+ :riemann {:host "localhost"
+           :port 6666}
  }

project.clj

@@ -11,8 +11,8 @@
                  [clj-wallhack "1.0.1"]
                  [clj-webdriver "0.6.1" :exclusions [org.clojure/core.cache]]
                  [cljs-ajax "0.5.3"]
-                 [com.amazonaws/aws-java-sdk-core "1.11.63"]
-                 [com.amazonaws/aws-java-sdk-ec2 "1.11.63"]
+                 [com.amazonaws/aws-java-sdk-core "1.11.77"]
+                 [com.amazonaws/aws-java-sdk-ec2 "1.11.77"]
                  [com.brweber2/clj-dns "0.0.2"]
                  [com.cemerick/url "0.1.1"]
                  [com.fzakaria/slf4j-timbre "0.3.4"]
@@ -46,7 +46,9 @@
                  [secretary "1.2.3"]
                  [slingshot "0.12.2"]
                  [sonian/carica "1.1.0" :exclusions [cheshire]]
-                 [venantius/accountant "0.1.7" :exclusions [org.clojure/tools.reader]]]
+                 [venantius/accountant "0.1.7" :exclusions [org.clojure/tools.reader]]
+                 [com.google.protobuf/protobuf-java "2.6.1"]
+                 [riemann "0.2.12" :exclusions [org.jsoup/jsoup io.netty/netty]]]
   :exclusions [cheshire]
   :scm {:name "git"
         :url "http://github.com/cardforcoin/shale"}

src/clj/shale/core.clj

@@ -12,6 +12,7 @@
             [shale.proxies :as proxies]
             [shale.sessions :as sessions]
             [shale.utils :refer (maybe-resolve-env-keyword) ]
+            [shale.riemann :as riemann]
             [system.components.repl-server :refer [new-repl-server]]
             [ring.adapter.jetty :as jetty])
   (:import clojure.lang.IPersistentMap
@@ -59,11 +60,12 @@
 (defn get-app-system-keyvals [conf]
   [:config conf
    :redis-conn (get-redis-config conf)
+   :riemann (riemann/new-riemann conf)
    :logger (component/using (logging/new-logger) [:config])
    :node-pool (component/using (nodes/new-node-pool conf)
-                               [:redis-conn :logger ])
+                               [:redis-conn :logger :riemann])
    :session-pool (component/using (sessions/new-session-pool conf)
-                                  [:redis-conn :node-pool :proxy-pool :logger])
+                                  [:redis-conn :node-pool :proxy-pool :logger :riemann])
    :proxy-pool (component/using (proxies/new-proxy-pool)
                                 [:config :redis-conn :logger])
    :app (component/using (handler/new-app)

src/clj/shale/nodes.clj

@@ -8,7 +8,8 @@
             [shale.logging :as logging]
             [shale.node-providers :as node-providers]
             [shale.redis :as redis]
-            [shale.utils :refer :all])
+            [shale.utils :refer :all]
+            [shale.riemann :as riemann])
   (:import java.util.UUID))
 
 (deftype ConfigNodeProvider [])
@@ -35,8 +36,9 @@
     {:node-list nodes} (node-providers/new-default-node-provider nodes)
     _ (node-providers/new-default-node-provider ["http://localhost:5555/wd/hub"])))
 
-(s/defrecord NodePool
+(defrecord NodePool
   [redis-conn
+   riemann
    logger
    node-provider
    default-session-limit
@@ -97,6 +99,14 @@
    id   :- s/Str]
   (redis/model-exists? (:redis-conn pool) redis/NodeInRedis id))
 
+(defn notify-node-modify [pool id tags max-sessions]
+  (riemann/send-event
+   (:riemann pool)
+   {:id id
+    :max-sessions max-sessions
+    :state "ok"
+    :tags tags}))
+
 (s/defn modify-node :- NodeView
   "Modify a node's url or tags in Redis."
   [pool :- NodePool
@@ -115,8 +125,15 @@
                                (if url {:url (str url)})
                                (if max-sessions {:max-sessions max-sessions})))
         (when tags (redis/sset-all node-tags-key tags))
+        (notify-node-modify pool id (into [] tags) max-sessions)
         (car/return (view-model pool id))))))
 
+(defn notify-node-create [pool id tags max-sessions]
+  {:id id
+   :state "ok"
+   :max-sessions max-sessions
+   :tags tags})
+
 (s/defn ^:always-validate create-node :- NodeView
   "Create a node in a given pool."
   [pool :- NodePool
@@ -128,10 +145,17 @@
       (let [id (gen-uuid)
             node-key (redis/model-key redis/NodeInRedis id)]
         (car/sadd (redis/model-ids-key redis/NodeInRedis) id)
+        (notify-node-create pool id (into [] tags) max-sessions)
         (car/return (modify-node pool id {:url url
                                           :tags tags
                                           :max-sessions max-sessions}))))))
 
+(defn notify-node-destroy [pool id]
+  (riemann/send-event
+   (:riemann pool)
+   {:id id
+    :state "expired"}))
+
 (s/defn ^:always-validate destroy-node
   [pool :- NodePool
    id   :- s/Str]
@@ -143,14 +167,19 @@
           (node-providers/remove-node (:node-provider pool) url)))
       (finally
         (redis/delete-model! (:redis-conn pool) redis/NodeInRedis id)
-        (car/del (redis/node-tags-key id)))))
+        (car/del (redis/node-tags-key id))
+        (notify-node-destroy pool id))))
   true)
 
 (defn ^:private to-set [s]
   (into #{} s))
 
 (def ^:private refresh-nodes-lock {})
 
+;; TODO
+;; - How should refresh riemann event look like?
+;; - Should it send also destroy and create event?
+
 (s/defn refresh-nodes
   "Syncs the node list with the backing node provider."
   [pool :- NodePool]

src/clj/shale/riemann.clj

@@ -0,0 +1,23 @@
+(ns shale.riemann
+  (require [riemann.client :as riemann]
+           [shale.logging :as logging]
+           [com.stuartsierra.component :as component]))
+
+(defrecord Riemann [config client]
+  component/Lifecycle
+  (start [cmp]
+    (logging/info "Starting the riemann client...")
+    (assoc cmp :client (riemann/tcp-client (:riemann config))))
+  (stop [cmp]
+    (logging/info "Stopping the riemann client...")
+    (assoc cmp :client nil)))
+
+(defn new-riemann [config]
+  (map->Riemann {:config config}))
+
+(defn send-event [riemann m]
+  (let [cl (:client riemann)
+        ev (assoc m :service "shale")]
+    (logging/debug (str "Sended to riemann: " ev))
+    @(riemann/send-event cl ev)))
+

src/clj/shale/sessions.clj

@@ -24,7 +24,8 @@
                                      resume-webdriver
                                      maybe-add-no-sandbox
                                      to-async
-                                     webdriver-capabilities]])
+                                     webdriver-capabilities]]
+            [shale.riemann :as riemann])
   (:import org.openqa.selenium.WebDriverException
            org.openqa.selenium.remote.UnreachableBrowserException
            org.xbill.DNS.Type
@@ -36,6 +37,7 @@
 
 (s/defrecord SessionPool
   [redis-conn
+   riemann
    node-pool
    proxy-pool
    logger
@@ -128,11 +130,16 @@
 (declare view-model view-model-exists? view-models view-model-ids
          resume-webdriver-from-id destroy-session)
 
+(defn notify-session-create-error [pool]
+  (riemann/send-event
+   (:riemann pool)
+   {:state "error"}))
+
 (defn start-webdriver!
   "Create a webdriver. Optionally specify a timeout in milliseconds.
 
   Throws an exception on timeout, but blocks forever by default."
-  [node capabilities & {:keys [timeout]}]
+  [pool node capabilities & {:keys [timeout]}]
   (logging/infof "start-webdriver! %s" node)
   (let [future-wd (future (new-webdriver node capabilities))
         wd (if (or (nil? timeout) (= 0 timeout))
@@ -141,7 +148,7 @@
     (if (= wd ::timeout)
         (do
           (future-cancel future-wd)
-          ;; TODO send to riemann
+          (notify-session-create-error pool)
           (logging/warn (format
                   "Timeout starting new webdriver on node %s"
                   node))
@@ -151,6 +158,12 @@
                       :node-url node})))
         wd)))
 
+(defn notify-session-resume-error [pool id]
+  (riemann/send-event
+   (:riemann pool)
+   {:id id
+    :state "error"}))
+
 (defn resume-webdriver-from-id
   "Resume and return a web driver from a session id (versus a webdriver id).
   Optionally include a timeout, in milliseconds.
@@ -172,7 +185,7 @@
       (if (= wd ::timeout)
         (do
           (future-cancel future-wd)
-          ;; TODO send to riemann
+          (notify-session-resume-error pool id)
           (logging/warn (format
                   "Timeout resuming session %s after %d ms against node %s"
                   webdriver-id
@@ -184,6 +197,7 @@
                       :timeout timeout
                       :node-url node-url})))
         wd))
+    ;; TODO Send riemann event for this?
     (throw
       (ex-info "Unknown session id." {:session-id id}))))
 
@@ -257,6 +271,15 @@
       :add (clojure.set/union #{tag} tags)
       :remove (disj tags tag))))
 
+
+(defn notify-session-modify [pool id tags reserved]
+  (riemann/send-event
+   (:riemann pool)
+   {:id id
+    :tags tags
+    :reserved reserved
+    :state "ok"}))
+
 (s/defn ^:always-validate modify-session :- (s/maybe SessionView)
   "Modify an existing session."
   [pool          :- SessionPool
@@ -284,14 +307,18 @@
                                 {:tags new-tags}))]
       (if (or (not go-to-url)
               (session-go-to-url-or-destroy-session id go-to-url))
-        (save-session-diff-to-redis pool id session-diff)))
+        (save-session-diff-to-redis pool id session-diff))
+      (let [{:keys [tags reserved]} session-diff
+            tags (into [] (map name tags))]
+        (notify-session-modify pool id tags reserved)))
     (view-model pool id)))
 
 (s/defn ^:always-validate get-node-or-err :- nodes/NodeView
   [pool      :- NodePool
    node-req  :- (s/maybe nodes/NodeRequirement)]
   (let [node (nodes/get-node pool node-req)]
     (when (nil? node)
+      (notify-session-create-error pool)
       (throw
         (ex-info "No suitable node found!"
                  {:user-visible true :status 503})))
@@ -312,16 +339,26 @@
    :reserved false
    :proxy-id nil})
 
+(defn notify-session-create [pool id tags reserved browser-name node-id]
+  (riemann/send-event
+   (:riemann pool)
+   {:id id
+    :tags tags
+    :reserved reserved
+    :browser-name browser-name
+    :node-id node-id
+    :state "ok"}))
+
 (s/defn ^:always-validate create-session :- SessionView
   [pool :- SessionPool
    options :- CreateArg]
   (logging/info (format "Creating a new session.\nOptions: %s"
                         (pretty options)))
-
-  (if (= 0 (count (nodes/nodes-under-capacity (:node-pool pool))))
+  (when (= 0 (count (nodes/nodes-under-capacity (:node-pool pool))))
+    (notify-session-create-error pool)
     (throw
-      (ex-info "All nodes are over capacity!"
-               {:user-visible true :status 503})))
+     (ex-info "All nodes are over capacity!"
+              {:user-visible true :status 503})))
   (let [{:keys [browser-name
                 capabilities
                 node-require
@@ -356,11 +393,16 @@
                                  requested-capabilities)
         id (gen-uuid)
         wd (start-webdriver!
+             pool
              (:url node)
              requested-capabilities
              :timeout (:start-webdriver-timeout pool))
         webdriver-id (remote-webdriver/session-id wd)
         actual-capabilities (webdriver-capabilities wd)]
+
+    (let [tags (into [] (map name tags))
+          node-id (:id node)]
+      (notify-session-create pool id tags reserved browser-name node-id))
     (last
       (car/wcar (:redis-conn pool)
         (car/sadd (redis/model-ids-key redis/SessionInRedis) id)
@@ -451,6 +493,9 @@
                           modifications)
           candidate)))))
 
+;; TODO
+;; - Send riemann error event for following catches?
+
 (s/defn ^:always-validate destroy-webdriver!
   [pool         :- SessionPool
    webdriver-id :- s/Str
@@ -504,6 +549,12 @@
     (when immediately
       (deref deferred))))
 
+(defn notify-session-destroy [pool id]
+  (riemann/send-event
+   (:riemann pool)
+   {:id id 
+    :state "expired"}))
+
 (defn destroy-session
   [pool id & {:keys [immediately] :or {immediately true}}]
   (s/validate SessionPool pool)
@@ -526,7 +577,8 @@
                             webdriver-id
                             node-url
                             (boolean immediately)
-                            hard-delete))))
+                            hard-delete)
+        (notify-session-destroy pool id))))
   true)
 
 (def view-model-defaults {:current-url nil
@@ -579,6 +631,10 @@
        (filter #(= (:webdriver-id %) webdriver-id))
        first))
 
+;; TODO
+;; - How should refresh riemann event look like?
+;; - Should it send also destroy and create event?
+
 (s/defn ^:always-validate refresh-session
   [pool :- SessionPool
    id   :- s/Str]

test-config.clj

@@ -2,4 +2,5 @@
  :node-max-sessions 6
  :port 5000
  :webdriver-timeout 3000
- :start-webdriver-timeout 5000}
+ :start-webdriver-timeout 5000
+ :riemann {:host "localhost" :port 6666}}