Merge branch 'kosmos-io:main' into fix_clustertree

rxy0210 · Nov 24, 2023 · 0308b3c · 0308b3c
2 parents 26ffa67 + 049b2ca
commit 0308b3c
Show file tree

Hide file tree

Showing 23 changed files with 390 additions and 194 deletions.
diff --git a/cmd/clusterlink/OWNERS b/cmd/clusterlink/OWNERS
@@ -0,0 +1,10 @@
+approvers:
+  - wuyingjun-lucky
+  - hanweisen
+  - wangyizhi1
+  - OrangeBao
+reviewers:
+  - wuyingjun-lucky
+  - hanweisen
+  - wangyizhi1
+  - OrangeBao
diff --git a/deploy/clusterlink-agent.yaml b/deploy/clusterlink-agent.yaml
@@ -46,6 +46,12 @@ spec:
           - mountPath: /etc/clusterlink/kubeconfig
             name: proxy-config
             readOnly: true
+          - mountPath: /run/xtables.lock
+            name: iptableslock
+            readOnly: false
+          - mountPath: /lib/modules
+            name: lib-modules
+            readOnly: true
       terminationGracePeriodSeconds: 30
       securityContext:
         privileged: true
@@ -55,3 +61,12 @@ spec:
         configMap:
           defaultMode: 420
           name: proxy-config
+      - hostPath:
+          path: /run/xtables.lock
+          type: FileOrCreate
+        name: iptableslock
+      - name: lib-modules
+        hostPath:
+          path: /lib/modules
+
+
diff --git a/deploy/clusterlink-elector.yml b/deploy/clusterlink-elector.yml
@@ -21,6 +21,23 @@ spec:
       labels:
         app: elector
     spec:
+      nodeAffinity:
+        requiredDuringSchedulingIgnoredDuringExecution:
+          nodeSelectorTerms:
+            - matchExpressions:
+                - key: kosmos.io/exclude
+                  operator: DoesNotExist
+      podAntiAffinity:
+        requiredDuringSchedulingIgnoredDuringExecution:
+          - labelSelector:
+              matchExpressions:
+                - key: app
+                  operator: In
+                  values:
+                    - elector
+            namespaces:
+              - kosmos-system
+            topologyKey: kubernetes.io/hostname
       containers:
         - command:
             - clusterlink-elector

diff --git a/pkg/apis/kosmos/v1alpha1/cluster_types.go b/pkg/apis/kosmos/v1alpha1/cluster_types.go
@@ -36,20 +36,20 @@ type ClusterSpec struct {
 	ImageRepository string `json:"imageRepository,omitempty"`
 
 	// +optional
-	ClusterLinkOptions ClusterLinkOptions `json:"clusterLinkOptions,omitempty"`
+	ClusterLinkOptions *ClusterLinkOptions `json:"clusterLinkOptions,omitempty"`
 
 	// +optional
-	ClusterTreeOptions ClusterTreeOptions `json:"clusterTreeOptions,omitempty"`
+	ClusterTreeOptions *ClusterTreeOptions `json:"clusterTreeOptions,omitempty"`
 }
 
 type ClusterStatus struct {
 	// ClusterLinkStatus contain the cluster network information
 	// +optional
-	ClusterLinkStatus ClusterLinkStatus `json:"clusterLinkStatus,omitempty"`
+	ClusterLinkStatus *ClusterLinkStatus `json:"clusterLinkStatus,omitempty"`
 
 	// ClusterTreeStatus contain the member cluster leafNode end status
 	// +optional
-	ClusterTreeStatus ClusterTreeStatus `json:"clusterTreeStatus,omitempty"`
+	ClusterTreeStatus *ClusterTreeStatus `json:"clusterTreeStatus,omitempty"`
 }
 
 type ClusterLinkOptions struct {
@@ -173,9 +173,15 @@ type ClusterList struct {
 }
 
 func (c *Cluster) IsP2P() bool {
+	if c.Spec.ClusterLinkOptions == nil {
+		return false
+	}
 	return c.Spec.ClusterLinkOptions.NetworkType == NetworkTypeP2P
 }
 
 func (c *Cluster) IsGateway() bool {
+	if c.Spec.ClusterLinkOptions == nil {
+		return false
+	}
 	return c.Spec.ClusterLinkOptions.NetworkType == NetWorkTypeGateWay
 }
diff --git a/pkg/apis/kosmos/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/kosmos/v1alpha1/zz_generated.deepcopy.go
diff --git a/pkg/clusterlink/controllers/cluster/cluster_controller.go b/pkg/clusterlink/controllers/cluster/cluster_controller.go
@@ -39,7 +39,6 @@ import (
 // KubeFlannelNetworkConfig
 const (
 	FlannelCNI             = "flannel"
-	KubeFlannelNamespace   = "kube-flannel"
 	KubeFlannelConfigMap   = "kube-flannel-cfg"
 	KubeFlannelNetworkConf = "net-conf.json"
 	KubeFlannelIPPool      = "Network"
@@ -371,7 +370,7 @@ func (c *Controller) initCalicoWatcherWithEtcdBackend(ctx context.Context, clust
 
 // todo by wuyingjun-lucky
 func (c *Controller) initFlannelInformer(context context.Context, cluster *clusterlinkv1alpha1.Cluster, kubeClient kubernetes.Interface) (SetClusterPodCIDRFun, error) {
-	informerFactory := informers.NewSharedInformerFactoryWithOptions(kubeClient, 0, informers.WithNamespace(KubeFlannelNamespace))
+	informerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
 	lister := informerFactory.Core().V1().ConfigMaps().Lister()
 	_, err := informerFactory.Core().V1().ConfigMaps().Informer().AddEventHandler(
 		cache.ResourceEventHandlerFuncs{

diff --git a/pkg/clusterlink/controllers/node/node_controller.go b/pkg/clusterlink/controllers/node/node_controller.go
@@ -43,15 +43,15 @@ type Reconciler struct {
 var predicatesFunc = predicate.Funcs{
 	CreateFunc: func(createEvent event.CreateEvent) bool {
 		node := createEvent.Object.(*corev1.Node)
-		return !utils.IsKosmosNode(node)
+		return !utils.IsKosmosNode(node) && !utils.IsExcludeNode(node)
 	},
 	UpdateFunc: func(updateEvent event.UpdateEvent) bool {
 		node := updateEvent.ObjectNew.(*corev1.Node)
-		return !utils.IsKosmosNode(node)
+		return !utils.IsKosmosNode(node) && !utils.IsExcludeNode(node)
 	},
 	DeleteFunc: func(deleteEvent event.DeleteEvent) bool {
 		node := deleteEvent.Object.(*corev1.Node)
-		return !utils.IsKosmosNode(node)
+		return !utils.IsKosmosNode(node) && !utils.IsExcludeNode(node)
 	},
 	GenericFunc: func(genericEvent event.GenericEvent) bool {
 		return false

diff --git a/pkg/clusterlink/network-manager/handlers/pod_routes.go b/pkg/clusterlink/network-manager/handlers/pod_routes.go
@@ -143,7 +143,17 @@ func BuildRoutes(ctx *Context, target *v1alpha1.ClusterNode, cidrs []string) {
 			}
 
 			gw := ctx.Filter.GetGatewayNodeByClusterName(n.Spec.ClusterName)
+			if gw == nil {
+				klog.Warning("cannot find gateway node, cluster name: %s", n.Spec.ClusterName)
+				continue
+			}
+
 			gwDev := ctx.GetDeviceFromResults(gw.Name, vxLocal)
+			if gwDev == nil {
+				klog.Warning("cannot find the gw dev, nodeName: %s, devName: %s", gw.Name, vxLocal)
+				continue
+			}
+
 			gwIP, _, err := net.ParseCIDR(gwDev.Addr)
 			if err != nil {
 				klog.Warning("cannot parse gw dev addr, nodeName: %s, devName: %s", gw.Name, vxLocal)

diff --git a/pkg/clusterlink/network-manager/handlers/vxlocal_mac_cache.go b/pkg/clusterlink/network-manager/handlers/vxlocal_mac_cache.go
@@ -20,6 +20,10 @@ func (h *VxLocalMacCache) Do(c *Context) (err error) {
 	for _, node := range nodes {
 		ipTypes := h.getSupportIPTypes(node, c)
 		gw := c.Filter.GetGatewayNodeByClusterName(node.Spec.ClusterName)
+		if gw == nil {
+			klog.Warning("cannot find gateway node, cluster name: %s", node.Spec.ClusterName)
+			continue
+		}
 
 		for _, ipType := range ipTypes {
 			fdb, arp, err := h.buildVxLocalCachesByNode(c, ipType, gw)

diff --git a/pkg/clustertree/cluster-manager/controllers/pod/leaf_pod_controller.go b/pkg/clustertree/cluster-manager/controllers/pod/leaf_pod_controller.go
@@ -8,6 +8,7 @@ import (
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/klog/v2"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
@@ -38,7 +39,7 @@ func (r *LeafPodReconciler) Reconcile(ctx context.Context, request reconcile.Req
 	if err := r.Get(ctx, request.NamespacedName, &pod); err != nil {
 		if apierrors.IsNotFound(err) {
 			// delete pod in root
-			if err := r.safeDeletePodInRootCluster(ctx, request); err != nil {
+			if err := DeletePodInRootCluster(ctx, request.NamespacedName, r.RootClient); err != nil {
 				return reconcile.Result{RequeueAfter: LeafPodRequeueTime}, nil
 			}
 			return reconcile.Result{}, nil
@@ -96,9 +97,9 @@ func NewLeafDeleteOption(pod *corev1.Pod) client.DeleteOption {
 	}
 }
 
-func (r *LeafPodReconciler) safeDeletePodInRootCluster(ctx context.Context, request reconcile.Request) error {
+func DeletePodInRootCluster(ctx context.Context, rootnamespacedname types.NamespacedName, rootClient client.Client) error {
 	rPod := corev1.Pod{}
-	err := r.RootClient.Get(ctx, request.NamespacedName, &rPod)
+	err := rootClient.Get(ctx, rootnamespacedname, &rPod)
 
 	if err != nil {
 		if apierrors.IsNotFound(err) {
@@ -111,7 +112,7 @@ func (r *LeafPodReconciler) safeDeletePodInRootCluster(ctx context.Context, requ
 	rPodCopy := rPod.DeepCopy()
 	deleteOption := NewRootDeleteOption(rPodCopy)
 
-	if err := r.RootClient.Delete(ctx, rPodCopy, deleteOption); err != nil {
+	if err := rootClient.Delete(ctx, rPodCopy, deleteOption); err != nil {
 		if !apierrors.IsNotFound(err) {
 			return err
 		}

diff --git a/pkg/clustertree/cluster-manager/controllers/pod/root_pod_controller.go b/pkg/clustertree/cluster-manager/controllers/pod/root_pod_controller.go
@@ -128,7 +128,7 @@ func (r *RootPodReconciler) Reconcile(ctx context.Context, request reconcile.Req
 					// wait for leaf resource init
 					return reconcile.Result{RequeueAfter: RootPodRequeueTime}, nil
 				}
-				if err := r.DeletePodInLeafCluster(ctx, lr, request.NamespacedName); err != nil {
+				if err := r.DeletePodInLeafCluster(ctx, lr, request.NamespacedName, false); err != nil {
 					klog.Errorf("delete pod in leaf error[1]: %v,  %s", err, request.NamespacedName)
 					return reconcile.Result{RequeueAfter: RootPodRequeueTime}, nil
 				}
@@ -185,7 +185,7 @@ func (r *RootPodReconciler) Reconcile(ctx context.Context, request reconcile.Req
 
 	// delete pod in leaf
 	if !rootpod.GetDeletionTimestamp().IsZero() {
-		if err := r.DeletePodInLeafCluster(ctx, lr, request.NamespacedName); err != nil {
+		if err := r.DeletePodInLeafCluster(ctx, lr, request.NamespacedName, true); err != nil {
 			klog.Errorf("delete pod in leaf error[1]: %v,  %s", err, request.NamespacedName)
 			return reconcile.Result{RequeueAfter: RootPodRequeueTime}, nil
 		}
@@ -802,14 +802,21 @@ func (r *RootPodReconciler) UpdatePodInLeafCluster(ctx context.Context, lr *leaf
 	return nil
 }
 
-func (r *RootPodReconciler) DeletePodInLeafCluster(ctx context.Context, lr *leafUtils.LeafResource, rootnamespacedname types.NamespacedName) error {
+func (r *RootPodReconciler) DeletePodInLeafCluster(ctx context.Context, lr *leafUtils.LeafResource, rootnamespacedname types.NamespacedName, cleanflag bool) error {
 	klog.V(4).Infof("Deleting pod %v/%+v", rootnamespacedname.Namespace, rootnamespacedname.Name)
 	leafPod := &corev1.Pod{}
 
+	cleanRootPodFunc := func() error {
+		return DeletePodInRootCluster(ctx, rootnamespacedname, r.Client)
+	}
+
 	err := lr.Client.Get(ctx, rootnamespacedname, leafPod)
 
 	if err != nil {
 		if errors.IsNotFound(err) {
+			if cleanflag {
+				return cleanRootPodFunc()
+			}
 			return nil
 		}
 		return err
@@ -825,6 +832,9 @@ func (r *RootPodReconciler) DeletePodInLeafCluster(ctx context.Context, lr *leaf
 	if err != nil {
 		if errors.IsNotFound(err) {
 			klog.V(4).Infof("Tried to delete pod %s/%s, but it did not exist in the cluster", leafPod.Namespace, leafPod.Name)
+			if cleanflag {
+				return cleanRootPodFunc()
+			}
 			return nil
 		}
 		return fmt.Errorf("could not delete pod: %v", err)

diff --git a/pkg/generated/openapi/zz_generated.openapi.go b/pkg/generated/openapi/zz_generated.openapi.go
diff --git a/pkg/kosmosctl/install/install.go b/pkg/kosmosctl/install/install.go
@@ -404,30 +404,6 @@ func (o *CommandInstallOptions) runClustertree() error {
 	}
 	klog.Info("Create CRD " + clustertreeCluster.Name + " successful.")
 
-	serviceExport, err := util.GenerateCustomResourceDefinition(manifest.ServiceExport, nil)
-	if err != nil {
-		return err
-	}
-	_, err = o.K8sExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().Create(context.Background(), serviceExport, metav1.CreateOptions{})
-	if err != nil {
-		if !apierrors.IsAlreadyExists(err) {
-			return fmt.Errorf("kosmosctl install clustertree run error, crd options failed: %v", err)
-		}
-	}
-	klog.Info("Create CRD " + serviceExport.Name + " successful.")
-
-	serviceImport, err := util.GenerateCustomResourceDefinition(manifest.ServiceImport, nil)
-	if err != nil {
-		return err
-	}
-	_, err = o.K8sExtensionsClient.ApiextensionsV1().CustomResourceDefinitions().Create(context.Background(), serviceImport, metav1.CreateOptions{})
-	if err != nil {
-		if !apierrors.IsAlreadyExists(err) {
-			return fmt.Errorf("kosmosctl install clustertree run error, crd options failed: %v", err)
-		}
-	}
-	klog.Info("Create CRD " + serviceImport.Name + " successful.")
-
 	klog.Info("Start creating kosmos-clustertree ConfigMap...")
 	clustertreeConfigMap := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{