k8s endpoints controller 分析

apiVersion: v1kind: Servicemetadata:  name: baidu-service  namespace: testspec:  type: ExternalName  externalName: www.baidu.com

// cmd/kube-controller-manager/app/core.gofunc startEndpointController(ctx ControllerContext) (http.Handler, 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"),    ctx.ComponentConfig.EndpointController.EndpointUpdatesBatchPeriod.Duration,  ).Run(int(ctx.ComponentConfig.EndpointController.ConcurrentEndpointSyncs), ctx.Stop)  return nil, true, nil}

// pkg/controller/endpoint/endpoints_controller.gofunc NewEndpointController(podInformer coreinformers.PodInformer, serviceInformer coreinformers.ServiceInformer,  endpointsInformer coreinformers.EndpointsInformer, client clientset.Interface, endpointUpdatesBatchPeriod time.Duration) *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 {    ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_controller", client.CoreV1().RESTClient().GetRateLimiter())  }  e := &EndpointController{    client:           client,    queue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "endpoint"),    workerLoopPeriod: time.Second,  }
  serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{    AddFunc: e.onServiceUpdate,    UpdateFunc: func(old, cur interface{}) {      e.onServiceUpdate(cur)    },    DeleteFunc: e.onServiceDelete,  })  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
  e.triggerTimeTracker = endpointutil.NewTriggerTimeTracker()  e.eventBroadcaster = broadcaster  e.eventRecorder = recorder
  e.endpointUpdatesBatchPeriod = endpointUpdatesBatchPeriod
  e.serviceSelectorCache = endpointutil.NewServiceSelectorCache()
  return e}

// pkg/controller/endpoint/endpoints_controller.gofunc (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")
  if !cache.WaitForNamedCacheSync("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}

// pkg/controller/endpoint/endpoints_controller.gofunc (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}

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))  }()
  namespace, name, err := cache.SplitMetaNamespaceKey(key)  if err != nil {    return err  }  service, err := e.serviceLister.Services(namespace).Get(name)  if err != nil {    if !errors.IsNotFound(err) {      return err    }
    // 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.DeleteService(namespace, name)    return nil  }
  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  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 ComputeEndpointLastChangeTriggerTime 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.    ComputeEndpointLastChangeTriggerTime(namespace, service, pods)
  subsets := []v1.EndpointSubset{}  var totalReadyEps int  var totalNotReadyEps int
  for _, pod := range pods {    if len(pod.Status.PodIP) == 0 {      klog.V(5).Infof("Failed to find an IP for pod %s/%s", pod.Namespace, pod.Name)      continue    }    if !tolerateUnreadyEndpoints && pod.DeletionTimestamp != nil {      klog.V(5).Infof("Pod is being deleted %s/%s", pod.Namespace, pod.Name)      continue    }
    ep, err := podToEndpointAddressForService(service, pod)    if err != nil {      // this will happen, if the cluster runs with some nodes configured as dual stack and some as not      // such as the case of an upgrade..      klog.V(2).Infof("failed to find endpoint for service:%v with ClusterIP:%v on pod:%v with error:%v", service.Name, service.Spec.ClusterIP, pod.Name, err)      continue    }
    epa := *ep    if endpointutil.ShouldSetHostname(pod, service) {      epa.Hostname = pod.Spec.Hostname    }
    // Allow headless service not to have ports.    if len(service.Spec.Ports) == 0 {      if service.Spec.ClusterIP == api.ClusterIPNone {        subsets, totalReadyEps, totalNotReadyEps = addEndpointSubset(subsets, pod, epa, nil, tolerateUnreadyEndpoints)        // No need to repack subsets for headless service without ports.      }    } else {      for i := range service.Spec.Ports {        servicePort := &service.Spec.Ports[i]
        portName := servicePort.Name        portProto := servicePort.Protocol        portNum, err := podutil.FindPort(pod, servicePort)        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 = endpoints.RepackSubsets(subsets)
  // 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    }  }
  createEndpoints := len(currentEndpoints.ResourceVersion) == 0
  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)  }
  if newEndpoints.Labels == nil {    newEndpoints.Labels = make(map[string]string)  }
  if !helper.IsServiceIPSet(service) {    newEndpoints.Labels = utillabels.CloneAndAddLabel(newEndpoints.Labels, v1.IsHeadlessService, "")  } else {    newEndpoints.Labels = utillabels.CloneAndRemoveLabel(newEndpoints.Labels, v1.IsHeadlessService)  }
  klog.V(4).Infof("Update endpoints for %v/%v, ready: %d not ready: %d", service.Namespace, service.Name, totalReadyEps, totalNotReadyEps)  if createEndpoints {    // No previous endpoints, create them    _, err = e.client.CoreV1().Endpoints(service.Namespace).Create(newEndpoints)  } else {    // Pre-existing    _, 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 the namespace is terminating, creates will continue to fail. Simply drop the item.      if errors.HasStatusCause(err, v1.NamespaceTerminatingCause) {        return nil      }    }
    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}

// pkg/controller/endpoint/endpoints_controller.gofunc addEndpointSubset(subsets []v1.EndpointSubset, pod *v1.Pod, epa v1.EndpointAddress,  epp *v1.EndpointPort, tolerateUnreadyEndpoints bool) ([]v1.EndpointSubset, int, int) {  var readyEps int  var notReadyEps int  ports := []v1.EndpointPort{}  if epp != nil {    ports = append(ports, *epp)  }  if tolerateUnreadyEndpoints || podutil.IsPodReady(pod) {    subsets = append(subsets, v1.EndpointSubset{      Addresses: []v1.EndpointAddress{epa},      Ports:     ports,    })    readyEps++  } else if shouldPodBeInEndpoints(pod) {    klog.V(5).Infof("Pod is out of service: %s/%s", pod.Namespace, pod.Name)    subsets = append(subsets, v1.EndpointSubset{      NotReadyAddresses: []v1.EndpointAddress{epa},      Ports:             ports,    })    notReadyEps++  }  return subsets, readyEps, notReadyEps}
func shouldPodBeInEndpoints(pod *v1.Pod) bool {  switch pod.Spec.RestartPolicy {  case v1.RestartPolicyNever:    return pod.Status.Phase != v1.PodFailed && pod.Status.Phase != v1.PodSucceeded  case v1.RestartPolicyOnFailure:    return pod.Status.Phase != v1.PodSucceeded  default:    return true  }}

// pkg/api/v1/pod/util.go// IsPodReady returns true if a pod is ready; false otherwise.func IsPodReady(pod *v1.Pod) bool {  return IsPodReadyConditionTrue(pod.Status)}
// GetPodReadyCondition extracts the pod ready condition from the given status and returns that.// Returns nil if the condition is not present.func GetPodReadyCondition(status v1.PodStatus) *v1.PodCondition {  _, condition := GetPodCondition(&status, v1.PodReady)  return condition}