Optimize spiderpool leader lease election

Signed-off-by: ty-dc <[email protected]>

pkg/applicationcontroller/app_controller.go

@@ -109,12 +109,11 @@ func (sac *SubnetAppController) SetupInformer(ctx context.Context, client kubern
 			select {
 			case <-ctx.Done():
 				return
-			default:
-			}
-
-			if !leader.IsElected() {
-				time.Sleep(sac.LeaderRetryElectGap)
-				continue
+			case isLeader := <-leader.IsElected():
+				if !isLeader {
+					time.Sleep(sac.LeaderRetryElectGap)
+					continue
+				}
 			}
 
 			innerCtx, innerCancel := context.WithCancel(ctx)
@@ -123,12 +122,13 @@ func (sac *SubnetAppController) SetupInformer(ctx context.Context, client kubern
 					select {
 					case <-innerCtx.Done():
 						return
-					default:
-					}
-
-					if !leader.IsElected() {
-						logger.Warn("Leader lost, stop Subnet App informer")
-						innerCancel()
+					case isLeader := <-leader.IsElected():
+						if !isLeader {
+							logger.Warn("Leader lost, stop SpiderSubnet informer")
+							innerCancel()
+							return
+						}
+					case <-ctx.Done():
 						return
 					}
 					time.Sleep(sac.LeaderRetryElectGap)

pkg/coordinatormanager/coordinator_informer.go

@@ -132,12 +132,11 @@ func (cc *CoordinatorController) SetupInformer(
 			select {
 			case <-ctx.Done():
 				return
-			default:
-			}
-
-			if !leader.IsElected() {
-				time.Sleep(cc.LeaderRetryElectGap)
-				continue
+			case isLeader := <-leader.IsElected():
+				if !isLeader {
+					time.Sleep(cc.LeaderRetryElectGap)
+					continue
+				}
 			}
 
 			innerCtx, innerCancel := context.WithCancel(ctx)
@@ -146,12 +145,13 @@ func (cc *CoordinatorController) SetupInformer(
 					select {
 					case <-innerCtx.Done():
 						return
-					default:
-					}
-
-					if !leader.IsElected() {
-						InformerLogger.Warn("Leader lost, stop Coordinator informer")
-						innerCancel()
+					case isLeader := <-leader.IsElected():
+						if !isLeader {
+							InformerLogger.Warn("Leader lost, stop Coordinator informer")
+							innerCancel()
+							return
+						}
+					case <-ctx.Done():
 						return
 					}
 					time.Sleep(cc.LeaderRetryElectGap)

pkg/dra/dra-controller/controller.go

@@ -30,12 +30,11 @@ func StartController(ctx context.Context,
 			select {
 			case <-ctx.Done():
 				return
-			default:
-			}
-
-			if !leader.IsElected() {
-				time.Sleep(leaderRetryElectGap)
-				continue
+			case isLeader := <-leader.IsElected():
+				if !isLeader {
+					time.Sleep(leaderRetryElectGap)
+					continue
+				}
 			}
 
 			innerCtx, innerCancel := context.WithCancel(ctx)
@@ -44,11 +43,12 @@ func StartController(ctx context.Context,
 					select {
 					case <-innerCtx.Done():
 						return
-					default:
-					}
-
-					if !leader.IsElected() {
-						innerCancel()
+					case isLeader := <-leader.IsElected():
+						if !isLeader {
+							innerCancel()
+							return
+						}
+					case <-ctx.Done():
 						return
 					}
 					time.Sleep(leaderRetryElectGap)
@@ -58,6 +58,7 @@ func StartController(ctx context.Context,
 			informerFactory.Start(innerCtx.Done())
 			controller.Run(1)
 		}
+
 	}()
 	return nil
 }

pkg/election/lease_election.go

@@ -25,7 +25,7 @@ var logger *zap.Logger
 type SpiderLeaseElector interface {
 	Run(ctx context.Context, clientSet kubernetes.Interface) error
 	// IsElected returns a boolean value to check current Elector whether is a leader
-	IsElected() bool
+	IsElected() chan bool
 	GetLeader() string
 }
 
@@ -40,7 +40,7 @@ type SpiderLeader struct {
 	leaseRetryPeriod    time.Duration
 	leaderRetryElectGap time.Duration
 
-	isLeader      bool
+	isLeader      chan bool
 	leaderElector *leaderelection.LeaderElector
 }
 
@@ -81,7 +81,7 @@ func NewLeaseElector(leaseLockNS, leaseLockName, leaseLockIdentity string,
 	}
 
 	sl := &SpiderLeader{
-		isLeader:            false,
+		isLeader:            make(chan bool, 1),
 		leaseLockName:       leaseLockName,
 		leaseLockNamespace:  leaseLockNS,
 		leaseLockIdentity:   leaseLockIdentity,
@@ -128,14 +128,14 @@ func (sl *SpiderLeader) register(clientSet kubernetes.Interface) error {
 		Callbacks: leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(_ context.Context) {
 				sl.Lock()
-				sl.isLeader = true
+				sl.isLeader <- true
 				sl.Unlock()
 				logger.Sugar().Infof("leader elected: %s/%s/%s", sl.leaseLockNamespace, sl.leaseLockName, sl.leaseLockIdentity)
 			},
 			OnStoppedLeading: func() {
 				// we can do cleanup here
 				sl.Lock()
-				sl.isLeader = false
+				sl.isLeader <- false
 				sl.Unlock()
 				logger.Sugar().Warnf("leader lost: %s/%s/%s", sl.leaseLockNamespace, sl.leaseLockName, sl.leaseLockIdentity)
 			},
@@ -150,7 +150,7 @@ func (sl *SpiderLeader) register(clientSet kubernetes.Interface) error {
 	return nil
 }
 
-func (sl *SpiderLeader) IsElected() bool {
+func (sl *SpiderLeader) IsElected() chan bool {
 	sl.RLock()
 	defer sl.RUnlock()
 

pkg/gcmanager/gc_manager.go

@@ -185,17 +185,24 @@ func (s *SpiderGC) Health() bool {
 	ctx, cancelFunc := context.WithTimeout(context.TODO(), waitForCacheSyncTimeout)
 	defer cancelFunc()
 
-	if s.leader.IsElected() {
-		if s.informerFactory == nil {
-			logger.Warn("the IP-GC manager pod informer is not ready")
-			return false
+	select {
+	case isLeader := <-s.leader.IsElected():
+		if !isLeader {
+			return true
 		}
+	case <-ctx.Done():
+		return false
+	}
 
-		waitForCacheSync := s.informerFactory.WaitForCacheSync(ctx.Done())
-		for _, isCacheSync := range waitForCacheSync {
-			if !isCacheSync {
-				return false
-			}
+	if s.informerFactory == nil {
+		logger.Warn("the IP-GC manager pod informer is not ready")
+		return false
+	}
+
+	waitForCacheSync := s.informerFactory.WaitForCacheSync(ctx.Done())
+	for _, isCacheSync := range waitForCacheSync {
+		if !isCacheSync {
+			return false
 		}
 	}
 

pkg/gcmanager/pod_informer.go

@@ -20,69 +20,73 @@ func (s *SpiderGC) startPodInformer(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		default:
-		}
-
-		if !s.leader.IsElected() {
-			time.Sleep(s.gcConfig.LeaderRetryElectGap)
-			continue
-		}
-
-		innerCtx, innerCancel := context.WithCancel(ctx)
-		go func() {
-			for {
-				select {
-				case <-innerCtx.Done():
-					return
-				default:
-				}
+		case isLeader := <-s.leader.IsElected():
+			// Proceed only if this pod is the leader
+			if !isLeader {
+				logger.Warn("Not the leader, waiting for leadership.")
+				time.Sleep(s.gcConfig.LeaderRetryElectGap)
+				continue
+			}
 
-				if !s.leader.IsElected() {
-					logger.Warn("Leader lost, stop IP GC pod informer")
-					innerCancel()
-					return
+			innerCtx, innerCancel := context.WithCancel(ctx)
+			defer innerCancel()
+
+			go func() {
+				for {
+					select {
+					case <-innerCtx.Done():
+						return
+					case isLeader := <-s.leader.IsElected():
+						if !isLeader {
+							logger.Warn("Leader lost, stopping IP GC pod informer.")
+							innerCancel()
+							return
+						}
+					}
+					time.Sleep(s.gcConfig.LeaderRetryElectGap)
 				}
-				time.Sleep(s.gcConfig.LeaderRetryElectGap)
+			}()
+
+			logger.Info("Create Pod informer")
+			informerFactory := informers.NewSharedInformerFactory(s.k8ClientSet, 0)
+			podInformer := informerFactory.Core().V1().Pods().Informer()
+			_, err := podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+				AddFunc:    s.onPodAdd,
+				UpdateFunc: s.onPodUpdate,
+				DeleteFunc: s.onPodDel,
+			})
+			if err != nil {
+				logger.Error(err.Error())
+				innerCancel()
+				continue
 			}
-		}()
-
-		logger.Info("create Pod informer")
-		informerFactory := informers.NewSharedInformerFactory(s.k8ClientSet, 0)
-		podInformer := informerFactory.Core().V1().Pods().Informer()
-		_, err := podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-			AddFunc:    s.onPodAdd,
-			UpdateFunc: s.onPodUpdate,
-			DeleteFunc: s.onPodDel,
-		})
-		if nil != err {
-			logger.Error(err.Error())
-			innerCancel()
-			continue
-		}
-		s.informerFactory = informerFactory
-		informerFactory.Start(innerCtx.Done())
-
-		// Let the leader trigger IP GC scan all.
-		// When the spiderpool-controller restarted, it will trigger IP GC scan all first.
-		// If the pod informer not starts and the user delete some pods, this will lead to IP leakage.
-		logger.Debug("try to trigger scan all with leader elected")
-		cacheSync := cache.WaitForCacheSync(innerCtx.Done())
-		if !cacheSync {
-			innerCancel()
-			continue
-		}
-		s.gcSignal <- struct{}{}
+			s.informerFactory = informerFactory
+			informerFactory.Start(innerCtx.Done())
+
+			logger.Debug("Triggering scan all with leader elected")
+			cacheSync := cache.WaitForCacheSync(innerCtx.Done())
+			if !cacheSync {
+				innerCancel()
+				continue
+			}
+			// Notify the system to trigger a GC scan
+			s.gcSignal <- struct{}{}
 
-		<-innerCtx.Done()
-		logger.Error("k8s pod informer broken")
+			// Wait for informer context to be done
+			<-innerCtx.Done()
+			logger.Error("K8s pod informer broken, restarting process")
+		}
 	}
 }
 
 // onPodAdd represents Pod informer Add Event
 func (s *SpiderGC) onPodAdd(obj interface{}) {
 	// backup controller could be elected as master
-	if !s.leader.IsElected() {
-		return
+	select {
+	case isLeader := <-s.leader.IsElected():
+		if !isLeader {
+			return
+		}
 	}
 
 	pod := obj.(*corev1.Pod)
@@ -104,8 +108,11 @@ func (s *SpiderGC) onPodAdd(obj interface{}) {
 // onPodUpdate represents Pod informer Update Event
 func (s *SpiderGC) onPodUpdate(oldObj interface{}, newObj interface{}) {
 	// backup controller could be elected as master
-	if !s.leader.IsElected() {
-		return
+	select {
+	case isLeader := <-s.leader.IsElected():
+		if !isLeader {
+			return
+		}
 	}
 
 	oldPod := oldObj.(*corev1.Pod)
@@ -128,8 +135,11 @@ func (s *SpiderGC) onPodUpdate(oldObj interface{}, newObj interface{}) {
 // onPodDel represents Pod informer Delete Event
 func (s *SpiderGC) onPodDel(obj interface{}) {
 	// backup controller could be elected as master
-	if !s.leader.IsElected() {
-		return
+	select {
+	case isLeader := <-s.leader.IsElected():
+		if !isLeader {
+			return
+		}
 	}
 
 	pod := obj.(*corev1.Pod)