9.2 EvictionManager源码解读

本节重点总结 :

• evictionManager初始化了两个相同的manager对象

  • evictionManager做本机驱逐pod的判定和厨房
  • evictionAdmitHandler用来kubelet创建Pod前进依据本机的资源压力进行准入检查

• evictionManager判断内存驱逐阈值有两种方法

  • 第一种使用内核的memcg的通知机制，更高效
    • 当内存使用第一时间达到门槛，会立刻通知kubelet，并触发synchronize()进行资源回收的流程。这样提高了eviction的实时性。
  • 第二种使用轮询检查的机制

• 优先回收节点资源再驱逐pod

• 驱逐pod调用kubelet的syncPod()方法，把podPhase=Failed更新进去。

evictionManager 接口定义

• 位置 D:\go_path\src\github.com\kubernetes\kubernetes\pkg\kubelet\eviction\types.go

// Manager evaluates when an eviction threshold for node stability has been met on the node.
type Manager interface {// Start 用来监控驱逐的阈值的Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, monitoringInterval time.Duration)// 返回内存是否有压力IsUnderMemoryPressure() bool// 返回磁盘是否有压力IsUnderDiskPressure() bool// 返回PID是否有压力IsUnderPIDPressure() bool
}

evictionManager 结构体字段解析

• 位置 D:\go_path\src\github.com\kubernetes\kubernetes\pkg\kubelet\eviction\eviction_manager.go

// managerImpl implements Manager
type managerImpl struct {//  used to track timeclock clock.Clock// config is how the manager is configuredconfig Config// killpod的方法killPodFunc KillPodFunc// 获取静态pod的mirror pod 方法mirrorPodFunc MirrorPodFunc//  当node出现diskPressure condition时，imageGC进行unused images删除操作以回收disk space。imageGC ImageGC// the interface that knows how to do container gccontainerGC ContainerGC// protects access to internal statesync.RWMutex//  当前节点条件集合nodeConditions []v1.NodeConditionType// 记录nodecontions上一次观察的时间nodeConditionsLastObservedAt nodeConditionsObservedAt// nodeRef is a reference to the nodenodeRef *v1.ObjectReference// used to record events about the noderecorder record.EventRecorder// 提供node和node上所有pods的最新status数据汇总，既NodeStats and []PodStats。summaryProvider stats.SummaryProvider// records when a threshold was first observedthresholdsFirstObservedAt thresholdsObservedAt// 保存已经触发但还没解决的Thresholds，包括那些处于grace period等待阶段的Thresholds。thresholdsMet []evictionapi.Threshold// s定义各Resource进行evict挑选时的排名方法。signalToRankFunc map[evictionapi.Signal]rankFunc// 定义各Resource进行回收时调用的方法signalToNodeReclaimFuncs map[evictionapi.Signal]nodeReclaimFuncs// last observations from synchronizelastObservations signalObservations// dedicatedImageFs indicates if imagefs is on a separate device from the rootfsdedicatedImageFs *bool// 内存阈值通知器集合thresholdNotifiers []ThresholdNotifier// 上次thresholdNotifiers发通知的时间thresholdsLastUpdated time.Time
}

evictionManager的初始化

• 入口在kubelet 的new中

	// setup eviction managerevictionManager, evictionAdmitHandler := eviction.NewManager(klet.resourceAnalyzer, evictionConfig, killPodNow(klet.podWorkers, kubeDeps.Recorder), klet.podManager.GetMirrorPodByPod, klet.imageManager, klet.containerGC, kubeDeps.Recorder, nodeRef, klet.clock)

• 返回的 evictionManager, evictionAdmitHandler 其实都是 evictionManager对象

func NewManager(summaryProvider stats.SummaryProvider,config Config,killPodFunc KillPodFunc,mirrorPodFunc MirrorPodFunc,imageGC ImageGC,containerGC ContainerGC,recorder record.EventRecorder,nodeRef *v1.ObjectReference,clock clock.Clock,
) (Manager, lifecycle.PodAdmitHandler) {manager := &managerImpl{clock:                        clock,killPodFunc:                  killPodFunc,mirrorPodFunc:                mirrorPodFunc,imageGC:                      imageGC,containerGC:                  containerGC,config:                       config,recorder:                     recorder,summaryProvider:              summaryProvider,nodeRef:                      nodeRef,nodeConditionsLastObservedAt: nodeConditionsObservedAt{},thresholdsFirstObservedAt:    thresholdsObservedAt{},dedicatedImageFs:             nil,thresholdNotifiers:           []ThresholdNotifier{},}return manager, manager
}

evictionAdmitHandler的作用

• evictionAdmitHandler用来kubelet创建Pod前进行准入检查，满足条件后才会继续创建Pod，通过Admit方法来检查
• 可以看到evictionAdmitHandler被添加到kubelet的admitHandlers中了

klet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)

Admit方法解读

evictionAdmitHandler的Admit方法是 用node的压力状态 做pod是否准入的依据，相当于影响调度的结果

• 由下列可以看出如果 node现在没有压力状态那么准入

	m.RLock()defer m.RUnlock()if len(m.nodeConditions) == 0 {return lifecycle.PodAdmitResult{Admit: true}}

• 如果是静态pod或者SystemCriticalPriority的高优则忽略节点压力状态准入

	// Admit Critical pods even under resource pressure since they are required for system stability.// https://github.com/kubernetes/kubernetes/issues/40573 has more details.if kubelettypes.IsCriticalPod(attrs.Pod) {return lifecycle.PodAdmitResult{Admit: true}}

• 如果当前节点只有内存的压力，那么pod的qos类型为 非BestEffort则准入

	if nodeOnlyHasMemoryPressureCondition {notBestEffort := v1.PodQOSBestEffort != v1qos.GetPodQOS(attrs.Pod)if notBestEffort {return lifecycle.PodAdmitResult{Admit: true}}

• 如果是BestEffort则要根据它的容忍类型判断

		// When node has memory pressure, check BestEffort Pod's toleration:// admit it if tolerates memory pressure taint, fail for other tolerations, e.g. DiskPressure.if v1helper.TolerationsTolerateTaint(attrs.Pod.Spec.Tolerations, &v1.Taint{Key:    v1.TaintNodeMemoryPressure,Effect: v1.TaintEffectNoSchedule,}) {return lifecycle.PodAdmitResult{Admit: true}}}

• 最后如果pod有disk压力就拒绝

	// reject pods when under memory pressure (if pod is best effort), or if under disk pressure.klog.InfoS("Failed to admit pod to node", "pod", klog.KObj(attrs.Pod), "nodeCondition", m.nodeConditions)return lifecycle.PodAdmitResult{Admit:   false,Reason:  Reason,Message: fmt.Sprintf(nodeConditionMessageFmt, m.nodeConditions),}

总结

• eviction manager对kube-scheduler的调度结果影响
• Kubelet会定期的将Node Condition传给kube-apiserver并存于etcd
• kube-scheduler watch到Node Condition Pressure之后，会根据以下策略，阻止更多Pods Bind到该Node。
  • 如果是内存压力，阻止BestEffort pod调度
  • 如果是磁盘压力，阻止其他pod调度

evictionManager的启动

• start方法，位置D:\go_path\src\github.com\kubernetes\kubernetes\pkg\kubelet\eviction\eviction_manager.go

• 如果启动了kubelet的参数 --experimental-kernel-memcg-notification

• 设置为 true 表示 kubelet 将会集成内核的 memcg 通知机制而不是使用轮询机制来 判断是否达到了内存驱逐阈值。

	if m.config.KernelMemcgNotification {for _, threshold := range m.config.Thresholds {if threshold.Signal == evictionapi.SignalMemoryAvailable || threshold.Signal == evictionapi.SignalAllocatableMemoryAvailable {notifier, err := NewMemoryThresholdNotifier(threshold, m.config.PodCgroupRoot, &CgroupNotifierFactory{}, thresholdHandler)if err != nil {klog.InfoS("Eviction manager: failed to create memory threshold notifier", "err", err)} else {go notifier.Start()m.thresholdNotifiers = append(m.thresholdNotifiers, notifier)}}}}

• 如果没开启则使用轮询机制

	// start the eviction manager monitoringgo func() {for {if evictedPods := m.synchronize(diskInfoProvider, podFunc); evictedPods != nil {klog.InfoS("Eviction manager: pods evicted, waiting for pod to be cleaned up", "pods", format.Pods(evictedPods))m.waitForPodsCleanup(podCleanedUpFunc, evictedPods)} else {time.Sleep(monitoringInterval)}}}()

synchronize解析

buildSignalToRankFunc

• 注册各个eviction signal所对应的资源排序方法

func buildSignalToRankFunc(withImageFs bool) map[evictionapi.Signal]rankFunc {signalToRankFunc := map[evictionapi.Signal]rankFunc{evictionapi.SignalMemoryAvailable:            rankMemoryPressure,evictionapi.SignalAllocatableMemoryAvailable: rankMemoryPressure,evictionapi.SignalPIDAvailable:               rankPIDPressure,}// usage of an imagefs is optionalif withImageFs {// with an imagefs, nodefs pod rank func for eviction only includes logs and local volumessignalToRankFunc[evictionapi.SignalNodeFsAvailable] = rankDiskPressureFunc([]fsStatsType{fsStatsLogs, fsStatsLocalVolumeSource}, v1.ResourceEphemeralStorage)signalToRankFunc[evictionapi.SignalNodeFsInodesFree] = rankDiskPressureFunc([]fsStatsType{fsStatsLogs, fsStatsLocalVolumeSource}, resourceInodes)// with an imagefs, imagefs pod rank func for eviction only includes rootfssignalToRankFunc[evictionapi.SignalImageFsAvailable] = rankDiskPressureFunc([]fsStatsType{fsStatsRoot}, v1.ResourceEphemeralStorage)signalToRankFunc[evictionapi.SignalImageFsInodesFree] = rankDiskPressureFunc([]fsStatsType{fsStatsRoot}, resourceInodes)} else {// without an imagefs, nodefs pod rank func for eviction looks at all fs stats.// since imagefs and nodefs share a common device, they share common ranking functions.signalToRankFunc[evictionapi.SignalNodeFsAvailable] = rankDiskPressureFunc([]fsStatsType{fsStatsRoot, fsStatsLogs, fsStatsLocalVolumeSource}, v1.ResourceEphemeralStorage)signalToRankFunc[evictionapi.SignalNodeFsInodesFree] = rankDiskPressureFunc([]fsStatsType{fsStatsRoot, fsStatsLogs, fsStatsLocalVolumeSource}, resourceInodes)signalToRankFunc[evictionapi.SignalImageFsAvailable] = rankDiskPressureFunc([]fsStatsType{fsStatsRoot, fsStatsLogs, fsStatsLocalVolumeSource}, v1.ResourceEphemeralStorage)signalToRankFunc[evictionapi.SignalImageFsInodesFree] = rankDiskPressureFunc([]fsStatsType{fsStatsRoot, fsStatsLogs, fsStatsLocalVolumeSource}, resourceInodes)}return signalToRankFunc
}

比如内存

• rankMemoryPressure底层调用exceedMemoryRequests来对pod内存进行排序
• pod内存来自于容器内存的叠加，容器内存使用采用workingSetBytes
• 排序的依据为
  • 如果pod 的内存使用是否达到它的requests则较大
  • 对于使用内存达到requests的则按照使用量再排序

func exceedMemoryRequests(stats statsFunc) cmpFunc {return func(p1, p2 *v1.Pod) int {p1Stats, p1Found := stats(p1)p2Stats, p2Found := stats(p2)if !p1Found || !p2Found {// prioritize evicting the pod for which no stats were foundreturn cmpBool(!p1Found, !p2Found)}p1Memory := memoryUsage(p1Stats.Memory)p2Memory := memoryUsage(p2Stats.Memory)p1ExceedsRequests := p1Memory.Cmp(v1resource.GetResourceRequestQuantity(p1, v1.ResourceMemory)) == 1p2ExceedsRequests := p2Memory.Cmp(v1resource.GetResourceRequestQuantity(p2, v1.ResourceMemory)) == 1// prioritize evicting the pod which exceeds its requestsreturn cmpBool(p1ExceedsRequests, p2ExceedsRequests)}
}

buildSignalToNodeReclaimFuncs

• 注册节点资源回收方法，例如imagefs.avaliable对用的是删除无用容器和无用镜像

// buildSignalToNodeReclaimFuncs returns reclaim functions associated with resources.
func buildSignalToNodeReclaimFuncs(imageGC ImageGC, containerGC ContainerGC, withImageFs bool) map[evictionapi.Signal]nodeReclaimFuncs {signalToReclaimFunc := map[evictionapi.Signal]nodeReclaimFuncs{}// usage of an imagefs is optionalif withImageFs {// with an imagefs, nodefs pressure should just delete logssignalToReclaimFunc[evictionapi.SignalNodeFsAvailable] = nodeReclaimFuncs{}signalToReclaimFunc[evictionapi.SignalNodeFsInodesFree] = nodeReclaimFuncs{}// with an imagefs, imagefs pressure should delete unused imagessignalToReclaimFunc[evictionapi.SignalImageFsAvailable] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}signalToReclaimFunc[evictionapi.SignalImageFsInodesFree] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}} else {// without an imagefs, nodefs pressure should delete logs, and unused images// since imagefs and nodefs share a common device, they share common reclaim functionssignalToReclaimFunc[evictionapi.SignalNodeFsAvailable] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}signalToReclaimFunc[evictionapi.SignalNodeFsInodesFree] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}signalToReclaimFunc[evictionapi.SignalImageFsAvailable] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}signalToReclaimFunc[evictionapi.SignalImageFsInodesFree] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}}return signalToReclaimFunc
}

synchronize 判定

• 通过podFunc 获取pod
• 获取node的状态summary

	activePods := podFunc()updateStats := truesummary, err := m.summaryProvider.Get(updateStats)if err != nil {klog.ErrorS(err, "Eviction manager: failed to get summary stats")return nil}

makeSignalObservations

• 据传入的节点summary，获取各个驱逐信号对应的singleObservations和Pod的StatsFunc

	observations, statsFunc := makeSignalObservations(summary)

• 返回值 signalObservations代表某一时间，资源的可用量和总量
• 返回值 statsFunc 后续对Pods进行Rank时需要用的方法
• 以内存为例

	if memory := summary.Node.Memory; memory != nil && memory.AvailableBytes != nil && memory.WorkingSetBytes != nil {result[evictionapi.SignalMemoryAvailable] = signalObservation{available: resource.NewQuantity(int64(*memory.AvailableBytes), resource.BinarySI),capacity:  resource.NewQuantity(int64(*memory.AvailableBytes+*memory.WorkingSetBytes), resource.BinarySI),time:      memory.Time,}}

回到 synchronize 判定

• 计算得出当前触发阈值的threshold集合

thresholds = thresholdsMet(thresholds, observations, false)

• merge之前的已经触发的但还没恢复的thresholds

	// determine the set of thresholds previously met that have not yet satisfied the associated min-reclaimif len(m.thresholdsMet) > 0 {thresholdsNotYetResolved := thresholdsMet(m.thresholdsMet, observations, true)thresholds = mergeThresholds(thresholds, thresholdsNotYetResolved)}debugLogThresholdsWithObservation("thresholds - reclaim not satisfied", thresholds, observations)

• 转换得到nodeConditions

	// the set of node conditions that are triggered by currently observed thresholdsnodeConditions := nodeConditions(thresholds)if len(nodeConditions) > 0 {klog.V(3).InfoS("Eviction manager: node conditions - observed", "nodeCondition", nodeConditions)}

• 过滤出满足优雅删除的thresholds

	// determine the set of thresholds we need to drive eviction behavior (i.e. all grace periods are met)thresholds = thresholdsMetGracePeriod(thresholdsFirstObservedAt, now)debugLogThresholdsWithObservation("thresholds - grace periods satisfied", thresholds, observations)

• 从这轮的观察结果中去掉上一轮的观察结果出现的thresholds

	// determine the set of thresholds whose stats have been updated since the last syncthresholds = thresholdsUpdatedStats(thresholds, observations, m.lastObservations)debugLogThresholdsWithObservation("thresholds - updated stats", thresholds, observations)

• 按照驱逐的优先级给thresholds排序

	// rank the thresholds by eviction prioritysort.Sort(byEvictionPriority(thresholds))thresholdToReclaim, resourceToReclaim, foundAny := getReclaimableThreshold(thresholds)if !foundAny {return nil}klog.InfoS("Eviction manager: attempting to reclaim", "resourceName", resourceToReclaim)

• 对应的将内存排在前面，优先驱逐

func (a byEvictionPriority) Less(i, j int) bool {_, jSignalHasResource := signalToResource[a[j].Signal]return a[i].Signal == evictionapi.SignalMemoryAvailable || a[i].Signal == evictionapi.SignalAllocatableMemoryAvailable || !jSignalHasResource
}

• 按照优先级最高的threshold回收节点资源，意思是先判断能否回收节点资源

	// check if there are node-level resources we can reclaim to reduce pressure before evicting end-user pods.if m.reclaimNodeLevelResources(thresholdToReclaim.Signal, resourceToReclaim) {klog.InfoS("Eviction manager: able to reduce resource pressure without evicting pods.", "resourceName", resourceToReclaim)return nil}

• 使用rank方法排序

	// rank the pods for evictionrank, ok := m.signalToRankFunc[thresholdToReclaim.Signal]if !ok {klog.ErrorS(nil, "Eviction manager: no ranking function for signal", "threshold", thresholdToReclaim.Signal)return nil}// the only candidates viable for eviction are those pods that had anything running.if len(activePods) == 0 {klog.ErrorS(nil, "Eviction manager: eviction thresholds have been met, but no pods are active to evict")return nil}// rank the running pods for eviction for the specified resourcerank(activePods, statsFunc)

• 执行驱逐动作

	// we kill at most a single pod during each eviction intervalfor i := range activePods {pod := activePods[i]gracePeriodOverride := int64(0)if !isHardEvictionThreshold(thresholdToReclaim) {gracePeriodOverride = m.config.MaxPodGracePeriodSeconds}message, annotations := evictionMessage(resourceToReclaim, pod, statsFunc)if m.evictPod(pod, gracePeriodOverride, message, annotations) {metrics.Evictions.WithLabelValues(string(thresholdToReclaim.Signal)).Inc()return []*v1.Pod{pod}}}klog.InfoS("Eviction manager: unable to evict any pods from the node")return nil

• 底层调用killPodFunc 驱逐pod

func (m *managerImpl) evictPod(pod *v1.Pod, gracePeriodOverride int64, evictMsg string, annotations map[string]string) bool {// If the pod is marked as critical and static, and support for critical pod annotations is enabled,// do not evict such pods. Static pods are not re-admitted after evictions.// https://github.com/kubernetes/kubernetes/issues/40573 has more details.if kubelettypes.IsCriticalPod(pod) {klog.ErrorS(nil, "Eviction manager: cannot evict a critical pod", "pod", klog.KObj(pod))return false}// record that we are evicting the podm.recorder.AnnotatedEventf(pod, annotations, v1.EventTypeWarning, Reason, evictMsg)// this is a blocking call and should only return when the pod and its containers are killed.klog.V(3).InfoS("Evicting pod", "pod", klog.KObj(pod), "podUID", pod.UID, "message", evictMsg)err := m.killPodFunc(pod, true, &gracePeriodOverride, func(status *v1.PodStatus) {status.Phase = v1.PodFailedstatus.Reason = Reasonstatus.Message = evictMsg})if err != nil {klog.ErrorS(err, "Eviction manager: pod failed to evict", "pod", klog.KObj(pod))} else {klog.InfoS("Eviction manager: pod is evicted successfully", "pod", klog.KObj(pod))}return true
}

killPodFunc对应的就是killPodNow

• 位置 D:\go_path\src\github.com\kubernetes\kubernetes\pkg\kubelet\pod_workers.go
• 调用podworker 执行 UpdatePod

		podWorkers.UpdatePod(UpdatePodOptions{Pod:        pod,UpdateType: kubetypes.SyncPodKill,KillPodOptions: &KillPodOptions{CompletedCh:                              ch,Evict:                                    isEvicted,PodStatusFunc:                            statusFn,PodTerminationGracePeriodSecondsOverride: gracePeriodOverride,},})

• 最终就是调用kubelet的syncPod()方法，把podPhase=Failed更新进去

本节重点总结 :

• evictionManager初始化了两个相同的manager对象

  • evictionManager做本机驱逐pod的判定和厨房
  • evictionAdmitHandler用来kubelet创建Pod前进依据本机的资源压力进行准入检查

• evictionManager判断内存驱逐阈值有两种方法

  • 第一种使用内核的memcg的通知机制，更高效
    • 当内存使用第一时间达到门槛，会立刻通知kubelet，并触发synchronize()进行资源回收的流程。这样提高了eviction的实时性。
  • 第二种使用轮询检查的机制

• 优先回收节点资源再驱逐pod

• 驱逐pod调用kubelet的syncPod()方法，把podPhase=Failed更新进去。