【kubernetes/k8s源码分析】kube-controller-manager之endpoint controller源码分析

蔚落 2022-06-09 10:47 137阅读 0赞

1 EndpointController 结构体

###  ###

*  client: 调用 apiserver 接口
 *  queue: 处理的队列

// EndpointController manages selector-based service endpoints.
    type EndpointController struct {
           client clientset.Interface
           
           serviceLister corelisters.ServiceLister
           servicesSynced cache.InformerSynced
           
           podLister corelisters.PodLister
           podsSynced cache.InformerSynced
           
           endpointsLister corelisters.EndpointsLister
           endpointsSynced cache.InformerSynced
    
           queue workqueue.RateLimitingInterface
    
           workerLoopPeriod time.Duration
    }

# 2 结构体初始化 #

###  ###

初始化 EndpointController 结构体，将 service 以及 pod 进行 serviceInformer podInformer 赋值

// NewEndpointController returns a new *EndpointController.
    func NewEndpointController(podInformer coreinformers.PodInformer, serviceInformer coreinformers.ServiceInformer,
           endpointsInformer coreinformers.EndpointsInformer, client clientset.Interface) *EndpointController {
           if client != nil && client.Core().RESTClient().GetRateLimiter() != nil {
                  metrics.RegisterMetricAndTrackRateLimiterUsage("endpoint_controller", client.Core().RESTClient().GetRateLimiter())
           }
           e := &EndpointController{
                  client:           client,
                  queue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "endpoint"),
                  workerLoopPeriod: time.Second,
           }
    
           serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
                  AddFunc: e.enqueueService,
                  UpdateFunc: func(old, cur interface{}) {
                         e.enqueueService(cur)
                  },
                  DeleteFunc: e.enqueueService,
           })
           e.serviceLister = serviceInformer.Lister()
           e.servicesSynced = serviceInformer.Informer().HasSynced
    
           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
    
           return e
    }

# 3 endpoint controller 启动 #

###  ###

goroutine 创建一个 endpointController 结构体并调用其 Run 方法

func startEndpointController(ctx ControllerContext) (bool, error) {
           go endpointcontroller.NewEndpointController(
                  ctx.InformerFactory.Core().V1().Pods(),
                  ctx.InformerFactory.Core().V1().Services(),
                  ctx.InformerFactory.Core().V1().Endpoints(),
                  ctx.ClientBuilder.ClientOrDie("endpoint-controller"),
           ).Run(int(ctx.Options.ConcurrentEndpointSyncs), ctx.Stop)
           return true, nil
    }

###  ###

3.1 Run 函数中 WaitForCacheSync 等待同步 pod service endpoint cache，然后启动几个 goroutine 工作处理

func (e *EndpointController) Run(workers int, stopCh <-chan struct{}) {
           defer utilruntime.HandleCrash()
           defer e.queue.ShutDown()
    
           if !controller.WaitForCacheSync("endpoint", stopCh, e.podsSynced, e.servicesSynced, e.endpointsSynced) {
                  return
           }
    
           for i := 0; i < workers; i++ {
                  go wait.Until(e.worker, e.workerLoopPeriod, stopCh)
           }
    
           go func() {
                  defer utilruntime.HandleCrash()
                  e.checkLeftoverEndpoints()
           }()
    
           <-stopCh
    }

###  ###

3.1.1 Worker循环处理，从队列中取出 key，调用 syncService 函数处理，第四章节讲解 syncService 函数

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
    }

# 4 syncService 函数 #

###  ###

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

###  ###

4.1 把 key 根据 / 拆分为 namespace 和 name，根据其获得 service

namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
           return err
    }
    service, err := e.serviceLister.Services(namespace).Get(name)
    if err != nil {
           // 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.Core().Endpoints(namespace).Delete(name, nil)
           if err != nil && !errors.IsNotFound(err) {
                  return err
           }
           return nil
    }

###  ###

4.2 获得相关的 pods

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
    }

###  ###

4,3 根据 namespace 和 name 查询所有 endpoints

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

###  ###

4.4 如果先前没有就调用 Create 接口创建，有的话就更新

createEndpoints := len(currentEndpoints.ResourceVersion) == 0
    if createEndpoints {
           // No previous endpoints, create them
           _, err = e.client.Core().Endpoints(service.Namespace).Create(newEndpoints)
    } else {
           // Pre-existing
           _, err = e.client.Core().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.
                  glog.V(5).Infof("Forbidden from creating endpoints: %v", err)
           }
           return err
    }

# 2 WaitForCacheSync 函数 #

###  ###

2.1 WaitForCacheSync 函数等待