endpointSlice
南风未起 2025-02-25 19:39:22 kcm
# 1.简介
# 1.1.endpoints
service会关联一个同名endpoints，匹配的pod信息均会记录到endpoints subsets。对于大规模集群来说，endpoints记录的后端池会非常庞大，某个pod变动就需要将endpoints完整分发到各节点的kubeproxy，造成网络性能压力。出于减轻服务跟踪压力的目的，endpoints会限制后端池数量(1000)，带来后端池不完整问题。
注意
endpoints对象可能占用大量api资源，限制service的网络端点数量，难以应对大规模集群场景。
# 1.2.endpoint slice
endpoint slice基于类似分片的方法解决endpoints伸缩性及跟踪问题，利用多个较小的endpoint slice跟踪服务的pod，某个pod变化仅会分发所在的endpoint slice，用于端点更新的数据将大大减少，改善kubeproxy的资源跟踪速度。
注意
1.
endpoint slice支持将ipv4和ipv6地址用于服务，基于endpoint slice的addressType字段按照IP跟踪地址2.
endpoint slice支持拓扑感知路由，以提示kubeproxy优先路由同一zone或region内的流量3.
endpoint slice会出现端点重复情况，监听使用必须处理端点出现在多个切片状况，进行端点去重
# 1.3.端点状态
endpoint slice会存储相关端点的状态，分别为ready/serving/terminating，以应对externalTrafficPolicy=local场景负载均衡器尚未探测健康检查，将流量路由到节点仅有终止pod情况，造成流量丢失。其中，serving条件的语义兼容终止pod提供服务情况，terminating条件标识终止pod，补充的条件可以使外部理解以前不可能的pod状态，以协同连接耗尽事件，继续为现有连接转发流量，新连接路由到其它非终止端点。
# 2.epscontroller
# 2.1.newController
newController()用于初始化eps调度器，监听service/pod/endpointSlice资源的变更事件及入队关联service，以触发eps更新，同时监听node资源变更事件，以维护eps拓扑，影响高负载的节点流量再平衡。// Controller manages selector-based service endpoint slices type Controller struct { client clientset.Interface ... // service监听 serviceLister corelisters.ServiceLister ... // pod监听 podLister corelisters.PodLister ... // endpoint slice监听 endpointSliceLister discoverylisters.EndpointSliceLister ... // node监听 nodeLister corelisters.NodeLister ... // reconciler is an util used to reconcile EndpointSlice changes. reconciler *reconciler // 暂存变化的service queue workqueue.RateLimitingInterface // endpoint slice容量 maxEndpointsPerSlice int32 // worker执行周期 workerLoopPeriod time.Duration ... // 缓存service标签选择器 serviceSelectorCache *endpointutil.ServiceSelectorCache // 拓扑感知提示,跟踪节点和端点在不同可用区的分布情况 topologyCache *topologycache.TopologyCache } // NewController creates and initializes a new Controller func NewController(...) *Controller { ... c := &Controller{ client: client, queue: workqueue.NewNamedRateLimitingQueue(workqueue.NewMaxOfRateLimiter( // 重试延迟[1s,100s] workqueue.NewItemExponentialFailureRateLimiter(defaultSyncBackOff, maxSyncBackOff), // 桶形限流器,10qps,100令牌 &workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)}, ), "endpoint_slice"), workerLoopPeriod: time.Second, } // service监听 serviceInformer.Informer().AddEventHandler(...) c.serviceLister = serviceInformer.Lister() c.servicesSynced = serviceInformer.Informer().HasSynced // pod监听 podInformer.Informer().AddEventHandler(...) c.podLister = podInformer.Lister() c.podsSynced = podInformer.Informer().HasSynced // node监听 c.nodeLister = nodeInformer.Lister() c.nodesSynced = nodeInformer.Informer().HasSynced // endpoint slice监听 endpointSliceInformer.Informer().AddEventHandler(...) c.endpointSliceLister = endpointSliceInformer.Lister() c.endpointSlicesSynced = endpointSliceInformer.Informer().HasSynced ... ... // 标签选择器缓存,避免频繁分配 c.serviceSelectorCache = endpointutil.NewServiceSelectorCache() // 拓扑提示 if utilfeature.DefaultFeatureGate.Enabled(features.TopologyAwareHints) { nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ // 节点新增检查拓扑分布 AddFunc: c.addNode, // nodeReady变化检查拓扑分布 UpdateFunc: c.updateNode, // 节点删除检查拓扑分布 DeleteFunc: c.deleteNode, }) c.topologyCache = topologycache.NewTopologyCache() } // 执行endpoint slice的调谐器 c.reconciler = &reconciler{ client: c.client, nodeLister: c.nodeLister, // default 100 maxEndpointsPerSlice: c.maxEndpointsPerSlice, ... topologyCache: c.topologyCache, } return c }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
注意
1.
zone和nodeCPU不存在时，service均视为overload入队调度，重置负载均衡比例，回退到原始的流量负载均衡策略2.
topology本质是流量的就近调度，A/B节点位于相同的zone，拓扑调度的结果就是A/B节点访问service的后端仅保留A/B节点的pod
# 2.2.startEndpointSliceController
startEndpointSliceController()会初始化eps controller对象，后台启动eps controller，以同步eps后端池及维护端点状态。func startEndpointSliceController(...) (controller.Interface, bool, error) { // 初始化及启动 go eps.NewController(...).Run(...) return nil, true, nil } // Run will not return until stopCh is closed. func (c *Controller) Run(workers int, stopCh <-chan struct{}) { ... // 同步完成 if !cache.WaitForNamedCacheSync("endpoint_slice", stopCh, c.podsSynced, c.servicesSynced, c.endpointSlicesSynced, c.nodesSynced) { return } // 初始化5个worker处理调度 for i := 0; i < workers; i++ { go wait.Until(c.worker, c.workerLoopPeriod, stopCh) } <-stopCh } // worker runs a worker thread that just dequeues items, processes them, and marks them done. func (c *Controller) worker() { for c.processNextWorkItem() { } } func (c *Controller) processNextWorkItem() bool { cKey, quit := c.queue.Get() if quit { return false } defer c.queue.Done(cKey) err := c.syncService(cKey.(string)) c.handleErr(err, cKey) 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
注意
核心逻辑入口是
syncService，内部最终依赖reconciler模块
# 2.3.syncService
syncService()会获取service/pods/eps，过滤标记为删除的eps，最终依赖reconciler.reconcile()执行真正的eps同步func (c *Controller) syncService(key string) error { ... // 获取service service, err := c.serviceLister.Services(namespace).Get(name) if apierrors.IsNotFound(err) { ... c.reconciler.deleteService(namespace, name) ... return nil } ... // 获取匹配pod pods, err := c.podLister.Pods(service.Namespace).List(podLabelSelector) ... // 获取eps endpointSlices, err := c.endpointSliceLister.EndpointSlices(service.Namespace).List(esLabelSelector) ... // 过滤标记删除的eps endpointSlices = dropEndpointSlicesPendingDeletion(endpointSlices) ... // eps同步处理 err = c.reconciler.reconcile(service, pods, endpointSlices, lastChangeTriggerTime) ... return nil } // reconcile takes a set of pods currently matching a service selector and // compares them with the endpoints already present in any existing endpoint // slices for the given service. func (r *reconciler) reconcile(...) error { ... // svc支持的addr type serviceSupportedAddressesTypes := getAddressTypesForService(service) // 根据svc支持的地址类型对eps分类 for _, existingSlice := range existingSlices { // eps地址类型不支持 if _, ok := serviceSupportedAddressesTypes[existingSlice.AddressType]; !ok { // 清理topologyCache缓存的eps if r.topologyCache != nil { svcKey, err := serviceControllerKey(existingSlice) ... r.topologyCache.RemoveHints(svcKey, existingSlice.AddressType) } // 记录到待删除列表 slicesToDelete = append(slicesToDelete, existingSlice) continue } //根据地址类型记录eps if _, ok := slicesByAddressType[existingSlice.AddressType]; !ok { slicesByAddressType[existingSlice.AddressType] = make([]*discovery.EndpointSlice, 0, 1) } slicesByAddressType[existingSlice.AddressType] = append(slicesByAddressType[existingSlice.AddressType], existingSlice) } // eps同步 for addressType := range serviceSupportedAddressesTypes { existingSlices := slicesByAddressType[addressType] r.reconcileByAddressType(service, pods, existingSlices, triggerTime, addressType) ... } // 清理地址类型不支持的eps for _, sliceToDelete := range slicesToDelete { r.client.DiscoveryV1().EndpointSlices(service.Namespace).Delete(context.TODO(), sliceToDelete.Name, metav1.DeleteOptions{}) ... } return utilerrors.NewAggregate(errs) } // reconcileByAddressType takes a set of pods currently matching a service selector and // compares them with the endpoints already present in any existing endpoint // slices for the given service. func (r *reconciler) reconcileByAddressType(...) error { ... // 遍历已存在的eps for _, existingSlice := range existingSlices { // 根据端口组合hash记录eps if ownedBy(existingSlice, service) { epHash := endpointutil.NewPortMapKey(existingSlice.Ports) existingSlicesByPortMap[epHash] = append(existingSlicesByPortMap[epHash], existingSlice) numExistingEndpoints += len(existingSlice.Endpoints) // 非svc子对象,记录到待回收列表 } else { slicesToDelete = append(slicesToDelete, existingSlice) } } ... // 遍历pod for _, pod := range pods { ... // 终态/无IP/终止中 if !endpointutil.ShouldPodBeInEndpoints(pod, includeTerminating) { continue } // 基于pod构造epp endpointPorts := getEndpointPorts(service, pod) // 当前epp组合hash epHash := endpointutil.NewPortMapKey(endpointPorts) ... // 端口组合对应元数据 if _, ok := desiredMetaByPortMap[epHash]; !ok { desiredMetaByPortMap[epHash] = &endpointMeta{ AddressType: addressType, Ports: endpointPorts, } } // 获取节点 node, err := r.nodeLister.Get(pod.Spec.NodeName) ... // 基于pod/node/svc构造ep endpoint := podToEndpoint(pod, node, service, addressType) // 端口组合对应端点 if len(endpoint.Addresses) > 0 { desiredEndpointsByPortMap[epHash].Insert(&endpoint) } } ... // 遍历端口组合对应端点 for portMap, desiredEndpoints := range desiredEndpointsByPortMap { ... // 计算创建/更新/删除的eps pmSlicesToCreate, pmSlicesToUpdate, pmSlicesToDelete, added, removed := r.reconcileByPortMapping( service, existingSlicesByPortMap[portMap], desiredEndpoints, desiredMetaByPortMap[portMap]) ... slicesToCreate = append(slicesToCreate, pmSlicesToCreate...) slicesToUpdate = append(slicesToUpdate, pmSlicesToUpdate...) slicesToDelete = append(slicesToDelete, pmSlicesToDelete...) } // 记录过期的eps至待删除列表 for portMap, existingSlices := range existingSlicesByPortMap { if _, ok := desiredEndpointsByPortMap[portMap]; !ok { for _, existingSlice := range existingSlices { slicesToDelete = append(slicesToDelete, existingSlice) } } } ... // 拓扑开启,计算流量分布策略,eps注入hints信息 if r.topologyCache != nil && hintsEnabled(service.Annotations) { slicesToCreate, slicesToUpdate = r.topologyCache.AddHints(si) } else { // 清理topologyCache缓存 if r.topologyCache != nil { r.topologyCache.RemoveHints(si.ServiceKey, addressType) } // eps去除hints信息 slicesToCreate, slicesToUpdate = topologycache.RemoveHintsFromSlices(si) } // 调用client更新资源 r.finalize(service, slicesToCreate, slicesToUpdate, slicesToDelete, triggerTime) ... return utilerrors.NewAggregate(errs) }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
注意
1.
eps会经过多次复用，待创建eps会尽可能复用待删除eps2.
eps hints分布经过多次分配，根据预期值平衡，zone内的endpoint数量最大为desired数量