[Agent] Fix merge of config (elastic#17399)

[Agent] Fix merge of config elastic#17399

x-pack/elastic-agent/pkg/agent/program/testdata/single_config-filebeat.yml

@@ -5,6 +5,8 @@ filebeat:
       - /var/log/hello1.log
       - /var/log/hello2.log
     index: logs-generic-default
+    vars:
+      var: value
 output:
   elasticsearch:
     hosts:

x-pack/elastic-agent/pkg/agent/program/testdata/single_config-metricbeat.yml

@@ -3,9 +3,11 @@ metricbeat:
   - module: docker
     metricsets: [status]
     index: metrics-docker.status-default
+    hosts: ["http://127.0.0.1:8080"]
   - module: apache
     metricsets: [info]
     index: metrics-generic-testing
+    hosts: ["http://apache.remote"]
 output:
   elasticsearch:
     hosts: [127.0.0.1:9200, 127.0.0.1:9300]

x-pack/elastic-agent/pkg/agent/program/testdata/single_config.yml

@@ -26,18 +26,24 @@ datasources:
       streams:
         - metricset: status
           dataset: docker.status
+      hosts: ["http://127.0.0.1:8080"]
     - type: logs
       streams:
         - paths:
           - /var/log/hello1.log
           - /var/log/hello2.log
+      vars:
+        var: value
   - namespace: testing
     use_output: default
     inputs:
       - type: apache/metrics
         streams:
           - enabled: true
             metricset: info
+            hosts: ["http://apache.remote"]
+        hosts: ["http://apache.local"]
+        id: apache-metrics-id
 
 settings.monitoring:
   use_output: monitoring

x-pack/elastic-agent/pkg/agent/transpiler/rules.go

@@ -49,6 +49,8 @@ func (r *RuleList) MarshalYAML() (interface{}, error) {
 			name = "copy"
 		case *CopyToListRule:
 			name = "copy_to_list"
+		case *CopyAllToListRule:
+			name = "copy_all_to_list"
 		case *RenameRule:
 			name = "rename"
 		case *TranslateRule:
@@ -123,6 +125,8 @@ func (r *RuleList) UnmarshalYAML(unmarshal func(interface{}) error) error {
 			r = &CopyRule{}
 		case "copy_to_list":
 			r = &CopyToListRule{}
+		case "copy_all_to_list":
+			r = &CopyAllToListRule{}
 		case "rename":
 			r = &RenameRule{}
 		case "translate":
@@ -231,8 +235,9 @@ func MakeArray(item Selector, to string) *MakeArrayRule {
 // CopyToListRule is a rule which copies a specified
 // node into every item in a provided list.
 type CopyToListRule struct {
-	Item Selector
-	To   string
+	Item      Selector
+	To        string
+	Overwrite bool
 }
 
 // Apply copies specified node into every item of the list.
@@ -260,17 +265,74 @@ func (r *CopyToListRule) Apply(ast *AST) error {
 			continue
 		}
 
+		if !r.Overwrite {
+			_, found := listItemMap.Find(r.Item)
+			if found {
+				continue
+			}
+		}
+
 		listItemMap.value = append(listItemMap.value, sourceNode.Clone())
 	}
 
 	return nil
 }
 
 // CopyToList creates a CopyToListRule
-func CopyToList(item Selector, to string) *CopyToListRule {
+func CopyToList(item Selector, to string, overwrite bool) *CopyToListRule {
 	return &CopyToListRule{
-		Item: item,
-		To:   to,
+		Item:      item,
+		To:        to,
+		Overwrite: overwrite,
+	}
+}
+
+// CopyAllToListRule is a rule which copies a all nodes
+// into every item in a provided list.
+type CopyAllToListRule struct {
+	To        string
+	Except    []string
+	Overwrite bool
+}
+
+// Apply copies all nodes into every item of the list.
+func (r *CopyAllToListRule) Apply(ast *AST) error {
+	// get list of nodes
+	astMap, err := ast.Map()
+	if err != nil {
+		return err
+	}
+
+	isFiltered := func(item string) bool {
+		for _, f := range r.Except {
+			if f == item {
+				return true
+			}
+		}
+
+		return false
+	}
+
+	// foreach node if not filtered out
+	for item := range astMap {
+		if isFiltered(item) {
+			continue
+		}
+
+		if err := CopyToList(item, r.To, r.Overwrite).Apply(ast); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// CopyAllToList creates a CopyAllToListRule
+func CopyAllToList(to string, overwrite bool, except ...string) *CopyAllToListRule {
+	return &CopyAllToListRule{
+		To:        to,
+		Except:    except,
+		Overwrite: overwrite,
 	}
 }
 

x-pack/elastic-agent/pkg/agent/transpiler/rules_test.go

@@ -481,7 +481,7 @@ inputs:
 `,
 			rule: &RuleList{
 				Rules: []Rule{
-					CopyToList("namespace", "inputs"),
+					CopyToList("namespace", "inputs", false),
 				},
 			},
 		},
@@ -561,7 +561,8 @@ func TestSerialization(t *testing.T) {
 		FilterValuesWithRegexp("inputs", "type", regexp.MustCompile("^metric/.*")),
 		ExtractListItem("path.p", "item", "target"),
 		InjectIndex("index-type"),
-		CopyToList("t1", "t2"),
+		CopyToList("t1", "t2", false),
+		CopyAllToList("t2", false, "a", "b"),
 	)
 
 	y := `- rename:
@@ -611,6 +612,13 @@ func TestSerialization(t *testing.T) {
 - copy_to_list:
     item: t1
     to: t2
+    overwrite: false
+- copy_all_to_list:
+    to: t2
+    except:
+    - a
+    - b
+    overwrite: false
 `
 
 	t.Run("serialize_rules", func(t *testing.T) {

x-pack/elastic-agent/spec/filebeat.yml

@@ -14,12 +14,10 @@ rules:
 - map:
     path: inputsstreams
     rules:
-    - copy_to_list:
-        item: type
-        to: streams
-    - copy_to_list:
-        item: processors
+    - copy_all_to_list:
         to: streams
+        overwrite: false
+        except: ["streams", "id", "enabled", "title", "package", "namespace", "constraints", "use_output"]
 
 - extract_list_items:
     path: inputsstreams

x-pack/elastic-agent/spec/metricbeat.yml

@@ -18,12 +18,10 @@ rules:
 - map:
     path: inputsstreams
     rules:
-    - copy_to_list:
-        item: type
-        to: streams
-    - copy_to_list:
-        item: processors
+    - copy_all_to_list:
         to: streams
+        overwrite: false
+        except: ["streams", "id", "enabled"]
 
 - extract_list_items:
     path: inputsstreams