Add Serializer plugins, and 'file' output plugin

Author: sparrc

Godeps

@@ -19,8 +19,7 @@ github.com/gorilla/context 1c83b3eabd45b6d76072b66b746c20815fb2872d
 github.com/gorilla/mux 26a6070f849969ba72b72256e9f14cf519751690
 github.com/hailocab/go-hostpool e80d13ce29ede4452c43dea11e79b9bc8a15b478
 github.com/influxdata/config bae7cb98197d842374d3b8403905924094930f24
-github.com/influxdata/influxdb a9552fdd91361819a792f337e5d9998859732a67
-github.com/influxdb/influxdb a9552fdd91361819a792f337e5d9998859732a67
+github.com/influxdata/influxdb ef571fc104dc24b77cd3710c156cd95e5cfd7aa5
 github.com/jmespath/go-jmespath c01cf91b011868172fdcd9f41838e80c9d716264
 github.com/klauspost/crc32 999f3125931f6557b991b2f8472172bdfa578d38
 github.com/lib/pq 8ad2b298cadd691a77015666a5372eae5dbfac8f

internal/config/config.go

@@ -16,6 +16,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/inputs"
 	"github.com/influxdata/telegraf/plugins/outputs"
 	"github.com/influxdata/telegraf/plugins/parsers"
+	"github.com/influxdata/telegraf/plugins/serializers"
 
 	"github.com/influxdata/config"
 	"github.com/naoina/toml/ast"
@@ -398,6 +399,17 @@ func (c *Config) addOutput(name string, table *ast.Table) error {
 	}
 	output := creator()
 
+	// If the output has a SetSerializer function, then this means it can write
+	// arbitrary types of output, so build the serializer and set it.
+	switch t := output.(type) {
+	case serializers.SerializerOutput:
+		serializer, err := buildSerializer(name, table)
+		if err != nil {
+			return err
+		}
+		t.SetSerializer(serializer)
+	}
+
 	outputConfig, err := buildOutput(name, table)
 	if err != nil {
 		return err
@@ -660,6 +672,37 @@ func buildParser(name string, tbl *ast.Table) (parsers.Parser, error) {
 	return parsers.NewParser(c)
 }
 
+// buildSerializer grabs the necessary entries from the ast.Table for creating
+// a serializers.Serializer object, and creates it, which can then be added onto
+// an Output object.
+func buildSerializer(name string, tbl *ast.Table) (serializers.Serializer, error) {
+	c := &serializers.Config{}
+
+	if node, ok := tbl.Fields["data_format"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.DataFormat = str.Value
+			}
+		}
+	}
+
+	if c.DataFormat == "" {
+		c.DataFormat = "influx"
+	}
+
+	if node, ok := tbl.Fields["prefix"]; ok {
+		if kv, ok := node.(*ast.KeyValue); ok {
+			if str, ok := kv.Value.(*ast.String); ok {
+				c.Prefix = str.Value
+			}
+		}
+	}
+
+	delete(tbl.Fields, "data_format")
+	delete(tbl.Fields, "prefix")
+	return serializers.NewSerializer(c)
+}
+
 // buildOutput parses output specific items from the ast.Table, builds the filter and returns an
 // internal_models.OutputConfig to be inserted into internal_models.RunningInput
 // Note: error exists in the return for future calls that might require error

plugins/outputs/all/all.go

@@ -5,6 +5,7 @@ import (
 	_ "github.com/influxdata/telegraf/plugins/outputs/amqp"
 	_ "github.com/influxdata/telegraf/plugins/outputs/cloudwatch"
 	_ "github.com/influxdata/telegraf/plugins/outputs/datadog"
+	_ "github.com/influxdata/telegraf/plugins/outputs/file"
 	_ "github.com/influxdata/telegraf/plugins/outputs/graphite"
 	_ "github.com/influxdata/telegraf/plugins/outputs/influxdb"
 	_ "github.com/influxdata/telegraf/plugins/outputs/kafka"

plugins/outputs/amqp/amqp.go

@@ -10,6 +10,8 @@ import (
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/internal"
 	"github.com/influxdata/telegraf/plugins/outputs"
+	"github.com/influxdata/telegraf/plugins/serializers"
+
 	"github.com/streadway/amqp"
 )
 
@@ -39,6 +41,8 @@ type AMQP struct {
 	channel *amqp.Channel
 	sync.Mutex
 	headers amqp.Table
+
+	serializer serializers.Serializer
 }
 
 const (
@@ -69,8 +73,18 @@ var sampleConfig = `
   # ssl_key = "/etc/telegraf/key.pem"
   ### Use SSL but skip chain & host verification
   # insecure_skip_verify = false
+
+  ### Data format to output. This can be "influx" or "graphite"
+  ### Each data format has it's own unique set of configuration options, read
+  ### more about them here:
+  ### https://github.com/influxdata/telegraf/blob/master/DATA_FORMATS_OUTPUT.md
+  data_format = "influx"
 `
 
+func (a *AMQP) SetSerializer(serializer serializers.Serializer) {
+	a.serializer = serializer
+}
+
 func (q *AMQP) Connect() error {
 	q.Lock()
 	defer q.Unlock()
@@ -147,18 +161,24 @@ func (q *AMQP) Write(metrics []telegraf.Metric) error {
 	}
 	var outbuf = make(map[string][][]byte)
 
-	for _, p := range metrics {
-		var value, key string
-		value = p.String()
-
+	for _, metric := range metrics {
+		var key string
 		if q.RoutingTag != "" {
-			if h, ok := p.Tags()[q.RoutingTag]; ok {
+			if h, ok := metric.Tags()[q.RoutingTag]; ok {
 				key = h
 			}
 		}
-		outbuf[key] = append(outbuf[key], []byte(value))
 
+		values, err := q.serializer.Serialize(metric)
+		if err != nil {
+			return err
+		}
+
+		for _, value := range values {
+			outbuf[key] = append(outbuf[key], []byte(value))
+		}
 	}
+
 	for key, buf := range outbuf {
 		err := q.channel.Publish(
 			q.Exchange, // exchange

plugins/outputs/amqp/amqp_test.go

@@ -3,6 +3,7 @@ package amqp
 import (
 	"testing"
 
+	"github.com/influxdata/telegraf/plugins/serializers"
 	"github.com/influxdata/telegraf/testutil"
 	"github.com/stretchr/testify/require"
 )
@@ -13,9 +14,11 @@ func TestConnectAndWrite(t *testing.T) {
 	}
 
 	var url = "amqp://" + testutil.GetLocalHost() + ":5672/"
+	s, _ := serializers.NewInfluxSerializer()
 	q := &AMQP{
-		URL:      url,
-		Exchange: "telegraf_test",
+		URL:        url,
+		Exchange:   "telegraf_test",
+		serializer: s,
 	}
 
 	// Verify that we can connect to the AMQP broker

plugins/outputs/file/README.md

@@ -0,0 +1 @@
+# file Output Plugin

plugins/outputs/file/file.go

@@ -0,0 +1,109 @@
+package file
+
+import (
+	"fmt"
+	"io"
+	"os"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/outputs"
+	"github.com/influxdata/telegraf/plugins/serializers"
+)
+
+type File struct {
+	Files []string
+
+	writer  io.Writer
+	closers []io.Closer
+
+	serializer serializers.Serializer
+}
+
+var sampleConfig = `
+  ### Files to write to, "stdout" is a specially handled file.
+  files = ["stdout", "/tmp/metrics.out"]
+
+  ### Data format to output. This can be "influx" or "graphite"
+  ### Each data format has it's own unique set of configuration options, read
+  ### more about them here:
+  ### https://github.com/influxdata/telegraf/blob/master/DATA_FORMATS_OUTPUT.md
+  data_format = "influx"
+`
+
+func (f *File) SetSerializer(serializer serializers.Serializer) {
+	f.serializer = serializer
+}
+
+func (f *File) Connect() error {
+	writers := []io.Writer{}
+	for _, file := range f.Files {
+		if file == "stdout" {
+			writers = append(writers, os.Stdout)
+			f.closers = append(f.closers, os.Stdout)
+		} else {
+			var of *os.File
+			var err error
+			if _, err := os.Stat(file); os.IsNotExist(err) {
+				of, err = os.Create(file)
+			} else {
+				of, err = os.OpenFile(file, os.O_APPEND|os.O_WRONLY, os.ModeAppend)
+			}
+
+			if err != nil {
+				return err
+			}
+			writers = append(writers, of)
+			f.closers = append(f.closers, of)
+		}
+	}
+	f.writer = io.MultiWriter(writers...)
+	return nil
+}
+
+func (f *File) Close() error {
+	var errS string
+	for _, c := range f.closers {
+		if err := c.Close(); err != nil {
+			errS += err.Error() + "\n"
+		}
+	}
+	if errS != "" {
+		return fmt.Errorf(errS)
+	}
+	return nil
+}
+
+func (f *File) SampleConfig() string {
+	return sampleConfig
+}
+
+func (f *File) Description() string {
+	return "Send telegraf metrics to file(s)"
+}
+
+func (f *File) Write(metrics []telegraf.Metric) error {
+	if len(metrics) == 0 {
+		return nil
+	}
+
+	for _, metric := range metrics {
+		values, err := f.serializer.Serialize(metric)
+		if err != nil {
+			return err
+		}
+
+		for _, value := range values {
+			_, err = f.writer.Write([]byte(value + "\n"))
+			if err != nil {
+				return fmt.Errorf("FAILED to write message: %s, %s", value, err)
+			}
+		}
+	}
+	return nil
+}
+
+func init() {
+	outputs.Add("file", func() telegraf.Output {
+		return &File{}
+	})
+}

plugins/outputs/file/file_test.go

@@ -0,0 +1 @@
+package file

plugins/outputs/graphite/graphite.go

@@ -3,14 +3,15 @@ package graphite
 import (
 	"errors"
 	"fmt"
-	"github.com/influxdata/telegraf"
-	"github.com/influxdata/telegraf/plugins/outputs"
 	"log"
 	"math/rand"
 	"net"
-	"sort"
 	"strings"
 	"time"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/plugins/outputs"
+	"github.com/influxdata/telegraf/plugins/serializers"
 )
 
 type Graphite struct {
@@ -71,42 +72,22 @@ func (g *Graphite) Description() string {
 func (g *Graphite) Write(metrics []telegraf.Metric) error {
 	// Prepare data
 	var bp []string
-	for _, metric := range metrics {
-		// Get name
-		name := metric.Name()
-		// Convert UnixNano to Unix timestamps
-		timestamp := metric.UnixNano() / 1000000000
-		tag_str := buildTags(metric)
+	s, err := serializers.NewGraphiteSerializer(g.Prefix)
+	if err != nil {
+		return err
+	}
 
-		for field_name, value := range metric.Fields() {
-			// Convert value
-			value_str := fmt.Sprintf("%#v", value)
-			// Write graphite metric
-			var graphitePoint string
-			if name == field_name {
-				graphitePoint = fmt.Sprintf("%s.%s %s %d\n",
-					tag_str,
-					strings.Replace(name, ".", "_", -1),
-					value_str,
-					timestamp)
-			} else {
-				graphitePoint = fmt.Sprintf("%s.%s.%s %s %d\n",
-					tag_str,
-					strings.Replace(name, ".", "_", -1),
-					strings.Replace(field_name, ".", "_", -1),
-					value_str,
-					timestamp)
-			}
-			if g.Prefix != "" {
-				graphitePoint = fmt.Sprintf("%s.%s", g.Prefix, graphitePoint)
-			}
-			bp = append(bp, graphitePoint)
+	for _, metric := range metrics {
+		gMetrics, err := s.Serialize(metric)
+		if err != nil {
+			log.Printf("Error serializing some metrics to graphite: %s", err.Error())
 		}
+		bp = append(bp, gMetrics...)
 	}
-	graphitePoints := strings.Join(bp, "")
+	graphitePoints := strings.Join(bp, "\n") + "\n"
 
 	// This will get set to nil if a successful write occurs
-	err := errors.New("Could not write to any Graphite server in cluster\n")
+	err = errors.New("Could not write to any Graphite server in cluster\n")
 
 	// Send data to a random server
 	p := rand.Perm(len(g.conns))
@@ -128,37 +109,6 @@ func (g *Graphite) Write(metrics []telegraf.Metric) error {
 	return err
 }
 
-func buildTags(metric telegraf.Metric) 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 {
-			tag_str += tag_value
-		} else {
-			tag_str += "." + tag_value
-		}
-	}
-	return tag_str
-}
-
 func init() {
 	outputs.Add("graphite", func() telegraf.Output {
 		return &Graphite{}