PS：源码会将一部分日志/监控等代码删除，周知

0、架构图

图片 1.png

1、replicaset成员变量

  1. type ReplicaSetController struct {
  2.     // 此controller控制的资源类型(通过gvk标识)
  3.     schema.GroupVersionKind
  5.     // k8s api客户端
  6.     kubeClient clientset.Interface
  8.     // pod控制器(负责pod的增删改)
  9.     podControl    controller.PodControlInterface
  11.     // 单次最大创建的pod个数
  12.     burstReplicas int
  14.     // 真正的同步函数
  15.     syncHandler func(rsKey string) error
  17.     // 控制rs是否需要sync，相当于rs的二级缓存
  18.     expectations *controller.UIDTrackingControllerExpectations
  20.     // rs查询接口
  21.     rsLister appslisters.ReplicaSetLister
  22.     // 是否rs已经同步过一次
  23.     rsListerSynced cache.InformerSynced
  25.     // pod的查询接口
  26.     podLister corelisters.PodLister
  27.     // 是否pod已经同步过一次
  28.     podListerSynced cache.InformerSynced
  30.     // 工作队列
  31.     queue workqueue.RateLimitingInterface
  32. }

2、主循环

  1. // controller启动函数
  2. // 同时开启workers个协程进行处理，不停的从workqueue读取发生变化的key，
  3. // 然后进行相应的处理
  4. func (rsc *ReplicaSetController) Run(workers int, stopCh <-chan struct{}) {
  5.     defer utilruntime.HandleCrash()
  6.     defer rsc.queue.ShutDown()
  8.     if !cache.WaitForNamedCacheSync(rsc.Kind, stopCh, rsc.podListerSynced, rsc.rsListerSynced) {
  9.         return
  10.     }
  12.     for i := 0; i < workers; i++ {
  13.         go wait.Until(rsc.worker, time.Second, stopCh)
  14.     }
  16.     <-stopCh
  17. }
  19. func (rsc *ReplicaSetController) worker() {
  20.     for rsc.processNextWorkItem() {
  21.     }
  22. }
  24. func (rsc *ReplicaSetController) processNextWorkItem() bool {
  25.     key, quit := rsc.queue.Get()
  26.     if quit {
  27.         return false
  28.     }
  29.     defer rsc.queue.Done(key)
  31.     err := rsc.syncHandler(key.(string))
  32.     if err == nil {
  33.         rsc.queue.Forget(key)
  34.         return true
  35.     }
  37.     utilruntime.HandleError(fmt.Errorf("sync %q failed with %v", key, err))
  38.     rsc.queue.AddRateLimited(key)
  40.     return true
  41. }

3、replicaset事件回调函数

1、addRS

  1. // 监听到rs的增加事件，直接将对应的key放入到workqueue即可
  2. func (rsc *ReplicaSetController) addRS(obj interface{}) {
  3.     rs := obj.(*apps.ReplicaSet)
  4.     rsc.enqueueRS(rs)
  5. }

2、updateRS

  1. func (rsc *ReplicaSetController) updateRS(old, cur interface{}) {
  2.     oldRS := old.(*apps.ReplicaSet)
  3.     curRS := cur.(*apps.ReplicaSet)
  5.     // oldRs删除事件丢失
  6.     if curRS.UID != oldRS.UID {
  7.         key, err := controller.KeyFunc(oldRS)
  8.         if err != nil {
  9.             utilruntime.HandleError(fmt.Errorf("couldn't get key for object %#v: %v", oldRS, err))
  10.             return
  11.         }
  12.         rsc.deleteRS(cache.DeletedFinalStateUnknown{
  13.             Key: key,
  14.             Obj: oldRS,
  15.         })
  16.     }
  17.     rsc.enqueueRS(curRS)
  18. }

3、deleteRS

  1. func (rsc *ReplicaSetController) deleteRS(obj interface{}) {
  2.     rs, ok := obj.(*apps.ReplicaSet)
  3.     if !ok {
  4.         // 若rs的删除时间丢失，会有updateRS封装oldObj为DeletedFinalStateUnknown并调用deleteRS
  5.         tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
  6.         if !ok {
  7.             utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
  8.             return
  9.         }
  10.         rs, ok = tombstone.Obj.(*apps.ReplicaSet)
  11.         if !ok {
  12.             utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a ReplicaSet %#v", obj))
  13.             return
  14.         }
  15.     }
  16.     key, err := controller.KeyFunc(rs)
  17.     if err != nil {
  18.         utilruntime.HandleError(fmt.Errorf("couldn't get key for object %#v: %v", rs, err))
  19.         return
  20.     }
  21.     rsc.expectations.DeleteExpectations(key)
  22.     rsc.queue.Add(key)
  23. }

4、addPod

  1. func (rsc *ReplicaSetController) addPod(obj interface{}) {
  2.     pod := obj.(*v1.Pod)
  3.     if pod.DeletionTimestamp != nil {
  4.         // 不太清楚为什么会走有这种情况
  5.         rsc.deletePod(pod)
  6.         return
  7.     }
  9.     // 获取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.         rsc.expectations.CreationObserved(rsKey)
  20.         rsc.queue.Add(rsKey)
  21.         return
  22.     }
  24.     rss := rsc.getPodReplicaSets(pod)
  25.     if len(rss) == 0 {
  26.         return
  27.     }
  28.     for _, rs := range rss {
  29.         rsc.enqueueRS(rs)
  30.     }
  31. }

5、updatePod

  1. func (rsc *ReplicaSetController) updatePod(old, cur interface{}) {
  2.     curPod := cur.(*v1.Pod)
  3.     oldPod := old.(*v1.Pod)
  4.     // 此处是resync触发的onUpdate事件，若版本号一致，则无须变更
  5.     // rs无须担心pod update丢失的事件，因为在后续的同步中，rs会认领pod
  6.     if curPod.ResourceVersion == oldPod.ResourceVersion {
  7.         return
  8.     }
  10.     labelChanged := !reflect.DeepEqual(curPod.Labels, oldPod.Labels)
  11.     // 用户手动设置删除时间戳，未调用api进行删除，出现此种情况
  12.     if curPod.DeletionTimestamp != nil {
  13.         rsc.deletePod(curPod)
  14.         // lable发生变更，需要删除旧版本pod，因为label变更会导致匹配到不同的rs，需要将rs都通知到
  15.         // 此处不会触发两次DeletionObserved，DeletionObserved函数会保证只删除一次
  16.         if labelChanged {
  17.             rsc.deletePod(oldPod)
  18.         }
  19.         return
  20.     }
  22.     curControllerRef := metav1.GetControllerOf(curPod)
  23.     oldControllerRef := metav1.GetControllerOf(oldPod)
  24.     controllerRefChanged := !reflect.DeepEqual(curControllerRef, oldControllerRef)
  25.     // 到此处说明新pod的变更非删除，倘若发生了ownerRef变更，且旧pod ownerRef不为空
  26.     // 需要将旧pod绑定的rs入队
  27.     if controllerRefChanged && oldControllerRef != nil {
  28.         if rs := rsc.resolveControllerRef(oldPod.Namespace, oldControllerRef); rs != nil {
  29.             rsc.enqueueRS(rs)
  30.         }
  31.     }
  33.     // 进入如下条件，需要如下情况
  34.     // 1、controllerRefChanged为true，oldControllerRef == nil
  35.     // 2、controllerRefChanged为false，oldControllerRef != nil
  36.     // 3、controllerRefChanged为true，oldControllerRef != nil并且curControllerRef != nil
  37.     if curControllerRef != nil {
  38.         rs := rsc.resolveControllerRef(curPod.Namespace, curControllerRef)
  39.         if rs == nil {
  40.             // 若新pod未找到对应的rs，直接return
  41.             return
  42.         }
  43.         rsc.enqueueRS(rs)
  44.         // 因为pod从ready到available状态需要时间
  45.         if !podutil.IsPodReady(oldPod) && podutil.IsPodReady(curPod) && rs.Spec.MinReadySeconds > 0 {
  46.             rsc.enqueueRSAfter(rs, (time.Duration(rs.Spec.MinReadySeconds)*time.Second)+time.Second)
  47.         }
  48.         return
  49.     }
  51.     // 逻辑走到这里需要如下情况
  52.     // 1、controllerRefChanged为false，oldControllerRef == nil，则newControllerRef == nil即为只是label发生了变化
  53.     // 2、controllerRefChanged为true，oldControllerRef != nil并且curControllerRef == nil即为controllerRefChanged
  54.     // 由于384行已经将旧pod对应rs入队，则此处无须入队，只需要将新pod对应rd入队
  55.     if labelChanged || controllerRefChanged {
  56.         rss := rsc.getPodReplicaSets(curPod)
  57.         if len(rss) == 0 {
  58.             return
  59.         }
  60.         for _, rs := range rss {
  61.             rsc.enqueueRS(rs)
  62.         }
  63.     }
  64. }

6、deletePod

  1. func (rsc *ReplicaSetController) deletePod(obj interface{}) {
  2.     pod, ok := obj.(*v1.Pod)
  3.     if !ok {
  4.         tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
  5.         if !ok {
  6.             utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %+v", obj))
  7.             return
  8.         }
  9.         pod, ok = tombstone.Obj.(*v1.Pod)
  10.         if !ok {
  11.             utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a pod %#v", obj))
  12.             return
  13.         }
  14.     }
  15.     // 孤儿Pod不会进行Expectation处理
  16.     controllerRef := metav1.GetControllerOf(pod)
  17.     if controllerRef == nil {
  18.         return
  19.     }
  20.     rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)
  21.     if rs == nil {
  22.         return
  23.     }
  24.     rsKey, err := controller.KeyFunc(rs)
  25.     if err != nil {
  26.         utilruntime.HandleError(fmt.Errorf("couldn't get key for object %#v: %v", rs, err))
  27.         return
  28.     }
  29.     rsc.expectations.DeletionObserved(rsKey, controller.PodKey(pod))
  30.     rsc.queue.Add(rsKey)
  31. }

4、replicaset同步函数

1、切割key获取namespace、name
2、查询informer获取相应的rs对象。若没有找到，则已经被删除，删除expectations里面的rs
3、根据expectations判断是否需要进行同步并转换selector
以下条件会进行sync
a. key未找到
b. key达到fulfilled状态
c. key已经过期
4、获取rs对应ns所有的pod，并过滤不健康的pod(不健康pod为已经成功结束/已经失败/处在删除状态)
5、认领pod/释放相应的pod
6、若需要同步并且rs不处于删除状态，进行sync

  1. func (rsc *ReplicaSetController) syncReplicaSet(key string) error {
  2.     namespace, name, err := cache.SplitMetaNamespaceKey(key)
  3.     if err != nil {
  4.         return err
  5.     }
  6.     rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)
  7.     if errors.IsNotFound(err) {
  8.         rsc.expectations.DeleteExpectations(key)
  9.         return nil
  10.     }
  11.     if err != nil {
  12.         return err
  13.     }
  15.     rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
  16.     selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
  17.     if err != nil {
  18.         utilruntime.HandleError(fmt.Errorf("error converting pod selector to selector: %v", err))
  19.         return nil
  20.     }
  22.     allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
  23.     if err != nil {
  24.         return err
  25.     }
  26.     filteredPods := controller.FilterActivePods(allPods)
  27.     filteredPods, err = rsc.claimPods(rs, selector, filteredPods)
  28.     if err != nil {
  29.         return err
  30.     }
  32.     var manageReplicasErr error
  33.     if rsNeedsSync && rs.DeletionTimestamp == nil {
  34.         manageReplicasErr = rsc.manageReplicas(filteredPods, rs)
  35.     }
  36.     rs = rs.DeepCopy()
  37.     newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)
  39.     updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)
  40.     if err != nil {
  41.         return err
  42.     }
  43.     if manageReplicasErr == nil && updatedRS.Spec.MinReadySeconds > 0 &&
  44.         updatedRS.Status.ReadyReplicas == *(updatedRS.Spec.Replicas) &&
  45.         updatedRS.Status.AvailableReplicas != *(updatedRS.Spec.Replicas) {
  46.         rsc.queue.AddAfter(key, time.Duration(updatedRS.Spec.MinReadySeconds)*time.Second)
  47.     }
  48.     return manageReplicasErr
  49. }

5、replicaset工具函数

  1. func (rsc *ReplicaSetController) manageReplicas(filteredPods []*v1.Pod, rs *apps.ReplicaSet) error {
  2.     diff := len(filteredPods) - int(*(rs.Spec.Replicas))
  3.     rsKey, err := controller.KeyFunc(rs)
  4.     if err != nil {
  5.         utilruntime.HandleError(fmt.Errorf("couldn't get key for %v %#v: %v", rsc.Kind, rs, err))
  6.         return nil
  7.     }
  8.     if diff < 0 {
  9.         diff *= -1
  10.         if diff > rsc.burstReplicas {
  11.             diff = rsc.burstReplicas
  12.         }
  13.         rsc.expectations.ExpectCreations(rsKey, diff)
  14.         successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {
  15.             err := rsc.podControl.CreatePodsWithControllerRef(rs.Namespace, &rs.Spec.Template, rs, metav1.NewControllerRef(rs, rsc.GroupVersionKind))
  16.             if err != nil {
  17.                 if errors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
  18.                     return nil
  19.                 }
  20.             }
  21.             return err
  22.         })
  23.         if skippedPods := diff - successfulCreations; skippedPods > 0 {
  24.             for i := 0; i < skippedPods; i++ {
  25.                 rsc.expectations.CreationObserved(rsKey)
  26.             }
  27.         }
  28.         return err
  29.     } else if diff > 0 {
  30.         if diff > rsc.burstReplicas {
  31.             diff = rsc.burstReplicas
  32.         }
  33.         relatedPods, err := rsc.getIndirectlyRelatedPods(rs)
  34.         utilruntime.HandleError(err)
  35.         podsToDelete := getPodsToDelete(filteredPods, relatedPods, diff)
  37.         rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))
  39.         errCh := make(chan error, diff)
  40.         var wg sync.WaitGroup
  41.         wg.Add(diff)
  42.         for _, pod := range podsToDelete {
  43.             go func(targetPod *v1.Pod) {
  44.                 defer wg.Done()
  45.                 if err := rsc.podControl.DeletePod(rs.Namespace, targetPod.Name, rs); err != nil {
  46.                     podKey := controller.PodKey(targetPod)
  47.                     rsc.expectations.DeletionObserved(rsKey, podKey)
  48.                     errCh <- err
  49.                 }
  50.             }(pod)
  51.         }
  52.         wg.Wait()
  54.         select {
  55.         case err := <-errCh:
  56.             if err != nil {
  57.                 return err
  58.             }
  59.         default:
  60.         }
  61.     }
  62.     return nil
  63. }
  65. func (rsc *ReplicaSetController) claimPods(rs *apps.ReplicaSet, selector labels.Selector, filteredPods []*v1.Pod) ([]*v1.Pod, error) {
  66.     canAdoptFunc := controller.RecheckDeletionTimestamp(func() (metav1.Object, error) {
  67.         fresh, err := rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace).Get(context.TODO(), rs.Name, metav1.GetOptions{})
  68.         if err != nil {
  69.             return nil, err
  70.         }
  71.         if fresh.UID != rs.UID {
  72.             return nil, fmt.Errorf("original %v %v/%v is gone: got uid %v, wanted %v", rsc.Kind, rs.Namespace, rs.Name, fresh.UID, rs.UID)
  73.         }
  74.         return fresh, nil
  75.     })
  76.     cm := controller.NewPodControllerRefManager(rsc.podControl, rs, selector, rsc.GroupVersionKind, canAdoptFunc)
  77.     return cm.ClaimPods(filteredPods)
  78. }
  80. // 获取与给定rs有相同owner的rs集合
  81. func (rsc *ReplicaSetController) getReplicaSetsWithSameController(rs *apps.ReplicaSet) []*apps.ReplicaSet {
  82.     controllerRef := metav1.GetControllerOf(rs)
  83.     if controllerRef == nil {
  84.         utilruntime.HandleError(fmt.Errorf("ReplicaSet has no controller: %v", rs))
  85.         return nil
  86.     }
  88.     allRSs, err := rsc.rsLister.ReplicaSets(rs.Namespace).List(labels.Everything())
  89.     if err != nil {
  90.         utilruntime.HandleError(err)
  91.         return nil
  92.     }
  94.     var relatedRSs []*apps.ReplicaSet
  95.     for _, r := range allRSs {
  96.         if ref := metav1.GetControllerOf(r); ref != nil && ref.UID == controllerRef.UID {
  97.             relatedRSs = append(relatedRSs, r)
  98.         }
  99.     }
  100.     return relatedRSs
  101. }
  103. // 获取关联给定pod的所有rs(通过label)
  104. func (rsc *ReplicaSetController) getPodReplicaSets(pod *v1.Pod) []*apps.ReplicaSet {
  105.     rss, err := rsc.rsLister.GetPodReplicaSets(pod)
  106.     if err != nil {
  107.         return nil
  108.     }
  109.     if len(rss) > 1 {
  110.         utilruntime.HandleError(fmt.Errorf("user error! more than one %v is selecting pods with labels: %+v", rsc.Kind, pod.Labels))
  111.     }
  112.     return rss
  113. }
  115. // 通过给定的引用信息找到rs
  116. func (rsc *ReplicaSetController) resolveControllerRef(namespace string, controllerRef *metav1.OwnerReference) *apps.ReplicaSet {
  117.     if controllerRef.Kind != rsc.Kind {
  118.         return nil
  119.     }
  120.     rs, err := rsc.rsLister.ReplicaSets(namespace).Get(controllerRef.Name)
  121.     if err != nil {
  122.         return nil
  123.     }
  124.     // 若UID不相同，则rs已经被重建，但是pod还未被删除
  125.     if rs.UID != controllerRef.UID {
  126.         return nil
  127.     }
  128.     return rs
  129. }
  131. func (rsc *ReplicaSetController) getIndirectlyRelatedPods(rs *apps.ReplicaSet) ([]*v1.Pod, error) {
  132.     var relatedPods []*v1.Pod
  133.     seen := make(map[types.UID]*apps.ReplicaSet)
  134.     for _, relatedRS := range rsc.getReplicaSetsWithSameController(rs) {
  135.         selector, err := metav1.LabelSelectorAsSelector(relatedRS.Spec.Selector)
  136.         if err != nil {
  137.             return nil, err
  138.         }
  139.         pods, err := rsc.podLister.Pods(relatedRS.Namespace).List(selector)
  140.         if err != nil {
  141.             return nil, err
  142.         }
  143.         for _, pod := range pods {
  144.             if otherRS, found := seen[pod.UID]; found {
  145.                 klog.V(5).Infof("Pod %s/%s is owned by both %v %s/%s and %v %s/%s", pod.Namespace, pod.Name, rsc.Kind, otherRS.Namespace, otherRS.Name, rsc.Kind, relatedRS.Namespace, relatedRS.Name)
  146.                 continue
  147.             }
  148.             seen[pod.UID] = relatedRS
  149.             relatedPods = append(relatedPods, pod)
  150.         }
  151.     }
  152.     return relatedPods, nil
  153. }
  155. // 指数级创建pod
  156. func slowStartBatch(count int, initialBatchSize int, fn func() error) (int, error) {
  157.     remaining := count
  158.     successes := 0
  159.     for batchSize := integer.IntMin(remaining, initialBatchSize); batchSize > 0; batchSize = integer.IntMin(2*batchSize, remaining) {
  160.         errCh := make(chan error, batchSize)
  161.         var wg sync.WaitGroup
  162.         wg.Add(batchSize)
  163.         for i := 0; i < batchSize; i++ {
  164.             go func() {
  165.                 defer wg.Done()
  166.                 if err := fn(); err != nil {
  167.                     errCh <- err
  168.                 }
  169.             }()
  170.         }
  171.         wg.Wait()
  172.         curSuccesses := batchSize - len(errCh)
  173.         successes += curSuccesses
  174.         if len(errCh) > 0 {
  175.             return successes, <-errCh
  176.         }
  177.         remaining -= batchSize
  178.     }
  179.     return successes, nil
  180. }
  182. func getPodsToDelete(filteredPods, relatedPods []*v1.Pod, diff int) []*v1.Pod {
  183.     if diff < len(filteredPods) {
  184.         podsWithRanks := getPodsRankedByRelatedPodsOnSameNode(filteredPods, relatedPods)
  185.         sort.Sort(podsWithRanks)
  186.     }
  187.     return filteredPods[:diff]
  188. }
  190. func getPodsRankedByRelatedPodsOnSameNode(podsToRank, relatedPods []*v1.Pod) controller.ActivePodsWithRanks {
  191.     podsOnNode := make(map[string]int)
  192.     for _, pod := range relatedPods {
  193.         if controller.IsPodActive(pod) {
  194.             podsOnNode[pod.Spec.NodeName]++
  195.         }
  196.     }
  197.     ranks := make([]int, len(podsToRank))
  198.     for i, pod := range podsToRank {
  199.         ranks[i] = podsOnNode[pod.Spec.NodeName]
  200.     }
  201.     return controller.ActivePodsWithRanks{Pods: podsToRank, Rank: ranks}
  202. }
  204. func getPodKeys(pods []*v1.Pod) []string {
  205.     podKeys := make([]string, 0, len(pods))
  206.     for _, pod := range pods {
  207.         podKeys = append(podKeys, controller.PodKey(pod))
  208.     }
  209.     return podKeys
  210. }