using cache worker set to handle the previous request of statefulset

Author: zwbrealm

api/v1alpha1/alluxioruntime_types.go

@@ -17,6 +17,7 @@ limitations under the License.
 package v1alpha1
 
 import (
+	"github.com/fluid-cloudnative/fluid/pkg/types/cacheworkerset"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
@@ -237,11 +238,18 @@ type AlluxioRuntimeSpec struct {
 	// +optional
 	RuntimeManagement RuntimeManagement `json:"management,omitempty"`
 
+	ScaleConfig AlluxioScaleSpec `json:"scale,omitempty"`
+
 	// ImagePullSecrets that will be used to pull images
 	// +optional
 	ImagePullSecrets []corev1.LocalObjectReference `json:"imagePullSecrets,omitempty"`
 }
 
+type AlluxioScaleSpec struct {
+	WorkerType   cacheworkerset.WorkerType `json:"workerType,omitempty"`
+	ScaleInSlots []int                     `json:"scaleInSlots,omitempty"`
+}
+
 // +kubebuilder:object:root=true
 // +kubebuilder:subresource:status
 // +kubebuilder:subresource:scale:specpath=.spec.replicas,statuspath=.status.currentWorkerNumberScheduled,selectorpath=.status.selector
@@ -259,7 +267,6 @@ type AlluxioRuntimeSpec struct {
 // +kubebuilder:resource:scope=Namespaced
 // +kubebuilder:resource:categories={fluid},shortName=alluxio
 // +genclient
-
 // AlluxioRuntime is the Schema for the alluxioruntimes API
 type AlluxioRuntime struct {
 	metav1.TypeMeta   `json:",inline"`

pkg/controllers/v1alpha1/datamigrate/status_handler.go

@@ -165,7 +165,7 @@ func (c *CronStatusHandler) GetOperationStatus(ctx cruntime.ReconcileRequestCont
 	if currentJob.Spec.Suspend != nil && *currentJob.Spec.Suspend {
 		ctx.Log.Info("scale the migrate workers statefulset", "name", utils.GetParallelOperationWorkersName(releaseName))
 		// scale the stateful set, the job acts as a worker.
-		err = kubeclient.ScaleStatefulSet(c.Client, utils.GetParallelOperationWorkersName(releaseName), c.dataMigrate.Namespace, c.dataMigrate.Spec.Parallelism-1)
+		err = kubeclient.ScaleCacheWorkerSet(c.Client, utils.GetParallelOperationWorkersName(releaseName), c.dataMigrate.Namespace, c.dataMigrate.Spec.Parallelism-1)
 		if err != nil {
 			return
 		}

pkg/ctrl/affinity.go

@@ -19,14 +19,40 @@ package ctrl
 import (
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
 	"github.com/fluid-cloudnative/fluid/pkg/common"
+	"github.com/fluid-cloudnative/fluid/pkg/types/cacheworkerset"
 	"github.com/fluid-cloudnative/fluid/pkg/utils"
+	openkruise "github.com/openkruise/kruise/apis/apps/v1beta1"
 
-	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func (e *Helper) checkWorkerAffinity(workers *appsv1.StatefulSet) (found bool) {
+func (e *Helper) checkWorkerAffinity(workers *cacheworkerset.CacheWorkerSet) (found bool) {
+
+	if workers.GetAffinity() == nil {
+		return
+	}
+
+	if workers.GetNodeAffinity() == nil {
+		return
+	}
+
+	if len(workers.GetNodeAffinityPreferredDuringSchedulingIgnoredDuringExecution()) == 0 {
+		return
+	}
+
+	for _, preferred := range workers.GetNodeAffinityPreferredDuringSchedulingIgnoredDuringExecution() {
+		for _, term := range preferred.Preference.MatchExpressions {
+			if term.Key == e.runtimeInfo.GetFuseLabelName() {
+				found = true
+				return
+			}
+		}
+	}
+
+	return
+}
+func (e *Helper) checkWorkerAffinityForAts(workers *openkruise.StatefulSet) (found bool) {
 
 	if workers.Spec.Template.Spec.Affinity == nil {
 		return
@@ -52,12 +78,14 @@ func (e *Helper) checkWorkerAffinity(workers *appsv1.StatefulSet) (found bool) {
 	return
 }
 
-// BuildWorkersAffinity builds workers affinity if it doesn't have
-func (e *Helper) BuildWorkersAffinity(workers *appsv1.StatefulSet) (workersToUpdate *appsv1.StatefulSet, err error) {
-	// TODO: for now, runtime affinity can't be set by user, so we can assume the affinity is nil in the first time.
-	// We need to enhance it in future
-	workersToUpdate = workers.DeepCopy()
-	if e.checkWorkerAffinity(workersToUpdate) {
+func (e *Helper) BuildCacheWorkersAffinity(workers *cacheworkerset.CacheWorkerSet) (workersToUpdate *cacheworkerset.CacheWorkerSet, err error) {
+	returnV, err := e.BuildWorkersAffinity(workers)
+	return returnV, err
+}
+
+func (e *Helper) BuildWorkersAffinityForAsts(workers *openkruise.StatefulSet) (workersToUpdate *openkruise.StatefulSet, err error) {
+
+	if e.checkWorkerAffinityForAts(workersToUpdate) {
 		return
 	}
 	var (
@@ -168,3 +196,145 @@ func (e *Helper) BuildWorkersAffinity(workers *appsv1.StatefulSet) (workersToUpd
 
 	return
 }
+
+// BuildWorkersAffinity builds workers affinity if it doesn't have
+func (e *Helper) BuildWorkersAffinity(workers *cacheworkerset.CacheWorkerSet) (workersToUpdate *cacheworkerset.CacheWorkerSet, err error) {
+	// TODO: for now, runtime affinity can't be set by user, so we can assume the affinity is nil in the first time.
+	// We need to enhance it in future
+
+	if e.checkWorkerAffinity(workersToUpdate) {
+		return
+	}
+	var (
+		name      = e.runtimeInfo.GetName()
+		namespace = e.runtimeInfo.GetNamespace()
+	)
+
+	if workersToUpdate.GetAffinity() == nil {
+		affinity := &corev1.Affinity{}
+		workersToUpdate.SetAffinity(affinity)
+		dataset, err := utils.GetDataset(e.client, name, namespace)
+		if err != nil {
+			return workersToUpdate, err
+		}
+		// 1. Set pod anti affinity(required) for same dataset (Current using port conflict for scheduling, no need to do)
+
+		// 2. Set pod anti affinity for the different dataset
+		if dataset.IsExclusiveMode() {
+			podAntiAffinity := &corev1.PodAntiAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+					{
+						LabelSelector: &metav1.LabelSelector{
+							MatchExpressions: []metav1.LabelSelectorRequirement{
+								{
+									Key:      "fluid.io/dataset",
+									Operator: metav1.LabelSelectorOpExists,
+								},
+							},
+						},
+						TopologyKey: "kubernetes.io/hostname",
+					},
+				},
+			}
+			workersToUpdate.SetPodAntiAffinity(podAntiAffinity)
+
+		} else {
+			podAntiAffinity := &corev1.PodAntiAffinity{
+				PreferredDuringSchedulingIgnoredDuringExecution: []corev1.WeightedPodAffinityTerm{
+					{
+						// The default weight is 50
+						Weight: 50,
+						PodAffinityTerm: corev1.PodAffinityTerm{
+							LabelSelector: &metav1.LabelSelector{
+								MatchExpressions: []metav1.LabelSelectorRequirement{
+									{
+										Key:      "fluid.io/dataset",
+										Operator: metav1.LabelSelectorOpExists,
+									},
+								},
+							},
+							TopologyKey: "kubernetes.io/hostname",
+						},
+					},
+				},
+
+				RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
+					{
+						LabelSelector: &metav1.LabelSelector{
+							MatchExpressions: []metav1.LabelSelectorRequirement{
+								{
+									Key:      "fluid.io/dataset-placement",
+									Operator: metav1.LabelSelectorOpIn,
+									Values:   []string{string(datav1alpha1.ExclusiveMode)},
+								},
+							},
+						},
+						TopologyKey: "kubernetes.io/hostname",
+					},
+				},
+			}
+			workersToUpdate.SetPodAntiAffinity(podAntiAffinity)
+
+			// TODO: remove this when EFC is ready for spread-first scheduling policy
+			// Currently EFC prefers binpack-first scheduling policy to spread-first scheduling policy. Set PreferredDuringSchedulingIgnoredDuringExecution to empty
+			// to avoid using spread-first scheduling policy
+			if e.runtimeInfo.GetRuntimeType() == common.EFCRuntime {
+				preferredDuringSchedulingIgnoredDuringExecution := []corev1.WeightedPodAffinityTerm{}
+				workersToUpdate.SetPodAntiAffinityPreferredDuringSchedulingIgnoredDuringExecution(preferredDuringSchedulingIgnoredDuringExecution)
+
+			}
+		}
+
+		// 3. Prefer to locate on the node which already has fuse on it
+		if workersToUpdate.GetNodeAffinity() == nil {
+			NodeAffinity := &corev1.NodeAffinity{}
+			workersToUpdate.SetNodeAffinity(NodeAffinity)
+
+		}
+
+		if len(workersToUpdate.GetNodeAffinityPreferredDuringSchedulingIgnoredDuringExecution()) == 0 {
+
+			PreferredDuringSchedulingIgnoredDuringExecution := []corev1.PreferredSchedulingTerm{}
+			workersToUpdate.SetNodeAffinityPreferredDuringSchedulingIgnoredDuringExecution(PreferredDuringSchedulingIgnoredDuringExecution)
+
+		}
+
+		//workersToUpdate.Spec.Template.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution =
+		//	append(workersToUpdate.Spec.Template.Spec.Affinity.NodeAffinity.PreferredDuringSchedulingIgnoredDuringExecution,
+		//		corev1.PreferredSchedulingTerm{
+		//			Weight: 100,
+		//			Preference: corev1.NodeSelectorTerm{
+		//				MatchExpressions: []corev1.NodeSelectorRequirement{
+		//					{
+		//						Key:      e.runtimeInfo.GetFuseLabelName(),
+		//						Operator: corev1.NodeSelectorOpIn,
+		//						Values:   []string{"true"},
+		//					},
+		//				},
+		//			},
+		//		})
+		PreferredDuringSchedulingIgnoredDuringExecution := corev1.PreferredSchedulingTerm{
+			Weight: 100,
+			Preference: corev1.NodeSelectorTerm{
+				MatchExpressions: []corev1.NodeSelectorRequirement{
+					{
+						Key:      e.runtimeInfo.GetFuseLabelName(),
+						Operator: corev1.NodeSelectorOpIn,
+						Values:   []string{"true"},
+					},
+				},
+			},
+		}
+		workersToUpdate.AppendNodeAffinityPreferredDuringSchedulingIgnoredDuringExecution(PreferredDuringSchedulingIgnoredDuringExecution)
+
+		// 3. set node affinity if possible
+		if dataset.Spec.NodeAffinity != nil {
+			if dataset.Spec.NodeAffinity.Required != nil {
+				workersToUpdate.SetNodeAffinityRequired(dataset.Spec.NodeAffinity.Required)
+
+			}
+		}
+	}
+
+	return
+}

pkg/ctrl/ctrl.go

@@ -21,14 +21,13 @@ import (
 	"reflect"
 
 	"github.com/fluid-cloudnative/fluid/pkg/ddc/base"
+	cacheworkerset "github.com/fluid-cloudnative/fluid/pkg/types/cacheworkerset"
 	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
-	appsv1 "k8s.io/api/apps/v1"
+	"github.com/go-logr/logr"
 	corev1 "k8s.io/api/core/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
-	"github.com/go-logr/logr"
-
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
 )
 
@@ -54,18 +53,19 @@ func BuildHelper(runtimeInfo base.RuntimeInfoInterface, client client.Client, lo
 // calls for a status update and finally returns error if anything unexpected happens.
 func (e *Helper) SetupWorkers(runtime base.RuntimeInterface,
 	currentStatus datav1alpha1.RuntimeStatus,
-	workers *appsv1.StatefulSet) (err error) {
+	workers *cacheworkerset.CacheWorkerSet) (err error) {
 
 	desireReplicas := runtime.Replicas()
-	if *workers.Spec.Replicas != desireReplicas {
+	if *workers.GetReplicas() != desireReplicas {
 		// workerToUpdate, err := e.buildWorkersAffinity(workers)
 
 		workerToUpdate, err := e.BuildWorkersAffinity(workers)
 		if err != nil {
 			return err
 		}
 
-		workerToUpdate.Spec.Replicas = &desireReplicas
+		//workerToUpdate.Spec.Replicas = &desireReplicas
+		workerToUpdate.SetReplicas(&desireReplicas)
 		err = e.client.Update(context.TODO(), workerToUpdate)
 		if err != nil {
 			return err
@@ -76,11 +76,11 @@ func (e *Helper) SetupWorkers(runtime base.RuntimeInterface,
 		e.log.V(1).Info("Nothing to do for syncing")
 	}
 
-	if *workers.Spec.Replicas != runtime.GetStatus().DesiredWorkerNumberScheduled {
+	if *workers.GetReplicas() != runtime.GetStatus().DesiredWorkerNumberScheduled {
 		statusToUpdate := runtime.GetStatus()
 
-		if workers.Status.ReadyReplicas > 0 {
-			if runtime.Replicas() == workers.Status.ReadyReplicas {
+		if workers.GetReadyReplicas() > 0 {
+			if runtime.Replicas() == workers.GetReadyReplicas() {
 				statusToUpdate.WorkerPhase = datav1alpha1.RuntimePhaseReady
 			} else {
 				statusToUpdate.WorkerPhase = datav1alpha1.RuntimePhasePartialReady
@@ -114,10 +114,10 @@ func (e *Helper) SetupWorkers(runtime base.RuntimeInterface,
 // CheckWorkersReady checks if workers are ready
 func (e *Helper) CheckWorkersReady(runtime base.RuntimeInterface,
 	currentStatus datav1alpha1.RuntimeStatus,
-	workers *appsv1.StatefulSet) (ready bool, err error) {
+	workers *cacheworkerset.CacheWorkerSet) (ready bool, err error) {
 
 	var (
-		phase datav1alpha1.RuntimePhase     = kubeclient.GetPhaseFromStatefulset(runtime.Replicas(), *workers)
+		phase datav1alpha1.RuntimePhase     = kubeclient.GetPhaseFromCacheWorkset(runtime.Replicas(), workers)
 		cond  datav1alpha1.RuntimeCondition = datav1alpha1.RuntimeCondition{}
 	)
 

pkg/ctrl/master.go

@@ -19,13 +19,14 @@ package ctrl
 import (
 	"context"
 	"fmt"
+	"github.com/fluid-cloudnative/fluid/pkg/types/cacheworkerset"
 	"reflect"
 
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
 	"github.com/fluid-cloudnative/fluid/pkg/ddc/base"
 	"github.com/fluid-cloudnative/fluid/pkg/utils"
 	"github.com/fluid-cloudnative/fluid/pkg/utils/kubeclient"
-	appsv1 "k8s.io/api/apps/v1"
+	_ "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/record"
 
@@ -37,13 +38,13 @@ import (
 // CheckMasterHealthy checks the sts healthy with role
 func (e *Helper) CheckMasterHealthy(recorder record.EventRecorder, runtime base.RuntimeInterface,
 	currentStatus datav1alpha1.RuntimeStatus,
-	sts *appsv1.StatefulSet) (err error) {
+	sts *cacheworkerset.CacheWorkerSet) (err error) {
 	var (
 		healthy             bool
 		selector            labels.Selector
 		unavailablePodNames []types.NamespacedName
 	)
-	if sts.Status.Replicas == sts.Status.ReadyReplicas {
+	if *sts.GetReplicas() == sts.GetReadyReplicas() {
 		healthy = true
 	}
 
@@ -67,7 +68,7 @@ func (e *Helper) CheckMasterHealthy(recorder record.EventRecorder, runtime base.
 	} else {
 		// 1. Update the status
 		cond := utils.NewRuntimeCondition(datav1alpha1.RuntimeMasterReady, "The master is not ready.",
-			fmt.Sprintf("The master %s in %s is not ready.", sts.Name, sts.Namespace), corev1.ConditionFalse)
+			fmt.Sprintf("The master %s in %s is not ready.", sts.GetName(), sts.GetNamespace()), corev1.ConditionFalse)
 		_, oldCond := utils.GetRuntimeCondition(statusToUpdate.Conditions, cond.Type)
 
 		if oldCond == nil || oldCond.Type != cond.Type {
@@ -79,22 +80,22 @@ func (e *Helper) CheckMasterHealthy(recorder record.EventRecorder, runtime base.
 
 		// 2. Record the event
 
-		selector, err = metav1.LabelSelectorAsSelector(sts.Spec.Selector)
+		selector, err = metav1.LabelSelectorAsSelector(sts.GetSelector())
 		if err != nil {
-			return fmt.Errorf("error converting StatefulSet %s in namespace %s selector: %v", sts.Name, sts.Namespace, err)
+			return fmt.Errorf("error converting StatefulSet %s in namespace %s selector: %v", sts.GetName(), sts.GetNamespace(), err)
 		}
 
-		unavailablePodNames, err = kubeclient.GetUnavailablePodNamesForStatefulSet(e.client, sts, selector)
+		unavailablePodNames, err = kubeclient.GetUnavailablePodNamesForCacheWorkerSet(e.client, sts, selector)
 		if err != nil {
 			return err
 		}
 
 		// 3. Set event
 		err = fmt.Errorf("the master %s in %s is not ready. The expected number is %d, the actual number is %d, the unhealthy pods are %v",
-			sts.Name,
-			sts.Namespace,
-			sts.Status.Replicas,
-			sts.Status.ReadyReplicas,
+			sts.GetName(),
+			sts.GetNamespace(),
+			sts.GetReplicas(),
+			sts.GetReadyReplicas(),
 			unavailablePodNames)
 
 		recorder.Eventf(runtime, corev1.EventTypeWarning, "MasterUnhealthy", err.Error())