Cherry-pick #18769 to 7.8: Guard against empty stream.datasource and namespace (#18806)

* [Ingest Manager] Guard against empty stream.datasource and namespace (#18769)

Author: michalpristas

x-pack/elastic-agent/CHANGELOG.asciidoc

@@ -40,6 +40,7 @@
 - Clean action store after enrolling to new configuration {pull}18656[18656]
 - Avoid watching monitor logs {pull}18723[18723]
 - Correctly report platform and family. {issue}18665[18665]
+- Guard against empty stream.datasource and namespace {pull}18769[18769]
 
 ==== New features
 

x-pack/elastic-agent/pkg/agent/program/supported.go

@@ -4,6 +4,7 @@ filebeat:
     paths:
       - /var/log/hello1.log
       - /var/log/hello2.log
+    dataset: generic
     index: logs-generic-default
     processors:
       - add_fields:

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

@@ -4,6 +4,7 @@ filebeat:
     paths:
       - /var/log/hello1.log
       - /var/log/hello2.log
+    dataset: generic
     index: logs-generic-default
     processors:
       - add_fields:

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

@@ -5,6 +5,7 @@ filebeat:
     paths:
       - /var/log/hello1.log
       - /var/log/hello2.log
+    dataset: generic
     index: logs-generic-default
     processors:
       - add_fields:

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

@@ -5,6 +5,7 @@ filebeat:
       - /var/log/hello1.log
       - /var/log/hello2.log
     index: logs-generic-default
+    dataset: generic
     vars:
       var: value
     processors:

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

@@ -11,6 +11,17 @@ metricbeat:
           type: metrics
           dataset: docker.status
           namespace: default
+  - module: docker
+    metricsets: [info]
+    index: metrics-generic-default
+    hosts: ["http://127.0.0.1:8080"]
+    processors:
+    - add_fields:
+        target: "stream"
+        fields:
+          type: metrics
+          dataset: generic
+          namespace: default
   - module: apache
     metricsets: [info]
     index: metrics-generic-testing

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

@@ -26,6 +26,8 @@ datasources:
       streams:
         - metricset: status
           dataset: docker.status
+        - metricset: info
+          dataset: ""
       hosts: ["http://127.0.0.1:8080"]
     - type: logs
       streams:

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

@@ -75,7 +75,8 @@ func (r *RuleList) MarshalYAML() (interface{}, error) {
 			name = "make_array"
 		case *RemoveKeyRule:
 			name = "remove_key"
-
+		case *FixStreamRule:
+			name = "fix_stream"
 		default:
 			return nil, fmt.Errorf("unknown rule of type %T", rule)
 		}
@@ -153,6 +154,8 @@ func (r *RuleList) UnmarshalYAML(unmarshal func(interface{}) error) error {
 			r = &MakeArrayRule{}
 		case "remove_key":
 			r = &RemoveKeyRule{}
+		case "fix_stream":
+			r = &FixStreamRule{}
 		default:
 			return fmt.Errorf("unknown rule of type %s", name)
 		}
@@ -345,6 +348,100 @@ func CopyAllToList(to, onMerge string, except ...string) *CopyAllToListRule {
 	}
 }
 
+// FixStreamRule fixes streams to contain default values
+// in case no value or invalid value are provided
+type FixStreamRule struct {
+}
+
+// Apply stream fixes.
+func (r *FixStreamRule) Apply(ast *AST) error {
+	const defaultNamespace = "default"
+	const defaultDataset = "generic"
+
+	datasourcesNode, found := Lookup(ast, "datasources")
+	if !found {
+		return nil
+	}
+
+	datasourcesList, ok := datasourcesNode.Value().(*List)
+	if !ok {
+		return nil
+	}
+
+	for _, datasourceNode := range datasourcesList.value {
+		nsNode, found := datasourceNode.Find("namespace")
+		if found {
+			nsKey, ok := nsNode.(*Key)
+			if ok {
+				if newNamespace := nsKey.value.String(); newNamespace == "" {
+					nsKey.value = &StrVal{value: defaultNamespace}
+				}
+			}
+		} else {
+			datasourceMap, ok := datasourceNode.(*Dict)
+			if !ok {
+				continue
+			}
+			datasourceMap.value = append(datasourceMap.value, &Key{
+				name:  "namespace",
+				value: &StrVal{value: defaultNamespace},
+			})
+		}
+
+		// get input
+		inputNode, found := datasourceNode.Find("inputs")
+		if !found {
+			continue
+		}
+
+		inputsList, ok := inputNode.Value().(*List)
+		if !ok {
+			continue
+		}
+
+		for _, inputNode := range inputsList.value {
+			streamsNode, ok := inputNode.Find("streams")
+			if !ok {
+				continue
+			}
+
+			streamsList, ok := streamsNode.Value().(*List)
+			if !ok {
+				continue
+			}
+
+			for _, streamNode := range streamsList.value {
+				streamMap, ok := streamNode.(*Dict)
+				if !ok {
+					continue
+				}
+
+				dsNode, found := streamNode.Find("dataset")
+				if found {
+					dsKey, ok := dsNode.(*Key)
+					if ok {
+						if newDataset := dsKey.value.String(); newDataset == "" {
+							dsKey.value = &StrVal{value: defaultDataset}
+						}
+					}
+				} else {
+					streamMap.value = append(streamMap.value, &Key{
+						name:  "dataset",
+						value: &StrVal{value: defaultDataset},
+					})
+				}
+			}
+		}
+	}
+
+	return nil
+}
+
+// FixStream creates a FixStreamRule
+func FixStream() *FixStreamRule {
+	return &FixStreamRule{}
+}
+
 // InjectIndexRule injects index to each input.
 // Index is in form {type}-{namespace}-{dataset-type}
 // type: is provided to the rule.
@@ -375,7 +472,9 @@ func (r *InjectIndexRule) Apply(ast *AST) error {
 		if found {
 			nsKey, ok := nsNode.(*Key)
 			if ok {
-				namespace = nsKey.value.String()
+				if newNamespace := nsKey.value.String(); newNamespace != "" {
+					namespace = newNamespace
+				}
 			}
 		}
 
@@ -413,7 +512,9 @@ func (r *InjectIndexRule) Apply(ast *AST) error {
 				if found {
 					dsKey, ok := dsNode.(*Key)
 					if ok {
-						dataset = dsKey.value.String()
+						if newDataset := dsKey.value.String(); newDataset != "" {
+							dataset = newDataset
+						}
 					}
 
 				}
@@ -464,7 +565,9 @@ func (r *InjectStreamProcessorRule) Apply(ast *AST) error {
 		if found {
 			nsKey, ok := nsNode.(*Key)
 			if ok {
-				namespace = nsKey.value.String()
+				if newNamespace := nsKey.value.String(); newNamespace != "" {
+					namespace = newNamespace
+				}
 			}
 		}
 
@@ -502,7 +605,9 @@ func (r *InjectStreamProcessorRule) Apply(ast *AST) error {
 				if found {
 					dsKey, ok := dsNode.(*Key)
 					if ok {
-						dataset = dsKey.value.String()
+						if newDataset := dsKey.value.String(); newDataset != "" {
+							dataset = newDataset
+						}
 					}
 				}
 

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

@@ -22,6 +22,86 @@ func TestRules(t *testing.T) {
 		expectedYAML string
 		rule         Rule
 	}{
+		"fix streams": {
+			givenYAML: `
+datasources:
+  - name: All default
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/error.log
+  - name: Specified namespace
+    namespace: nsns
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+  - name: Specified dataset
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: dsds
+  - name: All specified
+    namespace: nsns
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: dsds
+  - name: All specified with empty strings
+    namespace: ""
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: ""
+`,
+			expectedYAML: `
+datasources:
+  - name: All default
+    namespace: default
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/error.log
+          dataset: generic
+  - name: Specified namespace
+    namespace: nsns
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: generic
+  - name: Specified dataset
+    namespace: default
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: dsds
+  - name: All specified
+    namespace: nsns
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: dsds
+  - name: All specified with empty strings
+    namespace: default
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: generic
+`,
+			rule: &RuleList{
+				Rules: []Rule{
+					FixStream(),
+				},
+			},
+		},
+
 		"inject index": {
 			givenYAML: `
 datasources:
@@ -49,6 +129,13 @@ datasources:
       streams:
         - paths: /var/log/mysql/access.log
           dataset: dsds
+  - name: All specified with empty strings
+    namespace: ""
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: ""
 `,
 			expectedYAML: `
 datasources:
@@ -80,6 +167,14 @@ datasources:
         - paths: /var/log/mysql/access.log
           dataset: dsds
           index: mytype-dsds-nsns
+  - name: All specified with empty strings
+    namespace: ""
+    inputs:
+    - type: file
+      streams:
+        - paths: /var/log/mysql/access.log
+          dataset: ""
+          index: mytype-generic-default
 `,
 			rule: &RuleList{
 				Rules: []Rule{
@@ -564,6 +659,7 @@ func TestSerialization(t *testing.T) {
 		InjectStreamProcessor("insert_after", "index-type"),
 		CopyToList("t1", "t2", "insert_after"),
 		CopyAllToList("t2", "insert_before", "a", "b"),
+		FixStream(),
 	)
 
 	y := `- rename:
@@ -623,6 +719,7 @@ func TestSerialization(t *testing.T) {
     - a
     - b
     on_conflict: insert_before
+- fix_stream: {}
 `
 
 	t.Run("serialize_rules", func(t *testing.T) {