add librato output plugin, update datadog plugin to skip non-number m…

…etrics

outputs/all/all.go

@@ -5,6 +5,7 @@ import (
 	_ "github.com/influxdb/telegraf/outputs/datadog"
 	_ "github.com/influxdb/telegraf/outputs/influxdb"
 	_ "github.com/influxdb/telegraf/outputs/kafka"
+	_ "github.com/influxdb/telegraf/outputs/librato"
 	_ "github.com/influxdb/telegraf/outputs/mqtt"
 	_ "github.com/influxdb/telegraf/outputs/opentsdb"
 )

outputs/datadog/README.md

@@ -0,0 +1,7 @@
+# Datadog Output Plugin
+
+This plugin writes to the [Datadog Metrics API](http://docs.datadoghq.com/api/#metrics).
+
+If the point value being sent cannot be converted to a float64, the metric is skipped.
+
+Metrics are grouped by converting any `_` characters to `.` in the Point Name.

outputs/datadog/datadog.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"log"
 	"net/http"
 	"net/url"
 	"sort"
@@ -65,20 +66,25 @@ func (d *Datadog) Write(points []*client.Point) error {
 	if len(points) == 0 {
 		return nil
 	}
-	ts := TimeSeries{
-		Series: make([]*Metric, len(points)),
-	}
-	for index, pt := range points {
+	ts := TimeSeries{}
+	var tempSeries = make([]*Metric, len(points))
+	var acceptablePoints = 0
+	for _, pt := range points {
 		metric := &Metric{
 			Metric: strings.Replace(pt.Name(), "_", ".", -1),
 			Tags:   buildTags(pt.Tags()),
 			Host:   pt.Tags()["host"],
 		}
 		if p, err := buildPoint(pt); err == nil {
 			metric.Points[0] = p
+			tempSeries[acceptablePoints] = metric
+			acceptablePoints += 1
+		} else {
+			log.Printf("unable to build Metric for %s, skipping\n", pt.Name())
 		}
-		ts.Series[index] = metric
 	}
+	ts.Series = make([]*Metric, acceptablePoints)
+	copy(ts.Series, tempSeries[0:])
 	tsBytes, err := json.Marshal(ts)
 	if err != nil {
 		return fmt.Errorf("unable to marshal TimeSeries, %s\n", err.Error())

outputs/librato/README.md

@@ -0,0 +1,10 @@
+# Librato Output Plugin
+
+This plugin writes to the [Librato Metrics API](http://dev.librato.com/v1/metrics#metrics).
+
+The `source_tag` option in the Configuration file is used to send contextual information from
+Point Tags to the API.
+
+If the point value being sent cannot be converted to a float64, the metric is skipped.
+
+Currently, the plugin does not send any associated Point Tags.

outputs/librato/librato.go

@@ -0,0 +1,165 @@
+package librato
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+
+	"github.com/influxdb/influxdb/client/v2"
+	"github.com/influxdb/telegraf/duration"
+	"github.com/influxdb/telegraf/outputs"
+)
+
+type Librato struct {
+	User      string
+	Token     string
+	SourceTag string
+	Timeout   duration.Duration
+
+	apiUrl string
+	client *http.Client
+}
+
+var sampleConfig = `
+  # Librator API Docs
+  # http://dev.librato.com/v1/metrics-authentication
+
+  # Librato API user
+  token = "my-secret-token" # required.
+
+  # Librato API token
+  token = "my-secret-token" # required.
+
+  # Tag Field to populate source attribute (optional)
+  # This is typically the _hostname_ from which the metric was obtained.
+  source_tag = "hostname"
+
+  # Connection timeout.
+  # timeout = "5s"
+`
+
+type Metrics struct {
+	Gauges []*Gauge `json:"gauges"`
+}
+
+type Gauge struct {
+	Name        string  `json:"name"`
+	Value       float64 `json:"value"`
+	Source      string  `json:"source"`
+	MeasureTime int64   `json:"measure_time"`
+}
+
+const librato_api = "https://metrics-api.librato.com/v1/metrics"
+
+func NewLibrato(apiUrl string) *Librato {
+	return &Librato{
+		apiUrl: apiUrl,
+	}
+}
+
+func (l *Librato) Connect() error {
+	if l.User == "" || l.Token == "" {
+		return fmt.Errorf("user and token are required fields for librato output")
+	}
+	l.client = &http.Client{
+		Timeout: l.Timeout.Duration,
+	}
+	return nil
+}
+
+func (l *Librato) Write(points []*client.Point) error {
+	if len(points) == 0 {
+		return nil
+	}
+	metrics := Metrics{}
+	var tempGauges = make([]*Gauge, len(points))
+	var acceptablePoints = 0
+	for _, pt := range points {
+		if gauge, err := l.buildGauge(pt); err == nil {
+			tempGauges[acceptablePoints] = gauge
+			acceptablePoints += 1
+		} else {
+			log.Printf("unable to build Gauge for %s, skipping\n", pt.Name())
+		}
+	}
+	metrics.Gauges = make([]*Gauge, acceptablePoints)
+	copy(metrics.Gauges, tempGauges[0:])
+	metricsBytes, err := json.Marshal(metrics)
+	if err != nil {
+		return fmt.Errorf("unable to marshal Metrics, %s\n", err.Error())
+	}
+	req, err := http.NewRequest("POST", l.apiUrl, bytes.NewBuffer(metricsBytes))
+	if err != nil {
+		return fmt.Errorf("unable to create http.Request, %s\n", err.Error())
+	}
+	req.Header.Add("Content-Type", "application/json")
+	req.SetBasicAuth(l.User, l.Token)
+
+	resp, err := l.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("error POSTing metrics, %s\n", err.Error())
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != 200 {
+		return fmt.Errorf("received bad status code, %d\n", resp.StatusCode)
+	}
+
+	return nil
+}
+
+func (l *Librato) SampleConfig() string {
+	return sampleConfig
+}
+
+func (l *Librato) Description() string {
+	return "Configuration for Librato API to send metrics to."
+}
+
+func (l *Librato) buildGauge(pt *client.Point) (*Gauge, error) {
+	gauge := &Gauge{
+		Name:        pt.Name(),
+		MeasureTime: pt.Time().Unix(),
+	}
+	if err := gauge.setValue(pt.Fields()["value"]); err != nil {
+		return gauge, fmt.Errorf("unable to extract value from Fields, %s\n", err.Error())
+	}
+	if l.SourceTag != "" {
+		if source, ok := pt.Tags()[l.SourceTag]; ok {
+			gauge.Source = source
+		} else {
+			return gauge, fmt.Errorf("undeterminable Source type from Field, %s\n", l.SourceTag)
+		}
+	}
+	return gauge, nil
+}
+
+func (g *Gauge) setValue(v interface{}) error {
+	switch d := v.(type) {
+	case int:
+		g.Value = float64(int(d))
+	case int32:
+		g.Value = float64(int32(d))
+	case int64:
+		g.Value = float64(int64(d))
+	case float32:
+		g.Value = float64(d)
+	case float64:
+		g.Value = float64(d)
+	default:
+		return fmt.Errorf("undeterminable type %+v", d)
+	}
+	return nil
+}
+
+func (l *Librato) Close() error {
+	return nil
+}
+
+func init() {
+	outputs.Add("librato", func() outputs.Output {
+		return NewLibrato(librato_api)
+	})
+}