From b96dd8c60f341cf3607268e56fe0a9b9c15cb3a9 Mon Sep 17 00:00:00 2001 From: Thibault Cohen Date: Thu, 17 Dec 2015 15:36:31 -0500 Subject: [PATCH] Add Graphite output --- outputs/all/all.go | 1 + outputs/graphite/graphite.go | 134 +++++++++++++++++++++++++++++++++++ 2 files changed, 135 insertions(+) create mode 100644 outputs/graphite/graphite.go diff --git a/outputs/all/all.go b/outputs/all/all.go index 08ebf254994f7..e86ebea788d14 100644 --- a/outputs/all/all.go +++ b/outputs/all/all.go @@ -4,6 +4,7 @@ import ( _ "github.com/influxdb/telegraf/outputs/amon" _ "github.com/influxdb/telegraf/outputs/amqp" _ "github.com/influxdb/telegraf/outputs/datadog" + _ "github.com/influxdb/telegraf/outputs/graphite" _ "github.com/influxdb/telegraf/outputs/influxdb" _ "github.com/influxdb/telegraf/outputs/kafka" _ "github.com/influxdb/telegraf/outputs/kinesis" diff --git a/outputs/graphite/graphite.go b/outputs/graphite/graphite.go new file mode 100644 index 0000000000000..264d5c3fb7d55 --- /dev/null +++ b/outputs/graphite/graphite.go @@ -0,0 +1,134 @@ +package graphite + +import ( + "errors" + "fmt" + "github.com/influxdb/influxdb/client/v2" + "github.com/influxdb/telegraf/outputs" + "log" + "math/rand" + "net" + "strings" + "time" +) + +type Graphite struct { + // URL is only for backwards compatability + Servers []string + Prefix string + Timeout int + conns []net.Conn +} + +var sampleConfig = ` + # TCP raw endpoint for your graphite instance. + servers = ["mygraphiteserver:2003"] # default "localhost:2003" + # Prefix metrics name + prefix = "" # default "" + # Timeout in second + timeout = 2 # default 2 +` + +func (g *Graphite) Connect() error { + // Set default values + if g.Timeout <= 0 { + g.Timeout = 2 + } + if len(g.Servers) == 0 { + g.Servers = append(g.Servers, "localhost:2003") + } + // Get Connections + var conns []net.Conn + for _, server := range g.Servers { + conn, err := net.DialTimeout("tcp", server, time.Duration(g.Timeout)*time.Second) + if err == nil { + conns = append(conns, conn) + } + } + g.conns = conns + return nil +} + +func (g *Graphite) Close() error { + // Closing all connections + for _, conn := range g.conns { + conn.Close() + } + return nil +} + +func (g *Graphite) SampleConfig() string { + return sampleConfig +} + +func (g *Graphite) Description() string { + return "Configuration for Graphite server to send metrics to using TCP raw protocol" +} + +// Choose a random server in the cluster to write to until a successful write +// occurs, logging each unsuccessful. If all servers fail, return error. +func (g *Graphite) Write(points []*client.Point) error { + // Prepare data + var bp []string + for _, point := range points { + // Get name + names := strings.SplitN(point.Name(), "_", 2) + var name string + if len(names) > 1 { + name = names[1] + } else { + name = point.Name() + } + // Convert UnixNano to Unix timestamps + timestamp := point.UnixNano() / 1000000000 + + for field_name, value := range point.Fields() { + var graphitePoint string + if name == field_name { + graphitePoint = fmt.Sprintf("%s.%s %s %d\n", + strings.Replace(point.Tags()["host"], ".", "_", -1), + strings.Replace(name, ".", "_", -1), + value, + timestamp) + } else { + graphitePoint = fmt.Sprintf("%s.%s.%s %s %d\n", + strings.Replace(point.Tags()["host"], ".", "_", -1), + strings.Replace(name, ".", "_", -1), + strings.Replace(field_name, ".", "_", -1), + value, + timestamp) + } + if g.Prefix != "" { + graphitePoint = fmt.Sprintf("%s.%s", g.Prefix, graphitePoint) + } + bp = append(bp, graphitePoint) + //fmt.Printf(graphitePoint) + } + } + graphitePoints := strings.Join(bp, "\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") + + // Send data to a random server + p := rand.Perm(len(g.conns)) + for _, n := range p { + if _, e := fmt.Fprintf(g.conns[n], graphitePoints); e != nil { + // Error + log.Println("ERROR: " + err.Error()) + // Let's try the next one + } else { + // Success + err = nil + break + } + } + + return err +} + +func init() { + outputs.Add("graphite", func() outputs.Output { + return &Graphite{} + }) +}