Fix: skip fields level while validating

util/dataset.go

@@ -277,19 +277,32 @@ func (d *DataSet) validateRequiredFields() error {
 		return errors.Wrapf(err, "walking through fields files failed")
 	}
 
-	// Flatten all fields
-	for i, fields := range allFields {
-		allFields[i] = fields.Flatten()
+	err = validateRequiredFieldsData(allFields)
+	if err != nil {
+		return errors.Wrapf(err, "validating required fields data failed")
 	}
+	return nil
+}
+
+func validateRequiredFieldsData(allFields []MapStr) error {
+	allFields = flattenFieldsData(allFields)
 
-	// Verify required keys
+	var err error
 	err = requireField(allFields, "dataset.type", "constant_keyword", err)
 	err = requireField(allFields, "dataset.name", "constant_keyword", err)
 	err = requireField(allFields, "dataset.namespace", "constant_keyword", err)
 	err = requireField(allFields, "@timestamp", "date", err)
 	return err
 }
 
+func flattenFieldsData(allFields []MapStr) []MapStr {
+	var modified []MapStr
+	for _, fields := range allFields {
+		modified = append(modified, fields.FlattenWithoutChildrenKey("fields"))
+	}
+	return modified
+}
+
 func requireField(allFields []MapStr, searchedName, expectedType string, validationErr error) error {
 	if validationErr != nil {
 		return validationErr

util/dataset_test.go

@@ -0,0 +1,61 @@
+// 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 util
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestFlattenFieldsData_AlreadyFlat(t *testing.T) {
+	data := []MapStr{
+		{
+			"dataset.name": "name-1",
+			"dataset.type": "type-1",
+		},
+		{
+			"dataset.name": "name-2",
+			"dataset.type": "type-2",
+		},
+	}
+
+	flattened := flattenFieldsData(data)
+	require.Len(t, flattened, 2)
+	require.Equal(t, `{"dataset.name":"name-1","dataset.type":"type-1"}`, flattened[0].String())
+	require.Equal(t, `{"dataset.name":"name-2","dataset.type":"type-2"}`, flattened[1].String())
+}
+
+func TestFlattenFieldsData_Object(t *testing.T) {
+	data := []MapStr{
+		{
+			"dataset": MapStr{
+				"name": "name-1",
+				"type": "type-1",
+			},
+		},
+	}
+
+	flattened := flattenFieldsData(data)
+	require.Len(t, flattened, 1)
+	require.Equal(t, `{"dataset.name":"name-1","dataset.type":"type-1"}`, flattened[0].String())
+}
+
+func TestFlattenFieldsData_ObjectWithFields(t *testing.T) {
+	data := []MapStr{
+		{
+			"dataset": MapStr{
+				"fields": MapStr{
+					"name": "name-1",
+					"type": "type-1",
+				},
+			},
+		},
+	}
+
+	flattened := flattenFieldsData(data)
+	require.Len(t, flattened, 1)
+	require.Equal(t, `{"dataset.name":"name-1","dataset.type":"type-1"}`, flattened[0].String())
+}

util/mapstr.go

@@ -207,6 +207,30 @@ func flatten(prefix string, in, out MapStr) MapStr {
 	return out
 }
 
+func (m MapStr) FlattenWithoutChildrenKey(childrenKey string) MapStr {
+	return flattenWithoutChildrenKey("", childrenKey, m, MapStr{})
+}
+
+func flattenWithoutChildrenKey(prefix, childrenKey string, in, out MapStr) MapStr {
+	for k, v := range in {
+		var fullKey string
+		if prefix == "" {
+			fullKey = k
+		} else if k == childrenKey {
+			fullKey = prefix
+		} else {
+			fullKey = prefix + "." + k
+		}
+
+		if m, ok := tryToMapStr(v); ok {
+			flattenWithoutChildrenKey(fullKey, childrenKey, m, out)
+		} else {
+			out[fullKey] = v
+		}
+	}
+	return out
+}
+
 // toMapStr performs a type assertion on v and returns a MapStr. v can be either
 // a MapStr or a map[string]interface{}. If it's any other type or nil then
 // an error is returned.