gistfile1.clj

; Pick up service x events, sent by error detection in service 'y' stream
(streams
  (where (service "x")
         ;Index API service events for 60s we can watch for expiries info
         (with :ttl 60 (update-index (index)))

         (changed-state {:init "ok"} 
                        (service-x-state-change-email "errors@email.com"))

         (expired (with :state "ok" reinject))))

; Handle errors passed from dotcom MVC or node apps
(streams
  (where (service "y")

             (tagged "Log"

                 ;And high severity
                 (where (or (tagged "Error")
                            (tagged "Fatal"))

                        (where (tagged "live")
                               ;Send all logs to papertrail
                               log-to-papertrail

                               ;Looks like the API is down (more than 5 errors in 30s)
                               (where (description "service x errored")
                                      (fixed-time-window 30
                                         (fn [events]
                                           (if (> (count events) 5)

                                             ;Send back to top level streams with 60s TTL
                                             ((with {:host "service-x-host" 
                                                     :state "critical" 
                                                     :service "x"
                                                     :error-count (count events)}
                                                reinject) (first event)))))))

Here's an alternative, staring from L25

(where (description "service x errored")
  (with :metric 1.0
    (moving-time-window 30
      (smap folds/sum
        (over 5
          (with {:state "critical" :description "elevated error rate" :state "critical" :service "x"}
            store))))))

This assumes your streams are defined like so:

(let [index   (default :ttl 60 (index))
      mailer  (email "riemann")
      alert!    (changed-state {:init "ok"} (mailer "somewhere")
      store    (sdo alert! index)]
  (streams
    ...))