[Ingest Manager] Send datastreams fields (#20416)

elastic · Aug 4, 2020 · 24a03eb · 24a03eb
1 parent 7198f69
commit 24a03eb
Show file tree

Hide file tree

Showing 8 changed files with 92 additions and 0 deletions.
diff --git a/x-pack/elastic-agent/CHANGELOG.asciidoc b/x-pack/elastic-agent/CHANGELOG.asciidoc
@@ -98,3 +98,4 @@
 - Will retry to enroll if the server return a 429. {pull}19918[19811]
 - Add --staging option to enroll command {pull}20026[20026]
 - Add `event.dataset` to all events {pull}20076[20076]
+- Send datastreams fields {pull}20416[20416]
diff --git a/x-pack/elastic-agent/pkg/agent/operation/monitoring.go b/x-pack/elastic-agent/pkg/agent/operation/monitoring.go
@@ -198,6 +198,16 @@ func (o *Operator) getMonitoringFilebeatConfig(output interface{}) (map[string]i
 						},
 					},
 				},
+				{
+					"add_fields": map[string]interface{}{
+						"target": "datastream",
+						"fields": map[string]interface{}{
+							"type":      "logs",
+							"dataset":   "elastic.agent",
+							"namespace": "default",
+						},
+					},
+				},
 				{
 					"add_fields": map[string]interface{}{
 						"target": "event",
@@ -232,6 +242,16 @@ func (o *Operator) getMonitoringFilebeatConfig(output interface{}) (map[string]i
 							},
 						},
 					},
+					{
+						"add_fields": map[string]interface{}{
+							"target": "datastream",
+							"fields": map[string]interface{}{
+								"type":      "logs",
+								"dataset":   fmt.Sprintf("elastic.agent.%s", name),
+								"namespace": "default",
+							},
+						},
+					},
 					{
 						"add_fields": map[string]interface{}{
 							"target": "event",
@@ -282,6 +302,16 @@ func (o *Operator) getMonitoringMetricbeatConfig(output interface{}) (map[string
 						},
 					},
 				},
+				{
+					"add_fields": map[string]interface{}{
+						"target": "datastream",
+						"fields": map[string]interface{}{
+							"type":      "metrics",
+							"dataset":   fmt.Sprintf("elastic.agent.%s", name),
+							"namespace": "default",
+						},
+					},
+				},
 				{
 					"add_fields": map[string]interface{}{
 						"target": "event",

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
@@ -12,6 +12,12 @@ filebeat:
             type: logs
             name: generic
             namespace: default
+      - add_fields:
+          target: "datastream"
+          fields:
+            type: logs
+            dataset: generic
+            namespace: default
       - add_fields:
           target: "event"
           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
@@ -12,6 +12,12 @@ filebeat:
             type: logs
             name: generic
             namespace: default
+      - add_fields:
+          target: "datastream"
+          fields:
+            type: logs
+            dataset: generic
+            namespace: default
       - add_fields:
           target: "event"
           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
@@ -13,6 +13,12 @@ filebeat:
             type: logs
             name: generic
             namespace: default
+      - add_fields:
+          target: "datastream"
+          fields:
+            type: logs
+            dataset: generic
+            namespace: default
       - add_fields:
           target: "event"
           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
@@ -14,6 +14,12 @@ filebeat:
             type: logs
             name: generic
             namespace: default
+      - add_fields:
+          target: "datastream"
+          fields:
+            type: logs
+            dataset: generic
+            namespace: default
       - add_fields:
           target: "event"
           fields:
@@ -32,6 +38,12 @@ filebeat:
             type: testtype
             name: generic
             namespace: default
+      - add_fields:
+          target: "datastream"
+          fields:
+            type: testtype
+            dataset: generic
+            namespace: default
       - add_fields:
           target: "event"
           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
@@ -11,6 +11,12 @@ metricbeat:
           type: metrics
           name: docker.status
           namespace: default
+    - add_fields:
+        target: "datastream"
+        fields:
+          type: metrics
+          dataset: docker.status
+          namespace: default
     - add_fields:
         target: "event"
         fields:
@@ -26,6 +32,12 @@ metricbeat:
           type: metrics
           name: generic
           namespace: default
+    - add_fields:
+        target: "datastream"
+        fields:
+          type: metrics
+          dataset: generic
+          namespace: default
     - add_fields:
         target: "event"
         fields:
@@ -44,6 +56,12 @@ metricbeat:
             type: metrics
             name: generic
             namespace: testing
+      - add_fields:
+          target: "datastream"
+          fields:
+            type: metrics
+            dataset: generic
+            namespace: testing
       - add_fields:
           target: "event"
           fields:

diff --git a/x-pack/elastic-agent/pkg/agent/transpiler/rules.go b/x-pack/elastic-agent/pkg/agent/transpiler/rules.go
@@ -632,6 +632,7 @@ func (r *InjectStreamProcessorRule) Apply(ast *AST) error {
 				return errors.New("InjectStreamProcessorRule: processors is not a list")
 			}
 
+			// dataset
 			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: "fields", value: &Dict{value: []Node{
@@ -642,6 +643,18 @@ func (r *InjectStreamProcessorRule) Apply(ast *AST) error {
 			addFieldsMap := &Dict{value: []Node{&Key{"add_fields", processorMap}}}
 			processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, addFieldsMap)
 
+			// datastream
+			processorMap = &Dict{value: make([]Node, 0)}
+			processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "datastream"}})
+			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: "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{