Send metric tags as riemann tags and attributes

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 ### Features
 
+- [#1845](https://github.com/influxdata/telegraf/pull/1845): Send measurement as Riemann tag, add overwrite tags.
 - [#1782](https://github.com/influxdata/telegraf/pull/1782): Allow numeric and non-string values for tag_keys.
 - [#1694](https://github.com/influxdata/telegraf/pull/1694): Adding Gauge and Counter metric types.
 - [#1606](https://github.com/influxdata/telegraf/pull/1606): Remove carraige returns from exec plugin output on Windows
@@ -29,6 +30,7 @@
 
 ### Bugfixes
 
+- [#1501](https://github.com/influxdata/telegraf/issues/1501): Send metric tags as Riemann tags and attributes.
 - [#1746](https://github.com/influxdata/telegraf/issues/1746): Fix handling of non-string values for JSON keys listed in tag_keys.
 - [#1628](https://github.com/influxdata/telegraf/issues/1628): Fix mongodb input panic on version 2.2.
 - [#1733](https://github.com/influxdata/telegraf/issues/1733): Fix statsd scientific notation parsing

Godeps

@@ -1,7 +1,7 @@
 github.com/Shopify/sarama 8aadb476e66ca998f2f6bb3c993e9a2daa3666b9
 github.com/Sirupsen/logrus 219c8cb75c258c552e999735be6df753ffc7afdc
 github.com/aerospike/aerospike-client-go 7f3a312c3b2a60ac083ec6da296091c52c795c63
-github.com/amir/raidman 53c1b967405155bfc8758557863bf2e14f814687
+github.com/amir/raidman c74861fe6a7bb8ede0a010ce4485bdbb4fc4c985
 github.com/aws/aws-sdk-go 13a12060f716145019378a10e2806c174356b857
 github.com/beorn7/perks 3ac7bf7a47d159a033b107610db8a1b6575507a4
 github.com/cenkalti/backoff 4dc77674aceaabba2c7e3da25d4c823edfb73f99

etc/telegraf.conf

@@ -436,9 +436,12 @@
 #   url = "localhost:5555"
 #   ## transport protocol to use either tcp or udp
 #   transport = "tcp"
-#   ## separator to use between input name and field name in Riemann service name
+#   ## separator to use between measurement name and field name in Riemann service name
 #   separator = " "
-
+#   ## set measurement name as a Riemann tag instead of prepending it to the Riemann service name
+#   measurement_as_tag = false
+#   ## list of tag keys to specify, whose values get sent as Riemann tags. If empty, all Telegraf tag values will be sent to Riemann as tags.
+#   # riemann_tag_keys = ["telegraf","custom_tag"]
 
 
 ###############################################################################

plugins/outputs/riemann/riemann.go

@@ -12,9 +12,11 @@ import (
 )
 
 type Riemann struct {
-	URL       string
-	Transport string
-	Separator string
+	URL              string
+	Transport        string
+	Separator        string
+	MeasurementAsTag bool
+	RiemannTagKeys   []string
 
 	client *raidman.Client
 }
@@ -24,8 +26,12 @@ var sampleConfig = `
   url = "localhost:5555"
   ## transport protocol to use either tcp or udp
   transport = "tcp"
-  ## separator to use between input name and field name in Riemann service name
+  ## separator to use between measurement name and field name in Riemann service name
   separator = " "
+  ## set measurement name as a Riemann tag instead of prepending it to the Riemann service name
+  measurement_as_tag = false
+  ## list of tag keys to specify, whose values get sent as Riemann tags. If empty, all Telegraf tag values will be sent to Riemann as tags.
+  # riemann_tag_keys = ["telegraf","custom_tag"]
 `
 
 func (r *Riemann) Connect() error {
@@ -71,7 +77,7 @@ func (r *Riemann) Write(metrics []telegraf.Metric) error {
 
 	var events []*raidman.Event
 	for _, p := range metrics {
-		evs := buildEvents(p, r.Separator)
+		evs := r.buildEvents(p)
 		for _, ev := range evs {
 			events = append(events, ev)
 		}
@@ -87,7 +93,7 @@ func (r *Riemann) Write(metrics []telegraf.Metric) error {
 	return nil
 }
 
-func buildEvents(p telegraf.Metric, s string) []*raidman.Event {
+func (r *Riemann) buildEvents(p telegraf.Metric) []*raidman.Event {
 	events := []*raidman.Event{}
 	for fieldName, value := range p.Fields() {
 		host, ok := p.Tags()["host"]
@@ -101,13 +107,17 @@ func buildEvents(p telegraf.Metric, s string) []*raidman.Event {
 		}
 
 		event := &raidman.Event{
-			Host:    host,
-			Service: serviceName(s, p.Name(), p.Tags(), fieldName),
+			Host:       host,
+			Service:    r.service(p.Name(), fieldName),
+			Tags:       r.tags(p.Name(), p.Tags()),
+			Attributes: r.attributes(p.Name(), p.Tags()),
+			Time:       p.Time().Unix(),
 		}
 
 		switch value.(type) {
 		case string:
-			event.State = value.(string)
+			state := []byte(value.(string))
+			event.State = string(state[:254]) // Riemann states must be less than 255 bytes, e.g. "ok", "warning", "critical"
 		default:
 			event.Metric = value
 		}
@@ -118,30 +128,53 @@ func buildEvents(p telegraf.Metric, s string) []*raidman.Event {
 	return events
 }
 
-func serviceName(s string, n string, t map[string]string, f string) string {
-	serviceStrings := []string{}
-	serviceStrings = append(serviceStrings, n)
+func (r *Riemann) attributes(name string, tags map[string]string) map[string]string {
+	if r.MeasurementAsTag {
+		tags["measurement"] = name
+	}
+	return tags
+}
 
-	// we'll skip the 'host' tag
-	tagStrings := []string{}
-	tagNames := []string{}
+func (r *Riemann) tags(name string, tags map[string]string) []string {
+	var tagNames, tagValues []string
+
+	if r.MeasurementAsTag {
+		tagValues = append(tagValues, name)
+	}
 
-	for tagName := range t {
+	if len(r.RiemannTagKeys) > 0 {
+		for _, tagName := range r.RiemannTagKeys {
+			tagValue, ok := tags[tagName]
+			if ok {
+				tagValues = append(tagValues, tagValue)
+			}
+		}
+		return tagValues
+	}
+
+	for tagName := range tags {
 		tagNames = append(tagNames, tagName)
 	}
 	sort.Strings(tagNames)
 
 	for _, tagName := range tagNames {
-		if tagName != "host" {
-			tagStrings = append(tagStrings, t[tagName])
+		if tagName != "host" { // we'll skip the 'host' tag
+			tagValues = append(tagValues, tags[tagName])
 		}
 	}
-	var tagString string = strings.Join(tagStrings, s)
-	if tagString != "" {
-		serviceStrings = append(serviceStrings, tagString)
+
+	return tagValues
+}
+
+func (r *Riemann) service(name string, field string) string {
+	var serviceStrings []string
+
+	if !r.MeasurementAsTag {
+		serviceStrings = append(serviceStrings, name)
 	}
-	serviceStrings = append(serviceStrings, f)
-	return strings.Join(serviceStrings, s)
+	serviceStrings = append(serviceStrings, field)
+
+	return strings.Join(serviceStrings, r.Separator)
 }
 
 func init() {