Use event.kind to validate if there is any pipeline error

internal/testrunner/runners/system/tester.go

@@ -2429,8 +2429,8 @@ func writeSampleEvent(path string, doc common.MapStr, specVersion semver.Version
 func validateFields(docs []common.MapStr, fieldsValidator *fields.Validator) multierror.Error {
 	var multiErr multierror.Error
 	for _, doc := range docs {
-		if message, err := doc.GetValue("error.message"); err != common.ErrKeyNotFound {
-			multiErr = append(multiErr, fmt.Errorf("found error.message in event: %v", message))
+		if errorMessage := pipelineErrorMessage(doc); errorMessage != "" {
+			multiErr = append(multiErr, errors.New(errorMessage))
 			continue
 		}
 
@@ -2446,6 +2446,53 @@ func validateFields(docs []common.MapStr, fieldsValidator *fields.Validator) mul
 	return nil
 }
 
+// stringFromDocValue converts a document field value to a single string.
+// It handles synthetic source ([]any or []string) and plain string values.
+func stringsFromDocValue(value any) []string {
+	switch v := value.(type) {
+	case []any:
+		var values []string
+		for _, item := range v {
+			if s, ok := item.(string); ok {
+				values = append(values, s)
+			}
+		}
+		return values
+	case []string:
+		return v
+	case string:
+		return []string{v}
+	default:
+		return nil
+	}
+}
+
+// pipelineErrorMessage checks if the event has a pipeline_error and returns the error message if it does
+// If the event does not have a pipeline_error, it returns the empty string
+func pipelineErrorMessage(doc common.MapStr) string {
+	message, err := doc.GetValue("event.kind")
+	if err != nil {
+		// Skip any error (unexpected type for event.kind, key not found, etc.)
+		return ""
+	}
+	eventKind := stringsFromDocValue(message)
+	if !slices.Contains(eventKind, "pipeline_error") {
+		// Unexpected type for event.kind field, skip validation
+		// or it is not related to a pipeline error
+		return ""
+	}
+
+	errorMessage := []string{}
+	if errorMessageData, err := doc.GetValue("error.message"); err == nil {
+		errorMessage = stringsFromDocValue(errorMessageData)
+	}
+	if len(errorMessage) == 0 {
+		return "found pipeline_error in document: no error message"
+	}
+
+	return fmt.Sprintf("found pipeline_error in document with error message: %q", strings.Join(errorMessage, " "))
+}
+
 func listExceptionFields(docs []common.MapStr, fieldsValidator *fields.Validator) []string {
 	var allFields []string
 	visited := make(map[string]any)

internal/testrunner/runners/system/tester_test.go

@@ -522,3 +522,140 @@ func TestGetExpectedDatasetForTest(t *testing.T) {
 		})
 	}
 }
+
+func TestPipelineErrorMessage(t *testing.T) {
+	testCases := []struct {
+		name     string
+		doc      common.MapStr
+		expected string
+	}{
+		{
+			name:     "empty doc",
+			doc:      common.MapStr{},
+			expected: "",
+		},
+		{
+			name: "doc without event.kind",
+			doc: common.MapStr{
+				"message": "something",
+			},
+			expected: "",
+		},
+		{
+			name: "event.kind is not pipeline_error",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "event",
+				},
+			},
+			expected: "",
+		},
+		{
+			name: "event.kind is non-string",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": 42,
+				},
+			},
+			expected: "",
+		},
+		{
+			name: "pipeline_error without error.message",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "pipeline_error",
+				},
+			},
+			expected: "found pipeline_error in document: no error message",
+		},
+		{
+			name: "pipeline_error with empty error.message",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "pipeline_error",
+				},
+				"error": common.MapStr{
+					"message": "",
+				},
+			},
+			expected: "found pipeline_error in document with error message: \"\"",
+		},
+		{
+			name: "pipeline_error with non-string error.message",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "pipeline_error",
+				},
+				"error": common.MapStr{
+					"message": 123,
+				},
+			},
+			expected: "found pipeline_error in document: no error message",
+		},
+		{
+			name: "pipeline_error with error.message",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "pipeline_error",
+				},
+				"error": common.MapStr{
+					"message": "ingest pipeline failed",
+				},
+			},
+			expected: "found pipeline_error in document with error message: \"ingest pipeline failed\"",
+		},
+		{
+			name: "pipeline_error with error.message as array",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "pipeline_error",
+				},
+				"error": common.MapStr{
+					"message": []any{"ingest pipeline failed"},
+				},
+			},
+			expected: "found pipeline_error in document with error message: \"ingest pipeline failed\"",
+		},
+		{
+			name: "pipeline_error using synthetic source mode",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": []any{"pipeline_error"},
+				},
+				"error": common.MapStr{
+					"message": []any{"ingest pipeline failed"},
+				},
+			},
+			expected: "found pipeline_error in document with error message: \"ingest pipeline failed\"",
+		},
+		{
+			name: "unexpected type for event field",
+			doc: common.MapStr{
+				"event": []any{"foo"},
+				"error": common.MapStr{
+					"message": "ingest pipeline failed",
+				},
+			},
+			expected: "",
+		},
+		{
+			name: "unexpected type for error field",
+			doc: common.MapStr{
+				"event": common.MapStr{
+					"kind": "pipeline_error",
+				},
+				"error": []any{
+					"404 error code",
+				},
+			},
+			expected: "found pipeline_error in document: no error message",
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			got := pipelineErrorMessage(tc.doc)
+			assert.Equal(t, tc.expected, got)
+		})
+	}
+}

scripts/test-check-false-positives.sh

@@ -23,7 +23,7 @@ function cleanup() {
   if is_stack_created ; then
     # Dump stack logs
     # Required containers could not be running, so ignore the error
-    elastic-package stack dump -v --output "build/elastic-stack-dump/check-${SUFFIX_FOLDER_DUMP_LOGS}}" || true
+    elastic-package stack dump -v --output "build/elastic-stack-dump/check-${SUFFIX_FOLDER_DUMP_LOGS}" || true
 
     # Take down the stack
     elastic-package stack down -v

test/packages/false_positives/pipeline_error_system_tests.expected_errors

@@ -0,0 +1 @@
+<testcase name=\"system test: default\" classname=\"pipeline_error_system_tests.logs\" time=\".*\"> * <failure>test case failed: one or more errors found in documents stored in logs-pipeline_error_system_tests.logs-[[:digit:]]+ data stream: \[0\] found pipeline_error in document with error message: &#34;process logs failed&#34;</failure>

test/packages/false_positives/pipeline_error_system_tests/LICENSE.txt

@@ -0,0 +1,93 @@
+Elastic License 2.0
+
+URL: https://www.elastic.co/licensing/elastic-license
+
+## Acceptance
+
+By using the software, you agree to all of the terms and conditions below.
+
+## Copyright License
+
+The licensor grants you a non-exclusive, royalty-free, worldwide,
+non-sublicensable, non-transferable license to use, copy, distribute, make
+available, and prepare derivative works of the software, in each case subject to
+the limitations and conditions below.
+
+## Limitations
+
+You may not provide the software to third parties as a hosted or managed
+service, where the service provides users with access to any substantial set of
+the features or functionality of the software.
+
+You may not move, change, disable, or circumvent the license key functionality
+in the software, and you may not remove or obscure any functionality in the
+software that is protected by the license key.
+
+You may not alter, remove, or obscure any licensing, copyright, or other notices
+of the licensor in the software. Any use of the licensor’s trademarks is subject
+to applicable law.
+
+## Patents
+
+The licensor grants you a license, under any patent claims the licensor can
+license, or becomes able to license, to make, have made, use, sell, offer for
+sale, import and have imported the software, in each case subject to the
+limitations and conditions in this license. This license does not cover any
+patent claims that you cause to be infringed by modifications or additions to
+the software. If you or your company make any written claim that the software
+infringes or contributes to infringement of any patent, your patent license for
+the software granted under these terms ends immediately. If your company makes
+such a claim, your patent license ends immediately for work on behalf of your
+company.
+
+## Notices
+
+You must ensure that anyone who gets a copy of any part of the software from you
+also gets a copy of these terms.
+
+If you modify the software, you must include in any modified copies of the
+software prominent notices stating that you have modified the software.
+
+## No Other Rights
+
+These terms do not imply any licenses other than those expressly granted in
+these terms.
+
+## Termination
+
+If you use the software in violation of these terms, such use is not licensed,
+and your licenses will automatically terminate. If the licensor provides you
+with a notice of your violation, and you cease all violation of this license no
+later than 30 days after you receive that notice, your licenses will be
+reinstated retroactively. However, if you violate these terms after such
+reinstatement, any additional violation of these terms will cause your licenses
+to terminate automatically and permanently.
+
+## No Liability
+
+*As far as the law allows, the software comes as is, without any warranty or
+condition, and the licensor will not be liable to you for any damages arising
+out of these terms or the use or nature of the software, under any kind of
+legal claim.*
+
+## Definitions
+
+The **licensor** is the entity offering these terms, and the **software** is the
+software the licensor makes available under these terms, including any portion
+of it.
+
+**you** refers to the individual or entity agreeing to these terms.
+
+**your company** is any legal entity, sole proprietorship, or other kind of
+organization that you work for, plus all organizations that have control over,
+are under the control of, or are under common control with that
+organization. **control** means ownership of substantially all the assets of an
+entity, or the power to direct its management and policies by vote, contract, or
+otherwise. Control can be direct or indirect.
+
+**your licenses** are all the licenses granted to you for the software under
+these terms.
+
+**use** means anything you do with the software requiring one of your licenses.
+
+**trademark** means trademarks, service marks, and similar rights.

test/packages/false_positives/pipeline_error_system_tests/_dev/build/build.yml

@@ -0,0 +1,3 @@
+dependencies:
+  ecs:
+    reference: git@v8.17.0

test/packages/false_positives/pipeline_error_system_tests/changelog.yml

@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1 # FIXME Replace with the real PR link

test/packages/false_positives/runtime_fields/test-default-config.yml → test/packages/false_positives/pipeline_error_system_tests/data_stream/logs/_dev/test/system/test-default-config.yml

@@ -1,3 +1,4 @@
+---
 vars: ~
 data_stream:
   vars:

test/packages/false_positives/pipeline_error_system_tests/data_stream/logs/agent/stream/stream.yml.hbs

@@ -0,0 +1,7 @@
+paths:
+{{#each paths as |path i|}}
+  - {{path}}
+{{/each}}
+exclude_files: [".gz$"]
+processors:
+  - add_locale: ~

test/packages/false_positives/pipeline_error_system_tests/data_stream/logs/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,22 @@
+---
+description: Pipeline for processing sample logs
+processors:
+- set:
+    field: labels.foo
+    value: "bar"
+- set:
+    field: labels.version
+    value: "v1"
+- set:
+    field: metrics.bar
+    value: 10
+- set:
+    field: event.kind
+    value: pipeline_error
+- set:
+    field: error.message
+    value: process logs failed
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'

test/packages/false_positives/pipeline_error_system_tests/data_stream/logs/fields/base-fields.yml

@@ -0,0 +1,12 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: '@timestamp'
+  type: date
+  description: Event timestamp.

test/packages/false_positives/pipeline_error_system_tests/data_stream/logs/fields/fields.yml

@@ -0,0 +1,6 @@
+- name: input.type
+  type: keyword
+- name: log.offset
+  type: long
+- name: metrics.bar
+  type: long

test/packages/false_positives/pipeline_error_system_tests/data_stream/logs/manifest.yml

@@ -0,0 +1,17 @@
+title: "Logs"
+type: logs
+streams:
+  - input: logfile
+    title: Sample logs
+    description: Collect sample logs
+    vars:
+      - name: paths
+        type: text
+        title: Paths
+        multi: true
+        default:
+          - /var/log/*.log
+elasticsearch:
+  index_template:
+    mappings:
+      subobjects: false