Backport: Fix panic if user sets custom fields starting with host (#1…

…0935)

* Backport: Fix panic if user sets custom fields starting with host

The root cause has been fixed by #10801 by accident already.  This
selectively backports the checks to 6.7, as #10801 is to much of a
change.

CHANGELOG.next.asciidoc

@@ -16,6 +16,7 @@ https://github.com/elastic/beats/compare/v6.6.0...6.x[Check the HEAD diff]
 - Port settings have been deprecated in redis/logstash output and will be removed in 7.0. {pull}9915[9915]
 - Update the code of Central Management to align with the new returned format. {pull}10019[10019]
 - Allow Central Management to send events back to kibana. {issue}9382[9382]
+- Fix panic if fields settting is used to configure `hosts.x` fields. {issue}10824[10824] {pull}10935[10935]
 
 *Auditbeat*
 

libbeat/publisher/pipeline/processor.go

@@ -71,6 +71,7 @@ func newProcessorPipeline(
 	)
 
 	needsCopy := global.alwaysCopy || localProcessors != nil || global.processors != nil
+	builtin := global.builtinMeta
 
 	if !config.SkipNormalization {
 		// setup 1: generalize/normalize output (P)
@@ -102,13 +103,13 @@ func newProcessorPipeline(
 		// metadata will be merged into the fields.
 		// With dynamic fields potentially changing at any time, we need to copy,
 		// so we do not change shared structures be accident.
-		fieldsNeedsCopy := needsCopy || config.DynamicFields != nil || fields["beat"] != nil
+		fieldsNeedsCopy := needsCopy || config.DynamicFields != nil || hasKeyAnyOf(fields, builtin)
 		processors.add(makeAddFieldsProcessor("fields", fields, fieldsNeedsCopy))
 	}
 
 	if config.DynamicFields != nil {
 		checkCopy := func(m common.MapStr) bool {
-			return needsCopy || hasKey(m, "beat")
+			return needsCopy || hasKeyAnyOf(m, builtin)
 		}
 		processors.add(makeAddDynMetaProcessor("dynamicFields", config.DynamicFields, checkCopy))
 	}
@@ -117,8 +118,8 @@ func newProcessorPipeline(
 	processors.add(localProcessors)
 
 	// setup 6: add beats and host metadata
-	if meta := global.builtinMeta; len(meta) > 0 {
-		processors.add(makeAddFieldsProcessor("beatsMeta", meta, needsCopy))
+	if len(builtin) > 0 {
+		processors.add(makeAddFieldsProcessor("beatsMeta", builtin, needsCopy))
 	}
 
 	// setup 7: pipeline processors list
@@ -326,3 +327,12 @@ func hasKey(m common.MapStr, key string) bool {
 	_, exists := m[key]
 	return exists
 }
+
+func hasKeyAnyOf(m, builtin common.MapStr) bool {
+	for k := range builtin {
+		if hasKey(m, k) {
+			return true
+		}
+	}
+	return false
+}