Create a template system for the graphite serializer

closes #925
closes #879

docs/DATA_FORMATS_OUTPUT.md

@@ -1,5 +1,11 @@
 # Telegraf Output Data Formats
 
+Telegraf is able to serialize metrics into the following output data formats:
+
+1. [InfluxDB Line Protocol](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md#influx)
+1. [JSON](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md#json)
+1. [Graphite](https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md#graphite)
+
 Telegraf metrics, like InfluxDB
 [points](https://docs.influxdata.com/influxdb/v0.10/write_protocols/line/),
 are a combination of four basic parts:
@@ -29,7 +35,7 @@ config option, for example, in the `file` output plugin:
   ## Files to write to, "stdout" is a specially handled file.
   files = ["stdout"]
 
-  ## Data format to output. 
+  ## Data format to output.
 
   ## Each data format has it's own unique set of configuration options, read
   ## more about them here:
@@ -54,7 +60,7 @@ metrics are serialized directly into InfluxDB line-protocol.
   ## Files to write to, "stdout" is a specially handled file.
   files = ["stdout", "/tmp/metrics.out"]
 
-  ## Data format to output. 
+  ## Data format to output.
 
   ## Each data format has it's own unique set of configuration options, read
   ## more about them here:
@@ -89,7 +95,7 @@ tars.cpu-total.us-east-1.cpu.usage_idle 98.09 1455320690
   ## Files to write to, "stdout" is a specially handled file.
   files = ["stdout", "/tmp/metrics.out"]
 
-  ## Data format to output. 
+  ## Data format to output.
 
   ## Each data format has it's own unique set of configuration options, read
   ## more about them here:
@@ -126,7 +132,7 @@ The Json data format serialized Telegraf metrics in json format. The format is:
   ## Files to write to, "stdout" is a specially handled file.
   files = ["stdout", "/tmp/metrics.out"]
 
-  ## Data format to output. 
+  ## Data format to output.
 
   ## Each data format has it's own unique set of configuration options, read
   ## more about them here:

etc/telegraf.conf

@@ -178,6 +178,9 @@
 #   servers = ["localhost:2003"]
 #   ## Prefix metrics name
 #   prefix = ""
+#   ## Graphite output template
+#   ## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+#   template = "host.tags.measurement.field"
 #   ## timeout in seconds for the write connection to graphite
 #   timeout = 2
 

internal/config/config.go

@@ -850,8 +850,17 @@ func buildSerializer(name string, tbl *ast.Table) (serializers.Serializer, error
 		}
 	}
 
+	if node, ok := tbl.Fields["template"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.Template = str.Value
+			}
+		}
+	}
+
 	delete(tbl.Fields, "data_format")
 	delete(tbl.Fields, "prefix")
+	delete(tbl.Fields, "template")
 	return serializers.NewSerializer(c)
 }
 

plugins/outputs/graphite/README.md

@@ -1,13 +1,34 @@
 # Graphite Output Plugin
 
-This plugin writes to [Graphite](http://graphite.readthedocs.org/en/latest/index.html) via raw TCP.
+This plugin writes to [Graphite](http://graphite.readthedocs.org/en/latest/index.html)
+via raw TCP.
+
+## Configuration:
+
+```toml
+# Configuration for Graphite server to send metrics to
+[[outputs.graphite]]
+  ## TCP endpoint for your graphite instance.
+  servers = ["localhost:2003"]
+  ## Prefix metrics name
+  prefix = ""
+  ## Graphite output template
+  ## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+  template = "host.tags.measurement.field"
+  ## timeout in seconds for the write connection to graphite
+  timeout = 2
+```
 
 Parameters:
 
-    Servers []string
-    Prefix  string
-    Timeout int
+    Servers  []string
+    Prefix   string
+    Timeout  int
+    Template string
 
 * `servers`: List of strings, ["mygraphiteserver:2003"].
 * `prefix`: String use to prefix all sent metrics.
-* `timeout`: Connection timeout in second.
+* `timeout`: Connection timeout in seconds.
+* `template`: Template for graphite output format, see
+https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+for more details.

plugins/outputs/graphite/graphite.go

@@ -16,17 +16,21 @@ import (
 
 type Graphite struct {
 	// URL is only for backwards compatability
-	Servers []string
-	Prefix  string
-	Timeout int
-	conns   []net.Conn
+	Servers  []string
+	Prefix   string
+	Template string
+	Timeout  int
+	conns    []net.Conn
 }
 
 var sampleConfig = `
   ## TCP endpoint for your graphite instance.
   servers = ["localhost:2003"]
   ## Prefix metrics name
   prefix = ""
+  ## Graphite output template
+  ## see https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+  template = "host.tags.measurement.field"
   ## timeout in seconds for the write connection to graphite
   timeout = 2
 `
@@ -72,7 +76,7 @@ func (g *Graphite) Description() string {
 func (g *Graphite) Write(metrics []telegraf.Metric) error {
 	// Prepare data
 	var bp []string
-	s, err := serializers.NewGraphiteSerializer(g.Prefix)
+	s, err := serializers.NewGraphiteSerializer(g.Prefix, g.Template)
 	if err != nil {
 		return err
 	}

plugins/outputs/graphite/graphite_test.go

@@ -54,7 +54,7 @@ func TestGraphiteOK(t *testing.T) {
 	m1, _ := telegraf.NewMetric(
 		"mymeasurement",
 		map[string]string{"host": "192.168.0.1"},
-		map[string]interface{}{"mymeasurement": float64(3.14)},
+		map[string]interface{}{"myfield": float64(3.14)},
 		time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
 	)
 	m2, _ := telegraf.NewMetric(
@@ -90,7 +90,7 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup) {
 	reader := bufio.NewReader(conn)
 	tp := textproto.NewReader(reader)
 	data1, _ := tp.ReadLine()
-	assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement 3.14 1289430000", data1)
+	assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement.myfield 3.14 1289430000", data1)
 	data2, _ := tp.ReadLine()
 	assert.Equal(t, "my.prefix.192_168_0_1.mymeasurement.value 3.14 1289430000", data2)
 	data3, _ := tp.ReadLine()

plugins/outputs/librato/librato.go

@@ -7,6 +7,7 @@ import (
 	"io/ioutil"
 	"log"
 	"net/http"
+	"strings"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
@@ -152,17 +153,13 @@ func (l *Librato) Description() string {
 	return "Configuration for Librato API to send metrics to."
 }
 
-func (l *Librato) buildGaugeName(m telegraf.Metric, fieldName string) string {
-	// Use the GraphiteSerializer
-	graphiteSerializer := graphite.GraphiteSerializer{}
-	return graphiteSerializer.SerializeBucketName(m, fieldName)
-}
-
 func (l *Librato) buildGauges(m telegraf.Metric) ([]*Gauge, error) {
 	gauges := []*Gauge{}
+	graphiteSerializer := graphite.GraphiteSerializer{}
+	bucket := graphiteSerializer.SerializeBucketName(m.Name(), m.Tags())
 	for fieldName, value := range m.Fields() {
 		gauge := &Gauge{
-			Name:        l.buildGaugeName(m, fieldName),
+			Name:        strings.Replace(bucket, "FIELDNAME", fieldName, 1),
 			MeasureTime: m.Time().Unix(),
 		}
 		if !gauge.verifyValue(value) {

plugins/serializers/graphite/graphite.go

@@ -8,77 +8,101 @@ import (
 	"github.com/influxdata/telegraf"
 )
 
+const DEFAULT_TEMPLATE = "host.tags.measurement.field"
+
 type GraphiteSerializer struct {
-	Prefix string
+	Prefix   string
+	Template string
 }
 
-var sanitizedChars = strings.NewReplacer("/", "-", "@", "-", " ", "_")
+var sanitizedChars = strings.NewReplacer("/", "-", "@", "-", " ", "_", "..", ".")
 
 func (s *GraphiteSerializer) Serialize(metric telegraf.Metric) ([]string, error) {
 	out := []string{}
 
 	// Convert UnixNano to Unix timestamps
 	timestamp := metric.UnixNano() / 1000000000
 
+	bucket := s.SerializeBucketName(metric.Name(), metric.Tags())
+
 	for field_name, value := range metric.Fields() {
 		// Convert value
 		value_str := fmt.Sprintf("%#v", value)
 		// Write graphite metric
 		var graphitePoint string
 		graphitePoint = fmt.Sprintf("%s %s %d",
-			s.SerializeBucketName(metric, field_name),
+			// insert "field" section of template
+			strings.Replace(bucket, "FIELDNAME", field_name, 1),
 			value_str,
 			timestamp)
 		out = append(out, graphitePoint)
 	}
 	return out, nil
 }
 
-func (s *GraphiteSerializer) SerializeBucketName(metric telegraf.Metric, field_name string) string {
-	// Get the metric name
-	name := metric.Name()
+// SerializeBucketName will take the given measurement name and tags and
+// produce a graphite bucket. It will use the GraphiteSerializer.Template
+// to generate this, or DEFAULT_TEMPLATE.
+//
+// NOTE: SerializeBucketName replaces the "field" portion of the template with
+// FIELDNAME. It is up to the user to replace this. This is so that
+// SerializeBucketName can be called just once per measurement, rather than
+// once per field.
+func (s *GraphiteSerializer) SerializeBucketName(
+	measurement string,
+	tags map[string]string,
+) string {
+	if s.Template == "" {
+		s.Template = DEFAULT_TEMPLATE
+	}
+	tagsCopy := make(map[string]string)
+	for k, v := range tags {
+		tagsCopy[k] = v
+	}
 
-	// Convert UnixNano to Unix timestamps
-	tag_str := buildTags(metric)
+	var out []string
+	templateParts := strings.Split(s.Template, ".")
+	for _, templatePart := range templateParts {
+		switch templatePart {
+		case "measurement":
+			out = append(out, measurement)
+		case "tags":
+			// we will replace this later
+			out = append(out, "TAGS")
+		case "field":
+			// user of SerializeBucketName needs to replace this
+			out = append(out, "FIELDNAME")
+		default:
+			// This is a tag being applied
+			if tagvalue, ok := tagsCopy[templatePart]; ok {
+				out = append(out, strings.Replace(tagvalue, ".", "_", -1))
+				delete(tagsCopy, templatePart)
+			}
+		}
+	}
 
-	// Write graphite metric
-	var serializedBucketName string
-	if name == field_name {
-		serializedBucketName = fmt.Sprintf("%s.%s",
-			tag_str,
-			strings.Replace(name, ".", "_", -1))
-	} else {
-		serializedBucketName = fmt.Sprintf("%s.%s.%s",
-			tag_str,
-			strings.Replace(name, ".", "_", -1),
-			strings.Replace(field_name, ".", "_", -1))
+	// insert remaining tags into output name
+	for i, templatePart := range out {
+		if templatePart == "TAGS" {
+			out[i] = buildTags(tagsCopy)
+			break
+		}
 	}
-	if s.Prefix != "" {
-		serializedBucketName = fmt.Sprintf("%s.%s", s.Prefix, serializedBucketName)
+
+	if s.Prefix == "" {
+		return sanitizedChars.Replace(strings.Join(out, "."))
 	}
-	return serializedBucketName
+	return sanitizedChars.Replace(s.Prefix + "." + strings.Join(out, "."))
 }
 
-func buildTags(metric telegraf.Metric) string {
+func buildTags(tags map[string]string) string {
 	var keys []string
-	tags := metric.Tags()
 	for k := range tags {
-		if k == "host" {
-			continue
-		}
 		keys = append(keys, k)
 	}
 	sort.Strings(keys)
 
 	var tag_str string
-	if host, ok := tags["host"]; ok {
-		if len(keys) > 0 {
-			tag_str = strings.Replace(host, ".", "_", -1) + "."
-		} else {
-			tag_str = strings.Replace(host, ".", "_", -1)
-		}
-	}
-
 	for i, k := range keys {
 		tag_value := strings.Replace(tags[k], ".", "_", -1)
 		if i == 0 {
@@ -87,5 +111,5 @@ func buildTags(metric telegraf.Metric) string {
 			tag_str += "." + tag_value
 		}
 	}
-	return sanitizedChars.Replace(tag_str)
+	return tag_str
 }