[7.7] Deep merge event fields and metadata maps (#17958)

CHANGELOG.next.asciidoc

@@ -79,6 +79,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Do not rotate log files on startup when interval is configured and rotateonstartup is disabled. {pull}17613[17613]
 - Fix `setup.dashboards.index` setting not working. {pull}17749[17749]
 - Fix Elasticsearch license endpoint URL referenced in error message. {issue}17880[17880] {pull}18030[18030]
+- Change `decode_json_fields` processor, to merge parsed json objects with existing objects in the event instead of fully replacing them. {pull}17958[17958]
 
 *Auditbeat*
 
@@ -192,6 +193,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add `providers` setting to `add_cloud_metadata` processor. {pull}13812[13812]
 - Ensure that init containers are no longer tailed after they stop {pull}14394[14394]
 - Fingerprint processor adds a new xxhash hashing algorithm {pull}15418[15418]
+- When using the `decode_json_fields` processor, decoded fields are now deep-merged into existing event. {pull}17958[17958]
 
 *Filebeat*
 
@@ -250,6 +252,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Improve AWS cloudtrail field mappings {issue}16086[16086] {issue}16110[16110] {pull}17155[17155]
 - Release Google Cloud module as GA. {pull}17511[17511]
 - Update filebeat httpjson input to support pagination via Header and Okta module. {pull}16354[16354]
+- Change the `json.*` input settings implementation to merge parsed json objects with existing objects in the event instead of fully replacing them. {pull}17958[17958]
 
 *Heartbeat*
 

libbeat/common/jsontransform/jsonhelper.go

@@ -29,11 +29,12 @@ import (
 // WriteJSONKeys writes the json keys to the given event based on the overwriteKeys option and the addErrKey
 func WriteJSONKeys(event *beat.Event, keys map[string]interface{}, overwriteKeys bool, addErrKey bool) {
  if !overwriteKeys {
- for k, v := range keys {
- if _, exists := event.Fields[k]; !exists && k != "@timestamp" && k != "@metadata" {
- event.Fields[k] = v
- }
- }
+ // @timestamp and @metadata fields are root-level fields. We remove them so they
+ // don't become part of event.Fields.
+ removeKeys(keys, "@timestamp", "@metadata")
+
+ // Then, perform deep update without overwriting
+ event.Fields.DeepUpdateNoOverwrite(keys)
  return
  }
 
@@ -64,7 +65,7 @@ func WriteJSONKeys(event *beat.Event, keys map[string]interface{}, overwriteKeys
  }
 
  case map[string]interface{}:
- event.Meta.Update(common.MapStr(m))
+ event.Meta.DeepUpdate(common.MapStr(m))
 
  default:
  event.SetErrorWithOption(createJSONError("failed to update @metadata"), addErrKey)
@@ -83,13 +84,21 @@ func WriteJSONKeys(event *beat.Event, keys map[string]interface{}, overwriteKeys
  continue
  }
  event.Fields[k] = vstr
-
- default:
- event.Fields[k] = v
  }
  }
+
+ // We have accounted for @timestamp, @metadata, type above. So let's remove these keys and
+ // deep update the event with the rest of the keys.
+ removeKeys(keys, "@timestamp", "@metadata", "type")
+ event.Fields.DeepUpdate(keys)
 }
 
 func createJSONError(message string) common.MapStr {
  return common.MapStr{"message": message, "type": "json"}
 }
+
+func removeKeys(keys map[string]interface{}, names ...string) {
+ for _, name := range names {
+ delete(keys, name)
+ }
+}

libbeat/common/jsontransform/jsonhelper_test.go

@@ -0,0 +1,136 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package jsontransform
+
+import (
+ "testing"
+ "time"
+
+ "github.com/stretchr/testify/require"
+
+ "github.com/elastic/beats/v7/libbeat/beat"
+ "github.com/elastic/beats/v7/libbeat/common"
+)
+
+func TestWriteJSONKeys(t *testing.T) {
+ now := time.Now()
+ now = now.Round(time.Second)
+
+ eventTimestamp := time.Date(2020, 01, 01, 01, 01, 00, 0, time.UTC)
+ eventMetadata := common.MapStr{
+ "foo": "bar",
+ "baz": common.MapStr{
+ "qux": 17,
+ },
+ }
+ eventFields := common.MapStr{
+ "top_a": 23,
+ "top_b": common.MapStr{
+ "inner_c": "see",
+ "inner_d": "dee",
+ },
+ }
+
+ tests := map[string]struct {
+ keys map[string]interface{}
+ overwriteKeys bool
+ expectedMetadata common.MapStr
+ expectedTimestamp time.Time
+ expectedFields common.MapStr
+ }{
+ "overwrite_true": {
+ overwriteKeys: true,
+ keys: map[string]interface{}{
+ "@metadata": map[string]interface{}{
+ "foo": "NEW_bar",
+ "baz": map[string]interface{}{
+ "qux": "NEW_qux",
+ "durrr": "COMPLETELY_NEW",
+ },
+ },
+ "@timestamp": now.Format(time.RFC3339),
+ "top_b": map[string]interface{}{
+ "inner_d": "NEW_dee",
+ "inner_e": "COMPLETELY_NEW_e",
+ },
+ "top_c": "COMPLETELY_NEW_c",
+ },
+ expectedMetadata: common.MapStr{
+ "foo": "NEW_bar",
+ "baz": common.MapStr{
+ "qux": "NEW_qux",
+ "durrr": "COMPLETELY_NEW",
+ },
+ },
+ expectedTimestamp: now,
+ expectedFields: common.MapStr{
+ "top_a": 23,
+ "top_b": common.MapStr{
+ "inner_c": "see",
+ "inner_d": "NEW_dee",
+ "inner_e": "COMPLETELY_NEW_e",
+ },
+ "top_c": "COMPLETELY_NEW_c",
+ },
+ },
+ "overwrite_false": {
+ overwriteKeys: false,
+ keys: map[string]interface{}{
+ "@metadata": map[string]interface{}{
+ "foo": "NEW_bar",
+ "baz": map[string]interface{}{
+ "qux": "NEW_qux",
+ "durrr": "COMPLETELY_NEW",
+ },
+ },
+ "@timestamp": now.Format(time.RFC3339),
+ "top_b": map[string]interface{}{
+ "inner_d": "NEW_dee",
+ "inner_e": "COMPLETELY_NEW_e",
+ },
+ "top_c": "COMPLETELY_NEW_c",
+ },
+ expectedMetadata: eventMetadata.Clone(),
+ expectedTimestamp: eventTimestamp,
+ expectedFields: common.MapStr{
+ "top_a": 23,
+ "top_b": common.MapStr{
+ "inner_c": "see",
+ "inner_d": "dee",
+ "inner_e": "COMPLETELY_NEW_e",
+ },
+ "top_c": "COMPLETELY_NEW_c",
+ },
+ },
+ }
+
+ for name, test := range tests {
+ t.Run(name, func(t *testing.T) {
+ event := &beat.Event{
+ Timestamp: eventTimestamp,
+ Meta: eventMetadata.Clone(),
+ Fields: eventFields.Clone(),
+ }
+
+ WriteJSONKeys(event, test.keys, test.overwriteKeys, false)
+ require.Equal(t, test.expectedMetadata, event.Meta)
+ require.Equal(t, test.expectedTimestamp.UnixNano(), event.Timestamp.UnixNano())
+ require.Equal(t, test.expectedFields, event.Fields)
+ })
+ }
+}