evictManager | 南风未起｜云原生/Go技术分享博客

# 1.简介

# 1.1.作用

kubernetes出于保证节点的服务质量和稳定性，节点资源紧张时会驱动kubelet进行pod驱逐，evictionManager是驱逐的具体实现。pod调度时会进行准入控制，通过evictionManager管理的节点资源信息检查pod是否允许调度，同时内部基于定期监控及外部信号通知触发同步，根据资源状态驱逐节点pod释放资源。

# 1.2.定义

type managerImpl struct {
	...
    // killpod方法
	killPodFunc KillPodFunc
	// 获取mirrorPod方法
	mirrorPodFunc MirrorPodFunc
	// imageGC对象
	imageGC ImageGC
	// containerGC对象
	containerGC ContainerGC
	...
    // 当前节点condition
	nodeConditions []v1.NodeConditionType
	// 上一次观察node condition时间
	nodeConditionsLastObservedAt nodeConditionsObservedAt
	// node reference
	nodeRef *v1.ObjectReference
	// event记录器
	recorder record.EventRecorder
	...
    // 记录各个Threshold的第一次发现时间点
    thresholdsFirstObservedAt thresholdsObservedAt
    // 已触发未达到驱逐策略的Threshold
	thresholdsMet []evictionapi.Threshold
	// 存储软驱逐和硬驱逐中各个驱逐信号所对应的排序函数,用于计算被驱逐pod的顺序
	signalToRankFunc map[evictionapi.Signal]rankFunc
	// resource回收时调用的方法
	signalToNodeReclaimFuncs map[evictionapi.Signal]nodeReclaimFuncs
	...
    // 内存阈值通知器集合
	thresholdNotifiers []ThresholdNotifier
	// 上次thresholdNotifiers发通知的时间
	thresholdsLastUpdated time.Time
}

// 存储驱逐策略信息
type Threshold struct {
	// 驱逐信号
	Signal Signal
	// 驱逐信号对应资源的实际统计值与驱逐阈值之间的比较运算符
	Operator ThresholdOperator
	// 驱逐阈值
	Value ThresholdValue
	// 驱逐前持续的GracePeriod时间
	GracePeriod time.Duration
	// 触发驱逐后的资源最小回收值
	MinReclaim *ThresholdValue
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

# 1.3.接口

// Manager evaluates when an eviction threshold for node stability has been met on the node.
type Manager interface {
	// 监控驱逐的阈值
	Start(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc, podCleanedUpFunc PodCleanedUpFunc, monitoringInterval time.Duration)

	// 内存压力
	IsUnderMemoryPressure() bool

	// 磁盘压力
	IsUnderDiskPressure() bool

	// PID压力
	IsUnderPIDPressure() bool
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 1.4.初始化

eviction.NewManager初始化两个相同的manager对象，evictionManager用于本机驱逐pod的检查及回收，evictionAdmitHandler用于kubelet创建pod前的资源压力准入检查。

func NewMainKubelet(...
) (*Kubelet, error) {
    ...
	// setup eviction manager
    evictionManager, evictionAdmitHandler := eviction.NewManager(...)
    klet.evictionManager = evictionManager
    klet.admitHandlers.AddPodAdmitHandler(evictionAdmitHandler)
	...
}

1
2
3
4
5
6
7
8
9

补充

1.eviction manager会影响kube-scheduler的调度结果

2.kubelet定期将node condition传给apiserver更新到etcd

3.kube-scheduler监听到node condition pressure会调整策略，内存压力时阻止bestEffort pod调度，磁盘压力阻止所有pod调度

# 1.5.驱逐信号

--- memory.available
	内存可用量低,节点进入内存压力状态

--- nodefs.available
	跟文件系统容量,根分区磁盘空间不足,影响日志写入/volumn挂载

--- nodefs.inodesFree
	根文件系统`inode`,根分区`inode`用尽,无法创建新文件

--- imagefs.available
	容器文件镜像系统,镜像分区空间不足,无法拉取新镜像或创建容器

--- imagefs.inodesFree
	镜像文件系统`inode`,镜像分区`inode`用尽,容器写入失败

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 1.5.工作原理

eviction manager的核心逻辑是synchronize函数，通过monitor间隔10s或配置的内核监听notifier触发synchronize任务，两者共同驱动pod驱逐任务执行。

--- 实时驱逐
	1.`cgroups`允许向`cgroup.event_control`写入`<event_fd> <fd of memory.usage_in_bytes> <threshold>`
	2.`eventfd`设置`memory.usage_in_bytes`实现内存阈值超额触发内核通知
	3.`notifier`基于`epoll`监听`eventfd`，转发内核事件至`channel`通知`thresholdNotifier`执行`synchronize`

1
2
3
4

# 2.流程分析

# 2.1.启动

evictionManager会开启实时驱逐和轮询驱逐两个异步任务，实时驱逐依赖内核事件通知，轮询驱逐则间隔10s循环执行驱逐任务。

// Start starts the control loop to observe and response to low compute resources.
func (m *managerImpl) Start(...) {
   	// 处理器
	thresholdHandler := func(message string) {
        // 核心方法
		m.synchronize(diskInfoProvider, podFunc)
	}
    // 开启memcg通知,实时驱逐
	if m.config.KernelMemcgNotification {
		for _, threshold := range m.config.Thresholds {
            // cgroups只实现了memory事件监测
			if evictionapi.SignalMemoryAvailable || evictionapi.SignalAllocatableMemoryAvailable {
                // 初始化mem监听
				notifier, err := NewMemoryThresholdNotifier(threshold, m.config.PodCgroupRoot, &CgroupNotifierFactory{}, thresholdHandler)
				...
				// 启动事件内存检测
                go notifier.Start()
				m.thresholdNotifiers = append(m.thresholdNotifiers, notifier)	
			}
		}
	}
	// 开启monitor定时驱逐
	go func() {
		for {
            // 循环调用synchronize同步
			if evictedPods := m.synchronize(diskInfoProvider, podFunc); evictedPods != nil {
				// 阻塞等待驱逐的pod清理
				m.waitForPodsCleanup(podCleanedUpFunc, evictedPods)
			} else {
                // 间隔10s循环一次
				time.Sleep(monitoringInterval)
			}
		}
	}()
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# 2.2.mem事件通知

eviction manager实例化memoryThresholdNotifier对象时，会记录cgroup路径、阈值及事件管道，监听到的事件会基于channel传给synchronize方法执行。

// 初始化
func NewMemoryThresholdNotifier(threshold evictionapi.Threshold, cgroupRoot string, factory NotifierFactory, handler func(string)) (ThresholdNotifier, error) {
    // 获取cgroup的(获取系统支持的所有控制组,devices+freezer+/sys/fs/cgroup/cgroup.controllers支持的)
    cgroups, err := cm.GetCgroupSubsystems()
	...
    // 获取memory控制组路径(指向根cgroup)
	cgpath, found := cgroups.MountPoints["memory"]
	...
    // 基于node allocatable的内存策略(kubelet将cgroup划分为系统预留、自身组件预留及可分配的子控制组)
	if isAllocatableEvictionThreshold(threshold) {
		// cpath指向可分配子控制组
		cgpath += cgroupRoot
	}
	return &memoryThresholdNotifier{
		threshold:  threshold,
		cgroupPath: cgpath,
        // 事件管道
		events:     make(chan struct{}),
		handler:    handler,
		factory:    factory,
	}, nil
}

// 事件消费
func (m *memoryThresholdNotifier) Start() {
    // 消费事件管道数据
	for range m.events {
        // 调用synchronize同步
		m.handler(fmt.Sprintf("eviction manager: %s crossed", m.Description()))
	}
}

// 更新notifier
func (m *memoryThresholdNotifier) UpdateThreshold(summary *statsapi.Summary) error {
    // 节点内存状态信息
	memoryStats := summary.Node.Memory
    // 基于node allocatable的内存策略
	if isAllocatableEvictionThreshold(m.threshold) {
        // 获取pods内存状态(kubelet提供的统计信息,包括node上运行时占用、kubelet占用及pod申请的内存)
		allocatableContainer, err := getSysContainer(summary.Node.SystemContainers, statsapi.SystemContainerPods)
		...
        // 内存状态信息指向pods
		memoryStats = allocatableContainer.Memory
	}
    // 计算不活跃的内存(容器使用总内存-最近访问的活跃内存)作为可回收部分
	inactiveFile := resource.NewQuantity(int64(*memoryStats.UsageBytes-*memoryStats.WorkingSetBytes), resource.BinarySI)
    // 内存总量估算(可用内存≈总内存-活跃内存-不可回收的已用内存),不可回收内存无法感知,这里忽略
	capacity := resource.NewQuantity(int64(*memoryStats.AvailableBytes+*memoryStats.WorkingSetBytes), resource.BinarySI)
    // 计算驱逐阈值绝对值(阈值可指定为百分比)
	evictionThresholdQuantity := evictionapi.GetThresholdQuantity(m.threshold.Value, capacity)
    // usageBytes >= (capacity - evictionThreshold + inactive_file)触发驱逐
    // usageBytes = workingSetBytes + inactive_file
    // workingSetBytes >= capacity - evictionThreshold
    // 本质上是使用的内存超出可申请内存触发驱逐,由于workingSetBytes是kubelet计算出来的,没有直接文件用于观测,因此转为容器可用内存
	memcgThreshold := capacity.DeepCopy()
	memcgThreshold.Sub(*evictionThresholdQuantity)
	memcgThreshold.Add(*inactiveFile)
	...
    // 停掉旧的notifier(回收eventfd及epfd)
	if m.notifier != nil {
		m.notifier.Stop()
	}
    // 根据最新观测值初始化新的notifier
	newNotifier, err := m.factory.NewCgroupNotifier(m.cgroupPath, memoryUsageAttribute, memcgThreshold.Value())
	...
    // 启动最新的notifier
	m.notifier = newNotifier
	go m.notifier.Start(m.events)
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

补充

kubelet会统计几点的内存使用信息，分为runtime占用、kubelet占用及分配给pod的内存。

# 2.3.NewCgroupNotifier

NewCgroupNotifier用于修改cgroup memory配置触发内存通知，基于eventfd获取cgroups事件通知。eventfd捕获的事件基于channel

供thresholdNotifier消费，触发一次synchronize处理。

// cgroupNotifier工厂
func (n *CgroupNotifierFactory) NewCgroupNotifier(path, attribute string, threshold int64) (CgroupNotifier, error) {
	return NewCgroupNotifier(path, attribute, threshold)
}

// cgroupNotifier创建
func NewCgroupNotifier(path, attribute string, threshold int64) (CgroupNotifier, error) {
    // cgroupv2不支持cgroup.event_control,必须基于其它机制(memory.events/psi)
	if libcontainercgroups.IsCgroup2UnifiedMode() {
		return &disabledThresholdNotifier{}, nil
	}
	...
    // 打开memory.usage_in_bytes
	watchfd, err = unix.Open(fmt.Sprintf("%s/%s", path, attribute), unix.O_RDONLY|unix.O_CLOEXEC, 0)
	...
	defer unix.Close(watchfd)
	// 打开cgroup.event_control
    controlfd, err = unix.Open(fmt.Sprintf("%s/cgroup.event_control", path), unix.O_WRONLY|unix.O_CLOEXEC, 0)
	...
	defer unix.Close(controlfd)
    // 实例化eventfd
	eventfd, err = unix.Eventfd(0, unix.EFD_CLOEXEC)
	...
	defer func() {
		if err != nil {  unix.Close(eventfd)  }
	}()
    // 创建epoll
	epfd, err = unix.EpollCreate1(unix.EPOLL_CLOEXEC)
	...
	defer func() {
		if err != nil {  unix.Close(epfd)  }
	}()
    // 写入cgroup.event_control配置
    // <event_fd> <watchfd> <threshold>
	config := fmt.Sprintf("%d %d %d", eventfd, watchfd, threshold)
	_, err = unix.Write(controlfd, []byte(config))
	...
    // 返回cgroupNotifier
	return &linuxCgroupNotifier{
		eventfd: eventfd,
		epfd:    epfd,
		stop:    make(chan struct{}),
	}, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

cgroup通知

notifier基于内核的cgroups memory thresholds机制，cgroups允许用户态进程设置memory.usage_in_bytes达到某个阈值时内核向应用发送通知，具体做法就是向cgroup.event_control写入<event_fd> <fd of memory.usage_in_bytes> <threshold>。

# 2.4.notifier.start

notifier启动时通过epoll监听eventfd，监听到内核发送到事件时说明内存使用超出阈值，向channel发送信号触发驱逐。

func (n *linuxCgroupNotifier) Start(eventCh chan<- struct{}) {
    // eventfd事件注册到epoll进行epollin读监听
	err := unix.EpollCtl(n.epfd, unix.EPOLL_CTL_ADD, n.eventfd, &unix.EpollEvent{
		Fd:     int32(n.eventfd),
		Events: unix.EPOLLIN,
	})
	...
	buf := make([]byte, eventSize)
	for {
		select {
		case <-n.stop:
			return
		default:
		}
        // 阻塞等待事件,10s超时
		event, err := wait(n.epfd, n.eventfd, notifierRefreshInterval)
		...
		// 消费事件
		_, err = unix.Read(n.eventfd, buf)
		...
        // 推入eventCh,触发synchronize
		eventCh <- struct{}{}
	}
}

// epoll作为监听器,events作为事件目标
func wait(epfd, eventfd int, timeout time.Duration) (bool, error) {
	events := make([]unix.EpollEvent, numFdEvents+1)
    // 10s超时
	timeoutMS := int(timeout / time.Millisecond)
    // 阻塞等待事件触发
	n, err := unix.EpollWait(epfd, events, timeoutMS)
	...
	for _, event := range events[:n] {
		// 只关心eventfd事件
        if event.Fd == int32(eventfd) {
            // 对端关闭、文件描述符错误、有事件处理都要处理
			if event.Events&unix.EPOLLHUP != 0 || event.Events&unix.EPOLLERR != 0 || event.Events&unix.EPOLLIN != 0 {
				return true, nil
			}
		}
	}
	// An event occurred that we don't care about.
	return false, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

注意

synchronize执行时会检查10s内notifier是否有更新，进一步停止旧的notifier及启动新的notifier。

# 2.5.synchronize

synchronize是evict manager的核心处理方法，用于整理可驱逐资源，尝试进行imageGC和containerGC，同时针对优先级排序后的pod进行驱逐。synchronize驱逐是间歇性的，驱逐成功某个pod后就会等待下一轮任务。

// synchronize is the main control loop that enforces eviction thresholds.
// Returns the pod that was killed, or nil if no pod was killed.
func (m *managerImpl) synchronize(diskInfoProvider DiskInfoProvider, podFunc ActivePodsFunc) []*v1.Pod {
	...
    // 执行一次signal排序和回收函数初始化
	if m.dedicatedImageFs == nil {
        // 检查是否区分nodefs和imagefs的驱逐策略,用于构建回收函数时是否分别采集imagefs.available和 nodefs.available指标
		hasImageFs, ok := diskInfoProvider.HasDedicatedImageFs()
		...
		m.dedicatedImageFs = &hasImageFs
        // 各个signal的rankFunc排序注册到一个map
		m.signalToRankFunc = buildSignalToRankFunc(hasImageFs)
        // imageGC和containerGC注册到node的清理函数,驱逐pod前调用
		m.signalToNodeReclaimFuncs = buildSignalToNodeReclaimFuncs(m.imageGC, m.containerGC, hasImageFs)
	}
    // 获取活跃的pod列表(running且就绪的)
	activePods := podFunc()
	updateStats := true
    // 获取当前节点的资源使用情况,包括cpu、memory,fs,network,inode
	summary, err := m.summaryProvider.Get(updateStats)
	...
	// 上一次更新超过10s,调用UpdateThreshold重新启动notifier,更新cgroup.event_control
	if m.clock.Since(m.thresholdsLastUpdated) > notifierRefreshInterval {
		m.thresholdsLastUpdated = m.clock.Now()
		for _, notifier := range m.thresholdNotifiers {
            // 更新阈值/重新启动notifier
			notifier.UpdateThreshold(summary)
            ...
		}
	}

	// 根据传入的节点summary,获取各个驱逐信号对应的singleObservations
	// signalObservations代表某一时间资源状态信息
    // singleObservation包含获取资源占用的方法,即available、capacity、time方法
	observations, statsFunc := makeSignalObservations(summary)
	// 根据资源驱逐阈值和节点资源观测值计算当前触发阈值的threshold集合
	thresholds = thresholdsMet(thresholds, observations, false)
	...

	// 之前已经触发还未恢复的thresholds与当前的合并
	if len(m.thresholdsMet) > 0 {
		thresholdsNotYetResolved := thresholdsMet(m.thresholdsMet, observations, true)
		thresholds = mergeThresholds(thresholds, thresholdsNotYetResolved)
	}
	...
    // 过滤匹配阈值的条件类型
	nodeConditions := nodeConditions(thresholds)
	...
	// 满足压力过渡期内持续有效的nodeCondition
	nodeConditions = nodeConditionsObservedSince(nodeConditionsLastObservedAt, m.config.PressureTransitionPeriod, now)
	...
    // 观测值在period周期内不统计,其它超出阈值的观测值保留
	thresholds = thresholdsMetGracePeriod(thresholdsFirstObservedAt, now)
	...
    // 更新观察时间
	m.thresholdsFirstObservedAt = thresholdsFirstObservedAt
	m.nodeConditionsLastObservedAt = nodeConditionsLastObservedAt
    // 更新已触发未达到驱逐策略的Threshold
	m.thresholdsMet = thresholds

	// 再次过滤符合时间的阈值列表
	thresholds = thresholdsUpdatedStats(thresholds, observations, m.lastObservations)
	...
	// 更新上一次观测值
	m.lastObservations = observations
	m.Unlock()

	// 优先处理本地存储资源相关的驱逐
	if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
		if evictedPods := m.localStorageEviction(activePods, statsFunc); len(evictedPods) > 0 {
			return evictedPods
		}
	}
	...
	// 根据驱逐优先级排序
	sort.Sort(byEvictionPriority(thresholds))
    // 获取第一个threshold信号对应回收方法,后面的rank方法依赖这里的thresholdToReclaim
	thresholdToReclaim, resourceToReclaim, foundAny := getReclaimableThreshold(thresholds)
	...
	// 先尝试进行imageGC和containerGC垃圾回收清理
	if m.reclaimNodeLevelResources(thresholdToReclaim.Signal, resourceToReclaim) {
		return nil
	}
	// 获取rank方法
	rank, ok := m.signalToRankFunc[thresholdToReclaim.Signal]
	...
    // 对pods进行资源占比排序,最先被干掉的必定是资源占用最多的
	rank(activePods, statsFunc)
	...
	// we kill at most a single pod during each eviction interval
	for i := range activePods {
		pod := activePods[i]
		gracePeriodOverride := int64(0)
        // 不是硬驱逐,设置优雅退出时间
		if !isHardEvictionThreshold(thresholdToReclaim) {
			gracePeriodOverride = m.config.MaxPodGracePeriodSeconds
		}
		message, annotations := evictionMessage(resourceToReclaim, pod, statsFunc)
        // 只驱逐当前pod,成功后退出,等待下次触发
		if m.evictPod(pod, gracePeriodOverride, message, annotations) {
			return []*v1.Pod{pod}
		}
	}
	return nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

工作流程

1.构建信号rank函数和回收函数

2.notifier运行超过10s，更新threshold重新启动notifier

3.获取当前节点资源使用情况及活跃pods

4.针对每个信号，分别确定当前节点的资源使用情况是否达到驱逐阈值，没有的话退出当前循环

5.将所有的信号进行优先级排序，内存有关的信号先进行驱逐

6.向apiserver发送驱逐事件

7.所有活跃的pod进行优先级排序

8.根据顺序对第一个pod进行驱逐，退出等待下一轮触发

# 2.6.summaryProvider.Get

func (sp *summaryProviderImpl) Get(updateStats bool) (*statsapi.Summary, error) {
	// 获取nodeListener同步的节点信息
	node, err := sp.provider.GetNode()
	...
    // 获取containerManager维护的节点配置
	nodeConfig := sp.provider.GetNodeConfig()
    // 获取根cgroup统计数据,涉及CPU、内存、网络
	rootStats, networkStats, err := sp.provider.GetCgroupStats("/", updateStats)
	...
    // 获取根文件系统统计数据
	rootFsStats, err := sp.provider.RootFsStats()
	...
    // 获取镜像文件系统统计数据(/var/lib/containerd)
	imageFsStats, err := sp.provider.ImageFsStats()
	...
	var podStats []statsapi.PodStats
	if updateStats {
        // 获取pod统计信息,更新cpu usage的累加值
		podStats, err = sp.provider.ListPodStatsAndUpdateCPUNanoCoreUsage()
	} else {
        // 获取pod统计信息
		podStats, err = sp.provider.ListPodStats()
	}
	...
    // 获取进程资源限制统计信息
	rlimit, err := sp.provider.RlimitStats()
	...
	nodeStats := statsapi.NodeStats{
		NodeName:         node.Name,
		CPU:              rootStats.CPU,
		Memory:           rootStats.Memory,
		Network:          networkStats,
		StartTime:        sp.systemBootTime,
		Fs:               rootFsStats,
		Runtime:          &statsapi.RuntimeStats{ImageFs: imageFsStats},
		Rlimit:           rlimit,
        // 获取系统容器(kubelet/runtime/misc)资源使用信息,最终还是根据子cgroup获取cpu、内存和网络使用情况
		SystemContainers: sp.GetSystemContainersStats(nodeConfig, podStats, updateStats),
	}
	summary := statsapi.Summary{
		Node: nodeStats,
		Pods: podStats,
	}
	return &summary, nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

注意

summaryProvider获取的节点资源统计信息来自不同模块，涉及nodeListener、containerManager、cadvisor，各个模块维护的数据聚合后就是节点资源使用信息。

# 2.7.localStorageEviction

localStorageEviction用于监控pod和container的临时存储使用率，使用率超出上限时进行驱逐。

// 本地存储压力驱逐
func (m *managerImpl) localStorageEviction(pods []*v1.Pod, statsFunc statsFunc) []*v1.Pod {
	evicted := []*v1.Pod{}
    // 遍历activePods
	for _, pod := range pods {
        // 获取pod资源统计信息
		podStats, ok := statsFunc(pod)
		...
        // pod emptyDir超出
		if m.emptyDirLimitEviction(podStats, pod) {
			evicted = append(evicted, pod)
			continue
		}
		// pod ephemeral-storage超出
		if m.podEphemeralStorageLimitEviction(podStats, pod) {
			evicted = append(evicted, pod)
			continue
		}
		// container ephemeral-storage超出
		if m.containerEphemeralStorageLimitEviction(podStats, pod) {
			evicted = append(evicted, pod)
		}
	}

	return evicted
}

func (m *managerImpl) emptyDirLimitEviction(podStats statsapi.PodStats, pod *v1.Pod) bool {
	podVolumeUsed := make(map[string]*resource.Quantity)
    // 统计volumn使用信息
	for _, volume := range podStats.VolumeStats {
		podVolumeUsed[volume.Name] = resource.NewQuantity(int64(*volume.UsedBytes), resource.BinarySI)
	}
    // 遍历pod volumn定义
	for i := range pod.Spec.Volumes {
		source := &pod.Spec.Volumes[i].VolumeSource
        // emptyDir volumn定义且使用超限
		if source.EmptyDir != nil {
			size := source.EmptyDir.SizeLimit
			used := podVolumeUsed[pod.Spec.Volumes[i].Name]
			if used != nil && size != nil && size.Sign() == 1 && used.Cmp(*size) > 0 {
				// 驱逐pod
				if m.evictPod(pod, 0, fmt.Sprintf(emptyDirMessageFmt, pod.Spec.Volumes[i].Name, size.String()), nil) {
					return true
				}
				return false
			}
		}
	}
	return false
}

func (m *managerImpl) podEphemeralStorageLimitEviction(podStats statsapi.PodStats, pod *v1.Pod) bool {
    // 获取及检查pod Limit定义的ephemeral storage
	_, podLimits := apiv1resource.PodRequestsAndLimits(pod)
	_, found := podLimits[v1.ResourceEphemeralStorage]
	if !found {
		return false
	}

	// 统计pod ephemeral storage使用信息
	podEphemeralStorageTotalUsage := &resource.Quantity{}
	if podStats.EphemeralStorage != nil && podStats.EphemeralStorage.UsedBytes != nil {
		podEphemeralStorageTotalUsage = resource.NewQuantity(int64(*podStats.EphemeralStorage.UsedBytes), resource.BinarySI)
	}
    // 与pod limit指定的限制对比,超限驱逐pod
	podEphemeralStorageLimit := podLimits[v1.ResourceEphemeralStorage]
	if podEphemeralStorageTotalUsage.Cmp(podEphemeralStorageLimit) > 0 {
		// 驱逐pod
		if m.evictPod(pod, 0, fmt.Sprintf(podEphemeralStorageMessageFmt, podEphemeralStorageLimit.String()), nil) {
			return true
		}
		return false
	}
	return false
}

func (m *managerImpl) containerEphemeralStorageLimitEviction(podStats statsapi.PodStats, pod *v1.Pod) bool {
	// limit限制
    thresholdsMap := make(map[string]*resource.Quantity)
	for _, container := range pod.Spec.Containers {
		ephemeralLimit := container.Resources.Limits.StorageEphemeral()
		if ephemeralLimit != nil && ephemeralLimit.Value() != 0 {
			thresholdsMap[container.Name] = ephemeralLimit
		}
	}
	// 遍历container
	for _, containerStat := range podStats.Containers {
		containerUsed := diskUsage(containerStat.Logs)
        ...
        // limit < used
		if ephemeralStorageThreshold, ok := thresholdsMap[containerStat.Name]; ok {
			if ephemeralStorageThreshold.Cmp(*containerUsed) < 0 {
                // 驱逐pod
				if m.evictPod(pod, 0, fmt.Sprintf(containerEphemeralStorageMessageFmt, containerStat.Name, ephemeralStorageThreshold.String()), nil) {
					return true
				}
				return false
			}
		}
	}
	return false
}

补充

1.localStorageEviction用于触发emptyDir或ephemeral-storage使用超限的pod驱逐

2.本地存储驱逐保护可以避免节点资源耗尽，防止节点发生崩溃或无法调度pod现象

# 2.8.rankFunc

synchronize内部会调用buildSignalToRankFunc构建signal和rankFunc关系，pod驱逐时会根据定义的rankFunc排序，优先驱逐满足条件的pod。排序过程其实就是根据pod统计信息对比内存、磁盘、进程占用情况，根据资源占用情况优先驱逐使用多的。

// buildSignalToRankFunc returns ranking functions associated with resources
func buildSignalToRankFunc(withImageFs bool) map[evictionapi.Signal]rankFunc {
    // 内存和进程驱逐的排序函数
	signalToRankFunc := map[evictionapi.Signal]rankFunc{
		evictionapi.SignalMemoryAvailable:            rankMemoryPressure,
		evictionapi.SignalAllocatableMemoryAvailable: rankMemoryPressure,
		evictionapi.SignalPIDAvailable:               rankPIDPressure,
	}
    // imagefs文件系统是否独立影响rankDiskPressureFunc()关联的对比函数
    ...
    // nodefs imagefs
    signalToRankFunc[XXX] = rankDiskPressureFunc(statsType,diskResource)
	...
	return signalToRankFunc
}

// 1.优先驱逐内存使用超出request的pod
// 2.其次驱逐优先级低的pod
// 3.最后驱逐内存最接近request的pod
// 4.上述对比过程中，无统计数据的pod优先驱逐
func rankMemoryPressure(pods []*v1.Pod, stats statsFunc) {
	orderedBy(exceedMemoryRequests(stats), priority, memory(stats)).Sort(pods)
}

// 1.优先级低的优先驱逐
// 2.pid使用最多的优先驱逐
// 3.上述对比过程中,无统计数据的pod优先驱逐
func rankPIDPressure(pods []*v1.Pod, stats statsFunc) {
	orderedBy(priority, process(stats)).Sort(pods)
}

// 1.优先驱逐磁盘使用超出request的pod
// 2.其次驱逐优先级低的pod
// 3.最后驱逐磁盘最近request的pod
// 4.上述对比过程中，无统计数据的pod优先驱逐
func rankDiskPressureFunc(fsStatsToMeasure []fsStatsType, diskResource v1.ResourceName) rankFunc {
	return func(pods []*v1.Pod, stats statsFunc) {
		orderedBy(exceedDiskRequests(stats, fsStatsToMeasure, diskResource), priority, disk(stats, fsStatsToMeasure, diskResource)).Sort(pods)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40

# 2.9.reclaimFunc

synchronize内部会调用buildSignalToNodeReclaimFuncs构建signal和reclaimFunc关系，pod驱逐时会根据定义的reclaimFunc回收资源。

func buildSignalToNodeReclaimFuncs(imageGC ImageGC, containerGC ContainerGC, withImageFs bool) map[evictionapi.Signal]nodeReclaimFuncs {
	signalToReclaimFunc := map[evictionapi.Signal]nodeReclaimFuncs{}
	...
    // imagefs文件系统独立不会清理nodefs镜像和容器,其它情况均关联容器和镜像清理函数
    signalToReclaimFunc[XXX] = nodeReclaimFuncs{containerGC.DeleteAllUnusedContainers, imageGC.DeleteUnusedImages}
	return signalToReclaimFunc
}

// 正式驱逐前,执行reclaimFunc回收资源,回收后压力缓解就不再执行后面驱逐
func (m *managerImpl) reclaimNodeLevelResources(signalToReclaim evictionapi.Signal, resourceToReclaim v1.ResourceName) bool {
    // 获取回收函数
	nodeReclaimFuncs := m.signalToNodeReclaimFuncs[signalToReclaim]
    // 执行回收函数
	for _, nodeReclaimFunc := range nodeReclaimFuncs {
        // 这里其实执行的是containerGC和imageGC的未使用资源回收
		nodeReclaimFunc()
        ...
	}
    // 执行过回收
	if len(nodeReclaimFuncs) > 0 {
        // 获取节点资源统计信息
		summary, err := m.summaryProvider.Get(true)
		...
		// 获取资源观测值
		observations, _ := makeSignalObservations(summary)
		// 收集超出阈值的资源
		thresholds := thresholdsMet(m.config.Thresholds, observations, true)
		// 执行回收后没有超出阈值的资源,后续就不执行驱逐
		if len(thresholds) == 0 {
			return true
		}
	}
	return false
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

# 2.10.evictPod

evictPod就是执行驱逐的函数，驱逐pod其实就是干净利落的kill，被kill的pod再次调度时，由于node被打上资源压力condition，pod就不会继续调度到当前节点。

func (m *managerImpl) evictPod(pod *v1.Pod, gracePeriodOverride int64, evictMsg string, annotations map[string]string) bool {
    // static pod/mirror pod/critical pod(coredns kube-proxy等关键组件)不驱逐
	if kubelettypes.IsCriticalPod(pod) {
		return false
	}
	// 记录驱逐事件
	m.recorder.AnnotatedEventf(pod, annotations, v1.EventTypeWarning, Reason, evictMsg)
    // kill pod
	err := m.killPodFunc(pod, true, &gracePeriodOverride, func(status *v1.PodStatus) {
		status.Phase = v1.PodFailed
		status.Reason = Reason
		status.Message = evictMsg
	})
    ...
	return true
}

// pod删除
func killPodNow(podWorkers PodWorkers, recorder record.EventRecorder) eviction.KillPodFunc {
	return func(pod *v1.Pod, isEvicted bool, gracePeriodOverride *int64, statusFn func(*v1.PodStatus)) error {
		gracePeriod := int64(0)
        // evictManager配置了优雅退出时间
		if gracePeriodOverride != nil {
			gracePeriod = *gracePeriodOverride
        // pod设置了优雅退出时间
		} else if pod.Spec.TerminationGracePeriodSeconds != nil {
			gracePeriod = *pod.Spec.TerminationGracePeriodSeconds
		}
		// 设施kill pod超时时间
		timeout := int64(gracePeriod + (gracePeriod / 2))
		minTimeout := int64(10)
		if timeout < minTimeout {
			timeout = minTimeout
		}
		timeoutDuration := time.Duration(timeout) * time.Second
		// kill任务分发到podWorker
		ch := make(chan struct{}, 1)
		podWorkers.UpdatePod(UpdatePodOptions{
			Pod:        pod,
			UpdateType: kubetypes.SyncPodKill,
			KillPodOptions: &KillPodOptions{
				CompletedCh:                              ch,
				Evict:                                    isEvicted,
				PodStatusFunc:                            statusFn,
				PodTerminationGracePeriodSecondsOverride: gracePeriodOverride,
			},
		})
		// 等待kill完成
		select {
		case <-ch:
			return nil
		case <-time.After(timeoutDuration):
			recorder.Eventf(pod, v1.EventTypeWarning, events.ExceededGracePeriod, "Container runtime did not kill the pod within specified grace period.")
			return fmt.Errorf("timeout waiting to kill pod")
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57