Added capability to send metrics through Http API for OpenTSDB

CHANGELOG.md

@@ -9,6 +9,7 @@
 - [#1572](https://github.com/influxdata/telegraf/pull/1572): mesos improvements.
 - [#1513](https://github.com/influxdata/telegraf/issues/1513): Add Ceph Cluster Performance Statistics
 - [#1650](https://github.com/influxdata/telegraf/issues/1650): Ability to configure response_timeout in httpjson input.
+- [#1539](https://github.com/influxdata/telegraf/pull/1539): Added capability to send metrics through Http API for OpenTSDB.
 
 ### Bugfixes
 

plugins/outputs/opentsdb/README.md

@@ -1,6 +1,12 @@
 # OpenTSDB Output Plugin
 
-This plugin writes to a OpenTSDB instance using the "telnet" mode
+This plugin writes to an OpenTSDB instance using either the "telnet" or Http mode.
+
+Using the Http API is the recommended way of writing metrics since OpenTSDB 2.0
+To use Http mode, set useHttp to true in config. You can also control how many
+metrics is sent in each http request by setting batchSize in config.
+
+See http://opentsdb.net/docs/build/html/api_http/put.html for details.
 
 ## Transfer "Protocol" in the telnet mode
 
@@ -10,14 +16,14 @@ The expected input from OpenTSDB is specified in the following way:
 put <metric> <timestamp> <value> <tagk1=tagv1[ tagk2=tagv2 ...tagkN=tagvN]>
 ```
 
-The telegraf output plugin adds an optional prefix to the metric keys so 
+The telegraf output plugin adds an optional prefix to the metric keys so
 that a subamount can be selected.
 
 ```
 put <[prefix.]metric> <timestamp> <value> <tagk1=tagv1[ tagk2=tagv2 ...tagkN=tagvN]>
 ```
 
-### Example 
+### Example
 
 ```
 put nine.telegraf.system_load1 1441910356 0.430000 dc=homeoffice host=irimame scope=green
@@ -38,12 +44,12 @@ put nine.telegraf.ping_average_response_ms 1441910366 24.006000 dc=homeoffice ho
 ...
 ```
 
-## 
+##
 
-The OpenTSDB interface can be simulated with this reader:
+The OpenTSDB telnet interface can be simulated with this reader:
 
 ```
-// opentsdb_telnet_mode_mock.go 
+// opentsdb_telnet_mode_mock.go
 package main
 
 import (
@@ -75,4 +81,4 @@ func main() {
 
 ## Allowed values for metrics
 
-OpenTSDB allows `integers` and `floats` as input values
+OpenTSDB allows `integers` and `floats` as input values

plugins/outputs/opentsdb/opentsdb.go

@@ -3,10 +3,10 @@ package opentsdb
 import (
 	"fmt"
 	"net"
+	"net/url"
 	"sort"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/influxdata/telegraf"
 	"github.com/influxdata/telegraf/plugins/outputs"
@@ -18,6 +18,8 @@ type OpenTSDB struct {
 	Host string
 	Port int
 
+	HttpBatchSize int
+
 	Debug bool
 }
 
@@ -28,27 +30,41 @@ var sampleConfig = `
   ## prefix for metrics keys
   prefix = "my.specific.prefix."
 
-  ## Telnet Mode ##
-  ## DNS name of the OpenTSDB server in telnet mode
+  ## DNS name of the OpenTSDB server
+  ## Using "opentsdb.example.com" or "tcp://opentsdb.example.com" will use the
+  ## telnet API. "http://opentsdb.example.com" will use the Http API.
   host = "opentsdb.example.com"
 
-  ## Port of the OpenTSDB server in telnet mode
+  ## Port of the OpenTSDB server
   port = 4242
 
+  ## Number of data points to send to OpenTSDB in Http requests.
+  ## Not used with telnet API.
+  httpBatchSize = 50
+
   ## Debug true - Prints OpenTSDB communication
   debug = false
 `
 
-type MetricLine struct {
-	Metric    string
-	Timestamp int64
-	Value     string
-	Tags      string
+func ToLineFormat(tags map[string]string) string {
+	tagsArray := make([]string, len(tags))
+	index := 0
+	for k, v := range tags {
+		tagsArray[index] = fmt.Sprintf("%s=%s", k, v)
+		index++
+	}
+	sort.Strings(tagsArray)
+	return strings.Join(tagsArray, " ")
 }
 
 func (o *OpenTSDB) Connect() error {
 	// Test Connection to OpenTSDB Server
-	uri := fmt.Sprintf("%s:%d", o.Host, o.Port)
+	u, err := url.Parse(o.Host)
+	if err != nil {
+		return fmt.Errorf("Error in parsing host url: %s", err.Error())
+	}
+
+	uri := fmt.Sprintf("%s:%d", u.Host, o.Port)
 	tcpAddr, err := net.ResolveTCPAddr("tcp", uri)
 	if err != nil {
 		return fmt.Errorf("OpenTSDB: TCP address cannot be resolved")
@@ -65,10 +81,64 @@ func (o *OpenTSDB) Write(metrics []telegraf.Metric) error {
 	if len(metrics) == 0 {
 		return nil
 	}
-	now := time.Now()
 
+	u, err := url.Parse(o.Host)
+	if err != nil {
+		return fmt.Errorf("Error in parsing host url: %s", err.Error())
+	}
+
+	if u.Scheme == "" || u.Scheme == "tcp" {
+		return o.WriteTelnet(metrics, u)
+	} else if u.Scheme == "http" {
+		return o.WriteHttp(metrics, u)
+	} else {
+		return fmt.Errorf("Unknown scheme in host parameter.")
+	}
+}
+
+func (o *OpenTSDB) WriteHttp(metrics []telegraf.Metric, u *url.URL) error {
+	http := openTSDBHttp{
+		Host:      u.Host,
+		Port:      o.Port,
+		BatchSize: o.HttpBatchSize,
+		Debug:     o.Debug,
+	}
+
+	for _, m := range metrics {
+		now := m.UnixNano() / 1000000000
+		tags := cleanTags(m.Tags())
+
+		for fieldName, value := range m.Fields() {
+			metricValue, buildError := buildValue(value)
+			if buildError != nil {
+				fmt.Printf("OpenTSDB: %s\n", buildError.Error())
+				continue
+			}
+
+			metric := &HttpMetric{
+				Metric: sanitizedChars.Replace(fmt.Sprintf("%s%s_%s",
+					o.Prefix, m.Name(), fieldName)),
+				Tags:      tags,
+				Timestamp: now,
+				Value:     metricValue,
+			}
+
+			if err := http.sendDataPoint(metric); err != nil {
+				return err
+			}
+		}
+	}
+
+	if err := http.flush(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (o *OpenTSDB) WriteTelnet(metrics []telegraf.Metric, u *url.URL) error {
 	// Send Data with telnet / socket communication
-	uri := fmt.Sprintf("%s:%d", o.Host, o.Port)
+	uri := fmt.Sprintf("%s:%d", u.Host, o.Port)
 	tcpAddr, _ := net.ResolveTCPAddr("tcp", uri)
 	connection, err := net.DialTCP("tcp", nil, tcpAddr)
 	if err != nil {
@@ -77,9 +147,20 @@ func (o *OpenTSDB) Write(metrics []telegraf.Metric) error {
 	defer connection.Close()
 
 	for _, m := range metrics {
-		for _, metric := range buildMetrics(m, now, o.Prefix) {
+		now := m.UnixNano() / 1000000000
+		tags := ToLineFormat(cleanTags(m.Tags()))
+
+		for fieldName, value := range m.Fields() {
+			metricValue, buildError := buildValue(value)
+			if buildError != nil {
+				fmt.Printf("OpenTSDB: %s\n", buildError.Error())
+				continue
+			}
+
 			messageLine := fmt.Sprintf("put %s %v %s %s\n",
-				metric.Metric, metric.Timestamp, metric.Value, metric.Tags)
+				sanitizedChars.Replace(fmt.Sprintf("%s%s_%s", o.Prefix, m.Name(), fieldName)),
+				now, metricValue, tags)
+
 			if o.Debug {
 				fmt.Print(messageLine)
 			}
@@ -93,37 +174,12 @@ func (o *OpenTSDB) Write(metrics []telegraf.Metric) error {
 	return nil
 }
 
-func buildTags(mTags map[string]string) []string {
-	tags := make([]string, len(mTags))
-	index := 0
-	for k, v := range mTags {
-		tags[index] = sanitizedChars.Replace(fmt.Sprintf("%s=%s", k, v))
-		index++
-	}
-	sort.Strings(tags)
-	return tags
-}
-
-func buildMetrics(m telegraf.Metric, now time.Time, prefix string) []*MetricLine {
-	ret := []*MetricLine{}
-	for fieldName, value := range m.Fields() {
-		metric := &MetricLine{
-			Metric: sanitizedChars.Replace(fmt.Sprintf("%s%s_%s",
-				prefix, m.Name(), fieldName)),
-			Timestamp: now.Unix(),
-		}
-
-		metricValue, buildError := buildValue(value)
-		if buildError != nil {
-			fmt.Printf("OpenTSDB: %s\n", buildError.Error())
-			continue
-		}
-		metric.Value = metricValue
-		tagsSlice := buildTags(m.Tags())
-		metric.Tags = fmt.Sprint(strings.Join(tagsSlice, " "))
-		ret = append(ret, metric)
+func cleanTags(tags map[string]string) map[string]string {
+	tagSet := make(map[string]string, len(tags))
+	for k, v := range tags {
+		tagSet[sanitizedChars.Replace(k)] = sanitizedChars.Replace(v)
 	}
-	return ret
+	return tagSet
 }
 
 func buildValue(v interface{}) (string, error) {