k8s状态上报及调度实现分析

一时失言乱红尘 2023-02-14 06:48 58阅读 0赞

## 1. 概述 ##

下面的实现分析中会涉及到k8s三个组件，分别为:

*  scheduler: k8s的调度组件，这里负责筛选node来创建pod
 *  api-server: 这里负责对etcd的增删改查,在k8s中只有该组件可以对etcd进行操作
 *  kubelet: 是启动在node节点上的组件,负责pod的管理,监控,状态上报,pod的创建,删除等具体操作

注：scheduler,kubelet的资源数据的获取以及更新cache都是依赖api-server来实现的,具体的实现是通过list-watch机制来实现的([Kubernetes(k8s)代码解读-apiserver之list-watch篇][Kubernetes_k8s_-apiserver_list-watch])，api-server组件本身依靠etcd的watch实现，而组件之间通过http和长连接来实现

## 2. 调度的流程图 ##

具体调度可以参考: [k8s 调度][k8s]

![k8s调度][k8s 1]  
通过调用流程可以看到,组件之间的通信全部依赖list-watch机制,并且所有的数据更改在Api-Server中实现

## 3. 上报流程图 ##

![上报][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NzUyNjMy_size_16_color_FFFFFF_t_70]

## 4. 上报具体实现 ##

kublet的状态上报的触发是通过kublet.go中的Run方法来实现的：

// Run starts the kubelet reacting to config updates
    func (kl *Kubelet) Run(updates <-chan kubetypes.PodUpdate) { 
    	...
    	if kl.kubeClient != nil { 
    		// Start syncing node status immediately, this may set up things the runtime needs to run.
    		go wait.Until(kl.syncNodeStatus, kl.nodeStatusUpdateFrequency, wait.NeverStop)
    		go kl.fastStatusUpdateOnce()
    
    		// start syncing lease
    		go kl.nodeLeaseController.Run(wait.NeverStop)
    	}
    	go wait.Until(kl.updateRuntimeUp, 5*time.Second, wait.NeverStop)
    	...

这里的syncNodeStatus就是上报状态的方法,而这个方法的实现是在kublet\_node\_status.go文件中实现的:

// syncNodeStatus should be called periodically from a goroutine.
    // It 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.Errorf("Unable to update node status: %v", err)
    	}
    }

可以看到这里出了更新节点也包含了注册节点，这里的状态上报的主要实现就是updateNodeStatus:

// 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 { 
    	klog.V(5).Infof("Updating node status")
    	for i := 0; i < nodeStatusUpdateRetry; i++ { 
    		if err := kl.tryUpdateNodeStatus(i); err != nil { 
    			if i > 0 && kl.onRepeatedHeartbeatFailure != nil { 
    				kl.onRepeatedHeartbeatFailure()
    			}
    			klog.Errorf("Error updating node status, will retry: %v", err)
    		} else { 
    			return nil
    		}
    	}
    	return fmt.Errorf("update node status exceeds retry count")
    }

可以看到上报的逻辑会重试5次，具体的上报方法为:tryUpdateNodeStatus:

// tryUpdateNodeStatus tries to update node status to master if there is any
    // change or enough time passed from the last sync.
    func (kl *Kubelet) tryUpdateNodeStatus(tryNumber int) error { 
    	// In large clusters, GET and PUT operations on Node objects coming
    	// from here are the majority of load on apiserver and etcd.
    	// To reduce the load on etcd, we are serving GET operations from
    	// apiserver cache (the data might be slightly delayed but it doesn't
    	// seem to cause more conflict - the delays are pretty small).
    	// If it result in a conflict, all retries are served directly from etcd.
    	opts := metav1.GetOptions{ }
    	if tryNumber == 0 { 
    		util.FromApiserverCache(&opts)
    	}
    	node, err := kl.heartbeatClient.CoreV1().Nodes().Get(context.TODO(), string(kl.nodeName), opts)
    	if err != nil { 
    		return fmt.Errorf("error getting node %q: %v", kl.nodeName, err)
    	}
    
    	originalNode := node.DeepCopy()
    	if originalNode == nil { 
    		return fmt.Errorf("nil %q node object", kl.nodeName)
    	}
    
    	podCIDRChanged := false
    	if len(node.Spec.PodCIDRs) != 0 { 
    		// Pod CIDR could have been updated before, so we cannot rely on
    		// node.Spec.PodCIDR being non-empty. We also need to know if pod CIDR is
    		// actually changed.
    		podCIDRs := strings.Join(node.Spec.PodCIDRs, ",")
    		if podCIDRChanged, err = kl.updatePodCIDR(podCIDRs); err != nil { 
    			klog.Errorf(err.Error())
    		}
    	}
    
    	kl.setNodeStatus(node)
    
    	now := kl.clock.Now()
    	if now.Before(kl.lastStatusReportTime.Add(kl.nodeStatusReportFrequency)) { 
    		if !podCIDRChanged && !nodeStatusHasChanged(&originalNode.Status, &node.Status) { 
    			// We must mark the volumes as ReportedInUse in volume manager's dsw even
    			// if no changes were made to the node status (no volumes were added or removed
    			// from the VolumesInUse list).
    			//
    			// The reason is that on a kubelet restart, the volume manager's dsw is
    			// repopulated and the volume ReportedInUse is initialized to false, while the
    			// VolumesInUse list from the Node object still contains the state from the
    			// previous kubelet instantiation.
    			//
    			// Once the volumes are added to the dsw, the ReportedInUse field needs to be
    			// synced from the VolumesInUse list in the Node.Status.
    			//
    			// The MarkVolumesAsReportedInUse() call cannot be performed in dsw directly
    			// because it does not have access to the Node object.
    			// This also cannot be populated on node status manager init because the volume
    			// may not have been added to dsw at that time.
    			kl.volumeManager.MarkVolumesAsReportedInUse(node.Status.VolumesInUse)
    			return nil
    		}
    	}
    
    	// Patch the current status on the API server
    	updatedNode, _, err := nodeutil.PatchNodeStatus(kl.heartbeatClient.CoreV1(), types.NodeName(kl.nodeName), originalNode, node)
    	if err != nil { 
    		return err
    	}
    	kl.lastStatusReportTime = now
    	kl.setLastObservedNodeAddresses(updatedNode.Status.Addresses)
    	// If update finishes successfully, mark the volumeInUse as reportedInUse to indicate
    	// those volumes are already updated in the node's status
    	kl.volumeManager.MarkVolumesAsReportedInUse(updatedNode.Status.VolumesInUse)
    	return nil
    }

而具体的覆盖状态的方法是setNodeStatus:

// setNodeStatus fills in the Status fields of the given Node, overwriting
    // any fields that are currently set.
    // TODO(madhusudancs): Simplify the logic for setting node conditions and
    // refactor the node status condition code out to a different file.
    func (kl *Kubelet) setNodeStatus(node *v1.Node) { 
    	for i, f := range kl.setNodeStatusFuncs { 
    		klog.V(5).Infof("Setting node status at position %v", i)
    		if err := f(node); err != nil { 
    			klog.Errorf("Failed to set some node status fields: %s", err)
    		}
    	}
    }

setNodeStatusFunc是在kublet中的newMainKubelet中配置的:

func NewMainKubelet(kubeCfg *kubeletconfiginternal.KubeletConfiguration,
     		// ...
     		// Generating the status funcs should be the last thing we do,
        klet.setNodeStatusFuncs = klet.defaultNodeStatusFuncs()
    
        return klet, nil
    }

而defaultNodeStatusFuncs方法具体实现为：

// defaultNodeStatusFuncs is a factory that generates the default set of
    // setNodeStatus funcs
    func (kl *Kubelet) defaultNodeStatusFuncs() []func(*v1.Node) error { 
    	// if cloud is not nil, we expect the cloud resource sync manager to exist
    	var nodeAddressesFunc func() ([]v1.NodeAddress, error)
    	if kl.cloud != nil { 
    		nodeAddressesFunc = kl.cloudResourceSyncManager.NodeAddresses
    	}
    	var validateHostFunc func() error
    	if kl.appArmorValidator != nil { 
    		validateHostFunc = kl.appArmorValidator.ValidateHost
    	}
    	var setters []func(n *v1.Node) error
    	setters = append(setters,
    		nodestatus.NodeAddress(kl.nodeIP, kl.nodeIPValidator, kl.hostname, kl.hostnameOverridden, kl.externalCloudProvider, kl.cloud, nodeAddressesFunc),
    		nodestatus.MachineInfo(string(kl.nodeName), kl.maxPods, kl.podsPerCore, kl.GetCachedMachineInfo, kl.containerManager.GetCapacity,
    			kl.containerManager.GetDevicePluginResourceCapacity, kl.containerManager.GetNodeAllocatableReservation, kl.recordEvent),
    		nodestatus.VersionInfo(kl.cadvisor.VersionInfo, kl.containerRuntime.Type, kl.containerRuntime.Version),
    		nodestatus.DaemonEndpoints(kl.daemonEndpoints),
    		nodestatus.Images(kl.nodeStatusMaxImages, kl.imageManager.GetImageList),
    		nodestatus.GoRuntime(),
    	)
    	// Volume limits
    	setters = append(setters, nodestatus.VolumeLimits(kl.volumePluginMgr.ListVolumePluginWithLimits))
    
    	setters = append(setters,
    		nodestatus.MemoryPressureCondition(kl.clock.Now, kl.evictionManager.IsUnderMemoryPressure, kl.recordNodeStatusEvent),
    		nodestatus.DiskPressureCondition(kl.clock.Now, kl.evictionManager.IsUnderDiskPressure, kl.recordNodeStatusEvent),
    		nodestatus.PIDPressureCondition(kl.clock.Now, kl.evictionManager.IsUnderPIDPressure, kl.recordNodeStatusEvent),
    		nodestatus.ReadyCondition(kl.clock.Now, kl.runtimeState.runtimeErrors, kl.runtimeState.networkErrors, kl.runtimeState.storageErrors, validateHostFunc, kl.containerManager.Status, kl.recordNodeStatusEvent),
    		nodestatus.VolumesInUse(kl.volumeManager.ReconcilerStatesHasBeenSynced, kl.volumeManager.GetVolumesInUse),
    		// TODO(mtaufen): I decided not to move this setter for now, since all it does is send an event
    		// and record state back to the Kubelet runtime object. In the future, I'd like to isolate
    		// these side-effects by decoupling the decisions to send events and partial status recording
    		// from the Node setters.
    		kl.recordNodeSchedulableEvent,
    	)
    	return setters
    }

而此时的更改只是针对内存中的对象，还需要调用PatchNodeStatus来上报node状态到etcd(把更新的操作发送给Api-Server,k8s中只有Api-Server才可以操作etcd)：

// PatchNodeStatus patches node status.
    func PatchNodeStatus(c v1core.CoreV1Interface, nodeName types.NodeName, oldNode *v1.Node, newNode *v1.Node) (*v1.Node, []byte, error) { 
    	patchBytes, err := preparePatchBytesforNodeStatus(nodeName, oldNode, newNode)
    	if err != nil { 
    		return nil, nil, err
    	}
    
    	updatedNode, err := c.Nodes().Patch(context.TODO(), string(nodeName), types.StrategicMergePatchType, patchBytes, metav1.PatchOptions{ }, "status")
    	if err != nil { 
    		return nil, nil, fmt.Errorf("failed to patch status %q for node %q: %v", patchBytes, nodeName, err)
    	}
    	return updatedNode, patchBytes, nil
    }

[Kubernetes_k8s_-apiserver_list-watch]: https://www.kubernetes.org.cn/174.html
[k8s]: https://blog.csdn.net/qq_37050329/article/details/103542525
[k8s 1]: https://img-blog.csdnimg.cn/20200601151701322.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NzUyNjMy,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NzUyNjMy_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/2020060115172482.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2NzUyNjMy,size_16,color_FFFFFF,t_70