kube-controller之podgc

podgc controller负责对pod进行垃圾回收。

  • 如果Terminated状态的数量大于terminatedPodThreshold,删除部分Terminated状态的pod
  • 删除不再服务的node上的pod
  • 删除孤儿pod(一个pod绑定到了一个不存在的Node)
  • 删除Terminating状态且未被调度到某一Node上的pod
1
2
3
4
5
6
7
8
9
10
11
12
// cmd/kube-controller-manager/app/core.go
// 启动入口
func startPodGCController(ctx context.Context, controllerContext ControllerContext) (controller.Interface, bool, error) {
go podgc.NewPodGC(
ctx,
controllerContext.ClientBuilder.ClientOrDie("pod-garbage-collector"),
controllerContext.InformerFactory.Core().V1().Pods(),
controllerContext.InformerFactory.Core().V1().Nodes(),
int(controllerContext.ComponentConfig.PodGCController.TerminatedPodGCThreshold),
).Run(ctx)
return nil, true, 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
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
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
// pkg/controller/podgc/gc_controller.go

const (
// gcCheckPeriod defines frequency of running main controller loop
gcCheckPeriod = 20 * time.Second
// quarantineTime defines how long Orphaned GC waits for nodes to show up
// in an informer before issuing a GET call to check if they are truly gone
// 隔离时间,延迟入队时间,在quarantineTime秒后会再次判断node是否真正的消失
// 可以避免一些node短暂离开的情况
quarantineTime = 40 * time.Second

// field manager used to add pod failure condition and change the pod phase
fieldManager = "PodGC"
)

type PodGCController struct {
kubeClient clientset.Interface

podLister corelisters.PodLister
podListerSynced cache.InformerSynced
nodeLister corelisters.NodeLister
nodeListerSynced cache.InformerSynced

nodeQueue workqueue.DelayingInterface

terminatedPodThreshold int
gcCheckPeriod time.Duration
quarantineTime time.Duration
}

func init() {
// Register prometheus metrics
RegisterMetrics()
}

func NewPodGC(ctx context.Context, kubeClient clientset.Interface, podInformer coreinformers.PodInformer,
nodeInformer coreinformers.NodeInformer, terminatedPodThreshold int) *PodGCController {
return NewPodGCInternal(ctx, kubeClient, podInformer, nodeInformer, terminatedPodThreshold, gcCheckPeriod, quarantineTime)
}

// This function is only intended for integration tests
func NewPodGCInternal(ctx context.Context, kubeClient clientset.Interface, podInformer coreinformers.PodInformer,
nodeInformer coreinformers.NodeInformer, terminatedPodThreshold int, gcCheckPeriod, quarantineTime time.Duration) *PodGCController {
gcc := &PodGCController{
kubeClient: kubeClient,
terminatedPodThreshold: terminatedPodThreshold,
podLister: podInformer.Lister(),
podListerSynced: podInformer.Informer().HasSynced,
nodeLister: nodeInformer.Lister(),
nodeListerSynced: nodeInformer.Informer().HasSynced,
nodeQueue: workqueue.NewNamedDelayingQueue("orphaned_pods_nodes"),
gcCheckPeriod: gcCheckPeriod,
quarantineTime: quarantineTime,
}

return gcc
}

func (gcc *PodGCController) Run(ctx context.Context) {
defer utilruntime.HandleCrash()

klog.Infof("Starting GC controller")
defer gcc.nodeQueue.ShutDown()
defer klog.Infof("Shutting down GC controller")

if !cache.WaitForNamedCacheSync("GC", ctx.Done(), gcc.podListerSynced, gcc.nodeListerSynced) {
return
}

go wait.UntilWithContext(ctx, gcc.gc, gcc.gcCheckPeriod)

<-ctx.Done()
}

func (gcc *PodGCController) gc(ctx context.Context) {
// 返回indexer中的所有pod
pods, err := gcc.podLister.List(labels.Everything())
if err != nil {
klog.Errorf("Error while listing all pods: %v", err)
return
}
nodes, err := gcc.nodeLister.List(labels.Everything())
if err != nil {
klog.Errorf("Error while listing all nodes: %v", err)
return
}
// terminatedPodThreshold 系统中可以存在Terminated状态pod的数量
if gcc.terminatedPodThreshold > 0 {
// 删除部分Terminated状态的pod
gcc.gcTerminated(ctx, pods)
}
// 删除不再服务的node上的pod
if utilfeature.DefaultFeatureGate.Enabled(features.NodeOutOfServiceVolumeDetach) {
gcc.gcTerminating(ctx, pods)
}
// 删除孤儿pod(一个pod绑定到了一个不存在的Node)
gcc.gcOrphaned(ctx, pods, nodes)
// 删除Terminating状态且未被调度到某一Node上的pod
// 什么情况下会产生这种pod?? 将创建没来得及被调度就被删除了
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
}

// isPodTerminating returns true if the pod is terminating.
func isPodTerminating(pod *v1.Pod) bool {
return pod.ObjectMeta.DeletionTimestamp != nil
}

func (gcc *PodGCController) gcTerminating(ctx context.Context, pods []*v1.Pod) {
klog.V(4).Info("GC'ing terminating pods that are on out-of-service nodes")
terminatingPods := []*v1.Pod{}
for _, pod := range pods {
if isPodTerminating(pod) {
// 获取pod所在的node
node, err := gcc.nodeLister.Get(pod.Spec.NodeName)
if err != nil {
klog.Errorf("failed to get node %s : %s", pod.Spec.NodeName, err)
continue
}
// Add this pod to terminatingPods list only if the following conditions are met:
// 1. Node is not ready.
// 2. Node has `node.kubernetes.io/out-of-service` taint.
// 判断Node是否是Ready,Node是否包含node.kubernetes.io/out-of-service
// 同时满足这两个条件,则加入terminatingPods列表
if !nodeutil.IsNodeReady(node) && taints.TaintKeyExists(node.Spec.Taints, v1.TaintNodeOutOfService) {
klog.V(4).Infof("garbage collecting pod %s that is terminating. Phase [%v]", pod.Name, pod.Status.Phase)
terminatingPods = append(terminatingPods, pod)
}
}
}

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

klog.V(4).Infof("Garbage collecting %v pods that are terminating on node tainted with node.kubernetes.io/out-of-service", deleteCount)
// sort only when necessary
sort.Sort(byEvictionAndCreationTimestamp(terminatingPods))
var wait sync.WaitGroup
// 删除terminatingPods
for i := 0; i < deleteCount; i++ {
wait.Add(1)
go func(pod *v1.Pod) {
defer wait.Done()
deletingPodsTotal.WithLabelValues().Inc()
if err := gcc.markFailedAndDeletePod(ctx, pod); err != nil {
// ignore not founds
utilruntime.HandleError(err)
deletingPodsErrorTotal.WithLabelValues().Inc()
}
}(terminatingPods[i])
}
wait.Wait()
}

func (gcc *PodGCController) gcTerminated(ctx context.Context, pods []*v1.Pod) {
terminatedPods := []*v1.Pod{}
for _, pod := range pods {
if isPodTerminated(pod) {
terminatedPods = append(terminatedPods, pod)
}
}

terminatedPodCount := len(terminatedPods)
deleteCount := terminatedPodCount - gcc.terminatedPodThreshold

if deleteCount <= 0 {
return
}

klog.InfoS("Garbage collecting pods", "numPods", deleteCount)
// sort only when necessary
sort.Sort(byEvictionAndCreationTimestamp(terminatedPods))
var wait sync.WaitGroup
for i := 0; i < deleteCount; i++ {
wait.Add(1)
go func(pod *v1.Pod) {
defer wait.Done()
if err := gcc.markFailedAndDeletePod(ctx, pod); err != nil {
// ignore not founds
defer utilruntime.HandleError(err)
}
}(terminatedPods[i])
}
wait.Wait()
}

// gcOrphaned deletes pods that are bound to nodes that don't exist.
func (gcc *PodGCController) gcOrphaned(ctx context.Context, pods []*v1.Pod, nodes []*v1.Node) {
klog.V(4).Infof("GC'ing orphaned")
existingNodeNames := sets.NewString()
// 获取存在的NodeName
for _, node := range nodes {
existingNodeNames.Insert(node.Name)
}
// Add newly found unknown nodes to quarantine
for _, pod := range pods {
// pod的nodeName不为空且nodeName不存在于existingNodeNames
// 将NodeName入队,延迟时间为quarantineTime
if pod.Spec.NodeName != "" && !existingNodeNames.Has(pod.Spec.NodeName) {
gcc.nodeQueue.AddAfter(pod.Spec.NodeName, gcc.quarantineTime)
}
}
// Check if nodes are still missing after quarantine period
// 在隔离期时间过后检查node是否存在
deletedNodesNames, quit := gcc.discoverDeletedNodes(ctx, existingNodeNames)
if quit {
return
}
// 删除孤儿Pod
for _, pod := range pods {
// pod.Spec.NodeName 不在已删除的node集合,说明不是孤儿Pod
if !deletedNodesNames.Has(pod.Spec.NodeName) {
continue
}
klog.V(2).InfoS("Found orphaned Pod assigned to the Node, deleting.", "pod", klog.KObj(pod), "node", pod.Spec.NodeName)
condition := corev1apply.PodCondition().
WithType(v1.DisruptionTarget).
WithStatus(v1.ConditionTrue).
WithReason("DeletionByPodGC").
WithMessage("PodGC: node no longer exists").
WithLastTransitionTime(metav1.Now())
if err := gcc.markFailedAndDeletePodWithCondition(ctx, pod, condition); err != nil {
utilruntime.HandleError(err)
} else {
klog.InfoS("Forced deletion of orphaned Pod succeeded", "pod", klog.KObj(pod))
}
}
}

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
}
nodeName := item.(string)
if !existingNodeNames.Has(nodeName) {
exists, err := gcc.checkIfNodeExists(ctx, nodeName)
switch {
case err != nil:
klog.ErrorS(err, "Error while getting node", "node", klog.KRef("", nodeName))
// Node will be added back to the queue in the subsequent loop if still needed
case !exists:
deletedNodesNames.Insert(nodeName)
}
}
gcc.nodeQueue.Done(item)
}
return deletedNodesNames, false
}

func (gcc *PodGCController) checkIfNodeExists(ctx context.Context, name string) (bool, error) {
_, fetchErr := gcc.kubeClient.CoreV1().Nodes().Get(ctx, name, metav1.GetOptions{})
if errors.IsNotFound(fetchErr) {
return false, nil
}
return fetchErr == nil, fetchErr
}

// gcUnscheduledTerminating deletes pods that are terminating and haven't been scheduled to a particular node.
func (gcc *PodGCController) gcUnscheduledTerminating(ctx context.Context, pods []*v1.Pod) {
klog.V(4).Infof("GC'ing unscheduled pods which are terminating.")

for _, pod := range pods {
// 未被删除或已经调度到node
if pod.DeletionTimestamp == nil || len(pod.Spec.NodeName) > 0 {
continue
}

klog.V(2).InfoS("Found unscheduled terminating Pod not assigned to any Node, deleting.", "pod", klog.KObj(pod))
if err := gcc.markFailedAndDeletePod(ctx, pod); err != nil {
utilruntime.HandleError(err)
} else {
klog.InfoS("Forced deletion of unscheduled terminating Pod succeeded", "pod", klog.KObj(pod))
}
}
}

// byEvictionAndCreationTimestamp sorts a list by Evicted status and then creation timestamp,
// using their names as a tie breaker.
// Evicted pods will be deleted first to avoid impact on terminated pods created by controllers.
type byEvictionAndCreationTimestamp []*v1.Pod

func (o byEvictionAndCreationTimestamp) Len() int { return len(o) }
func (o byEvictionAndCreationTimestamp) Swap(i, j int) { o[i], o[j] = o[j], o[i] }

func (o byEvictionAndCreationTimestamp) Less(i, j int) bool {
iEvicted, jEvicted := eviction.PodIsEvicted(o[i].Status), eviction.PodIsEvicted(o[j].Status)
// Evicted pod is smaller
if iEvicted != jEvicted {
return iEvicted
}
if o[i].CreationTimestamp.Equal(&o[j].CreationTimestamp) {
return o[i].Name < o[j].Name
}
return o[i].CreationTimestamp.Before(&o[j].CreationTimestamp)
}

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 {
klog.InfoS("PodGC is force deleting Pod", "pod", klog.KRef(pod.Namespace, pod.Name))
if utilfeature.DefaultFeatureGate.Enabled(features.PodDisruptionConditions) {

// Mark the pod as failed - this is especially important in case the pod
// is orphaned, in which case the pod would remain in the Running phase
// forever as there is no kubelet running to change the phase.
if pod.Status.Phase != v1.PodSucceeded && pod.Status.Phase != v1.PodFailed {
podApply := corev1apply.Pod(pod.Name, pod.Namespace).WithStatus(corev1apply.PodStatus())
// we don't need to extract the pod apply configuration and can send
// only phase and the DisruptionTarget condition as PodGC would not
// own other fields. If the DisruptionTarget condition is owned by
// PodGC it means that it is in the Failed phase, so sending the
// condition will not be re-attempted.
podApply.Status.WithPhase(v1.PodFailed)
if condition != nil {
podApply.Status.WithConditions(condition)
}
if _, err := gcc.kubeClient.CoreV1().Pods(pod.Namespace).ApplyStatus(ctx, podApply, metav1.ApplyOptions{FieldManager: fieldManager, Force: true}); err != nil {
return err
}
}
}
return gcc.kubeClient.CoreV1().Pods(pod.Namespace).Delete(ctx, pod.Name, *metav1.NewDeleteOptions(0))
}

REF:
1.cmd/kube-controller-manager/app/core.go
2.pkg/controller/podgc/gc_controller.go