feature: support one-to-many model for the leaf node feature

Signed-off-by: qiuming520 <[email protected]>

Author: qiuming520

pkg/clustertree/cluster-manager/controllers/pod/root_pod_controller.go

@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
 	"github.com/kosmos.io/kosmos/cmd/clustertree/cluster-manager/app/options"
+	kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
 	"github.com/kosmos.io/kosmos/pkg/clustertree/cluster-manager/extensions/daemonset"
 	leafUtils "github.com/kosmos.io/kosmos/pkg/clustertree/cluster-manager/utils"
 	"github.com/kosmos.io/kosmos/pkg/utils"
@@ -198,7 +199,7 @@ func (r *RootPodReconciler) Reconcile(ctx context.Context, request reconcile.Req
 	// create pod in leaf
 	if err != nil {
 		if errors.IsNotFound(err) {
-			if err := r.CreatePodInLeafCluster(ctx, lr, &rootpod); err != nil {
+			if err := r.CreatePodInLeafCluster(ctx, lr, &rootpod, r.GlobalLeafManager.GetClusterNode(rootpod.Spec.NodeName).LeafNodeSelector); err != nil {
 				klog.Errorf("create pod inleaf error, err: %s", err)
 				return reconcile.Result{RequeueAfter: RootPodRequeueTime}, nil
 			} else {
@@ -700,7 +701,7 @@ func (r *RootPodReconciler) createVolumes(ctx context.Context, lr *leafUtils.Lea
 	return nil
 }
 
-func (r *RootPodReconciler) CreatePodInLeafCluster(ctx context.Context, lr *leafUtils.LeafResource, pod *corev1.Pod) error {
+func (r *RootPodReconciler) CreatePodInLeafCluster(ctx context.Context, lr *leafUtils.LeafResource, pod *corev1.Pod, nodeSelector kosmosv1alpha1.NodeSelector) error {
 	if err := podutils.PopulateEnvironmentVariables(ctx, pod, r.envResourceManager); err != nil {
 		// span.SetStatus(err)
 		return err
@@ -711,7 +712,7 @@ func (r *RootPodReconciler) CreatePodInLeafCluster(ctx context.Context, lr *leaf
 		return fmt.Errorf("clusternode info is nil , name: %s", pod.Spec.NodeName)
 	}
 
-	basicPod := podutils.FitPod(pod, lr.IgnoreLabels, clusterNodeInfo.LeafMode == leafUtils.ALL)
+	basicPod := podutils.FitPod(pod, lr.IgnoreLabels, clusterNodeInfo.LeafMode, nodeSelector)
 	klog.V(4).Infof("Creating pod %v/%+v", pod.Namespace, pod.Name)
 
 	// create ns

pkg/clustertree/cluster-manager/utils/leaf_model_handler.go

@@ -9,7 +9,6 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/util/retry"
 	"k8s.io/klog/v2"
@@ -295,47 +294,33 @@ func (h ClassificationModelHandler) GetLeafModelType() LeafModelType {
 // GetLeafNodes returns nodes in leaf cluster by the rootNode
 func (h ClassificationModelHandler) GetLeafNodes(ctx context.Context, _ *corev1.Node, selector kosmosv1alpha1.NodeSelector) (*corev1.NodeList, error) {
 	nodesInLeaf, err := h.LeafClientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{
-		LabelSelector: selector.LabelSelector.String(),
+		LabelSelector: metav1.FormatLabelSelector(selector.LabelSelector),
 	})
 	if err != nil {
 		return nil, err
 	}
 	return nodesInLeaf, nil
 }
 
-func join(nodeList *corev1.NodeList, sep string) string {
-	nodes := nodeList.Items
-	if len(nodes) == 0 {
-		return ""
-	}
-
-	result := nodes[0].Name
-	for _, node := range nodes[1:] {
-		result += sep + node.Name
-	}
-
-	return result
-}
-
 // GetLeafPods returns pods in leaf cluster by the rootNode
 func (h ClassificationModelHandler) GetLeafPods(ctx context.Context, rootNode *corev1.Node, selector kosmosv1alpha1.NodeSelector) (pods *corev1.PodList, err error) {
 	nodesInLeafs, err := h.GetLeafNodes(ctx, rootNode, selector)
 	if err != nil {
 		return nil, err
 	}
 
-	fieldSelector := join(nodesInLeafs, ",")
-	if fieldSelector == "" {
-		klog.Warningf("have leaf node in rootNode :v%", rootNode.Name)
-		return
-	}
-
-	fieldSelector = "spec.nodeName in (" + join(nodesInLeafs, ",") + ")"
-	pods, err = h.LeafClientset.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
-		FieldSelector: fieldSelector,
-	})
-	if err != nil {
-		return nil, err
+	for _, node := range nodesInLeafs.Items {
+		podsInNode, err := h.LeafClientset.CoreV1().Pods(metav1.NamespaceAll).List(ctx, metav1.ListOptions{
+			FieldSelector: fmt.Sprintf("spec.nodeName=%s", node.Name),
+		})
+		if err != nil {
+			return nil, err
+		}
+		if pods == nil {
+			pods = podsInNode
+		} else {
+			pods.Items = append(pods.Items, podsInNode.Items...)
+		}
 	}
 
 	return pods, nil
@@ -368,13 +353,16 @@ func (h ClassificationModelHandler) UpdateNodeStatus(ctx context.Context, nodes
 			}
 
 			nodesInLeaf, err := h.LeafClientset.CoreV1().Nodes().List(ctx, metav1.ListOptions{
-				LabelSelector: selector.LabelSelector.String(),
+				LabelSelector: metav1.FormatLabelSelector(selector.LabelSelector),
 			})
 			if err != nil {
 				// TODO: If a node is accidentally deleted, recreate it
 				return fmt.Errorf("cannot get node in leaf cluster while update node status %s, err: %v", nodeCopy.Name, err)
 			}
 
+			if len(nodesInLeaf.Items) == 0 {
+				return fmt.Errorf("cannot get node in leaf cluster while update node status, leaf node item is 0")
+			}
 			rootCopy := nodeRoot.DeepCopy()
 
 			// TODO: Aggregation the resources of the leaf nodes
@@ -402,23 +390,25 @@ func (h ClassificationModelHandler) UpdateNodeStatus(ctx context.Context, nodes
 // CreateNodeInRoot creates the node in root cluster
 func (h ClassificationModelHandler) CreateNodeInRoot(ctx context.Context, cluster *kosmosv1alpha1.Cluster, listenPort int32, gitVersion string) ([]*corev1.Node, map[string]kosmosv1alpha1.NodeSelector, error) {
 	nodes := make([]*corev1.Node, 0)
-
 	leafNodeSelectors := make(map[string]kosmosv1alpha1.NodeSelector)
-	for _, leafModel := range cluster.Spec.ClusterTreeOptions.LeafModels {
-		if reflect.DeepEqual(leafModel.NodeSelector.LabelSelector, metav1.LabelSelector{}) {
+
+	for i, leafModel := range cluster.Spec.ClusterTreeOptions.LeafModels {
+		if !reflect.DeepEqual(leafModel.NodeSelector.LabelSelector, metav1.LabelSelector{}) {
 			var nodeName string
 			if leafModel.NodeSelector.NodeName != "" {
-				nodeName = leafModel.NodeSelector.NodeName
+				nodeName = fmt.Sprintf("%v%v%v%v", utils.KosmosNodePrefix, leafModel.NodeSelector.NodeName, "-", i)
 			} else {
-				nodeName = fmt.Sprintf("%s%s%s%s", utils.KosmosNodePrefix, cluster.Name, "_", string(uuid.NewUUID()))
+				nodeName = fmt.Sprintf("%v%v%v%v", utils.KosmosNodePrefix, cluster.Name, "-", i)
 			}
 
+			if len(nodeName) > 63 {
+				nodeName = nodeName[:63]
+			}
 			node, err := h.RootClientset.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
 			if err != nil {
 				if !errors.IsNotFound(err) {
 					return nil, nil, err
 				}
-
 				node = utils.BuildNodeTemplate(nodeName)
 				nodeAnnotations := node.GetAnnotations()
 				if nodeAnnotations == nil {
@@ -427,14 +417,14 @@ func (h ClassificationModelHandler) CreateNodeInRoot(ctx context.Context, cluste
 				nodeAnnotations[utils.KosmosNodeOwnedByClusterAnnotations] = cluster.Name
 				node.SetAnnotations(nodeAnnotations)
 
-				if leafModel.NodeSelector.LabelSelector.MatchLabels != nil {
-					nodeLables := node.GetLabels()
-					if nodeLables == nil {
-						nodeLables = make(map[string]string)
-					}
-					nodeLables[utils.KosmosNodeOwnedByClusterAnnotations] = cluster.Name
-					node.SetLabels(leafModel.NodeSelector.LabelSelector.MatchLabels)
+				nodeLabels := node.GetLabels()
+				if nodeLabels == nil {
+					nodeLabels = make(map[string]string, 1)
 				}
+				for key, value := range leafModel.NodeSelector.LabelSelector.MatchLabels {
+					nodeLabels[key] = value
+				}
+				node.SetLabels(nodeLabels)
 
 				node.Status.NodeInfo.KubeletVersion = gitVersion
 				node.Status.DaemonEndpoints = corev1.NodeDaemonEndpoints{
@@ -451,6 +441,7 @@ func (h ClassificationModelHandler) CreateNodeInRoot(ctx context.Context, cluste
 				}
 			}
 			nodes = append(nodes, node)
+			leafNodeSelectors[nodeName] = leafModel.NodeSelector
 		}
 	}
 	return nodes, leafNodeSelectors, nil
@@ -459,18 +450,18 @@ func (h ClassificationModelHandler) CreateNodeInRoot(ctx context.Context, cluste
 // NewLeafModelHandler create a LeafModelHandler for Cluster
 func NewLeafModelHandler(cluster *kosmosv1alpha1.Cluster, root, leafClient client.Client, rootClientset, leafClientset kubernetes.Interface) LeafModelHandler {
 	leafModels := cluster.Spec.ClusterTreeOptions.LeafModels
-	if leafModels != nil {
-		if reflect.DeepEqual(leafModels[0].NodeSelector, kosmosv1alpha1.NodeSelector{}) {
-			return &DispersionModelHandler{
+	if leafModels != nil && !reflect.DeepEqual(leafModels[0].NodeSelector, kosmosv1alpha1.NodeSelector{}) {
+		if leafModels[0].NodeSelector.LabelSelector != nil {
+			// support nodeSelector mode
+			return &ClassificationModelHandler{
 				Cluster:       cluster,
 				LeafClient:    leafClient,
 				RootClient:    root,
 				RootClientset: rootClientset,
 				LeafClientset: leafClientset,
 			}
 		} else {
-			// support nodeSelector mode
-			return &ClassificationModelHandler{
+			return &DispersionModelHandler{
 				Cluster:       cluster,
 				LeafClient:    leafClient,
 				RootClient:    root,

pkg/clustertree/cluster-manager/utils/leaf_resource_manager.go

@@ -30,8 +30,9 @@ const (
 )
 
 type ClusterNode struct {
-	NodeName string
-	LeafMode LeafMode
+	NodeName         string
+	LeafMode         LeafMode
+	LeafNodeSelector kosmosv1alpha1.NodeSelector
 }
 
 type LeafResource struct {
@@ -106,8 +107,9 @@ func (l *leafResourceManager) AddLeafResource(lptr *LeafResource, cluster *kosmo
 		if leafModels != nil && leafModels[i].NodeSelector.LabelSelector != nil {
 			// TODO: support labelselector
 			clusterNodes = append(clusterNodes, ClusterNode{
-				NodeName: trimNamePrefix(n.Name),
-				LeafMode: Party,
+				NodeName:         trimNamePrefix(n.Name),
+				LeafMode:         Party,
+				LeafNodeSelector: leafModels[i].NodeSelector,
 			})
 		} else if leafModels != nil && len(leafModels[i].NodeSelector.NodeName) > 0 {
 			clusterNodes = append(clusterNodes, ClusterNode{

pkg/utils/podutils/pod.go

@@ -10,6 +10,8 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/klog"
 
+	kosmosv1alpha1 "github.com/kosmos.io/kosmos/pkg/apis/kosmos/v1alpha1"
+	clustertreeutil "github.com/kosmos.io/kosmos/pkg/clustertree/cluster-manager/utils"
 	"github.com/kosmos.io/kosmos/pkg/utils"
 )
 
@@ -141,7 +143,7 @@ func FitUnstructuredObjMeta(unstructuredObj *unstructured.Unstructured) {
 	}
 }
 
-func FitPod(pod *corev1.Pod, ignoreLabels []string, cleanNodeName bool) *corev1.Pod {
+func FitPod(pod *corev1.Pod, ignoreLabels []string, leafMode clustertreeutil.LeafMode, nodeSelector kosmosv1alpha1.NodeSelector) *corev1.Pod {
 	vols := []corev1.Volume{}
 	for _, v := range pod.Spec.Volumes {
 		if strings.HasPrefix(v.Name, "default-token") {
@@ -170,10 +172,14 @@ func FitPod(pod *corev1.Pod, ignoreLabels []string, cleanNodeName bool) *corev1.
 		podCopy.Spec.SchedulerName = ""
 	}
 
-	if cleanNodeName {
+	if leafMode != clustertreeutil.Node {
 		podCopy.Spec.NodeName = ""
 	}
 
+	if leafMode == clustertreeutil.Party {
+		podCopy.Spec.Affinity = fitNodeAffinity(pod.Spec.Affinity, nodeSelector)
+	}
+
 	tripped := FitLabels(podCopy.ObjectMeta.Labels, ignoreLabels)
 	if tripped != nil {
 		trippedStr, err := json.Marshal(tripped)
@@ -186,6 +192,64 @@ func FitPod(pod *corev1.Pod, ignoreLabels []string, cleanNodeName bool) *corev1.
 	return podCopy
 }
 
+func fitNodeAffinity(affinity *corev1.Affinity, nodeSelector kosmosv1alpha1.NodeSelector) (cpAffinity *corev1.Affinity) {
+	nodeSelectorTerms := make([]corev1.NodeSelectorTerm, 0)
+	nodeSelectorTerm := corev1.NodeSelectorTerm{
+		MatchExpressions: make([]corev1.NodeSelectorRequirement, 0),
+	}
+	if nodeSelector.LabelSelector.MatchLabels != nil {
+		for key, value := range nodeSelector.LabelSelector.MatchLabels {
+			selector := corev1.NodeSelectorRequirement{
+				Key:      key,
+				Operator: corev1.NodeSelectorOpIn,
+				Values:   []string{value},
+			}
+			nodeSelectorTerm.MatchExpressions = append(nodeSelectorTerm.MatchExpressions, selector)
+		}
+	}
+
+	if nodeSelector.LabelSelector.MatchExpressions != nil {
+		for _, item := range nodeSelector.LabelSelector.MatchExpressions {
+			selector := corev1.NodeSelectorRequirement{
+				Key:      item.Key,
+				Operator: corev1.NodeSelectorOperator(item.Operator),
+				Values:   item.Values,
+			}
+			nodeSelectorTerm.MatchExpressions = append(nodeSelectorTerm.MatchExpressions, selector)
+		}
+	}
+	nodeSelectorTerms = append(nodeSelectorTerms, nodeSelectorTerm)
+
+	if affinity == nil {
+		cpAffinity = &corev1.Affinity{
+			NodeAffinity: &corev1.NodeAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+					NodeSelectorTerms: nodeSelectorTerms,
+				},
+			},
+		}
+	} else {
+		cpAffinity = affinity.DeepCopy()
+		if cpAffinity.NodeAffinity == nil {
+			cpAffinity.NodeAffinity = &corev1.NodeAffinity{
+				RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
+					NodeSelectorTerms: nodeSelectorTerms,
+				},
+			}
+		} else if cpAffinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution == nil {
+			cpAffinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution = &corev1.NodeSelector{
+				NodeSelectorTerms: nodeSelectorTerms,
+			}
+		} else if cpAffinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms == nil {
+			cpAffinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = nodeSelectorTerms
+		} else {
+			//nodeSelectorTerms = append(nodeSelectorTerms, cpAffinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms...)
+			cpAffinity.NodeAffinity.RequiredDuringSchedulingIgnoredDuringExecution.NodeSelectorTerms = nodeSelectorTerms
+		}
+	}
+	return cpAffinity
+}
+
 func fitContainers(containers []corev1.Container) []corev1.Container {
 	var newContainers []corev1.Container
 

pkg/utils/resources.go

@@ -52,26 +52,28 @@ func SubResourceList(base, list corev1.ResourceList) {
 // lifted from https://github.com/kubernetes/kubernetes/blob/v1.21.8/staging/src/k8s.io/kubectl/pkg/describe/describe.go#L4051
 func GetPodsTotalRequestsAndLimits(podList *corev1.PodList) (reqs corev1.ResourceList, limits corev1.ResourceList) {
 	reqs, limits = corev1.ResourceList{}, corev1.ResourceList{}
-	for _, p := range podList.Items {
-		pod := p
-		if IsVirtualPod(&pod) {
-			continue
-		}
-		podReqs, podLimits := v1resource.PodRequestsAndLimits(&pod)
-		for podReqName, podReqValue := range podReqs {
-			if value, ok := reqs[podReqName]; !ok {
-				reqs[podReqName] = podReqValue.DeepCopy()
-			} else {
-				value.Add(podReqValue)
-				reqs[podReqName] = value
+	if podList.Items != nil {
+		for _, p := range podList.Items {
+			pod := p
+			if IsVirtualPod(&pod) {
+				continue
 			}
-		}
-		for podLimitName, podLimitValue := range podLimits {
-			if value, ok := limits[podLimitName]; !ok {
-				limits[podLimitName] = podLimitValue.DeepCopy()
-			} else {
-				value.Add(podLimitValue)
-				limits[podLimitName] = value
+			podReqs, podLimits := v1resource.PodRequestsAndLimits(&pod)
+			for podReqName, podReqValue := range podReqs {
+				if value, ok := reqs[podReqName]; !ok {
+					reqs[podReqName] = podReqValue.DeepCopy()
+				} else {
+					value.Add(podReqValue)
+					reqs[podReqName] = value
+				}
+			}
+			for podLimitName, podLimitValue := range podLimits {
+				if value, ok := limits[podLimitName]; !ok {
+					limits[podLimitName] = podLimitValue.DeepCopy()
+				} else {
+					value.Add(podLimitValue)
+					limits[podLimitName] = value
+				}
 			}
 		}
 	}