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

# 前言

job关联的Pod设置postStart钩子可以正常触发，设置preStop钩子不会触发，无法进一步执行终止前的清理工作，但job对象是支持定义的。
注意

job Pod由kubelet管理，理论上postStart和preStop都应触发，下面会分析kubelet回收Pod流程，查找preStop未触发原因。

# 原理

ops002

job pod会委派给kubelet创建或回收，由于创建阶段与普通Pod没什么区别，所以postStart钩子可以正常触发，下面主要看preStop触发

# pleg

job pod是一次性任务，执行完成后会主动退出，这种退出不是kubelet基于资源压力或污点容忍发起，也无法基于apiserver监听。

// Start spawns a goroutine to relist periodically.
func (g *GenericPLEG) Start() {
	...
    // 间隔1s交互一次CRI查询所有container,组装出pod及生成事件
    go wait.Until(g.Relist, g.relistDuration.RelistPeriod, g.stopCh)
}

1
2
3
4
5
6

由于这种主动退出信号kubelet无法感知，只能借助pleg模块轮询检测CRI获取container，组装Pod及生成Pod相关事件。

// Relist queries the container runtime for list of pods/containers, compare
// with the internal pods/containers, and generates events accordingly.
func (g *GenericPLEG) Relist() {
	...

	// 基于CRI查询及组装所有Pod
    // 先查出所有sandbox,再基于sandbox uid关联其它container
	podList, err := g.runtime.GetPods(ctx, true)
	if err != nil {
		return
	}
	...
	pods := kubecontainer.Pods(podList)
	...
    g.podRecords.setCurrent(pods)
    ...
    
	// Compare the old and the current pods, and generate events.
	for pid := range g.podRecords {
		oldPod := g.podRecords.getOld(pid)
		pod := g.podRecords.getCurrent(pid)
		// Get all containers in the old and the new pod.
		allContainers := getContainersFromPods(oldPod, pod)
		for _, container := range allContainers {
            // 生成事件(关键)
			events := computeEvents(oldPod, pod, &container.ID)
            // 记录到eventsByPodID
			for _, e := range events {
				updateEvents(eventsByPodID, e)
			}
		}
	}
	...

	// If there are events associated with a pod, we should update the podCache.
	for pid, events := range eventsByPodID {
        // curPod
		pod := g.podRecords.getCurrent(pid)
        // cacheEnable
		if g.cacheEnabled() {
			// 交互CRI更新podStatus缓存
			if err, updated := g.updateCache(ctx, pod, pid); err != nil {
				// 失败记录到重试队列
				needsReinspection[pid] = pod
				continue
			} else {
				// 成功由重试队列删除
				delete(g.podsToReinspect, pid)
				...
			}
		}
        
        // 转置(old=cur,cur=nil)
		g.podRecords.update(pid)
		...

        // 向kubelet mainLoop发送事件
		for i := range events {
			// Filter out events that are not reliable and no other components use yet.
            // containerChanged事件代表不稳定
			if events[i].Type == ContainerChanged {
				continue
			}
            
			select {
            // 发事件到eventChannel,pleg.eventChannel是kubelet主循环的事件来源之一
            // job pod终止发送的是ContainerDied事件
			case g.eventChannel <- events[i]:
			default:
				klog.ErrorS(nil, "Event channel is full, discard this relist() cycle event")
			}
            ...
		}
	}

    // podStatus更新重试
	if g.cacheEnabled() {
		// reinspect any pods that failed inspection during the previous relist
		if len(g.podsToReinspect) > 0 {
			for pid, pod := range g.podsToReinspect {
                // 再次尝试更新podStatus
				if err, _ := g.updateCache(ctx, pod, pid); err != nil {
					// Rely on updateCache calling GetPodStatus to log the actual error.
					needsReinspection[pid] = pod
				}
			}
		}
		...
	}

	// 记录重试队列
	g.podsToReinspect = needsReinspection
}

computeEvents()会根据container在新旧Pod的状态计算containerEvent，这里我们只关心job pod退出对应的事件。

func computeEvents(oldPod, newPod *kubecontainer.Pod, cid *kubecontainer.ContainerID) []*PodLifecycleEvent {
	var pid types.UID
	if oldPod != nil {
		pid = oldPod.ID
	} else if newPod != nil {
		pid = newPod.ID
	}
	oldState := getContainerState(oldPod, cid)
	newState := getContainerState(newPod, cid)
	return generateEvents(pid, cid.ID, oldState, newState)
}

func generateEvents(podID types.UID, cid string, oldState, newState plegContainerState) []*PodLifecycleEvent {
	if newState == oldState {
		return nil
	}

	switch newState {
    // other-->running
	case plegContainerRunning:
		return []*PodLifecycleEvent{{ID: podID, Type: ContainerStarted, Data: cid}}
    // other-->exited
	case plegContainerExited:
        // job对应container就发这种事件
		return []*PodLifecycleEvent{{ID: podID, Type: ContainerDied, Data: cid}}
    // other-->unknown(不稳定)
	case plegContainerUnknown:
		return []*PodLifecycleEvent{{ID: podID, Type: ContainerChanged, Data: cid}}
	// other-->non exist
    case plegContainerNonExistent:
		switch oldState {
        // exited-->non exist
		case plegContainerExited:
			// We already reported that the container died before.
			return []*PodLifecycleEvent{{ID: podID, Type: ContainerRemoved, Data: cid}}
        // other-->non exist
		default:
			return []*PodLifecycleEvent{{ID: podID, Type: ContainerDied, Data: cid}, {ID: podID, Type: ContainerRemoved, Data: cid}}
		}
	default:
		panic(fmt.Sprintf("unrecognized container state: %v", newState))
	}
}

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

注意

job pod终止后，发送的事件依次是ContainerDied-->ContainerRemoved

# kubelet

pleg发送的事件会由kubelet主循环进行处理，主要是syncLoopIteration大循环，会监听pleg.eventChannel进行事件消费。

func (kl *Kubelet) syncLoopIteration(...) bool {
	select {
    ...
    // 这里忽略GC协程、Probe探测、静态配置、Http及来自apiserver监听的事件,只看pleg模块
	case e := <-plegCh:
		...
        // 非ContainerRemoved事件
        if isSyncPodWorthy(e) {
			// 获取podManager缓存的Pod
			if pod, ok := kl.podManager.GetPodByUID(e.ID); ok {
                // dispatch分发
				handler.HandlePodSyncs([]*v1.Pod{pod})
			}
		}

        // ContainerDied事件
		if e.Type == pleg.ContainerDied {
			if containerID, ok := e.Data.(string); ok {
                // 清理对应容器
				kl.cleanUpContainersInPod(e.ID, containerID)
			}
		}
    ...
	}
	return 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

先来看handler.HandlePodSyncs()，主要负责将相关的Pod分发到处理模块，基于updateType进行生命周期管理。

// HandlePodSyncs is the callback in the syncHandler interface for pods
// that should be dispatched to pod workers for sync.
func (kl *Kubelet) HandlePodSyncs(pods []*v1.Pod) {
	...
	for _, pod := range pods {
        // 获取mirror pod(主要用于apiserver这类静态pod)
		mirrorPod, _ := kl.podManager.GetMirrorPodByPod(pod)
        // 分发sync事件
		kl.dispatchWork(pod, kubetypes.SyncPodSync, mirrorPod, start)
	}
}

// dispatchWork starts the asynchronous sync of the pod in a pod worker.
// If the pod has completed termination, dispatchWork will perform no action.
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,	// SyncPodSync
		StartTime:  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

kl.podWorkers.UpdatePod()是核心处理模块，负责轮转Pod state及根据state执行终止、回收、清理或创建工作。

// UpdatePod carries a configuration change or termination state to a pod. A pod is either runnable,
// terminating, or terminated, and will transition to terminating if: deleted on the apiserver,
// discovered to have a terminal phase (Succeeded or Failed), or evicted by the kubelet.
func (p *podWorkers) UpdatePod(options UpdatePodOptions) {
	...
    uid, ns, name = options.Pod.UID, options.Pod.Namespace, options.Pod.Name
	...
    
    // 状态机检查
	status, ok := p.podSyncStatuses[uid]
	if !ok {
        // 先假设Pod首次处理
		firstTime = true
        // 设置默认状态
		status = &podSyncStatus{
			syncedAt: now,
			fullname: kubecontainer.BuildPodFullName(name, ns),
		}
        
        // APIServer Pod走到终态(重启状态重建,基于APIServer对象状态和运行时容器状态推断)
		if options.Pod.Status.Phase == v1.PodFailed || options.Pod.Status.Phase == v1.PodSucceeded {
            // 这里的podCache就是和pleg共用的podCache
			if statusCache, err := p.podCache.Get(uid); err == nil {
                // 运行时检测到Pod container均终止
				if isPodStatusCacheTerminal(statusCache) {
					// 设置终止状态
					status = &podSyncStatus{
						terminatedAt:       now,
						terminatingAt:      now,
						syncedAt:           now,
						startedTerminating: true,
						finished:           false,
						fullname:           kubecontainer.BuildPodFullName(name, ns),
					}
				}
			}
		}
        // 记录syncStatus
		p.podSyncStatuses[uid] = status
	}

	pod := options.Pod
	...

    // Pod不是第一次被kubelet观测到&处于terminating状态,但收到了create事件
    // 静态Pod重置或apiserver由于抖动重发对象,这里延迟处理
	if !firstTime && status.IsTerminationRequested() {
		if options.UpdateType == kubetypes.SyncPodCreate {
			status.restartRequested = true
			return
		}
	}

	// Pod已结束生命周期
	if status.IsFinished() {
		return
	}
	...
    
	// job pod正常情况会进这里,重启重建状态不会进这里
	if !status.IsTerminationRequested() {
		switch {
        // 运行时容器存在,对象不存在的Pod
        case isRuntimePod:
			status.deleted = true
			status.terminatingAt = now
			becameTerminating = true
        // 正常删除Pod
		case pod.DeletionTimestamp != nil:
			status.deleted = true
			status.terminatingAt = now
			becameTerminating = true
        // 走到终态的Pod
		case pod.Status.Phase == v1.PodFailed, pod.Status.Phase == v1.PodSucceeded:
			status.terminatingAt = now
			becameTerminating = true
        // 由于驱逐触发清理的Pod
		case options.UpdateType == kubetypes.SyncPodKill:
			if options.KillPodOptions != nil && options.KillPodOptions.Evict {
				status.evicted = true
			}
			status.terminatingAt = now
			becameTerminating = true
		}
	}

	// 一旦kill开始,状态只能按照Running → Terminating → Terminated推进
	var wasGracePeriodShortened bool
	switch {
    // Pod已走到Terminated状态
	case status.IsTerminated():
		// 已经cleanup过但运行时容器依然存在
		if isRuntimePod {
			return
		}

        // 通知驱逐模块kill结束
		if options.KillPodOptions != nil {
			if ch := options.KillPodOptions.CompletedCh; ch != nil {
				close(ch)
			}
		}
		options.KillPodOptions = nil
	// Pod处于Terminating状态
	case status.IsTerminationRequested():
		...
		// 重新计算优雅退出剩余时间
		gracePeriod, gracePeriodShortened := calculateEffectiveGracePeriod(status, pod, options.KillPodOptions)
		...
		status.gracePeriod = gracePeriod
		// always set grace period for syncTerminatingPod so we don't have to recalculate, will never be zero.
		options.KillPodOptions.PodTerminationGracePeriodSecondsOverride = &gracePeriod
	// 正常生命周期
	default:
        // kill请求(无效),直接通知上游结束kill
		if options.KillPodOptions != nil {
			if ch := options.KillPodOptions.CompletedCh; ch != nil {
				close(ch)
			}
            // 重置非法参数
			options.KillPodOptions = nil
		}
	}

	// 一个pod对应一个pod worker
	podUpdates, exists := p.podUpdates[uid]
    // pod worker还不存在
	if !exists {
		// 初始化pod worker对应channel
		podUpdates = make(chan struct{}, 1)
		p.podUpdates[uid] = podUpdates

		// 静态pod
		if kubetypes.IsStaticPod(pod) {
            // 将同名Pod UID记录到队列,基于收到顺序启动
			p.waitingToStartStaticPodsByFullname[status.fullname] =
				append(p.waitingToStartStaticPodsByFullname[status.fullname], uid)
		}
		...
		
        outCh = podUpdates

		// spawn a pod worker
		go func() {
			// 启动pod worker
			p.podWorkerLoop(uid, outCh)
		}()
	}
	...

    // 标记pod正在处理
	status.working = true
	select {
    // 发送信号
	case podUpdates <- struct{}{}:
	default:
	}
	
    // Pod状态转换/优雅时间缩容,需取消旧操作重新处理
	if (becameTerminating || wasGracePeriodShortened) && status.cancelFn != nil {
		status.cancelFn()
		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
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

p.podWorkerLoop()负责处理关联Pod的完整生命周期，是Pod串行化处理机制的心脏，确保Pod状态迁移及回收有序执行。

// podWorkerLoop manages sequential state updates to a pod in a goroutine, exiting once final state is reached.
func (p *podWorkers) podWorkerLoop(podUID types.UID, podUpdates <-chan struct{}) {
	...
    // 监听到信号
	for range podUpdates {
        // 计算sync状态
        // update.WorkType取的是status的检查结果
        // 基于terminatedAt/terminatingAt检测是否终止
		ctx, update, canStart, canEverStart, ok := p.startPodSync(podUID)
        // sync失败或暂时无法启动,等待下一轮调度(job pod已经启动过,这里是准入的)
		if !ok || !canEverStart || !canStart {
			continue
		}
        
		...
		podUID, podRef := podUIDAndRefForUpdate(update.Options)
		...
		
        // 检测Pod终止状态
		err := func() error {
			...
			switch {
			case update.Options.RunningPod != nil:
				// 收到runningPod,不需要其它额外状态
			default:
                // 获取缓存的podStatus
				status = p.podCache.GetNewerThan(update.Options.Pod.UID, lastSyncTime)
				...
			}

			switch {
            // Terminated Pod处理
			case update.WorkType == TerminatedPod:
				err = p.podSyncer.SyncTerminatedPod(ctx, update.Options.Pod, status)
            // Terminating Pod处理
			case update.WorkType == TerminatingPod:
				...
				// if we only have a running pod, terminate it directly
                err = p.podSyncer.SyncTerminatingPod(ctx, update.Options.Pod, status, gracePeriod, ...)
			// 正常运行
			default:
                // 进行状态同步
				isTerminal, err = p.podSyncer.SyncPod(ctx, update.Options.UpdateType, update.Options.Pod, ...)
			}
			...
            
			return err
		}()

		switch {
		...
		case update.WorkType == TerminatedPod:
			// we can shut down the worker
			p.completeTerminated(podUID)
			...
			return
		case update.WorkType == TerminatingPod:
			// pods that don't exist in config don't need to be terminated, other loops will clean them up
			if update.Options.RunningPod != nil {
				p.completeTerminatingRuntimePod(podUID)
				...
				return
			}
			// otherwise we move to the terminating phase
			p.completeTerminating(podUID)
			phaseTransition = true

		case isTerminal:
			// if syncPod indicated we are now terminal, set the appropriate pod status to move to terminating
			p.completeSync(podUID)
			phaseTransition = true
		}

		// queue a retry if necessary, then put the next event in the channel if any
		p.completeWork(podUID, phaseTransition, err)
		...
	}
}

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

由上述程序可以看出，job pod终止后会先进入Terminating流程进行相关运行时容器资源回收，Terminating分支会调用preStop钩子。

// SyncTerminatingPod is expected to terminate all running containers in a pod.
func (kl *Kubelet) SyncTerminatingPod(_ context.Context, pod *v1.Pod, ...) error {
	...
    // 这里忽略基于podStatus更新statusManager维护的Pod状态,进而patch到APIServer
    ...
    
	// 终止探针
	kl.probeManager.StopLivenessAndStartup(pod)
	...
   	
    
    // 基于运行时容器状态组装出一个runningPod对象
    p := kubecontainer.ConvertPodStatusToRunningPod(kl.getRuntime().Type(), podStatus)
    // 执行回调及终止容器
	kl.killPod(ctx, pod, p, gracePeriod)
    ...

	// 移除探针
	kl.probeManager.RemovePod(pod)
	...
    
    // 再次获取运行时容器状态
    podStatus, err := kl.containerRuntime.GetPodStatus(ctx, pod.UID, pod.Name, pod.Namespace)
	...
    
    // 所有容器未Running,terminating阶段结束,更新最新的podStatus
    kl.statusManager.SetPodStatus(pod, apiPodStatus)
	return nil
}

// ConvertPodStatusToRunningPod returns Pod given PodStatus and container runtime string.
func ConvertPodStatusToRunningPod(runtimeName string, podStatus *PodStatus) Pod {
	runningPod := Pod{
		ID:        podStatus.ID,
		Name:      podStatus.Name,
		Namespace: podStatus.Namespace,
	}
    
    // 这个podStatus来自pleg.podCache缓存的容器状态Pod
	for _, containerStatus := range podStatus.ContainerStatuses {
        // job pod没有任何运行中的容器,不会组装任何container
		if containerStatus.State != ContainerStateRunning {
			continue
		}
		container := &Container{
			ID:                   containerStatus.ID,
			Name:                 containerStatus.Name,
			Image:                containerStatus.Image,
			ImageID:              containerStatus.ImageID,
			Hash:                 containerStatus.Hash,
			HashWithoutResources: containerStatus.HashWithoutResources,
			State:                containerStatus.State,
		}
		runningPod.Containers = append(runningPod.Containers, container)
	}

	// Populate sandboxes in kubecontainer.Pod
	for _, sandbox := range podStatus.SandboxStatuses {
		runningPod.Sandboxes = append(runningPod.Sandboxes, &Container{
			ID:    ContainerID{Type: runtimeName, ID: sandbox.Id},
			State: SandboxToContainerState(sandbox.State),
		})
	}
	return runningPod
}

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

kl.killPod()会结束Pod容器的运行状态及执行preStop回调，由于job pod对应runningPod没有存活容器，不会触发killContainer。

// killPod instructs the container runtime to kill the pod.
func (kl *Kubelet) killPod(ctx context.Context, pod *v1.Pod, ...) error {
    // 执行回调及终止容器
	kl.containerRuntime.KillPod(ctx, pod, p, gracePeriodOverride)
    ...
    
    // 重置container qos等级
	kl.containerManager.UpdateQOSCgroups()
    ...
    
	return nil
}

// KillPod kills all the containers of a pod. Pod may be nil, running pod must not be.
func (m *kubeGenericRuntimeManager) KillPod(ctx context.Context, pod *v1.Pod, ...) error {
	err := m.killPodWithSyncResult(ctx, pod, runningPod, gracePeriodOverride)
	return err.Error()
}

// killPodWithSyncResult kills a runningPod and returns SyncResult.
// 容器和sandbox所谓的kill都是只终止进程,不会直接回收,回收由GC模块完成
func (m *kubeGenericRuntimeManager) killPodWithSyncResult(ctx context.Context, pod *v1.Pod, ...) (...) {
    // 先终止容器
	m.killContainersWithSyncResult(ctx, pod, runningPod, gracePeriodOverride)
	...

	// 再终止所有sandbox
	for _, podSandbox := range runningPod.Sandboxes {
		m.runtimeService.StopPodSandbox(ctx, podSandbox.ID.ID)
        ...
	}

	return
}

// killContainersWithSyncResult kills all pod's containers with sync results.
// job pod不会触发m.killContainer
func (m *kubeGenericRuntimeManager) killContainersWithSyncResult(ctx context.Context, ...) (...) {
	...

	wg.Add(len(runningPod.Containers))
	for _, container := range runningPod.Containers {
        // 由于job Pod没有任何存活容器,组装的runningPod没有任何container对象,不会触发killContainer
		go func(container *kubecontainer.Container) {
			...
            
            // 终止容器
			m.killContainer(ctx, pod, container.ID, container.Name, "", reasonUnknown, gracePeriodOverride)
            ...
		}(container)
	}
	wg.Wait()
	...
    
	return
}

// killContainer kills a container through the following steps:
// * Run the pre-stop lifecycle hooks (if applicable).
// * Stop the container.
func (m *kubeGenericRuntimeManager) killContainer(ctx context.Context, ...) error {
	containerSpec = kubecontainer.GetContainerSpec(pod, containerName)
    ...

	// From this point, pod and container must be non-nil.
	gracePeriod := setTerminationGracePeriod(pod, containerSpec, containerName, containerID, reason)
    
	// Run the pre-stop lifecycle hooks if applicable and if there is enough time to run it
	if containerSpec.Lifecycle != nil && containerSpec.Lifecycle.PreStop != nil && gracePeriod > 0 {
		gracePeriod = gracePeriod - m.executePreStopHook(ctx, pod, containerID, containerSpec, gracePeriod)
	}
    
	// always give containers a minimal shutdown window to avoid unnecessary SIGKILLs
	if gracePeriod < minimumGracePeriodInSeconds {
		gracePeriod = minimumGracePeriodInSeconds
	}
	if gracePeriodOverride != nil {
		gracePeriod = *gracePeriodOverride
	}
	...
    
    // 终止容器
	m.runtimeService.StopContainer(ctx, containerID.ID, gracePeriod)

	return nil
}

terminating阶段完成后，p.completeTerminating会更新podStatus及设置terminatedAt标记，用以触发下一轮的定时调度。

// completeTerminating is invoked when syncTerminatingPod completes successfully, which means
// no container is running, no container will be started in the future, and we are ready for
// cleanup.
func (p *podWorkers) completeTerminating(podUID types.UID) {
	...

    // 获取podSyncStatus
	status, ok := p.podSyncStatuses[podUID]
	...
    
    // 设置terminatedAt标记
	status.terminatedAt = p.clock.Now()
	...

	// 必要时会再次触发入队
	p.requeueLastPodUpdate(podUID, status)
}

// creates a new pending pod update from the most recently executed update if no update is already queued, and // then notifies the pod worker goroutine of the update.
func (p *podWorkers) requeueLastPodUpdate(podUID types.UID, status *podSyncStatus) {
	// 还有待更新项 || 最近未处理过更新
	if status.pendingUpdate != nil || status.activeUpdate == nil {
		return
	}
    
    // 基于最近过的更新再生成一次信号
	copied := *status.activeUpdate
	status.pendingUpdate = &copied

	// notify the pod worker
	status.working = true
	select {
	case p.podUpdates[podUID] <- struct{}{}:
	default:
	}
}

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

p.podUpdates再次推入信号后，pod worker会进行terminated流程调度，主要回收卷资源及内部缓存，确保Pod仅剩对象，相关资源已释放。

// 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 {
	...

    // 更新statusMgr维护的状态缓存,用于异步更新APIServer对象状态
	kl.statusManager.SetPodStatus(pod, apiPodStatus)

	// volume卸载完成
	if err := kl.volumeManager.WaitForUnmount(pod); err != nil {
		return err
	}

	if !kl.keepTerminatedPodVolumes {
		// 间隔一段事件检测volume清理完成,直到超时
		wait.PollUntilContextCancel(ctx, 100*time.Millisecond, true, func(ctx context.Context) (bool, error) {
			volumesExist := kl.podVolumesExist(pod.UID)
			return !volumesExist, nil
		})
        ...
	}

	// 清理secret/cm-->Pod缓存
	if kl.secretManager != nil {
		kl.secretManager.UnregisterPod(pod)
	}
	if kl.configMapManager != nil {
		kl.configMapManager.UnregisterPod(pod)
	}

	// 清理cgroup对应qos数据
	if kl.cgroupsPerQOS {
		pcm := kl.containerManager.NewPodContainerManager()
		name, _ := pcm.GetPodContainerName(pod)
		pcm.Destroy(name)
        ...
	}

    // 释放用户命名空间
	kl.usernsManager.Release(pod.UID)

	// 更新pod状态为terminated
	kl.statusManager.TerminatePod(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

此外，p.completeWork()会在sync/terminating阶段后主动入队一次Pod，由sync模块定时触发一次分发处理，确保资源清理干净。

// completeWork requeues on error or the next sync interval and then immediately executes any pending
// work.
func (p *podWorkers) completeWork(podUID types.UID, phaseTransition bool, syncErr error) {
	// Requeue the last update if the last sync returned error.
	switch {
	case phaseTransition:
        // sync/terminating处理后会主动入队Pod
		p.workQueue.Enqueue(podUID, 0)
	...
	}
	...
}

// syncLoopIteration reads from various channels and dispatches pods to the
// given handler.
//
// Arguments:
// 1.  configCh:       a channel to read config events from
// 2.  handler:        the SyncHandler to dispatch pods to
// 3.  syncCh:         a channel to read periodic sync events from
// 4.  housekeepingCh: a channel to read housekeeping events from
// 5.  plegCh:         a channel to read PLEG updates from
func (kl *Kubelet) syncLoopIteration(ctx context.Context, ...) bool {
	select {
	...
	case <-syncCh:
		// 这里会消费上面的workQueue
		podsToSync := kl.getPodsToSync()
		if len(podsToSync) == 0 {
			break
		}
        / 重复分发处理流程,如果requeueLastPodUpdate已经触发terminated处理,这次分发会提前结束,因为finished=true
		handler.HandlePodSyncs(podsToSync)
	...
}
    
// Get pods which should be resynchronized. Currently, the following pod should be resynchronized:
//   - pod whose work is ready.
//   - internal modules that request sync of a pod.
func (kl *Kubelet) getPodsToSync() []*v1.Pod {
	allPods := kl.podManager.GetPods()
	podUIDs := kl.workQueue.GetWork()
	podUIDSet := sets.NewString()
	for _, podUID := range podUIDs {
		podUIDSet.Insert(string(podUID))
	}
	var podsToSync []*v1.Pod
	for _, pod := range allPods {
        // workQueue记录的Pod会触发dispatch
		if podUIDSet.Has(string(pod.UID)) {
			// The work of the pod is ready
			podsToSync = append(podsToSync, pod)
			continue
		}
		...
	}
	return podsToSync
}

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