Add Serializer plugins, and 'file' output plugin

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/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{}

plugins/outputs/graphite/graphite_test.go

@@ -43,6 +43,8 @@ func TestGraphiteOK(t *testing.T) {
 	// Start TCP server
 	wg.Add(1)
 	go TCPServer(t, &wg)
+	// Give the fake graphite TCP server some time to start:
+	time.Sleep(time.Millisecond * 100)
 
 	// Init plugin
 	g := Graphite{
@@ -95,32 +97,3 @@ func TCPServer(t *testing.T, wg *sync.WaitGroup) {
 	assert.Equal(t, "my.prefix.192_168_0_1.my_measurement.value 3.14 1289430000", data3)
 	conn.Close()
 }
-
-func TestGraphiteTags(t *testing.T) {
-	m1, _ := telegraf.NewMetric(
-		"mymeasurement",
-		map[string]string{"host": "192.168.0.1"},
-		map[string]interface{}{"value": float64(3.14)},
-		time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
-	)
-	m2, _ := telegraf.NewMetric(
-		"mymeasurement",
-		map[string]string{"host": "192.168.0.1", "afoo": "first", "bfoo": "second"},
-		map[string]interface{}{"value": float64(3.14)},
-		time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
-	)
-	m3, _ := telegraf.NewMetric(
-		"mymeasurement",
-		map[string]string{"afoo": "first", "bfoo": "second"},
-		map[string]interface{}{"value": float64(3.14)},
-		time.Date(2010, time.November, 10, 23, 0, 0, 0, time.UTC),
-	)
-
-	tags1 := buildTags(m1)
-	tags2 := buildTags(m2)
-	tags3 := buildTags(m3)
-
-	assert.Equal(t, "192_168_0_1", tags1)
-	assert.Equal(t, "192_168_0_1.first.second", tags2)
-	assert.Equal(t, "first.second", tags3)
-}