diff --git a/x-pack/elastic-agent/CHANGELOG.asciidoc b/x-pack/elastic-agent/CHANGELOG.asciidoc
index c6d6c7af732..e287cc3b35b 100644
--- a/x-pack/elastic-agent/CHANGELOG.asciidoc
+++ b/x-pack/elastic-agent/CHANGELOG.asciidoc
@@ -68,3 +68,4 @@
 - Use shorter hash for application differentiator {pull}18770[18770]
 - When not port are specified and the https is used fallback to 443 {pull}18844[18844]
 - Agent verifies packages before using them {pull}18876[18876]
+- Change stream.* to dataset.* fields {pull}18967[18967]
diff --git a/x-pack/elastic-agent/pkg/agent/operation/monitoring.go b/x-pack/elastic-agent/pkg/agent/operation/monitoring.go
index de852f7b13a..1cbf29152a5 100644
--- a/x-pack/elastic-agent/pkg/agent/operation/monitoring.go
+++ b/x-pack/elastic-agent/pkg/agent/operation/monitoring.go
@@ -180,7 +180,17 @@ func (o *Operator) getMonitoringFilebeatConfig(output interface{}) (map[string]i
 					"paths": paths,
 					"index": "logs-agent-default",
 					"processors": []map[string]interface{}{
-						map[string]interface{}{
+						{
+							"add_fields": map[string]interface{}{
+								"target": "dataset",
+								"fields": map[string]interface{}{
+									"type":      "logs",
+									"name":      "agent",
+									"namespace": "default",
+								},
+							},
+						},
+						{
 							"add_fields": map[string]interface{}{
 								"target": "stream",
 								"fields": map[string]interface{}{
@@ -220,7 +230,17 @@ func (o *Operator) getMonitoringMetricbeatConfig(output interface{}) (map[string
 					"hosts":      hosts,
 					"index":      "metrics-agent-default",
 					"processors": []map[string]interface{}{
-						map[string]interface{}{
+						{
+							"add_fields": map[string]interface{}{
+								"target": "dataset",
+								"fields": map[string]interface{}{
+									"type":      "metrics",
+									"name":      "agent",
+									"namespace": "default",
+								},
+							},
+						},
+						{
 							"add_fields": map[string]interface{}{
 								"target": "stream",
 								"fields": map[string]interface{}{
diff --git a/x-pack/elastic-agent/pkg/agent/program/testdata/constraints_config-filebeat.yml b/x-pack/elastic-agent/pkg/agent/program/testdata/constraints_config-filebeat.yml
index a620d40de2e..d4a38b09c6b 100644
--- a/x-pack/elastic-agent/pkg/agent/program/testdata/constraints_config-filebeat.yml
+++ b/x-pack/elastic-agent/pkg/agent/program/testdata/constraints_config-filebeat.yml
@@ -7,6 +7,12 @@ filebeat:
     dataset: generic
     index: logs-generic-default
     processors:
+      - add_fields:
+          target: "dataset"
+          fields:
+            type: logs
+            name: generic
+            namespace: default
       - add_fields:
           target: "stream"
           fields:
diff --git a/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_output_true-filebeat.yml b/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_output_true-filebeat.yml
index 71c95920829..bc79c306334 100644
--- a/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_output_true-filebeat.yml
+++ b/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_output_true-filebeat.yml
@@ -7,6 +7,12 @@ filebeat:
     dataset: generic
     index: logs-generic-default
     processors:
+      - add_fields:
+          target: "dataset"
+          fields:
+            type: logs
+            name: generic
+            namespace: default
       - add_fields:
           target: "stream"
           fields:
diff --git a/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_true-filebeat.yml b/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_true-filebeat.yml
index 176df2f7675..9ad59d63f6f 100644
--- a/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_true-filebeat.yml
+++ b/x-pack/elastic-agent/pkg/agent/program/testdata/enabled_true-filebeat.yml
@@ -8,6 +8,12 @@ filebeat:
     dataset: generic
     index: logs-generic-default
     processors:
+      - add_fields:
+          target: "dataset"
+          fields:
+            type: logs
+            name: generic
+            namespace: default
       - add_fields:
           target: "stream"
           fields:
diff --git a/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-filebeat.yml b/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-filebeat.yml
index 442457c5dc1..59346260ca2 100644
--- a/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-filebeat.yml
+++ b/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-filebeat.yml
@@ -9,6 +9,12 @@ filebeat:
     vars:
       var: value
     processors:
+      - add_fields:
+          target: "dataset"
+          fields:
+            type: logs
+            name: generic
+            namespace: default
       - add_fields:
           target: "stream"
           fields:
diff --git a/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-metricbeat.yml b/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-metricbeat.yml
index af734fb28d8..df6ac491bb8 100644
--- a/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-metricbeat.yml
+++ b/x-pack/elastic-agent/pkg/agent/program/testdata/single_config-metricbeat.yml
@@ -5,6 +5,12 @@ metricbeat:
     index: metrics-docker.status-default
     hosts: ["http://127.0.0.1:8080"]
     processors:
+    - add_fields:
+        target: "dataset"
+        fields:
+          type: metrics
+          name: docker.status
+          namespace: default
     - add_fields:
         target: "stream"
         fields:
@@ -16,6 +22,12 @@ metricbeat:
     index: metrics-generic-default
     hosts: ["http://127.0.0.1:8080"]
     processors:
+    - add_fields:
+        target: "dataset"
+        fields:
+          type: metrics
+          name: generic
+          namespace: default
     - add_fields:
         target: "stream"
         fields:
@@ -30,12 +42,19 @@ metricbeat:
       - add_fields:
           fields:
             should_be: first
+      - add_fields:
+          target: "dataset"
+          fields:
+            type: metrics
+            name: generic
+            namespace: testing
       - add_fields:
           target: "stream"
           fields:
             type: metrics
             dataset: generic
             namespace: testing
+
 output:
   elasticsearch:
     hosts: [127.0.0.1:9200, 127.0.0.1:9300]
diff --git a/x-pack/elastic-agent/pkg/agent/transpiler/rules.go b/x-pack/elastic-agent/pkg/agent/transpiler/rules.go
index 4554912e066..2e3fb60b6fa 100644
--- a/x-pack/elastic-agent/pkg/agent/transpiler/rules.go
+++ b/x-pack/elastic-agent/pkg/agent/transpiler/rules.go
@@ -628,15 +628,29 @@ func (r *InjectStreamProcessorRule) Apply(ast *AST) error {
 				}
 
 				processorMap := &Dict{value: make([]Node, 0)}
-				processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "stream"}})
+				processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "dataset"}})
 				processorMap.value = append(processorMap.value, &Key{name: "fields", value: &Dict{value: []Node{
 					&Key{name: "type", value: &StrVal{value: r.Type}},
 					&Key{name: "namespace", value: &StrVal{value: namespace}},
-					&Key{name: "dataset", value: &StrVal{value: dataset}},
+					&Key{name: "name", value: &StrVal{value: dataset}},
 				}}})
 
 				addFieldsMap := &Dict{value: []Node{&Key{"add_fields", processorMap}}}
 				processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, addFieldsMap)
+
+				// add this for backwards compatibility remove later
+				streamProcessorMap := &Dict{value: make([]Node, 0)}
+				streamProcessorMap.value = append(streamProcessorMap.value, &Key{name: "target", value: &StrVal{value: "stream"}})
+				streamProcessorMap.value = append(streamProcessorMap.value, &Key{name: "fields", value: &Dict{value: []Node{
+					&Key{name: "type", value: &StrVal{value: r.Type}},
+					&Key{name: "namespace", value: &StrVal{value: namespace}},
+					&Key{name: "dataset", value: &StrVal{value: dataset}},
+				}}})
+
+				streamAddFieldsMap := &Dict{value: []Node{&Key{"add_fields", streamProcessorMap}}}
+
+				processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, streamAddFieldsMap)
+				// end of backward compatibility section
 			}
 		}
 	}