Skip to content

Commit

Permalink
Add Serializer plugins, and 'file' output plugin
Browse files Browse the repository at this point in the history
  • Loading branch information
sparrc committed Feb 12, 2016
1 parent 72f5c9b commit 45463c7
Show file tree
Hide file tree
Showing 22 changed files with 855 additions and 155 deletions.
3 changes: 1 addition & 2 deletions Godeps
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
43 changes: 43 additions & 0 deletions internal/config/config.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
1 change: 1 addition & 0 deletions plugins/outputs/all/all.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
32 changes: 26 additions & 6 deletions plugins/outputs/amqp/amqp.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"
)

Expand Down Expand Up @@ -39,6 +41,8 @@ type AMQP struct {
channel *amqp.Channel
sync.Mutex
headers amqp.Table

serializer serializers.Serializer
}

const (
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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
Expand Down
7 changes: 5 additions & 2 deletions plugins/outputs/amqp/amqp_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package amqp
import (
"testing"

"github.com/influxdata/telegraf/plugins/serializers"
"github.com/influxdata/telegraf/testutil"
"github.com/stretchr/testify/require"
)
Expand All @@ -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
Expand Down
1 change: 1 addition & 0 deletions plugins/outputs/file/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
# file Output Plugin
109 changes: 109 additions & 0 deletions plugins/outputs/file/file.go
Original file line number Diff line number Diff line change
@@ -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{}
})
}
Loading

0 comments on commit 45463c7

Please sign in to comment.