nodelifecycle

1.入口

1.1.start

• startNodeLifecycleController()负责实例化nodeLifecycleController，基于informer及monitor监听资源变化。

• // NewNodeLifecycleController returns a new taint controller.
  func NewNodeLifecycleController(...) (*Controller, error) {
  	...
  
  	// 实例化nlc
  	nc := &Controller{
  		...
  		knownNodeSet:                make(map[string]*v1.Node),		// 已知node集合
  		nodeHealthMap:               newNodeHealthMap(),			// 健康node缓存
  		...
          nodeMonitorPeriod:           nodeMonitorPeriod,				// 心跳、lease及nodeCondition采样间隔(5min)
          nodeStartupGracePeriod:      nodeStartupGracePeriod,		// 新节点不健康的宽限期(60s)
          nodeMonitorGracePeriod:      nodeMonitorGracePeriod,		// 最后一次心跳到失联的宽限期(40s)
  		nodeUpdateWorkerSize:        scheduler.UpdateWorkerSize,	// worker数量
  		zoneNoExecuteTainter:        make(map[string]*scheduler.RateLimitedTimedQueue),	// zone延迟驱逐任务队列
  		nodesToRetry:                sync.Map{},					// 待重试node缓存
  		zoneStates:                  make(map[string]ZoneState),	// zone状态缓存
          evictionLimiterQPS:          evictionLimiterQPS,			// 正常情况下驱逐速率(0.1/s)
          secondaryEvictionLimiterQPS: secondaryEvictionLimiterQPS,	// zone异常或部分故障时驱逐速率(0.01/s)
          largeClusterThreshold:       largeClusterThreshold,			// 驱逐限流阈值(50个异常节点)
          unhealthyZoneThreshold:      unhealthyZoneThreshold,		// zone视为不健康的异常节点比例(0.55)
  		nodeUpdateQueue:             workqueue.NewNamed("node_lifecycle_controller"),
  		podUpdateQueue:              workqueue.NewNamedRateLimitingQueue(DefaultControllerRateLimiter(), "nlc"),
  	}
  
      // 注册node驱逐速率及zone状态计算回调
  	nc.enterPartialDisruptionFunc = nc.ReducedQPSFunc
  	nc.enterFullDisruptionFunc = nc.HealthyQPSFunc
  	nc.computeZoneStateFunc = nc.ComputeZoneState
  
      // podInformer监听,相关对象推入queue
  	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc: func(obj interface{}) {
  			...
  			nc.podUpdated(nil, pod)
  			if nc.taintManager != nil {
  				nc.taintManager.PodUpdated(nil, pod)
  			}
  		},
  		UpdateFunc: func(prev, obj interface{}) {
  			...
  			nc.podUpdated(prevPod, newPod)
  			if nc.taintManager != nil {
  				nc.taintManager.PodUpdated(prevPod, newPod)
  			}
  		},
  		DeleteFunc: func(obj interface{}) {
  			...
              
  			nc.podUpdated(pod, nil)
  			if nc.taintManager != nil {
  				nc.taintManager.PodUpdated(pod, nil)
  			}
  		},
  	})
      ...
      
      // pod缓存索引
  	podInformer.Informer().AddIndexers(cache.Indexers{
  		nodeNameKeyIndex: func(obj interface{}) ([]string, error) {
  			...
              
  			return []string{pod.Spec.NodeName}, nil
  		},
  	})
  
  	podIndexer := podInformer.Informer().GetIndexer()
  
      
      // 基于node分组获取Pod回调
      nc.getPodsAssignedToNode = func(nodeName string) ([]*v1.Pod, error) {
  		objs, err := podIndexer.ByIndex(nodeNameKeyIndex, nodeName)
  		...
          
  		pods := make([]*v1.Pod, 0, len(objs))
  		for _, obj := range objs {
  			pod, ok := obj.(*v1.Pod)
  			...
              
  			pods = append(pods, pod)
  		}
  		return pods, nil
  	}
      
  	nc.podLister = podInformer.Lister()
  	nc.nodeLister = nodeInformer.Lister()
  
      // 初始化taintManager
  	nc.taintManager = scheduler.NewNoExecuteTaintManager(ctx, kubeClient, nc.podLister, nc.nodeLister, nc.getPodsAssignedToNode)
      
      // 注册nodeInformer回调
  	nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc: controllerutil.CreateAddNodeHandler(func(node *v1.Node) error {
  			nc.taintManager.NodeUpdated(nil, node)
              nc.nodeUpdateQueue.Add(node.Name)
  			return nil
  		}),
  		UpdateFunc: controllerutil.CreateUpdateNodeHandler(func(oldNode, newNode *v1.Node) error {
  			nc.taintManager.NodeUpdated(oldNode, newNode)
              nc.nodeUpdateQueue.Add(newNode.Name)
  			return nil
  		}),
  		DeleteFunc: controllerutil.CreateDeleteNodeHandler(func(node *v1.Node) error {
  			nc.taintManager.NodeUpdated(node, nil)
              nc.nodesToRetry.Delete(node.Name)
  			return nil
  		}),
  	})
  	...
  
  	nc.leaseLister = leaseInformer.Lister()
  	...
  
  	nc.daemonSetStore = daemonSetInformer.Lister()
  	...
  
  	return nc, nil
  }
  
  func startNodeLifecycleController(...) (controller.Interface, bool, error) {
  	lifecycleController, err := lifecyclecontroller.NewNodeLifecycleController(...)
  	...
      
  	go lifecycleController.Run(ctx)
  	return nil, true, nil
  }
  

• 

• 注意

  taintManager和nodeLifeCycle controller都会watch node变化，但处理策略不同，后面会提到

1.2.run

• nc.Run()会启动多个协程运行taintManager和多个worker，以消费node/pod事件、触发驱逐及监控nodelifecycle监控状态。

• // Run starts an asynchronous loop that monitors the status of cluster nodes.
  func (nc *Controller) Run(ctx context.Context) {
  	...
  
  	// Close node update queue to cleanup go routine.
  	defer nc.nodeUpdateQueue.ShutDown()
  	defer nc.podUpdateQueue.ShutDown()
  	...
  
      // 阻塞至同步完成
  	if !cache.WaitForNamedCacheSync("taint", ctx.Done(), nc.leaseInformerSynced, nc.nodeInformerSynced, nc.podInformerSynced, nc.daemonSetInformerSynced) {
  		return
  	}
  
      // 启动taintManager
  	go nc.taintManager.Run(ctx)
  
  	// 激活8个worker处理nodeUpdateQueue事件
  	for i := 0; i < scheduler.UpdateWorkerSize; i++ {
  		go wait.UntilWithContext(ctx, nc.doNodeProcessingPassWorker, time.Second)
  	}
  
      
      // 激活4个worker处理podUpdateQueue事件
  	for i := 0; i < podUpdateWorkerSize; i++ {
  		go wait.UntilWithContext(ctx, nc.doPodProcessingWorker, time.Second)
  	}
  
  	// 节流器,周期处理taint注入
  	go wait.UntilWithContext(ctx, nc.doNoExecuteTaintingPass, scheduler.NodeEvictionPeriod)
  
  	// 执行monitor监控监控
  	go wait.UntilWithContext(ctx, func(ctx context.Context) {
  		nc.monitorNodeHealth(ctx)
          ...
  	}, nc.nodeMonitorPeriod)
  
  	<-ctx.Done()
  }
  

• 

• 注意

  run会启动多个协程处理taint注入、取消、健康监控及驱逐

2.taintMgr

2.1.newMgr

• taintManager会监听node资源变化，基于node粒度分组Pod，检查Pod污点容忍及推容即将驱逐的Pod至evictQueue，由异步协程驱逐。

• // creates a new NoExecuteTaintManager that will use passed clientset to communicate with the API server.
  func NewNoExecuteTaintManager(...) *NoExecuteTaintManager {
  	...
  
      // 初始化taintManager
  	tm := &NoExecuteTaintManager{
  		client:                c,
  		...
  		podLister:             podLister,
  		nodeLister:            nodeLister,
  		getPodsAssignedToNode: getPodsAssignedToNode,					// pod分组器
  		taintedNodes:          make(map[string][]v1.Taint),				// 已注入污点的node
  
  		nodeUpdateQueue: workqueue.NewNamed("noexec_taint_node"),
  		podUpdateQueue:  workqueue.NewNamed("noexec_taint_pod"),		// 用不到
  	}	
      
      // 驱逐队列
  	tm.taintEvictionQueue = CreateWorkerQueue(deletePodHandler(c, tm.emitPodDeletionEvent))
  
  	return tm
  }
  
  // creates a new TimedWorkerQueue for workers that will execute given function `f`.
  func CreateWorkerQueue(f func(ctx context.Context, args *WorkArgs) error) *TimedWorkerQueue {
  	return &TimedWorkerQueue{
          // 驱逐的Pod都放在这里
  		workers:  make(map[string]*TimedWorker),
  		workFunc: f,
  		clock:    clock.RealClock{},
  	}
  }
  
  func deletePodHandler(...) func(ctx context.Context, args *WorkArgs) error {
  	return func(ctx context.Context, args *WorkArgs) error {
  		...
          
          // 尝试5次
  		for i := 0; i < retries; i++ {
  			// 标记PodDisruptionCondition及删除Pod
              err = addConditionAndDeletePod(ctx, c, name, ns)
  			if err == nil {
  				break
  			}
  			time.Sleep(10 * time.Millisecond)
  		}
  		return err
  	}
  }
  

• 

• 补充

  nodeUpdateQueue事件会被taintManaget主循环消费，检出的evictPod推入timedWorkerQueue供deletePodHandler删除

2.2.run

• taintManager.Run()会处理nodeUpdateQueue事件转发到nodeUpdateChannels，执行tc.worker消费nodeChannel数据进行驱逐。

• // Run starts NoExecuteTaintManager which will run in loop until `stopCh` is closed.
  func (tc *NoExecuteTaintManager) Run(ctx context.Context) {
  	...
  
  	defer tc.nodeUpdateQueue.ShutDown()
  	defer tc.podUpdateQueue.ShutDown()
  
      // 初始化8个nodeUpdateChan和podUpdateChan
  	for i := 0; i < UpdateWorkerSize; i++ {
  		tc.nodeUpdateChannels = append(tc.nodeUpdateChannels, make(chan nodeUpdateItem, NodeUpdateChannelSize))
  		tc.podUpdateChannels = append(tc.podUpdateChannels, make(chan podUpdateItem, podUpdateChannelSize))
  	}
  
  	// taking work items out of the workqueues and putting them into channels.
  	go func(stopCh <-chan struct{}) {
  		for {
              // nodeUpdateQueue消费
  			item, shutdown := tc.nodeUpdateQueue.Get()
  			...
              
  			nodeUpdate := item.(nodeUpdateItem)
  			hash := hash(nodeUpdate.nodeName, UpdateWorkerSize)		// hash%8,决定放在哪个nodeUpdateChan
              
  			select {
              // 终止
  			case <-stopCh:
  				tc.nodeUpdateQueue.Done(item)
  				return
              // nodeUpdate转入nodeUpdateChan
  			case tc.nodeUpdateChannels[hash] <- nodeUpdate:
  			}
  		}
  	}(ctx.Done())
  
      // taking work items out of the workqueues and putting them into channels.
  	go func(stopCh <-chan struct{}) {
  		for {
              // podUpdateQueue消费
  			item, shutdown := tc.podUpdateQueue.Get()
  			...
              
  			podUpdate := item.(podUpdateItem)
  			hash := hash(podUpdate.nodeName, UpdateWorkerSize)		// hash%8,决定放在哪个podUpdateChan
  			select {
              // 终止
  			case <-stopCh:
  				tc.podUpdateQueue.Done(item)
  				return
              // podUpdate转入podUpdateChan
  			case tc.podUpdateChannels[hash] <- podUpdate:
  			}
  		}
  	}(ctx.Done())
  	...
      
      // 激活8个worker消费updateChan
  	for i := 0; i < UpdateWorkerSize; i++ {
  		go tc.worker(ctx, i, wg.Done, ctx.Done())
  	}
  	...
  }
  

• 

• 注意

  nodeUpdateQueue和podUpdateQueue仅暂存资源，最后还是会路由到updateChan供worker并发处理

2.3.worker

• tc.worker()执行handleNodeUpdate和handlePodUpdate处理updateChan数据，优先处理nodeUpdateChan的数据。

• func (tc *NoExecuteTaintManager) worker(ctx context.Context, worker int, done func(), stopCh <-chan struct{}) {
  	
  	for {
  		select {
  		case <-stopCh:
  			return
          // node事件
  		case nodeUpdate := <-tc.nodeUpdateChannels[worker]:
  			tc.handleNodeUpdate(ctx, nodeUpdate)
  			tc.nodeUpdateQueue.Done(nodeUpdate)
          // pod事件
  		case podUpdate := <-tc.podUpdateChannels[worker]:
  
  		priority:
              // 优先处理node事件
  			for {
  				select {
  				case nodeUpdate := <-tc.nodeUpdateChannels[worker]:
  					tc.handleNodeUpdate(ctx, nodeUpdate)
  					tc.nodeUpdateQueue.Done(nodeUpdate)
  				default:
  					break priority
  				}
  			}
              
  			// node处理完毕再处理Pod事件
  			tc.handlePodUpdate(ctx, podUpdate)
  			tc.podUpdateQueue.Done(podUpdate)
  		}
  	}
  }
  
  func (tc *NoExecuteTaintManager) handleNodeUpdate(ctx context.Context, nodeUpdate nodeUpdateItem) {
  	node, err := tc.nodeLister.Get(nodeUpdate.nodeName)
      
      if apierrors.IsNotFound(err) {
          ...
          delete(tc.taintedNodes, nodeUpdate.nodeName)
          return
      }
  	...
  
  	// 获取node taints
  	taints := getNoExecuteTaints(node.Spec.Taints)
  	func() {
  		...
          
          // 更新taintedNodes
  		if len(taints) == 0 {
  			delete(tc.taintedNodes, node.Name)
  		} else {
  			tc.taintedNodes[node.Name] = taints
  		}
  	}()
  
  	// 获取node节点上的Pod
  	pods, err := tc.getPodsAssignedToNode(node.Name)
  	...
      
  	if len(pods) == 0 {
  		return
  	}
      
  	// 终止驱逐及生成事件
  	if len(taints) == 0 {
  		for i := range pods {
  			tc.cancelWorkWithEvent(logger, NamespacedName{Namespace: pods[i].Namespace, Name: pods[i].Name})
  		}
  		return
  	}
  	...
      
      // 检查Pod驱逐条件及注册worker
  	for _, pod := range pods {
  		podNamespacedName := types.NamespacedName{Namespace: pod.Namespace, Name: pod.Name}
  		tc.processPodOnNode(ctx, podNamespacedName, node.Name, pod.Spec.Tolerations, taints, now)
  	}
  }
  
  func (tc *NoExecuteTaintManager) handlePodUpdate(ctx context.Context, podUpdate podUpdateItem) {
  	pod, err := tc.podLister.Pods(podUpdate.podNamespace).Get(podUpdate.podName)
      ...
      
      // Pod不存在终止驱逐及生成事件
      if apierrors.IsNotFound(err) {
          ...
          tc.cancelWorkWithEvent(logger, podNamespacedName)
          return
      }
  	...
  
  	// 已经调度到其它节点
  	if pod.Spec.NodeName != podUpdate.nodeName {
  		return
  	}
  	...
      
      // 未调到任何节点
  	if nodeName == "" {
  		return
  	}
      
      // node上有驱逐taint
  	taints, ok := func() ([]v1.Taint, bool) {
  		...
  		taints, ok := tc.taintedNodes[nodeName]
  		return taints, ok
  	}()
  	...
      
      // 执行驱逐检查
  	tc.processPodOnNode(ctx, podNamespacedName, nodeName, pod.Spec.Tolerations, taints, time.Now())
  }
  

• 

• 注意

  tc.worker基于channel的node/pod触发驱逐检查，Pod满足驱逐条件会注册evictTask执行workFunc实现驱逐

2.4.process

• tc.processPodOnNode()会基于Pod和node污点及容忍度检查驱逐条件，满足驱逐的Pod会注册evictWork执行workFunc删除。

• func (tc *NoExecuteTaintManager) processPodOnNode(...) {
  	// node没有驱逐taint
  	if len(taints) == 0 {
          // 取消驱逐worker
  		tc.cancelWorkWithEvent(logger, podNamespacedName)
  	}
      
      // 检查Pod针对taint容忍度
  	allTolerated, usedTolerations := v1helper.GetMatchingTolerations(taints, tolerations)
      // 未完全匹配
  	if !allTolerated {
  		// 取消旧的驱逐worker
  		tc.cancelWorkWithEvent(logger, podNamespacedName)
          // 注册新的驱逐worker
  		tc.taintEvictionQueue.AddWork(ctx, NewWorkArgs(pod.Name, pod.Namespace), time.Now(), time.Now())
  		return
  	}
      
      // 获取容忍时间
  	minTolerationTime := getMinTolerationTime(usedTolerations)
  	
      // 一直容忍
  	if minTolerationTime < 0 {
  		// 取消驱逐worker
  		tc.cancelWorkWithEvent(logger, podNamespacedName)
  		return
  	}
  
      // 最近一次驱逐计划
  	startTime := now
  	triggerTime := startTime.Add(minTolerationTime)
      
      // 获取已注册的驱逐worker
  	scheduledEviction := tc.taintEvictionQueue.GetWorkerUnsafe(podNamespacedName.String())
  	if scheduledEviction != nil {
          // 旧计划在最近计划前,沿用旧计划
  		startTime = scheduledEviction.CreatedAt
  		if startTime.Add(minTolerationTime).Before(triggerTime) {
  			return
  		}
          
          // 否则先取消驱逐worker
  		tc.cancelWorkWithEvent(logger, podNamespacedName)
  	}
      
      // 再基于新计划注册驱逐worker
  	tc.taintEvictionQueue.AddWork(ctx, NewWorkArgs(pod.Name, pod.Namespace), startTime, triggerTime)
  }
  
  // AddWork adds a work to the WorkerQueue which will be executed not earlier than `fireAt`.
  func (q *TimedWorkerQueue) AddWork(ctx context.Context, args *WorkArgs, createdAt time.Time, fireAt time.Time) {
  
  	q.Lock()
  	defer q.Unlock()
      
      // 驱逐worker存在
  	if _, exists := q.workers[key]; exists {
  		return
  	}
      
      // 注册驱逐worker(执行+清理worker)
  	worker := createWorker(ctx, args, createdAt, fireAt, q.getWrappedWorkerFunc(key), q.clock)
  	q.workers[key] = worker
  }
  

• 

• 注意

  tc.processPodOnNode()会检查Pod对于污点的容忍，无法容忍或延迟容忍会注册及执行evictworker，执行成功或5 次重试后清理

3.doProcess

3.1.doNodeProcess

• nc.doNodeProcessingPassWorker()会处理nc.nodeUpdateQueue事件，为node添加合适的NoSchedule taint及label。

• func (nc *Controller) doNodeProcessingPassWorker(ctx context.Context) {
  	logger := klog.FromContext(ctx)
  	for {
  		obj, shutdown := nc.nodeUpdateQueue.Get()
  		...
          
          // taint处理
  		nodeName := obj.(string)
  		nc.doNoScheduleTaintingPass(ctx, nodeName)
          ...
          
          // 补充labels:
          //      beta.kubernetes.io/arch: amd64
          //      beta.kubernetes.io/os: linux
          //      kubernetes.io/arch: amd64
          //      kubernetes.io/os: linux
  		nc.reconcileNodeLabels(ctx, nodeName)
          ...
          
  		nc.nodeUpdateQueue.Done(nodeName)
  	}
  }
  
  func (nc *Controller) doNoScheduleTaintingPass(ctx context.Context, nodeName string) error {
  	node, err := nc.nodeLister.Get(nodeName)
  	...
  
  	// node condition转为taints
  	for _, condition := range node.Status.Conditions {
  		if taintMap, found := nodeConditionToTaintKeyStatusMap[condition.Type]; found {
  			if taintKey, found := taintMap[condition.Status]; found {
  				taints = append(taints, v1.Taint{
  					Key:    taintKey,
  					Effect: v1.TaintEffectNoSchedule,
  				})
  			}
  		}
  	}
      
      // node不可调度,补充Unschedulable taint
  	if node.Spec.Unschedulable {
  		taints = append(taints, v1.Taint{
  			Key:    v1.TaintNodeUnschedulable,
  			Effect: v1.TaintEffectNoSchedule,
  		})
  	}
  
  	// 补充node上关于NoSchedule类型的taint
  	nodeTaints := taintutils.TaintSetFilter(node.Spec.Taints, func(t *v1.Taint) bool {
  		// only NoSchedule taints are candidates to be compared with "taints" later
  		if t.Effect != v1.TaintEffectNoSchedule {
  			return false
  		}
  		// Find unschedulable taint of node.
  		if t.Key == v1.TaintNodeUnschedulable {
  			return true
  		}
  		// Find node condition taints of node.
  		_, found := taintKeyToNodeConditionMap[t.Key]
  		return found
  	})
      
      // 对比获取待新增的taint及待删除的taint
  	taintsToAdd, taintsToDel := taintutils.TaintSetDiff(taints, nodeTaints)
  	
      
  	if len(taintsToAdd) == 0 && len(taintsToDel) == 0 {
  		return nil
  	}
      
      // Patch到节点上
  	controllerutil.SwapNodeControllerTaint(ctx, nc.kubeClient, taintsToAdd, taintsToDel, node)
      ...
      
  	return nil
  }
  

• 

• 注意

  doNodeProcessingPassWorker做的事较为简单，就是补充label及根据node condition打taint，触发taintMgr驱逐

3.2.doPodProcess

• nc.doPodProcessingWorker()会基于Pod事件检查node健康状态，node不健康会标记Pod为notReady状态，以触发taint检查及驱逐。

• func (nc *Controller) doPodProcessingWorker(ctx context.Context) {
  	for {
  		obj, shutdown := nc.podUpdateQueue.Get()
  		...
  
  		podItem := obj.(podUpdateItem)
  		nc.processPod(ctx, podItem)
  	}
  }
  
  // processPod is processing events of assigning pods to nodes.
  func (nc *Controller) processPod(ctx context.Context, podItem podUpdateItem) {
  	defer nc.podUpdateQueue.Done(podItem)
      
  	pod, err := nc.podLister.Pods(podItem.namespace).Get(podItem.name)
  	if err != nil {
  		if apierrors.IsNotFound(err) {
  			return
  		}
  		nc.podUpdateQueue.AddRateLimited(podItem)
  		return
  	}
  
  	nodeName := pod.Spec.NodeName
  
  	nodeHealth := nc.nodeHealthMap.getDeepCopy(nodeName)
  	if nodeHealth == nil {
  		return
  	}
  
      // node不存在
  	_, err = nc.nodeLister.Get(nodeName)
  	if err != nil {
  		nc.podUpdateQueue.AddRateLimited(podItem)
  		return
  	}
  
      // readyCondition不存在
  	_, currentReadyCondition := controllerutil.GetNodeCondition(nodeHealth.status, v1.NodeReady)
  	if currentReadyCondition == nil {
  		return
  	}
  	...
      
      // node不健康,标记Pod为notReady
  	if currentReadyCondition.Status != v1.ConditionTrue {
  		if err := controllerutil.MarkPodsNotReady(ctx, nc.kubeClient, nc.recorder, pods, nodeName); err != nil {
  			nc.podUpdateQueue.AddRateLimited(podItem)
  		}
  	}
  }
  

• 

• 补充

  doPodProcessingWorker只处理Pod状态，根据node健康状态标记Pod为notReady

3.3.doNoExecute

• nc.doNoExecuteTaintingPass()会处理zoneNoExecuteTainter队列任务，会进行一定限速，异常node由monitorNodeHealth推入队列。

• func (nc *Controller) doNoExecuteTaintingPass(ctx context.Context) {
  	...
      
  	func() {
  		...
  
  		// 记录zone集合
  		for k := range nc.zoneNoExecuteTainter {
  			zoneNoExecuteTainterKeys = append(zoneNoExecuteTainterKeys, k)
  		}
  	}()
      
  	// 以zone为单位处理异常node
  	for _, k := range zoneNoExecuteTainterKeys {
  		...
          
          // 获取taintQueue
          zoneNoExecuteTainterWorker = nc.zoneNoExecuteTainter[k]
  
          // 回调执行
  		zoneNoExecuteTainterWorker.Try(logger, func(value scheduler.TimedValue) (bool, time.Duration) {
              // 获取node
  			node, err := nc.nodeLister.Get(value.Value)
  			...
              
              // 获取readyCondition
  			_, condition := controllerutil.GetNodeCondition(&node.Status, v1.NodeReady)
  			...
              
  			switch condition.Status {
              // notReady
  			case v1.ConditionFalse:
                  // 添加NoExecute NotReady污点
  				taintToAdd = *NotReadyTaintTemplate
                  // 清理NoExecute Unreachable污点
  				oppositeTaint = *UnreachableTaintTemplate
              // 未知
  			case v1.ConditionUnknown:
                  // 添加NoExecute Unreachable污点
  				taintToAdd = *UnreachableTaintTemplate
                  // 清理NoExecute NotReady污点
  				oppositeTaint = *NotReadyTaintTemplate
  			default:
  				return true, 0
  			}
              
              // 更新到node
  			result := controllerutil.SwapNodeControllerTaint(ctx, nc.kubeClient, []*v1.Taint{&taintToAdd}, []*v1.Taint{&oppositeTaint}, node)
  			...
  
  			return result, 0
  		})
  	}
  }
  

• 

• 注意

  doNoExecute会基于zone获取异常node，检查readyCondition条件，补充或移除NoExecute Taint触发驱逐

4.monitor

4.1.monitorHealth

• nc.monitorNodeHealth()会监控node状态，将检测到的异常node加入zoneNoExecuteTainter，基于zone推送node及关联驱逐速率。

• // This function will taint nodes who are not ready or not reachable for a long period of time.
  func (nc *Controller) monitorNodeHealth(ctx context.Context) error {
  	...
  
  	// 获取node列表
  	nodes, err := nc.nodeLister.List(labels.Everything())
  	...
      
      // 基于knownNodeSet分类节点
  	added, deleted, newZoneRepresentatives := nc.classifyNodes(nodes)
  	
      // newZone注册到zoneStates及zoneNoExecuteTainter
  	for i := range newZoneRepresentatives {
  		nc.addPodEvictorForNewZone(logger, newZoneRepresentatives[i])
  	}
      
      // newNode注册
  	for i := range added {
          // 记录节点已知
  		nc.knownNodeSet[added[i].Name] = added[i]
          // nodeZone注册到zoneStates及zoneNoExecuteTainter
  		nc.addPodEvictorForNewZone(logger, added[i])
          // newNode清理NoExecute Taint及zoneNoExecuteTainter队列数据
  		nc.markNodeAsReachable(ctx, added[i])
  	}
  
      // 待删除节点由knownNodeSet移除
  	for i := range deleted {
  		delete(nc.knownNodeSet, deleted[i].Name)
  	}
  	...
      
  	updateNodeFunc := func(piece int) {
  		...
  
  		node := nodes[piece].DeepCopy()
  
          // 间隔20ms执行一次,直到成功或100ms超时
  		if err := wait.PollImmediate(20ms, 100ms, func() (bool, error) {
  			...
              
              // 尝试更新node状态及nodeHealth缓存
  			_, observedReadyCond, curReadyCond, err = nc.tryUpdateNodeHealth(ctx, node)
  			if err == nil {
  				return true, nil
  			}
              
  			// 更新失败,尝试获取node
  			node, err = nc.kubeClient.CoreV1().Nodes().Get(ctx, node.Name, metav1.GetOptions{})
  			...
              
  			return false, nil
  		}); err != nil {
  			return
  		}
  
  		// node未被排除规模检查
  		if !isNodeExcludedFromDisruptionChecks(node) {
  			zoneToNodeConditionsLock.Lock()
              // 记录curReadyCond
  			zoneToNodeConditions[nodeZone] = append(zoneToNodeConditions[nodeZone], curReadyCond)
  			zoneToNodeConditionsLock.Unlock()
  		}
  
  		if curReadyCond != nil {
              // 获取node上的Pod
  			pods, err := nc.getPodsAssignedToNode(node.Name)
  			if err != nil {
                  // ready-->notReady
  				if currentReadyCondition.Status != True && observedReadyCondition.Status == True {
                      // 记录到重试
  					nc.nodesToRetry.Store(node.Name, struct{}{})
  				}
  				return
  			}
              
              // node taint更新及加入zoneNoExecuteTainter
  			nc.processTaintBaseEviction(ctx, node, &observedReadyCondition)
  
              // 加载待重试node
  			_, needsRetry := nc.nodesToRetry.Load(node.Name)
              
  			switch {
              // ready-->notReady
  			case currentReadyCondition.Status != True && observedReadyCondition.Status == True:
  				fallthrough
              // 刚变为notReady或之前notReady且处理失败
  			case needsRetry && observedReadyCondition.Status != True:
                  // 更新Pod为notReady
  				if err = controllerutil.MarkPodsNotReady(ctx,kubeClient, recorder, pods, nodeName); err != nil {
  					// 更新失败,node加入重试
                      nc.nodesToRetry.Store(node.Name, struct{}{})
  					return
  				}
  			}
  		}
          
          // node处理成功清理重试状态
  		nc.nodesToRetry.Delete(node.Name)
  	}
  
      // 激活8个协程并发处理node
  	workqueue.ParallelizeUntil(ctx, nc.nodeUpdateWorkerSize, len(nodes), updateNodeFunc)
  
      // 驱逐速率控制
  	nc.handleDisruption(ctx, zoneToNodeConditions, nodes)
  
  	return nil
  }
  

• 

• 注意

  monitorHealth会更新controller维护的节点状态信息，更新驱逐速率，设置node及Pod状态及推送node至zoneNoExecuteTainter

4.2.tryUpdateNode

• nc.tryUpdateNodeHealth()基于node.status更新nodeHealthMap记录最近状态，根据上一次状态检查及更新本次node状态。

• // tryUpdateNodeHealth checks a given node's conditions and tries to update it.
  func (nc *Controller) tryUpdateNodeHealth(ctx context.Context, node *v1.Node) (...) {
      // 获取上一次node状态
  	nodeHealth := nc.nodeHealthMap.getDeepCopy(node.Name)
  	defer func() {
          // 记录本次node状态
  		nc.nodeHealthMap.set(node.Name, nodeHealth)
  	}()
  	...
      
      // 获取node readyCondition
  	_, currentReadyCondition := controllerutil.GetNodeCondition(&node.Status, v1.NodeReady)
      
      // ready状态未知
  	if currentReadyCondition == nil {
  		// 伪造observedReadyCondition
  		observedReadyCondition = v1.NodeCondition{
  			Type:               v1.NodeReady,
  			Status:             v1.ConditionUnknown,
  			LastHeartbeatTime:  node.CreationTimestamp,
  			LastTransitionTime: node.CreationTimestamp,
  		}
          
          // 启动就绪宽限期60s
  		gracePeriod = nc.nodeStartupGracePeriod
          
          // 更新nodeHealth
  		if nodeHealth != nil {
  			nodeHealth.status = &node.Status
  		} else {
  			nodeHealth = &nodeHealthData{
  				status:                   &node.Status,
  				probeTimestamp:           node.CreationTimestamp,
  				readyTransitionTimestamp: node.CreationTimestamp,
  			}
  		}
      // ready状态已知
  	} else {
  		// observedReadyCondition设置为获取到的ready状态
  		observedReadyCondition = *currentReadyCondition
          // 心跳失联宽限期40s
  		gracePeriod = nc.nodeMonitorGracePeriod
  	}
  	...
      
      // 基于nodeHealth获取readyCondition及lease
  	if nodeHealth != nil {
  		_, savedCondition = controllerutil.GetNodeCondition(nodeHealth.status, v1.NodeReady)
  		savedLease = nodeHealth.lease
  	}
      
  	// nodeHealth为空进行构造
  	if nodeHealth == nil {
  		nodeHealth = &nodeHealthData{
  			status:                   &node.Status,
  			probeTimestamp:           nc.now(),
  			readyTransitionTimestamp: nc.now(),
  		}
      // 上一次readyCondition为空本次不为空,更新nodeHealth
  	} else if savedCondition == nil && currentReadyCondition != nil {
  		nodeHealth = &nodeHealthData{
  			status:                   &node.Status,
  			probeTimestamp:           nc.now(),
  			readyTransitionTimestamp: nc.now(),
  		}
      // 上一次readyCondition不为空本次为空,更新nodeHealth
  	} else if savedCondition != nil && currentReadyCondition == nil {
  		nodeHealth = &nodeHealthData{
  			status:                   &node.Status,
  			probeTimestamp:           nc.now(),
  			readyTransitionTimestamp: nc.now(),
  		}
      // readyCondition一直在且心跳事件不同
      } else if savedCondition != nil && currentReadyCondition != nil && LastHeartbeatTimeDiff() {
  		...
          
          // 基于ready状态语义调整时间设置transitionTime
  		if savedCondition.LastTransitionTime != currentReadyCondition.LastTransitionTime {
  			transitionTime = nc.now()
  		} else {
  			transitionTime = nodeHealth.readyTransitionTimestamp
  		}
          
          // 设置nodeHealth
  		nodeHealth = &nodeHealthData{
  			status:                   &node.Status,
  			probeTimestamp:           nc.now(),
  			readyTransitionTimestamp: transitionTime,
  		}
  	}
  
      // 获取nodeLease
  	observedLease, _ := nc.leaseLister.Leases(v1.NamespaceNodeLease).Get(node.Name)
      
      // lease未缓存过或续期过
  	if observedLease != nil && (savedLease == nil || savedLease.Spec.RenewTime.Before(observedLease.RenewTime)){
  		// 更新lease及探测时间
          nodeHealth.lease = observedLease
  		nodeHealth.probeTimestamp = nc.now()
  	}
  
      // lease距离上次续期超过宽限期
  	if nc.now().After(nodeHealth.probeTimestamp.Add(gracePeriod)) {
  		...
  
  		nowTimestamp := nc.now()
          // 遍历设置Ready/MemoryPressure/DiskPressure/PIDPressure
  		for _, nodeConditionType := range nodeConditionTypes {
              // 获取对应condition
  			_, currentCondition := controllerutil.GetNodeCondition(&node.Status, nodeConditionType)
              
              // condition未设置过
  			if currentCondition == nil {
                  // 设为unKnown
  				node.Status.Conditions = append(node.Status.Conditions, v1.NodeCondition{
  					Type:               nodeConditionType,
  					Status:             v1.ConditionUnknown,
  					Reason:             "NodeStatusNeverUpdated",
  					Message:            "Kubelet never posted node status.",
  					LastHeartbeatTime:  node.CreationTimestamp,
  					LastTransitionTime: nowTimestamp,
  				})
              // condition上报过,已失联
  			} else {
                  // 重置为unKnown状态
  				if currentCondition.Status != v1.ConditionUnknown {
  					currentCondition.Status = v1.ConditionUnknown
  					currentCondition.Reason = "NodeStatusUnknown"
  					currentCondition.Message = "Kubelet stopped posting node status."
  					currentCondition.LastTransitionTime = nowTimestamp
  				}
  			}
  		}
          
  		// 获取readyCondition
  		_, currentReadyCondition = controllerutil.GetNodeCondition(&node.Status, v1.NodeReady)
  
          // 距离上次观测readyCondition变化
  		if !apiequality.Semantic.DeepEqual(currentReadyCondition, &observedReadyCondition) {
              // 更新node状态
  			nc.kubeClient.CoreV1().Nodes().UpdateStatus(ctx, node, metav1.UpdateOptions{})
              ...
              
              // 更新nodeHealth缓存
  			nodeHealth = &nodeHealthData{
  				status:                   &node.Status,
  				probeTimestamp:           nodeHealth.probeTimestamp,
  				readyTransitionTimestamp: nc.now(),
  				lease:                    observedLease,
  			}
  			return gracePeriod, observedReadyCondition, currentReadyCondition, nil
  		}
  	}
  
  	return gracePeriod, observedReadyCondition, currentReadyCondition, nil
  }
  

• 

• 注意

  tryUpdateNodeHealth主要更新node.status.condition及维护nodeHealth缓存的节点状态信息

4.3.processTaint

• nc.processTaintBaseEviction()基于观测到的observedReadyCondition更新node taint及推送node至zoneNoExecuteTainter。

• func (nc *Controller) processTaintBaseEviction(...) {
  	...
  	
  	switch observedReadyCondition.Status {
      // 观测到readyCondition为false
  	case v1.ConditionFalse:
  		// UnReachable Taint已存在
  		if taintutils.TaintExists(node.Spec.Taints, UnreachableTaintTemplate) {
  			taintToAdd := *NotReadyTaintTemplate
              // 设置NotReady Taint,清理UnReachable Taint
  			controllerutil.SwapNodeControllerTaint(ctx, nc.kubeClient, []*v1.Taint{&taintToAdd}, []*v1.Taint{UnreachableTaintTemplate}, node)
              ...
  		} else if nc.markNodeForTainting(node, v1.ConditionFalse) {  }
      // 观测到readyCondition为UnKnown
  	case v1.ConditionUnknown:
  		// NotReady Taint已存在
  		if taintutils.TaintExists(node.Spec.Taints, NotReadyTaintTemplate) {
  			taintToAdd := *UnreachableTaintTemplate
              // 设置UnReachable Taint,清理NotReady Taint
  			controllerutil.SwapNodeControllerTaint(ctx, nc.kubeClient, []*v1.Taint{&taintToAdd}, []*v1.Taint{NotReadyTaintTemplate}, node)
              ...
  		} else if nc.markNodeForTainting(node, v1.ConditionUnknown) {  }
      // 观测到readyCondition为true
  	case v1.ConditionTrue:
          // 清理NotReady Taint和UnReachable Taint,取消zoneNoExecuteTainter队列的node
  		removed, err := nc.markNodeAsReachable(ctx, node)
  		...
  	}
  }
  
  // Taint标记及zoneNoExecuteTainter推送
  func (nc *Controller) markNodeForTainting(node *v1.Node, status v1.ConditionStatus) bool {
  	...
      
      // readyCondition为False
  	if status == v1.ConditionFalse {
          // NotReady Taint不存在
  		if !taintutils.TaintExists(node.Spec.Taints, NotReadyTaintTemplate) {
              // 清理zoneNoExecuteTainter推送的node
  			nc.zoneNoExecuteTainter[nodetopology.GetZoneKey(node)].Remove(node.Name)
  		}
  	}
  
      // readyCondition为UnKnown
  	if status == v1.ConditionUnknown {
          // UnReachable Taint不存在
  		if !taintutils.TaintExists(node.Spec.Taints, UnreachableTaintTemplate) {
              // 清理zoneNoExecuteTainter推送的node
  			nc.zoneNoExecuteTainter[nodetopology.GetZoneKey(node)].Remove(node.Name)
  		}
  	}
  
      // 重新推送node至zoneNoExecuteTainter,由doNoExecute打Taint
  	return nc.zoneNoExecuteTainter[nodetopology.GetZoneKey(node)].Add(node.Name, string(node.UID))
  }
  

• 

• 注意

  nc.processTaintBaseEviction()会交换Taint，未Patch过Taint的node推到zoneNoExecuteTainter由doNoExecute处理

4.4.handleDisruption

• nc.handleDisruption()会检测zone出现的node状态，计算zoneStates，unhealthy node大量出现会基于zone设置不同的驱逐速率。

• func (nc *Controller) handleDisruption(...) {
  	...
      
      // 遍历zone
  	for k, v := range zoneToNodeConditions {
          // 计算zone内node异常状态
  		unhealthy, newState := nc.computeZoneStateFunc(v)
  		if newState != stateFullDisruption {
  			allAreFullyDisrupted = false
  		}
          
          // 更新zoneState
  		newZoneStates[k] = newState
  		if _, had := nc.zoneStates[k]; !had {
  			nc.zoneStates[k] = stateInitial
  		}
  	}
  	...
      
      // 检测zone历史状态
  	for k, v := range nc.zoneStates {
          // zone内无node
  		if _, have := zoneToNodeConditions[k]; !have {
              // 清理zoneState
  			delete(nc.zoneStates, k)
  			continue
  		}
          
          // 检查历史异常状态
  		if v != stateFullDisruption {
  			allWasFullyDisrupted = false
  			break
  		}
  	}
  
  	// 不全为FullDisruption
  	if !allAreFullyDisrupted || !allWasFullyDisrupted {
  		// 非FullyDisrupted-->FullyDisrupted
  		if allAreFullyDisrupted {
              // 此时nc无法区分节点真实状态,索性去除Taint和zoneNoExecuteTainter注册的node
  			for i := range nodes {
  				_, err := nc.markNodeAsReachable(ctx, nodes[i])
  				...
  			}
              
  			// 切换limiter,冻结破坏性操作
  			for k := range nc.zoneStates {
  				nc.zoneNoExecuteTainter[k].SwapLimiter(0)
  			}
              
              // 记录节点状态
  			for k := range nc.zoneStates {
  				nc.zoneStates[k] = stateFullDisruption
  			}
  			// All rate limiters are updated, so we can return early here.
  			return
  		}
          
  		// FullyDisrupted-->非FullyDisrupted
  		if allWasFullyDisrupted {
  			...
              
              // 部分或全部节点正常,更新nodeHealth缓存
  			for i := range nodes {
  				v := nc.nodeHealthMap.getDeepCopy(nodes[i].Name)
  				v.probeTimestamp = now
  				v.readyTransitionTimestamp = now
  				nc.nodeHealthMap.set(nodes[i].Name, v)
  			}
              
  			// 基于异常节点数量切换limiter及更新zoneState
  			for k := range nc.zoneStates {
  				nc.setLimiterInZone(k, len(zoneToNodeConditions[k]), newZoneStates[k])
  				nc.zoneStates[k] = newZoneStates[k]
  			}
  			return
  		}
          
  		// 部分或全部节点正常了
  		for k, v := range nc.zoneStates {
  			newState := newZoneStates[k]
              // zoneState无变化
  			if v == newState {
  				continue
  			}
              
              // 有变化,基于异常节点数量切换limiter
  			nc.setLimiterInZone(k, len(zoneToNodeConditions[k]), newState)
              // 更新zoneState
  			nc.zoneStates[k] = newState
  		}
  	}
  }
  

• 

• 补充

  nc.handleDisruption()主要基于算出的zone状态切换rateLimiter，限制doNoExecute标记node Taint速度，避免触发大规模驱逐

4.5.swapLimiter

• ComputeZoneState()算出zone状态后，nc。setLimiterInZone()或zoneNoExecuteTainter[k].SwapLimiter()会切换限流队列速率。

• // ComputeZoneState returns a slice of NodeReadyConditions for all Nodes in a given zone.
  func (nc *Controller) ComputeZoneState(nodeReadyConditions []*v1.NodeCondition) (int, ZoneState) {
  	...
      
  	for i := range nodeReadyConditions {
  		if nodeReadyConditions[i] != nil && nodeReadyConditions[i].Status == v1.ConditionTrue {
  			readyNodes++
  		} else {
  			notReadyNodes++
  		}
  	}
      
  	switch {
      // zone下节点均异常(FullDisruption)
  	case readyNodes == 0 && notReadyNodes > 0:
  		return notReadyNodes, stateFullDisruption
      // zone下异常节点2个以上且比例超出0.55(PartialDisruption)
  	case notReadyNodes > 2 && float32(notReadyNodes)/float32(notReadyNodes+readyNodes) >= nc.unhealthyZoneThreshold:
  		return notReadyNodes, statePartialDisruption
      // 正常
  	default:
  		return notReadyNodes, stateNormal
  	}
  }
  
  // safely swaps current limiter for this queue with the passed one if capacities or qps's differ.
  func (q *RateLimitedTimedQueue) SwapLimiter(newQPS float32) {
  	q.limiterLock.Lock()
  	defer q.limiterLock.Unlock()
      
      // 驱逐速率无变化
  	if q.limiter.QPS() == newQPS {
  		return
  	}
      ...
      
      // 冻结,禁止消费
  	if newQPS <= 0 {
  		newLimiter = flowcontrol.NewFakeNeverRateLimiter()
      // 初始化tokenBucket
  	} else {
          // 初始token仅1个,10s生成一个
  		newLimiter = flowcontrol.NewTokenBucketRateLimiter(newQPS, EvictionRateLimiterBurst)
  
          // 此刻不该放行(无token)
  		if q.limiter.TryAccept() == false {
              // 消耗掉刚生成的一个token
  			newLimiter.TryAccept()
  		}
  	}
      
      // 替换限流器
  	q.limiter.Stop()
  	q.limiter = newLimiter
  }
  
  func (nc *Controller) setLimiterInZone(zone string, zoneSize int, state ZoneState) {
  	switch state {
      // 正常
  	case stateNormal:
          // 初始化qps=0.1的limiter
  		nc.zoneNoExecuteTainter[zone].SwapLimiter(nc.evictionLimiterQPS)
      // 半数异常
  	case statePartialDisruption:
          // 异常节点数量超出50,qps=0.01,未超出50则qps=0
  		nc.zoneNoExecuteTainter[zone].SwapLimiter(
  			nc.enterPartialDisruptionFunc(zoneSize))
      // 全节点异常
  	case stateFullDisruption:
          // qps=0.1
  		nc.zoneNoExecuteTainter[zone].SwapLimiter(
  			nc.enterFullDisruptionFunc(zoneSize))
  	}
  }
  

• 

• 注意

  swapLimiter()切换rateLimitQueue实现会调整消费速率，以限制doNoExecute标记node Taint，避免大规模驱逐