Use Riemann for auto scale of VM simple example to start

[Dan Hunt]

I have grand plans for using Riemann to allow users to specify metrics and actions for VM processing.

However I cannot get off the starting block due to my lack of a "ding" moment as Pete put's it.

I am trying to start simple by detecting the state change of the service metric called load.

I want to call a url on the state change and pass the state.

What I get is the url being called when Riemann first runs but nothing afterwards.

I see in the std output the info and the state changing but I do not get the url called.

I also cannot get the actual state to appear in the post (have tried :state but that did not work ... however this is a nice to have considering I am failing on the basics).

(require '[clj-http.client :as client])

(let [host "127.0.0.1"]

  (tcp-server {:host host})

  (udp-server {:host host})

  (ws-server  {:host host}))

(def graph (graphite {:host "localhost"}))

(defn srims [msg]

  (fn [event]

    #(info "this never appears" %)

  )

)

(let [index (index)]

  ; Inbound events will be passed to these streams:

  (streams

    ; graph all

    graph

    (where (and (service "load") (> metric 20))

      (srims "testing")

    )

    (by [:host :service]

      (where (service "load")

        (changed :state

          (srims "testing")

          #(info "this appears:" %)

         ; the call to the service happens at the start but never again

         (client/post "http://10.111.96.222:8080/cloud-srims/event" {:query-params {"q" "state", "cb" "fred"}})

        )

      )

    )

  )

)

I see this in the stdoutput as I run stress to push the load up.

UDP server 127.0.0.1 5555 16384 online

INFO [2015-08-16 17:28:40,878] main - riemann.core - Hyperspace core online

INFO [2015-08-16 17:28:43,476] defaultEventExecutorGroup-2-1 - riemann.config - this appears: #riemann.codec.Event{:host vagrant-ubuntu-trusty-64, :service load, :state ok, :description 1-minute load average/core is 0.58, :metric 0.58, :tags nil, :time 1439746123, :ttl 10.0}

INFO [2015-08-16 17:29:30,878] defaultEventExecutorGroup-2-1 - riemann.config - this appears: #riemann.codec.Event{:host vagrant-ubuntu-trusty-64, :service load, :state warning, :description 1-minute load average/core is 6.66, :metric 6.66, :tags nil, :time 1439746170, :ttl 10.0}

INFO [2015-08-16 17:29:35,881] defaultEventExecutorGroup-2-1 - riemann.config - this appears: #riemann.codec.Event{:host vagrant-ubuntu-trusty-64, :service load, :state critical, :description 1-minute load average/core is 12.53, :metric 12.53, :tags nil, :time 1439746175, :ttl 10.0}

I have trawled so many posts but cannot piece this simple use case together.

rgds

Dan

[ap...@aphyr.com]

I think you want to drop the (fn [event...]) wrapper--the info line is already a fn of an event, which is why it's not being called.

--
You received this message because you are subscribed to the Google Groups "Riemann Users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to riemann-user...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Dan Hunt]

thanks that now displays. 

rgds

Dan 

[Dan Hunt]

I have now modified the config to concentrate on the state change and I am confused again.

With the below script the url is called only at the start of Riemann but the info message in the srims function does not appear. I suspect Riemann or the clj-http library calls the url to check it can resolve the host.

After the start nothing is displayed even though I can see the state change:

UDP server 127.0.0.1 5555 16384 online

INFO [2015-08-16 20:06:30,189] main - riemann.core - Hyperspace core online

INFO [2015-08-16 20:08:01,381] defaultEventExecutorGroup-2-1 - riemann.config - service load breached 5 #riemann.codec.Event{:host vagrant-ubuntu-trusty-64, :service load, :state warning, :description 1-minute load average/core is 6.41, :metric 6.41, :tags nil, :time 1439755681, :ttl 10.0}

INFO [2015-08-16 20:08:06,366] defaultEventExecutorGroup-2-1 - riemann.config - service load breached 5 #riemann.codec.Event{:host vagrant-ubuntu-trusty-64, :service load, :state critical, :description 1-minute load average/core is 12.3, :metric 12.3, :tags nil, :time 1439755686, :ttl 10.0}

I am afraid of this post turning into a tutorial on the basics of Riemann but I welcome help on this simple scenario.

(require '[clj-http.client :as client])

(let [host "127.0.0.1"]

  (tcp-server {:host host})

  (udp-server {:host host})

  (ws-server  {:host host}))

(def graph (graphite {:host "localhost"}))

(defn srims [msg]

  #(info "this does not appear however the url is called on startup" %)

  (client/post "http://10.109.163.119:8080/cloud-srims/event" {:query-params {"q" "hellostate", "cb" "fred"}})

)

(let [index (index)]

  ; Inbound events will be passed to these streams:

  (streams

    ; graph all

    graph

    (where (and (service "load") (> metric 5))

      #(info "service load breached 5" %)

    )

    (by [:host :service]

      (where (service "load")

        (changed :state

          (srims "testing")

        )

      )

    )

  )

)

thanks

Dan

[ap...@aphyr.com]

You're invoking srims at config time, where it makes the HTTP call. You might want to read the howto section on the riemann streaming model--talks a bit about how streams as functions works.

--

[Dan Hunt]

the new script works. thanks

(require '[clj-http.client :as client])

(let [host "127.0.0.1"]

  (tcp-server {:host host})

  (udp-server {:host host})

  (ws-server  {:host host}))

(def graph (graphite {:host "localhost"}))

(let [index (index)]

  ; Inbound events will be passed to these streams:

  (streams

    ; graph all

    graph

    (where (and (service "load") (> metric 0))

      #(info "service load is" %)

    )

    (by [:host :service]

      (where (service "load")

        (changed :state {:init "ok"}

          (fn [event]

            (client/get "http://10.109.129.36:8080/cloud-srims/event" {:query-params {"q" "hellostate", "cb" :state}})

            (info "new state changed ****" event)

          )

        )

      )

    )

  )

)

I will now see how to pass the right parameters to the url such as the state (previous attempts have just sent :state and not the actual value .... so if someone wants to jump in a save me here then great else i'll post when i find it.

rgds

Dan

[DNA]

the search continues ...
so I now need the simplest example of Riemann reading from Kafka.
i have the plugin for kafka-riemann.
no tests in the plugin so I have tried sending json using the kafka-console-producer.sh with json decoder but no success.

started by sending "hello word"

echo "Hello, World" | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic events > /dev/null

got error:
ERROR [2016-02-15 01:29:11,778] clojure-agent-send-off-pool-0 - riemann.plugin.kafka - could not decode msg
com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'Hello': was expecting ('true', 'false' or 'null')
 at [Source: java.io.StringReader@4983a458; line: 1, column: 6]
        at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1419)
 

riemann config is:
(defn my-json-decoder
  "Decode kafka message into a riemann event"
  [input]
(load-plugins)
  (let [decoded-msg (cheshire.core/parse-string (String. input) true)]
    (map riemann.common/event decoded-msg))

(kafka/kafka-consumer {:topic "events"
                       :zookeeper.connect "localhost:2181"
                       :group.id "riemann.consumer"
                       :auto.offset.reset "smallest"
                       :auto.commit.enable "false"
                       :decoder my-json-decoder})

So what is the json I should send to be valid ?

thanks
Dan

[Brian Lalor]

It looks like it should be a json representation of a riemann event.

--
You received this message because you are subscribed to the Google Groups "Riemann Users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to riemann-user...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

— 

Brian Lalor

bla...@bravo5.org

[DNA]

@Brian
cool so I tried the following json in my test:
vagrant@vagrant-ubuntu-trusty-64:~/kafka_2.11-0.9.0.0$ echo '{"host":"dan","service":"test","state":"ok","metric":"0.13"}' | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic events > /dev/null

and Riemann raised "Key must be integer" exception:
ERROR [2016-02-15 13:35:41,922] clojure-agent-send-off-pool-1 - riemann.plugin.kafka - interrupted consumption
java.lang.IllegalArgumentException: Key must be integer
        at clojure.lang.APersistentVector.invoke(APersistentVector.java:284)
        at riemann.common$event.invoke(common.clj:137)
        at clojure.core$map$fn__4245.invoke(core.clj:2559)
        at clojure.lang.LazySeq.sval(LazySeq.java:40)
        at clojure.lang.LazySeq.seq(LazySeq.java:49)
        at clojure.lang.RT.seq(RT.java:484)
        at clojure.core$seq.invoke(core.clj:133)
        at riemann.plugin.kafka$start_kafka_thread$fn__456.invoke(kafka.clj:53)
        at clojure.core$binding_conveyor_fn$fn__4145.invoke(core.clj:1910)
        at clojure.lang.AFn.call(AFn.java:18)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:745)
INFO [2016-02-15 13:35:41,931] clojure-agent-send-off-pool-1 - kafka.consumer.ZookeeperConsumerConnector - [riemann.consumer_vagrant-ubuntu-trusty-64-1455543323173-87613d6e], ZKConsumerConnector shutting down

-------
Here is my Riemann config:

(load-plugins)

(defn my-json-decoder
  "Decode kafka message into a riemann event"
  [input]

  (let [decoded-msg (cheshire.core/parse-string (String. input) true)]

    (map riemann.common/event decoded-msg)))

(kafka/kafka-consumer {:topic "events"
                       :zookeeper.connect "localhost:2181"
                       :group.id "riemann.consumer"
                       :auto.offset.reset "smallest"
                       :auto.commit.enable "false"
                       :decoder my-json-decoder})

(let [host "127.0.0.1"]
  (tcp-server {:host host})

)

(let [index (index)]
  ; Inbound events will be passed to these streams:
  (streams

    prn
))

[Brian Lalor]

I’m pretty sure the metric needs to be a number.

--
You received this message because you are subscribed to the Google Groups "Riemann Users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to riemann-user...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

— 

Brian Lalor

bla...@bravo5.org

[DNA]

I tried that but same exception:

vagrant@vagrant-ubuntu-trusty-64:~/kafka_2.11-0.9.0.0$ echo '{"host":"dan","service":"test","state":"ok","metric":1.0}' | ~/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic events > /dev/null
thanks

Dan 

[Aphyr]

On 02/14/2016 05:44 PM, DNA wrote:
> (map riemann.common/event decoded-msg)

Looks like it expects each message to be an array of events, perhaps?

> (defn my-json-decoder
> "Decode kafka message into a riemann event"
> [input]
> (load-plugins)

Also are you sure you want to call load-plugins every time a message arrives?

--Kyle