Cherry-pick #14407 to 7.x: Libbeat: Do not overwrite agent.*, ecs.version, and host.name

CHANGELOG.next.asciidoc

@@ -26,6 +26,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Allow Metricbeat's beat module to read monitoring information over a named pipe or unix domain socket. {pull}14558[14558]
 - Remove version information from default ILM policy for improved upgrade experience on custom policies. {pull}14745[14745]
 - Running `setup` cmd respects `setup.ilm.overwrite` setting for improved support of custom policies. {pull}14741[14741]
+- Libbeat: Do not overwrite agent.*, ecs.version, and host.name. {pull}14407[14407]
 
 *Auditbeat*
 

libbeat/common/mapstr.go

@@ -64,33 +64,51 @@ func (m MapStr) Update(d MapStr) {
 // DeepUpdate recursively copies the key-value pairs from d to this map.
 // If the key is present and a map as well, the sub-map will be updated recursively
 // via DeepUpdate.
+// DeepUpdateNoOverwrite is a version of this function that does not
+// overwrite existing values.
 func (m MapStr) DeepUpdate(d MapStr) {
+	m.deepUpdateMap(d, true)
+}
+
+// DeepUpdateNoOverwrite recursively copies the key-value pairs from d to this map.
+// If a key is already present it will not be overwritten.
+// DeepUpdate is a version of this function that overwrites existing values.
+func (m MapStr) DeepUpdateNoOverwrite(d MapStr) {
+	m.deepUpdateMap(d, false)
+}
+
+func (m MapStr) deepUpdateMap(d MapStr, overwrite bool) {
 	for k, v := range d {
 		switch val := v.(type) {
 		case map[string]interface{}:
-			m[k] = deepUpdateValue(m[k], MapStr(val))
+			m[k] = deepUpdateValue(m[k], MapStr(val), overwrite)
 		case MapStr:
-			m[k] = deepUpdateValue(m[k], val)
+			m[k] = deepUpdateValue(m[k], val, overwrite)
 		default:
-			m[k] = v
+			if overwrite {
+				m[k] = v
+			} else if _, exists := m[k]; !exists {
+				m[k] = v
+			}
 		}
 	}
 }
 
-func deepUpdateValue(old interface{}, val MapStr) interface{} {
+func deepUpdateValue(old interface{}, val MapStr, overwrite bool) interface{} {
 	if old == nil {
 		return val
 	}
 
 	switch sub := old.(type) {
 	case MapStr:
-		sub.DeepUpdate(val)
+		sub.deepUpdateMap(val, overwrite)
 		return sub
 	case map[string]interface{}:
 		tmp := MapStr(sub)
-		tmp.DeepUpdate(val)
+		tmp.deepUpdateMap(val, overwrite)
 		return tmp
 	default:
+		// This should never happen
 		return val
 	}
 }

libbeat/processors/actions/add_fields.go

@@ -29,8 +29,9 @@ import (
 )
 
 type addFields struct {
-	fields common.MapStr
-	shared bool
+	fields    common.MapStr
+	shared    bool
+	overwrite bool
 }
 
 // FieldsKey is the default target key for the add_fields processor.
@@ -66,8 +67,8 @@ func CreateAddFields(c *common.Config) (processors.Processor, error) {
 // NewAddFields creates a new processor adding the given fields to events.
 // Set `shared` true if there is the chance of labels being changed/modified by
 // subsequent processors.
-func NewAddFields(fields common.MapStr, shared bool) processors.Processor {
-	return &addFields{fields: fields, shared: shared}
+func NewAddFields(fields common.MapStr, shared bool, overwrite bool) processors.Processor {
+	return &addFields{fields: fields, shared: shared, overwrite: overwrite}
 }
 
 func (af *addFields) Run(event *beat.Event) (*beat.Event, error) {
@@ -76,7 +77,12 @@ func (af *addFields) Run(event *beat.Event) (*beat.Event, error) {
 		fields = fields.Clone()
 	}
 
-	event.Fields.DeepUpdate(fields)
+	if af.overwrite {
+		event.Fields.DeepUpdate(fields)
+	} else {
+		event.Fields.DeepUpdateNoOverwrite(fields)
+	}
+
 	return event, nil
 }
 
@@ -99,5 +105,5 @@ func makeFieldsProcessor(target string, fields common.MapStr, shared bool) proce
 		}
 	}
 
-	return NewAddFields(fields, shared)
+	return NewAddFields(fields, shared, true)
 }

libbeat/processors/actions/add_labels.go

@@ -56,5 +56,5 @@ func createAddLabels(c *common.Config) (processors.Processor, error) {
 func NewAddLabels(labels common.MapStr, shared bool) processors.Processor {
 	return NewAddFields(common.MapStr{
 		LabelsKey: labels.Flatten(),
-	}, shared)
+	}, shared, true)
 }

libbeat/publisher/processing/default.go

@@ -295,7 +295,7 @@ func (b *builder) Create(cfg beat.ProcessingConfig, drop bool) (beat.Processor,
 		// With dynamic fields potentially changing at any time, we need to copy,
 		// so we do not change shared structures be accident.
 		fieldsNeedsCopy := needsCopy || cfg.DynamicFields != nil || hasKeyAnyOf(fields, builtin)
-		processors.add(actions.NewAddFields(fields, fieldsNeedsCopy))
+		processors.add(actions.NewAddFields(fields, fieldsNeedsCopy, true))
 	}
 
 	if cfg.DynamicFields != nil {
@@ -310,7 +310,7 @@ func (b *builder) Create(cfg beat.ProcessingConfig, drop bool) (beat.Processor,
 
 	// setup 6: add beats and host metadata
 	if meta := builtin; len(meta) > 0 {
-		processors.add(actions.NewAddFields(meta, needsCopy))
+		processors.add(actions.NewAddFields(meta, needsCopy, false))
 	}
 
 	// setup 8: pipeline processors list

libbeat/publisher/processing/default_test.go

@@ -165,7 +165,7 @@ func TestProcessorsConfigs(t *testing.T) {
 			local: beat.ProcessingConfig{
 				Processor: func() beat.ProcessorList {
 					g := newGroup("test", logp.L())
-					g.add(actions.NewAddFields(common.MapStr{"custom": "value"}, true))
+					g.add(actions.NewAddFields(common.MapStr{"custom": "value"}, true, true))
 					return g
 				}(),
 			},