kubelet源码 删除pod（二）

kubelet源码 删除pod（二）

本文中含有k8s的一个bug，我也正在努力提交PR，不过会不会被merge就不清楚了。

kubernetes PR地址

pod_workers.go是主要处理pod变化的文件，在1.22版本后对这个文件进行了比较大的修改。把属于pod处理的工作都放在了这个文件里。并且对pod分段处理，如审查、标记状态、处理状态等。

1、options为pod的一些基本信息。runningPod是运行中 的pod，如果runningPod存在，并且pod配置不存在，则代表是孤儿pod，只能进行删除。如果pod和runningPod都存在，则代表都会被更新，所以只保留pod即可。

2.对整个结构体进行加锁，避免污染数据。
根据uid取出当前pod状态，12行中，如果pod是不是个孤儿pod并且状态是失败或者已完成的，则记录他的状态。13行是从本地缓存中获得运行时的pod状态（不是这次要更新的状态）这里14行的函数，判断的是pod下的容器是否运行状态。流程3介绍。

  p.podLock.Lock()
	defer p.podLock.Unlock()

	now := time.Now()
	status, ok := p.podSyncStatuses[uid]
	if !ok {
		klog.V(4).InfoS("Pod is being synced for the first time", "pod", klog.KObj(pod), "podUID", pod.UID)
		status = &podSyncStatus{
			syncedAt: now,
			fullname: kubecontainer.GetPodFullName(pod),
		}
		if !isRuntimePod && (pod.Status.Phase == v1.PodFailed || pod.Status.Phase == v1.PodSucceeded) {
			if statusCache, err := p.podCache.Get(pod.UID); err == nil {
				if isPodStatusCacheTerminal(statusCache) {
					status = &podSyncStatus{
						terminatedAt:       now,
						terminatingAt:      now,
						syncedAt:           now,
						startedTerminating: true,
						finished:           true,
						fullname:           kubecontainer.GetPodFullName(pod),
					}
				}
			}
		}
		p.podSyncStatuses[uid] = status
	}
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

3.判断pod是否已经停止了。遍历容器查看是否有运行中的容器。还要判断“sb"是否有运行中的。如果都为0，则代表pod已经运行完成。然后代码2中的30行，将状态存到pod的状态流的map中

func isPodStatusCacheTerminal(status *kubecontainer.PodStatus) bool {
	runningContainers := 0
	runningSandboxes := 0
	for _, container := range status.ContainerStatuses {
		if container.State == kubecontainer.ContainerStateRunning {
			runningContainers++
		}
	}
	for _, sb := range status.SandboxStatuses {
		if sb.State == runtimeapi.PodSandboxState_SANDBOX_READY {
			runningSandboxes++
		}
	}
	return runningContainers == 0 && runningSandboxes == 0
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

4.第一行判断为该pod是否已经停止了。如果已经停止了，并且这次的更新又是创建，则标记pod状态为重启启动。这个场景一般是静态pod才会出现，通常是具有相同UID。如果是此情况则后续会重新启动。
第9行判断是pod状态是否是已完成了。不归pod_workers管了，后续housekeeping会去清除他。

	if status.IsTerminationRequested() {
		if options.UpdateType == kubetypes.SyncPodCreate {
			status.restartRequested = true
			klog.V(4).InfoS("Pod is terminating but has been requested to restart with same UID, will be reconciled later", "pod", klog.KObj(pod), "podUID", pod.UID)
			return
		}
	}

	if status.IsFinished() {
		klog.V(4).InfoS("Pod is finished processing, no further updates", "pod", klog.KObj(pod), "podUID", pod.UID)
		return
	}
1
2
3
4
5
6
7
8
9
10
11
12

特殊，下面流程的5.6需要这几个函数，在这里标注一下

func (s *podSyncStatus) IsWorking() bool              { return s.working }
func (s *podSyncStatus) IsTerminationRequested() bool { return !s.terminatingAt.IsZero() }
func (s *podSyncStatus) IsTerminationStarted() bool   { return s.startedTerminating }
func (s *podSyncStatus) IsTerminated() bool           { return !s.terminatedAt.IsZero() }
func (s *podSyncStatus) IsFinished() bool             { return s.finished }
func (s *podSyncStatus) IsEvicted() bool              { return s.evicted }
func (s *podSyncStatus) IsDeleted() bool              { return s.deleted }
1
2
3
4
5
6
7

5.上面的一些特殊场景的处理都做完了，到这开始对当前pod进行标记状态了。

• 先给becameTerminating标定为false，这个变量或许会用做是否这个pod杠开始进行删除。
• 第4行case判断是否是孤儿pod，如果是孤儿pod，标记删除（这个标记是apiserver上也删除了），删除初始时间是现在，标记pod刚开始删除
• 第9行case判断的是是否为优雅的删除（DeletionTimestamp就是grace period的时间，如果未传，默认30s)
• 第14行，如果pod的状态是失败或者运行完成，只更新删除初始时间和刚开始删除，但是不会标记apiserver上已删除
• 第18行，如果这个pod是删除。如果还是一个被驱逐的pod，则标记驱逐状态（KillPodOptions为pod的终止行为）

	var becameTerminating bool
	if !status.IsTerminationRequested() {
		switch {
		case isRuntimePod:
			klog.V(4).InfoS("Pod is orphaned and must be torn down", "pod", klog.KObj(pod), "podUID", pod.UID)
			status.deleted = true    //如果为true，代表apiserver上也已经删除
			status.terminatingAt = now   //删除的开始时间
			becameTerminating = true     //刚开始进入删除流程
		case pod.DeletionTimestamp != nil:
			klog.V(4).InfoS("Pod is marked for graceful deletion, begin teardown", "pod", klog.KObj(pod), "podUID", pod.UID)
			status.deleted = true
			status.terminatingAt = now
			becameTerminating = true
		case pod.Status.Phase == v1.PodFailed, pod.Status.Phase == v1.PodSucceeded:
			klog.V(4).InfoS("Pod is in a terminal phase (success/failed), begin teardown", "pod", klog.KObj(pod), "podUID", pod.UID)
			status.terminatingAt = now
			becameTerminating = true
		case options.UpdateType == kubetypes.SyncPodKill:
			if options.KillPodOptions != nil && options.KillPodOptions.Evict {
				klog.V(4).InfoS("Pod is being evicted by the kubelet, begin teardown", "pod", klog.KObj(pod), "podUID", pod.UID)
				status.evicted = true
			} else {
				klog.V(4).InfoS("Pod is being removed by the kubelet, begin teardown", "pod", klog.KObj(pod), "podUID", pod.UID)
			}
			status.terminatingAt = now
			becameTerminating = 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

6.workType标记这个pod声明周期的状态（sync同步、terminating终止中、terminated清理）
wasGracePeriodShortened代表是否缩短优雅删除时间;例，第一次优雅删除（grace period）时间是30s，第二次是10s，则代表缩短）。这个官方存在bug，无法成功缩短优雅删除时间 后面会介绍bug原因

• 第4行case判断是否pod已经删除完成了，第5行判断是否是孤儿pod（可能是旧缓存）直接忽略就可以。第12行验证一下是否是completed状态，如果是的话，关闭这个管道并且置空。
• 第19行，如果pod刚刚开始准备删除，如果是cpmpleted状态，暂存到notifyPostTerminating（后续会停止成功后统一关闭管道）PodStatusFunc记录的是kill pod的状态。然后判断优雅删除的时间（流程7）
• 第39行同理

  var workType PodWorkType 
	var wasGracePeriodShortened bool
	switch {
	case status.IsTerminated():
		if isRuntimePod {
			klog.V(3).InfoS("Pod is waiting for termination, ignoring runtime-only kill until after pod worker is fully terminated", "pod", klog.KObj(pod), "podUID", pod.UID)
			return
		}

		workType = TerminatedPodWork

		if options.KillPodOptions != nil {
			if ch := options.KillPodOptions.CompletedCh; ch != nil {
				close(ch)
			}
		}
		options.KillPodOptions = nil

	case status.IsTerminationRequested():
		workType = TerminatingPodWork
		if options.KillPodOptions == nil {
			options.KillPodOptions = &KillPodOptions{}
		}

		if ch := options.KillPodOptions.CompletedCh; ch != nil {
			status.notifyPostTerminating = append(status.notifyPostTerminating, ch)
		}
		if fn := options.KillPodOptions.PodStatusFunc; fn != nil {
			status.statusPostTerminating = append(status.statusPostTerminating, fn)
		}

		gracePeriod, gracePeriodShortened := calculateEffectiveGracePeriod(status, pod, options.KillPodOptions)

		wasGracePeriodShortened = gracePeriodShortened
		status.gracePeriod = gracePeriod

		options.KillPodOptions.PodTerminationGracePeriodSecondsOverride = &gracePeriod

	default:
		workType = SyncPodWork

		if options.KillPodOptions != nil {
			if ch := options.KillPodOptions.CompletedCh; ch != nil {
				close(ch)
			}
			options.KillPodOptions = 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

7.优雅时间只能缩短，如果不是缩短则不用更改时间。优雅删除时间不能为0

• 第3.4行的判断就是验证新的请求时间是否缩短。
• 9行判断一下pod的删除状态（可能存在驱逐）
• 15行判断优雅时间是否为0，如果为0并且pod的配置文件删除时间不为空，则替换
• 最终的优雅删除时间如果还小于1，则等于1**（这里还存在一个bug，因为强制删除–force的删除时间就是0，如果这里强行判断等于1，那等于强制删除会采用默认的30s，导致apiserver和kubelet不同步）**

func calculateEffectiveGracePeriod(status *podSyncStatus, pod *v1.Pod, options *KillPodOptions) (int64, bool) {
   gracePeriod := status.gracePeriod
   if override := pod.DeletionGracePeriodSeconds; override != nil {
      if gracePeriod == 0 || *override < gracePeriod {
         gracePeriod = *override
      }
   }
   if options != nil {
      if override := options.PodTerminationGracePeriodSecondsOverride; override != nil {
         if gracePeriod == 0 || *override < gracePeriod {
            gracePeriod = *override
         }
      }
   }
   if gracePeriod == 0 && pod.Spec.TerminationGracePeriodSeconds != nil {
      gracePeriod = *pod.Spec.TerminationGracePeriodSeconds
   }
   if gracePeriod < 1 {
      gracePeriod = 1
   }
   return gracePeriod, status.gracePeriod != 0 && status.gracePeriod != gracePeriod
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

8.到此，pod的status状态记录就都完成了，该准备一下pod的期望状态然后进行处理了。

• 6行根据uid获取一下这个pod是否有一个更新的处理管道请求。如果没有则创建，管道长度为1，同时只能更新一个请求。
• 11行如果是静态pod，确保静态pod按照UpdatePod收到的顺序启动
• 16行的无作用，测试使用临时加的。
• 25行，给一个pod初始化成功（能走到这一流程，说明kubelet刚重新启动或pod第一次创建）
  每个pod有一个自己的goroutine专门处理自己的更新，并且同时间只能处理一个

9、完成这个函数的工作。此函数的任务就是对pod进行一些审查，并且对status状态链路进行更新，然后标记一下状态

• 第一行，如果没在工作工作中，把上面配置好的work结构体推入到管。
• 如果pod正在进行更新中，则进入第7行，如果已经存在一次为成功更新的pod请求，覆盖一下时间。并且第13行覆盖上次的更新，因为多次更新时，只采用最新的一次。
• 15行，如果（正在删除中或者优雅删除缩短了）并且pod链路的退出信号已经被注册（这个注册是在managePodLoop函数中，最新的1.27版本已经不存在managePodLoop函数，更名为podWorkerLoop。下一篇会说到）。则直接关闭信号（为了是关闭上一次正在处理的更新）这里和流程6的bug相呼应，这里用context来取消阻塞的goroutine，但是后续并没有监听ctx.Done()导致cancel失败
• 最终一行中则是经过上方多层判断，如果无误，则进行pod的处理

	if !status.IsWorking() {
		status.working = true
		podUpdates <- work
		return
	}

	if undelivered, ok := p.lastUndeliveredWorkUpdate[pod.UID]; ok {
		if !undelivered.Options.StartTime.IsZero() && undelivered.Options.StartTime.Before(work.Options.StartTime) {
			work.Options.StartTime = undelivered.Options.StartTime
		}
	}

	p.lastUndeliveredWorkUpdate[pod.UID] = work

	if (becameTerminating || wasGracePeriodShortened) && status.cancelFn != nil {
		klog.V(3).InfoS("Cancelling current pod sync", "pod", klog.KObj(pod), "podUID", pod.UID, "updateType", work.WorkType)
		status.cancelFn()
		return
	}

	go func() {
			defer runtime.HandleCrash()
			p.managePodLoop(outCh)
		}()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

上一篇 kubelet源码 删除pod（一）
下一篇 kubelet源码 删除pod（三）