processor ends when next one start

haetamoudi · haetamoudi · commit 9d2902f391c2 · 2024-09-06T15:57:08.000+02:00
diff --git a/internal/elasticsearch/ingest/processors.go b/internal/elasticsearch/ingest/processors.go
@@ -5,8 +5,9 @@
 package ingest
 
 import (
+	"bufio"
+	"bytes"
 	"fmt"
-	"strings"
 
 	"gopkg.in/yaml.v3"
 )
@@ -59,6 +60,7 @@ func processorsFromYAML(content []byte) (procs []Processor, err error) {
 	if err = yaml.Unmarshal(content, &p); err != nil {
 		return nil, err
 	}
+
 	for idx, entry := range p.Processors {
 		if entry.Kind != yaml.MappingNode || len(entry.Content) != 2 {
 			return nil, fmt.Errorf("processor#%d is not a single-key map (kind:%v content:%d)", idx, entry.Kind, len(entry.Content))
@@ -71,28 +73,88 @@ func processorsFromYAML(content []byte) (procs []Processor, err error) {
 			return nil, fmt.Errorf("error decoding processor#%d type: %w", idx, err)
 		}
 		proc.FirstLine = entry.Line
-		proc.LastLine = lastLine(&entry)
+		lastLine, err := getProcessorLastLine(idx, p.Processors, proc, content)
+		if err != nil {
+			return nil, err
+		}
+		proc.LastLine = lastLine
+
 		procs = append(procs, proc)
 	}
-	return procs, nil
+	return procs, err
+}
+
+// getProcessorLastLine determines the last line number for the given processor.
+func getProcessorLastLine(idx int, processors []yaml.Node, currentProcessor Processor, content []byte) (int, error) {
+	if idx < len(processors)-1 {
+		var endProcessor = processors[idx+1].Line - 1
+		if endProcessor < currentProcessor.FirstLine {
+			return currentProcessor.FirstLine, nil
+		} else {
+			return processors[idx+1].Line - 1, nil
+		}
+	}
+
+	return nextProcessorOrEndOfPipeline(content)
 }
 
-// lastLine returns the last (greater) line number used by a yaml.Node.
-func lastLine(node *yaml.Node) int {
+// lastProcessorLine get the line before the node after the processors node. If there is none, it returns the end of file line
+func nextProcessorOrEndOfPipeline(content []byte) (int, error) {
+	var root yaml.Node
+	if err := yaml.Unmarshal(content, &root); err != nil {
+		return 0, fmt.Errorf("error unmarshaling YAML: %v", err)
+	}
+
+	var nodes []*yaml.Node
+	extractNodesFromMapping(&root, &nodes)
+	for i, node := range nodes {
+
+		if node.Value == "processors" {
+			if i < len(nodes)-1 {
+
+				return nodes[i+1].Line - 1, nil
+			}
+		}
+
+	}
+	return countLinesInBytes(content)
+}
+
+// extractNodesFromMapping recursively extracts all nodes from MappingNodes within DocumentNodes.
+func extractNodesFromMapping(node *yaml.Node, nodes *[]*yaml.Node) {
 	if node == nil {
-		return 0
+		return
 	}
-	last := node.Line
-	for _, inner := range node.Content {
-		if line := lastLine(inner); line > last {
-			last = line
+
+	if node.Kind == yaml.DocumentNode {
+		for _, child := range node.Content {
+			extractNodesFromMapping(child, nodes)
 		}
-		// For scalar node with multiline content, calculate the last line based on line breaks
-		if inner.Kind == yaml.ScalarNode && strings.Contains(inner.Value, "\n") {
-			lineCount := strings.Count(inner.Value, "\n")
-			last += lineCount
+		return
+	}
+
+	if node.Kind == yaml.MappingNode {
+		for _, child := range node.Content {
+			if child.Kind == yaml.MappingNode || child.Kind == yaml.ScalarNode {
+				*nodes = append(*nodes, child)
+			}
+			extractNodesFromMapping(child, nodes)
 		}
 	}
+}
+
+// countLinesInBytes counts the number of lines in the given byte slice.
+func countLinesInBytes(data []byte) (int, error) {
+	scanner := bufio.NewScanner(bytes.NewReader(data))
+	lineCount := 0
+
+	for scanner.Scan() {
+		lineCount++
+	}
+
+	if err := scanner.Err(); err != nil {
+		return 0, fmt.Errorf("error reading data: %w", err)
+	}
 
-	return last
+	return lineCount, nil
 }
diff --git a/internal/elasticsearch/ingest/processors_test.go b/internal/elasticsearch/ingest/processors_test.go
@@ -24,61 +24,104 @@ func TestResource_Processors(t *testing.T) {
 			content: []byte(`---
 description: Made up pipeline
 processors:
-# First processor.
-- grok:
-    tag: Extract header
-    field: message
-    patterns:
-    - \[%{APACHE_TIME:apache.error.timestamp}\] \[%{LOGLEVEL:log.level}\]( \[client
-      %{IPORHOST:source.address}(:%{POSINT:source.port})?\])? %{GREEDYDATA:message}
-    - \[%{APACHE_TIME:apache.error.timestamp}\] \[%{DATA:apache.error.module}:%{LOGLEVEL:log.level}\]
-      \[pid %{NUMBER:process.pid:long}(:tid %{NUMBER:process.thread.id:long})?\](
-      \[client %{IPORHOST:source.address}(:%{POSINT:source.port})?\])? %{GREEDYDATA:message}
-    pattern_definitions:
-      APACHE_TIME: '%{DAY} %{MONTH} %{MONTHDAY} %{TIME} %{YEAR}'
-    ignore_missing: true
+    - grok:
+        tag: Extract header
+        field: message
+        patterns:
+        - \[%{APACHE_TIME:apache.error.timestamp}\] \[%{LOGLEVEL:log.level}\]( \[client%{IPORHOST:source.address}(:%{POSINT:source.port})?\])? %{GREEDYDATA:message}
+        - \[%{APACHE_TIME:apache.error.timestamp}\] \[%{DATA:apache.error.module}:%{LOGLEVEL:log.level}\]
+        ignore_missing: true
 
-- date:
-    field: apache.error.timestamp
-    target_field: '@timestamp'
-    formats:
-    - EEE MMM dd H:m:s yyyy
-    - EEE MMM dd H:m:s.SSSSSS yyyy
-    on_failure:
-    - append:
+    - date:
+        field: apache.error.timestamp
+        target_field: '@timestamp'
+        formats:
+        - EEE MMM dd H:m:s yyyy
+        - EEE MMM dd H:m:s.SSSSSS yyyy
+        on_failure:
+        - append:
+            field: error.message
+            value: '{{ _ingest.on_failure_message }}'
+    - set:
+        description: Set event category
+        field: event.category
+        value: web
+    # Some script
+    - script:
+        lang: painless
+
+    - grok:
+        field: source.address
+        ignore_missing: true
+        patterns:
+        - ^(%{IP:source.ip}|%{HOSTNAME:source.domain})$
+    - rename:
+        field: source.as.organization_name
+        target_field: source.as.organization.name
+        ignore_missing: true
+on_failure:
+    - set:
         field: error.message
         value: '{{ _ingest.on_failure_message }}'
-- set:
-    description: Set event category
-    field: event.category
-    value: web
-# Some script
-- script:
-    lang: painless
-    source: >-
-      [...]
+`),
+			expected: []Processor{
+				{Type: "grok", FirstLine: 4, LastLine: 11},
+				{Type: "date", FirstLine: 12, LastLine: 21},
+				{Type: "set", FirstLine: 22, LastLine: 26},
+				{Type: "script", FirstLine: 27, LastLine: 29},
+				{Type: "grok", FirstLine: 30, LastLine: 34},
+				{Type: "rename", FirstLine: 35, LastLine: 38},
+			},
+		},
+		{
+			name:   "Yaml pipeline",
+			format: "yml",
+			content: []byte(`---
+description: Made up pipeline
+processors:
+    - grok:
+        tag: Extract header
+        field: message
+        patterns:
+        - \[%{APACHE_TIME:apache.error.timestamp}\] \[%{LOGLEVEL:log.level}\]( \[client%{IPORHOST:source.address}(:%{POSINT:source.port})?\])? %{GREEDYDATA:message}
+        - \[%{APACHE_TIME:apache.error.timestamp}\] \[%{DATA:apache.error.module}:%{LOGLEVEL:log.level}\]
+        ignore_missing: true
 
-- grok:
-    field: source.address
-    ignore_missing: true
-    patterns:
-    - ^(%{IP:source.ip}|%{HOSTNAME:source.domain})$
-- rename:
-    field: source.as.organization_name
-    target_field: source.as.organization.name
-    ignore_missing: true
-on_failure:
-- set:
-    field: error.message
-    value: '{{ _ingest.on_failure_message }}'
+    - date:
+        field: apache.error.timestamp
+        target_field: '@timestamp'
+        formats:
+        - EEE MMM dd H:m:s yyyy
+        - EEE MMM dd H:m:s.SSSSSS yyyy
+        on_failure:
+        - append:
+            field: error.message
+            value: '{{ _ingest.on_failure_message }}'
+    - set:
+        description: Set event category
+        field: event.category
+        value: web
+    # Some script
+    - script:
+        lang: painless
+
+    - grok:
+        field: source.address
+        ignore_missing: true
+        patterns:
+        - ^(%{IP:source.ip}|%{HOSTNAME:source.domain})$
+    - rename:
+        field: source.as.organization_name
+        target_field: source.as.organization.name
+        ignore_missing: true
 `),
 			expected: []Processor{
-				{Type: "grok", FirstLine: 5, LastLine: 16},
-				{Type: "date", FirstLine: 18, LastLine: 27},
-				{Type: "set", FirstLine: 28, LastLine: 31},
-				{Type: "script", FirstLine: 33, LastLine: 35},
-				{Type: "grok", FirstLine: 38, LastLine: 42},
-				{Type: "rename", FirstLine: 43, LastLine: 46},
+				{Type: "grok", FirstLine: 4, LastLine: 11},
+				{Type: "date", FirstLine: 12, LastLine: 21},
+				{Type: "set", FirstLine: 22, LastLine: 26},
+				{Type: "script", FirstLine: 27, LastLine: 29},
+				{Type: "grok", FirstLine: 30, LastLine: 34},
+				{Type: "rename", FirstLine: 35, LastLine: 38},
 			},
 		},
 		{
@@ -109,10 +152,10 @@ on_failure:
 `),
 			expected: []Processor{
 				{Type: "drop", FirstLine: 3, LastLine: 3},
-				{Type: "set", FirstLine: 4, LastLine: 7},
+				{Type: "set", FirstLine: 4, LastLine: 8},
 				{Type: "remove", FirstLine: 9, LastLine: 9},
 				{Type: "set", FirstLine: 9, LastLine: 9},
-				{Type: "set", FirstLine: 10, LastLine: 13},
+				{Type: "set", FirstLine: 10, LastLine: 15},
 			},
 		},
 		{
@@ -155,7 +198,7 @@ on_failure:
 				"processors": [{"drop": {"if":"ctx.drop!=null"}}]
 			  }`),
 			expected: []Processor{
-				{Type: "drop", FirstLine: 3, LastLine: 3},
+				{Type: "drop", FirstLine: 3, LastLine: 4},
 			},
 		},
 		{
@@ -173,7 +216,7 @@ on_failure:
   ]
 }`),
 			expected: []Processor{
-				{Type: "script", FirstLine: 3, LastLine: 10},
+				{Type: "script", FirstLine: 3, LastLine: 11},
 				//   Source will be processed as multiline:
 				//  "source": """
 				// 		String[] envSplit = ctx['env'].splitOnToken(params['delimiter']);
@@ -222,6 +265,22 @@ processors:
 				{Type: "script", FirstLine: 3, LastLine: 6},
 			},
 		},
+		{
+			name:   "Yaml script with empty line characters",
+			format: "yml",
+			content: []byte(`---
+processors:
+  - script:
+      description: Do something.
+      tag: script_drop_null_empty_values
+      lang: painless
+      source: "def a = b \n
+	  ; def b = 2; \n"
+`),
+			expected: []Processor{
+				{Type: "script", FirstLine: 3, LastLine: 8},
+			},
+		},
 		{
 			name:   "Yaml empty processor",
 			format: "yml",
@@ -249,7 +308,7 @@ processors:
         def b = 2;
 `),
 			expected: []Processor{
-				{Type: "set", FirstLine: 4, LastLine: 6},
+				{Type: "set", FirstLine: 4, LastLine: 8},
 				{Type: "script", FirstLine: 9, LastLine: 12},
 			},
 		},
