telegraf/plugins/parsers/json/parser.go

227 lines
5.1 KiB
Go

package json
import (
"bytes"
"encoding/json"
"fmt"
"strconv"
"strings"
"time"
"github.com/influxdata/telegraf"
"github.com/jmoiron/jsonq"
)
type JSONParser struct {
MetricName string
TagKeys []string
DefaultTags map[string]string
JSONPaths []string
}
func (p *JSONParser) parseArray(buf []byte) ([]telegraf.Metric, error) {
metrics := make([]telegraf.Metric, 0)
var jsonOut []map[string]interface{}
err := json.Unmarshal(buf, &jsonOut)
if err != nil {
err = fmt.Errorf("unable to parse out as JSON Array, %s", err)
return nil, err
}
for _, item := range jsonOut {
metrics, err = p.parseObject(metrics, item)
}
return metrics, nil
}
func (p *JSONParser) parseObject(metrics []telegraf.Metric, jsonOut map[string]interface{}) ([]telegraf.Metric, error) {
tags := make(map[string]string)
for k, v := range p.DefaultTags {
tags[k] = v
}
for _, tag := range p.TagKeys {
switch v := jsonOut[tag].(type) {
case string:
tags[tag] = v
case bool:
tags[tag] = strconv.FormatBool(v)
case float64:
tags[tag] = strconv.FormatFloat(v, 'f', -1, 64)
}
delete(jsonOut, tag)
}
f := JSONFlattener{}
err := f.FlattenJSON("", jsonOut)
if err != nil {
return nil, err
}
metric, err := telegraf.NewMetric(p.MetricName, tags, f.Fields, time.Now().UTC())
if err != nil {
return nil, err
}
return append(metrics, metric), nil
}
func (p *JSONParser) parsePath(metrics []telegraf.Metric, jsonOut map[string]interface{}) ([]telegraf.Metric, error) {
jq := jsonq.NewQuery(jsonOut)
tags := make(map[string]string)
for k, v := range p.DefaultTags {
tags[k] = v
}
for _, tag := range p.TagKeys {
if v, err := jq.String(tag); err == nil {
tags[tag] = v
}
}
for _, path := range p.JSONPaths {
pathByNode := strings.Split(path, ".")
if a, err := jq.ArrayOfObjects(pathByNode...); err == nil {
metrics, err = p.parseArrayPath(metrics, a, tags)
if err != nil {
return nil, err
}
parentObj, err := jq.Object(pathByNode[:len(pathByNode)-1]...)
if err != nil {
return nil, err
}
delete(parentObj, pathByNode[len(pathByNode)-1])
}
if o, err := jq.Object(pathByNode...); err == nil {
metrics, err = p.parseObjectPath(metrics, o, tags)
if err != nil {
return nil, err
}
parentObj, err := jq.Object(pathByNode[:len(pathByNode)-1]...)
if err != nil {
return nil, err
}
delete(parentObj, pathByNode[len(pathByNode)-1])
}
}
return metrics, nil
}
func (p *JSONParser) parseArrayPath(metrics []telegraf.Metric, jsonOut []map[string]interface{}, tags map[string]string) ([]telegraf.Metric, error) {
var err error
for _, doc := range jsonOut {
metrics, err = p.parseObjectPath(metrics, doc, tags)
if err != nil {
return nil, err
}
}
return metrics, nil
}
func (p *JSONParser) parseObjectPath(metrics []telegraf.Metric, jsonOut map[string]interface{}, tags map[string]string) ([]telegraf.Metric, error) {
jq := jsonq.NewQuery(jsonOut)
for _, tag := range p.TagKeys {
if v, err := jq.String(tag); err == nil {
tags[tag] = v
}
}
f := JSONFlattener{}
err := f.FlattenJSON("", jsonOut)
if err != nil {
return nil, err
}
metric, err := telegraf.NewMetric(p.MetricName, tags, f.Fields, time.Now().UTC())
if err != nil {
return nil, err
}
return append(metrics, metric), nil
}
func (p *JSONParser) Parse(buf []byte) ([]telegraf.Metric, error) {
if !isarray(buf) {
metrics := make([]telegraf.Metric, 0)
var jsonOut map[string]interface{}
err := json.Unmarshal(buf, &jsonOut)
if err != nil {
err = fmt.Errorf("unable to parse out as JSON, %s", err)
return nil, err
}
if len(p.JSONPaths) > 0 {
metrics, err = p.parsePath(metrics, jsonOut)
}
return p.parseObject(metrics, jsonOut)
}
return p.parseArray(buf)
}
func (p *JSONParser) ParseLine(line string) (telegraf.Metric, error) {
metrics, err := p.Parse([]byte(line + "\n"))
if err != nil {
return nil, err
}
if len(metrics) < 1 {
return nil, fmt.Errorf("Can not parse the line: %s, for data format: influx ", line)
}
return metrics[0], nil
}
func (p *JSONParser) SetDefaultTags(tags map[string]string) {
p.DefaultTags = tags
}
type JSONFlattener struct {
Fields map[string]interface{}
}
// FlattenJSON flattens nested maps/interfaces into a fields map
func (f *JSONFlattener) FlattenJSON(
fieldname string,
v interface{},
) error {
if f.Fields == nil {
f.Fields = make(map[string]interface{})
}
fieldname = strings.Trim(fieldname, "_")
switch t := v.(type) {
case map[string]interface{}:
for k, v := range t {
err := f.FlattenJSON(fieldname+"_"+k+"_", v)
if err != nil {
return err
}
}
case []interface{}:
for i, v := range t {
k := strconv.Itoa(i)
err := f.FlattenJSON(fieldname+"_"+k+"_", v)
if err != nil {
return nil
}
}
case float64:
f.Fields[fieldname] = t
case bool, string, nil:
// ignored types
return nil
default:
return fmt.Errorf("JSON Flattener: got unexpected type %T with value %v (%s)",
t, t, fieldname)
}
return nil
}
func isarray(buf []byte) bool {
ia := bytes.IndexByte(buf, '[')
ib := bytes.IndexByte(buf, '{')
if ia > -1 && ia < ib {
return true
} else {
return false
}
}