Skip to content

Commit 8b522f7

Browse files
taylor-swansontaylor-swanson
authored
feat: Add validator for pipeline tags (#1010)
Add validation rule SVR00006 to verify all ingest pipeline processors have a valid tag (excludes global pipeline on_failure). Add validation to verify all ingest pipeline processor tags are unique (excludes global pipeline on_failure).
1 parent 322b9b2 commit 8b522f7

File tree

35 files changed

+718
-2
lines changed

35 files changed

+718
-2
lines changed

‎README.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ to [JSON Schema](https://json-schema.org/), but with a couple of differences:
6868

6969
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.
7070

71-
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.
71+
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).
7272

7373
# Specification Versioning
7474

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

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,50 @@ func (r *runtimeField) UnmarshalYAML(value *yaml.Node) error {
9191
return r.unmarshalString(value.Value)
9292
}
9393

94+
type position struct {
95+
file string
96+
line int
97+
column int
98+
}
99+
100+
func (p position) String() string {
101+
return fmt.Sprintf("%s:%d:%d", p.file, p.line, p.column)
102+
}
103+
104+
type processor struct {
105+
Type string
106+
Attributes map[string]any
107+
OnFailure []processor
108+
109+
position position
110+
}
111+
112+
func (p *processor) UnmarshalYAML(value *yaml.Node) error {
113+
var procMap map[string]struct {
114+
Attributes map[string]any `yaml:",inline"`
115+
OnFailure []processor `yaml:"on_failure"`
116+
}
117+
if err := value.Decode(&procMap); err != nil {
118+
return err
119+
}
120+
121+
for k, v := range procMap {
122+
p.Type = k
123+
p.Attributes = v.Attributes
124+
p.OnFailure = v.OnFailure
125+
break
126+
}
127+
128+
p.position.line = value.Line
129+
p.position.column = value.Column
130+
131+
return nil
132+
}
133+
134+
type ingestPipeline struct {
135+
Processors []processor `yaml:"processors"`
136+
}
137+
94138
type field struct {
95139
Name string `yaml:"name"`
96140
Type string `yaml:"type"`
@@ -112,6 +156,12 @@ type fieldFileMetadata struct {
112156
fullFilePath string
113157
}
114158

159+
type pipelineFileMetadata struct {
160+
dataStream string
161+
filePath string
162+
fullFilePath string
163+
}
164+
115165
type validateFunc func(fileMetadata fieldFileMetadata, f field) specerrors.ValidationErrors
116166

117167
func validateFields(fsys fspath.FS, validate validateFunc) specerrors.ValidationErrors {
@@ -247,6 +297,23 @@ func readFieldsFolder(fsys fspath.FS, fieldsDir string) ([]string, error) {
247297
return fieldsFiles, nil
248298
}
249299

300+
func readPipelinesFolder(fsys fspath.FS, pipelinesDir string) ([]string, error) {
301+
var pipelineFiles []string
302+
entries, err := fs.ReadDir(fsys, pipelinesDir)
303+
if errors.Is(err, os.ErrNotExist) {
304+
return []string{}, nil
305+
}
306+
if err != nil {
307+
return nil, fmt.Errorf("can't list pipelines directory (path: %s): %w", fsys.Path(pipelinesDir), err)
308+
}
309+
310+
for _, v := range entries {
311+
pipelineFiles = append(pipelineFiles, path.Join(pipelinesDir, v.Name()))
312+
}
313+
314+
return pipelineFiles, nil
315+
}
316+
250317
func unmarshalFields(fsys fspath.FS, fieldsPath string) (fields, error) {
251318
content, err := fs.ReadFile(fsys, fieldsPath)
252319
if err != nil {
@@ -292,4 +359,25 @@ func listTransforms(fsys fspath.FS) ([]string, error) {
292359
list[i] = transform.Name()
293360
}
294361
return list, nil
362+
363+
}
364+
365+
func listPipelineFiles(fsys fspath.FS, dataStream string) ([]pipelineFileMetadata, error) {
366+
var pipelineFileMetadatas []pipelineFileMetadata
367+
368+
ingestPipelineDir := path.Join(dataStreamDir, dataStream, "elasticsearch", "ingest_pipeline")
369+
pipelineFiles, err := readPipelinesFolder(fsys, ingestPipelineDir)
370+
if err != nil {
371+
return nil, fmt.Errorf("cannot read pipeline files from integration package: %w", err)
372+
}
373+
374+
for _, file := range pipelineFiles {
375+
pipelineFileMetadatas = append(pipelineFileMetadatas, pipelineFileMetadata{
376+
filePath: file,
377+
fullFilePath: fsys.Path(file),
378+
dataStream: dataStream,
379+
})
380+
}
381+
382+
return pipelineFileMetadatas, nil
295383
}

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

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
2+
// or more contributor license agreements. Licensed under the Elastic License;
3+
// you may not use this file except in compliance with the Elastic License.
4+
5+
package semantic
6+
7+
import (
8+
"fmt"
9+
"io/fs"
10+
11+
"gopkg.in/yaml.v3"
12+
13+
"github.com/elastic/package-spec/v3/code/go/internal/fspath"
14+
"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
15+
)
16+
17+
// ValidatePipelineTags validates ingest pipeline processor tags.
18+
func ValidatePipelineTags(fsys fspath.FS) specerrors.ValidationErrors {
19+
dataStreams, err := listDataStreams(fsys)
20+
if err != nil {
21+
return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
22+
}
23+
24+
var errors specerrors.ValidationErrors
25+
for _, dataStream := range dataStreams {
26+
pipelineFiles, err := listPipelineFiles(fsys, dataStream)
27+
if err != nil {
28+
return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
29+
}
30+
31+
for _, pipelineFile := range pipelineFiles {
32+
content, err := fs.ReadFile(fsys, pipelineFile.filePath)
33+
if err != nil {
34+
return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
35+
}
36+
37+
var pipeline ingestPipeline
38+
if err = yaml.Unmarshal(content, &pipeline); err != nil {
39+
return specerrors.ValidationErrors{specerrors.NewStructuredError(err, specerrors.UnassignedCode)}
40+
}
41+
42+
if vErrs := validatePipelineTags(&pipeline, pipelineFile.fullFilePath); len(vErrs) > 0 {
43+
errors = append(errors, vErrs...)
44+
}
45+
}
46+
}
47+
48+
return errors
49+
}
50+
51+
func validatePipelineTags(pipeline *ingestPipeline, filename string) specerrors.ValidationErrors {
52+
var errors specerrors.ValidationErrors
53+
54+
seen := map[string]struct{}{}
55+
for _, proc := range pipeline.Processors {
56+
procErrors := checkPipelineTag(&proc, seen, filename)
57+
errors = append(errors, procErrors...)
58+
}
59+
60+
return errors
61+
}
62+
63+
func checkPipelineTag(proc *processor, seen map[string]struct{}, filename string) specerrors.ValidationErrors {
64+
var errors specerrors.ValidationErrors
65+
66+
for _, subProc := range proc.OnFailure {
67+
subErrors := checkPipelineTag(&subProc, seen, filename)
68+
errors = append(errors, subErrors...)
69+
}
70+
71+
raw, ok := proc.Attributes["tag"]
72+
if !ok {
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+
return errors
75+
}
76+
77+
tag, ok := raw.(string)
78+
if !ok {
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+
return errors
81+
}
82+
if tag == "" {
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+
return errors
85+
}
86+
87+
if _, dup := seen[tag]; dup {
88+
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))
89+
return errors
90+
}
91+
92+
seen[tag] = struct{}{}
93+
94+
return errors
95+
}

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

Lines changed: 150 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,150 @@
1+
// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
2+
// or more contributor license agreements. Licensed under the Elastic License;
3+
// you may not use this file except in compliance with the Elastic License.
4+
5+
package semantic
6+
7+
import (
8+
"testing"
9+
10+
"github.com/stretchr/testify/assert"
11+
"github.com/stretchr/testify/require"
12+
"gopkg.in/yaml.v3"
13+
)
14+
15+
func TestValidatePipelineTags(t *testing.T) {
16+
testCases := []struct {
17+
name string
18+
pipeline string
19+
errors []string
20+
}{
21+
{
22+
name: "good",
23+
pipeline: `
24+
processors:
25+
- set:
26+
tag: set_1
27+
field: key1
28+
value: value1
29+
- set:
30+
tag: set_2
31+
field: key2
32+
value: value2
33+
on_failure:
34+
- set:
35+
tag: onfail_1
36+
field: fail_key_1
37+
key: fail_value_1
38+
`,
39+
},
40+
{
41+
name: "missing-tag",
42+
pipeline: `
43+
processors:
44+
- set:
45+
field: key1
46+
value: value1
47+
on_failure:
48+
- set:
49+
tag: onfail_1
50+
field: fail_key_1
51+
key: fail_value_1
52+
`,
53+
errors: []string{
54+
`file "default.yml" is invalid: set processor at line 3 missing required tag (SVR00006)`,
55+
},
56+
},
57+
{
58+
name: "missing-tag-nested",
59+
pipeline: `
60+
processors:
61+
- set:
62+
tag: set_1
63+
field: key1
64+
value: value1
65+
on_failure:
66+
- set:
67+
field: fail_key_1
68+
value: fail_value_1
69+
`,
70+
errors: []string{
71+
`file "default.yml" is invalid: set processor at line 8 missing required tag (SVR00006)`,
72+
},
73+
},
74+
{
75+
name: "duplicate-tag",
76+
pipeline: `
77+
processors:
78+
- set:
79+
tag: set_1
80+
field: key1
81+
value: value1
82+
- set:
83+
tag: set_1
84+
field: key2
85+
value: value2
86+
`,
87+
errors: []string{
88+
`file "default.yml" is invalid: set processor at line 7 has duplicate tag value: "set_1"`,
89+
},
90+
},
91+
{
92+
name: "duplicate-nested-tag",
93+
pipeline: `
94+
processors:
95+
- set:
96+
tag: set_1
97+
field: key1
98+
value: value1
99+
on_failure:
100+
- set:
101+
tag: set_1
102+
field: fail_key_1
103+
value: fail_value_1
104+
`,
105+
errors: []string{
106+
`file "default.yml" is invalid: set processor at line 3 has duplicate tag value: "set_1"`,
107+
},
108+
},
109+
{
110+
name: "invalid-tag-value",
111+
pipeline: `
112+
processors:
113+
- set:
114+
tag: 1
115+
field: key1
116+
value: value1
117+
`,
118+
errors: []string{
119+
`file "default.yml" is invalid: set processor at line 3 has invalid tag value (SVR00006)`,
120+
},
121+
},
122+
{
123+
name: "empty-tag-value",
124+
pipeline: `
125+
processors:
126+
- set:
127+
tag: ''
128+
field: key1
129+
value: value1
130+
`,
131+
errors: []string{
132+
`file "default.yml" is invalid: set processor at line 3 has empty tag value (SVR00006)`,
133+
},
134+
},
135+
}
136+
137+
for _, tc := range testCases {
138+
t.Run(tc.name, func(t *testing.T) {
139+
var pipeline ingestPipeline
140+
err := yaml.Unmarshal([]byte(tc.pipeline), &pipeline)
141+
require.NoError(t, err)
142+
143+
errors := validatePipelineTags(&pipeline, "default.yml")
144+
assert.Len(t, errors, len(tc.errors))
145+
for _, err := range errors {
146+
assert.Contains(t, tc.errors, err.Error())
147+
}
148+
})
149+
}
150+
}

‎code/go/internal/validator/spec.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -224,6 +224,7 @@ func (s Spec) rules(pkgType string, rootSpec spectypes.ItemSpec) validationRules
224224
{fn: semantic.ValidateInputPackagesPolicyTemplates, types: []string{"input"}},
225225
{fn: semantic.ValidateMinimumAgentVersion},
226226
{fn: semantic.ValidateIntegrationPolicyTemplates, types: []string{"integration"}},
227+
{fn: semantic.ValidatePipelineTags, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
227228
}
228229

229230
var validationRules validationRules

‎code/go/pkg/specerrors/constants.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,4 +21,5 @@ const (
2121
CodeKibanaDanglingObjectsIDs = "SVR00003"
2222
CodeVisualizationByValue = "SVR00004"
2323
CodeMinimumKibanaVersion = "SVR00005"
24+
CodePipelineTagRequired = "SVR00006"
2425
)

‎code/go/pkg/validator/validator_test.go‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -857,6 +857,12 @@ func TestValidateIngestPipelines(t *testing.T) {
857857
"field processors.2.remove.if: rename \"message\" to \"event.original\" processor requires remove \"message\" processor with if: 'ctx.event?.original != null' (JSE00001)",
858858
},
859859
},
860+
"bad_pipeline_tags": {
861+
"example": []string{
862+
"set processor at line 4 missing required tag (SVR00006)",
863+
"set processor at line 15 has duplicate tag value: \"set_sample_field\"",
864+
},
865+
},
860866
}
861867

862868
for pkgName, pipelines := range tests {

0 commit comments

Comments
 (0)