From 8482c40a911715b9d6e37ba01cd346939acb2933 Mon Sep 17 00:00:00 2001
From: maxunt <max.untrecht@gmail.com>
Date: Thu, 14 Jun 2018 13:17:32 -0700
Subject: [PATCH] Fix selection of tags under nested objects in the JSON parser
 (#4284)

---
 plugins/parsers/json/parser.go      | 57 ++++++++++++++++++++++-------
 plugins/parsers/json/parser_test.go | 27 ++++++++++++++
 2 files changed, 70 insertions(+), 14 deletions(-)

diff --git a/plugins/parsers/json/parser.go b/plugins/parsers/json/parser.go
index 8a3d15be7..62d17c76a 100644
--- a/plugins/parsers/json/parser.go
+++ b/plugins/parsers/json/parser.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"log"
 	"strconv"
 	"strings"
 	"time"
@@ -44,25 +45,15 @@ func (p *JSONParser) parseObject(metrics []telegraf.Metric, jsonOut map[string]i
 		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)
+	err := f.FullFlattenJSON("", jsonOut, true, true)
 	if err != nil {
 		return nil, err
 	}
 
-	metric, err := metric.New(p.MetricName, tags, f.Fields, time.Now().UTC())
+	tags, nFields := p.switchFieldToTag(tags, f.Fields)
+
+	metric, err := metric.New(p.MetricName, tags, nFields, time.Now().UTC())
 
 	if err != nil {
 		return nil, err
@@ -70,6 +61,43 @@ func (p *JSONParser) parseObject(metrics []telegraf.Metric, jsonOut map[string]i
 	return append(metrics, metric), nil
 }
 
+//will take in field map with strings and bools,
+//search for TagKeys that match fieldnames and add them to tags
+//will delete any strings/bools that shouldn't be fields
+//assumes that any non-numeric values in TagKeys should be displayed as tags
+func (p *JSONParser) switchFieldToTag(tags map[string]string, fields map[string]interface{}) (map[string]string, map[string]interface{}) {
+	for _, name := range p.TagKeys {
+		//switch any fields in tagkeys into tags
+		if fields[name] == nil {
+			continue
+		}
+		switch value := fields[name].(type) {
+		case string:
+			tags[name] = value
+			delete(fields, name)
+		case bool:
+			tags[name] = strconv.FormatBool(value)
+			delete(fields, name)
+		case float64:
+			tags[name] = strconv.FormatFloat(value, 'f', -1, 64)
+			delete(fields, name)
+		default:
+			log.Printf("E! [parsers.json] Unrecognized type %T", value)
+		}
+	}
+
+	//remove any additional string/bool values from fields
+	for k := range fields {
+		switch fields[k].(type) {
+		case string:
+			delete(fields, k)
+		case bool:
+			delete(fields, k)
+		}
+	}
+	return tags, fields
+}
+
 func (p *JSONParser) Parse(buf []byte) ([]telegraf.Metric, error) {
 	buf = bytes.TrimSpace(buf)
 	buf = bytes.TrimPrefix(buf, utf8BOM)
@@ -119,6 +147,7 @@ func (f *JSONFlattener) FlattenJSON(
 	if f.Fields == nil {
 		f.Fields = make(map[string]interface{})
 	}
+
 	return f.FullFlattenJSON(fieldname, v, false, false)
 }
 
diff --git a/plugins/parsers/json/parser_test.go b/plugins/parsers/json/parser_test.go
index c4f946364..c26b209a2 100644
--- a/plugins/parsers/json/parser_test.go
+++ b/plugins/parsers/json/parser_test.go
@@ -4,6 +4,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 const (
@@ -440,3 +441,29 @@ func TestHttpJsonBOM(t *testing.T) {
 	_, err := parser.Parse(jsonBOM)
 	assert.NoError(t, err)
 }
+
+//for testing issue #4260
+func TestJSONParseNestedArray(t *testing.T) {
+	testString := `{
+	"total_devices": 5,
+	"total_threads": 10,
+	"shares": {
+	"total": 5,
+	"accepted": 5,
+	"rejected": 0,
+	"avg_find_time": 4,
+	"tester": "work",
+	"tester2": "don't want this",
+	"tester3": 7.93
+	}
+	}`
+
+	parser := JSONParser{
+		MetricName: "json_test",
+		TagKeys:    []string{"total_devices", "total_threads", "shares_tester", "shares_tester3"},
+	}
+
+	metrics, err := parser.Parse([]byte(testString))
+	require.NoError(t, err)
+	require.Equal(t, len(parser.TagKeys), len(metrics[0].Tags()))
+}