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

kubelet主流程分析中可以看出，它的工作核心就是围绕syncLoop完成不同工作，并根据上报的信息管理Pod生命周期，这些操作都是configCh下的HandlePods具体完成。

处理流程

# 1.configCh事件

kubelet会主动监听file/apiserver/http的事件源，出现pod变更时会将事件写入configCh管道，根据事件类型ADD/UPDATE...供syncLoop消费。

# 1.1.新增事件

// ADD事件(pod创建/kubelet重启)
func (kl *Kubelet) HandlePodAdditions(pods []*v1.Pod) {
	...
    // 根据时间排序的pods
	for _, pod := range pods {
        // 获取已存在的pod
		existingPods := kl.podManager.GetPods()
		// podManager缓存pod期望状态
		kl.podManager.AddPod(pod)
        // 注入kubernetes.io/config.mirror注解的static pod处理(file/http获得的,会同步创建mirror pod)
		if kubetypes.IsMirrorPod(pod) {
			kl.handleMirrorPod(pod, start)
			continue
		}
		// pod未被标记为终止
		if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
            // 筛选需要处理的pod
            // 1.已经终止的/succeed或failed且未请求终止的排除
            // 2.保留的是running/pending/containerCreating/terminating状态的pod
			activePods := kl.filterOutInactivePods(existingPods)
            // pod准入检查(资源申请记录到cpuManager和MemoryManager)
			if ok, reason, message := kl.canAdmitPod(activePods, pod); !ok {
                // 未通过pod状态标记为拒绝
				kl.rejectPod(pod, reason, message)
				continue
			}
		}
        // 任务分发给worker
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)
	}
}

func (kl *Kubelet) handleMirrorPod(mirrorPod *v1.Pod, start time.Time) {
	// 获取mirror pod关联的pod分发
	if pod, ok := kl.podManager.GetPodByMirrorPod(mirrorPod); ok {
		kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
	}
}

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

主要任务

1.按照创建时间给pods进行排序

2.将pod添加到podManager，podManager是statusManager/volumeManager/runtimeManager依赖

3.校验pod是否能在该节点运行，校验内容包括资源、安全策略、节点是否就绪

4.调用dispatchWork把pod分配给worker做异步处理，实现pod创建

# 1.2.更新事件

// HandlePodUpdates is the callback in the SyncHandler interface for pods being updated from a config source.
func (kl *Kubelet) HandlePodUpdates(pods []*v1.Pod) {
	for _, pod := range pods {
        // 更新podManager缓存的pod期望状态
		kl.podManager.UpdatePod(pod)
        // 注入kubernetes.io/config.mirror注解的static pod处理(file/http获得的,会同步创建mirror pod)
		if kubetypes.IsMirrorPod(pod) {
			kl.handleMirrorPod(pod, start)
			continue
		}
        // 分发处理
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
	}
}

func (kl *Kubelet) handleMirrorPod(mirrorPod *v1.Pod, start time.Time) {
	// 获取mirror pod关联的pod分发
	if pod, ok := kl.podManager.GetPodByMirrorPod(mirrorPod); ok {
		kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 1.3.移除事件

// HandlePodRemoves is the callback in the SyncHandler interface for pods being removed from a config source.
func (kl *Kubelet) HandlePodRemoves(pods []*v1.Pod) {
	for _, pod := range pods {
        // 删除podManager缓存的pod期望状态
		kl.podManager.DeletePod(pod)
        // 注入kubernetes.io/config.mirror注解的static pod处理(file/http获得的,会同步创建mirror pod)
		if kubetypes.IsMirrorPod(pod) {
			kl.handleMirrorPod(pod, start)
			continue
		} 
		// 删除pod
		if err := kl.deletePod(pod); err != nil {
			klog.V(2).InfoS("Failed to delete pod", "pod", klog.KObj(pod), "err", err)
		}
	}
}

func (kl *Kubelet) handleMirrorPod(mirrorPod *v1.Pod, start time.Time) {
	// 获取mirror pod关联的pod及分发处理
	if pod, ok := kl.podManager.GetPodByMirrorPod(mirrorPod); ok {
		kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
	}
}

// deletePod deletes the pod from the internal state of the kubelet.
func (kl *Kubelet) deletePod(pod *v1.Pod) error {
	...
    // worker处理pod删除(类似dispatchWork,内部也会调用kl.podWorkers.UpdatePod)
	kl.podWorkers.UpdatePod(UpdatePodOptions{
		Pod:        pod,
		UpdateType: kubetypes.SyncPodKill,
	})
	// We leave the volume/directory cleanup to the periodic cleanup routine.
	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

# 1.4.协调事件

// HandlePodReconcile is the callback in the SyncHandler interface for pods that should be reconciled.
func (kl *Kubelet) HandlePodReconcile(pods []*v1.Pod) {
	for _, pod := range pods {
		// 更新podManager缓存的pod状态
		kl.podManager.UpdatePod(pod)
        // 定义readinessGates且Ready的实际状态与期望状态不一致
		if status.NeedToReconcilePodReadiness(pod) {
			mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
            // 分发同步
			kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
		}

		// pod标记为驱逐
		if eviction.PodIsEvicted(pod.Status) {
            // 获取pod状态,清理pod关联容器
			if podStatus, err := kl.podCache.Get(pod.UID); err == nil {
				kl.containerDeletor.deleteContainersInPod("", podStatus, true)
			}
		}
	}
}

func (p *podContainerDeletor) deleteContainersInPod(filterContainerID string, podStatus *kubecontainer.PodStatus, removeAll bool) {
	...
    // pod关联的容器
	for _, candidate := range getContainersToDeleteInPod(filterContainerID, podStatus, containersToKeep) {
		select {
        // 推入工作队列(异步调用运行时删除容器)
		case p.worker <- candidate.ID:
		...
		}
	}
}

func newPodContainerDeletor(runtime kubecontainer.Runtime, containersToKeep int) *podContainerDeletor {
	buffer := make(chan kubecontainer.ContainerID, containerDeletorBufferLimit)
	go wait.Until(func() {
		for {
			id := <-buffer
            // 异步调用运行时删除容器
			runtime.DeleteContainer(id)
            ...
		}
	}, 0, wait.NeverStop)
	return &podContainerDeletor{
		worker:           buffer,
		containersToKeep: containersToKeep,
	}
}

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

# 1.5.删除事件

// HandlePodUpdates is the callback in the SyncHandler interface for pods being updated from a config source.
func (kl *Kubelet) HandlePodUpdates(pods []*v1.Pod) {
	for _, pod := range pods {
        // 更新podManager缓存的pod
		kl.podManager.UpdatePod(pod)
        // 注入kubernetes.io/config.mirror注解的static pod处理(file/http获得的,会同步创建mirror pod)
		if kubetypes.IsMirrorPod(pod) {
			kl.handleMirrorPod(pod, start)
			continue
		}
        // 分发处理
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
	}
}

func (kl *Kubelet) handleMirrorPod(mirrorPod *v1.Pod, start time.Time) {
	// 获取mirror pod关联的pod分发
	if pod, ok := kl.podManager.GetPodByMirrorPod(mirrorPod); ok {
		kl.dispatchWork(pod, kubetypes.SyncPodUpdate, mirrorPod, start)
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 2.pleg事件

kubelet.pleg会每秒调用reList同步运行时的容器，进而生成pod生命周期的相关事件写入plegCh管道，根据不同事件类型分发处理。

# 2.1.同步事件

// 非Remove事件,同步pod状态
if isSyncPodWorthy(e) {
    // 获取podManager缓存的pod
    if pod, ok := kl.podManager.GetPodByUID(e.ID); ok {
        // 同步pod状态
        handler.HandlePodSyncs([]*v1.Pod{pod})
    }
}

func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
	for _, pod := range pods {
        // 依然是dispatchWork分发
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
	}
}

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

# 2.2.退出事件

// 退出事件
if e.Type == pleg.ContainerDied {
    // 根据containerID清理退出容器
    if containerID, ok := e.Data.(string); ok {
        kl.cleanUpContainersInPod(e.ID, containerID)
    }
}

// Delete the eligible dead container instances in a pod.
func (kl *Kubelet) cleanUpContainersInPod(podID types.UID, exitedContainerID string) {
    // 获取缓存的podStatus
	if podStatus, err := kl.podCache.Get(podID); err == nil {
		// 驱逐或删除的pod已同步,标记pod容器全部删除
		removeAll := kl.podWorkers.ShouldPodContentBeRemoved(podID)
		kl.containerDeletor.deleteContainersInPod(exitedContainerID, podStatus, removeAll)
	}
}

func (p *podContainerDeletor) deleteContainersInPod(filterContainerID string, podStatus *kubecontainer.PodStatus, removeAll bool) {
	containersToKeep := p.containersToKeep
	...	
	// 删除退出的容器
	for _, candidate := range getContainersToDeleteInPod(filterContainerID, podStatus, containersToKeep) {
		select {
        // 写入worker管道
		case p.worker <- candidate.ID:
		default:
			klog.InfoS("Failed to issue the request to remove container", "containerID", candidate.ID)
		}
	}
}

func newPodContainerDeletor(runtime kubecontainer.Runtime, containersToKeep int) *podContainerDeletor {
	buffer := make(chan kubecontainer.ContainerID, containerDeletorBufferLimit)
	go wait.Until(func() {
		for {
			id := <-buffer
            // 异步调用运行时删除容器
			runtime.DeleteContainer(id)
            ...
		}
	}, 0, wait.NeverStop)
	return &podContainerDeletor{
		worker:           buffer,
		containersToKeep: containersToKeep,
	}
}

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

# 3.syncCh事件

# 3.1.计算待同步pod

// 获取待同步的pod
func (kl *Kubelet) getPodsToSync() []*v1.Pod {
    // podManager缓存的所有pod
	allPods := kl.podManager.GetPods()
    // queue中已同步的pod
	podUIDs := kl.workQueue.GetWork()
    // allPods的所有podID
	podUIDSet := sets.NewString()
	...
	for _, pod := range allPods {
         // 待同步的pod
		if podUIDSet.Has(string(pod.UID)) {
			// The work of the pod is ready
			podsToSync = append(podsToSync, pod)
			continue
		}
        // 注册的syncLoop handler(activeDeadlineHandler)
		for _, podSyncLoopHandler := range kl.PodSyncLoopHandlers {
            // 检测是否需要同步(超出最长运行时间)
			if podSyncLoopHandler.ShouldSync(pod) {
				podsToSync = append(podsToSync, pod)
				break
			}
		}
	}
	return podsToSync
}

// 获取过去同步的pod(当前时间还未同步的)
func (q *basicWorkQueue) GetWork() []types.UID {
	q.lock.Lock()
	defer q.lock.Unlock()
	now := q.clock.Now()
	var items []types.UID
	for k, v := range q.queue {
		if v.Before(now) {
			items = append(items, k)
			delete(q.queue, k)
		}
	}
	return items
}

func (m *activeDeadlineHandler) ShouldSync(pod *v1.Pod) bool {
	// 检测是否超出最长运行时间
    return m.pastActiveDeadline(pod)
}

// pastActiveDeadline returns true if the pod has been active for more than its ActiveDeadlineSeconds
func (m *activeDeadlineHandler) pastActiveDeadline(pod *v1.Pod) bool {
	// 未开启运行限制
	if pod.Spec.ActiveDeadlineSeconds == nil {
		return false
	}
	// 获取statusManager缓存的podStatus
	podStatus, ok := m.podStatusProvider.GetPodStatus(pod.UID)
	if !ok {
		podStatus = pod.Status
	}
	// 还未开始运行
	if podStatus.StartTime.IsZero() {
		return false
	}
	// 检测运行时间是否超出限制
	start := podStatus.StartTime.Time
	duration := m.clock.Since(start)
	allowedDuration := time.Duration(*pod.Spec.ActiveDeadlineSeconds) * time.Second
	return duration >= allowedDuration
}

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

# 3.2.同步处理

// 分发处理
func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
	start := kl.clock.Now()
	for _, pod := range pods {
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
	}
}

1
2
3
4
5
6
7
8

# 4.livenessCh事件

case update := <-kl.livenessManager.Updates():
	// liveness探测状态为Failure
    if update.Result == proberesults.Failure {
        // 探测同步
        handleProbeSync(kl, update, handler, "liveness", "unhealthy")
    }

func handleProbeSync(kl *Kubelet, update proberesults.Update, handler SyncHandler, probe, status string) {
	// 获取podManager缓存的pod
	pod, ok := kl.podManager.GetPodByUID(update.PodUID)
	if !ok {
		return
	}
	// 处理pod同步
	handler.HandlePodSyncs([]*v1.Pod{pod})
}

// 分发处理
func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
	start := kl.clock.Now()
	for _, pod := range pods {
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
	}
}

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

# 5.readinessCh

case update := <-kl.readinessManager.Updates():
	// 计算readiness探测结果
    ready := update.Result == proberesults.Success
	// 更新statusManager缓存的pod container状态
    kl.statusManager.SetContainerReadiness(update.PodUID, update.ContainerID, ready)
	// 处理探测同步
    handleProbeSync(kl, update, handler, "readiness", status)

func handleProbeSync(kl *Kubelet, update proberesults.Update, handler SyncHandler, probe, status string) {
	// 获取podManager缓存的pod
	pod, ok := kl.podManager.GetPodByUID(update.PodUID)
	if !ok {
		return
	}
	// 处理pod同步
	handler.HandlePodSyncs([]*v1.Pod{pod})
}

// 分发处理
func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
	start := kl.clock.Now()
	for _, pod := range pods {
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
	}
}

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

# 6.startCh事件

case update := <-kl.startupManager.Updates():
	// 计算启动结果
    started := update.Result == proberesults.Success
	// 更新statusManager缓存的pod container状态
    kl.statusManager.SetContainerStartup(update.PodUID, update.ContainerID, started)

    status := "unhealthy"
    if started {
        status = "started"
    }
    handleProbeSync(kl, update, handler, "startup", status)

func handleProbeSync(kl *Kubelet, update proberesults.Update, handler SyncHandler, probe, status string) {
	// 获取podManager缓存的pod
	pod, ok := kl.podManager.GetPodByUID(update.PodUID)
	if !ok {
		return
	}
	// 处理pod同步
	handler.HandlePodSyncs([]*v1.Pod{pod})
}

// 分发处理
func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
	start := kl.clock.Now()
	for _, pod := range pods {
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
	}
}

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

# 7.housekeepingCh事件

// 家务事件,2s执行一次,清理not running的pod/未完成清理的pod
case <-housekeepingCh:
	// api/http/file配置源就绪
    if kl.sourcesReady.AllReady() {
        ...
        handler.HandlePodCleanups()
    }

// pod清理
func (kl *Kubelet) HandlePodCleanups() error {
	...
    // 扫描pod使用到的cgroup
	if kl.cgroupsPerQOS {
		pcm := kl.containerManager.NewPodContainerManager()
		cgroupPods, err = pcm.GetAllPodsFromCgroups()
		...
	}
    // 获取podManager缓存的所有pod和mirrorPod(Handle处理pod时会更新podManager缓存)
	allPods, mirrorPods := kl.podManager.GetPodsAndMirrorPods()
    // 获取有状态pod列表(事件分发时更新podWorkers的状态缓存)
	workingPods := kl.podWorkers.SyncKnownPods(allPods)
	allPodsByUID := make(map[types.UID]*v1.Pod)
	for _, pod := range allPods {
		allPodsByUID[pod.UID] = pod
	}
	// 根据已知的pod状态划分workingPod
	runningPods := make(map[types.UID]sets.Empty)
	possiblyRunningPods := make(map[types.UID]sets.Empty)
	restartablePods := make(map[types.UID]sets.Empty)
	for uid, sync := range workingPods {
		switch sync {
		case SyncPod:
			runningPods[uid] = struct{}{}
			possiblyRunningPods[uid] = struct{}{}
		case TerminatingPod:
			possiblyRunningPods[uid] = struct{}{}
		case TerminatedAndRecreatedPod:
			restartablePods[uid] = struct{}{}
		}
	}
    // 遍历清理terminating pod关联的探针任务(start/liveness/readiness探针会注册到probeManager.workers异步执行)
	kl.probeManager.CleanupPods(possiblyRunningPods)
    // 获取缓存的运行时pod(runtimeCache定期同步运行时的pod)
	runningRuntimePods, err := kl.runtimeCache.GetPods()
	...
    // 遍历运行时pod列表,workingPod没有且podManager未缓存,删除pod(pod已删除/运行时缓存未更新情况)
	for _, runningPod := range runningRuntimePods {
		switch workerState, ok := workingPods[runningPod.ID]; {
		case ok && workerState == SyncPod, ok && workerState == TerminatingPod:
			// if the pod worker is already in charge of this pod, we don't need to do anything
			continue
		default:
			// podManager未缓存pod
			if _, ok := allPodsByUID[runningPod.ID]; !ok {
                // 分发删除pod
				kl.podWorkers.UpdatePod(UpdatePodOptions{
					UpdateType: kubetypes.SyncPodKill,
					RunningPod: runningPod,
					KillPodOptions: &KillPodOptions{
						PodTerminationGracePeriodSecondsOverride: &one,
					},
				})
			}
		}
	}
	// 更新statusManager同步的pod status
	kl.removeOrphanedPodStatuses(allPods, mirrorPods)
    // 再次获取运行时的pod(非缓存,避免删除的未同步导致延迟清理)
	runningRuntimePods, err = kl.containerRuntime.GetPods(false)
	...
	// 清理孤儿pod的挂载卷占用
	err = kl.cleanupOrphanedPodDirs(allPods, runningRuntimePods)
	...
    // 删除孤儿pod的mirror pod
	kl.deleteOrphanedMirrorPods()
	// 清理pod的cgroup占用
	if kl.cgroupsPerQOS {
		pcm := kl.containerManager.NewPodContainerManager()
		kl.cleanupOrphanedPodCgroups(pcm, cgroupPods, possiblyRunningPods)
	}
    // 清理backoff entry(重试控制)
	kl.backOff.GC()
	// 遍历重启的pod
	for uid := range restartablePods {
		...
        // pod不存在/succeed/failed跳过
		if kl.isAdmittedPodTerminal(pod) {
			continue
		}
		// 分发重启terminal pod
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
		kl.dispatchWork(pod, kubetypes.SyncPodCreate, mirrorPod, start)
	}
	return nil
}

# 8.分发

# 8.1.dispatchWork

func (kl *Kubelet) dispatchWork(pod *v1.Pod, syncType kubetypes.SyncPodType, mirrorPod *v1.Pod, start time.Time) {
	// Run the sync in an async worker.
	kl.podWorkers.UpdatePod(UpdatePodOptions{
		Pod:        pod,
		MirrorPod:  mirrorPod,
		UpdateType: syncType,
		StartTime:  start,
	})
	...
}

1
2
3
4
5
6
7
8
9
10

封装一个UpdatePodOptions结构体丢给podWorkers.UpdatePod去执行

# 8.2.podWorkers.UpdatePod

func (p *podWorkers) UpdatePod(options UpdatePodOptions) {
	pod := options.Pod
	...
	uid := pod.UID
	...
	// 1.podUpdates列表没有podUpdate,新建channel启动异步协程管理pod生命周期
	podUpdates, exists := p.podUpdates[uid]
	if !exists {
		podUpdates = make(chan podWork, 1)
		p.podUpdates[uid] = podUpdates
		...
		go func() {
            // pod变更处理
			defer runtime.HandleCrash()
			p.managePodLoop(outCh)
		}()
	}

	// 2.podWorker未运行请求,下发更新事件
	if !status.IsWorking() {
		status.working = true
		podUpdates <- work
		return
	}
    ...
    // 否则加入待处理缓存
	p.lastUndeliveredWorkUpdate[pod.UID] = work
    ...
}

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

UpdatePod会加锁后获取podUpdates数组里的数据，不存在时会创建一个channel然后执行异步协程

# 8.3.managePodLoop

func (p *podWorkers) managePodLoop(podUpdates <-chan podWork) {
    ...
    // 1.遍历channel
	for update := range podUpdates {
		pod := update.Options.Pod
		...
		err := func() error {
			...
			switch {
			case update.Options.RunningPod != nil:
				// orphan pod不需要获取状态,因为配置及状态已被清理
			default:
                // 1.获取cache中的pod status，保证worker不会提前开始(pleg每次reList触发通知)
				status, err = p.podCache.GetNewerThan(pod.UID, lastSyncTime)
			}
			...
			switch {
            // 2.已终止的pod资源回收
			case update.WorkType == TerminatedPodWork:
				err = p.syncTerminatedPodFn(ctx, pod, status)
			// 3.正在终止pod删除
			case update.WorkType == TerminatingPodWork:
				...
                // 标记pod正在终止,取出优雅删除回调
				podStatusFn := p.acknowledgeTerminating(pod)
				// 终止pod
				p.syncTerminatingPodFn(ctx, pod, status, update.Options.RunningPod, gracePeriod, podStatusFn)
			default:
                // 4.pod变更同步
				isTerminal, err = p.syncPodFn(ctx, update.Options.UpdateType, pod, update.Options.MirrorPod, status)
			}
			return err
		}()
		switch {
		case err == context.Canceled:
			// ctx cancel
		case err != nil:
			// sync err
		case update.WorkType == TerminatedPodWork:
			// 清除pod状态及缓存
			p.completeTerminated(pod)
            ...
			return
		case update.WorkType == TerminatingPodWork:
			// orphan pod删除完成,清除pod状态及缓存
			if update.Options.RunningPod != nil {
				p.completeTerminatingRuntimePod(pod)
				...
				return
			}
			// terminating pod状态及缓存更新
			p.completeTerminating(pod)
			phaseTransition = true
		case isTerminal:
			// pod同步状态及缓存更新
			p.completeSync(pod)
			phaseTransition = true
		}
		// 收尾工作
		p.completeWork(pod, phaseTransition, err)
	}
}

// completeTerminated is invoked after syncTerminatedPod completes successfully and means we
// can stop the pod worker. The pod is finalized at this point.
func (p *podWorkers) completeTerminated(pod *v1.Pod) {
    // 关闭pod协程管道(podUpdates),取消p.PodUpdates的管道登记及才处理的队列缓存
	p.cleanupPodUpdates(pod.UID)
	// 更新pod状态链路
	if status, ok := p.podSyncStatuses[pod.UID]; ok {
		// 标记完成
		status.finished = true
        // 关闭工作状态(不阻塞后续处理)
		status.working = false
		// static pod
		if p.startedStaticPodsByFullname[status.fullname] == pod.UID {
            // pod已经被删除,清理static pod缓存
			delete(p.startedStaticPodsByFullname, status.fullname)
		}
	}
}

// completeTerminatingRuntimePod is invoked when syncTerminatingPod completes successfully.
func (p *podWorkers) completeTerminatingRuntimePod(pod *v1.Pod) {
	...
    // 更新链路状态
	if status, ok := p.podSyncStatuses[pod.UID]; ok {
        // 标记完成
		status.finished = true
        // 关闭工作状态(不阻塞后续变更)
		status.working = false
		// static pod
		if p.startedStaticPodsByFullname[status.fullname] == pod.UID {
            // pod已经被删除,清理static pod缓存
			delete(p.startedStaticPodsByFullname, status.fullname)
		}
	}
	// 关闭pod协程管道(podUpdates),删除pod的map信息及未执行变化
	p.cleanupPodUpdates(pod.UID)
}

// terminating pod完成通知
func (p *podWorkers) completeTerminating(pod *v1.Pod) {
	...
    // 更新链路状态
	if status, ok := p.podSyncStatuses[pod.UID]; ok {
		...
        // 关闭管道,通知驱逐pod完成(evitManager使用)
		for _, ch := range status.notifyPostTerminating {
			close(ch)
		}
        ...
	}
	// 将最后一个未更新状态变为当前pod,标记为terminated
	p.lastUndeliveredWorkUpdate[pod.UID] = podWork{
		WorkType: TerminatedPodWork,
		Options: UpdatePodOptions{
			Pod: pod,
		},
	}
}

// sync pod通知完成
func (p *podWorkers) completeSync(pod *v1.Pod) {
	...
	// 更新链路状态
	if status, ok := p.podSyncStatuses[pod.UID]; ok {
		...
		status.startedTerminating = true
	}
	// 将最后一个未更新状态变为当前pod,标记为terminating pod
	p.lastUndeliveredWorkUpdate[pod.UID] = podWork{
		WorkType: TerminatingPodWork,
		Options: UpdatePodOptions{
			Pod: pod,
		},
	}
}

// completeWork requeues on error or the next sync interval and then immediately executes any pending work.
func (p *podWorkers) completeWork(pod *v1.Pod, phaseTransition bool, syncErr error) {
	// 完成的pod加入workqueue,供syncCh函数处理
	switch {
	...
        p.workQueue.Enqueue(pod.UID,delay)
    ...
	}
	p.completeWorkQueueNext(pod.UID)
}

// completeWorkQueueNext holds the lock and either queues the next work item for the worker or
// clears the working status.
func (p *podWorkers) completeWorkQueueNext(uid types.UID) {
	...
    // pod还有请求
	if workUpdate, exists := p.lastUndeliveredWorkUpdate[uid]; exists {
        // 推入podWorker.chan处理
		p.podUpdates[uid] <- workUpdate
        // 清理已发送请求
		delete(p.lastUndeliveredWorkUpdate, uid)
	} else {
        // pod工作状态标记为false
		p.podSyncStatuses[uid].working = 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
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

作用

这个方法会遍历channel里的数据，然后调用syncPodFn方法，kubelet在执行NewMainKubelet方法的时候调用newPodWorkers方法设置syncPodFn。

# 9.syncTerminatedPodFn

// syncTerminatedPod cleans up a pod that has terminated (has no running containers).
func (kl *Kubelet) syncTerminatedPod(ctx context.Context,pod *v1.Pod,podStatus *kubecontainer.PodStatus) error {
	// 更新statusManager的pod status
	apiPodStatus := kl.generateAPIPodStatus(pod, podStatus)
	kl.statusManager.SetPodStatus(pod, apiPodStatus)

	// 等待volumn卸载
	if err := kl.volumeManager.WaitForUnmount(pod); err != nil {
		return err
	}
	...
    // secretManager注销pod(缓存及secretRef更新)
	kl.secretManager.UnregisterPod(pod)
	...
    // configmapManager管理器注销pod(缓存及cmRef更新)
	kl.configMapManager.UnregisterPod(pod)

	// 移除pod的cgroup
	if kl.cgroupsPerQOS {
		pcm := kl.containerManager.NewPodContainerManager()
		name, _ := pcm.GetPodContainerName(pod)
		pcm.Destroy(name)
        ...
	}
	// mark the final pod status
	kl.statusManager.TerminatePod(pod)
	...
	return nil
}

// 更新pod状态
func (m *manager) TerminatePod(pod *v1.Pod) {
	...
	// pod完成初始化
	if hasPodInitialized(pod) {
		for i := range status.ContainerStatuses {
			if status.ContainerStatuses[i].State.Terminated != nil {
				continue
			}
            // 标记为terminated状态
			status.ContainerStatuses[i].State = v1.ContainerState{
				Terminated: &v1.ContainerStateTerminated{
					Reason:   "ContainerStatusUnknown",
					Message:  "The container could not be located when the pod was terminated",
					ExitCode: 137,
				},
			}
		}
	}

	// 标记已初始化的容器
	for i := range initializedContainers(status.InitContainerStatuses) {
		if status.InitContainerStatuses[i].State.Terminated != nil {
			continue
		}
        // 标记为terminated状态
		status.InitContainerStatuses[i].State = v1.ContainerState{
			Terminated: &v1.ContainerStateTerminated{
				Reason:   "ContainerStatusUnknown",
				Message:  "The container could not be located when the pod was terminated",
				ExitCode: 137,
			},
		}
	}
    //更新pod status(状态更新请求发送到m.podStatusChannel)
	m.updateStatusInternal(pod, status, true)
}

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

# 10.syncTerminatingPodFn

// syncTerminatingPod is expected to terminate all running containers in a pod
func (kl *Kubelet) syncTerminatingPod(ctx context.Context, pod *v1.Pod, podStatus *kubecontainer.PodStatus, runningPod *kubecontainer.Pod, gracePeriod *int64, podStatusFn func(*v1.PodStatus)) error {
	// orphan pod删除
	if runningPod != nil {
		...
        // kill pod
		kl.killPod(pod, *runningPod, gracePeriod)
		return nil
	}
	// pod状态转换
	apiPodStatus := kl.generateAPIPodStatus(pod, podStatus)
    // 执行优雅删除回调
	if podStatusFn != nil {
		podStatusFn(&apiPodStatus)
	}
    // 更新statusManager的pod status
	kl.statusManager.SetPodStatus(pod, apiPodStatus)
	...
    // 停止存活探测
	kl.probeManager.StopLivenessAndStartup(pod)
    // pod status转换为running pod(便于遍历pod容器删除)
	p := kubecontainer.ConvertPodStatusToRunningPod(kl.getRuntime().Type(), podStatus)
    // 调用CRI删除pod
	kl.killPod(pod, p, gracePeriod)
    ...
	// 移除探测任务
	kl.probeManager.RemovePod(pod)
	// 获取运行时的容器组装pod状态
	podStatus, err := kl.containerRuntime.GetPodStatus(pod.UID, pod.Name, pod.Namespace)
	...
	var runningContainers []string
	...
	var containers []container
	...
	for _, s := range podStatus.ContainerStatuses {
		if s.State == kubecontainer.ContainerStateRunning {
			runningContainers = append(runningContainers, s.ID.String())
		}
		...
	}
	...
    // pod还有剩余容器运行,报错
	if len(runningContainers) > 0 {
		return fmt.Errorf("detected running containers after a successful KillPod, CRI violation: %v", runningContainers)
	}
	// 终止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

# 11.syncPodFn

# 11.1.入口

func (kl *Kubelet) syncPod(ctx context.Context, updateType kubetypes.SyncPodType, pod, mirrorPod *v1.Pod, podStatus *kubecontainer.PodStatus) (isTerminal bool, err error) {
	...
    // pod信息和podStatus(缓存对象信息)与内部oldStatus历史信息对比,转换为status运行信息
    apiPodStatus := kl.generateAPIPodStatus(pod, podStatus, false)
    ...
    // 1.pod是运行完成或失败终态,记录状态返回
	if apiPodStatus.Phase == v1.PodSucceeded || apiPodStatus.Phase == v1.PodFailed {
		kl.statusManager.SetPodStatus(pod, apiPodStatus)
		isTerminal = true
		return isTerminal, nil
	}
    // 2.校验该pod是否能运行(资源申请记录到cpuManager和MemoryManager)
	runnable := kl.canRunPod(pod)
    // 如果不能运行,回写container的等待原因
	if !runnable.Admit {
		// Pod is not runnable; and update the Pod and Container statuses to why.
		if apiPodStatus.Phase != v1.PodFailed && apiPodStatus.Phase != v1.PodSucceeded {
			apiPodStatus.Phase = v1.PodPending
		}
		apiPodStatus.Reason = runnable.Reason
		apiPodStatus.Message = runnable.Message
		// 回写容器等待原因
		const waitingReason = "Blocked"
		for _, cs := range apiPodStatus.InitContainerStatuses {
			if cs.State.Waiting != nil {
				cs.State.Waiting.Reason = waitingReason
			}
		}
		for _, cs := range apiPodStatus.ContainerStatuses {
			if cs.State.Waiting != nil {
				cs.State.Waiting.Reason = waitingReason
			}
		}
	}
	...
	// 3.更新statusManager的pod status
	kl.statusManager.SetPodStatus(pod, apiPodStatus)
	// 4.如果校验没通过则kill掉pod
	if !runnable.Admit {
		...
        // kill pod
		kl.killPod(pod, p, nil)
        ...
		return false, syncErr
	}
    // 5.校验网络插件是否准备好(非hostNet模式)
	kl.runtimeState.networkErrors()
    ...
	// 6.非终止请求,注册pod使用的secret及cm
	if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
        kl.secretManager.RegisterPod(pod)
		...	
		kl.configMapManager
        ...
	}
	// 7.初始化pcm,用于cgroup创建
	pcm := kl.containerManager.NewPodContainerManager()
	// 8.非终止请求,创建或更新pod
	if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
        // 9.校验该pod是否首次创建(是否存在startAt)
		firstSync := true
		for _, containerStatus := range apiPodStatus.ContainerStatuses {
			if containerStatus.State.Running != nil {
				firstSync = false
				break
			}
		}
		// 10.pod的cgroup不存在且非首次启动
		podKilled := false
		if !pcm.Exists(pod) && !firstSync {
			...
            // kill pod
            kl.killPod(pod, p, nil)
            podKilled = true
			...
		}
		// 11.pod未kill掉或重启策略不是restartNever
		if !(podKilled && pod.Spec.RestartPolicy == v1.RestartPolicyNever) {
            // pod cgroup不存在则创建
			if !pcm.Exists(pod) {
                // 更新QoS级别的cgroups,确保系统级cgroups层次结构正确(k8s维护的QoS树,用于资源优先级检查)
                // 1.所有容器指定limits和requests且两者相等,优先级最高,机会不会被系统回收
                // 2.至少一个容器指定requests,允许资源空闲时占用更多资源
                // 3.所有容器均未指定requests和limits,优先级最低
				kl.containerManager.UpdateQOSCgroups()
                ...
                // 创建pod的cgroups,确保pod资源限制生效
				pcm.EnsureExists(pod)
                ...
			}
		}
	}
    // 12.当前pod是static pod(来自配置文件)
	if kubetypes.IsStaticPod(pod) {
		deleted := false
        // mirror pod不为空
		if mirrorPod != nil {
            // mirror pod正在删除或mirror pod不是pod镜像
			if mirrorPod.DeletionTimestamp != nil || !kl.podManager.IsMirrorPodOf(mirrorPod, pod) {
				...
                // 删除mirror pod
				deleted, err = kl.podManager.DeleteMirrorPod(podFullName, &mirrorPod.ObjectMeta.UID)
				...
			}
		}
        // mirror pod为空或已删除
		if mirrorPod == nil || deleted {
            // 获取node
			node, err := kl.GetNode()
			...
            // node未删除,创建pod的mirror pod
            if node.DeletionTimestamp == nil{
                // 创建mirror pod
				kl.podManager.CreateMirrorPod(pod)
            }

		}
	}
	// 13.创建pod文件目录
	kl.makePodDataDirs(pod)
    ...
	// 14.非终止请求，需要等待attach/mount volumns
	if !kl.podWorkers.IsPodTerminationRequested(pod.UID) {
		// Wait for volumes to attach/mount
		kl.volumeManager.WaitForAttachAndMount(pod)
        ...
	}
	// 15.获取拉取镜像的secret
	pullSecrets := kl.getPullSecretsForPod(pod)
	// 16.增加pod探针
	kl.probeManager.AddPod(pod)
	// 17.调用CRI同步pod
	result := kl.containerRuntime.SyncPod(pod, podStatus, pullSecrets, kl.backOff)
    // 缓存执行结果(影响后续入队的pod status计算)
	kl.reasonCache.Update(pod.UID, result)
    // 非crash或imagePull失败返回err
	if err := result.Error(); err != nil {
		for _, r := range result.SyncResults {
			if r.Error != kubecontainer.ErrCrashLoopBackOff && r.Error != images.ErrImagePullBackOff {
				return false, err
			}
		}
		return false, nil
	}
	return false, nil
}

该方法主要为创建pod做一些准备工作，包括：

1.校验该pod能否运行，如果不能则回写container的等待原因，然后更新状态管理器的状态

2.如果校验没通过则kill掉pod并提前结束

3.校验网络插件是否准备好，没有则提前结束

4.该pod的cgroup不存在时则创建

5.为静态pod创建镜像

6.创建pod文件目录，等待volumns attach/mount

7.拉取pod secrets

8.调用containerRuntime.SyncPod真正创建pod

# 11.2.syncPod

kl.containerRuntime.SyncPod其实调用的是kubeGenericRuntimeManager.syncPod，kubeGenericRuntimeManager是更底层的运行时管理抽象，内部包装容器运行时和镜像运行时，遵循外观模式。

func (m *kubeGenericRuntimeManager) SyncPod(pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, backOff *flowcontrol.Backoff) (result kubecontainer.PodSyncResult) {
	// 1.计算pod变更行为,涉及sandbox变更及container变更
	podContainerChanges := m.computePodActions(pod, podStatus)
    ....
	// 2.sandbox变化,删除pod
	if podContainerChanges.KillPod {
		...
        // kill pod
		killResult := m.killPodWithSyncResult(pod, kubecontainer.ConvertPodStatusToRunningPod(m.runtimeName, podStatus), nil)
        ...
        // 创建sandbox,需要移除所有init容器(重建需要再次初始化)
		if podContainerChanges.CreateSandbox {
			m.purgeInitContainers(pod, podStatus)
		}
	} else {
		// 3.杀掉containersToKill列表中的container
		for containerID, containerInfo := range podContainerChanges.ContainersToKill {
			...
            m.killContainer(pod,containerID,containerInfo.name,containerInfo.message,containerInfo.reason,nil)
		}
	}
	// 4.清理同名的init container
	m.pruneInitContainersBeforeStart(pod, podStatus)
	...
	// 5.为pod创建sandbox
	podSandboxID := podContainerChanges.SandboxID
	if podContainerChanges.CreateSandbox {
		...
        // 调用CRI创建sandbox
		podSandboxID, msg, err = m.createPodSandbox(pod, podContainerChanges.Attempt)
		...
        // 查询sandbox状态
        resp, err := m.runtimeService.PodSandboxStatus(podSandboxID, false)
        ...
        
	}
	// 6.基于pod生成sandbox配置,涉及dns/hostName/端口映射
	podSandboxConfig, err := m.generatePodSandboxConfig(pod, podContainerChanges.Attempt)
    ...
    // 7.容器启动函数
	start := func(typeName, metricLabel string, spec *startSpec) error {
		...
        // CRI调用
		m.startContainer(podSandboxID, podSandboxConfig, spec, pod, podStatus, pullSecrets, podIP, podIPs)
        ...
	}
	// 8.临时容器相关
	if utilfeature.DefaultFeatureGate.Enabled(features.EphemeralContainers) {
        // 临时容器创建
		for _, idx := range podContainerChanges.EphemeralContainersToStart {
			start("ephemeral container", metrics.EphemeralContainer, ephemeralContainerStartSpec(&pod.Spec.EphemeralContainers[idx]))
		}
	}
	// 9.启动init容器
	if container := podContainerChanges.NextInitContainerToStart; container != nil {
		// Start the next init container.
		start("init container", metrics.InitContainer, containerStartSpec(container))
        ...
	}
	// 10.启动业务容器
	for _, idx := range podContainerChanges.ContainersToStart {
		start("container", metrics.Container, containerStartSpec(&pod.Spec.Containers[idx]))
	}
	return
}

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

主要流程

1.调用computePodActions计算pod变化，涉及sandbox重建及container增删

2.删除sandbox变化的pod及init容器

3.清理不应该再运行的container

4.调用pruneInitContainersBeforeStart()清理同名的init container

5.调用createPodSandbox()创建sandbox，同时获取sandbox状态

6.开启ephemeral feature特性，启动定义的临时容器

7.获取init container及启动

8.获取business container及启动

# 11.3.computePodActions

func (m *kubeGenericRuntimeManager) computePodActions(pod *v1.Pod, podStatus *kubecontainer.PodStatus) podActions {
	// 1.计算pod sandbox是否变化
    // a.sandbox不存在,pod删除重建
    // b.readySandbox多个,pod删除重建
    // c.第一个sandbox不是ready,pod删除重建
    // d.sandbox网络命名空间与pod期望的网络命名空间不一致,pod删除重建
    // e.非hostNetwork模式sandbox未分配IP,pod删除重建
	createPodSandbox, attempt, sandboxID := m.podSandboxChanged(pod, podStatus)
	changes := podActions{
		KillPod:           createPodSandbox,
		CreateSandbox:     createPodSandbox,
		SandboxID:         sandboxID,
		Attempt:           attempt,
		ContainersToStart: []int{},
		ContainersToKill:  make(map[kubecontainer.ContainerID]containerToKillInfo),
	}
	// 2.新建sandbox
	if createPodSandbox {
        // pod未设置重启策略&非首次创建&已有容器启动,不需要重建sandbox
		if !shouldRestartOnFailure(pod) && attempt != 0 && len(podStatus.ContainerStatuses) != 0 {
			changes.CreateSandbox = false
			return changes
		}
        // 3.将所有的container加入到需要启动的队列中(除了已启动且重启策略为RestartPolicyOnFailure的Pod)
        var containersToStart []int
		for idx, c := range pod.Spec.Containers {
            // restartOnFaliure且container启动成功不收集
			if pod.Spec.RestartPolicy == v1.RestartPolicyOnFailure && containerSucceeded(&c, podStatus) {
				continue
			}
			containersToStart = append(containersToStart, idx)
		}
        // 没有需要重启的容器
        if len(containersToStart) == 0 {
            // 找init容器
			_, _, done := findNextInitContainerToRun(pod, podStatus)
            // init容器都成功,不需要重建sandbox
			if done {
				changes.CreateSandbox = false
				return changes
			}
		}
		...
        // init容器不为0
		if len(pod.Spec.InitContainers) != 0 {
			// 4.获取首个init容器用于重启
			changes.NextInitContainerToStart = &pod.Spec.InitContainers[0]
			return changes
		}
        // 登记需要重启的业务容器
		changes.ContainersToStart = containersToStart
		return changes
	}
    // sandbox不需要重启
    // 5.登记需要启动的临时容器(未启动过的)
	if utilfeature.DefaultFeatureGate.Enabled(features.EphemeralContainers) {
		for i := range pod.Spec.EphemeralContainers {
			c := (*v1.Container)(&pod.Spec.EphemeralContainers[i].EphemeralContainerCommon)
			// Ephemeral Containers are never restarted
			if podStatus.FindContainerStatusByName(c.Name) == nil {
				changes.EphemeralContainersToStart = append(changes.EphemeralContainersToStart, i)
			}
		}
	}
	// 6.检查init container运行状态
	initLastStatus, next, done := findNextInitContainerToRun(pod, podStatus)
    // 7.init container未执行完成
	if !done {
        // 获取的init container不为空
		if next != nil {
            // init container运行失败且未设置不重启
			if initFailed && !shouldRestartOnFailure(pod) {
                // 重建pod
				changes.KillPod = true
			} else {
				// init container未知状态
				if initLastStatus != nil && initLastStatus.State == kubecontainer.ContainerStateUnknown {
					// 登记init container清理
                    changes.ContainersToKill[initLastStatus.ID] = containerToKillInfo{...}
				}
                // 登记init container启动
				changes.NextInitContainerToStart = next
			}
		}
		// 8.init container未全部完成,但是没有下一个需要执行的,说明存在running的init container,不再继续向下处理
		return changes
	}
	// 8.init已完成，计算需要kill&start的工作container
	keepCount := 0
	// 9.校验containers列表的状态
	for idx, container := range pod.Spec.Containers {
		containerStatus := podStatus.FindContainerStatusByName(container.Name)
		// 10.container未成功执行,先调用post-stop钩子走完生命周期
		if containerStatus != nil && containerStatus.State != kubecontainer.ContainerStateRunning {
			m.internalLifecycle.PostStopContainer(containerStatus.ID.ID)
            ...
		}
		// 11.如果container status不存在或未运行,根据重启策略决定容器是否重启
		if containerStatus == nil || containerStatus.State != kubecontainer.ContainerStateRunning {
            // 未运行过/未知或已创建状态/异常退出状态均需要重启
			if kubecontainer.ShouldContainerBeRestarted(&container, pod, podStatus) {
				// container登记到重启项
				changes.ContainersToStart = append(changes.ContainersToStart, idx)
                // container状态未知,登记到容器删除项
				if containerStatus != nil && containerStatus.State == kubecontainer.ContainerStateUnknown {
					// 12.如果container的状态是unknown，先kill再启动保证容器唯一
					changes.ContainersToKill[containerStatus.ID] = containerToKillInfo{...}
				}
			}
			continue
		}
        ...
        // 13.此时容器处于running状态
		restart := shouldRestartOnFailure(pod)
        ...
        // 1.container change强制重启
        // 2.liveness探测失败根据策略重启
        // 3.startup探测失败根据策略重启
        ...
		// 14.登记需要重启的容器
		if restart {
			changes.ContainersToStart = append(changes.ContainersToStart, idx)
		}
		// 15.登记需要删除的容器
		changes.ContainersToKill[containerStatus.ID] = containerToKillInfo{...}
	}
    // 需要保持和启动的容器都是0,清理sandbox
	if keepCount == 0 && len(changes.ContainersToStart) == 0 {
		changes.KillPod = true
	}
	return changes
}

主要流程

1.计算sandbox/container变化

2.sandbox有变化，计算需要重启的init/business container

3.sandbox无变化，计算需要重启及删除的init/business container，检查是否需要killPod

# 11.4.killPodWithSyncResult

func (m *kubeGenericRuntimeManager) killPodWithSyncResult(pod *v1.Pod, runningPod kubecontainer.Pod, gracePeriodOverride *int64) (result kubecontainer.PodSyncResult) {
    // 调用CRI停止容器
	killContainerResults := m.killContainersWithSyncResult(pod, runningPod, gracePeriodOverride)
	...
	// 调用CRI停止sandbox容器
	for _, podSandbox := range runningPod.Sandboxes {
		m.runtimeService.StopPodSandbox(podSandbox.ID.ID)
        ...
	}
	return
}

// killContainersWithSyncResult kills all pod's containers with sync results.
func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(pod *v1.Pod, runningPod kubecontainer.Pod, gracePeriodOverride *int64) (syncResults []*kubecontainer.SyncResult) {
	...
	for _, container := range runningPod.Containers {
		go func(container *kubecontainer.Container) {
			...
            // 停止容器
            m.killContainer(pod, container.ID, container.Name, "", reasonUnknown, gracePeriodOverride)
			...
		}(container)
	}
	...
	return
}

// sandbox的删除由containerGC触发
func (r *remoteRuntimeService) StopPodSandbox(podSandBoxID string) (err error) {
	....
    // 调用CRI停止sandbox
	r.runtimeClient.StopPodSandbox(ctx, &runtimeapi.StopPodSandboxRequest{PodSandboxId: podSandBoxID})
	....
	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

# 11.5.purgeInitContainers

func (m *kubeGenericRuntimeManager) purgeInitContainers(pod *v1.Pod, podStatus *kubecontainer.PodStatus) {
	...
	for name := range initContainerNames {
		for _, status := range podStatus.ContainerStatuses {
			if status.Name != name {
				continue
			}
            ...
            // 调用CRI删除容器
			m.removeContainer(status.ID.ID)
            ...
		}
	}
}

func (m *kubeGenericRuntimeManager) removeContainer(containerID string) error {
	// 调用post-stop回调,走完容器生命周期
	m.internalLifecycle.PostStopContainer(containerID)
	...
    // 移除容器日志
    m.removeContainerLog(containerID)
    ...
    // 调用CRI移除容器
	return m.runtimeService.RemoveContainer(containerID)
}

// removeContainerLog removes the container log.
func (m *kubeGenericRuntimeManager) removeContainerLog(containerID string) error {
    // 调用日志管理器删除旋转日志文件
	err := m.logManager.Clean(containerID)
	...
	// 调用CRI查询容器状态
	resp, err := m.runtimeService.ContainerStatus(containerID, false)
	...
    // 容器标签解析容器名称、pod名称、命名空间
	labeledInfo := getContainerInfoFromLabels(status.Labels)
    // 生成旧版日志符号链接路径
	legacySymlink := legacyLogSymlink(containerID, labeledInfo.ContainerName, labeledInfo.PodName,
		labeledInfo.PodNamespace)
    ...
    // 删除符号链接,避免新旧日志路径并存导致混淆
	m.osInterface.Remove(legacySymlink)
	return nil
}

func (c *containerLogManager) Clean(containerID string) error {
	...
    // 调用CRI获取容器状态
	resp, err := c.runtimeService.ContainerStatus(containerID, false)
	...
    // 获取容器日志文件
	pattern := fmt.Sprintf("%s*", resp.GetStatus().GetLogPath())
	logs, err := c.osInterface.Glob(pattern)
	...
    // 删除容器日志目录
	for _, l := range logs {
		c.osInterface.Remove(l)
        ...
	}
	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

# 11.6.killContainer

// killContainer kills a container through the following steps:
// * Run the pre-stop lifecycle hooks (if applicable).
// * Stop the container.
func (m *kubeGenericRuntimeManager) killContainer(pod *v1.Pod, containerID kubecontainer.ContainerID, containerName string, message string, reason containerKillReason, gracePeriodOverride *int64) error {
	var containerSpec *v1.Container
	if pod != nil {
        // pod中获取containerSpec
		containerSpec =kubecontainer.GetContainerSpec(pod, containerName)
        ...
	} else {
		// pod为空,从container labels恢复pod和容器信息
		pod, containerSpec = m.restoreSpecsFromContainerLabels(containerID)
		...
	}

    // 优雅删除时间(默认为2)
	gracePeriod := int64(minimumGracePeriodInSeconds)
	switch {
    // 删除时指定了优雅删除时间(kubectl delete --grace-period)或apiserver默认给的30s
	case pod.DeletionGracePeriodSeconds != nil:
		gracePeriod = *pod.DeletionGracePeriodSeconds
    // pod配置了优雅删除时间
	case pod.Spec.TerminationGracePeriodSeconds != nil:
		gracePeriod = *pod.Spec.TerminationGracePeriodSeconds
		switch reason {
        // 启动探针失败
		case reasonStartupProbe:
            // 启动探测配置了优雅删除时间
			if containerSpec.StartupProbe!= nil&containerSpec.StartupProbe.TerminationGracePeriodSeconds!=nil {
                // 以启动探针的优雅删除时间为准
				gracePeriod = *containerSpec.StartupProbe.TerminationGracePeriodSeconds
			}
        // 存活探针失败
		case reasonLivenessProbe:
  			// 存活探针配置了优雅删除时间
			if containerSpec.LivenessProbe!=nil&containerSpec.LivenessProbe.TerminationGracePeriodSeconds!=nil {
				// 以存活探针的优雅删除时间为准
                gracePeriod = *containerSpec.LivenessProbe.TerminationGracePeriodSeconds
			}
		}
	}
	...
    // 执行pre-stop钩子
	if containerSpec.Lifecycle != nil && containerSpec.Lifecycle.PreStop != nil && gracePeriod > 0 {
        // 优雅删除时间更新
		gracePeriod = gracePeriod - m.executePreStopHook(pod, containerID, containerSpec, gracePeriod)
	}
	// 优雅删除时间低于默认值进行更新
	if gracePeriod < minimumGracePeriodInSeconds {
		gracePeriod = minimumGracePeriodInSeconds
	}
	...
    // 调用CRI停止容器(剩余的优雅退出时间作为超时时间)
	m.runtimeService.StopContainer(containerID.ID, gracePeriod)
	...
	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

# 11.7.pruneInitContainersBeforeStart

func (m *kubeGenericRuntimeManager) pruneInitContainersBeforeStart(pod *v1.Pod, podStatus *kubecontainer.PodStatus) {
	...
    // 遍历init container
	for name := range initContainerNames {
		count := 0
		for _, status := range podStatus.ContainerStatuses {
            // 非异常退出/Unknown的容器跳过
			if status.Name != name ||
				(status.State != kubecontainer.ContainerStateExited &&
					status.State != kubecontainer.ContainerStateUnknown) {
				continue
			}
			// 保留第一个同名容器
			count++
			// keep the first init container for this name
			if count == 1 {
				continue
			}
			...
            // 基于ID删除其它容器
            m.removeContainer(status.ID.ID)
            ...
		}
	}
}

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

# 11.8.createPodSandbox

// createPodSandbox creates a pod sandbox and returns (podSandBoxID, message, error).
func (m *kubeGenericRuntimeManager) createPodSandbox(pod *v1.Pod, attempt uint32) (string, string, error) {
	// 生成sandbox配置
    podSandboxConfig, err := m.generatePodSandboxConfig(pod, attempt)
	...
    // 创建pod日志目录(/var/log/pods/<podUID>)
	err = m.osInterface.MkdirAll(podSandboxConfig.LogDirectory, 0755)
	...
	// runtimeClassManager获取运行时
	runtimeHandler := ""
	if m.runtimeClassManager != nil {
		runtimeHandler, err = m.runtimeClassManager.LookupRuntimeHandler(pod.Spec.RuntimeClassName)
		...
	}
	// 调用CRI运行sandbox
	podSandBoxID, err := m.runtimeService.RunPodSandbox(podSandboxConfig, runtimeHandler)
	...
	return podSandBoxID, "", nil
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

# 11.9.startContainer

func (m *kubeGenericRuntimeManager) startContainer(podSandboxID string, podSandboxConfig *runtimeapi.PodSandboxConfig, spec *startSpec, pod *v1.Pod, podStatus *kubecontainer.PodStatus, pullSecrets []v1.Secret, podIP string, podIPs []string) (string, error) {
	container := spec.container
	// 1.拉取镜像
	imageRef, msg, err := m.imagePuller.EnsureImageExists(pod, container, pullSecrets, podSandboxConfig)
	...
	// 2.生成container配置
	containerConfig, cleanupAction, err := m.generateContainerConfig(container, pod, restartCount, podIP, imageRef, podIPs, target)
	if cleanupAction != nil {
		defer cleanupAction()
	}
	...
    // 3.调用internal-pre-create生命周期钩子(更新container配置中CPU和Memory亲和性信息,作为容器runtime config一部分)
	m.internalLifecycle.PreCreateContainer(pod, container, containerConfig)
	...
	// 4.调用CRI接口创建container
	containerID, err := m.runtimeService.CreateContainer(podSandboxID, containerConfig, podSandboxConfig)
	...
    // 5.调用internal-pre-start生命周期钩子(cpuManager和memoryManager登记容器)
	err = m.internalLifecycle.PreStartContainer(pod, container, containerID)
	...
	// 6.调用CRI接口启动container
	m.runtimeService.StartContainer(containerID)
 	...
	// 7.容器生命周期钩子
	if container.Lifecycle != nil && container.Lifecycle.PostStart != nil {
		...
        // 8.执行post-start钩子
		msg, handlerErr := m.runner.Run(kubeContainerID, pod, container, container.Lifecycle.PostStart)
		if handlerErr != nil {
			...
            // 9.post-start钩子执行失败视为严重错误,杀掉容器
			m.killContainer(pod, kubeContainerID, container.Name, "FailedPostStartHook", reasonFailedPostStartHook, nil)
            ...
		}
	}
	return "", nil
}

// EnsureImageExists pulls the image for the specified pod and container.
func (m *imageManager) EnsureImageExists(pod *v1.Pod, container *v1.Container, pullSecrets []v1.Secret, podSandboxConfig *runtimeapi.PodSandboxConfig) (string, string, error) {
	...
    // 镜像tag(默认为latest)
	image, err := applyDefaultImageTag(container.Image)
	...
    // 初始化镜像定义(镜像tag+pod annotations)
	spec := kubecontainer.ImageSpec{
		Image:       image,
		Annotations: podAnnotations,
	}
    // 调用CRI获取镜像引用信息(镜像签名及spec数据)
	imageRef, err := m.imageService.GetImageRef(spec)
	...
    // 生成backOffKey,检查是否拉取失败过,记录事件
    backOffKey := fmt.Sprintf("%s_%s", pod.UID, container.Image)
	if m.backOff.IsInBackOffSinceUpdate(backOffKey, m.backOff.Clock.Now()) {
		return "", msg, ErrImagePullBackOff
	}
    ...
	// 镜像拉取
	m.puller.pullImage(spec, pullSecrets, pullChan, podSandboxConfig)
	...
    // 清理相关的backOff状态(清理失败超出2分钟的backOff记录)
	m.backOff.GC()
	return imagePullResult.imageRef, "", 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

流程梳理

1.检查镜像是否存在，不存在调用CRI拉取镜像

2.生成container相关配置，调用internal-pre-create钩子更新关联的cpu及memory资源

3.调用CRI接口创建container

4.调用internal-pre-start钩子向cpuManager和memoryManager登记container，确认资源绑定

5.调用CRI启动container

6.执行用户设置的post-start钩子