feat: Add validator for pipeline tags

README.md

@@ -68,7 +68,7 @@ to [JSON Schema](https://json-schema.org/), but with a couple of differences:
 
 Expected package files, e.g. `manifest.yml` themselves have a structure to their contents. This structure is described in specification files using JSON schema (this is point 2. above). These specification files are also written as YAML for readability.
 
-Note that the specification files primarily define the structure (syntax) of a package's contents. To a limited extent they may also define some semantics, e.g. enumeration values for certain fields. Richer semantics, however, will need to be expressed as validation code.
+Note that the specification files primarily define the structure (syntax) of a package's contents. To a limited extent they may also define some semantics, e.g. enumeration values for certain fields. Richer semantics, however, will need to be expressed as [validation code](docs/validations.md).
 
 # Specification Versioning
 

code/go/internal/validator/semantic/types.go

@@ -91,6 +91,50 @@ func (r *runtimeField) UnmarshalYAML(value *yaml.Node) error {
 	return r.unmarshalString(value.Value)
 }
 
+type position struct {
+	file   string
+	line   int
+	column int
+}
+
+func (p position) String() string {
+	return fmt.Sprintf("%s:%d:%d", p.file, p.line, p.column)
+}
+
+type processor struct {
+	Type       string
+	Attributes map[string]any
+	OnFailure  []processor
+
+	position position
+}
+
+func (p *processor) UnmarshalYAML(value *yaml.Node) error {
+	var procMap map[string]struct {
+		Attributes map[string]any `yaml:",inline"`
+		OnFailure  []processor    `yaml:"on_failure"`
+	}
+	if err := value.Decode(&procMap); err != nil {
+		return err
+	}
+
+	for k, v := range procMap {
+		p.Type = k
+		p.Attributes = v.Attributes
+		p.OnFailure = v.OnFailure
+		break
+	}
+
+	p.position.line = value.Line
+	p.position.column = value.Column
+
+	return nil
+}
+
+type ingestPipeline struct {
+	Processors []processor `yaml:"processors"`
+}
+
 type field struct {
 	Name       string `yaml:"name"`
 	Type       string `yaml:"type"`
@@ -111,6 +155,12 @@ type fieldFileMetadata struct {
 	fullFilePath string
 }
 
+type pipelineFileMetadata struct {
+	dataStream   string
+	filePath     string
+	fullFilePath string
+}
+
 type validateFunc func(fileMetadata fieldFileMetadata, f field) specerrors.ValidationErrors
 
 func validateFields(fsys fspath.FS, validate validateFunc) specerrors.ValidationErrors {
@@ -219,6 +269,23 @@ func readFieldsFolder(fsys fspath.FS, fieldsDir string) ([]string, error) {
 	return fieldsFiles, nil
 }
 
+func readPipelinesFolder(fsys fspath.FS, pipelinesDir string) ([]string, error) {
+	var pipelineFiles []string
+	entries, err := fs.ReadDir(fsys, pipelinesDir)
+	if errors.Is(err, os.ErrNotExist) {
+		return []string{}, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("can't list pipelines directory (path: %s): %w", fsys.Path(pipelinesDir), err)
+	}
+
+	for _, v := range entries {
+		pipelineFiles = append(pipelineFiles, path.Join(pipelinesDir, v.Name()))
+	}
+
+	return pipelineFiles, nil
+}
+
 func unmarshalFields(fsys fspath.FS, fieldsPath string) (fields, error) {
 	content, err := fs.ReadFile(fsys, fieldsPath)
 	if err != nil {
@@ -248,3 +315,23 @@ func listDataStreams(fsys fspath.FS) ([]string, error) {
 	}
 	return list, nil
 }
+
+func listPipelineFiles(fsys fspath.FS, dataStream string) ([]pipelineFileMetadata, error) {
+	var pipelineFileMetadatas []pipelineFileMetadata
+
+	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)
+	}
+
+	for _, file := range pipelineFiles {
+		pipelineFileMetadatas = append(pipelineFileMetadatas, pipelineFileMetadata{
+			filePath:     file,
+			fullFilePath: fsys.Path(file),
+			dataStream:   dataStream,
+		})
+	}
+
+	return pipelineFileMetadatas, nil
+}

code/go/internal/validator/semantic/validate_pipeline_tags.go

@@ -0,0 +1,95 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"fmt"
+	"io/fs"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
+)
+
+// ValidatePipelineTags validates ingest pipeline processor tags.
+func ValidatePipelineTags(fsys fspath.FS) specerrors.ValidationErrors {
+	dataStreams, err := listDataStreams(fsys)
+	if err != nil {
+		return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
+	}
+
+	var errors specerrors.ValidationErrors
+	for _, dataStream := range dataStreams {
+		pipelineFiles, err := listPipelineFiles(fsys, dataStream)
+		if err != nil {
+			return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
+		}
+
+		for _, pipelineFile := range pipelineFiles {
+			content, err := fs.ReadFile(fsys, pipelineFile.filePath)
+			if err != nil {
+				return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
+			}
+
+			var pipeline ingestPipeline
+			if err = yaml.Unmarshal(content, &pipeline); err != nil {
+				return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
+			}
+
+			if vErrs := validatePipelineTags(&pipeline, pipelineFile.fullFilePath); len(vErrs) > 0 {
+				errors = append(errors, vErrs...)
+			}
+		}
+	}
+
+	return errors
+}
+
+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, filename)
+		errors = append(errors, procErrors...)
+	}
+
+	return errors
+}
+
+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, filename)
+		errors = append(errors, subErrors...)
+	}
+
+	raw, ok := proc.Attributes["tag"]
+	if !ok {
+		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("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("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.NewStructuredErrorf("file %q is invalid: %s processor at line %d has duplicate tag value: %q", filename, proc.Type, proc.position.line, tag))
+		return errors
+	}
+
+	seen[tag] = struct{}{}
+
+	return errors
+}

code/go/internal/validator/semantic/validate_pipeline_tags_test.go

@@ -0,0 +1,150 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v3"
+)
+
+func TestValidatePipelineTags(t *testing.T) {
+	testCases := []struct {
+		name     string
+		pipeline string
+		errors   []string
+	}{
+		{
+			name: "good",
+			pipeline: `
+processors:
+  - set:
+      tag: set_1
+      field: key1
+      value: value1
+  - set:
+      tag: set_2
+      field: key2
+      value: value2
+      on_failure:
+        - set:
+            tag: onfail_1
+            field: fail_key_1
+            key: fail_value_1
+`,
+		},
+		{
+			name: "missing-tag",
+			pipeline: `
+processors:
+  - set:
+      field: key1
+      value: value1
+      on_failure:
+        - set:
+            tag: onfail_1
+            field: fail_key_1
+            key: fail_value_1
+`,
+			errors: []string{
+				`file "default.yml" is invalid: set processor at line 3 missing required tag (SVR00006)`,
+			},
+		},
+		{
+			name: "missing-tag-nested",
+			pipeline: `
+processors:
+  - set:
+      tag: set_1
+      field: key1
+      value: value1
+      on_failure:
+        - set:
+            field: fail_key_1
+            value: fail_value_1
+`,
+			errors: []string{
+				`file "default.yml" is invalid: set processor at line 8 missing required tag (SVR00006)`,
+			},
+		},
+		{
+			name: "duplicate-tag",
+			pipeline: `
+processors:
+  - set:
+      tag: set_1
+      field: key1
+      value: value1
+  - set:
+      tag: set_1
+      field: key2
+      value: value2
+`,
+			errors: []string{
+				`file "default.yml" is invalid: set processor at line 7 has duplicate tag value: "set_1"`,
+			},
+		},
+		{
+			name: "duplicate-nested-tag",
+			pipeline: `
+processors:
+  - set:
+      tag: set_1
+      field: key1
+      value: value1
+      on_failure:
+        - set:
+            tag: set_1
+            field: fail_key_1
+            value: fail_value_1
+`,
+			errors: []string{
+				`file "default.yml" is invalid: set processor at line 3 has duplicate tag value: "set_1"`,
+			},
+		},
+		{
+			name: "invalid-tag-value",
+			pipeline: `
+processors:
+  - set:
+      tag: 1
+      field: key1
+      value: value1
+`,
+			errors: []string{
+				`file "default.yml" is invalid: set processor at line 3 has invalid tag value (SVR00006)`,
+			},
+		},
+		{
+			name: "empty-tag-value",
+			pipeline: `
+processors:
+  - set:
+      tag: ''
+      field: key1
+      value: value1
+`,
+			errors: []string{
+				`file "default.yml" is invalid: set processor at line 3 has empty tag value (SVR00006)`,
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var pipeline ingestPipeline
+			err := yaml.Unmarshal([]byte(tc.pipeline), &pipeline)
+			require.NoError(t, err)
+
+			errors := validatePipelineTags(&pipeline, "default.yml")
+			assert.Len(t, errors, len(tc.errors))
+			for _, err := range errors {
+				assert.Contains(t, tc.errors, err.Error())
+			}
+		})
+	}
+}

code/go/internal/validator/spec.go

@@ -224,6 +224,7 @@ func (s Spec) rules(pkgType string, rootSpec spectypes.ItemSpec) validationRules
 		{fn: semantic.ValidateInputPackagesPolicyTemplates, types: []string{"input"}},
 		{fn: semantic.ValidateMinimumAgentVersion},
 		{fn: semantic.ValidateIntegrationPolicyTemplates, types: []string{"integration"}},
+		{fn: semantic.ValidatePipelineTags, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
 	}
 
 	var validationRules validationRules

code/go/pkg/specerrors/constants.go

@@ -21,4 +21,5 @@ const (
 	CodeKibanaDanglingObjectsIDs            = "SVR00003"
 	CodeVisualizationByValue                = "SVR00004"
 	CodeMinimumKibanaVersion                = "SVR00005"
+	CodePipelineTagRequired                 = "SVR00006"
 )

code/go/pkg/validator/validator_test.go

@@ -835,6 +835,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\"",
+			},
+		},
 	}
 
 	for pkgName, pipelines := range tests {