sending tags with def influxdb to influxDB

[Nina Zimmermann]

I get server stats via collectD send to Riemann.

In Riemann I filter in different streams and would like to add additional tags to some events before I send them to influxDB.

For some reason the attributes get recognized and forwarded as fields to influxDB but the tags don't get listed. Also the :tag-fields parameter in the def influxdb doesn't seem to work anymore. It does in the def influxdb-09, but that one doesn't connect with my influx instance.

I'd highly appreciate some help. I am new to Riemann and Clojure and quite puzzled over this. The custom tags are important for the influxDB design and performance. I can't imagine it's not possible to adjust them in Riemann.

I use riemann 0.2.11 and influxDB 0.9

Here ist the riemann.config I use to testing the tagging functionality:

(logging/init {:file "/var/log/riemann/riemann.log"})

(def influxdb-creds { :version :0.9

              :host "localhost"

              :db "riemann_071116"

              :port 8086

              :tag-fields #{:host :sys :env}}) ;doesn't transmit

                              

;; Riemann log file location

(logging/init {:file "/var/log/riemann/riemann.log"})

(let [host "0.0.0.0"]

  (tcp-server {:host host})

  (udp-server {:host host})

  (ws-server  {:host host}))

(periodically-expire 10 {:keep-keys [:host :service :tags, :state, :description, :metric]})

(let [index (index)

      influxBatchSender (batch 100 1/10

         (async-queue! :agg {:queue-size 1000

                             :core-pool-size 4

                             :max-pool-size 32

                             :keep-alive-time 60000}

                       (influxdb influxdb-creds)))]

  ; Inbound events will be passed to these streams:

  (streams

    (default :ttl 60

  

      ; add tag to all events - does not work

      (tag "foo" index)

      ; Index all events immediately.

      (where (not (tagged "notification"))

        index)

      (where (not (service #"^riemann."))

      (where (service "myservice") 

      

      influxBatchSender)))))

;print events to log

(streams

  (where (tagged "test")

  ;     prn

      #(info %)))

[Mathieu Corbin]

Hello,

I just tried your configuration and everything works for me :

- First, i think it's better to have only one entrypoint (streams) on your config file.

- If i replace '(tag "foo" index)' by "'(tag "foo" #(info "Tagged foo : " %))'i can see the tagged events in the log file. What is not working for you?

- I also did '(with :sys "sys!" influxBatchSender)' for adding a :sys tag. I can see the tags (host and sys, because the :env keyword does not exists on the event) with "SHOW TAG KEYS" using the influxdb client.
The :tag of an event and the tag in influxdb (with the :tag-fields configuration) are two separates concepts.

--
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-users+unsubscribe@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Kyle Kingsbury]

This may be slightly confusing: Riemann events are maps of keys to values. One of those keys is :tags, whose value is a sequence of strings. (with :sys "sys!" ...) adds a :sys key to an event, but leaves its tags unchanged.

I think Influx uses the terms a little differently, which might account for some confusion ;)

--Kyle

To unsubscribe from this group and stop receiving emails from it, send an email to riemann-user...@googlegroups.com.

[Nina]

Thank you very much Mathieu and aphyr, your answers were very helpful!

As far as I understood now there is a difference between riemann tags and influx tags and the function (tag "new-tag-name") to add a tag to an riemann event and the keyword: tag-fields #{} to feed tags to influx. 

So my problem is that there are changing tags I want to send to influxDB, not always the static same.

Via the statsD plugin which is incorporated in collectD I get very long service names, e.g. "rt.test,farm=farmid,farm_node_id=5,host=hostid,country=countrykey,customer=customername".

I have a function that returns the first bit of the lang string for the measurement name ("test") and one that goes through the string and returns a map like this {:farm farmid, :farm_node_id 5, :host hostid :country countrykey :customer customername}:

namespace influx.clj:

(ns testing.etc.influx

 (require [clojure.string :as str]))

(defn measurement [statsdstring]

       (first(str/split statsdstring #",")))

(defn tags [statsdstring]

   (def tagsvector (rest(str/split statsdstring #",")))

   (reduce (fn [m v] (assoc m (keyword (first v)) (second v))) {} (map (fn [s] (str/split s #"=")) tagsvector)))

So I want to use these results to send them to influxDB and save them as :table-name and :tag-fields

Here is my new riemann.config:

(require [testing.etc.influx :refer :all])

(def influxdb-creds { :version :0.9

             :host "localhost"

             :db "riemann_091116"

             :port 8086

             :tag-fields #{}

              :table-name ""})

                             

;; Riemann log file location

(logging/init {:file "/var/log/riemann/riemann.log"})

(let [host "0.0.0.0"]

 (tcp-server {:host host})

 (udp-server {:host host})

 (ws-server  {:host host}))

(periodically-expire 10 {:keep-keys [:host :service :tags, :state, :description, :metric]})

(let [index (index)

     influxBatchSender (batch 100 1/10

        (async-queue! :agg {:queue-size 1000

                            :core-pool-size 4

                            :max-pool-size 32

                            :keep-alive-time 60000}

                      (influxdb influxdb-creds)))]

  ; Inbound events will be passed to these streams:

 (streams

   (default :ttl 60

 

     ; Index events immediately.

     (where (not (tagged "notification"))

       index)

      (where (service "myservice")

       (where (service #"^statsd")

         (with :table-name (measurement (#(info (:service %)))) :tag-fields (tags (#(info (:service %)))) influxBatchSender))))))

;print events to log

(streams

 (where (service "myservice")

    #(info %)))

So with

<(with :table-name (measurement (#(info (:service %)))) :tag-fields (tags (#(info (:service %)))) influxBatchSender)>

I try to do this <(with :table-name (measurement (<value of :service as string>)) :tag-fields (tags (<value of :service as string>)) influxBatchSender)>

When I run this I get this error:

ERROR [2016-11-09 16:09:21,078] main - riemann.bin - Couldn't start

clojure.lang.Compiler$CompilerException: clojure.lang.ArityException: Wrong number of args (0) passed to: config/eval53/fn--143, compiling:(/etc/riemann/riemann.config:25:1)

at clojure.lang.Compiler.load(Compiler.java:7391)

at clojure.lang.Compiler.loadFile(Compiler.java:7317)

at clojure.lang.RT$3.invoke(RT.java:320)

at riemann.config$include.invokeStatic(config.clj:406)

at riemann.config$include.invoke(config.clj:384)

at riemann.bin$_main.invokeStatic(bin.clj:99)

at riemann.bin$_main.invoke(bin.clj:85)

at clojure.lang.AFn.applyToHelper(AFn.java:156)

at clojure.lang.AFn.applyTo(AFn.java:144)

at riemann.bin.main(Unknown Source)

Caused by: clojure.lang.ArityException: Wrong number of args (0) passed to: config/eval53/fn--143

at clojure.lang.AFn.throwArity(AFn.java:429)

at clojure.lang.AFn.invoke(AFn.java:28)

at riemann.config$eval53.invokeStatic(riemann.config:51)

at riemann.config$eval53.invoke(riemann.config:25)

at clojure.lang.Compiler.eval(Compiler.java:6927)

at clojure.lang.Compiler.load(Compiler.java:7379)

... 9 common frames omitted

I understand that no args are given to the function config/eval53/fn--143.

Now my questions:

- I persume the config/eval53/fn--143 is the (measurement (#(info (:service %))))-call. How can I check that/find out what eval53/fn--143 really is?

- How do I get to the value of the event key :service and cast it into an string?

- Or do I missunderstand the error and it's something different?

And a general questions:

- why should there be just one initial stream? For performance reasons?

Help is still highly appreciated. I'm just getting back into functional programming and into riemann and clojure. It's so much fun but so frustrating when I cannot understand the errors deeply.

Thank you, Nina

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.


[Mathieu Corbin]

Hi,

In clojure, you can get the value for a key in a map like this :

- If the key is a keyword, just do (:key mymap)

- If the key is everything else (string, number...), do (get mymap "key"), or if you want a default value (get mymap "key" "default value if key doesn't exists")

So, if you have an event and you want the service, you can do (:service event).

In this line :

(with :table-name (measurement (#(info (:service %)))) :tag-fields (tags (#(info (:service %)))) influxBatchSender)



You are doing (measurement (#(info (:service %)))). You probably want to get the :service value of your event. You don't have to call info. Info is a function for logging.  Unfortunaly, as far as i know, you cannot access the event in the "with" stream. You can do (with :table-name "foo") but you cannot do (with :table-name (:service event)).

But you can use smap, like this:

(smap
 (fn [event] (assoc event :table-name (measurement (:service event)) :tag-fields (tags (:service event))))
 influxBatchSender)

I hope this can help you ;)

To unsubscribe from this group and stop receiving emails from it, send an email to riemann-users+unsubscribe@googlegroups.com.

[Nina]

Hi Mathieu

Yes, that helped a lot. Thank you so much! I learned a lot today rewriting and testing.

So I am transmitting to influx with the desired tags. After going through the source code of the riemann influx api I learned that :tag-fields is a sequence with keys only rather than a key/value map (I tried to feed that but got a cast exeption).

My problem in the future will be widely changing tags differing on the :service string I get from the hosts. I am looking for a way to feed the keys of my origin to :tag-fields.

If I run the config bellow it's not working. I again get a cast error, internaly (in the event point function?) a cast from map to number is tried and I don't quite understand how I can feed a map into :tag-fields dynamically instead of having to declare it statically.

The call that throws the error is this one in the riemann.config, without it it works but with static tag keys: :tag-fields (keys (gettags (:service ev)))

Here is my riemann.config:

(require '[testing.influx :refer :all])

(def influxdb-creds { :version :0.9

             :host "localhost"

             :db "riemann_091116"

             :port 8086

             :tag-fields #{:farm :farm_node_id :host}})

                             

;; Riemann log file location

(logging/init {:file "/var/log/riemann/riemann.log"})

(let [host "0.0.0.0"]

 (tcp-server {:host host})

 (udp-server {:host host})

 (ws-server  {:host host}))

(periodically-expire 10 {:keep-keys [:host :service :tags, :state, :description, :metric]})

(let [index (index)

     influxBatchSender (batch 100 1/10

        (async-queue! :agg {:queue-size 1000

                            :core-pool-size 4

                            :max-pool-size 32

                            :keep-alive-time 60000}

         

                      (influxdb influxdb-creds)))]

    (streams

   (default :ttl 60

 

     (where (service #"^statsd")

        (tag "influx" #(info "Tagged with influx: " %))

        (smap

          (fn [ev] (conj (assoc ev :service (getmeasurement (:service ev)) :tag-fields (keys (gettags (:service ev))))

                         (gettags (:service ev))))

         influxBatchSender)))))

Here is the gettags function from my influx namespace:

(defn gettags [statsdstring]

    (def tagsvector (rest(str/split statsdstring #",")))

    (reduce (fn [m v] (assoc m (keyword (first v)) (second v))) {} (map (fn [s] (str/split s #"=")) tagsvector)))

And here the error I get:

WARN [2016-11-10 18:54:19,444] pool-1-thread-4 - riemann.streams - riemann.influxdb$influxdb_9$stream__9428@5c5aa320 threw

java.lang.ClassCastException: clojure.lang.APersistentMap$KeySeq cannot be cast to java.lang.Number

 at clojure.lang.Numbers.isNeg(Numbers.java:100)

 at clojure.pprint$fixed_float.invokeStatic(cl_format.clj:676)

 at clojure.pprint$fixed_float.invoke(cl_format.clj:672)

 at clojure.lang.AFn.applyToHelper(AFn.java:160)

 at clojure.lang.AFn.applyTo(AFn.java:144)

 at clojure.core$apply.invokeStatic(core.clj:646)

 at clojure.pprint$execute_format$fn__9038.invoke(cl_format.clj:1907)

 at clojure.lang.AFn.applyToHelper(AFn.java:156)

 at clojure.lang.AFn.applyTo(AFn.java:144)

 at clojure.core$apply.invokeStatic(core.clj:646)

 at clojure.pprint$map_passing_context.invokeStatic(utilities.clj:30)

 at clojure.pprint$execute_format.invokeStatic(cl_format.clj:1879)

 at clojure.pprint$execute_format.invoke(cl_format.clj:1879)

 at clojure.pprint$execute_format$fn__9036.invoke(cl_format.clj:1893)

 at clojure.pprint$execute_format.invokeStatic(cl_format.clj:1892)

 at clojure.pprint$execute_format.invoke(cl_format.clj:1879)

 at clojure.pprint$cl_format.invokeStatic(cl_format.clj:64)

 at clojure.pprint$cl_format.doInvoke(cl_format.clj:27)

 at clojure.lang.RestFn.invoke(RestFn.java:442)

 at riemann.influxdb$kv_encode_9$fn__9379.invoke(influxdb.clj:25)

 at clojure.core$map$fn__4785.invoke(core.clj:2646)

 at clojure.lang.LazySeq.sval(LazySeq.java:40)

 at clojure.lang.LazySeq.seq(LazySeq.java:49)

 at clojure.lang.Cons.next(Cons.java:39)

 at clojure.lang.RT.next(RT.java:688)

 at clojure.core$next__4341.invokeStatic(core.clj:64)

 at clojure.string$join.invokeStatic(string.clj:191)

 at clojure.string$join.invoke(string.clj:180)

 at riemann.influxdb$kv_encode_9.invokeStatic(influxdb.clj:21)

 at riemann.influxdb$kv_encode_9.invoke(influxdb.clj:20)

 at riemann.influxdb$lineprotocol_encode_9.invokeStatic(influxdb.clj:29)

 at riemann.influxdb$lineprotocol_encode_9.invoke(influxdb.clj:28)

 at clojure.core$map$fn__4785.invoke(core.clj:2644)

 at clojure.lang.LazySeq.sval(LazySeq.java:40)

 at clojure.lang.LazySeq.seq(LazySeq.java:49)

 at clojure.lang.LazySeq.first(LazySeq.java:71)

 at clojure.lang.RT.first(RT.java:667)

 at clojure.core$first__4339.invokeStatic(core.clj:55)

 at clojure.string$join.invokeStatic(string.clj:180)

 at clojure.string$join.invoke(string.clj:180)

 at riemann.influxdb$influxdb_9$stream__9428.invoke(influxdb.clj:181)

 at riemann.streams$execute_on$stream__6765$runner__6766$fn__6777.invoke(streams.clj:268)

 at riemann.streams$execute_on$stream__6765$runner__6766.invoke(streams.clj:268)

 at clojure.lang.AFn.applyToHelper(AFn.java:152)

 at clojure.lang.AFn.applyTo(AFn.java:144)

 at clojure.core$apply.invokeStatic(core.clj:646)

 at clojure.core$with_bindings_STAR_.invokeStatic(core.clj:1881)

 at clojure.core$with_bindings_STAR_.doInvoke(core.clj:1881)

 at clojure.lang.RestFn.invoke(RestFn.java:425)

 at clojure.lang.AFn.applyToHelper(AFn.java:156)

 at clojure.lang.RestFn.applyTo(RestFn.java:132)

 at clojure.core$apply.invokeStatic(core.clj:650)

 at clojure.core$bound_fn_STAR_$fn__4671.doInvoke(core.clj:1911)

 at clojure.lang.RestFn.invoke(RestFn.java:397)

 at clojure.lang.AFn.run(AFn.java:22)

 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)

Can anyone help?

[Mathieu Corbin]

:tag-fields is a set, and is an influxdb option (like :db, scheme...) and not an event option. You need to do :

(influxdb {:tag-fields #{:tag1 :tag2 :tag3}})

I am not very familiar with influxdb. If i read the documentation (https://docs.influxdata.com/influxdb/v1.0/concepts/key_concepts/), there are fields and tags. In riemann we have :tag-fields (docstring : "A set of event fields to map into InfluxDB series tags") and :tags (docstring: "A common map of tags to apply to all points") in the influxdb options.

The :metric key of the event is transformed into a field called "value". Other events keys are also transformed into fields  (docstring : "The event's `metric` is converted to the `value` field, and any additional event fields which are not standard Riemann properties or in `tag-fields` will also be present").

This is a bit confusing. Can anyone confirm this ?

For the clojure part, It is not good practice to have def inside defn or def in clojure. For example, your gettags function can be :

(defn get-tags [statd-string]
  (->> (rest (str/split statd-string #","))   ;; get the tags
       (map #(str/split % #"="))              ;; split => [key, value]
       (map #(assoc % 0 (keyword (first %)))) ;; transform keyword => [:key, value]
       (into {})))                            ;; array => map



->> is a threading macro (http://clojure.org/guides/threading_macros)



To unsubscribe from this group and stop receiving emails from it, send an email to riemann-users+unsubscribe@googlegroups.com.

[Nina]

I can confirm from my experience throughout the past 2 weeks with collectD -> riemann -> influxDB that the only "standard key" arriving in influx as an indexed tag is the riemann :host key. All other event keys are given to influxDB as fields. That is exactly my problem because it leads to a lot of fields and only one tag per measurement in influxDB which is not so good for A. performance and B. filtering because you need tags in influxDB for a where clause. Fields are what you aggregate on. That's why I am looking for a way to transform certain riemann event keys into influxDB tags and leave certain riemann event keys the way they are to just forward them as fields to influxDB. If anyone found a way that works with changing key values I'd be very happy to see that solution. I think I'm getting close to one, my influxDB server just crashed so no more testing for today but once I find one I will post it here.

Thank you for the threading macro hint. I come from scala and missed the point notation to chain things together. Now there it is. Awesome.

[Mathieu Corbin]

I think the only way to have more key actually is to do :

(influxdb {:tag-fields #{:tag1 :tag2 :tag3}})

and with events like {:host "host" :service "service" :tag1 "value1" :tag2 "value2" :tag3 "value3"} you will have the :tag1, :tag2, :tag3 added to influxdb.

So you need to know beforehand the keys.

You can also have multiples "influxdb" functions, each with a different :tag-fields, and send events to them according to their keys.

If you think it can be useful to have :tag-fields on events, you can open an issue on github and i will take a look at this in a near future.

To unsubscribe from this group and stop receiving emails from it, send an email to riemann-users+unsubscribe@googlegroups.com.

[K. Arun]

Hello Mathieu,

Per-event :tag-fields would help greatly. I'm running into the exact same issue as Nina - I have a heterogeneous set of events that need to be sent to InfluxDB, and each type of event requires a different set of (InfluxDB) tags to be set. As Nina mentioned, InfluxDB, somewhat counter-intuitively, indexes (per point) tags and not fields. Therefore, if your InfluxDB queries depend on slicing and dicing fields (rather than tags), performance suffers.

I tried dynamically adjusting my version of influxBatchSender to set a different set of :tag-fields per event, but haven't had much success composing the streams. As is perhaps indicated in the async-queue! documentation, I see compilation errors like:

clojure.lang.Compiler$CompilerException: java.lang.IllegalArgumentException: won't conj service: #object[

riemann.service.ExecutorServiceService 0xc4d2c44 "riemann.service.ExecutorServiceService@c4d2c44"] wouldc

onflict with (#object[riemann.service.ExecutorServiceService 0x13866329 "riemann.service.ExecutorServiceS

ervice@13866329"]), compiling:(fnc-riemann.config:34:10)

Arun

[K. Arun]

Filed https://github.com/riemann/riemann/issues/742. 

Thanks,

Arun

[Mathieu Corbin]

Hi,

I will look at this next week. I think it can be easily done.

Mathieu

[Aphyr]

On 11/19/2016 09:59 AM, K. Arun wrote:
> Hello Mathieu,
>
> Per-event :tag-fields would help greatly.

It might be easier to offer a lower-level influxdb stream which, instead of
taking Riemann events, accepts {:tags {...}, :keys {...}}, or some structure
better-suited to InfluxDB's data model. No sense in making folks jump through
double hoops, having to build up a custom :tag-fields list and then having an
interpreter figure out how to construct the underlying data structure.

--Kyle

[Nina]

InfluxDB data points consist of the measurement (sort of the table name), a timestamp and at least one field (the metric's value). Tags are not required but useful for aggregations. Basically tags are needed for WHERE and GROUP BY queries; fields for all kinds of aggregations. Because tags are indexed the performance of queries can be optimized by using them, but it should be kept in mind that with the amount of tags the index and storage grows - the more unchained tags, the higher the cardinality. So just sending everything as tags instead of fields to InfluxDB by default would be no good in my opinion.

A working example maybe helps:

I currently get host events send to riemann via collectD and custom metric events (e.g. counter, gauge, timer) via statsD. The host data aggregation works all fine. The metric data I want to extract from statsD depends on the metric type and sender (host). statsD usually sends a string that consists of <pluginname>/<datatype>-<name>,<host=hostname>,<custom1=string>,<custom2=int></value>. That string gets send to riemann as the :service value and :type_instance value (minus the plugin name). The :service value gets send to InfluxDB as measurement name.

At the moment I iterate through the statsD string in riemann, filter out the information I want to store as tag in InfluxDB and hand it over to the :tag-fields key in riemann's InflluxDBSender. To shorten the measurement name I filter the <name> part and give that to the event as the new :service value. That doesn't feel very efficient, for now it's the only working way I found though. As mentioned in the issue, it only works with pre-defined InfluxDB tag keys in the sender function. A way to decide the key values for influxDB tags in a dynamic manner in riemann would be great. Unfortunately I do not know enough about streams and clojure yet to be of any help how to implement that. 

I saw there is an active issue for supporting the current InfluxDB version 1.x:  https://github.com/riemann/riemann/issues/739

Maybe it makes sense to bring these two issues together? Although I do think that this "tagging feature" discussed here would be useful for the 0.9 users as well in the future.

Thank you for getting into this.

Nina

[Mathieu Corbin]

I have added the ability to add :tag-fields by event in my PR (https://github.com/riemann/riemann/pull/741).

However, Aphyr is right. The actual influxdb code is a bit confusing (need to generate a :tag-fields set, all event keys not in :tag-fields are converted to influx fields  so need to dissoc key from the event map if you don't want it). I tried in my PR to keep backward compatibility with the actual code (same options).

I think we also need the ability to specify by event the consistency level, the retention policy, and the database name (https://github.com/riemann/riemann/issues/730).

I don't know what is the best solution. Keep the actual code (for compatibility) and create a new influx stream, or modify the actual influx stream and broke backward compatibility ? Or keep backward compatibility (like in my PR) ? Or use the :version stream option (https://github.com/riemann/riemann/blob/master/src/riemann/influxdb.clj#L198), and create the new stream only if a specific version is specified, and deprecate the old influx stream later.

[Mathieu Corbin]

I have worked on this issue and i think the best solution is to :

- Drop the influxdb 0.8 support

- Use the current behavior by default when the influxdb stream in called (for compatibility), with news features (consistency level, per-event :tag-fields and :precision keys).

- If the :version influxdb option is set to something like ":new-stream", use a new influxdb stream.

The new stream will receive events like :

{:time 1428366765
 :tags {:foo "bar"
           :bar "baz"}
 :precision :seconds
 :db "riemann_test"
 :consistency "ALL"
 :retention "autogen"
 :measurement "measurement"
 :fields {:alice "bob"}}

=> We can set per event the consistency level, the db, the retention, and the time precision. Tags and fields are just map so easy to modify (no more :tag-fields).

What do you think ?

[Nina]

I think that sounds good! And it would be great to incorporate those features in the influxdb 1.0 and higher support as well.

Thank you for working on this. When do you think will this be implemented?

To unsubscribe from this group and stop receiving emails from it, send an email to riemann-user...@googlegroups.com.

[Mathieu Corbin]

I have updated my PR : https://github.com/riemann/riemann/pull/741

There is a lot of change so i need tests/reviews ;)

To unsubscribe from this group and stop receiving emails from it, send an email to riemann-users+unsubscribe@googlegroups.com.

[Nina]

I have been terribly buzzy with another project. I will look into this over the holidays. Thank you very much!