prestop优雅退出 | 南风未起｜云原生/Go技术分享博客

# 1.apiserver

# 1.1.入口

del pod由kube-apiserver处理，请求会路由转到store模块删除及设置deleteionTimeStamp和deletionGracePeriodSeconds。

// Delete removes the item from storage.
// options can be mutated by rest.BeforeDelete due to a graceful deletion strategy.
func (e *Store) Delete(...) (runtime.Object, bool, error) {
	...

	// 获取对象	
	e.Storage.Get(ctx, key, storage.GetOptions{}, obj)
    ...

    // 前置检查
	graceful, pendingGraceful, err := rest.BeforeDelete(e.DeleteStrategy, ctx, obj, options)
	...
    
    // 正在删除(terminating)
	if pendingGraceful {
        // 返回正在删除的对象
		out, err := e.finalizeDelete(ctx, obj, false, options)
		return out, false, err
	}
    ...

    // finalizer更新检查(前台删除/后台删除)
	shouldUpdateFinalizers, _ := deletionFinalizersForGarbageCollection(ctx, e, accessor, options)
	// 优雅退出/带finalizer/需更新finalizer
	if graceful || pendingFinalizers || shouldUpdateFinalizers {
        // 再次检查删除条件,设置deletionTimeStamp和指定的deletionGracePeriodSeconds
		err, ignoreNotFound, deleteImmediately, out, lastExisting = e.updateForGracefulDeletionAndFinalizers(ctx, name, key, options, preconditions, deleteValidation, obj)
		...
	}

	// 非立即删除直接返回
	if !deleteImmediately || err != nil {
		return out, false, err
	}

	// dry-run模拟删除及返回模拟对象
	if dryrun.IsDryRun(options.DryRun) && out != nil {
		return out, true, nil
	}
	...
    
	// 真正执行物理删除
	e.Storage.Delete(ctx, key, out, &preconditions, storage.ValidateObjectFunc(deleteValidation), ..., nil)
    ...
    
    // hook收尾及生成返回对象
	out, err = e.finalizeDelete(ctx, out, true, options)
	return out, true, 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

注意

store触发的deletionGracePeriodSeconds设置是用户指定的，默认的30s限制是反序列化阶段做的

# 1.2.默认规则

向apiserver发送请求查询Pod会进行反序列化，对应的反序列化器会填充一些Pod默认值，其中就包括terminationGracePeriodSeconds。

// 若设置过不会触发
func SetDefaults_PodSpec(obj *v1.PodSpec) {
	if obj.DNSPolicy == "" {
		obj.DNSPolicy = v1.DNSClusterFirst
	}
    
	if obj.RestartPolicy == "" {
		obj.RestartPolicy = v1.RestartPolicyAlways
	}
    
	if obj.HostNetwork {
		defaultHostNetworkPorts(&obj.Containers)
		defaultHostNetworkPorts(&obj.InitContainers)
	}
    
	if obj.SecurityContext == nil {
		obj.SecurityContext = &v1.PodSecurityContext{}
	}
    
	if obj.TerminationGracePeriodSeconds == nil {
		period := int64(30)
		obj.TerminationGracePeriodSeconds = &period
	}
    
	if obj.SchedulerName == "" {
		obj.SchedulerName = v1.DefaultSchedulerName
	}
}

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

注意

deletionGracePeriodSeconds不允许续期，仅允许变短，多次删除可以改小优雅退出时间

# 2.kubelet

pre-stop是container级别，killPod流程会并发执行liveness container的pre-stop hook进行相关前置清理。

// 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 {
        // 存活的容器才会触发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)
    ...

	// 优雅时间计算,用户设置的--->spec默认的30s-->启动探针失败设置的-->存活探针失败设置的
	gracePeriod := setTerminationGracePeriod(pod, containerSpec, containerName, containerID, reason)
    
	// pre-stop执行时间算入优雅退出时间
	if containerSpec.Lifecycle != nil && containerSpec.Lifecycle.PreStop != nil && gracePeriod > 0 {
		gracePeriod = gracePeriod - m.executePreStopHook(ctx, pod, containerID, containerSpec, gracePeriod)
	}
    
	// 最小2s
	if gracePeriod < minimumGracePeriodInSeconds {
		gracePeriod = minimumGracePeriodInSeconds
	}
	...
    
    // 终止容器
	m.runtimeService.StopContainer(ctx, 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

注意

pre-stop执行时间算入优雅退出时间，会影响endpoint后端池流量，延迟容器终止可以尽可能处理剩余流量

# 3.endpoint

endpoint controller会监听Pod删除时间，计算service对应的endpoint后端池，后端池会过滤删除状态或未就绪的Pod。

func (e *Controller) syncService(ctx context.Context, key string) error {
	...

    // 获取svc
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	...
	service, err := e.serviceLister.Services(namespace).Get(name)
	...
    
    // externalName service由dns直接转出去,不走endpoint
	if service.Spec.Type == v1.ServiceTypeExternalName {
		// services with Type ExternalName receive no endpoints from this controller
		return nil
	}

    // services without a selector receive no endpoints from this controller
	if service.Spec.Selector == nil {
		return nil
	}

	// service关联的pod
	pods := e.podLister.Pods(service.Namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated())
	...

	// 计算后端池
	for _, pod := range pods {
        // pod条件检查
		if !endpointutil.ShouldPodBeInEndpoints(pod, service.Spec.PublishNotReadyAddresses) {
			continue
		}

        // 基于pod初始化endpoint
		ep, err := podToEndpointAddressForService(service, pod)
		...
		epa := *ep
		...

        // headless类型后端池(供DNS同步查询)
		if len(service.Spec.Ports) == 0 {
			if service.Spec.ClusterIP == api.ClusterIPNone {
				subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, service.Spec.PublishNotReadyAddresses)
			}
        // clusterIP/NodePort类型后端池
		} else {
			for i := range service.Spec.Ports {
				...
				epp := endpointPortFromServicePort(servicePort, portNum)
				subsets, readyEps, notReadyEps = addEndpointSubset(subsets, pod, epa, epp, service.Spec.PublishNotReadyAddresses)
				...
			}
		}
	}
    
	subsets = endpoints.RepackSubsets(subsets)

	// See if there's actually an update here.
	currentEndpoints, err := e.endpointsLister.Endpoints(service.Namespace).Get(service.Name)
	...
    
    // 基于后端池生成新的endpoints对象
    newEndpoints := currentEndpoints.DeepCopy()
	newEndpoints.Subsets = subsets
    ...

    // 后端池最大1000个,避免影响集群流量造成延迟(endpointslice修复了这个问题)
	if truncateEndpoints(newEndpoints) {
		newEndpoints.Annotations[v1.EndpointsOverCapacity] = truncated
	} else {
		delete(newEndpoints.Annotations, v1.EndpointsOverCapacity)
	}
	...

	if createEndpoints {
		// 创建后端池
		e.client.CoreV1().Endpoints(service.Namespace).Create(ctx, newEndpoints, metav1.CreateOptions{})
	} else {
		// 更新后端池
		e.client.CoreV1().Endpoints(service.Namespace).Update(ctx, newEndpoints, metav1.UpdateOptions{})
	}
	...
    
	return nil
}

// returns true if a specified pod should be in an Endpoints or EndpointSlice resource. 
// Terminating pods are only included if includeTerminating is true.
func ShouldPodBeInEndpoints(pod *v1.Pod, includeTerminating bool) bool {
	// success/failed phase pod
	if podutil.IsPodTerminal(pod) {
		return false
	}

    // 未分配CNI的Pod
	if len(pod.Status.PodIP) == 0 && len(pod.Status.PodIPs) == 0 {
		return false
	}

    // 正在删除的Pod
	if !includeTerminating && pod.DeletionTimestamp != nil {
		return false
	}

	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
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

注意

epcontroller监听pod删除事件，过滤删除中的Pod重建后端池subnets，kube-proxy基于后端池重建iptables/ipvs规则

# 4.结论

--- pre-stop的业务价值
	1.优雅退出依赖应用监听sigterm信号尽可能处理流量,应用未监听可能直接退出
	2.kubelet、epcontroller和kube-proxy三个组件独立工作，相互配合依赖apiserver推送的对象
	3.epcontroller和kube-proxy组件相对来说串行，kube-proxy基于epcontroller重建的后端池修改iptables/ipvs规则
	4.kubelet和epcontroer并行工作,无法确定先stopcontainer还是先重建后端池
	5.kubelet先stopcontainer,epcontroller暂未摘除后端池会出现业务流量中断
	6.pre-stop设置一定时间可以延后stopcontainer,给epcontroller和kube-proxy更长时间重建规则及等待容器处理剩余流量

1
2
3
4
5
6
7