Cherry-pick #20402 to 7.x: Send datastreams fields

x-pack/elastic-agent/pkg/agent/operation/monitoring.go

@@ -190,10 +190,10 @@ func (o *Operator) getMonitoringFilebeatConfig(output interface{}) (map[string]i
 			"processors": []map[string]interface{}{
 				{
 					"add_fields": map[string]interface{}{
-						"target": "dataset",
+						"target": "data_stream",
 						"fields": map[string]interface{}{
 							"type":      "logs",
-							"name":      "elastic.agent",
+							"dataset":   "elastic.agent",
 							"namespace": "default",
 						},
 					},
@@ -224,10 +224,10 @@ func (o *Operator) getMonitoringFilebeatConfig(output interface{}) (map[string]i
 				"processors": []map[string]interface{}{
 					{
 						"add_fields": map[string]interface{}{
-							"target": "dataset",
+							"target": "data_stream",
 							"fields": map[string]interface{}{
 								"type":      "logs",
-								"name":      fmt.Sprintf("elastic.agent.%s", name),
+								"dataset":   fmt.Sprintf("elastic.agent.%s", name),
 								"namespace": "default",
 							},
 						},
@@ -274,10 +274,10 @@ func (o *Operator) getMonitoringMetricbeatConfig(output interface{}) (map[string
 			"processors": []map[string]interface{}{
 				{
 					"add_fields": map[string]interface{}{
-						"target": "dataset",
+						"target": "data_stream",
 						"fields": map[string]interface{}{
 							"type":      "metrics",
-							"name":      fmt.Sprintf("elastic.agent.%s", name),
+							"dataset":   fmt.Sprintf("elastic.agent.%s", name),
 							"namespace": "default",
 						},
 					},

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

@@ -7,10 +7,10 @@ filebeat:
     index: logs-generic-default
     processors:
       - add_fields:
-          target: "dataset"
+          target: "data_stream"
           fields:
             type: logs
-            name: generic
+            dataset: generic
             namespace: default
       - add_fields:
           target: "event"

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

@@ -7,10 +7,10 @@ filebeat:
     index: logs-generic-default
     processors:
       - add_fields:
-          target: "dataset"
+          target: "data_stream"
           fields:
             type: logs
-            name: generic
+            dataset: generic
             namespace: default
       - add_fields:
           target: "event"

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

@@ -8,10 +8,10 @@ filebeat:
     index: logs-generic-default
     processors:
       - add_fields:
-          target: "dataset"
+          target: "data_stream"
           fields:
             type: logs
-            name: generic
+            dataset: generic
             namespace: default
       - add_fields:
           target: "event"

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

@@ -9,10 +9,10 @@ filebeat:
       var: value
     processors:
       - add_fields:
-          target: "dataset"
+          target: "data_stream"
           fields:
             type: logs
-            name: generic
+            dataset: generic
             namespace: default
       - add_fields:
           target: "event"
@@ -27,10 +27,10 @@ filebeat:
       var: value
     processors:
       - add_fields:
-          target: "dataset"
+          target: "data_stream"
           fields:
             type: testtype
-            name: generic
+            dataset: generic
             namespace: default
       - add_fields:
           target: "event"

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

@@ -6,10 +6,10 @@ metricbeat:
     hosts: ["http://127.0.0.1:8080"]
     processors:
     - add_fields:
-        target: "dataset"
+        target: "data_stream"
         fields:
           type: metrics
-          name: docker.status
+          dataset: docker.status
           namespace: default
     - add_fields:
         target: "event"
@@ -21,10 +21,10 @@ metricbeat:
     hosts: ["http://127.0.0.1:8080"]
     processors:
     - add_fields:
-        target: "dataset"
+        target: "data_stream"
         fields:
           type: metrics
-          name: generic
+          dataset: generic
           namespace: default
     - add_fields:
         target: "event"
@@ -39,10 +39,10 @@ metricbeat:
           fields:
             should_be: first
       - add_fields:
-          target: "dataset"
+          target: "data_stream"
           fields:
             type: metrics
-            name: generic
+            dataset: generic
             namespace: testing
       - add_fields:
           target: "event"

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

@@ -632,16 +632,18 @@ func (r *InjectStreamProcessorRule) Apply(ast *AST) error {
 				return errors.New("InjectStreamProcessorRule: processors is not a list")
 			}
 
+			// datastream
 			processorMap := &Dict{value: make([]Node, 0)}
-			processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "dataset"}})
+			processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "data_stream"}})
 			processorMap.value = append(processorMap.value, &Key{name: "fields", value: &Dict{value: []Node{
 				&Key{name: "type", value: &StrVal{value: datasetType}},
 				&Key{name: "namespace", value: &StrVal{value: namespace}},
-				&Key{name: "name", value: &StrVal{value: dataset}},
+				&Key{name: "dataset", value: &StrVal{value: dataset}},
 			}}})
 			addFieldsMap := &Dict{value: []Node{&Key{"add_fields", processorMap}}}
 			processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, addFieldsMap)
 
+			// event
 			processorMap = &Dict{value: make([]Node, 0)}
 			processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "event"}})
 			processorMap.value = append(processorMap.value, &Key{name: "fields", value: &Dict{value: []Node{