[Elastic Agent] Fix fleet-server.yml spec to not overwrite existing keys

x-pack/elastic-agent/CHANGELOG.next.asciidoc

@@ -66,6 +66,7 @@
 - Fix AckBatch to do nothing when no actions passed {pull}25562[25562]
 - Add error log entry when listener creation fails {issue}23483[23482]
 - Handle case where policy doesn't contain Fleet connection information {pull}25707[25707]
+- Fix fleet-server.yml spec to not overwrite existing keys {pull}25741[25741]
 
 ==== New features
 

x-pack/elastic-agent/pkg/agent/program/testdata/fleet_server-fleet-server.yml

@@ -14,3 +14,18 @@ output:
 inputs:
   - id: fleet-server-id
     type: fleet-server
+    name: fleet_server-1
+    revision: 6
+    server:
+      host: 0.0.0.0
+      port: 8220
+      ssl:
+        verification_mode: none
+      limits:
+        max_connections: 40
+      runtime:
+        gc_percent: 50
+      timeouts:
+        read: 5m
+    policy:
+      id: copy-policy-id

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

@@ -11,6 +11,12 @@ fleet:
   host:
     id: host-agent-id
   server:
+    host: 127.0.0.1
+    port: 8822
+    ssl:
+      verification_mode: none
+    policy:
+      id: copy-policy-id
     output:
       elasticsearch:
         hosts: [ 127.0.0.1:9200, 127.0.0.1:9300 ]
@@ -31,4 +37,15 @@ inputs:
     type: fleet-server
     use_output: default
     data_stream:
-      type: default
+      namespace: default
+    name: fleet_server-1
+    revision: 6
+    server:
+      host: 0.0.0.0
+      port: 8220
+      limits:
+        max_connections: 40
+      runtime:
+        gc_percent: 50
+      timeouts:
+        read: 5m

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

@@ -88,6 +88,8 @@ func (r *RuleList) MarshalYAML() (interface{}, error) {
 			name = "remove_key"
 		case *FixStreamRule:
 			name = "fix_stream"
+		case *InsertDefaultsRule:
+			name = "insert_defaults"
 		default:
 			return nil, fmt.Errorf("unknown rule of type %T", rule)
 		}
@@ -171,6 +173,8 @@ func (r *RuleList) UnmarshalYAML(unmarshal func(interface{}) error) error {
 			r = &RemoveKeyRule{}
 		case "fix_stream":
 			r = &FixStreamRule{}
+		case "insert_defaults":
+			r = &InsertDefaultsRule{}
 		default:
 			return fmt.Errorf("unknown rule of type %s", name)
 		}
@@ -1430,6 +1434,77 @@ func (r *FilterValuesWithRegexpRule) Apply(_ AgentInfo, ast *AST) (err error) {
 	return nil
 }
 
+// InsertDefaultsRule inserts selected paths into keys if they do not exist.
+//
+// In the case that an exiting key already exists then it is not inserted.
+type InsertDefaultsRule struct {
+	Selectors []Selector
+	Path      string
+}
+
+// Apply applies select into rule.
+func (r *InsertDefaultsRule) Apply(_ AgentInfo, ast *AST) (err error) {
+	defer func() {
+		if err != nil {
+			err = errors.New(err, "failed to select data into configuration")
+		}
+	}()
+
+	insertTo := ast.root
+	for _, part := range splitPath(r.Path) {
+		n, ok := insertTo.Find(part)
+		if !ok {
+			insertTo = nil
+			break
+		}
+		insertTo = n
+	}
+
+	// path completely missing; easy path is just to insert all selectors
+	if insertTo == nil {
+		target := &Dict{}
+		for _, selector := range r.Selectors {
+			lookupNode, ok := Lookup(ast.Clone(), selector)
+			if !ok {
+				continue
+			}
+			target.value = append(target.value, lookupNode.Clone())
+		}
+		if len(target.value) > 0 {
+			return Insert(ast, target, r.Path)
+		}
+		return nil
+	}
+
+	// path does exist, so we insert the keys only if they don't exist
+	for _, selector := range r.Selectors {
+		lookupNode, ok := Lookup(ast.Clone(), selector)
+		if !ok {
+			continue
+		}
+		switch lt := lookupNode.(type) {
+		case *Key:
+			_, ok := insertTo.Find(lt.name)
+			if !ok {
+				// doesn't exist; insert it
+				if err := Insert(ast, lt, r.Path); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+// InsertDefaults creates a InsertDefaultsRule
+func InsertDefaults(path string, selectors ...Selector) *InsertDefaultsRule {
+	return &InsertDefaultsRule{
+		Selectors: selectors,
+		Path:      path,
+	}
+}
+
 // NewRuleList returns a new list of rules to be executed.
 func NewRuleList(rules ...Rule) *RuleList {
 	return &RuleList{Rules: rules}

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

@@ -653,6 +653,66 @@ logs:
 				},
 			},
 		},
+		"insert defaults into existing": {
+			givenYAML: `
+level_one:
+  key1: val1
+  key2:
+    d_key1: val2
+    d_key2: val3
+  level_two:
+    key2:
+      d_key3: val3
+      d_key4: val4
+rest: of
+`,
+			expectedYAML: `
+level_one:
+  key1: val1
+  key2:
+    d_key1: val2
+    d_key2: val3
+  level_two:
+    key1: val1
+    key2:
+      d_key3: val3
+      d_key4: val4
+rest: of
+`,
+			rule: &RuleList{
+				Rules: []Rule{
+					InsertDefaults("level_one.level_two", "level_one.key1", "level_one.key2"),
+				},
+			},
+		},
+		"insert defaults into not existing": {
+			givenYAML: `
+level_one:
+  key1: val1
+  key2:
+    d_key1: val2
+    d_key2: val3
+rest: of
+`,
+			expectedYAML: `
+level_one:
+  key1: val1
+  key2:
+    d_key1: val2
+    d_key2: val3
+  level_two:
+    key1: val1
+    key2:
+      d_key1: val2
+      d_key2: val3
+rest: of
+`,
+			rule: &RuleList{
+				Rules: []Rule{
+					InsertDefaults("level_one.level_two", "level_one.key1", "level_one.key2"),
+				},
+			},
+		},
 	}
 
 	for name, test := range testcases {
@@ -716,6 +776,7 @@ func TestSerialization(t *testing.T) {
 		CopyAllToList("t2", "insert_before", "a", "b"),
 		FixStream(),
 		SelectInto("target", "s1", "s2"),
+		InsertDefaults("target", "s1", "s2"),
 	)
 
 	y := `- rename:
@@ -781,6 +842,11 @@ func TestSerialization(t *testing.T) {
     - s1
     - s2
     path: target
+- insert_defauls:
+    selectors:
+    - s1
+    - s2
+    path: target
 `
 
 	t.Run("serialize_rules", func(t *testing.T) {

x-pack/elastic-agent/spec/fleet-server.yml

@@ -28,7 +28,7 @@ rules:
       selectors: [ fleet.server.policy.id ]
       path: inputs.0.policy
 
-  - select_into:
+  - insert_defaults:
       selectors:
         - fleet.server.host
         - fleet.server.port