Include the validation of transform fields (#1027)

Include the validation of the fields defined in transforms as
part of the rest of the fields (data streams and input packages).
Thereby, fields in transforms will run the same validations as the
other fields with one exception about the required fields. It is not
required to defined `data_stream.{type,namespace,dataset}` and 
`@timestamp` fields in transforms.

Author: mrodm

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

@@ -107,6 +107,7 @@ type field struct {
 
 type fieldFileMetadata struct {
 	dataStream   string
+	transform    string
 	filePath     string
 	fullFilePath string
 }
@@ -180,6 +181,7 @@ func listFieldsFiles(fsys fspath.FS) ([]fieldFileMetadata, error) {
 					filePath:     file,
 					fullFilePath: fsys.Path(file),
 					dataStream:   dataStream,
+					transform:    "",
 				})
 		}
 	}
@@ -197,9 +199,35 @@ func listFieldsFiles(fsys fspath.FS) ([]fieldFileMetadata, error) {
 				filePath:     file,
 				fullFilePath: fsys.Path(file),
 				dataStream:   "",
+				transform:    "",
 			})
 	}
 
+	// transform definitions
+	transforms, err := listTransforms(fsys)
+	if err != nil {
+		return nil, err
+	}
+
+	for _, transform := range transforms {
+		fieldsDir := path.Join("elasticsearch", "transform", transform, "fields")
+		transformFieldsFiles, err := readFieldsFolder(fsys, fieldsDir)
+		if err != nil {
+			return nil, fmt.Errorf("cannot read fields file from integration packages: %w", err)
+		}
+
+		for _, file := range transformFieldsFiles {
+			fieldsFilesMetadata = append(
+				fieldsFilesMetadata,
+				fieldFileMetadata{
+					filePath:     file,
+					fullFilePath: fsys.Path(file),
+					dataStream:   "",
+					transform:    transform,
+				})
+		}
+	}
+
 	return fieldsFilesMetadata, nil
 }
 
@@ -248,3 +276,20 @@ func listDataStreams(fsys fspath.FS) ([]string, error) {
 	}
 	return list, nil
 }
+
+func listTransforms(fsys fspath.FS) ([]string, error) {
+	transformDirectory := path.Join("elasticsearch", "transform")
+	transforms, err := fs.ReadDir(fsys, transformDirectory)
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("can't list transforms directory: %w", err)
+	}
+
+	list := make([]string, len(transforms))
+	for i, transform := range transforms {
+		list[i] = transform.Name()
+	}
+	return list, nil
+}

code/go/internal/validator/semantic/types_test.go

@@ -25,87 +25,103 @@ func TestListFieldsFiles(t *testing.T) {
 		{
 			pkgName: "good_v2",
 			expected: []fieldFileMetadata{
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/foo/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/foo/fields/base-fields.yml"),
 					dataStream:   "foo",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/foo/fields/external-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/foo/fields/external-fields.yml"),
 					dataStream:   "foo",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/foo/fields/some_fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/foo/fields/some_fields.yml"),
 					dataStream:   "foo",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/hidden_data_stream/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/hidden_data_stream/fields/base-fields.yml"),
 					dataStream:   "hidden_data_stream",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/hidden_data_stream/fields/some_fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/hidden_data_stream/fields/some_fields.yml"),
 					dataStream:   "hidden_data_stream",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/ilm_policy/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/ilm_policy/fields/base-fields.yml"),
 					dataStream:   "ilm_policy",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/ilm_policy/fields/some_fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/ilm_policy/fields/some_fields.yml"),
 					dataStream:   "ilm_policy",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/k8s_data_stream/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/k8s_data_stream/fields/base-fields.yml"),
 					dataStream:   "k8s_data_stream",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/k8s_data_stream_no_definitions/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/k8s_data_stream_no_definitions/fields/base-fields.yml"),
 					dataStream:   "k8s_data_stream_no_definitions",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/pe/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/pe/fields/base-fields.yml"),
 					dataStream:   "pe",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/pe/fields/some_fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/pe/fields/some_fields.yml"),
 					dataStream:   "pe",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/routing_rules/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/routing_rules/fields/base-fields.yml"),
 					dataStream:   "routing_rules",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/skipped_tests/fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/skipped_tests/fields/base-fields.yml"),
 					dataStream:   "skipped_tests",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "data_stream/skipped_tests/fields/some_fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/data_stream/skipped_tests/fields/some_fields.yml"),
 					dataStream:   "skipped_tests",
 				},
+				// transforms
+				{
+					filePath:     "elasticsearch/transform/metadata_current/fields/fields.yml",
+					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/elasticsearch/transform/metadata_current/fields/fields.yml"),
+					transform:    "metadata_current",
+				},
+				{
+					filePath:     "elasticsearch/transform/metadata_united/fields/base-fields.yml",
+					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/elasticsearch/transform/metadata_united/fields/base-fields.yml"),
+					transform:    "metadata_united",
+				},
+				{
+					filePath:     "elasticsearch/transform/metadata_united/fields/fields.yml",
+					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_v2/elasticsearch/transform/metadata_united/fields/fields.yml"),
+					transform:    "metadata_united",
+				},
 			},
 		},
 		{
 			pkgName: "good_input",
 			expected: []fieldFileMetadata{
-				fieldFileMetadata{
+				{
 					filePath:     "fields/base-fields.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_input/fields/base-fields.yml"),
 					dataStream:   "",
 				},
-				fieldFileMetadata{
+				{
 					filePath:     "fields/input.yml",
 					fullFilePath: filepath.FromSlash("../../../../../test/packages/good_input/fields/input.yml"),
 					dataStream:   "",

code/go/internal/validator/semantic/validate_fields_limits.go

@@ -18,12 +18,20 @@ func ValidateFieldsLimits(limit int) func(fspath.FS) specerrors.ValidationErrors
 
 func validateFieldsLimits(fsys fspath.FS, limit int) specerrors.ValidationErrors {
 	counts := make(map[string]int)
+	// Created a new map to avoid collisions with data stream names
+	transformCounts := make(map[string]int)
 	countField := func(metadata fieldFileMetadata, f field) specerrors.ValidationErrors {
 		if len(f.Fields) > 0 {
 			// Don't count groups
 			return nil
 		}
 
+		if metadata.transform != "" {
+			count := transformCounts[metadata.transform]
+			transformCounts[metadata.transform] = count + 1
+			return nil
+		}
+
 		count := counts[metadata.dataStream]
 		counts[metadata.dataStream] = count + 1
 		return nil
@@ -33,11 +41,19 @@ func validateFieldsLimits(fsys fspath.FS, limit int) specerrors.ValidationErrors
 	if err != nil {
 		return err
 	}
-
 	var errs specerrors.ValidationErrors
 	for id, count := range counts {
 		if count > limit {
-			errs = append(errs, specerrors.NewStructuredErrorf("data stream %s has more than %d fields (%d)", id, limit, count))
+			if id != "" {
+				errs = append(errs, specerrors.NewStructuredErrorf("data stream %s has more than %d fields (%d)", id, limit, count))
+			} else {
+				errs = append(errs, specerrors.NewStructuredErrorf("input package has more than %d fields (%d)", limit, count))
+			}
+		}
+	}
+	for id, count := range transformCounts {
+		if count > limit {
+			errs = append(errs, specerrors.NewStructuredErrorf("transform %s has more than %d fields (%d)", id, limit, count))
 		}
 	}
 	return errs

code/go/internal/validator/semantic/validate_fields_limits_test.go

@@ -0,0 +1,105 @@
+// 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 (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+)
+
+func TestValidateFieldsLimits(t *testing.T) {
+	t.Run("one data stream with too many fields", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.MkdirAll(filepath.Join(d, "data_stream", "test", "fields"), 0o755)
+		require.NoError(t, err)
+		err = os.MkdirAll(filepath.Join(d, "data_stream", "foo", "fields"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "data_stream", "foo", "fields", "fields.yml"), []byte(`
+- name: field1
+  type: keyword
+- name: field2
+  type: keyword
+- name: field3
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "data_stream", "foo", "fields", "more-fields.yml"), []byte(`
+- name: field1
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "data_stream", "test", "fields", "fields.yml"), []byte(`
+- name: field1
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := validateFieldsLimits(fspath.DirFS(d), 1)
+		require.Len(t, errs, 1)
+		assert.EqualError(t, errs[0], "data stream foo has more than 1 fields (4)")
+	})
+	t.Run("one transform with too many fields", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.MkdirAll(filepath.Join(d, "elasticsearch", "transform", "foo", "fields"), 0o755)
+		require.NoError(t, err)
+		err = os.MkdirAll(filepath.Join(d, "elasticsearch", "transform", "bar", "fields"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "elasticsearch", "transform", "foo", "fields", "fields.yml"), []byte(`
+- name: field1
+  type: keyword
+- name: field2
+  type: keyword
+- name: field3
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "elasticsearch", "transform", "foo", "fields", "more-fields.yml"), []byte(`
+- name: field1
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "elasticsearch", "transform", "bar", "fields", "fields.yml"), []byte(`
+- name: field1
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := validateFieldsLimits(fspath.DirFS(d), 1)
+		require.Len(t, errs, 1)
+		assert.EqualError(t, errs[0], "transform foo has more than 1 fields (4)")
+	})
+	t.Run("input package with too many fields", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.MkdirAll(filepath.Join(d, "fields"), 0o755)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "fields", "fields.yml"), []byte(`
+- name: field1
+  type: keyword
+- name: field2
+  type: keyword
+- name: field3
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+		err = os.WriteFile(filepath.Join(d, "fields", "more-fields.yml"), []byte(`
+- name: field1
+  type: keyword
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := validateFieldsLimits(fspath.DirFS(d), 1)
+		require.Len(t, errs, 1)
+		assert.EqualError(t, errs[0], "input package has more than 1 fields (4)")
+	})
+}