[WIP] Move dataset.spec.mounts length validation logic from the CRD definition down into the controller

api/v1alpha1/dataset_types.go

@@ -127,7 +127,6 @@ type DatasetSpec struct {
 	// Mount Points to be mounted on cache runtime. <br>
 	// This field can be empty because some runtimes don't need to mount external storage (e.g.
 	// <a href="https://v6d.io/">Vineyard</a>).
-	// +kubebuilder:validation:MinItems=1
 	// +kubebuilder:validation:UniqueItems=false
 	// +optional
 	Mounts []Mount `json:"mounts,omitempty"`

config/crd/bases/data.fluid.io_datasets.yaml

@@ -159,7 +159,6 @@ spec:
                   required:
                   - mountPoint
                   type: object
-                minItems: 1
                 type: array
               nodeAffinity:
                 description: |-

pkg/common/constraints.go

@@ -0,0 +1,35 @@
+/*
+Copyright 2020 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.
+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 common
+
+type datasetMountsLengthConstraint struct {
+	AllowZeroMountPoints      bool
+	AllowUnlimitedMountPoints bool
+	AllowedMaxMountsLength    int
+}
+
+var MountsLengthConstraintMap = map[string]datasetMountsLengthConstraint{
+	AlluxioEngineImpl:    {false, true, -1},
+	JindoFSEngineImpl:    {false, true, -1},
+	JindoFSxEngineImpl:   {false, true, -1},
+	JindoCacheEngineImpl: {false, true, -1},
+	GooseFSEngineImpl:    {false, true, -1},
+	JuiceFSEngineImpl:    {false, true, -1},
+	ThinEngineImpl:       {true, true, -1},
+	EFCEngineImpl:        {false, true, -1},
+	VineyardEngineImpl:   {true, true, -1},
+}

pkg/controllers/runtime_controller.go

@@ -134,29 +134,20 @@ func (r *RuntimeReconciler) ReconcileInternal(ctx cruntime.ReconcileRequestConte
 		return result, err
 	}
 
-	if dataset != nil {
+	if ctx.Dataset != nil {
 		// 7.Add the OwnerReference of runtime and requeue
 		if !utils.ContainsOwners(objectMeta.GetOwnerReferences(), dataset) {
 			return r.AddOwnerAndRequeue(ctx, dataset)
 		}
-		if !dataset.CanbeBound(ctx.Name, ctx.Namespace, ctx.Category) {
-			ctx.Log.Info("the dataset can't be bound to the runtime, because it's already bound to another runtime ",
-				"dataset", dataset.Name)
-			r.Recorder.Eventf(runtime, corev1.EventTypeWarning,
-				common.ErrorProcessRuntimeReason,
-				"the dataset can't be bound to the runtime, because it's already bound to another runtime %s",
-				dataset.Name)
-			return utils.RequeueAfterInterval(time.Duration(20 * time.Second))
-		}
-		// check reference dataset support
-		isSupport, reason := r.CheckIfReferenceDatasetIsSupported(ctx)
-		if !isSupport {
+
+		// 8. check if runtime can be bounded to the dataset
+		if canBeBound, reason := r.CheckIfDatasetReadyToBound(ctx); !canBeBound {
 			ctx.Log.Info(reason, "dataset", dataset.Name)
 			r.Recorder.Eventf(runtime, corev1.EventTypeWarning, common.ErrorProcessRuntimeReason, reason)
 			return utils.RequeueAfterInterval(time.Duration(20 * time.Second))
 		}
 
-		// 8. Add Finalizer of runtime and requeue
+		// 9. Add Finalizer of runtime and requeue
 		if !utils.ContainsString(objectMeta.GetFinalizers(), ctx.FinalizerName) {
 			return r.implement.AddFinalizerAndRequeue(ctx, ctx.FinalizerName)
 		} else {
@@ -169,7 +160,7 @@ func (r *RuntimeReconciler) ReconcileInternal(ctx cruntime.ReconcileRequestConte
 		return utils.RequeueAfterInterval(time.Duration(5 * time.Second))
 	}
 
-	// 9.Start to reconcile runtime
+	// 10.Start to reconcile runtime
 	return r.implement.ReconcileRuntime(engine, ctx)
 }
 
@@ -187,7 +178,7 @@ func (r *RuntimeReconciler) ReconcileRuntimeDeletion(engine base.Engine, ctx cru
 		return utils.RequeueAfterInterval(time.Duration(20 * time.Second))
 	}
 
-	// 1. Delete the implementation of the the runtime
+	// 1. Delete the implementation of the runtime
 	err = engine.Shutdown()
 	if err != nil {
 		r.Recorder.Eventf(ctx.Runtime, corev1.EventTypeWarning, common.ErrorProcessRuntimeReason, "Failed to shutdown engine %v", err)
@@ -370,6 +361,52 @@ func (r *RuntimeReconciler) CheckIfReferenceDatasetIsSupported(ctx cruntime.Reco
 	return true, ""
 }
 
+func (r *RuntimeReconciler) CheckIfDatasetMountLengthIsSupported(ctx cruntime.ReconcileRequestContext) (bool, string) {
+	constraint := common.MountsLengthConstraintMap[ctx.EngineImpl]
+	switch len(ctx.Dataset.Spec.Mounts) {
+	case 0:
+		if constraint.AllowZeroMountPoints {
+			return true, "the mounts length of dataset can not be zero"
+		}
+	default:
+		if !constraint.AllowUnlimitedMountPoints {
+			if len(ctx.Dataset.Spec.Mounts) > constraint.AllowedMaxMountsLength {
+				return false, fmt.Sprintf("the mounts length of dataset is %d, which is larger than the limit %d", len(ctx.Dataset.Spec.Mounts), constraint.AllowedMaxMountsLength)
+			}
+		}
+	}
+	return true, ""
+}
+
+func (r *RuntimeReconciler) CheckIfDatasetReadyToBound(ctx cruntime.ReconcileRequestContext) (isSupported bool, reason string) {
+	dataset := ctx.Dataset
+	// 1. check if dataset has bound with another runtime
+	if !dataset.CanbeBound(ctx.Name, ctx.Namespace, ctx.Category) {
+		reason = "the dataset can't be bound to the runtime, because it's already bound to another runtime "
+		return
+	}
+
+	// 2. check reference dataset support
+	isSupported, reason = r.CheckIfReferenceDatasetIsSupported(ctx)
+	if !isSupported {
+		return
+	}
+	mounted := base.GetPhysicalDatasetFromMounts(ctx.Dataset.Spec.Mounts)
+	if len(mounted) > 0 && ctx.RuntimeType != common.ThinRuntime {
+		isSupported = false
+		reason = "dataset mounting another dataset can only use thin runtime"
+		return
+	}
+
+	// 3. check mounts length
+	isSupported, reason = r.CheckIfDatasetMountLengthIsSupported(ctx)
+	if !isSupported {
+		return
+	}
+	isSupported = true
+	return
+}
+
 func (r *RuntimeReconciler) ReportDatasetNotReadyCondition(ctx cruntime.ReconcileRequestContext, notReadyReason error) error {
 	datasetToReport := ctx.Dataset
 	err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {