fixes

taylor-swanson · taylor-swanson · commit 0e3ba16ca9e1 · 2025-11-11T10:51:48.000-06:00
diff --git a/code/go/internal/validator/semantic/types.go b/code/go/internal/validator/semantic/types.go
@@ -319,7 +319,7 @@ func listDataStreams(fsys fspath.FS) ([]string, error) {
 func listPipelineFiles(fsys fspath.FS, dataStream string) ([]pipelineFileMetadata, error) {
 	var pipelineFileMetadatas []pipelineFileMetadata
 
-	ingestPipelineDir := path.Join(dataStreamDir, "elasticsearch", "ingest_pipeline")
+	ingestPipelineDir := path.Join(dataStreamDir, dataStream, "elasticsearch", "ingest_pipeline")
 	pipelineFiles, err := readPipelinesFolder(fsys, ingestPipelineDir)
 	if err != nil {
 		return nil, fmt.Errorf("cannot read pipeline files from integration package: %w", err)
diff --git a/code/go/internal/validator/semantic/validate_pipeline_tags.go b/code/go/internal/validator/semantic/validate_pipeline_tags.go
@@ -39,7 +39,7 @@ func ValidatePipelineTags(fsys fspath.FS) specerrors.ValidationErrors {
 				return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
 			}
 
-			if vErrs := validatePipelineTags(&pipeline); len(vErrs) > 0 {
+			if vErrs := validatePipelineTags(&pipeline, pipelineFile.fullFilePath); len(vErrs) > 0 {
 				errors = append(errors, vErrs...)
 			}
 		}
@@ -48,44 +48,44 @@ func ValidatePipelineTags(fsys fspath.FS) specerrors.ValidationErrors {
 	return errors
 }
 
-func validatePipelineTags(pipeline *ingestPipeline) specerrors.ValidationErrors {
+func validatePipelineTags(pipeline *ingestPipeline, filename string) specerrors.ValidationErrors {
 	var errors specerrors.ValidationErrors
 
 	seen := map[string]struct{}{}
 	for _, proc := range pipeline.Processors {
-		procErrors := checkPipelineTag(proc, seen)
+		procErrors := checkPipelineTag(proc, seen, filename)
 		errors = append(errors, procErrors...)
 	}
 
 	return errors
 }
 
-func checkPipelineTag(proc *processor, seen map[string]struct{}) specerrors.ValidationErrors {
+func checkPipelineTag(proc *processor, seen map[string]struct{}, filename string) specerrors.ValidationErrors {
 	var errors specerrors.ValidationErrors
 
 	for _, subProc := range proc.OnFailure {
-		subErrors := checkPipelineTag(subProc, seen)
+		subErrors := checkPipelineTag(subProc, seen, filename)
 		errors = append(errors, subErrors...)
 	}
 
 	raw, ok := proc.Attributes["tag"]
 	if !ok {
-		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d missing required tag", proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))
+		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d missing required tag", filename, proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))
 		return errors
 	}
 
 	tag, ok := raw.(string)
 	if !ok {
-		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d has invalid tag value", proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))
+		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d has invalid tag value", filename, proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))
 		return errors
 	}
 	if tag == "" {
-		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d has empty tag value", proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))
+		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d has empty tag value", filename, proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))
 		return errors
 	}
 
 	if _, dup := seen[tag]; dup {
-		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d has duplicate tag value: %q", proc.Type, proc.position.line, tag), specerrors.CodePipelineTagDuplicate))
+		errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d has duplicate tag value: %q", filename, proc.Type, proc.position.line, tag), specerrors.CodePipelineTagDuplicate))
 		return errors
 	}
 
diff --git a/code/go/internal/validator/semantic/validate_pipeline_tags_test.go b/code/go/internal/validator/semantic/validate_pipeline_tags_test.go
@@ -51,7 +51,7 @@ processors:
             key: fail_value_1
 `,
 			errors: []string{
-				`set processor at line 3 missing required tag (SVR00006)`,
+				`file "default.yml" is invalid: set processor at line 3 missing required tag (SVR00006)`,
 			},
 		},
 		{
@@ -68,7 +68,7 @@ processors:
             value: fail_value_1
 `,
 			errors: []string{
-				`set processor at line 8 missing required tag (SVR00006)`,
+				`file "default.yml" is invalid: set processor at line 8 missing required tag (SVR00006)`,
 			},
 		},
 		{
@@ -85,7 +85,7 @@ processors:
       value: value2
 `,
 			errors: []string{
-				`set processor at line 7 has duplicate tag value: "set_1" (SVR00007)`,
+				`file "default.yml" is invalid: set processor at line 7 has duplicate tag value: "set_1" (SVR00007)`,
 			},
 		},
 		{
@@ -103,7 +103,7 @@ processors:
             value: fail_value_1
 `,
 			errors: []string{
-				`set processor at line 3 has duplicate tag value: "set_1" (SVR00007)`,
+				`file "default.yml" is invalid: set processor at line 3 has duplicate tag value: "set_1" (SVR00007)`,
 			},
 		},
 		{
@@ -116,7 +116,7 @@ processors:
       value: value1
 `,
 			errors: []string{
-				`set processor at line 3 has invalid tag value (SVR00006)`,
+				`file "default.yml" is invalid: set processor at line 3 has invalid tag value (SVR00006)`,
 			},
 		},
 		{
@@ -129,7 +129,7 @@ processors:
       value: value1
 `,
 			errors: []string{
-				`set processor at line 3 has empty tag value (SVR00006)`,
+				`file "default.yml" is invalid: set processor at line 3 has empty tag value (SVR00006)`,
 			},
 		},
 	}
@@ -140,7 +140,7 @@ processors:
 			err := yaml.Unmarshal([]byte(tc.pipeline), &pipeline)
 			require.NoError(t, err)
 
-			errors := validatePipelineTags(&pipeline)
+			errors := validatePipelineTags(&pipeline, "default.yml")
 			assert.Len(t, errors, len(tc.errors))
 			for _, err := range errors {
 				assert.Contains(t, tc.errors, err.Error())
diff --git a/code/go/pkg/validator/validator_test.go b/code/go/pkg/validator/validator_test.go
@@ -829,6 +829,12 @@ func TestValidateIngestPipelines(t *testing.T) {
 				"field processors.2.remove.if: rename \"message\" to \"event.original\" processor requires remove \"message\" processor with if: 'ctx.event?.original != null' (JSE00001)",
 			},
 		},
+		"bad_pipeline_tags": {
+			"example": []string{
+				"set processor at line 4 missing required tag (SVR00006)",
+				"set processor at line 15 has duplicate tag value: \"set_sample_field\" (SVR00007)",
+			},
+		},
 	}
 
 	for pkgName, pipelines := range tests {
diff --git a/test/packages/bad_agent_version_v3/data_stream/agent_settings/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_agent_version_v3/data_stream/agent_settings/elasticsearch/ingest_pipeline/default.yml
@@ -2,6 +2,7 @@
 description: Pipeline for processing sample logs
 processors:
 - set:
+    tag: set_sample_field
     field: sample_field
     value: "1"
 on_failure:
diff --git a/test/packages/bad_agent_version_v3/data_stream/ecs_import_mappings/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_agent_version_v3/data_stream/ecs_import_mappings/elasticsearch/ingest_pipeline/default.yml
@@ -2,6 +2,7 @@
 description: Pipeline for processing sample logs
 processors:
 - set:
+    tag: set_sample_field
     field: sample_field
     value: "1"
 on_failure:
diff --git a/test/packages/bad_agent_version_v3/data_stream/foo/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_agent_version_v3/data_stream/foo/elasticsearch/ingest_pipeline/default.yml
@@ -3,6 +3,7 @@ description: Pipeline for Microsoft DHCP
 
 processors:
   - rename:
+      tag: rename_foo_to_message
       field: foo
       target_field: message
       ignore_missing: true
diff --git a/test/packages/bad_agent_version_v3/data_stream/rename_message/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_agent_version_v3/data_stream/rename_message/elasticsearch/ingest_pipeline/default.yml
@@ -3,12 +3,14 @@ description: Test pipeline with renaming processor
 processors:
     # cf. https://github.com/elastic/integrations/pull/7026/files#diff-dafc5693f21abb74b294987b502cdc5770f22cccff3c5a8fee0d48510791b811
     - rename:
+        tag: rename_message_to_event_original
         field: message
         target_field: event.original
         ignore_missing: true
         if: 'ctx.event?.original == null'
         description: 'Renames the original `message` field to `event.original` to store a copy of the original message. The `event.original` field is not touched if the document already has one; it may happen when Logstash sends the document.'
     - remove:
+        tag: remove_message
         field: message
         ignore_missing: true
         if: 'ctx.event?.original != null'
diff --git a/test/packages/bad_agent_version_v3/data_stream/root/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_agent_version_v3/data_stream/root/elasticsearch/ingest_pipeline/default.yml
@@ -2,6 +2,7 @@
 description: Pipeline for processing sample logs
 processors:
 - set:
+    tag: set_sample_field
     field: sample_field
     value: "1"
 on_failure:
diff --git a/test/packages/bad_agent_version_v3/data_stream/routing_rules/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_agent_version_v3/data_stream/routing_rules/elasticsearch/ingest_pipeline/default.yml
@@ -2,6 +2,7 @@
 description: Pipeline for processing sample logs
 processors:
 - set:
+    tag: set_sample_field
     field: sample_field
     value: "1"
 on_failure:
diff --git a/test/packages/bad_pipeline_tags/LICENSE.txt b/test/packages/bad_pipeline_tags/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.
diff --git a/test/packages/bad_pipeline_tags/changelog.yml b/test/packages/bad_pipeline_tags/changelog.yml
@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "1.0.0"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1 # FIXME Replace with the real PR link
diff --git a/test/packages/bad_pipeline_tags/data_stream/example/agent/stream/stream.yml.hbs b/test/packages/bad_pipeline_tags/data_stream/example/agent/stream/stream.yml.hbs
@@ -0,0 +1,7 @@
+paths:
+{{#each paths as |path i|}}
+  - {{path}}
+{{/each}}
+exclude_files: [".gz$"]
+processors:
+  - add_locale: ~
diff --git a/test/packages/bad_pipeline_tags/data_stream/example/elasticsearch/ingest_pipeline/default.yml b/test/packages/bad_pipeline_tags/data_stream/example/elasticsearch/ingest_pipeline/default.yml
@@ -0,0 +1,22 @@
+---
+description: Pipeline for processing sample logs
+processors:
+- set:
+    field: sample_field
+    value: "1"
+- set:
+    tag: valid_tag
+    field: sample_field
+    value: "1"
+- set:
+    tag: set_sample_field
+    field: sample_field
+    value: "1"
+- set:
+    tag: set_sample_field
+    field: sample_field
+    value: "1"
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'
diff --git a/test/packages/bad_pipeline_tags/data_stream/example/fields/base-fields.yml b/test/packages/bad_pipeline_tags/data_stream/example/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.
diff --git a/test/packages/bad_pipeline_tags/data_stream/example/manifest.yml b/test/packages/bad_pipeline_tags/data_stream/example/manifest.yml
@@ -0,0 +1,13 @@
+title: "Test Data Stream"
+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
diff --git a/test/packages/bad_pipeline_tags/docs/README.md b/test/packages/bad_pipeline_tags/docs/README.md
diff --git a/test/packages/bad_pipeline_tags/img/sample-logo.svg b/test/packages/bad_pipeline_tags/img/sample-logo.svg
diff --git a/test/packages/bad_pipeline_tags/img/sample-screenshot.png b/test/packages/bad_pipeline_tags/img/sample-screenshot.png
diff --git a/test/packages/bad_pipeline_tags/manifest.yml b/test/packages/bad_pipeline_tags/manifest.yml
diff --git a/test/packages/bad_pipeline_tags/validation.yml b/test/packages/bad_pipeline_tags/validation.yml
diff --git a/test/packages/good_v3/data_stream/foo/elasticsearch/ingest_pipeline/default.yml b/test/packages/good_v3/data_stream/foo/elasticsearch/ingest_pipeline/default.yml
diff --git a/test/packages/good_v3/data_stream/routing_rules/elasticsearch/ingest_pipeline/default.yml b/test/packages/good_v3/data_stream/routing_rules/elasticsearch/ingest_pipeline/default.yml

Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ func ValidatePipelineTags(fsys fspath.FS) specerrors.ValidationErrors {`
`39`	`39`	`return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}`
`40`	`40`	`}`
`41`	`41`
`42`		`- if vErrs := validatePipelineTags(&pipeline); len(vErrs) > 0 {`
	`42`	`+ if vErrs := validatePipelineTags(&pipeline, pipelineFile.fullFilePath); len(vErrs) > 0 {`
`43`	`43`	`errors = append(errors, vErrs...)`
`44`	`44`	`}`
`45`	`45`	`}`
`@@ -48,44 +48,44 @@ func ValidatePipelineTags(fsys fspath.FS) specerrors.ValidationErrors {`
`48`	`48`	`return errors`
`49`	`49`	`}`
`50`	`50`
`51`		`-func validatePipelineTags(pipeline *ingestPipeline) specerrors.ValidationErrors {`
	`51`	`+func validatePipelineTags(pipeline *ingestPipeline, filename string) specerrors.ValidationErrors {`
`52`	`52`	`var errors specerrors.ValidationErrors`
`53`	`53`
`54`	`54`	`seen := map[string]struct{}{}`
`55`	`55`	`for _, proc := range pipeline.Processors {`
`56`		`- procErrors := checkPipelineTag(proc, seen)`
	`56`	`+ procErrors := checkPipelineTag(proc, seen, filename)`
`57`	`57`	`errors = append(errors, procErrors...)`
`58`	`58`	`}`
`59`	`59`
`60`	`60`	`return errors`
`61`	`61`	`}`
`62`	`62`
`63`		`-func checkPipelineTag(proc *processor, seen map[string]struct{}) specerrors.ValidationErrors {`
	`63`	`+func checkPipelineTag(proc *processor, seen map[string]struct{}, filename string) specerrors.ValidationErrors {`
`64`	`64`	`var errors specerrors.ValidationErrors`
`65`	`65`
`66`	`66`	`for _, subProc := range proc.OnFailure {`
`67`		`- subErrors := checkPipelineTag(subProc, seen)`
	`67`	`+ subErrors := checkPipelineTag(subProc, seen, filename)`
`68`	`68`	`errors = append(errors, subErrors...)`
`69`	`69`	`}`
`70`	`70`
`71`	`71`	`raw, ok := proc.Attributes["tag"]`
`72`	`72`	`if !ok {`
`73`		`- errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d missing required tag", proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))`
	`73`	`+ errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d missing required tag", filename, proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))`
`74`	`74`	`return errors`
`75`	`75`	`}`
`76`	`76`
`77`	`77`	`tag, ok := raw.(string)`
`78`	`78`	`if !ok {`
`79`		`- errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d has invalid tag value", proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))`
	`79`	`+ errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d has invalid tag value", filename, proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))`
`80`	`80`	`return errors`
`81`	`81`	`}`
`82`	`82`	`if tag == "" {`
`83`		`- errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d has empty tag value", proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))`
	`83`	`+ errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d has empty tag value", filename, proc.Type, proc.position.line), specerrors.CodePipelineTagRequired))`
`84`	`84`	`return errors`
`85`	`85`	`}`
`86`	`86`
`87`	`87`	`if _, dup := seen[tag]; dup {`
`88`		`- errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("%s processor at line %d has duplicate tag value: %q", proc.Type, proc.position.line, tag), specerrors.CodePipelineTagDuplicate))`
	`88`	`+ errors = append(errors, specerrors.NewStructuredError(fmt.Errorf("file %q is invalid: %s processor at line %d has duplicate tag value: %q", filename, proc.Type, proc.position.line, tag), specerrors.CodePipelineTagDuplicate))`
`89`	`89`	`return errors`
`90`	`90`	`}`
`91`	`91`