Splunk Metrics serializer

plugins/serializers/registry.go

@@ -9,6 +9,7 @@ import (
 	"github.com/influxdata/telegraf/plugins/serializers/graphite"
 	"github.com/influxdata/telegraf/plugins/serializers/influx"
 	"github.com/influxdata/telegraf/plugins/serializers/json"
+	"github.com/influxdata/telegraf/plugins/serializers/splunkmetric"
 )
 
 // SerializerOutput is an interface for output plugins that are able to
@@ -73,6 +74,8 @@ func NewSerializer(config *Config) (Serializer, error) {
 		serializer, err = NewGraphiteSerializer(config.Prefix, config.Template, config.GraphiteTagSupport)
 	case "json":
 		serializer, err = NewJsonSerializer(config.TimestampUnits)
+	case "splunkmetric":
+		serializer, err = NewSplunkmetricSerializer(config.TimestampUnits)
 	default:
 		err = fmt.Errorf("Invalid data format: %s", config.DataFormat)
 	}
@@ -83,6 +86,10 @@ func NewJsonSerializer(timestampUnits time.Duration) (Serializer, error) {
 	return json.NewSerializer(timestampUnits)
 }
 
+func NewSplunkmetricSerializer(timestampUnits time.Duration) (Serializer, error) {
+	return splunkmetric.NewSerializer(timestampUnits)
+}
+
 func NewInfluxSerializerConfig(config *Config) (Serializer, error) {
 	var sort influx.FieldSortOrder
 	if config.InfluxSortFields {

plugins/serializers/splunkmetric/README.md

@@ -0,0 +1,85 @@
+# Splunk Metrics serialzier
+
+This serializer formats and outputs the metric data in a format that can be consumed by a Splunk metrics index. It can be used to write to a file using the file output, or for sending metrics to a HEC using the standard telegraf HTTP output. If you're using the HTTP output, this serializer knows how to batch the metrics so you don't end up with an HTTP POST per metric.
+
+Th data is output in a format that conforms to the specified Splunk HEC JSON format as found here: [Send metrics in JSON format](http://dev.splunk.com/view/event-collector/SP-CAAAFDN).
+
+An example event looks like:
+```javascript
+{
+  "time": 1529708430,
+  "event": "metric",
+  "host": "patas-mbp",
+  "fields": {
+    "_value": 0.6,
+    "cpu": "cpu0",
+    "dc": "mobile",
+    "metric_name": "cpu.usage_user",
+    "user": "ronnocol"
+  }
+}
+```
+In the above snippet, the following keys are dimensions:
+* cpu
+* dc
+* user
+
+## Using with the HTTP output
+
+To send this data to a Splunk HEC, you can use the HTTP output, there are some custom headers that you need to add
+to manage the HEC authorization, here's a sample config for an HTTP output:
+
+```toml
+[[outputs.http]]
+#   ## URL is the address to send metrics to
+   url = "https://localhost:8088/services/collector"
+#
+#   ## Timeout for HTTP message
+#   # timeout = "5s"
+#
+#   ## HTTP method, one of: "POST" or "PUT"
+#   # method = "POST"
+#
+#   ## HTTP Basic Auth credentials
+#   # username = "username"
+#   # password = "pa$$word"
+#
+#   ## Optional TLS Config
+#   # tls_ca = "/etc/telegraf/ca.pem"
+#   # tls_cert = "/etc/telegraf/cert.pem"
+#   # tls_key = "/etc/telegraf/key.pem"
+#   ## Use TLS but skip chain & host verification
+#   # insecure_skip_verify = false
+#
+#   ## Data format to output.
+#   ## Each data format has it's own unique set of configuration options, read
+#   ## more about them here:
+#   ## https://github.com/influxdata/telegraf/blob/master/docs/DATA_FORMATS_OUTPUT.md
+   data_format = "splunkmetric"
+#
+#   ## Additional HTTP headers
+    [outputs.http.headers]
+#   # Should be set manually to "application/json" for json data_format
+      Content-Type = "application/json"
+      Authorization = "Splunk xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
+      X-Splunk-Request-Channel = "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"
+```
+
+## Overrides
+You can override the default values for the HEC token you are using by adding additional tags to the config file.
+
+The following aspects of the token can be overriden with tags:
+* index
+* source
+
+You can either use `[global_tags]` or using a more advanced configuration as documented [here](https://github.com/influxdata/telegraf/blob/master/docs/CONFIGURATION.md).
+
+Such as this example which overrides the index just on the cpu metric:
+```toml
+[[inputs.cpu]]
+  percpu = false
+  totalcpu = true
+  [inputs.cpu.tags]
+    index = "cpu_metrics"
+```
+

plugins/serializers/splunkmetric/splunkmetric.go

@@ -0,0 +1,135 @@
+package splunkmetric
+
+import (
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/influxdata/telegraf"
+)
+
+type serializer struct {
+	TimestampUnits time.Duration
+}
+
+func NewSerializer(timestampUnits time.Duration) (*serializer, error) {
+	s := &serializer{
+		TimestampUnits: truncateDuration(timestampUnits),
+	}
+	return s, nil
+}
+
+func (s *serializer) Serialize(metric telegraf.Metric) ([]byte, error) {
+	var serialized string
+
+	m, err := s.createObject(metric)
+	if err == nil {
+		serialized = m + "\n"
+	}
+
+	return []byte(serialized), nil
+}
+
+func (s *serializer) SerializeBatch(metrics []telegraf.Metric) ([]byte, error) {
+
+	var serialized string
+
+	var objects []string
+
+	for _, metric := range metrics {
+		m, err := s.createObject(metric)
+		if err == nil {
+			objects = append(objects, m)
+		}
+	}
+
+	for _, m := range objects {
+		serialized = serialized + m + "\n"
+	}
+
+	return []byte(serialized), nil
+}
+
+func (s *serializer) createObject(metric telegraf.Metric) (metricString string, err error) {
+
+	/* Splunk supports one metric per line and has the following required names:
+	 ** metric_name: The name of the metric
+	 ** _value:      The value for the metric
+	 ** _time:       The timestamp for the metric
+	 ** All other index fields become deminsions.
+	 */
+	type HECTimeSeries struct {
+		Time   float64                `json:"time"`
+		Event  string                 `json:"event"`
+		Host   string                 `json:"host,omitempty"`
+		Index  string                 `json:"index,omitempty"`
+		Source string                 `json:"source,omitempty"`
+		Fields map[string]interface{} `json:"fields"`
+	}
+
+	dataGroup := HECTimeSeries{}
+
+	for k, v := range metric.Fields() {
+
+		if !verifyValue(v) {
+			err = errors.New("can not parse value")
+			return "", err
+		}
+
+		obj := map[string]interface{}{}
+		obj["metric_name"] = metric.Name() + "." + k
+		obj["_value"] = v
+
+		dataGroup.Event = "metric"
+		dataGroup.Time = float64(metric.Time().UnixNano() / int64(s.TimestampUnits))
+		dataGroup.Fields = obj
+
+		// Break tags out into key(n)=value(t) pairs
+		for n, t := range metric.Tags() {
+			if n == "host" {
+				dataGroup.Host = t
+			} else if n == "index" {
+				dataGroup.Index = t
+			} else if n == "source" {
+				dataGroup.Source = t
+			} else {
+				dataGroup.Fields[n] = t
+			}
+		}
+		dataGroup.Fields["metric_name"] = metric.Name() + "." + k
+		dataGroup.Fields["_value"] = v
+	}
+
+	metricJson, err := json.Marshal(dataGroup)
+
+	if err != nil {
+		return "", err
+	}
+
+	metricString = string(metricJson)
+	return metricString, nil
+}
+
+func verifyValue(v interface{}) bool {
+	switch v.(type) {
+	case string:
+		return false
+	}
+	return true
+}
+
+func truncateDuration(units time.Duration) time.Duration {
+	// Default precision is 1s
+	if units <= 0 {
+		return time.Second
+	}
+
+	// Search for the power of ten less than the duration
+	d := time.Nanosecond
+	for {
+		if d*10 > units {
+			return d
+		}
+		d = d * 10
+	}
+}

plugins/serializers/splunkmetric/splunkmetric_test.go

@@ -0,0 +1,108 @@
+package splunkmetric
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/influxdata/telegraf"
+	"github.com/influxdata/telegraf/metric"
+)
+
+func MustMetric(v telegraf.Metric, err error) telegraf.Metric {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}
+
+func TestSerializeMetricFloat(t *testing.T) {
+	now := time.Unix(0, 0)
+	tags := map[string]string{
+		"cpu": "cpu0",
+	}
+	fields := map[string]interface{}{
+		"usage_idle": float64(91.5),
+	}
+	m, err := metric.New("cpu", tags, fields, now)
+	assert.NoError(t, err)
+
+	s, _ := NewSerializer(0)
+	var buf []byte
+	buf, err = s.Serialize(m)
+	assert.NoError(t, err)
+	expS := `{"time":0,"event":"metric","fields":{"_value":91.5,"cpu":"cpu0","metric_name":"cpu.usage_idle"}}` + "\n"
+	assert.Equal(t, string(expS), string(buf))
+}
+
+func TestSerializeMetricInt(t *testing.T) {
+	now := time.Unix(0, 0)
+	tags := map[string]string{
+		"cpu": "cpu0",
+	}
+	fields := map[string]interface{}{
+		"usage_idle": int64(90),
+	}
+	m, err := metric.New("cpu", tags, fields, now)
+	assert.NoError(t, err)
+
+	s, _ := NewSerializer(0)
+	var buf []byte
+	buf, err = s.Serialize(m)
+	assert.NoError(t, err)
+
+	expS := `{"time":0,"event":"metric","fields":{"_value":90,"cpu":"cpu0","metric_name":"cpu.usage_idle"}}` + "\n"
+	assert.Equal(t, string(expS), string(buf))
+}
+
+func TestSerializeMetricString(t *testing.T) {
+	now := time.Now()
+	tags := map[string]string{
+		"cpu": "cpu0",
+	}
+	fields := map[string]interface{}{
+		"usage_idle": "foobar",
+	}
+	m, err := metric.New("cpu", tags, fields, now)
+	assert.NoError(t, err)
+
+	s, _ := NewSerializer(0)
+	var buf []byte
+	buf, err = s.Serialize(m)
+	assert.NoError(t, err)
+
+	expS := ""
+	assert.Equal(t, string(expS), string(buf))
+}
+
+func TestSerializeBatch(t *testing.T) {
+	m := MustMetric(
+		metric.New(
+			"cpu",
+			map[string]string{},
+			map[string]interface{}{
+				"value": 42.0,
+			},
+			time.Unix(0, 0),
+		),
+	)
+	n := MustMetric(
+		metric.New(
+			"cpu",
+			map[string]string{},
+			map[string]interface{}{
+				"value": 92.0,
+			},
+			time.Unix(0, 0),
+		),
+	)
+
+	metrics := []telegraf.Metric{m, n}
+	s, _ := NewSerializer(0)
+	buf, err := s.SerializeBatch(metrics)
+	assert.NoError(t, err)
+
+	expS := `{"time":0,"event":"metric","fields":{"_value":42,"metric_name":"cpu.value"}}` + "\n" + `{"time":0,"event":"metric","fields":{"_value":92,"metric_name":"cpu.value"}}` + "\n"
+	assert.Equal(t, string(expS), string(buf))
+}