在前面的文章中已经介绍了 deployment controller 的设计与实现，deployment 控制的是 replicaset，而 replicaset 控制 pod 的创建与删除，deployment 通过控制 replicaset 实现了滚动更新、回滚等操作。而 replicaset 会直接控制 pod 的创建与删除，本文会继续从源码层面分析 replicaset 的设计与实现。

在分析源码前先考虑一下 replicaset 的使用场景，在平时的操作中其实我们并不会直接操作 replicaset，replicaset 也仅有几个简单的操作，创建、删除、更新等，但其地位是非常重要的，replicaset 的主要功能就是通过 add/del pod 来达到期望的状态。

ReplicaSetController 源码分析

kubernetes 版本: v1.16

启动流程

首先来看 replicaSetController 对象初始化以及启动的代码，在 startReplicaSetController 中有两个比较重要的变量：

  • BurstReplicas：用来控制在一个 syncLoop 过程中 rs 最多能创建的 pod 数量，设置上限值是为了避免单个 rs 影响整个系统，默认值为 500；
  • ConcurrentRSSyncs：指的是需要启动多少个 goroutine 处理 informer 队列中的对象，默认值为 5；

k8s.io/kubernetes/cmd/kube-controller-manager/app/apps.go:69

  1. func startReplicaSetController(ctx ControllerContext) (http.Handler, bool, error) {
  2.     if !ctx.AvailableResources[schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "replicasets"}] {
  3.         return nil, false, nil
  4.     }
  5.     go replicaset.NewReplicaSetController(
  6.         ctx.InformerFactory.Apps().V1().ReplicaSets(),
  7.         ctx.InformerFactory.Core().V1().Pods(),
  8.         ctx.ClientBuilder.ClientOrDie("replicaset-controller"),
  9.         replicaset.BurstReplicas,
  10.     ).Run(int(ctx.ComponentConfig.ReplicaSetController.ConcurrentRSSyncs), ctx.Stop)
  11.     return nil, true, nil
  12. }

下面是 replicaSetController 初始化的具体步骤，可以看到其会监听 pod 以及 rs 两个对象的事件。

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:109

  1. func NewReplicaSetController(......) *ReplicaSetController {
  2.     ......
  3.     // 1、此处调用 NewBaseController
  4.     return NewBaseController(rsInformer, podInformer, kubeClient, burstReplicas,
  5.         apps.SchemeGroupVersion.WithKind("ReplicaSet"),
  6.         "replicaset_controller",
  7.         "replicaset",
  8.         controller.RealPodControl{
  9.             KubeClient: kubeClient,
  10.             Recorder:   eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "replicaset-controller"}),
  11.         },
  12.     )
  13. }
  15. func NewBaseController(......) *ReplicaSetController {
  16.     ......
  17.     // 2、ReplicaSetController 初始化
  18.     rsc := &ReplicaSetController{
  19.         GroupVersionKind: gvk,
  20.         kubeClient:       kubeClient,
  21.         podControl:       podControl,
  22.         burstReplicas:    burstReplicas,
  23.         // 3、expectations 的初始化
  24.         expectations:     controller.NewUIDTrackingControllerExpectations(controller.NewControllerExpectations()),
  25.         queue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName),
  26.     }
  28.     // 4、rsInformer 中注册的 EventHandler
  29.     rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  30.         AddFunc:    rsc.enqueueReplicaSet,
  31.         UpdateFunc: rsc.updateRS,
  32.         DeleteFunc: rsc.enqueueReplicaSet,
  33.     })
  34.     ......
  36.     // 5、podInformer 中注册的 EventHandler
  37.     podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  38.         AddFunc: rsc.addPod,
  39.         UpdateFunc: rsc.updatePod,
  40.         DeleteFunc: rsc.deletePod,
  41.     })
  42.     ......
  44.     return rsc
  45. }

replicaSetController 初始化完成后会调用 Run 方法启动 5 个 goroutine 处理 informer 队列中的事件并进行 sync 操作，kube-controller-manager 中每个 controller 的启动操作都是如下所示流程。

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:177

  1. func (rsc *ReplicaSetController) Run(workers int, stopCh <-chan struct{}) {
  2.     ......
  4.     // 1、等待 informer 同步缓存
  5.     if !cache.WaitForNamedCacheSync(rsc.Kind, stopCh, rsc.podListerSynced, rsc.rsListerSynced) {
  6.         return
  7.     }
  9.     // 2、启动 5 个 goroutine 执行 worker 方法
  10.     for i := 0; i < workers; i++ {
  11.         go wait.Until(rsc.worker, time.Second, stopCh)
  12.     }
  14.     <-stopCh
  15. }
  17. // 3、worker 方法中调用 rocessNextWorkItem
  18. func (rsc *ReplicaSetController) worker() {
  19.     for rsc.processNextWorkItem() {
  20.     }
  21. }
  23. func (rsc *ReplicaSetController) processNextWorkItem() bool {
  24.     // 4、从队列中取出对象
  25.     key, quit := rsc.queue.Get()
  26.     if quit {
  27.         return false
  28.     }
  29.     defer rsc.queue.Done(key)
  31.     // 5、执行 sync 操作
  32.     err := rsc.syncHandler(key.(string))
  33.     ......
  35.     return true
  36. }

EventHandler

初始化 replicaSetController 时，其中有一个 expectations 字段，这是 rs 中一个比较特殊的机制，为了说清楚 expectations，先来看一下 controller 中所注册的 eventHandler，replicaSetController 会 watch pod 和 replicaSet 两个对象，eventHandler 中注册了对这两种对象的 add、update、delete 三个操作。

addPod
  • 1、判断 pod 是否处于删除状态；
  • 2、获取该 pod 关联的 rs 以及 rsKey，入队 rs 并更新 rsKey 的 expectations；
  • 3、若 pod 对象没体现出关联的 rs 则为孤儿 pod，遍历 rsList 查找匹配的 rs，若该 rs.Namespace == pod.Namespace 并且 rs.Spec.Selector 匹配 pod.Labels，则说明该 pod 应该与此 rs 关联，将匹配的 rs 入队；

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:255

  1. func (rsc *ReplicaSetController) addPod(obj interface{}) {
  2.     pod := obj.(*v1.Pod)
  4.     if pod.DeletionTimestamp != nil {
  5.         rsc.deletePod(pod)
  6.         return
  7.     }
  9.     // 1、获取 pod 所关联的 rs
  10.     if controllerRef := metav1.GetControllerOf(pod); controllerRef != nil {
  11.         rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)
  12.         if rs == nil {
  13.             return
  14.         }
  15.         rsKey, err := controller.KeyFunc(rs)
  16.         if err != nil {
  17.             return
  18.         }
  19.         // 2、更新 expectations，rsKey 的 add - 1
  20.         rsc.expectations.CreationObserved(rsKey)
  21.         rsc.enqueueReplicaSet(rs)
  22.         return
  23.     }
  26.     rss := rsc.getPodReplicaSets(pod)
  27.     if len(rss) == 0 {
  28.         return
  29.     }
  31.     for _, rs := range rss {
  32.         rsc.enqueueReplicaSet(rs)
  33.     }
  34. }
updatePod
  • 1、如果 pod label 改变或者处于删除状态，则直接删除；
  • 2、如果 pod 的 OwnerReference 发生改变，此时 oldRS 需要创建 pod，将 oldRS 入队；
  • 3、获取 pod 关联的 rs，入队 rs，若 pod 当前处于 ready 并非 available 状态，则会再次将该 rs 加入到延迟队列中，因为 pod 从 ready 到 available 状态需要触发一次 status 的更新；
  • 4、否则为孤儿 pod，遍历 rsList 查找匹配的 rs，若找到则将 rs 入队；

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:298

  1. func (rsc *ReplicaSetController) updatePod(old, cur interface{}) {
  2.     curPod := cur.(*v1.Pod)
  3.     oldPod := old.(*v1.Pod)
  4.     if curPod.ResourceVersion == oldPod.ResourceVersion {
  5.         return
  6.     }
  8.     // 1、如果 pod label 改变或者处于删除状态，则直接删除
  9.     labelChanged := !reflect.DeepEqual(curPod.Labels, oldPod.Labels)
  10.     if curPod.DeletionTimestamp != nil {
  11.         rsc.deletePod(curPod)
  12.         if labelChanged {
  13.             rsc.deletePod(oldPod)
  14.         }
  15.         return
  16.     }
  18.     // 2、如果 pod 的 OwnerReference 发生改变，将 oldRS 入队
  19.     curControllerRef := metav1.GetControllerOf(curPod)
  20.     oldControllerRef := metav1.GetControllerOf(oldPod)
  21.     controllerRefChanged := !reflect.DeepEqual(curControllerRef, oldControllerRef)
  22.     if controllerRefChanged && oldControllerRef != nil {
  23.         if rs := rsc.resolveControllerRef(oldPod.Namespace, oldControllerRef); rs != nil {
  24.             rsc.enqueueReplicaSet(rs)
  25.         }
  26.     }
  28.     // 3、获取 pod 关联的 rs，入队 rs
  29.     if curControllerRef != nil {
  30.         rs := rsc.resolveControllerRef(curPod.Namespace, curControllerRef)
  31.         if rs == nil {
  32.             return
  33.         }
  35.         rsc.enqueueReplicaSet(rs)
  36.         if !podutil.IsPodReady(oldPod) && podutil.IsPodReady(curPod) && rs.Spec.MinReadySeconds > 0 {
  37.             rsc.enqueueReplicaSetAfter(rs, (time.Duration(rs.Spec.MinReadySeconds)*time.Second)+time.Second)
  38.         }
  39.         return
  40.     }
  43.     // 4、查找匹配的 rs
  44.     if labelChanged || controllerRefChanged {
  45.         rss := rsc.getPodReplicaSets(curPod)
  46.         if len(rss) == 0 {
  47.             return
  48.         }
  49.         for _, rs := range rss {
  50.             rsc.enqueueReplicaSet(rs)
  51.         }
  52.     }
  53. }
deletePod
  • 1、确认该对象是否为 pod；
  • 2、判断是否为孤儿 pod；
  • 3、获取其对应的 rs 以及 rsKey；
  • 4、更新 expectations 中 rsKey 的 del 值；
  • 5、将 rs 入队；

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:372

  1. func (rsc *ReplicaSetController) deletePod(obj interface{}) {
  2.     pod, ok := obj.(*v1.Pod)
  4.     if !ok {
  5.         ......
  6.     }
  8.     controllerRef := metav1.GetControllerOf(pod)
  9.     if controllerRef == nil {
  10.         return
  11.     }
  12.     rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)
  13.     if rs == nil {
  14.         return
  15.     }
  16.     rsKey, err := controller.KeyFunc(rs)
  17.     if err != nil {
  18.         return
  19.     }
  20.     // 更新 expectations，该 rsKey 的 del - 1
  21.     rsc.expectations.DeletionObserved(rsKey, controller.PodKey(pod))
  22.     rsc.enqueueReplicaSet(rs)
  23. }
AddRS 和 DeleteRS

以上两个操作仅仅是将对应的 rs 入队。

UpdateRS

其实 updateRS 也仅仅是将对应的 rs 进行入队，不过多了一个打印日志的操作，如下所示：

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:232

  1. func (rsc *ReplicaSetController) updateRS(old, cur interface{}) {
  2.     oldRS := old.(*apps.ReplicaSet)
  3.     curRS := cur.(*apps.ReplicaSet)
  5.     if *(oldRS.Spec.Replicas) != *(curRS.Spec.Replicas) {
  6.         klog.V(4).Infof("%v %v updated. Desired pod count change: %d->%d", rsc.Kind, curRS.Name, *(oldRS.Spec.Replicas), *(curRS.Spec.Replicas))
  7.     }
  8.     rsc.enqueueReplicaSet(cur)
  9. }

至于 expectations 机制会在下文进行分析。

syncReplicaSet

syncReplicaSet 是 controller 的核心方法，它会驱动 controller 所控制的对象达到期望状态，主要逻辑如下所示：

  • 1、根据 ns/name 获取 rs 对象；
  • 2、调用 expectations.SatisfiedExpectations 判断是否需要执行真正的 sync 操作；
  • 3、获取所有 pod list；
  • 4、根据 pod label 进行过滤获取与该 rs 关联的 pod 列表，对于其中的孤儿 pod 若与该 rs label 匹配则进行关联，若已关联的 pod 与 rs label 不匹配则解除关联关系；
  • 5、调用 manageReplicas 进行同步 pod 操作，add/del pod；
  • 6、计算 rs 当前的 status 并进行更新；
  • 7、若 rs 设置了 MinReadySeconds 字段则将该 rs 加入到延迟队列中；

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:562

  1. func (rsc *ReplicaSetController) syncReplicaSet(key string) error {
  2.         ......
  4.     namespace, name, err := cache.SplitMetaNamespaceKey(key)
  5.     if err != nil {
  6.         return err
  7.     }
  9.     // 1、根据 ns/name 从 informer cache 中获取 rs 对象，
  10.     // 若 rs 已经被删除则直接删除 expectations 中的对象
  11.     rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)
  12.     if errors.IsNotFound(err) {
  13.         rsc.expectations.DeleteExpectations(key)
  14.         return nil
  15.     }
  16.     ......
  18.     // 2、判断该 rs 是否需要执行 sync 操作
  19.     rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
  20.     selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
  21.     if err != nil {
  22.         ......
  23.     }
  25.     // 3、获取所有 pod list
  26.     allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
  27.         ......
  29.     // 4、过滤掉异常 pod，处于删除状态或者 failed 状态的 pod 都为非 active 状态
  30.     filteredPods := controller.FilterActivePods(allPods)
  32.     // 5、检查所有 pod，根据 pod 并进行 adopt 与 release 操作，最后获取与该 rs 关联的 pod list
  33.     filteredPods, err = rsc.claimPods(rs, selector, filteredPods)
  34.     ......
  36.     // 6、若需要 sync 则执行 manageReplicas 创建/删除 pod
  37.     var manageReplicasErr error
  38.     if rsNeedsSync && rs.DeletionTimestamp == nil {
  39.         manageReplicasErr = rsc.manageReplicas(filteredPods, rs)
  40.     }
  41.     rs = rs.DeepCopy()
  42.     // 7、计算 rs 当前的 status
  43.     newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)
  45.     // 8、更新 rs status
  46.     updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)
  49.     // 9、判断是否需要将 rs 加入到延迟队列中
  50.     if manageReplicasErr == nil && updatedRS.Spec.MinReadySeconds > 0 &&
  51.         updatedRS.Status.ReadyReplicas == *(updatedRS.Spec.Replicas) &&
  52.         updatedRS.Status.AvailableReplicas != *(updatedRS.Spec.Replicas) {
  53.         rsc.enqueueReplicaSetAfter(updatedRS, time.Duration(updatedRS.Spec.MinReadySeconds)*time.Second)
  54.     }
  55.     return manageReplicasErr
  56. }

在 syncReplicaSet 方法中有几个重要的操作分别为：rsc.expectations.SatisfiedExpectations、rsc.manageReplicas、calculateStatus，下面一一进行分析。

SatisfiedExpectations

该方法主要判断 rs 是否需要执行真正的同步操作，若需要 add/del pod 或者 expectations 已过期则需要进行同步操作。

k8s.io/kubernetes/pkg/controller/controller_utils.go:181

  1. func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {
  2.     // 1、若该 key 存在时，判断是否满足条件或者是否超过同步周期
  3.     if exp, exists, err := r.GetExpectations(controllerKey); exists {
  4.         if exp.Fulfilled() {
  5.             return true
  6.         } else if exp.isExpired() {
  7.             return true
  8.         } else {
  9.             return false
  10.         }
  11.     } else if err != nil {
  12.         ......
  13.     } else {
  14.         // 2、该 rs 可能为新创建的，需要进行 sync
  15.         ......
  16.     }
  17.     return true
  18. }
  20. // 3、若 add <= 0 且 del <= 0 说明本地观察到的状态已经为期望状态了
  21. func (e *ControlleeExpectations) Fulfilled() bool {
  22.     return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0
  23. }
  25. // 4、判断 key 是否过期，ExpectationsTimeout 默认值为 5 * time.Minute
  26. func (exp *ControlleeExpectations) isExpired() bool {
  27.     return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout
  28. }
manageReplicas

manageReplicas 是最核心的方法，它会计算 replicaSet 需要创建或者删除多少个 pod 并调用 apiserver 的接口进行操作，在此阶段仅仅是调用 apiserver 的接口进行创建，并不保证 pod 成功运行，如果在某一轮，未能成功创建的所有 Pod 对象，则不再创建剩余的 pod。一个周期内最多只能创建或删除 500 个 pod，若超过上限值未创建完成的 pod 数会在下一个 syncLoop 继续进行处理。

该方法主要逻辑如下所示：

  • 1、计算已存在 pod 数与期望数的差异；
  • 2、如果 diff < 0 说明 rs 实际的 pod 数未达到期望值需要继续创建 pod，首先会将需要创建的 pod 数在 expectations 中进行记录，然后调用 slowStartBatch 创建所需要的 pod，slowStartBatch 以指数级增长的方式批量创建 pod，创建 pod 过程中若出现 timeout err 则忽略，若为其他 err 则终止创建操作并更新 expectations；
  • 3、如果 diff > 0 说明可能是一次缩容操作需要删除多余的 pod，如果需要删除全部的 pod 则直接进行删除，否则会通过 getPodsToDelete 方法筛选出需要删除的 pod，具体的筛选策略在下文会将到，然后并发删除这些 pod，对于删除失败操作也会记录在 expectations 中；

在 slowStartBatch 中会调用 rsc.podControl.CreatePodsWithControllerRef 方法创建 pod，若创建 pod 失败会判断是否为创建超时错误，或者可能是超时后失败，但此时认为超时并不影响后续的批量创建动作，大家知道，创建 pod 操作提交到 apiserver 后会经过认证、鉴权、以及动态访问控制三个步骤，此过程有可能会超时，即使真的创建失败了，等到 expectations 过期后在下一个 syncLoop 时会重新创建。

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:459

  1. func (rsc *ReplicaSetController) manageReplicas(......) error {
  2.     // 1、计算已存在 pod 数与期望数的差异
  3.     diff := len(filteredPods) - int(*(rs.Spec.Replicas))
  4.     rsKey, err := controller.KeyFunc(rs)
  5.     if err != nil {
  6.         ......
  7.     }
  8.     2、如果 <0，则需要创建 pod
  9.     if diff < 0 {
  10.         diff *= -1
  11.         3、判断需要创建的 pod 数是否超过单次 sync 上限值 500
  12.         if diff > rsc.burstReplicas {
  13.             diff = rsc.burstReplicas
  14.         }
  16.         4、在 expectations 中进行记录，若该 key 已经存在会进行覆盖
  17.         rsc.expectations.ExpectCreations(rsKey, diff)
  19.         5、调用 slowStartBatch 创建所需要的 pod
  20.         successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {
  21.             err := rsc.podControl.CreatePodsWithControllerRef(rs.Namespace, &rs.Spec.Template, rs, metav1.NewControllerRef(rs, rsc.GroupVersionKind))
  22.             // 6、若为 timeout err 则忽略
  23.             if err != nil && errors.IsTimeout(err) {
  24.                 return nil
  25.             }
  26.             return err
  27.         })
  29.         // 7、计算未创建的 pod 数，并记录在 expectations 中
  30.         // 若 pod 创建成功，informer watch 到事件后会在 addPod handler 中更新 expectations
  31.         if skippedPods := diff - successfulCreations; skippedPods > 0 {
  32.             for i := 0; i < skippedPods; i++ {
  33.                 rsc.expectations.CreationObserved(rsKey)
  34.             }
  35.         }
  36.         return err
  37.     } else if diff > 0 {
  38.         // 8、若 diff >0 说明需要删除多创建的 pod
  39.         if diff > rsc.burstReplicas {
  40.             diff = rsc.burstReplicas
  41.         }
  43.         // 9、getPodsToDelete 会按照一定的策略找出需要删除的 pod 列表
  44.         podsToDelete := getPodsToDelete(filteredPods, diff)
  46.         // 10、在 expectations 中进行记录，若该 key 已经存在会进行覆盖
  47.         rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))
  49.         // 11、进行并发删除的操作
  50.         errCh := make(chan error, diff)
  51.         var wg sync.WaitGroup
  52.         wg.Add(diff)
  53.         for _, pod := range podsToDelete {
  54.             go func(targetPod *v1.Pod) {
  55.                 defer wg.Done()
  56.                 if err := rsc.podControl.DeletePod(rs.Namespace, targetPod.Name, rs); err != nil {
  57.                     podKey := controller.PodKey(targetPod)
  58.                     // 12、某次删除操作若失败会记录在 expectations 中
  59.                     rsc.expectations.DeletionObserved(rsKey, podKey)
  60.                     errCh <- err
  61.                 }
  62.             }(pod)
  63.         }
  64.         wg.Wait()
  66.         // 13、返回其中一条 err
  67.         select {
  68.         case err := <-errCh:
  69.             if err != nil {
  70.                 return err
  71.             }
  72.         default:
  73.         }
  74.     }
  76.     return nil
  77. }

slowStartBatch 会批量创建出已计算出的 diff pod 数，创建的 pod 数依次为 1、2、4、8……，呈指数级增长，其方法如下所示：

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:658

  1. func slowStartBatch(count int, initialBatchSize int, fn func() error) (int, error) {
  2.     remaining := count
  3.     successes := 0
  4.     for batchSize := integer.IntMin(remaining, initialBatchSize); batchSize > 0; batchSize = integer.IntMin(2*batchSize, remaining) {
  5.         errCh := make(chan error, batchSize)
  6.         var wg sync.WaitGroup
  7.         wg.Add(batchSize)
  8.         for i := 0; i < batchSize; i++ {
  9.             go func() {
  10.                 defer wg.Done()
  11.                 if err := fn(); err != nil {
  12.                     errCh <- err
  13.                 }
  14.             }()
  15.         }
  16.         wg.Wait()
  17.         curSuccesses := batchSize - len(errCh)
  18.         successes += curSuccesses
  19.         if len(errCh) > 0 {
  20.             return successes, <-errCh
  21.         }
  22.         remaining -= batchSize
  23.     }
  24.     return successes, nil
  25. }

若 diff > 0 时再删除 pod 阶段会调用getPodsToDelete 对 pod 进行筛选操作，此阶段会选出最劣质的 pod，下面是用到的 6 种筛选方法：

  • 1、判断是否绑定了 node：Unassigned < assigned；
  • 2、判断 pod phase：PodPending < PodUnknown < PodRunning；
  • 3、判断 pod 状态：Not ready < ready；
  • 4、若 pod 都为 ready，则按运行时间排序，运行时间最短会被删除：empty time < less time < more time；
  • 5、根据 pod 重启次数排序：higher restart counts < lower restart counts；
  • 6、按 pod 创建时间进行排序：Empty creation time pods < newer pods < older pods；

上面的几个排序规则遵循互斥原则，从上到下进行匹配，符合条件则排序完成，代码如下所示：

k8s.io/kubernetes/pkg/controller/replicaset/replica_set.go:684

  1. func getPodsToDelete(filteredPods []*v1.Pod, diff int) []*v1.Pod {
  2.     if diff < len(filteredPods) {
  3.         sort.Sort(controller.ActivePods(filteredPods))
  4.     }
  5.     return filteredPods[:diff]
  6. }

k8s.io/kubernetes/pkg/controller/controller_utils.go:735

  1. type ActivePods []*v1.Pod
  3. func (s ActivePods) Len() int      { return len(s) }
  4. func (s ActivePods) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
  6. func (s ActivePods) Less(i, j int) bool {
  7.     // 1. Unassigned < assigned
  8.     if s[i].Spec.NodeName != s[j].Spec.NodeName && (len(s[i].Spec.NodeName) == 0 || len(s[j].Spec.NodeName) == 0) {
  9.         return len(s[i].Spec.NodeName) == 0
  10.     }
  12.     // 2. PodPending < PodUnknown < PodRunning
  13.     m := map[v1.PodPhase]int{v1.PodPending: 0, v1.PodUnknown: 1, v1.PodRunning: 2}
  14.     if m[s[i].Status.Phase] != m[s[j].Status.Phase] {
  15.         return m[s[i].Status.Phase] < m[s[j].Status.Phase]
  16.     }
  18.     // 3. Not ready < ready
  19.     if podutil.IsPodReady(s[i]) != podutil.IsPodReady(s[j]) {
  20.         return !podutil.IsPodReady(s[i])
  21.     }
  23.     // 4. Been ready for empty time < less time < more time
  24.     if podutil.IsPodReady(s[i]) && podutil.IsPodReady(s[j]) && !podReadyTime(s[i]).Equal(podReadyTime(s[j])) {
  25.         return afterOrZero(podReadyTime(s[i]), podReadyTime(s[j]))
  26.     }
  28.     // 5. Pods with containers with higher restart counts < lower restart counts
  29.     if maxContainerRestarts(s[i]) != maxContainerRestarts(s[j]) {
  30.         return maxContainerRestarts(s[i]) > maxContainerRestarts(s[j])
  31.     }
  33.     // 6. Empty creation time pods < newer pods < older pods
  34.     if !s[i].CreationTimestamp.Equal(&s[j].CreationTimestamp) {
  35.         return afterOrZero(&s[i].CreationTimestamp, &s[j].CreationTimestamp)
  36.     }
  37.     return false
  38. }
calculateStatus

calculateStatus 会通过当前 pod 的状态计算出 rs 中 status 字段值，status 字段如下所示：

  1. status:
  2.   availableReplicas: 10
  3.   fullyLabeledReplicas: 10
  4.   observedGeneration: 1
  5.   readyReplicas: 10
  6.   replicas: 10

k8s.io/kubernetes/pkg/controller/replicaset/replica_set_utils.go:85

  1. func calculateStatus(......) apps.ReplicaSetStatus {
  2.     newStatus := rs.Status
  3.     fullyLabeledReplicasCount := 0
  4.     readyReplicasCount := 0
  5.     availableReplicasCount := 0
  6.     templateLabel := labels.Set(rs.Spec.Template.Labels).AsSelectorPreValidated()
  7.     for _, pod := range filteredPods {
  8.         if templateLabel.Matches(labels.Set(pod.Labels)) {
  9.             fullyLabeledReplicasCount++
  10.         }
  11.         if podutil.IsPodReady(pod) {
  12.             readyReplicasCount++
  13.             if podutil.IsPodAvailable(pod, rs.Spec.MinReadySeconds, metav1.Now()) {
  14.                 availableReplicasCount++
  15.             }
  16.         }
  17.     }
  19.     failureCond := GetCondition(rs.Status, apps.ReplicaSetReplicaFailure)
  20.     if manageReplicasErr != nil && failureCond == nil {
  21.         var reason string
  22.         if diff := len(filteredPods) - int(*(rs.Spec.Replicas)); diff < 0 {
  23.             reason = "FailedCreate"
  24.         } else if diff > 0 {
  25.             reason = "FailedDelete"
  26.         }
  27.         cond := NewReplicaSetCondition(apps.ReplicaSetReplicaFailure, v1.ConditionTrue, reason, manageReplicasErr.Error())
  28.         SetCondition(&newStatus, cond)
  29.     } else if manageReplicasErr == nil && failureCond != nil {
  30.         RemoveCondition(&newStatus, apps.ReplicaSetReplicaFailure)
  31.     }
  33.     newStatus.Replicas = int32(len(filteredPods))
  34.     newStatus.FullyLabeledReplicas = int32(fullyLabeledReplicasCount)
  35.     newStatus.ReadyReplicas = int32(readyReplicasCount)
  36.     newStatus.AvailableReplicas = int32(availableReplicasCount)
  37.     return newStatus
  38. }

expectations 机制

通过上面的分析可知，在 rs 每次入队后进行 sync 操作时，首先需要判断该 rs 是否满足 expectations 机制，那么这个 expectations 的目的是什么？其实，rs 除了有 informer 的缓存外，还有一个本地缓存就是 expectations，expectations 会记录 rs 所有对象需要 add/del 的 pod 数量，若两者都为 0 则说明该 rs 所期望创建的 pod 或者删除的 pod 数已经被满足，若不满足则说明某次在 syncLoop 中创建或者删除 pod 时有失败的操作，则需要等待 expectations 过期后再次同步该 rs。

通过上面对 eventHandler 的分析，再来总结一下触发 replicaSet 对象发生同步事件的条件：

  • 1、与 rs 相关的：AddRS、UpdateRS、DeleteRS；
  • 2、与 pod 相关的：AddPod、UpdatePod、DeletePod；
  • 3、informer 二级缓存的同步；

但是所有的更新事件是否都需要执行 sync 操作？对于除 rs.Spec.Replicas 之外的更新操作其实都没必要执行 sync 操作，因为 spec 其他字段和 status 的更新都不需要创建或者删除 pod。

在 sync 操作真正开始之前，依据 expectations 机制进行判断，确定是否要真正地启动一次 sync，因为在 eventHandler 阶段也会更新 expectations 值，从上面的 eventHandler 中可以看到在 addPod 中会调用 rsc.expectations.CreationObserved 更新 rsKey 的 expectations，将其 add 值 -1，在 deletePod 中调用 rsc.expectations.DeletionObserved 将其 del 值 -1。所以等到 sync 时，若 controllerKey(name 或者 ns/name)满足 expectations 机制则进行 sync 操作，而 updatePod 并不会修改 expectations，所以，expectations 的设计就是当需要创建或删除 pod 才会触发对应的 sync 操作，expectations 机制的目的就是减少不必要的 sync 操作。

什么条件下 expectations 机制会满足？

  • 1、当 expectations 中不存在 rsKey 时，也就说首次创建 rs 时；
  • 2、当 expectations 中 del 以及 add 值都为 0 时，即 rs 所需要创建或者删除的 pod 数都已满足；
  • 3、当 expectations 过期时，即超过 5 分钟未进行 sync 操作；

最后再看一下 expectations 中用到的几个方法：

  1.  // 创建了一个 pod 说明 expectations 中对应的 key add 期望值需要减少一个 pod， add -1
  2.  CreationObserved(controllerKey string)
  4.  // 删除了一个 pod 说明 expectations 中对应的 key del 期望值需要减少一个 pod， del - 1
  5.  DeletionObserved(controllerKey string)
  7.  // 写入 key 需要 add 的 pod 数量
  8.  ExpectCreations(controllerKey string, adds int) error
  10.  // 写入 key 需要 del 的 pod 数量
  11.  ExpectDeletions(controllerKey string, dels int) error
  13.  // 删除该 key
  14.  DeleteExpectations(controllerKey string)

当在 syncLoop 中发现满足条件时，会执行 manageReplicas 方法，在 manageReplicas 中无论是为 rs 创建还是删除 pod 都会调用 ExpectCreations 和 ExpectDeletions 为 rsKey 创建 expectations 对象。

总结

本文主要从源码层面分析了 replicaSetController 的设计与实现，但是不得不说其在设计方面考虑了很多因素，文中只提到了笔者理解了或者思考后稍有了解的一些机制，至于其他设计思想还得自行阅读代码体会。

下面以一个流程图总结下创建 rs 的主要流程。

  1.                                     SatisfiedExpectations
  2.                                     (expectations 中不存在
  3.                                      rsKey，rsNeedsSync
  4.                                      为 true)
  5.                                               |              判断 add/del pod
  6.                                               |                     |
  7.                                               |                     ∨
  8.                                               |             创建 expectations 对象,
  9.                                               |             并设置 add/del 值
  10.                                               ∨                     |
  11. create rs --> syncReplicaSet -->       manageReplicas  -->          ∨
  12.                                        (为 rs 创建 pod)       调用 slowStartBatch 批量创建 pod/
  13.                                               |               删除筛选出的多余 pod
  14.                                               |                     |
  15.                                               |                     ∨
  16.                                               |               更新 expectations 对象
  17.                                               ∨
  18.                                     updateReplicaSetStatus
  19.                                     (更新 rs 的 status
  20.                                     subResource)

参考：

https://keyla.vip/k8s/3-master/controller/replica-set/