Add @metadata.input_id and @metadata.stream_id when injecting streams

CHANGELOG.next.asciidoc

@@ -183,3 +183,4 @@
 - Save the agent configuration and the state encrypted on the disk. {issue}535[535] {pull}398[398]
 - Bump node.js version for heartbeat/synthetics to 16.15.0
 - Support scheduled actions and cancellation of pending actions. {issue}393[393] {pull}419[419]
+- Add `@metadata.input_id` and `@metadata.stream_id` when applying the inject stream processor {pull}527[527]

internal/pkg/agent/program/testdata/audit_config-auditbeat.yml

@@ -24,6 +24,10 @@ auditbeat:
             fields:
               dataset: auditd_manager.auditd
             target: event
+        - add_fields:
+            fields:
+              stream_id: audit/auditd-auditd_manager.auditd-6d35f37c-2243-4229-9fba-ccf39305b536
+            target: '@metadata'
         - add_fields:
             fields:
               id: agent-id
@@ -59,6 +63,10 @@ auditbeat:
             fields:
               dataset: auditd_manager.file_integrity
             target: event
+        - add_fields:
+            fields:
+              stream_id: fim_1
+            target: '@metadata'
         - add_fields:
             fields:
               id: agent-id
@@ -89,6 +97,10 @@ auditbeat:
             fields:
               dataset: auditd_manager.file_integrity
             target: event
+        - add_fields:
+            fields:
+              stream_id: fim_2
+            target: '@metadata'
         - add_fields:
             fields:
               id: agent-id
@@ -117,6 +129,10 @@ auditbeat:
             fields:
               dataset: audit_system.socket
             target: event
+        - add_fields:
+            fields:
+              stream_id: id-auditd-system-socket
+            target: '@metadata'
         - add_fields:
             fields:
               id: agent-id
@@ -143,6 +159,10 @@ auditbeat:
             fields:
               dataset: audit_system.process
             target: event
+        - add_fields:
+            fields:
+              stream_id: id-auditd-system-process
+            target: '@metadata'
         - add_fields:
             fields:
               id: agent-id

internal/pkg/agent/program/testdata/logstash_config-metricbeat.yml

@@ -58,6 +58,10 @@ metricbeat:
       - add_fields:
           fields:
             should_be: first
+      - add_fields:
+          target: "@metadata"
+          fields:
+            input_id: apache-metrics-id
       - add_fields:
           target: "data_stream"
           fields:

internal/pkg/agent/program/testdata/namespace-metricbeat.yml

@@ -58,6 +58,10 @@ metricbeat:
       - add_fields:
           fields:
             should_be: first
+      - add_fields:
+          target: "@metadata"
+          fields:
+            input_id: apache-metrics-id
       - add_fields:
           target: "data_stream"
           fields:
@@ -83,7 +87,7 @@ output:
     hosts: [127.0.0.1:9200, 127.0.0.1:9300]
     headers:
       h1: test-header
-                    
+
     namespace: test_namespace
     username: elastic
     password: changeme

internal/pkg/agent/program/testdata/single_config-metricbeat.yml

@@ -58,6 +58,10 @@ metricbeat:
       - add_fields:
           fields:
             should_be: first
+      - add_fields:
+          target: "@metadata"
+          fields:
+            input_id: apache-metrics-id
       - add_fields:
           target: "data_stream"
           fields:

internal/pkg/agent/program/testdata/usecases/generated/namespace.metricbeat.golden.yml

@@ -64,6 +64,10 @@ metricbeat:
     - add_fields:
         fields:
           should_be: first
+    - add_fields:
+        fields:
+          input_id: apache-metrics-id
+        target: '@metadata'
     - add_fields:
         fields:
           dataset: generic

internal/pkg/agent/program/testdata/usecases/generated/single_config.filebeat.golden.yml

@@ -6,6 +6,10 @@ filebeat:
     - /var/log/hello1.log
     - /var/log/hello2.log
     processors:
+    - add_fields:
+        fields:
+          input_id: logfile-1
+        target: '@metadata'
     - add_fields:
         fields:
           dataset: generic
@@ -35,6 +39,10 @@ filebeat:
     - /var/log/hello3.log
     - /var/log/hello4.log
     processors:
+    - add_fields:
+        fields:
+          input_id: logfile-2
+        target: '@metadata'
     - add_fields:
         fields:
           dataset: generic

internal/pkg/agent/program/testdata/usecases/generated/single_config.metricbeat.golden.yml

@@ -64,6 +64,10 @@ metricbeat:
     - add_fields:
         fields:
           should_be: first
+    - add_fields:
+        fields:
+          input_id: apache-metrics-id
+        target: '@metadata'
     - add_fields:
         fields:
           dataset: generic

internal/pkg/agent/transpiler/rules.go

@@ -669,6 +669,42 @@ func (r *InjectStreamProcessorRule) Apply(_ AgentInfo, ast *AST) (err error) {
 		namespace := datastreamNamespaceFromInputNode(inputNode)
 		datastreamType := datastreamTypeFromInputNode(inputNode, r.Type)
 
+		var inputID *StrVal
+		inputIDNode, found := inputNode.Find("id")
+		if found {
+			inputID, _ = inputIDNode.Value().(*StrVal)
+		}
+
+		if inputID != nil {
+			// get input-level processors node
+			processorsNode, found := inputNode.Find("processors")
+			if !found {
+				processorsNode = &Key{
+					name:  "processors",
+					value: &List{value: make([]Node, 0)},
+				}
+
+				inputMap, ok := inputNode.(*Dict)
+				if ok {
+					inputMap.value = append(inputMap.value, processorsNode)
+				}
+			}
+
+			processorsList, ok := processorsNode.Value().(*List)
+			if !ok {
+				return errors.New("InjectStreamProcessorRule: input processors is not a list")
+			}
+
+			// inject `input_id` on the input level
+			processorMap := &Dict{value: make([]Node, 0)}
+			processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "@metadata"}})
+			processorMap.value = append(processorMap.value, &Key{name: "fields", value: &Dict{value: []Node{
+				&Key{name: "input_id", value: inputID},
+			}}})
+			addFieldsMap := &Dict{value: []Node{&Key{"add_fields", processorMap}}}
+			processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, addFieldsMap)
+		}
+
 		streamsNode, ok := inputNode.Find("streams")
 		if !ok {
 			continue
@@ -680,6 +716,12 @@ func (r *InjectStreamProcessorRule) Apply(_ AgentInfo, ast *AST) (err error) {
 		}
 
 		for _, streamNode := range streamsList.value {
+			var streamID *StrVal
+			streamIDNode, ok := streamNode.Find("id")
+			if ok {
+				streamID, _ = streamIDNode.Value().(*StrVal)
+			}
+
 			streamMap, ok := streamNode.(*Dict)
 			if !ok {
 				continue
@@ -722,6 +764,17 @@ func (r *InjectStreamProcessorRule) Apply(_ AgentInfo, ast *AST) (err error) {
 			}}})
 			addFieldsMap = &Dict{value: []Node{&Key{"add_fields", processorMap}}}
 			processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, addFieldsMap)
+
+			if streamID != nil {
+				// source stream
+				processorMap = &Dict{value: make([]Node, 0)}
+				processorMap.value = append(processorMap.value, &Key{name: "target", value: &StrVal{value: "@metadata"}})
+				processorMap.value = append(processorMap.value, &Key{name: "fields", value: &Dict{value: []Node{
+					&Key{name: "stream_id", value: streamID.Clone()},
+				}}})
+				addFieldsMap = &Dict{value: []Node{&Key{"add_fields", processorMap}}}
+				processorsList.value = mergeStrategy(r.OnConflict).InjectItem(processorsList.value, addFieldsMap)
+			}
 		}
 	}
 

internal/pkg/agent/transpiler/rules_test.go

@@ -165,6 +165,114 @@ inputs:
 			},
 		},
 
+		"inject stream": {
+			givenYAML: `
+inputs:
+  - name: No streams, no IDs
+    type: file
+  - name: With streams and IDs
+    id: input-id
+    type: file
+    data_stream.namespace: nsns
+    streams:
+      - paths: /var/log/mysql/error.log
+        id: stream-id
+        data_stream.dataset: dsds
+  - name: With processors
+    id: input-id
+    type: file
+    data_stream.namespace: nsns
+    processors:
+      - add_fields:
+          target: some
+          fields:
+            dataset: value
+    streams:
+      - paths: /var/log/mysql/error.log
+        id: stream-id
+        data_stream.dataset: dsds
+        processors:
+          - add_fields:
+              target: another
+              fields:
+                dataset: value
+`,
+			expectedYAML: `
+inputs:
+  - name: No streams, no IDs
+    type: file
+  - name: With streams and IDs
+    id: input-id
+    type: file
+    data_stream.namespace: nsns
+    processors:
+      - add_fields:
+          target: "@metadata"
+          fields:
+            input_id: input-id
+    streams:
+      - paths: /var/log/mysql/error.log
+        id: stream-id
+        data_stream.dataset: dsds
+        processors:
+          - add_fields:
+              target: data_stream
+              fields:
+                type: stream-type
+                namespace: nsns
+                dataset: dsds
+          - add_fields:
+              target: event
+              fields:
+                dataset: dsds
+          - add_fields:
+              target: "@metadata"
+              fields:
+                stream_id: stream-id
+  - name: With processors
+    id: input-id
+    type: file
+    data_stream.namespace: nsns
+    processors:
+      - add_fields:
+          target: some
+          fields:
+            dataset: value
+      - add_fields:
+          target: "@metadata"
+          fields:
+            input_id: input-id
+    streams:
+      - paths: /var/log/mysql/error.log
+        id: stream-id
+        data_stream.dataset: dsds
+        processors:
+          - add_fields:
+              target: another
+              fields:
+                dataset: value
+          - add_fields:
+              target: data_stream
+              fields:
+                type: stream-type
+                namespace: nsns
+                dataset: dsds
+          - add_fields:
+              target: event
+              fields:
+                dataset: dsds
+          - add_fields:
+              target: "@metadata"
+              fields:
+                stream_id: stream-id
+`,
+			rule: &RuleList{
+				Rules: []Rule{
+					InjectStreamProcessor("insert_after", "stream-type"),
+				},
+			},
+		},
+
 		"inject agent info": {
 			givenYAML: `
 inputs: