kubernetes源码-EndpointSlice Controller 原理和源码分析

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

func startEndpointSliceController(ctx ControllerContext) (http.Handler, bool, error) {
  go endpointslicecontroller.NewController(
    ctx.InformerFactory.Core().V1().Pods(),
    ctx.InformerFactory.Core().V1().Services(),
    ctx.InformerFactory.Core().V1().Nodes(),
    ctx.InformerFactory.Discovery().V1().EndpointSlices(),
    ctx.ComponentConfig.EndpointSliceController.MaxEndpointsPerSlice,
    ctx.ClientBuilder.ClientOrDie("endpointslice-controller"),
    ctx.ComponentConfig.EndpointSliceController.EndpointUpdatesBatchPeriod.Duration,
  ).Run(int(ctx.ComponentConfig.EndpointSliceController.ConcurrentServiceEndpointSyncs), ctx.Stop)
  return nil, true, 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
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

// NewController creates and initializes a new Controller
func NewController(podInformer coreinformers.PodInformer,
  serviceInformer coreinformers.ServiceInformer,
  nodeInformer coreinformers.NodeInformer,
  endpointSliceInformer discoveryinformers.EndpointSliceInformer,
  maxEndpointsPerSlice int32,
  client clientset.Interface,
  endpointUpdatesBatchPeriod time.Duration,
) *Controller {
  broadcaster := record.NewBroadcaster()
  broadcaster.StartStructuredLogging(0)
  broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})
  recorder := broadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "endpoint-slice-controller"})

  if client != nil && client.CoreV1().RESTClient().GetRateLimiter() != nil {
    ratelimiter.RegisterMetricAndTrackRateLimiterUsage("endpoint_slice_controller", client.DiscoveryV1().RESTClient().GetRateLimiter())
  }

  endpointslicemetrics.RegisterMetrics()

  c := &Controller{
    client: client,
    // This is similar to the DefaultControllerRateLimiter, just with a
    // significantly higher default backoff (1s vs 5ms). This controller
    // processes events that can require significant EndpointSlice changes,
    // such as an update to a Service or Deployment. A more significant
    // rate limit back off here helps ensure that the Controller does not
    // overwhelm the API Server.
    queue: workqueue.NewNamedRateLimitingQueue(workqueue.NewMaxOfRateLimiter(
      workqueue.NewItemExponentialFailureRateLimiter(defaultSyncBackOff, maxSyncBackOff),
      // 10 qps, 100 bucket size. This is only for retry speed and its
      // only the overall factor (not per item).
      &workqueue.BucketRateLimiter{Limiter: rate.NewLimiter(rate.Limit(10), 100)},
    ), "endpoint_slice"),
    workerLoopPeriod: time.Second,
  }

  serviceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc: c.onServiceUpdate,
    UpdateFunc: func(old, cur interface{}) {
      c.onServiceUpdate(cur)
    },
    DeleteFunc: c.onServiceDelete,
  })
  c.serviceLister = serviceInformer.Lister()
  c.servicesSynced = serviceInformer.Informer().HasSynced

  podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc:    c.addPod,
    UpdateFunc: c.updatePod,
    DeleteFunc: c.deletePod,
  })
  c.podLister = podInformer.Lister()
  c.podsSynced = podInformer.Informer().HasSynced

  c.nodeLister = nodeInformer.Lister()
  c.nodesSynced = nodeInformer.Informer().HasSynced

  endpointSliceInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
    AddFunc:    c.onEndpointSliceAdd,
    UpdateFunc: c.onEndpointSliceUpdate,
    DeleteFunc: c.onEndpointSliceDelete,
  })

  c.endpointSliceLister = endpointSliceInformer.Lister()
  c.endpointSlicesSynced = endpointSliceInformer.Informer().HasSynced
  c.endpointSliceTracker = endpointsliceutil.NewEndpointSliceTracker()

  c.maxEndpointsPerSlice = maxEndpointsPerSlice

  c.triggerTimeTracker = endpointutil.NewTriggerTimeTracker()

  c.eventBroadcaster = broadcaster
  c.eventRecorder = recorder

  c.endpointUpdatesBatchPeriod = endpointUpdatesBatchPeriod
  c.serviceSelectorCache = endpointutil.NewServiceSelectorCache()

  if utilfeature.DefaultFeatureGate.Enabled(features.TopologyAwareHints) {
    nodeInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
      AddFunc:    c.addNode,
      UpdateFunc: c.updateNode,
      DeleteFunc: c.deleteNode,
    })

    c.topologyCache = topologycache.NewTopologyCache()
  }

  c.reconciler = &reconciler{
    client:               c.client,
    nodeLister:           c.nodeLister,
    maxEndpointsPerSlice: c.maxEndpointsPerSlice,
    endpointSliceTracker: c.endpointSliceTracker,
    metricsCache:         endpointslicemetrics.NewCache(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

// Run will not return until stopCh is closed.
func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
  defer utilruntime.HandleCrash()
  defer c.queue.ShutDown()

  klog.Infof("Starting endpoint slice controller")
  defer klog.Infof("Shutting down endpoint slice controller")

  if !cache.WaitForNamedCacheSync("endpoint_slice", stopCh, c.podsSynced, c.servicesSynced, c.endpointSlicesSynced, c.nodesSynced) {
    return
  }

  for i := 0; i < workers; i++ {
    go wait.Until(c.worker, c.workerLoopPeriod, stopCh)
  }

  go func() {
    defer utilruntime.HandleCrash()
  }()

  <-stopCh
}

// worker runs a worker thread that just dequeues items, processes them, and
// marks them done. You may run as many of these in parallel as you wish; the
// workqueue guarantees that they will not end up processing the same service
// at the same time
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
}

func (c *Controller) handleErr(err error, key interface{}) {
  trackSync(err)

  if err == nil {
    c.queue.Forget(key)
    return
  }

  if c.queue.NumRequeues(key) < maxRetries {
    klog.Warningf("Error syncing endpoint slices for service %q, retrying. Error: %v", key, err)
    c.queue.AddRateLimited(key)
    return
  }

  klog.Warningf("Retry budget exceeded, dropping service %q out of the queue: %v", key, err)
  c.queue.Forget(key)
  utilruntime.HandleError(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
50
51
52
53
54
55
56
57
58
59

func (r *reconciler) reconcile(service *corev1.Service, pods []*corev1.Pod, existingSlices []*discovery.EndpointSlice, triggerTime time.Time) error {
  slicesToDelete := []*discovery.EndpointSlice{}                                    // slices that are no longer  matching any address the service has
  errs := []error{}                                                                 // all errors generated in the process of reconciling
  slicesByAddressType := make(map[discovery.AddressType][]*discovery.EndpointSlice) // slices by address type

  // addresses that this service supports [o(1) find]
  serviceSupportedAddressesTypes := getAddressTypesForService(service)

  // loop through slices identifying their address type.
  // slices that no longer match address type supported by services
  // go to delete, other slices goes to the reconciler machinery
  // for further adjustment
  for _, existingSlice := range existingSlices {
    // service no longer supports that address type, add it to deleted slices
    if _, ok := serviceSupportedAddressesTypes[existingSlice.AddressType]; !ok {
      if r.topologyCache != nil {
        svcKey, err := serviceControllerKey(existingSlice)
        if err != nil {
          klog.Warningf("Couldn't get key to remove EndpointSlice from topology cache %+v: %v", existingSlice, err)
        } else {
          r.topologyCache.RemoveHints(svcKey, existingSlice.AddressType)
        }
      }

      slicesToDelete = append(slicesToDelete, existingSlice)
      continue
    }

    // add list if it is not on our map
    if _, ok := slicesByAddressType[existingSlice.AddressType]; !ok {
      slicesByAddressType[existingSlice.AddressType] = make([]*discovery.EndpointSlice, 0, 1)
    }

    slicesByAddressType[existingSlice.AddressType] = append(slicesByAddressType[existingSlice.AddressType], existingSlice)
  }

  // reconcile for existing.
  for addressType := range serviceSupportedAddressesTypes {
    existingSlices := slicesByAddressType[addressType]
    err := r.reconcileByAddressType(service, pods, existingSlices, triggerTime, addressType)
    if err != nil {
      errs = append(errs, err)
    }
  }

  // delete those which are of addressType that is no longer supported
  // by the service
  for _, sliceToDelete := range slicesToDelete {
    err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Delete(context.TODO(), sliceToDelete.Name, metav1.DeleteOptions{})
    if err != nil {
      errs = append(errs, fmt.Errorf("error deleting %s EndpointSlice for Service %s/%s: %w", sliceToDelete.Name, service.Namespace, service.Name, err))
    } else {
      r.endpointSliceTracker.ExpectDeletion(sliceToDelete)
      metrics.EndpointSliceChanges.WithLabelValues("delete").Inc()
    }
  }

  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

func (r *reconciler) reconcileByAddressType(service *corev1.Service, pods []*corev1.Pod, existingSlices []*discovery.EndpointSlice, triggerTime time.Time, addressType discovery.AddressType) error {

  slicesToCreate := []*discovery.EndpointSlice{}
  slicesToUpdate := []*discovery.EndpointSlice{}
  slicesToDelete := []*discovery.EndpointSlice{}

  // Build data structures for existing state.
  existingSlicesByPortMap := map[endpointutil.PortMapKey][]*discovery.EndpointSlice{}
  numExistingEndpoints := 0
  for _, existingSlice := range existingSlices {
    if ownedBy(existingSlice, service) {
      epHash := endpointutil.NewPortMapKey(existingSlice.Ports)
      existingSlicesByPortMap[epHash] = append(existingSlicesByPortMap[epHash], existingSlice)
      numExistingEndpoints += len(existingSlice.Endpoints)
    } else {
      slicesToDelete = append(slicesToDelete, existingSlice)
    }
  }

  // Build data structures for desired state.
  desiredMetaByPortMap := map[endpointutil.PortMapKey]*endpointMeta{}
  desiredEndpointsByPortMap := map[endpointutil.PortMapKey]endpointsliceutil.EndpointSet{}
  numDesiredEndpoints := 0

  for _, pod := range pods {
    includeTerminating := service.Spec.PublishNotReadyAddresses || utilfeature.DefaultFeatureGate.Enabled(features.EndpointSliceTerminatingCondition)
    if !endpointutil.ShouldPodBeInEndpoints(pod, includeTerminating) {
      continue
    }

    endpointPorts := getEndpointPorts(service, pod)
    epHash := endpointutil.NewPortMapKey(endpointPorts)
    if _, ok := desiredEndpointsByPortMap[epHash]; !ok {
      desiredEndpointsByPortMap[epHash] = endpointsliceutil.EndpointSet{}
    }

    if _, ok := desiredMetaByPortMap[epHash]; !ok {
      desiredMetaByPortMap[epHash] = &endpointMeta{
        AddressType: addressType,
        Ports:       endpointPorts,
      }
    }

    node, err := r.nodeLister.Get(pod.Spec.NodeName)
    if err != nil {
      return err
    }
    endpoint := podToEndpoint(pod, node, service, addressType)
    if len(endpoint.Addresses) > 0 {
      desiredEndpointsByPortMap[epHash].Insert(&endpoint)
      numDesiredEndpoints++
    }
  }

  spMetrics := metrics.NewServicePortCache()
  totalAdded := 0
  totalRemoved := 0

  // Determine changes necessary for each group of slices by port map.
  for portMap, desiredEndpoints := range desiredEndpointsByPortMap {
    numEndpoints := len(desiredEndpoints)
    pmSlicesToCreate, pmSlicesToUpdate, pmSlicesToDelete, added, removed := r.reconcileByPortMapping(
      service, existingSlicesByPortMap[portMap], desiredEndpoints, desiredMetaByPortMap[portMap])

    totalAdded += added
    totalRemoved += removed

    spMetrics.Set(portMap, metrics.EfficiencyInfo{
      Endpoints: numEndpoints,
      Slices:    len(existingSlicesByPortMap[portMap]) + len(pmSlicesToCreate) - len(pmSlicesToDelete),
    })

    slicesToCreate = append(slicesToCreate, pmSlicesToCreate...)
    slicesToUpdate = append(slicesToUpdate, pmSlicesToUpdate...)
    slicesToDelete = append(slicesToDelete, pmSlicesToDelete...)
  }

  // If there are unique sets of ports that are no longer desired, mark
  // the corresponding endpoint slices for deletion.
  for portMap, existingSlices := range existingSlicesByPortMap {
    if _, ok := desiredEndpointsByPortMap[portMap]; !ok {
      for _, existingSlice := range existingSlices {
        slicesToDelete = append(slicesToDelete, existingSlice)
      }
    }
  }

  // When no endpoint slices would usually exist, we need to add a placeholder.
  if len(existingSlices) == len(slicesToDelete) && len(slicesToCreate) < 1 {
    placeholderSlice := newEndpointSlice(service, &endpointMeta{Ports: []discovery.EndpointPort{}, AddressType: addressType})
    slicesToCreate = append(slicesToCreate, placeholderSlice)
    spMetrics.Set(endpointutil.NewPortMapKey(placeholderSlice.Ports), metrics.EfficiencyInfo{
      Endpoints: 0,
      Slices:    1,
    })
  }

  metrics.EndpointsAddedPerSync.WithLabelValues().Observe(float64(totalAdded))
  metrics.EndpointsRemovedPerSync.WithLabelValues().Observe(float64(totalRemoved))

  serviceNN := types.NamespacedName{Name: service.Name, Namespace: service.Namespace}
  r.metricsCache.UpdateServicePortCache(serviceNN, spMetrics)

  // Topology hints are assigned per address type. This means it is
  // theoretically possible for endpoints of one address type to be assigned
  // hints while another endpoints of another address type are not.
  si := &topologycache.SliceInfo{
    ServiceKey: fmt.Sprintf("%s/%s", service.Namespace, service.Name),
    ToCreate:   slicesToCreate,
    ToUpdate:   slicesToUpdate,
    Unchanged:  unchangedSlices(existingSlices, slicesToUpdate, slicesToDelete),
  }

  if r.topologyCache != nil && hintsEnabled(service.Annotations) {
    slicesToCreate, slicesToUpdate = r.topologyCache.AddHints(si)
  } else {
    if r.topologyCache != nil {
      r.topologyCache.RemoveHints(si.ServiceKey, addressType)
    }
    slicesToCreate, slicesToUpdate = topologycache.RemoveHintsFromSlices(si)
  }

  return r.finalize(service, slicesToCreate, slicesToUpdate, slicesToDelete, triggerTime)
}

 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

func (r *reconciler) finalize(
  service *corev1.Service,
  slicesToCreate,
  slicesToUpdate,
  slicesToDelete []*discovery.EndpointSlice,
  triggerTime time.Time,
) error {
  // If there are slices to create and delete, change the creates to updates
  // of the slices that would otherwise be deleted.
  for i := 0; i < len(slicesToDelete); {
    if len(slicesToCreate) == 0 {
      break
    }
    sliceToDelete := slicesToDelete[i]
    slice := slicesToCreate[len(slicesToCreate)-1]
    // Only update EndpointSlices that are owned by this Service and have
    // the same AddressType. We need to avoid updating EndpointSlices that
    // are being garbage collected for an old Service with the same name.
    // The AddressType field is immutable. Since Services also consider
    // IPFamily immutable, the only case where this should matter will be
    // the migration from IP to IPv4 and IPv6 AddressTypes, where there's a
    // chance EndpointSlices with an IP AddressType would otherwise be
    // updated to IPv4 or IPv6 without this check.
    if sliceToDelete.AddressType == slice.AddressType && ownedBy(sliceToDelete, service) {
      slice.Name = sliceToDelete.Name
      slicesToCreate = slicesToCreate[:len(slicesToCreate)-1]
      slicesToUpdate = append(slicesToUpdate, slice)
      slicesToDelete = append(slicesToDelete[:i], slicesToDelete[i+1:]...)
    } else {
      i++
    }
  }

  // Don't create new EndpointSlices if the Service is pending deletion. This
  // is to avoid a potential race condition with the garbage collector where
  // it tries to delete EndpointSlices as this controller replaces them.
  if service.DeletionTimestamp == nil {
    for _, endpointSlice := range slicesToCreate {
      addTriggerTimeAnnotation(endpointSlice, triggerTime)
      createdSlice, err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Create(context.TODO(), endpointSlice, metav1.CreateOptions{})
      if err != nil {
        // If the namespace is terminating, creates will continue to fail. Simply drop the item.
        if errors.HasStatusCause(err, corev1.NamespaceTerminatingCause) {
          return nil
        }
        return fmt.Errorf("failed to create EndpointSlice for Service %s/%s: %v", service.Namespace, service.Name, err)
      }
      r.endpointSliceTracker.Update(createdSlice)
      metrics.EndpointSliceChanges.WithLabelValues("create").Inc()
    }
  }

  for _, endpointSlice := range slicesToUpdate {
    addTriggerTimeAnnotation(endpointSlice, triggerTime)
    updatedSlice, err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Update(context.TODO(), endpointSlice, metav1.UpdateOptions{})
    if err != nil {
      return fmt.Errorf("failed to update %s EndpointSlice for Service %s/%s: %v", endpointSlice.Name, service.Namespace, service.Name, err)
    }
    r.endpointSliceTracker.Update(updatedSlice)
    metrics.EndpointSliceChanges.WithLabelValues("update").Inc()
  }

  for _, endpointSlice := range slicesToDelete {
    err := r.client.DiscoveryV1().EndpointSlices(service.Namespace).Delete(context.TODO(), endpointSlice.Name, metav1.DeleteOptions{})
    if err != nil {
      return fmt.Errorf("failed to delete %s EndpointSlice for Service %s/%s: %v", endpointSlice.Name, service.Namespace, service.Name, err)
    }
    r.endpointSliceTracker.ExpectDeletion(endpointSlice)
    metrics.EndpointSliceChanges.WithLabelValues("delete").Inc()
  }

  topologyLabel := "Disabled"
  if r.topologyCache != nil && hintsEnabled(service.Annotations) {
    topologyLabel = "Auto"
  }

  numSlicesChanged := len(slicesToCreate) + len(slicesToUpdate) + len(slicesToDelete)
  metrics.EndpointSlicesChangedPerSync.WithLabelValues(topologyLabel).Observe(float64(numSlicesChanged))

  return nil
}