Validate that fields are defined only once per data stream (#309)

Author: jsoriano

code/go/internal/validator/semantic/validate_unique_fields.go

@@ -0,0 +1,59 @@
+// 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 (
+	"sort"
+	"strings"
+
+	ve "github.com/elastic/package-spec/code/go/internal/errors"
+	"github.com/elastic/package-spec/code/go/internal/fspath"
+	"github.com/pkg/errors"
+)
+
+// ValidateUniqueFields verifies that any field is defined only once on each data stream.
+func ValidateUniqueFields(fsys fspath.FS) ve.ValidationErrors {
+	// data_stream -> field -> files
+	fields := make(map[string]map[string][]string)
+
+	countField := func(fieldsFile string, f field) ve.ValidationErrors {
+		if len(f.Fields) > 0 {
+			// Don't count groups
+			return nil
+		}
+
+		dataStream, err := dataStreamFromFieldsPath(fsys.Path(), fieldsFile)
+		if err != nil {
+			return ve.ValidationErrors{err}
+		}
+
+		dsMap, found := fields[dataStream]
+		if !found {
+			dsMap = make(map[string][]string)
+			fields[dataStream] = dsMap
+		}
+		dsMap[f.Name] = append(dsMap[f.Name], fieldsFile)
+		return nil
+	}
+
+	err := validateFields(fsys, countField)
+	if err != nil {
+		return err
+	}
+
+	var errs ve.ValidationErrors
+	for dataStream, dataStreamFields := range fields {
+		for field, files := range dataStreamFields {
+			if len(files) > 1 {
+				sort.Strings(files)
+				errs = append(errs,
+					errors.Errorf("field %q is defined multiple times for data stream %q, found in: %s",
+						field, dataStream, strings.Join(files, ", ")))
+			}
+		}
+
+	}
+	return errs
+}

code/go/internal/validator/spec.go

@@ -71,6 +71,7 @@ func (s Spec) ValidatePackage(pkg Package) ve.ValidationErrors {
 		semantic.ValidatePrerelease,
 		semantic.ValidateFieldGroups,
 		semantic.ValidateFieldsLimits(rootSpec.Limits.FieldsPerDataStreamLimit),
+		semantic.ValidateUniqueFields,
 		semantic.ValidateDimensionFields,
 		semantic.ValidateRequiredFields,
 	}

code/go/pkg/validator/validator_test.go

@@ -264,6 +264,30 @@ func TestValidateVersionIntegrity(t *testing.T) {
 	}
 }
 
+func TestValidateDuplicatedFields(t *testing.T) {
+	tests := map[string]string{
+		"bad_duplicated_fields": "field \"event.dataset\" is defined multiple times for data stream \"wrong\", found in: ../../../../test/packages/bad_duplicated_fields/data_stream/wrong/fields/base-fields.yml, ../../../../test/packages/bad_duplicated_fields/data_stream/wrong/fields/ecs.yml",
+	}
+
+	for pkgName, expectedErrorMessage := range tests {
+		t.Run(pkgName, func(t *testing.T) {
+			errs := ValidateFromPath(filepath.Join("..", "..", "..", "..", "test", "packages", pkgName))
+			require.Error(t, errs)
+			vErrs, ok := errs.(errors.ValidationErrors)
+			require.True(t, ok)
+
+			assert.Len(t, vErrs, 1)
+
+			var errMessages []string
+			for _, vErr := range vErrs {
+				errMessages = append(errMessages, vErr.Error())
+			}
+			require.Contains(t, errMessages, expectedErrorMessage)
+		})
+	}
+
+}
+
 func requireErrorMessage(t *testing.T, pkgName string, invalidItemsPerFolder map[string][]string, expectedErrorMessage string) {
 	pkgRootPath := filepath.Join("..", "..", "..", "..", "test", "packages", pkgName)
 

test/packages/bad_duplicated_fields/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/309

test/packages/bad_duplicated_fields/data_stream/good/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/bad_duplicated_fields/data_stream/good/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,10 @@
+---
+description: Pipeline for processing sample logs
+processors:
+- set:
+    field: sample_field
+    value: "1"
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'

test/packages/bad_duplicated_fields/data_stream/good/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/bad_duplicated_fields/data_stream/good/manifest.yml

@@ -0,0 +1,13 @@
+title: "Valid 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

test/packages/bad_duplicated_fields/data_stream/wrong/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/bad_duplicated_fields/data_stream/wrong/elasticsearch/ingest_pipeline/default.yml

@@ -0,0 +1,10 @@
+---
+description: Pipeline for processing sample logs
+processors:
+- set:
+    field: sample_field
+    value: "1"
+on_failure:
+- set:
+    field: error.message
+    value: '{{ _ingest.on_failure_message }}'