nodeStatus

1.nodeStatus

1.1.简介

• kubelet运行会向apiserver自注册成一个node资源，之后间隔10s进行节点状态上报，根据本地容器运行时检查podcidr与apiserver存储一致性，与上次同步间隔5min修改节点status。

• 

• 至关重要的node

  1.kubelet周期同步node状态至apiserver

  2.kube-controller-mgr监听node，决定node进行污点注入或驱逐

  3.kube-scheduler监听node用于调度pod的算法数据来源

1.2.状态上报

• kubelet会启动周期执行的syncNodeStatus()的协程上报节点状态，此外会快速触发一次状态上报，确保节点启动后快速将状态推到apiserver，促使调度器将pod更快调度到节点上。

• // Run starts the kubelet reacting to config updates
  func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
  	...
      if kl.kubeClient != nil {
  		...
          // 加入抖动,避免同步洪峰造成apiserver压力
          go wait.JitterUntil(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, 0.04, true, wait.NeverStop)
          // 快速执行一次状态更新,确保调度器更快分发pod
  		go kl.fastStatusUpdateOnce()
  		...
  	}
      ...
  }
  
  // 状态快速上报
  func (kl *Kubelet) fastStatusUpdateOnce() {
  	for {
          // 休眠100ms
  		time.Sleep(100 * time.Millisecond)
          // 获取node
  		node, err := kl.GetNode()
  		...
          // 节点网络配置完成
  		if len(node.Spec.PodCIDRs) != 0 {
  			podCIDRs := strings.Join(node.Spec.PodCIDRs, ",")
              // 更新运行时的podcidr
  			if _, err := kl.updatePodCIDR(podCIDRs); err != nil {
  				continue
  			}
              // 更新runtime状态可用
  			kl.updateRuntimeUp()
              // 向apiserver上报node状态
  			kl.syncNodeStatus()
  			return
  		}
  	}
  }
  

• kl.updatePodCIDR()用于将内部runtimeState的podcidr更新为最新值，通知底层运行时或网络插件更新网络配置。

• // updatePodCIDR updates the pod CIDR in the runtime state if it is different
  // from the current CIDR. Return true if pod CIDR is actually changed.
  func (kl *Kubelet) updatePodCIDR(cidr string) (bool, error) {
  	kl.updatePodCIDRMux.Lock()
  	defer kl.updatePodCIDRMux.Unlock()
  
      // 获取内部缓存的podcidr
  	podCIDR := kl.runtimeState.podCIDR()
  
      // cidr一致无需更新
  	if podCIDR == cidr {
  		return false, nil
  	}
  
  	// kubelet -> generic runtime -> runtime shim -> network plugin
      // 调用运行时更新底层CNI的cidr配置
  	kl.getRuntime().UpdatePodCIDR(cidr)
      ...
  	// 内部缓存最新的cidr
  	kl.runtimeState.setPodCIDR(cidr)
  	return true, nil
  }
  

• kl.updateRuntimeUp()用于检测运行时状态，完成运行时健康检查和依赖模块初始化，更新内部维护的运行时状态信息。

• // check and update the container runtime
  func (kl *Kubelet) updateRuntimeUp() {
  	kl.updateRuntimeMux.Lock()
  	defer kl.updateRuntimeMux.Unlock()
  
      // 获取运行时状态
  	s, err := kl.containerRuntime.Status()
  	...
  	// 网络状态检查
  	networkReady := s.GetRuntimeCondition(kubecontainer.NetworkReady)
  	if networkReady == nil || !networkReady.Status {
  		kl.runtimeState.setNetworkState(fmt.Errorf("container runtime network not ready: %v", networkReady))
  	} else {
  		// Set nil if the container runtime network is ready.
  		kl.runtimeState.setNetworkState(nil)
  	}
  	// 检查运行时状态
  	runtimeReady := s.GetRuntimeCondition(kubecontainer.RuntimeReady)
  	// If RuntimeReady is not set or is false, report an error.
  	if runtimeReady == nil || !runtimeReady.Status {
  		kl.runtimeState.setRuntimeState(fmt.Errorf("container runtime not ready: %v", runtimeReady))
  		return
  	}
  	kl.runtimeState.setRuntimeState(nil)
      // 运行相关模块
  	kl.oneTimeInitializer.Do(kl.initializeRuntimeDependentModules)
      // 更新运行时同步时间戳
  	kl.runtimeState.setRuntimeSync(kl.clock.Now())
  }
  

• kl.syncNodeStatus()用于向apiserver注册节点，同步最新的node状态至apiserver，确保集群中node对象反映当前节点的真实状态。

• // syncNodeStatus synchronizes node status to master if there is any change or enough time
  // passed from the last sync, registering the kubelet first if necessary.
  func (kl *Kubelet) syncNodeStatus() {
  	kl.syncNodeStatusMux.Lock()
  	defer kl.syncNodeStatusMux.Unlock()
  
  	if kl.kubeClient == nil || kl.heartbeatClient == nil {
  		return
  	}
      
      // 自注册
  	if kl.registerNode {
  		// This will exit immediately if it doesn't need to do anything.
  		kl.registerWithAPIServer()
  	}
      
      // 节点状态更新
  	if err := kl.updateNodeStatus(); err != nil {
  		klog.ErrorS(err, "Unable to update node status")
  	}
  }
  
  // registerWithAPIServer registers the node with the cluster master.
  func (kl *Kubelet) registerWithAPIServer() {
      // 注册完成
  	if kl.registrationCompleted {
  		return
  	}
  	...
  
  	for {
          // 2^n*100ms~7s
  		time.Sleep(step)
  
  		// 初始化节点
  		node, err := kl.initialNode(context.TODO())
  		if err != nil {
  			continue
  		}
  
          // 节点上报(创建或对比patch)
  		registered := kl.tryRegisterWithAPIServer(node)
  		if registered {
  			// 上报完成,取消自注册
  			kl.registrationCompleted = true
  			return
  		}
  	}
  }
  
  // updateNodeStatus updates node status to master with retries if there is any
  // change or enough time passed from the last sync.
  func (kl *Kubelet) updateNodeStatus() error {
      // 重试5次
  	for i := 0; i < nodeStatusUpdateRetry; i++ {
          // 对比更新节点状态
  		if err := kl.tryUpdateNodeStatus(i); err != nil {
  			if i > 0 && kl.onRepeatedHeartbeatFailure != nil {
                  // 失败回调
                  // 1.基于HTTP/1.1连接,强制关闭所有活跃连接,触发建立新连接
                  // 2.基于HTTP/2连接,周期检测连接状态,关闭空闲连接
  				kl.onRepeatedHeartbeatFailure()
  			}
  		} else {
  			return nil
  		}
  	}
  	return fmt.Errorf("update node status exceeds retry count")
  }
  

• 

• 注意

  1.kubelet会调用容器运行时健康检查获取状态，更新节点运行时和网络健康状态

  2.evictManager检测到的节点及节点组件基础信息和资源压力会在状态上报时更新到节点

2.nodeLease

2.1.简介

• nodeLease是一种用于优化节点心跳机制的API资源，旨在解决节点心跳依赖更新node.status.ready(大对象)造成的etcd压力，将节点心跳控制在频繁更新的一个lease对象上，降低服务端负载。

• 

2.2.nodeLeaseController

• kubelet启动后会后台执行周期10s的协程，调用nodeLeaseController.Run()定时更新lease对象，带上节点的身份信息续约，长时间未续约标记节点为notReady。

• // Run starts the kubelet reacting to config updates
  func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) {
  	...
      if kl.kubeClient != nil {
          // start syncing lease
  		go kl.nodeLeaseController.Run(wait.NeverStop)
      }
      ...
  }
  
  // NewController constructs and returns a controller
  func NewController(...) Controller {
  	var leaseClient coordclientset.LeaseInterface
  	if client != nil {
  		leaseClient = client.CoordinationV1().Leases(leaseNamespace)
  	}
  	return &controller{
  		client:                     client,
  		leaseClient:                leaseClient,
          // 节点身份(nodeName)
  		holderIdentity:             holderIdentity,
          // kube-node-lease
  		leaseNamespace:             leaseNamespace,
          // 租约有效期(40s)
  		leaseDurationSeconds:       leaseDurationSeconds,
          // 续约间隔(10s)
  		renewInterval:              renewInterval,
  		clock:                      clock,
          // 失败回调(HTTP1.1关闭所有连接,HTTP2.0关闭空闲连接)
  		onRepeatedHeartbeatFailure: onRepeatedHeartbeatFailure,
          // lease对象的ownerRef设置
  		newLeasePostProcessFunc:    newLeasePostProcessFunc,
  	}
  }
  
  // Run runs the controller
  func (c *controller) Run(stopCh <-chan struct{}) {
  	if c.leaseClient == nil {
  		return
  	}
      // 间隔10s续约一次
  	wait.JitterUntil(c.sync, c.renewInterval, 0.04, true, stopCh)
  }
  
  func (c *controller) sync() {
      // 优先基于缓存的lease更新
  	if c.latestLease != nil {
  		// 重试更新
  		err := c.retryUpdateLease(c.latestLease)
  		...
  	}
  
      // 获取或创建lease
  	lease, created := c.backoffEnsureLease()
      // 缓存lease
  	c.latestLease = lease
  	// we don't need to update the lease if we just created it
  	if !created && lease != nil {
  		c.retryUpdateLease(lease)
          ...
  	}
  }
  
  // retryUpdateLease attempts to update the lease for maxUpdateRetries,
  // call this once you're sure the lease has been created
  func (c *controller) retryUpdateLease(base *coordinationv1.Lease) error {
      // 最大重试5次
  	for i := 0; i < maxUpdateRetries; i++ {
          // 初始化或更新lease对象
  		leaseToUpdate, _ := c.newLease(base)
  		lease, err := c.leaseClient.Update(context.TODO(), leaseToUpdate, metav1.UpdateOptions{})
  		if err == nil {
  			c.latestLease = lease
  			return nil
  		}
  		
          // 回退确认lease是否存在(不存在更新,周期为[2^n*100ms,7s]).
  		if apierrors.IsConflict(err) {
  			base, _ = c.backoffEnsureLease()
  			continue
  		}
          
          // 失败回调(释放底层conn)
  		if i > 0 && c.onRepeatedHeartbeatFailure != nil {
  			c.onRepeatedHeartbeatFailure()
  		}
  	}
  	return fmt.Errorf("failed %d attempts to update lease", maxUpdateRetries)
  }
  

• 

• 注意

  1.kube-node-lease命名空间创建节点名的lease对象

  2.lease更新独立于nodeStatus发生

  3.lease按照周期10s触发更新

  4.lease租约有效期为40s，过期后节点转为notReady