The Garbage Collection of Pods

// pkg/controller/podgc/gc_controller.go

// Pod GC controller 最终使用的方法
func (gcc *PodGCController) gc(ctx context.Context) {
	// 列举出当前集群中所有 pod 和 node 的资源
	pods, err := gcc.podLister.List(labels.Everything())
	nodes, err := gcc.nodeLister.List(labels.Everything())

	if gcc.terminatedPodThreshold > 0 { // 该阈值小于等于0，说明不启用 Pod GC，只进行一些其他的回收工作
		gcc.gcTerminated(ctx, pods)
	}
	if utilfeature.DefaultFeatureGate.Enabled(features.NodeOutOfServiceVolumeDetach) {
		gcc.gcTerminating(ctx, pods)
	}
	gcc.gcOrphaned(ctx, pods, nodes)
	gcc.gcUnscheduledTerminating(ctx, pods)
}

func isPodTerminated(pod *v1.Pod) bool {
	if phase := pod.Status.Phase; phase != v1.PodPending && phase != v1.PodRunning && phase != v1.PodUnknown {
		return true
	}
	return false
}

func (gcc *PodGCController) gcTerminated(ctx context.Context, pods []*v1.Pod) {
	terminatedPods := []*v1.Pod{}
	for _, pod := range pods {
		if isPodTerminated(pod) {
			terminatedPods = append(terminatedPods, pod)  // 收集所有处于已终止状态的 pods
		}
	}

	terminatedPodCount := len(terminatedPods)
	deleteCount := terminatedPodCount - gcc.terminatedPodThreshold
	if deleteCount <= 0 {  // 不及 pod 回收的阈值时，就终止此次回收
		return
	}

	sort.Sort(byEvictionAndCreationTimestamp(terminatedPods))  // 按驱逐状态和 pod 创建时间戳排序
	var wait sync.WaitGroup
	for i := 0; i < deleteCount; i++ {
		wait.Add(1)
		go func(pod *v1.Pod) {
			defer wait.Done()
			gcc.markFailedAndDeletePod(ctx, pod)  // 执行删除
		}(terminatedPods[i])
	}
	wait.Wait()
}

func (gcc *PodGCController) gcOrphaned(ctx context.Context, pods []*v1.Pod, nodes []*v1.Node) {
	existingNodeNames := sets.NewString()
	for _, node := range nodes {
		existingNodeNames.Insert(node.Name)
	}
	// 将新找到的、未知的 node 进行隔离
	for _, pod := range pods {
		if pod.Spec.NodeName != "" && !existingNodeNames.Has(pod.Spec.NodeName) {
			gcc.nodeQueue.AddAfter(pod.Spec.NodeName, gcc.quarantineTime) // 在经过 quarantineTime 的隔离期之后再加入 node 队列
		}
	}
	// 检查 node 在隔离期之后是否还属于未知状态
	deletedNodesNames, quit := gcc.discoverDeletedNodes(ctx, existingNodeNames)
	if quit {
		return
	}

	for _, pod := range pods {
		if !deletedNodesNames.Has(pod.Spec.NodeName) {  // 将不属于任何 node 的 pod 删除
			continue
		}
		condition := corev1apply.PodCondition().
			WithType(v1.DisruptionTarget).
			WithStatus(v1.ConditionTrue).
			WithReason("DeletionByPodGC").
			WithMessage("PodGC: node no longer exists").
			WithLastTransitionTime(metav1.Now())
		gcc.markFailedAndDeletePodWithCondition(ctx, pod, condition)  // 执行删除
	}
}

func (gcc *PodGCController) discoverDeletedNodes(ctx context.Context, existingNodeNames sets.String) (sets.String, bool) {
	deletedNodesNames := sets.NewString()
	for gcc.nodeQueue.Len() > 0 {
		item, quit := gcc.nodeQueue.Get()
		if quit {
			return nil, true  // quit
		}
		nodeName := item.(string)
		if !existingNodeNames.Has(nodeName) {  // 仍然属于未知的 node 的话
			exists, err := gcc.checkIfNodeExists(ctx, nodeName) // 通过 kube-client 检查对应 node 是否真实存在
			switch {
			case err != nil:
				// ...
			case !exists:
				// 对于不存在的 node，加入到删除名单中
				deletedNodesNames.Insert(nodeName)
			}
		}
		gcc.nodeQueue.Done(item)
	}
	return deletedNodesNames, false
}

func (gcc *PodGCController) gcUnscheduledTerminating(ctx context.Context, pods []*v1.Pod) {
	for _, pod := range pods {
		if pod.DeletionTimestamp == nil || len(pod.Spec.NodeName) > 0 {
			continue
		}
		gcc.markFailedAndDeletePod(ctx, pod)  // 执行删除
	}
}

func (gcc *PodGCController) gcTerminating(ctx context.Context, pods []*v1.Pod) {
	terminatingPods := []*v1.Pod{}
	for _, pod := range pods {
		if isPodTerminating(pod) {  // => pod.ObjectMeta.DeletionTimestamp != nil
			node, err := gcc.nodeLister.Get(pod.Spec.NodeName)

			// 同时满足下列两个条件时，pod 才会被加入到 terminatingPods 列表中：
			// 1. Node 没有 ready
			// 2. 但是 Node 有 `node.kubernetes.io/out-of-service` 污点
			if !nodeutil.IsNodeReady(node) && taints.TaintKeyExists(node.Spec.Taints, v1.TaintNodeOutOfService) {
				terminatingPods = append(terminatingPods, pod)
			}
		}
	}

	deleteCount := len(terminatingPods)
	if deleteCount == 0 {
		return
	}

	sort.Sort(byEvictionAndCreationTimestamp(terminatingPods))  // 按驱逐状态和 pod 创建时间戳排序
	var wait sync.WaitGroup
	for i := 0; i < deleteCount; i++ {
		wait.Add(1)
		go func(pod *v1.Pod) {
			defer wait.Done()
			gcc.markFailedAndDeletePod(ctx, pod)  // 执行删除
		}(terminatingPods[i])
	}
	wait.Wait()
}

func (gcc *PodGCController) markFailedAndDeletePod(ctx context.Context, pod *v1.Pod) error {
	return gcc.markFailedAndDeletePodWithCondition(ctx, pod, nil)
}

func (gcc *PodGCController) markFailedAndDeletePodWithCondition(ctx context.Context, pod *v1.Pod, condition *corev1apply.PodConditionApplyConfiguration) error {
	if utilfeature.DefaultFeatureGate.Enabled(features.PodDisruptionConditions) {
		// 对于处于运行中阶段的 Pod，进行清楚原因的设置
		if pod.Status.Phase != v1.PodSucceeded && pod.Status.Phase != v1.PodFailed {
			podApply := corev1apply.Pod(pod.Name, pod.Namespace).WithStatus(corev1apply.PodStatus())
			podApply.Status.WithPhase(v1.PodFailed)
			// 只有在 gcOrphaned 调用下该 condition 才不为 nil，传入的 condition 就是 `DelectionByPodGC`
			if condition != nil {
				podApply.Status.WithConditions(condition)
			}
			gcc.kubeClient.CoreV1().Pods(pod.Namespace).ApplyStatus(ctx, podApply, metav1.ApplyOptions{FieldManager: fieldManager, Force: true})  // => fieldManager := "PodGC"
		}
	}
	return gcc.kubeClient.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, *metav1.NewDeleteOptions(0))
}