Validate fields mount name and mount path in Dataset

pkg/common/env_names.go

@@ -28,4 +28,6 @@ const (
 	EnvEnableRuntimeInfoCache = "ENABLE_RUNTIMEINFO_CACHE"
 
 	EnvRuntimeInfoCacheTTL = "RUNTIMEINFO_CACHE_TTL"
+
+	EnvEnableMountValidation = "ENABLE_MOUNT_VALIDATION"
 )

pkg/controllers/v1alpha1/dataset/dataset_controller.go

@@ -18,12 +18,8 @@ import (
 	"context"
 	"errors"
 	"reflect"
-	"strings"
 	"time"
 
-	"k8s.io/apimachinery/pkg/util/validation"
-	"k8s.io/apimachinery/pkg/util/validation/field"
-
 	"github.com/fluid-cloudnative/fluid/pkg/common"
 	"github.com/fluid-cloudnative/fluid/pkg/controllers/deploy"
 	"github.com/fluid-cloudnative/fluid/pkg/ddc/base"
@@ -41,6 +37,7 @@ import (
 
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
 	"github.com/fluid-cloudnative/fluid/pkg/utils"
+	fluidvalidation "github.com/fluid-cloudnative/fluid/pkg/utils/validation"
 )
 
 const (
@@ -124,9 +121,12 @@ func (r *DatasetReconciler) reconcileDataset(ctx reconcileRequestContext, needRe
 	log := ctx.Log.WithName("reconcileDataset")
 	log.V(1).Info("process the dataset", "dataset", ctx.Dataset)
 
-	// 0. Validate name is prefixed with a number such as "20-hbase".
-	if errs := validation.IsDNS1035Label(ctx.Dataset.ObjectMeta.Name); len(ctx.Dataset.ObjectMeta.Name) > 0 && len(errs) > 0 {
-		err := field.Invalid(field.NewPath("metadata").Child("name"), ctx.Dataset.ObjectMeta.Name, strings.Join(errs, ","))
+	// 0. Validate the dataset.
+	// Users can set this environment variable to 'false' to skip the validation of the mount field in dataset
+	// Default is true
+	if err := fluidvalidation.IsValidDataset(
+		ctx.Dataset, utils.GetBoolValueFromEnv(common.EnvEnableMountValidation, true),
+	); err != nil {
 		ctx.Log.Error(err, "Failed to create dataset", "DatasetCreateError", ctx)
 		r.Recorder.Eventf(&ctx.Dataset, v1.EventTypeWarning, common.ErrorCreateDataset, "Failed to create dataset because err: %v", err)
 		return utils.RequeueIfError(err)

pkg/utils/validation/dataset.go

@@ -0,0 +1,55 @@
+/*
+Copyright 2024 The Fluid Author.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package validation
+
+import (
+	"strings"
+
+	"github.com/fluid-cloudnative/fluid/api/v1alpha1"
+	"k8s.io/apimachinery/pkg/util/validation"
+	"k8s.io/apimachinery/pkg/util/validation/field"
+)
+
+func IsValidDataset(dataset v1alpha1.Dataset, enableMountValidation bool) error {
+	if errs := validation.IsDNS1035Label(dataset.ObjectMeta.Name); len(dataset.ObjectMeta.Name) > 0 && len(errs) > 0 {
+		return field.Invalid(field.NewPath("metadata").Child("name"), dataset.ObjectMeta.Name, strings.Join(errs, ","))
+	}
+
+	// 0.1 Validate the mount name and mount path
+	// Users can set the environment variable to 'false' to disable this validation
+	// Default is true
+	if !enableMountValidation {
+		return nil
+	}
+	for _, mount := range dataset.Spec.Mounts {
+		// The field mount.Name and mount.Path is optional
+		// Empty name or path is allowed
+		if len(mount.Name) != 0 {
+			// If users set the mount.Name, it should comply with the DNS1035 rule.
+			if errs := validation.IsDNS1035Label(mount.Name); len(errs) > 0 {
+				return field.Invalid(field.NewPath("spec").Child("mounts").Child("name"), mount.Name, strings.Join(errs, ","))
+			}
+		}
+		if len(mount.Path) != 0 {
+			// If users set the mount.Path, check it.
+			if err := IsValidMountPath(mount.Path); err != nil {
+				return field.Invalid(field.NewPath("spec").Child("mounts").Child("path"), mount.Path, err.Error())
+			}
+		}
+	}
+	return nil
+}

pkg/utils/validation/dataset_test.go

@@ -0,0 +1,182 @@
+/*
+Copyright 2024 The Fluid Author.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package validation
+
+import (
+	"testing"
+
+	"github.com/fluid-cloudnative/fluid/api/v1alpha1"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+const mountPoint1 string = "https://mirrors.bit.edu.cn/apache/spark/"
+const validMountName1 string = "spark"
+
+const mountPoint2 string = "https://mirrors.bit.edu.cn/apache/flink/"
+const validMountName2 string = "flink"
+
+const validMountPath1 string = "/test"
+const validMountPath2 string = "mnt/test"
+
+func TestIsValidDatasetWithValidDataset(t *testing.T) {
+	type testCase struct {
+		name                  string
+		input                 v1alpha1.Dataset
+		enableMountValidation bool
+	}
+
+	testCases := []testCase{
+		{
+			name:                  "validDatasetWithSingleMount",
+			enableMountValidation: true,
+			input: v1alpha1.Dataset{
+				ObjectMeta: v1.ObjectMeta{
+					Name: "demo",
+				},
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Name:       validMountName1,
+							Path:       validMountPath1,
+						},
+					},
+				},
+			},
+		},
+		{
+			name:                  "validDatasetWithMultiMount",
+			enableMountValidation: true,
+			input: v1alpha1.Dataset{
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Name:       validMountName1,
+						},
+						{
+							MountPoint: mountPoint2,
+							Name:       validMountName2,
+							Path:       validMountPath2,
+						},
+					},
+				},
+			},
+		},
+		{
+			name:                  "validDatasetWithDisableMountValidation",
+			enableMountValidation: false,
+			input: v1alpha1.Dataset{
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Path:       "/${TEST}",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range testCases {
+		got := IsValidDataset(test.input, test.enableMountValidation)
+		if got != nil {
+			t.Errorf("testcase %s failed, expect no error happened, but got an error: %s", test.name, got.Error())
+		}
+	}
+}
+
+func TestIsValidDatasetWithInvalidDataset(t *testing.T) {
+	type testCase struct {
+		name  string
+		input v1alpha1.Dataset
+	}
+
+	testCases := []testCase{
+		{
+			name: "invalidDatasetMountName",
+			input: v1alpha1.Dataset{
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Name:       "$(cat /etc/passwd > /test.txt)",
+							Path:       validMountPath1,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "invalidDatasetName",
+			input: v1alpha1.Dataset{
+				ObjectMeta: v1.ObjectMeta{
+					Name: "20-hbase",
+				},
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Name:       validMountName1,
+							Path:       validMountPath1,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "invalidDatasetMountPath",
+			input: v1alpha1.Dataset{
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Name:       validMountName1,
+							Path:       "/$(cat /etc/passwd > /test.txt)",
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "invalidDatasetMountPathInSecondMount",
+			input: v1alpha1.Dataset{
+				Spec: v1alpha1.DatasetSpec{
+					Mounts: []v1alpha1.Mount{
+						{
+							MountPoint: mountPoint1,
+							Name:       validMountName1,
+						},
+						{
+							MountPoint: mountPoint2,
+							Name:       validMountName2,
+							Path:       "/test/$(cat /etc/passwd > /test.txt)",
+						},
+					},
+				},
+			},
+		},
+	}
+
+	for _, test := range testCases {
+		got := IsValidDataset(test.input, true)
+		if got == nil {
+			t.Errorf("testcase %s failed, expect an error happened, but got no error", test.name)
+		}
+	}
+}

pkg/utils/validation/validation.go → pkg/utils/validation/path.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2023 The Fluid Authors.
+Copyright 2024 The Fluid Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -24,6 +24,22 @@ import (
 	"k8s.io/apimachinery/pkg/util/validation"
 )
 
+const invalidPartOfPathErrMsg string = "every part of the path shuold follow the relaxed DNS (RFC 1123) rule which additionally allows upper case alphabetic character and character '_'"
+
+func isValidPartsOfPath(partsOfPath []string) error {
+	for _, part := range partsOfPath {
+		// Convert characters to lowercase and replace underscores with hyphens
+		part = strings.ToLower(part)
+		part = strings.Replace(part, "_", "-", -1)
+
+		// If the component fails the DNS 1123 conformity test, the function returns an error
+		if len(validation.IsDNS1123Label(part)) > 0 {
+			return fmt.Errorf(invalidPartOfPathErrMsg)
+		}
+	}
+	return nil
+}
+
 const invalidMountRootErrMsgFmt string = "invalid mount root path '%s': %s"
 
 func IsValidMountRoot(path string) error {
@@ -37,18 +53,32 @@ func IsValidMountRoot(path string) error {
 	// The path is an absolute path and to avoid an empty part, we omit the first '/'
 	parts := strings.Split(filepath.Clean(path)[1:], string(filepath.Separator))
 
-	for _, part := range parts {
-		// Convert characters to lowercase and replace underscores with hyphens
-		part = strings.ToLower(part)
-		part = strings.Replace(part, "_", "-", -1)
+	// Validate every part of the path
+	if err := isValidPartsOfPath(parts); err != nil {
+		return fmt.Errorf(invalidMountRootErrMsgFmt, path, err.Error())
+	}
 
-		// If the component fails the DNS 1123 conformity test, the function returns an error
-		errs := validation.IsDNS1123Label(part)
-		if len(errs) > 0 {
-			return fmt.Errorf(invalidMountRootErrMsgFmt, path, "every directory name in the mount root path shuold follow the relaxed DNS (RFC 1123) rule which additionally allows upper case alphabetic character and character '_'")
-		}
+	// If all components pass the check, the function returns nil
+	return nil
+}
+
+func IsValidMountPath(path string) error {
+	if len(path) == 0 {
+		return fmt.Errorf("the mount path is empty")
+	}
+	// Normalize the path and split it into components
+	path = filepath.Clean(path)
+	if filepath.IsAbs(path) {
+		// The path is an absolute path and to avoid an empty part, we omit the first '/'
+		path = path[1:]
+	}
+	parts := strings.Split(path, string(filepath.Separator))
+
+	// Validate every part of the path
+	if err := isValidPartsOfPath(parts); err != nil {
+		return err
 	}
 
-	// If all components pass the DNS 1123 check, the function returns nil
+	// If all components pass the check, the function returns nil
 	return nil
 }