telegraf/plugins/serializers/splunkmetric/splunkmetric.go

127 lines
3.2 KiB
Go
Raw Normal View History

2018-09-11 20:01:08 +00:00
package splunkmetric
import (
"encoding/json"
"fmt"
"log"
"github.com/influxdata/telegraf"
)
type serializer struct {
HecRouting bool
}
func NewSerializer(splunkmetric_hec_routing bool) (*serializer, error) {
s := &serializer{
HecRouting: splunkmetric_hec_routing,
}
return s, nil
}
func (s *serializer) Serialize(metric telegraf.Metric) ([]byte, error) {
m, err := s.createObject(metric)
if err != nil {
return nil, fmt.Errorf("D! [serializer.splunkmetric] Dropping invalid metric: %s", metric.Name())
}
return m, nil
}
func (s *serializer) SerializeBatch(metrics []telegraf.Metric) ([]byte, error) {
var serialized []byte
for _, metric := range metrics {
m, err := s.createObject(metric)
if err != nil {
return nil, fmt.Errorf("D! [serializer.splunkmetric] Dropping invalid metric: %s", metric.Name())
} else if m != nil {
serialized = append(serialized, m...)
}
}
return serialized, nil
}
func (s *serializer) createObject(metric telegraf.Metric) (metricGroup []byte, err error) {
/* Splunk supports one metric json object, and does _not_ support an array of JSON objects.
** Splunk has the following required names for the metric store:
** metric_name: The name of the metric
** _value: The value for the metric
** time: The timestamp for the metric
2018-10-19 18:12:01 +00:00
** All other index fields become dimensions.
2018-09-11 20:01:08 +00:00
*/
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{}
var metricJson []byte
for _, field := range metric.FieldList() {
if !verifyValue(field.Value) {
log.Printf("D! Can not parse value: %v for key: %v", field.Value, field.Key)
continue
}
obj := map[string]interface{}{}
obj["metric_name"] = metric.Name() + "." + field.Key
obj["_value"] = field.Value
dataGroup.Event = "metric"
// Convert ns to float seconds since epoch.
dataGroup.Time = float64(metric.Time().UnixNano()) / float64(1000000000)
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() + "." + field.Key
dataGroup.Fields["_value"] = field.Value
switch s.HecRouting {
case true:
// Output the data as a fields array and host,index,time,source overrides for the HEC.
metricJson, err = json.Marshal(dataGroup)
default:
// Just output the data and the time, useful for file based outuputs
dataGroup.Fields["time"] = dataGroup.Time
metricJson, err = json.Marshal(dataGroup.Fields)
}
metricGroup = append(metricGroup, metricJson...)
if err != nil {
return nil, err
}
}
return metricGroup, nil
}
func verifyValue(v interface{}) bool {
switch v.(type) {
case string:
return false
}
return true
}