json_decode_fields processor

.gitignore

@@ -4,6 +4,7 @@
 /build
 /*/data
 /*/logs
+/.vscode
 
 # Files
 .DS_Store

CHANGELOG.asciidoc

@@ -32,6 +32,7 @@ https://github.com/elastic/beats/compare/v5.0.0...master[Check the HEAD diff]
 ==== Bugfixes
 
 *Affecting all Beats*
+- Added decode_json_fields processor for decoding fields containing JSON strings. {pull}2605[2605]
 
 *Metricbeat*
 

filebeat/harvester/reader/json.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 
 	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/jsontransform"
 	"github.com/elastic/beats/libbeat/logp"
 )
 
@@ -69,51 +70,10 @@ func unmarshal(text []byte, fields *map[string]interface{}) error {
 	if err != nil {
 		return err
 	}
-	transformNumbersDict(*fields)
+	jsontransform.TransformNumbers(*fields)
 	return nil
 }
 
-// transformNumbersDict walks a json decoded tree an replaces json.Number
-// with int64, float64, or string, in this order of preference (i.e. if it
-// parses as an int, use int. if it parses as a float, use float. etc).
-func transformNumbersDict(dict common.MapStr) {
-	for k, v := range dict {
-		switch vv := v.(type) {
-		case json.Number:
-			dict[k] = transformNumber(vv)
-		case map[string]interface{}:
-			transformNumbersDict(vv)
-		case []interface{}:
-			transformNumbersArray(vv)
-		}
-	}
-}
-
-func transformNumber(value json.Number) interface{} {
-	i64, err := value.Int64()
-	if err == nil {
-		return i64
-	}
-	f64, err := value.Float64()
-	if err == nil {
-		return f64
-	}
-	return value.String()
-}
-
-func transformNumbersArray(arr []interface{}) {
-	for i, v := range arr {
-		switch vv := v.(type) {
-		case json.Number:
-			arr[i] = transformNumber(vv)
-		case map[string]interface{}:
-			transformNumbersDict(vv)
-		case []interface{}:
-			transformNumbersArray(vv)
-		}
-	}
-}
-
 // Next decodes JSON and returns the filled Line object.
 func (r *JSON) Next() (Message, error) {
 	message, err := r.reader.Next()

filebeat/tests/system/test_json.py

@@ -261,7 +261,7 @@ def test_with_generic_filtering(self):
                 message_key="message",
                 keys_under_root=True,
                 overwrite_keys=True,
-                add_error_key=True,
+                add_error_key=True
                 ),
             processors=[{
                 "drop_fields": {
@@ -305,7 +305,7 @@ def test_with_generic_filtering_remove_headers(self):
                 message_key="message",
                 keys_under_root=True,
                 overwrite_keys=True,
-                add_error_key=True,
+                add_error_key=True
                 ),
             processors=[{
                 "drop_fields": {

filebeat/tests/system/test_registrar.py

@@ -766,6 +766,10 @@ def test_clean_inactive(self):
             lambda: self.log_contains_count("Registry file updated") > 1,
             max_timeout=15)
 
+        if os.name == "nt":
+            # On windows registry recreation can take a bit longer
+            time.sleep(1)
+
         data = self.get_registry()
         assert len(data) == 2
 
@@ -834,6 +838,10 @@ def test_clean_removed(self):
             lambda: self.log_contains_count("Registry file updated") > 1,
             max_timeout=15)
 
+        if os.name == "nt":
+            # On windows registry recration can take a bit longer
+            time.sleep(1)
+
         data = self.get_registry()
         assert len(data) == 2
 

libbeat/common/jsontransform/transform.go

@@ -0,0 +1,48 @@
+package jsontransform
+
+import (
+	"encoding/json"
+
+	"github.com/elastic/beats/libbeat/common"
+)
+
+// TransformNumbers walks a json decoded tree an replaces json.Number
+// with int64, float64, or string, in this order of preference (i.e. if it
+// parses as an int, use int. if it parses as a float, use float. etc).
+func TransformNumbers(dict common.MapStr) {
+	for k, v := range dict {
+		switch vv := v.(type) {
+		case json.Number:
+			dict[k] = transformNumber(vv)
+		case map[string]interface{}:
+			TransformNumbers(vv)
+		case []interface{}:
+			transformNumbersArray(vv)
+		}
+	}
+}
+
+func transformNumber(value json.Number) interface{} {
+	i64, err := value.Int64()
+	if err == nil {
+		return i64
+	}
+	f64, err := value.Float64()
+	if err == nil {
+		return f64
+	}
+	return value.String()
+}
+
+func transformNumbersArray(arr []interface{}) {
+	for i, v := range arr {
+		switch vv := v.(type) {
+		case json.Number:
+			arr[i] = transformNumber(vv)
+		case map[string]interface{}:
+			TransformNumbers(vv)
+		case []interface{}:
+			transformNumbersArray(vv)
+		}
+	}
+}

libbeat/processors/actions/decode_json_fields.go

@@ -0,0 +1,101 @@
+package actions
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"strings"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/common/jsontransform"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/elastic/beats/libbeat/processors"
+	"github.com/pkg/errors"
+)
+
+type decodeJSONFields struct {
+	Fields []string
+}
+
+var debug = logp.MakeDebug("filters")
+
+func init() {
+	processors.RegisterPlugin("decode_json_fields", configChecked(newDecodeJSONFields,
+		requireFields("fields"), allowedFields("fields", "when")))
+}
+
+func newDecodeJSONFields(c common.Config) (processors.Processor, error) {
+	config := struct {
+		Fields []string `config:"fields"`
+	}{}
+	err := c.Unpack(&config)
+	if err != nil {
+		logp.Warn("Error unpacking config for decode_json_fields")
+		return nil, fmt.Errorf("fail to unpack the decode_json_fields configuration: %s", err)
+	}
+
+	f := decodeJSONFields{Fields: config.Fields}
+	return f, nil
+}
+
+func (f decodeJSONFields) Run(event common.MapStr) (common.MapStr, error) {
+	var errs []string
+
+	for _, field := range f.Fields {
+		data, err := event.GetValue(field)
+		if err != nil && errors.Cause(err) != common.ErrKeyNotFound {
+			debug("Error trying to GetValue for field : %s in event : %v", field, event)
+			errs = append(errs, err.Error())
+			continue
+		}
+		text, ok := data.(string)
+		if ok {
+			var output map[string]interface{}
+			err := unmarshal([]byte(text), &output)
+			if err != nil {
+				debug("Error trying to unmarshal %s", event[field])
+				errs = append(errs, err.Error())
+				continue
+			}
+
+			_, err = event.Put(field, output)
+			if err != nil {
+				debug("Error trying to Put value %v for field : %s", output, field)
+				errs = append(errs, err.Error())
+				continue
+			}
+		}
+	}
+
+	return event, fmt.Errorf(strings.Join(errs, ", "))
+}
+
+// unmarshal is equivalent with json.Unmarshal but it converts numbers
+// to int64 where possible, instead of using always float64.
+func unmarshal(text []byte, fields *map[string]interface{}) error {
+	dec := json.NewDecoder(bytes.NewReader(text))
+	dec.UseNumber()
+	err := dec.Decode(fields)
+	if err != nil {
+		return err
+	}
+
+	//Iterate through all the fields to perform deep parsing
+	for k, v := range *fields {
+		switch vv := v.(type) {
+		case string:
+			var output map[string]interface{}
+			sErr := unmarshal([]byte(vv), &output)
+			if sErr == nil {
+				(*fields)[k] = output
+			}
+		}
+	}
+
+	jsontransform.TransformNumbers(*fields)
+	return nil
+}
+
+func (f decodeJSONFields) String() string {
+	return "decode_json_fields=" + strings.Join(f.Fields, ", ")
+}

libbeat/processors/actions/decode_json_fields_test.go

@@ -0,0 +1,100 @@
+package actions
+
+import (
+	"testing"
+
+	"github.com/elastic/beats/libbeat/common"
+	"github.com/elastic/beats/libbeat/logp"
+	"github.com/stretchr/testify/assert"
+)
+
+var fields = [1]string{"msg"}
+var config, _ = common.NewConfigFrom(map[string]interface{}{
+	"fields": fields,
+})
+
+func TestMissingKey(t *testing.T) {
+	input := common.MapStr{
+		"pipeline": "us1",
+	}
+
+	actual := getActualValue(t, config, input)
+
+	expected := common.MapStr{
+		"pipeline": "us1",
+	}
+
+	assert.Equal(t, expected.String(), actual.String())
+}
+
+func TestFieldNotString(t *testing.T) {
+	input := common.MapStr{
+		"msg":      123,
+		"pipeline": "us1",
+	}
+
+	actual := getActualValue(t, config, input)
+
+	expected := common.MapStr{
+		"msg":      123,
+		"pipeline": "us1",
+	}
+
+	assert.Equal(t, expected.String(), actual.String())
+
+}
+
+func TestInvalidJSON(t *testing.T) {
+	input := common.MapStr{
+		"msg":      "{\"log\":\"{\\\"level\\\":\\\"info\\\"}\",\"stream\":\"stderr\",\"count\":3",
+		"pipeline": "us1",
+	}
+
+	actual := getActualValue(t, config, input)
+
+	expected := common.MapStr{
+		"msg":      "{\"log\":\"{\\\"level\\\":\\\"info\\\"}\",\"stream\":\"stderr\",\"count\":3",
+		"pipeline": "us1",
+	}
+	assert.Equal(t, expected.String(), actual.String())
+
+}
+
+func TestValidJSON(t *testing.T) {
+	input := common.MapStr{
+		"msg":      "{\"log\":\"{\\\"level\\\":\\\"info\\\"}\",\"stream\":\"stderr\",\"count\":3}",
+		"pipeline": "us1",
+	}
+
+	actual := getActualValue(t, config, input)
+
+	expected := common.MapStr{
+		"msg": map[string]interface{}{
+			"log": map[string]interface{}{
+				"level": "info",
+			},
+			"stream": "stderr",
+			"count":  3,
+		},
+		"pipeline": "us1",
+	}
+
+	assert.Equal(t, expected.String(), actual.String())
+
+}
+
+func getActualValue(t *testing.T, config *common.Config, input common.MapStr) common.MapStr {
+	if testing.Verbose() {
+		logp.LogInit(logp.LOG_DEBUG, "", false, true, []string{"*"})
+	}
+
+	p, err := newDecodeJSONFields(*config)
+	if err != nil {
+		logp.Err("Error initializing decode_json_fields")
+		t.Fatal(err)
+	}
+
+	actual, err := p.Run(input)
+
+	return actual
+}