透过真实场景分析K8S的EndpointController的源码

场景重现
最近遇到一个问题，在K8S的几台机器上中创建了Glusterfs的集群，通过官方的教程一步步的来利用Glusterfs创建Volume以及PV，不过只是创建了每个Volume的Endpoint，并没有相对应的创建Service实例（官方说创建Service会使Endpoint持久化，当时并没有理会），然后在一次集群重启的时候发现Endpoint实例并没有启动起来，很疑惑，像其他的K8S对象，例如POD，Deployment，Service都启动起来了，但是Endpoint并没有，带着这个问题看了下官方的Issue，并没有什么有效的解答，大家可以参考一下Issue: Endpoints are not persistented

1. 探究源码
   1.1 源码版本

基于k8s release-1.13
1.2 源码目录结构
由于我们重点看Endpoint部分，因此我们只看Endpoint相关的源码
Endpoint
1.3 Endpoint的初始化
文件位置： endpoints_controller.go

// NewEndpointController returns a new *EndpointController.
//我们可以看到在Endpoint初始化的时候，已经注册了三个informer，分别是podInformer，serviceInformer，endpointsInformer
func NewEndpointController(podInformer coreinformers.PodInformer, serviceInformer coreinformers.ServiceInformer,

endpointsInformer coreinformers.EndpointsInformer, client clientset.Interface) *EndpointController { broadcaster := record.NewBroadcaster() broadcaster.StartLogging(klog.Infof) broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")}) recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "endpoint-controller"}) if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil { metrics.RegisterMetricAndTrackRateLimiterUsage("endpoint_controller", client.CoreV1().RESTClient().GetRateLimiter()) } e := &EndpointController{ client: client, queue: workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "endpoint"), workerLoopPeriod: time.Second, } //这里对service进行watch操作，并注册了对应的add\update\del等操作 serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{

//add：以添加的service的namespace/name形式为key，并将该key加入 queue

 AddFunc: e.enqueueService,

//update：以更新后的service的namespace/name形式为key，并将该key加入 queue

 UpdateFunc: func(old, cur interface{}) { e.enqueueService(cur) },

//delete：以删除的service的namespace/name形式为key，并将该key加入 queue

 DeleteFunc: e.enqueueService, }) e.serviceLister = serviceInformer.Lister() e.servicesSynced = serviceInformer.Informer().HasSynced //这里对pod进行watch操作，并注册了对应的add\update\del等操作 podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: e.addPod, UpdateFunc: e.updatePod, DeleteFunc: e.deletePod, }) e.podLister = podInformer.Lister() e.podsSynced = podInformer.Informer().HasSynced e.endpointsLister = endpointsInformer.Lister() e.endpointsSynced = endpointsInformer.Informer().HasSynced e.triggerTimeTracker = NewTriggerTimeTracker() e.eventBroadcaster = broadcaster e.eventRecorder = recorder return e

}
我们看看pod注册的Handler引用了哪些函数

1.3.1 e.addPod
func (e *EndpointController) addPod(obj interface{}) {
//实例化一个pod对象

pod := obj.(*v1.Pod) services, err := e.getPodServiceMemberships(pod) if err != nil { utilruntime.HandleError(fmt.Errorf("Unable to get pod %s/%s's service memberships: %v", pod.Namespace, pod.Name, err)) return }

//将service集合以namespace/name为key逐个加入到queue中

for key := range services { e.queue.Add(key) }

}

func (e EndpointController) getPodServiceMemberships(pod v1.Pod) (sets.String, error) {

set := sets.String{}

//获取pod与service的映射关系

services, err := e.serviceLister.GetPodServices(pod) if err != nil { // don't log this error because this function makes pointless // errors when no services match. return set, nil }

//查找逻辑为逐个对比service的selector与该pod的label，如果service的selector为该pod label的子集，则表示该pod属于service

for i := range services { key, err := controller.KeyFunc(services[i]) if err != nil { return nil, err } set.Insert(key) } return set, nil

}
1.3.2 e.updatePod
func (e *EndpointController) updatePod(old, cur interface{}) {

newPod := cur.(*v1.Pod) oldPod := old.(*v1.Pod)

//比较两者的ResourceVersion，对比更新后的pod与原pod，如果两者的资源版本相等，则直接返回，不进行入队操作

if newPod.ResourceVersion == oldPod.ResourceVersion { // Periodic resync will send update events for all known pods. // Two different versions of the same pod will always have different RVs. return }

//判断pod相关信息是否发生改变

podChangedFlag := podChanged(oldPod, newPod) // Check if the pod labels have changed, indicating a possible // change in the service membership labelsChanged := false

//判断两者的label是否已经不一致，或者hostname或subdomain已改变

if !reflect.DeepEqual(newPod.Labels, oldPod.Labels) || !hostNameAndDomainAreEqual(newPod, oldPod) { labelsChanged = true } // If both the pod and labels are unchanged, no update is needed if !podChangedFlag && !labelsChanged { return }

//判断错误，则获取对应的service和pod映射关系

services, err := e.getPodServiceMemberships(newPod) if err != nil { utilruntime.HandleError(fmt.Errorf("Unable to get pod %v/%v's service memberships: %v", newPod.Namespace, newPod.Name, err)) return } if labelsChanged { oldServices, err := e.getPodServiceMemberships(oldPod) if err != nil { utilruntime.HandleError(fmt.Errorf("Unable to get pod %v/%v's service memberships: %v", oldPod.Namespace, oldPod.Name, err)) return } services = determineNeededServiceUpdates(oldServices, services, podChangedFlag) } for key := range services { e.queue.Add(key) }

}

func podChanged(oldPod, newPod *v1.Pod) bool {
//podChanged函数，其检测逻辑为，如果新旧两个pod的DeletionTimestamp字段不等则返回true，否则继续判断两者的就绪状态，如果不等则返回true，最后再判断新旧pod的ip、nodename、namespace、UID是否相等，如果相等则返回false，否则返回true。将返回结果赋值给podChangedFlag

// If the pod's deletion timestamp is set, remove endpoint from ready address. if newPod.DeletionTimestamp != oldPod.DeletionTimestamp { return true } // If the pod's readiness has changed, the associated endpoint address // will move from the unready endpoints set to the ready endpoints. // So for the purposes of an endpoint, a readiness change on a pod // means we have a changed pod. if podutil.IsPodReady(oldPod) != podutil.IsPodReady(newPod) { return true } // Convert the pod to an EndpointAddress, clear inert fields, // and see if they are the same. newEndpointAddress := podToEndpointAddress(newPod) oldEndpointAddress := podToEndpointAddress(oldPod) // Ignore the ResourceVersion because it changes // with every pod update. This allows the comparison to // show equality if all other relevant fields match. newEndpointAddress.TargetRef.ResourceVersion = "" oldEndpointAddress.TargetRef.ResourceVersion = "" if reflect.DeepEqual(newEndpointAddress, oldEndpointAddress) { // The pod has not changed in any way that impacts the endpoints return false } return true

}
1.4 Endpoint-Controller具体逻辑
// Run will not return until stopCh is closed. workers determines how many
// endpoints will be handled in parallel.
func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) {

defer utilruntime.HandleCrash() defer e.queue.ShutDown() klog.Infof("Starting endpoint controller") defer klog.Infof("Shutting down endpoint controller")

// 等待pod、service、endpoint列表同步

if !controller.WaitForCacheSync("endpoint", stopCh, e.podsSynced, e.servicesSynced, e.endpointsSynced) { return } // 这里workers数为kube-controller-manager启动参数中的--concurrent-endpoint-syncs决定，默认为5，workerLoopPeriod为1秒 for i := 0; i < workers; i++ {

// 执行worker函数，for死循环处理queue中的key

 go wait.Until(e.worker, e.workerLoopPeriod, stopCh) } go func() { defer utilruntime.HandleCrash() e.checkLeftoverEndpoints() }() <-stopCh

}

func (e *EndpointController) worker() {

for e.processNextWorkItem() { }

}

func (e *EndpointController) processNextWorkItem() bool {

eKey, quit := e.queue.Get() if quit { return false } defer e.queue.Done(eKey) err := e.syncService(eKey.(string)) e.handleErr(err, eKey) return true

}

endpointController的主要逻辑在syncService函数

func (e *EndpointController) syncService(key string) error {

startTime := time.Now() defer func() { klog.V(4).Infof("Finished syncing service %q endpoints. (%v)", key, time.Since(startTime)) }()

// 根据key获取service的namespace和name

namespace, name, err := cache.SplitMetaNamespaceKey(key) if err != nil { return err } service, err := e.serviceLister.Services(namespace).Get(name) if err != nil {

// 如果service已经被删除，则也要删除对用的endpoint资源

 // Delete the corresponding endpoint, as the service has been deleted. // TODO: Please note that this will delete an endpoint when a // service is deleted. However, if we're down at the time when // the service is deleted, we will miss that deletion, so this // doesn't completely solve the problem. See #6877. err = e.client.CoreV1().Endpoints(namespace).Delete(name, nil) if err != nil && !errors.IsNotFound(err) { return err } e.triggerTimeTracker.DeleteEndpoints(namespace, name) return nil } // 如果service的.spec.selector字段为空，直接返回，endpointController不处理这种情况 if service.Spec.Selector == nil { // services without a selector receive no endpoints from this controller; // these services will receive the endpoints that are created out-of-band via the REST API. return nil } klog.V(5).Infof("About to update endpoints for service %q", key) pods, err := e.podLister.Pods(service.Namespace).List(labels.Set(service.Spec.Selector).AsSelectorPreValidated()) if err != nil { // Since we're getting stuff from a local cache, it is // basically impossible to get this error. return err } // If the user specified the older (deprecated) annotation, we have to respect it. tolerateUnreadyEndpoints := service.Spec.PublishNotReadyAddresses //如果service的注解含有key为service.alpha.kubernetes.io/tolerate-unready-endpoints的值，该值为bool类型，默认tolerateUnreadyEndpoints值为false if v, ok := service.Annotations[TolerateUnreadyEndpointsAnnotation]; ok { b, err := strconv.ParseBool(v) if err == nil { tolerateUnreadyEndpoints = b } else { utilruntime.HandleError(fmt.Errorf("Failed to parse annotation %v: %v", TolerateUnreadyEndpointsAnnotation, err)) } } // We call ComputeEndpointsLastChangeTriggerTime here to make sure that the state of the trigger // time tracker gets updated even if the sync turns out to be no-op and we don't update the // endpoints object. endpointsLastChangeTriggerTime := e.triggerTimeTracker. ComputeEndpointsLastChangeTriggerTime(namespace, name, service, pods) subsets := []v1.EndpointSubset{} var totalReadyEps int var totalNotReadyEps int //循环处理pod列表 for _, pod := range pods { // pod的podIp为空，则continue for循环 if len(pod.Status.PodIP) == 0 { klog.V(5).Infof("Failed to find an IP for pod %s/%s", pod.Namespace, pod.Name) continue } // 如果该pod正在被删除，则continue for循环 if !tolerateUnreadyEndpoints && pod.DeletionTimestamp != nil {

// 获取该pod的信息，输出EndpointAddress结构体变量

 klog.V(5).Infof("Pod is being deleted %s/%s", pod.Namespace, pod.Name) continue } epa := *podToEndpointAddress(pod) hostname := pod.Spec.Hostname // 如果pod存在hostname，则最后的FQDN为hostname.subdomain.namespace.svc.cluster.local if len(hostname) > 0 && pod.Spec.Subdomain == service.Name && service.Namespace == pod.Namespace { epa.Hostname = hostname } // Allow headless service not to have ports. // 允许headless service没有端口 if len(service.Spec.Ports) == 0 { if service.Spec.ClusterIP == api.ClusterIPNone {

// 1、如果tolerateUnreadyEndpoints为true，允许未就绪的pod也列入Addresses列表，如果tolerateUnreadyEndpoints为false但pod状态为ready则将pod列入Addresses列表;

 // 2、检测pod的重启策略，如果重启策略为Never，pod的运行状态不为Failed且不是Succeeded，将该pod列入NotReadyAddresses，如果重启策略为OnFailure并且pod的运行状态不为Succeeded，将该pod列入NotReadyAddresses，其它情况也将该pod列入NotReadyAddresses； subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, tolerateUnreadyEndpoints) // No need to repack subsets for headless service without ports. } } else { // 循环service的ports端口 for i := range service.Spec.Ports { servicePort := &service.Spec.Ports[i] portName := servicePort.Name portProto := servicePort.Protocol portNum, err := podutil.FindPort(pod, servicePort) // 如果service中的port在pod中不存在，则继续for循环 if err != nil { klog.V(4).Infof("Failed to find port for service %s/%s: %v", service.Namespace, service.Name, err) continue } var readyEps, notReadyEps int epp := &v1.EndpointPort{Name: portName, Port: int32(portNum), Protocol: portProto} subsets, readyEps, notReadyEps = addEndpointSubset(subsets, pod, epa, epp, tolerateUnreadyEndpoints) totalReadyEps = totalReadyEps + readyEps totalNotReadyEps = totalNotReadyEps + notReadyEps } } } // 重新整理subsets subsets = endpoints.RepackSubsets(subsets) // 如果endpoint不存在(通常该情况是新建一个service的情况)，则新建一个，如果是其他未知错误，则返回err // See if there's actually an update here. currentEndpoints, err := e.endpointsLister.Endpoints(service.Namespace).Get(service.Name) if err != nil { if errors.IsNotFound(err) { currentEndpoints = &v1.Endpoints{ ObjectMeta: metav1.ObjectMeta{ Name: service.Name, Labels: service.Labels, }, } } else { return err } } // currentEndpoints的资源版本为空时，表示要创建endpoint createEndpoints := len(currentEndpoints.ResourceVersion) == 0 // 如果当前currentEndpoints的subset列表和重新整理后的subsets相等，并且label与service的label一致，则忽略本次更新操作 if !createEndpoints && apiequality.Semantic.DeepEqual(currentEndpoints.Subsets, subsets) && apiequality.Semantic.DeepEqual(currentEndpoints.Labels, service.Labels) { klog.V(5).Infof("endpoints are equal for %s/%s, skipping update", service.Namespace, service.Name) return nil } newEndpoints := currentEndpoints.DeepCopy() newEndpoints.Subsets = subsets newEndpoints.Labels = service.Labels if newEndpoints.Annotations == nil { newEndpoints.Annotations = make(map[string]string) } if !endpointsLastChangeTriggerTime.IsZero() { newEndpoints.Annotations[v1.EndpointsLastChangeTriggerTime] = endpointsLastChangeTriggerTime.Format(time.RFC3339Nano) } else { // No new trigger time, clear the annotation. delete(newEndpoints.Annotations, v1.EndpointsLastChangeTriggerTime) } klog.V(4).Infof("Update endpoints for %v/%v, ready: %d not ready: %d", service.Namespace, service.Name, totalReadyEps, totalNotReadyEps) if createEndpoints { // 如果没有与service同命名空间和同名的endpoint，则生成新的endpoint // No previous endpoints, create them _, err = e.client.CoreV1().Endpoints(service.Namespace).Create(newEndpoints) } else { // Pre-existing // 已经存在与service同命名空间和同名的endpoint，需要更新endpoint _, err = e.client.CoreV1().Endpoints(service.Namespace).Update(newEndpoints) } if err != nil { if createEndpoints && errors.IsForbidden(err) { // A request is forbidden primarily for two reasons: // 1. namespace is terminating, endpoint creation is not allowed by default. // 2. policy is misconfigured, in which case no service would function anywhere. // Given the frequency of 1, we log at a lower level. klog.V(5).Infof("Forbidden from creating endpoints: %v", err) } if createEndpoints { e.eventRecorder.Eventf(newEndpoints, v1.EventTypeWarning, "FailedToCreateEndpoint", "Failed to create endpoint for service %v/%v: %v", service.Namespace, service.Name, err) } else { e.eventRecorder.Eventf(newEndpoints, v1.EventTypeWarning, "FailedToUpdateEndpoint", "Failed to update endpoint %v/%v: %v", service.Namespace, service.Name, err) } return err } return nil

}
1.5 Endpoint检测
之前说的是当Endpoint和Service绑定的时候Service和Pod改变时的一系列操作，现在我们回到问题，如果Endpoint单独存在，K8S是如何检测并且删除的？
我们重新看看Run函数中的

go func() {

 defer utilruntime.HandleCrash() e.checkLeftoverEndpoints() }()

K8S在运行Run函数的时候启动了一个协程去检测当前所有的Endpoint

// checkLeftoverEndpoints lists all currently existing endpoints and adds their
// service to the queue. This will detect endpoints that exist with no
// corresponding service; these endpoints need to be deleted. We only need to
// do this once on startup, because in steady-state these are detected (but
// some stragglers could have been left behind if the endpoint controller
// reboots).
func (e *EndpointController) checkLeftoverEndpoints() {
//拉取当前所有的endpoint对象

list, err := e.endpointsLister.List(labels.Everything()) if err != nil { utilruntime.HandleError(fmt.Errorf("Unable to list endpoints (%v); orphaned endpoints will not be cleaned up. (They're pretty harmless, but you can restart this component if you want another attempt made.)", err)) return }

//轮询所有endpoint

for _, ep := range list { if _, ok := ep.Annotations[resourcelock.LeaderElectionRecordAnnotationKey]; ok { // when there are multiple controller-manager instances, // we observe that it will delete leader-election endpoints after 5min // and cause re-election // so skip the delete here // as leader-election only have endpoints without service continue } key, err := controller.KeyFunc(ep) if err != nil { utilruntime.HandleError(fmt.Errorf("Unable to get key for endpoint %#v", ep)) continue }

//假如此处endpoint没有对应的service，猜想会把endpoint的name当成key传入queue，然后在之前的逻辑中判断获取service name错误，于是删除endpoint

 e.queue.Add(key) }

}

1. 总结
   一句话，遇到如上问题有两种解决的方式：

创建Service的时候使用Selector，这样可以自动创建Endpoint
在创建Endpoint还需要创建Service，这样才可以持久化Endpoint