Riemann - Build a stream dynamically from a map

Question

I have the following function which gets a map with service name and threshold. It checks if the service crossed a defined threshold and then calls multiple downstream children on the event.

(defn tc
  [s & children]
   (where
     (and (service (:service_name s)) (not (expired? event)))
       (by [:host :service]
         (where (> metric (:threshold s)
           (with :state "critical" 
             (apply sdo children)))))))

I would like to build a stream dynamically using a vector of maps:

(def services [{:service "cpu/usage" :threshold 90}
               {:service "memory/usage" :threshold 90}])

When trying to run it in a stream i'm getting the following warning:

(streams
  (doseq [s services] (tc s prn)))

WARN [2015-01-05 14:27:07,187] Thread-15 - riemann.core - instrumentation service caught
java.lang.NullPointerException
  at riemann.core$stream_BANG_$fn__11140.invoke(core.clj:19)
  at riemann.core$stream_BANG_.invoke(core.clj:18)
  at riemann.core$instrumentation_service$measure__11149.invoke(core.clj:57)
  at riemann.service.ThreadService$thread_service_runner__8782$fn__8783.invoke(service.clj:66)
  at riemann.service.ThreadService$thread_service_runner__8782.invoke(service.clj:65)
  at clojure.lang.AFn.run(AFn.java:22)
  at java.lang.Thread.run(Thread.java:701)

It works, if i run the streams function inside the doseq. This one works and gives the following output:

(doseq [s services]
  (streams (tc s prn)))

#riemann.codec.Event{:host "testhost", :service "memory/usage", :state "critical", :description nil, :metric 91.0, :tags nil, :time 1420460856, :ttl 60.0}

are those your exact events or have they been elided for the question? — Arthur Ulfeldt, Jan 05 '15 at 23:17
@ArthurUlfeldt, I'm not sure that the problem is in the event. Since the function does work, if i use the following code (doseq [s services] (streams (tc s prn))). I don't want to create a stream for each services, but one streams to all services. — Igor Liner, Jan 06 '15 at 08:20
`(with :service "everything" ... process combined stream here ..) `would combine everything into a single service — Arthur Ulfeldt, Jan 06 '15 at 19:49
@ArthurUlfeldt, what i would like to do is have a vector of maps. Each map describe a service and a threshold. I would like to check that if a service crosses the defined threshold, then the event state is set to "critical". What i'm trying to do is to run the streams function using this dataset. — Igor Liner, Jan 07 '15 at 08:07

score 0 · Answer 1 · answered Jan 05 '15 at 23:15

It seems to blow up if your events don't have all the required fields, here's a sample from a similar project where I build an event from a sequence of events (reducing) It's not exactly what you are doing though I'm generating events in the same way:

{:service (:service (first events))
 :metric (->> events count)
 :host "All-counts"
 :state "OK"
 :time (:time (last events))
 :ttl default-interval}

I got NPE specifically when time was missing. If you can't inherit it form somewhere, just make it up (use now for instance) without a reasonable value here, event expiration will not work and you'll run out of RAM

Riemann - Build a stream dynamically from a map

1 Answers1