fix:1.root pvc could be lost possibly; 2.leaf pvc spec could not be m…

pkg/clustertree/cluster-manager/cluster_controller.go

@@ -291,15 +291,15 @@ func (c *ClusterController) setupControllers(mgr manager.Manager, cluster *kosmo
 		return fmt.Errorf("error starting podUpstreamReconciler %s: %v", podcontrollers.LeafPodControllerName, err)
 	}
 
-	err := c.setupStorageControllers(mgr, nodes, leafClientset, cluster.Name)
+	err := c.setupStorageControllers(mgr, utils.IsOne2OneMode(cluster), cluster.Name)
 	if err != nil {
 		return err
 	}
 
 	return nil
 }
 
-func (c *ClusterController) setupStorageControllers(mgr manager.Manager, nodes []*corev1.Node, leafClient kubernetes.Interface, clustername string) error {
+func (c *ClusterController) setupStorageControllers(mgr manager.Manager, isOne2OneMode bool, clustername string) error {
 	leafPVCController := pvc.LeafPVCController{
 		LeafClient:    mgr.GetClient(),
 		RootClient:    c.Root,
@@ -315,6 +315,7 @@ func (c *ClusterController) setupStorageControllers(mgr manager.Manager, nodes [
 		RootClient:    c.Root,
 		RootClientSet: c.RootClientset,
 		ClusterName:   clustername,
+		IsOne2OneMode: isOne2OneMode,
 	}
 	if err := leafPVController.SetupWithManager(mgr); err != nil {
 		return fmt.Errorf("error starting leaf pv controller %v", err)

pkg/clustertree/cluster-manager/controllers/pv/leaf_pv_controller.go

@@ -33,12 +33,13 @@ type LeafPVController struct {
 	RootClient    client.Client
 	RootClientSet kubernetes.Interface
 	ClusterName   string
+	IsOne2OneMode bool
 }
 
 func (l *LeafPVController) Reconcile(ctx context.Context, request reconcile.Request) (reconcile.Result, error) {
 	pv := &v1.PersistentVolume{}
-	err := l.LeafClient.Get(ctx, request.NamespacedName, pv)
 	pvNeedDelete := false
+	err := l.LeafClient.Get(ctx, request.NamespacedName, pv)
 	if err != nil {
 		if !errors.IsNotFound(err) {
 			klog.Errorf("get pv from leaf cluster failed, error: %v", err)
@@ -56,7 +57,7 @@ func (l *LeafPVController) Reconcile(ctx context.Context, request reconcile.Requ
 			return reconcile.Result{RequeueAfter: LeafPVRequeueTime}, nil
 		}
 
-		if pv.DeletionTimestamp != nil {
+		if pvNeedDelete || pv.DeletionTimestamp != nil {
 			return reconcile.Result{}, nil
 		}
 
@@ -84,7 +85,7 @@ func (l *LeafPVController) Reconcile(ctx context.Context, request reconcile.Requ
 		}
 
 		rootPV = pv.DeepCopy()
-		filterPV(rootPV, l.ClusterName)
+		filterPV(rootPV, utils.NodeAffinity4RootPV(pv, l.IsOne2OneMode, l.ClusterName))
 		nn := types.NamespacedName{
 			Name:      rootPV.Spec.ClaimRef.Name,
 			Namespace: rootPV.Spec.ClaimRef.Namespace,
@@ -128,7 +129,7 @@ func (l *LeafPVController) Reconcile(ctx context.Context, request reconcile.Requ
 		return reconcile.Result{}, nil
 	}
 
-	filterPV(rootPV, l.ClusterName)
+	filterPV(rootPV, utils.NodeAffinity4RootPV(pv, l.IsOne2OneMode, l.ClusterName))
 	if pvCopy.Spec.ClaimRef != nil || rootPV.Spec.ClaimRef == nil {
 		nn := types.NamespacedName{
 			Name:      pvCopy.Spec.ClaimRef.Name,

pkg/clustertree/cluster-manager/controllers/pvc/leaf_pvc_controller.go

@@ -10,6 +10,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	mergetypes "k8s.io/apimachinery/pkg/types"
+	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -65,6 +66,29 @@ func (l *LeafPVCController) Reconcile(ctx context.Context, request reconcile.Req
 	if reflect.DeepEqual(rootPVC.Status, pvcCopy.Status) {
 		return reconcile.Result{}, nil
 	}
+
+	//when root pvc is not bound, it's status can't be changed to bound
+	if pvcCopy.Status.Phase == v1.ClaimBound {
+		err = wait.PollImmediate(500*time.Millisecond, 1*time.Minute, func() (bool, error) {
+			if rootPVC.Spec.VolumeName == "" {
+				klog.Warningf("pvc namespace: %q, name: %q is not bounded", request.NamespacedName.Namespace,
+					request.NamespacedName.Name)
+				err = l.RootClient.Get(ctx, request.NamespacedName, rootPVC)
+				if err != nil {
+					return false, err
+				}
+				return false, nil
+			}
+			return true, nil
+		})
+		if err != nil {
+			if !errors.IsNotFound(err) {
+				return reconcile.Result{RequeueAfter: LeafPVCRequeueTime}, nil
+			}
+			return reconcile.Result{}, nil
+		}
+	}
+
 	if err = filterPVC(pvcCopy, l.ClusterName); err != nil {
 		return reconcile.Result{}, nil
 	}

pkg/clustertree/cluster-manager/controllers/pvc/root_pvc_controller.go

@@ -71,9 +71,11 @@ func (r *RootPVCController) Reconcile(ctx context.Context, request reconcile.Req
 		return reconcile.Result{}, nil
 	}*/
 
-	if reflect.DeepEqual(pvcOld.Spec, pvc.Spec) {
+	if reflect.DeepEqual(pvcOld.Spec.Resources.Requests, pvc.Spec.Resources.Requests) {
 		return reconcile.Result{}, nil
 	}
+	pvcOld.Spec.Resources.Requests = pvc.Spec.Resources.Requests
+	pvc.Spec = pvcOld.Spec
 
 	pvc.Annotations = pvcOld.Annotations
 	pvc.ObjectMeta.UID = pvcOld.ObjectMeta.UID

pkg/utils/pvpvc.go

@@ -1,9 +1,12 @@
 package utils
 
 import (
+	"fmt"
 	"reflect"
 
 	v1 "k8s.io/api/core/v1"
+
+	kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
 )
 
 func IsPVEqual(pv *v1.PersistentVolume, clone *v1.PersistentVolume) bool {
@@ -14,3 +17,26 @@ func IsPVEqual(pv *v1.PersistentVolume, clone *v1.PersistentVolume) bool {
 	}
 	return false
 }
+
+func IsOne2OneMode(cluster *kosmosv1alpha1.Cluster) bool {
+	return cluster.Spec.ClusterTreeOptions.LeafModels != nil
+}
+
+func NodeAffinity4RootPV(pv *v1.PersistentVolume, isOne2OneMode bool, clusterName string) string {
+	node4RootPV := fmt.Sprintf("%s%s", KosmosNodePrefix, clusterName)
+	if isOne2OneMode {
+		for _, v := range pv.Spec.NodeAffinity.Required.NodeSelectorTerms {
+			for _, val := range v.MatchFields {
+				if val.Key == NodeHostnameValue || val.Key == NodeHostnameValueBeta {
+					node4RootPV = val.Values[0]
+				}
+			}
+			for _, val := range v.MatchExpressions {
+				if val.Key == NodeHostnameValue || val.Key == NodeHostnameValueBeta {
+					node4RootPV = val.Values[0]
+				}
+			}
+		}
+	}
+	return node4RootPV
+}