官方文档：https://openkruise.io/zh-cn/docs/cloneset.html

0、前置概念

0.1、readinessGates

https://kubernetes.io/docs/concepts/workloads/pods/pod-lifecycle/#pod-readiness-gate

1、Spec定义

  1. type CloneSetSpec struct {
  2.     // 副本数量，默认为1
  3.     Replicas *int32 `json:"replicas,omitempty"`
  5.     Selector *metav1.LabelSelector `json:"selector"`
  7.     // pod模板
  8.     Template v1.PodTemplateSpec `json:"template"`
  10.     // 存储模板
  11.     VolumeClaimTemplates []v1.PersistentVolumeClaim `json:"volumeClaimTemplates,omitempty"`
  13.     // 扩/缩容策略
  14.     ScaleStrategy CloneSetScaleStrategy `json:"scaleStrategy,omitempty"`
  16.     // 更新策略
  17.     UpdateStrategy CloneSetUpdateStrategy `json:"updateStrategy,omitempty"`
  19.     // 历史版本保留数量，默认为10
  20.     RevisionHistoryLimit *int32 `json:"revisionHistoryLimit,omitempty"`
  22.     // pod从ready状态持续该时间，才认为pod是可用状态
  23.     // 默认为0
  24.     MinReadySeconds int32 `json:"minReadySeconds,omitempty"`
  26.     // 删除和原地升级自定义hook
  27.     Lifecycle *Lifecycle `json:"lifecycle,omitempty"`
  28. }
  30. type CloneSetScaleStrategy struct {
  31.     // 缩容的时候指定pod进行删除
  32.     // 若数量没有发生变化，则删除再创建
  33.     PodsToDelete []string `json:"podsToDelete,omitempty"`
  34. }
  36. // 升级策略
  37. type CloneSetUpdateStrategy struct {
  38.     // 升级策略
  39.     // ReCreate、InPlaceIfPossible、InPlaceOnly
  40.     Type CloneSetUpdateStrategyType `json:"type,omitempty"`
  42.     // 保留旧版本的数量，默认为0
  43.     Partition *int32 `json:"partition,omitempty"`
  45.     // 最大不可用数量，默认20%
  46.     MaxUnavailable *intstr.IntOrString `json:"maxUnavailable,omitempty"`
  48.     // 最大弹性数量
  49.     // 如果发布的时候设置了 maxSurge，控制器会先多扩出来 maxSurge 数量的 Pod（此时 Pod 总数为 (replicas+maxSurge))，
  50.     // 然后再开始发布存量的 Pod。 然后，当新版本 Pod 数量已经满足 partition 要求之后，控制器会再把多余的 maxSurge 数量的 Pod 删除掉，
  51.     // 保证最终的 Pod 数量符合 replicas。
  53.     // 要说明的是，maxSurge 不允许配合 InPlaceOnly 更新模式使用。 
  54.     // 另外，如果是与 InPlaceIfPossible 策略配合使用，控制器会先扩出来 maxSurge 数量的 Pod，
  55.     // 再对存量 Pod 做原地升级。
  56.     MaxSurge *intstr.IntOrString `json:"maxSurge,omitempty"`
  58.     // CloneSet是否暂停
  59.     Paused bool `json:"paused,omitempty"`
  61.     // 更新优先级
  62.     PriorityStrategy *UpdatePriorityStrategy `json:"priorityStrategy,omitempty"`
  63.     // 打散方式更新
  64.     ScatterStrategy CloneSetUpdateScatterStrategy `json:"scatterStrategy,omitempty"`
  66.     // 原地更新策略
  67.     InPlaceUpdateStrategy *InPlaceUpdateStrategy `json:"inPlaceUpdateStrategy,omitempty"`
  68. }
  70. type CloneSetUpdateScatterStrategy []CloneSetUpdateScatterTerm
  72. type CloneSetUpdateScatterTerm struct {
  73.     Key   string `json:"key"`
  74.     Value string `json:"value"`
  75. }
  77. type CloneSetUpdateStrategyType string
  79. type InPlaceUpdateStrategy struct {
  80.     GracePeriodSeconds int32 `json:"gracePeriodSeconds,omitempty"`
  81. }

2、调谐函数

  1. func (r *ReconcileCloneSet) doReconcile(request reconcile.Request) (res reconcile.Result, retErr error) {
  2.     // fetch
  3.     instance := &appsv1alpha1.CloneSet{}
  4.     err := r.Get(context.TODO(), request.NamespacedName, instance)
  5.     if err != nil {
  6.         // 删除事件
  7.         if errors.IsNotFound(err) {
  8.             clonesetutils.ScaleExpectations.DeleteExpectations(request.String())
  9.             clonesetutils.UpdateExpectations.DeleteExpectations(request.String())
  10.             return reconcile.Result{}, nil
  11.         }
  12.         return reconcile.Result{}, err
  13.     }
  15.     coreControl := clonesetcore.New(instance)
  16.     // 测试特性，永远返回false
  17.     if coreControl.IsInitializing() {
  18.         return reconcile.Result{}, nil
  19.     }
  21.     // 构造选择器
  22.     selector, err := metav1.LabelSelectorAsSelector(instance.Spec.Selector)
  23.     if err != nil {
  24.         return reconcile.Result{}, nil
  25.     }
  27.     // 判断是否处在扩容状态
  28.     // 若还处在扩容状态，则跳过本次循环
  29.     if scaleSatisfied, unsatisfiedDuration, _ := clonesetutils.ScaleExpectations.SatisfiedExpectations(request.String()); !scaleSatisfied {
  30.         if unsatisfiedDuration >= expectations.ExpectationTimeout {
  31.             return reconcile.Result{}, nil
  32.         }
  33.         return reconcile.Result{RequeueAfter: expectations.ExpectationTimeout - unsatisfiedDuration}, nil
  34.     }
  36.     // 通过ownerRef索引获取此CloneSet关联的pod/pvc
  37.     filteredPods, filteredPVCs, err := r.getOwnedResource(instance)
  38.     if err != nil {
  39.         return reconcile.Result{}, err
  40.     }
  42.     // 认领pod，释放label不符合的pod/认领label符合的孤儿pod
  43.     filteredPods, err = r.claimPods(instance, filteredPods)
  44.     if err != nil {
  45.         return reconcile.Result{}, err
  46.     }
  48.     // 获取历史版本
  49.     // cloneset的历史版本是通过ControllerRevison维护，
  50.     // 不同于Deployment通过直接维护repliset来维护历史Spec
  51.     revisions, err := r.controllerHistory.ListControllerRevisions(instance, selector)
  52.     if err != nil {
  53.         return reconcile.Result{}, err
  54.     }
  55.     // 版本排序
  56.     history.SortControllerRevisions(revisions)
  58.     // 获取当前pod的Spec版本和即将要更新的Spec版本
  59.     // 若当前Spec版本为nil，则当前Spec版本为即将要更新的Spec版本
  60.     currentRevision, updateRevision, collisionCount, err := r.getActiveRevisions(instance, revisions, clonesetutils.GetPodsRevisions(filteredPods))
  61.     if err != nil {
  62.         return reconcile.Result{}, err
  63.     }
  65.     // 有可能是pod变更触发controller循环
  66.     // 需要判断一下是否有pod已经变更了版本
  67.     // 此处不在回调函数触发，是因为pod必须先要经过认领，
  68.     // 为了减少认领操作，统一到ControllerLoop进行处理
  69.     for _, pod := range filteredPods {
  70.         clonesetutils.UpdateExpectations.ObserveUpdated(request.String(), updateRevision.Name, pod)
  71.     }
  72.     // 是否正在进行的更新还没有完成
  73.     if updateSatisfied, unsatisfiedDuration, _ := clonesetutils.UpdateExpectations.SatisfiedExpectations(request.String(), updateRevision.Name); !updateSatisfied {
  74.         if unsatisfiedDuration >= expectations.ExpectationTimeout {
  75.             return reconcile.Result{}, nil
  76.         }
  77.         return reconcile.Result{RequeueAfter: expectations.ExpectationTimeout - unsatisfiedDuration}, nil
  78.     }
  80.     for _, pod := range filteredPods {
  81.         // 必须等待客户自定义回调全部处理完成才能进行下一轮同步
  82.         // ResourceVersionExpectations是为了减少等待用户hook而无效的loop
  83.         // 若进入了等待用户定义hook的状态，则Except Pod，只有当Pod发生变化才假定用户自定义hook完成
  84.         // 否则跳过loop
  85.         if isSatisfied, unsatisfiedDuration := clonesetutils.ResourceVersionExpectations.IsSatisfied(pod); !isSatisfied {
  86.             if unsatisfiedDuration >= expectations.ExpectationTimeout {
  87.                 return reconcile.Result{}, nil
  88.             }
  89.             return reconcile.Result{RequeueAfter: expectations.ExpectationTimeout - unsatisfiedDuration}, nil
  90.         }
  91.     }
  93.     // 初始化新的Status
  94.     newStatus := appsv1alpha1.CloneSetStatus{
  95.         ObservedGeneration: instance.Generation,
  96.         UpdateRevision:     updateRevision.Name,
  97.         CollisionCount:     new(int32),
  98.         LabelSelector:      selector.String(),
  99.     }
  100.     *newStatus.CollisionCount = collisionCount
  102.     // 开始进行扩/缩容、升级
  103.     delayDuration, syncErr := r.syncCloneSet(instance, &newStatus, currentRevision, updateRevision, revisions, filteredPods, filteredPVCs)
  105.     // 更新CloneSet状态
  106.     if err = r.statusUpdater.UpdateCloneSetStatus(instance, &newStatus, filteredPods); err != nil {
  107.         return reconcile.Result{}, err
  108.     }
  110.     // 对于用户指定的不存在的pod进行修正
  111.     // 针对PodsToDelete参数
  112.     if err = r.truncatePodsToDelete(instance, filteredPods); err != nil {
  113.         klog.Warningf("Failed to truncate podsToDelete for %s: %v", request, err)
  114.     }
  116.     // 删除超过给定缓存历史Spec数量的旧版本ControllerRevision
  117.     if err = r.truncateHistory(instance, filteredPods, revisions, currentRevision, updateRevision); err != nil {
  118.         klog.Errorf("Failed to truncate history for %s: %v", request, err)
  119.     }
  121.     // 若符合同步成功、给定启动时间、可用数量少于期望数量
  122.     // 则给定启动时间之后再将给Controller Key入对，更新Status
  123.     if syncErr == nil && instance.Spec.MinReadySeconds > 0 && newStatus.AvailableReplicas != newStatus.ReadyReplicas {
  124.         minReadyDuration := time.Second * time.Duration(instance.Spec.MinReadySeconds)
  125.         if delayDuration == 0 || minReadyDuration < delayDuration {
  126.             delayDuration = minReadyDuration
  127.         }
  128.     }
  129.     return reconcile.Result{RequeueAfter: delayDuration}, syncErr
  130. }

3、同步函数

3.1、同步函数

  1. func (r *ReconcileCloneSet) syncCloneSet(
  2.     instance *appsv1alpha1.CloneSet, newStatus *appsv1alpha1.CloneSetStatus,
  3.     currentRevision, updateRevision *apps.ControllerRevision, revisions []*apps.ControllerRevision,
  4.     filteredPods []*v1.Pod, filteredPVCs []*v1.PersistentVolumeClaim,
  5. ) (time.Duration, error) {
  7.     // 若已经被删除，则直接return
  8.     var delayDuration time.Duration
  9.     if instance.DeletionTimestamp != nil {
  10.         return delayDuration, nil
  11.     }
  13.     // 根据Revision复原CloneSet对象
  14.     currentSet, err := r.revisionControl.ApplyRevision(instance, currentRevision)
  15.     if err != nil {
  16.         return delayDuration, err
  17.     }
  18.     updateSet, err := r.revisionControl.ApplyRevision(instance, updateRevision)
  19.     if err != nil {
  20.         return delayDuration, err
  21.     }
  23.     var scaling bool
  24.     var podsScaleErr error
  25.     var podsUpdateErr error
  27.     // 开始扩容
  28.     scaling, podsScaleErr = r.scaleControl.Manage(currentSet, updateSet, currentRevision.Name, updateRevision.Name, filteredPods, filteredPVCs)
  29.     if podsScaleErr != nil {
  30.         newStatus.Conditions = append(newStatus.Conditions, appsv1alpha1.CloneSetCondition{
  31.             Type:               appsv1alpha1.CloneSetConditionFailedScale,
  32.             Status:             v1.ConditionTrue,
  33.             LastTransitionTime: metav1.Now(),
  34.             Message:            podsScaleErr.Error(),
  35.         })
  36.         err = podsScaleErr
  37.     }
  38.     // 若处在扩容状态，则直接return
  39.     // 否则才进行更新
  40.     if scaling {
  41.         return delayDuration, podsScaleErr
  42.     }
  44.     // 开始更新
  45.     delayDuration, podsUpdateErr = r.updateControl.Manage(updateSet, updateRevision, revisions, filteredPods, filteredPVCs)
  46.     if podsUpdateErr != nil {
  47.         newStatus.Conditions = append(newStatus.Conditions, appsv1alpha1.CloneSetCondition{
  48.             Type:               appsv1alpha1.CloneSetConditionFailedUpdate,
  49.             Status:             v1.ConditionTrue,
  50.             LastTransitionTime: metav1.Now(),
  51.             Message:            podsUpdateErr.Error(),
  52.         })
  53.         if err == nil {
  54.             err = podsUpdateErr
  55.         }
  56.     }
  57.     return delayDuration, err
  58. }

3.2、扩容

3.2.1、扩容核心接口

1、首先通过managePreparingDelete和deletePods函数进行副本数量的控制 2、其次才进行升级相关的数量控制

  1. func (r *realControl) Manage(
  2.     currentCS, updateCS *appsv1alpha1.CloneSet,
  3.     currentRevision, updateRevision string,
  4.     pods []*v1.Pod, pvcs []*v1.PersistentVolumeClaim,
  5. ) (bool, error) {
  7.     if updateCS.Spec.Replicas == nil {
  8.         return false, fmt.Errorf("spec.Replicas is nil")
  9.     }
  11.     coreControl := clonesetcore.New(updateCS)
  12.     if !coreControl.IsReadyToScale() {
  13.         return false, nil
  14.     }
  16.     // 获取计划要删除的pod
  17.     // 包含用户通过PodToDelete指定和已经处于PreDelete状态的pod
  18.     podsSpecifiedToDelete, podsInPreDelete := getPlannedDeletedPods(updateCS, pods)
  20.     // 由于用户指定要删除的pod和PreDelete状态的pod有可能有重合，因此需要merge
  21.     // 下面的逻辑只要发生了pod变更或者删除操作，就直接返回
  22.     podsToDelete := util.MergePods(podsSpecifiedToDelete, podsInPreDelete)
  23.     if len(podsToDelete) > 0 {
  24.         if modified, err := r.managePreparingDelete(updateCS, pods, podsInPreDelete, len(podsToDelete)); err != nil || modified {
  25.             return modified, err
  26.         }
  27.         if modified, err := r.deletePods(updateCS, podsToDelete, pvcs); err != nil || modified {
  28.             return modified, err
  29.         }
  30.     }
  32.     // 根据版本获取已经更新的pod和还未更新的pod，并通过旧版本保留数和弹性数量计算diff值
  33.     updatedPods, notUpdatedPods := clonesetutils.SplitPodsByRevision(pods, updateRevision)
  34.     diff, currentRevDiff := calculateDiffs(updateCS, updateRevision == currentRevision, len(pods), len(notUpdatedPods))
  36.     // 小于0则需要创建pod
  37.     if diff < 0 {
  38.         // 需要创建的总数
  39.         expectedCreations := diff * -1
  40.         // 缺少的当前的版本的数量
  41.         // 针对当前旧版本数量少于给定保留旧版本数量的情况
  42.         expectedCurrentCreations := 0
  43.         if currentRevDiff < 0 {
  44.             expectedCurrentCreations = currentRevDiff * -1
  45.         }
  46.         // 可用实例id
  47.         availableIDs := getOrGenAvailableIDs(expectedCreations, pods, pvcs)
  48.         // 存在的pvc名字
  49.         existingPVCNames := sets.NewString()
  50.         for _, pvc := range pvcs {
  51.             existingPVCNames.Insert(pvc.Name)
  52.         }
  53.         // 创建pod
  54.         return r.createPods(expectedCreations, expectedCurrentCreations,
  55.             currentCS, updateCS, currentRevision, updateRevision, availableIDs.List(), existingPVCNames)
  57.     } else if diff > 0 {
  58.         // 否则删除pod
  59.         // 若当前存在要删除的pod，则先等待删除完成，再删除剩余pod
  60.         if len(podsToDelete) > 0 {
  61.             return false, nil
  62.         }
  63.         // 选择要删除的pod
  64.         podsToDelete := choosePodsToDelete(diff, currentRevDiff, notUpdatedPods, updatedPods)
  65.         return r.deletePods(updateCS, podsToDelete, pvcs)
  66.     }
  67.     return false, nil
  68. }
  70. func (r *realControl) managePreparingDelete(cs *appsv1alpha1.CloneSet, pods, podsInPreDelete []*v1.Pod, numToDelete int) (bool, error) {
  71.     // diff为还需要删除的数量
  72.     // 若要删除的数量，删除完成后副本数量达不到期望，将PreDelete的pod转为Normal状态
  73.     diff := int(*cs.Spec.Replicas) - len(pods) + numToDelete
  74.     var modified bool
  75.     for _, pod := range podsInPreDelete {
  76.         if diff <= 0 {
  77.             return modified, nil
  78.         }
  79.         if isPodSpecifiedDelete(cs, pod) {
  80.             continue
  81.         }
  82.         if patched, err := lifecycle.PatchPodLifecycle(r, pod, appsv1alpha1.LifecycleStateNormal); err != nil {
  83.             return modified, err
  84.         } else if patched {
  85.             modified = true
  86.             clonesetutils.ResourceVersionExpectations.Expect(pod)
  87.         }
  88.         diff--
  89.     }
  90.     return modified, nil
  91. }
  93. func (r *realControl) createPods(
  94.     expectedCreations, expectedCurrentCreations int,
  95.     currentCS, updateCS *appsv1alpha1.CloneSet,
  96.     currentRevision, updateRevision string,
  97.     availableIDs []string, existingPVCNames sets.String,
  98. ) (bool, error) {
  100.     coreControl := clonesetcore.New(updateCS)
  101.     newPods, err := coreControl.NewVersionedPods(currentCS, updateCS, currentRevision, updateRevision,
  102.         expectedCreations, expectedCurrentCreations, availableIDs)
  103.     if err != nil {
  104.         return false, err
  105.     }
  107.     podsCreationChan := make(chan *v1.Pod, len(newPods))
  108.     for _, p := range newPods {
  109.         clonesetutils.ScaleExpectations.ExpectScale(clonesetutils.GetControllerKey(updateCS), expectations.Create, p.Name)
  110.         podsCreationChan <- p
  111.     }
  113.     var created int64
  114.     successPodNames := sync.Map{}
  115.     _, err = clonesetutils.DoItSlowly(len(newPods), initialBatchSize, func() error {
  116.         pod := <-podsCreationChan
  117.         cs := updateCS
  118.         if pod.Labels[apps.ControllerRevisionHashLabelKey] == currentRevision {
  119.             cs = currentCS
  120.         }
  121.         lifecycle.SetPodLifecycle(appsv1alpha1.LifecycleStateNormal)(pod)
  123.         var createErr error
  124.         if createErr = r.createOnePod(cs, pod, existingPVCNames); createErr != nil {
  125.             return createErr
  126.         }
  128.         atomic.AddInt64(&created, 1)
  130.         successPodNames.Store(pod.Name, struct{}{})
  131.         return nil
  132.     })
  134.     for _, pod := range newPods {
  135.         if _, ok := successPodNames.Load(pod.Name); !ok {
  136.             clonesetutils.ScaleExpectations.ObserveScale(clonesetutils.GetControllerKey(updateCS), expectations.Create, pod.Name)
  137.         }
  138.     }
  140.     if created == 0 {
  141.         return false, err
  142.     }
  143.     return true, err
  144. }
  146. func (r *realControl) createOnePod(cs *appsv1alpha1.CloneSet, pod *v1.Pod, existingPVCNames sets.String) error {
  147.     claims := clonesetutils.GetPersistentVolumeClaims(cs, pod)
  148.     for _, c := range claims {
  149.         if existingPVCNames.Has(c.Name) {
  150.             continue
  151.         }
  152.         clonesetutils.ScaleExpectations.ExpectScale(clonesetutils.GetControllerKey(cs), expectations.Create, c.Name)
  153.         if err := r.Create(context.TODO(), &c); err != nil {
  154.             clonesetutils.ScaleExpectations.ObserveScale(clonesetutils.GetControllerKey(cs), expectations.Create, c.Name)
  155.             r.recorder.Eventf(cs, v1.EventTypeWarning, "FailedCreate", "failed to create pvc: %v, pvc: %v", err, util.DumpJSON(c))
  156.             return err
  157.         }
  158.     }
  160.     if err := r.Create(context.TODO(), pod); err != nil {
  161.         r.recorder.Eventf(cs, v1.EventTypeWarning, "FailedCreate", "failed to create pod: %v, pod: %v", err, util.DumpJSON(pod))
  162.         return err
  163.     }
  165.     r.recorder.Eventf(cs, v1.EventTypeNormal, "SuccessfulCreate", "succeed to create pod %s", pod.Name)
  166.     return nil
  167. }
  169. func (r *realControl) deletePods(cs *appsv1alpha1.CloneSet, podsToDelete []*v1.Pod, pvcs []*v1.PersistentVolumeClaim) (bool, error) {
  170.     var modified bool
  171.     for _, pod := range podsToDelete {
  172.         if cs.Spec.Lifecycle != nil && lifecycle.IsPodHooked(cs.Spec.Lifecycle.PreDelete, pod) {
  173.             if patched, err := lifecycle.PatchPodLifecycle(r, pod, appsv1alpha1.LifecycleStatePreparingDelete); err != nil {
  174.                 return false, err
  175.             } else if patched {
  176.                 modified = true
  177.                 clonesetutils.ResourceVersionExpectations.Expect(pod)
  178.             }
  179.             continue
  180.         }
  182.         clonesetutils.ScaleExpectations.ExpectScale(clonesetutils.GetControllerKey(cs), expectations.Delete, pod.Name)
  183.         if err := r.Delete(context.TODO(), pod); err != nil {
  184.             clonesetutils.ScaleExpectations.ObserveScale(clonesetutils.GetControllerKey(cs), expectations.Delete, pod.Name)
  185.             r.recorder.Eventf(cs, v1.EventTypeWarning, "FailedDelete", "failed to delete pod %s: %v", pod.Name, err)
  186.             return modified, err
  187.         }
  188.         modified = true
  189.         r.recorder.Event(cs, v1.EventTypeNormal, "SuccessfulDelete", fmt.Sprintf("succeed to delete pod %s", pod.Name))
  191.         for _, pvc := range pvcs {
  192.             if pvc.Labels[appsv1alpha1.CloneSetInstanceID] != pod.Labels[appsv1alpha1.CloneSetInstanceID] {
  193.                 continue
  194.             }
  196.             clonesetutils.ScaleExpectations.ExpectScale(clonesetutils.GetControllerKey(cs), expectations.Delete, pvc.Name)
  197.             if err := r.Delete(context.TODO(), pvc); err != nil {
  198.                 clonesetutils.ScaleExpectations.ObserveScale(clonesetutils.GetControllerKey(cs), expectations.Delete, pvc.Name)
  199.                 r.recorder.Eventf(cs, v1.EventTypeWarning, "FailedDelete", "failed to delete pvc %s: %v", pvc.Name, err)
  200.                 return modified, err
  201.             }
  202.         }
  203.     }
  205.     return modified, nil
  206. }

3.2.2、扩容工具函数

  1. // 判断pod是否被指定删除
  2. // 1、判断是否有apps.kruise.io/specified-delete Label
  3. // 2、是否出现在PodsToDelete参数中
  4. func isPodSpecifiedDelete(cs *appsv1alpha1.CloneSet, pod *v1.Pod) bool {
  5.     if specifieddelete.IsSpecifiedDelete(pod) {
  6.         return true
  7.     }
  8.     for _, name := range cs.Spec.ScaleStrategy.PodsToDelete {
  9.         if name == pod.Name {
  10.             return true
  11.         }
  12.     }
  13.     return false
  14. }
  16. // 获取被指定删除的pod和处于PreDelete状态的Pod
  17. func getPlannedDeletedPods(cs *appsv1alpha1.CloneSet, pods []*v1.Pod) ([]*v1.Pod, []*v1.Pod) {
  18.     var podsSpecifiedToDelete []*v1.Pod
  19.     var podsInPreDelete []*v1.Pod
  20.     for _, pod := range pods {
  21.         if isPodSpecifiedDelete(cs, pod) {
  22.             podsSpecifiedToDelete = append(podsSpecifiedToDelete, pod)
  23.         }
  24.         // 根据lifecycle.apps.kruise.io/state Label进行校验
  25.         if lifecycle.GetPodLifecycleState(pod) == appsv1alpha1.LifecycleStatePreparingDelete {
  26.             podsInPreDelete = append(podsInPreDelete, pod)
  27.         }
  28.     }
  29.     return podsSpecifiedToDelete, podsInPreDelete
  30. }
  32. // 获取可用id，如果可以复用，则进行复用
  33. func getOrGenAvailableIDs(num int, pods []*v1.Pod, pvcs []*v1.PersistentVolumeClaim) sets.String {
  34.     existingIDs := sets.NewString()
  35.     availableIDs := sets.NewString()
  36.     for _, pvc := range pvcs {
  37.         if id := pvc.Labels[appsv1alpha1.CloneSetInstanceID]; len(id) > 0 {
  38.             existingIDs.Insert(id)
  39.             availableIDs.Insert(id)
  40.         }
  41.     }
  42.     for _, pod := range pods {
  43.         if id := pod.Labels[appsv1alpha1.CloneSetInstanceID]; len(id) > 0 {
  44.             existingIDs.Insert(id)
  45.             availableIDs.Delete(id)
  46.         }
  47.     }
  48.     retIDs := sets.NewString()
  49.     for i := 0; i < num; i++ {
  50.         id := getOrGenInstanceID(existingIDs, availableIDs)
  51.         retIDs.Insert(id)
  52.     }
  53.     return retIDs
  54. }
  56. // 获取或者生成一个InstanceID
  57. func getOrGenInstanceID(existingIDs, availableIDs sets.String) string {
  58.     id, _ := availableIDs.PopAny()
  59.     if len(id) == 0 {
  60.         for {
  61.             id = rand.String(LengthOfInstanceID)
  62.             if !existingIDs.Has(id) {
  63.                 break
  64.             }
  65.         }
  66.     }
  67.     return id
  68. }
  70. // 计算diff(若小于0，则需要创建；否则需要删除)
  71. func calculateDiffs(cs *appsv1alpha1.CloneSet, revConsistent bool, totalPods int, notUpdatedPods int) (totalDiff int, currentRevDiff int) {
  72.     var maxSurge int
  74.     // 若同时有升级，则需要考虑保留旧版本数量
  75.     if !revConsistent {
  76.         if cs.Spec.UpdateStrategy.Partition != nil {
  77.             currentRevDiff = notUpdatedPods - integer.IntMin(int(*cs.Spec.UpdateStrategy.Partition), int(*cs.Spec.Replicas))
  78.         }
  79.         if currentRevDiff > 0 {
  80.             if cs.Spec.UpdateStrategy.MaxSurge != nil {
  81.                 maxSurge, _ = intstrutil.GetValueFromIntOrPercent(cs.Spec.UpdateStrategy.MaxSurge, int(*cs.Spec.Replicas), true)
  82.                 maxSurge = integer.IntMin(maxSurge, currentRevDiff)
  83.             }
  84.         }
  85.     }
  87.     // 若只是扩缩容，则减去副本数量(弹性扩容数量为0)
  88.     totalDiff = totalPods - int(*cs.Spec.Replicas) - maxSurge
  89.     return
  90. }
  92. // 选择要删除的pod
  93. func choosePodsToDelete(totalDiff int, currentRevDiff int, notUpdatedPods, updatedPods []*v1.Pod) []*v1.Pod {
  94.     choose := func(pods []*v1.Pod, diff int) []*v1.Pod {
  95.         if diff < len(pods) {
  96.             sort.Sort(kubecontroller.ActivePods(pods))
  97.         } else if diff > len(pods) {
  98.             klog.Warningf("Diff > len(pods) in choosePodsToDelete func which is not expected.")
  99.             return pods
  100.         }
  101.         return pods[:diff]
  102.     }
  104.     var podsToDelete []*v1.Pod
  106.     if currentRevDiff >= totalDiff {
  107.         // 若旧版本diff数量大于totalDiff
  108.         // 则直接从旧版本中删除totalDiff pod
  109.         podsToDelete = choose(notUpdatedPods, totalDiff)
  110.     } else if currentRevDiff > 0 {
  111.         // 否则先从notUpdatedPods删除currentRevDiff pods
  112.         // 然后从updatedPods删除剩余pods
  113.         podsToDelete = choose(notUpdatedPods, currentRevDiff)
  114.         podsToDelete = append(podsToDelete, choose(updatedPods, totalDiff-currentRevDiff)...)
  115.     } else {
  116.         // 从updatedPods删除totalDiff pods
  117.         podsToDelete = choose(updatedPods, totalDiff)
  118.     }
  119.     return podsToDelete
  120. }

3.3、升级

3.3.1、升级核心接口

  1. func (c *realControl) Manage(cs *appsv1alpha1.CloneSet,
  2.     updateRevision *apps.ControllerRevision, revisions []*apps.ControllerRevision,
  3.     pods []*v1.Pod, pvcs []*v1.PersistentVolumeClaim,
  4. ) (time.Duration, error) {
  6.     requeueDuration := requeueduration.Duration{}
  7.     coreControl := clonesetcore.New(cs)
  9.     if cs.Spec.UpdateStrategy.Paused {
  10.         return requeueDuration.Get(), nil
  11.     }
  13.     // 刷新所有的pod状态
  14.     var modified bool
  15.     for _, pod := range pods {
  16.         patched, duration, err := c.refreshPodState(cs, coreControl, pod)
  17.         if err != nil {
  18.             return 0, err
  19.         } else if duration > 0 {
  20.             requeueDuration.Update(duration)
  21.         }
  22.         if patched {
  23.             modified = true
  24.         }
  25.     }
  26.     if modified {
  27.         return requeueDuration.Get(), nil
  28.     }
  30.     // 找到所有等待更新的pod
  31.     // 满足：
  32.     //     1、当前pod的revision不为给定的revision
  33.     //     2、不处于PreparingDelete、Updated状态
  34.     var waitUpdateIndexes []int
  35.     for i := range pods {
  36.         if coreControl.IsPodUpdatePaused(pods[i]) {
  37.             continue
  38.         }
  39.         if clonesetutils.GetPodRevision(pods[i]) != updateRevision.Name {
  40.             switch lifecycle.GetPodLifecycleState(pods[i]) {
  41.             case appsv1alpha1.LifecycleStatePreparingDelete, appsv1alpha1.LifecycleStateUpdated:
  42.             default:
  43.                 waitUpdateIndexes = append(waitUpdateIndexes, i)
  44.             }
  45.         }
  46.     }
  48.     // 根据优先级策略和打散策略给待升级的pod进行排序
  49.     waitUpdateIndexes = sortUpdateIndexes(coreControl, cs.Spec.UpdateStrategy, pods, waitUpdateIndexes)
  51.     // 计算最大可以扩容的数量
  52.     needToUpdateCount := calculateUpdateCount(coreControl, cs.Spec.UpdateStrategy, cs.Spec.MinReadySeconds, int(*cs.Spec.Replicas), waitUpdateIndexes, pods)
  53.     if needToUpdateCount < len(waitUpdateIndexes) {
  54.         waitUpdateIndexes = waitUpdateIndexes[:needToUpdateCount]
  55.     }
  57.     // 更新pod
  58.     for _, idx := range waitUpdateIndexes {
  59.         pod := pods[idx]
  60.         if duration, err := c.updatePod(cs, coreControl, updateRevision, revisions, pod, pvcs); err != nil {
  61.             return requeueDuration.Get(), err
  62.         } else if duration > 0 {
  63.             requeueDuration.Update(duration)
  64.         }
  65.     }
  67.     return requeueDuration.Get(), nil
  68. }

3.3.2、升级工具函数

  1. func (c *realControl) refreshPodState(cs *appsv1alpha1.CloneSet, coreControl clonesetcore.Control, pod *v1.Pod) (bool, time.Duration, error) {
  2.     opts := coreControl.GetUpdateOptions()
  3.     res := c.inplaceControl.Refresh(pod, opts)
  4.     if res.RefreshErr != nil {
  5.         return false, 0, res.RefreshErr
  6.     }
  8.     var state appsv1alpha1.LifecycleStateType
  9.     switch lifecycle.GetPodLifecycleState(pod) {
  10.     case appsv1alpha1.LifecycleStateUpdating:
  11.         checkFunc := inplaceupdate.CheckInPlaceUpdateCompleted
  12.         if opts != nil && opts.CustomizeCheckUpdateCompleted != nil {
  13.             checkFunc = opts.CustomizeCheckUpdateCompleted
  14.         }
  15.         if checkFunc(pod) == nil {
  16.             if cs.Spec.Lifecycle != nil && !lifecycle.IsPodHooked(cs.Spec.Lifecycle.InPlaceUpdate, pod) {
  17.                 state = appsv1alpha1.LifecycleStateUpdated
  18.             } else {
  19.                 state = appsv1alpha1.LifecycleStateNormal
  20.             }
  21.         }
  22.     case appsv1alpha1.LifecycleStateUpdated:
  23.         if cs.Spec.Lifecycle == nil ||
  24.             cs.Spec.Lifecycle.InPlaceUpdate == nil ||
  25.             lifecycle.IsPodHooked(cs.Spec.Lifecycle.InPlaceUpdate, pod) {
  26.             state = appsv1alpha1.LifecycleStateNormal
  27.         }
  28.     }
  30.     if state != "" {
  31.         if patched, err := lifecycle.PatchPodLifecycle(c, pod, state); err != nil {
  32.             return false, 0, err
  33.         } else if patched {
  34.             clonesetutils.ResourceVersionExpectations.Expect(pod)
  35.             return true, res.DelayDuration, nil
  36.         }
  37.     }
  38.     return false, res.DelayDuration, nil
  39. }
  41. func (c *realControl) updatePod(cs *appsv1alpha1.CloneSet, coreControl clonesetcore.Control,
  42.     updateRevision *apps.ControllerRevision, revisions []*apps.ControllerRevision,
  43.     pod *v1.Pod, pvcs []*v1.PersistentVolumeClaim,
  44. ) (time.Duration, error) {
  46.     if cs.Spec.UpdateStrategy.Type == appsv1alpha1.InPlaceIfPossibleCloneSetUpdateStrategyType ||
  47.         cs.Spec.UpdateStrategy.Type == appsv1alpha1.InPlaceOnlyCloneSetUpdateStrategyType {
  48.         var oldRevision *apps.ControllerRevision
  49.         for _, r := range revisions {
  50.             if r.Name == clonesetutils.GetPodRevision(pod) {
  51.                 oldRevision = r
  52.                 break
  53.             }
  54.         }
  56.         if c.inplaceControl.CanUpdateInPlace(oldRevision, updateRevision, coreControl.GetUpdateOptions()) {
  57.             if cs.Spec.Lifecycle != nil && lifecycle.IsPodHooked(cs.Spec.Lifecycle.InPlaceUpdate, pod) {
  58.                 if patched, err := lifecycle.PatchPodLifecycle(c, pod, appsv1alpha1.LifecycleStatePreparingUpdate); err != nil {
  59.                     return 0, err
  60.                 } else if patched {
  61.                     clonesetutils.ResourceVersionExpectations.Expect(pod)
  62.                 }
  63.                 return 0, nil
  64.             }
  66.             opts := coreControl.GetUpdateOptions()
  67.             opts.AdditionalFuncs = append(opts.AdditionalFuncs, lifecycle.SetPodLifecycle(appsv1alpha1.LifecycleStateUpdating))
  68.             res := c.inplaceControl.Update(pod, oldRevision, updateRevision, opts)
  69.             if res.InPlaceUpdate {
  70.                 if res.UpdateErr == nil {
  71.                     clonesetutils.UpdateExpectations.ExpectUpdated(clonesetutils.GetControllerKey(cs), updateRevision.Name, pod)
  72.                     return res.DelayDuration, nil
  73.                 }
  74.                 return res.DelayDuration, res.UpdateErr
  75.             }
  76.         }
  78.         if cs.Spec.UpdateStrategy.Type == appsv1alpha1.InPlaceOnlyCloneSetUpdateStrategyType {
  79.             return 0, fmt.Errorf("find Pod %s update strategy is InPlaceOnly but can not update in-place", pod.Name)
  80.         }
  81.     }
  83.     // 若更新策略不为原地升级类型，直接修改Pod labels apps.kruise.io/specified-delete为true
  84.     if patched, err := specifieddelete.PatchPodSpecifiedDelete(c, pod, "true"); err != nil {
  85.         return 0, err
  86.     } else if patched {
  87.         clonesetutils.ResourceVersionExpectations.Expect(pod)
  88.     }
  89.     return 0, nil
  90. }
  92. func sortUpdateIndexes(coreControl clonesetcore.Control, strategy appsv1alpha1.CloneSetUpdateStrategy, pods []*v1.Pod, waitUpdateIndexes []int) []int {
  93.     sort.Slice(waitUpdateIndexes, coreControl.GetPodsSortFunc(pods, waitUpdateIndexes))
  95.     if strategy.PriorityStrategy != nil {
  96.         waitUpdateIndexes = updatesort.NewPrioritySorter(strategy.PriorityStrategy).Sort(pods, waitUpdateIndexes)
  97.     }
  98.     if strategy.ScatterStrategy != nil {
  99.         waitUpdateIndexes = updatesort.NewScatterSorter(strategy.ScatterStrategy).Sort(pods, waitUpdateIndexes)
  100.     }
  102.     sort.Slice(waitUpdateIndexes, func(i, j int) bool {
  103.         preparingUpdateI := lifecycle.GetPodLifecycleState(pods[waitUpdateIndexes[i]]) == appsv1alpha1.LifecycleStatePreparingUpdate
  104.         preparingUpdateJ := lifecycle.GetPodLifecycleState(pods[waitUpdateIndexes[j]]) == appsv1alpha1.LifecycleStatePreparingUpdate
  105.         if preparingUpdateI != preparingUpdateJ {
  106.             return preparingUpdateI
  107.         }
  108.         return false
  109.     })
  110.     return waitUpdateIndexes
  111. }
  113. // 计算要更新的数量
  114. func calculateUpdateCount(coreControl clonesetcore.Control, strategy appsv1alpha1.CloneSetUpdateStrategy, minReadySeconds int32, totalReplicas int, waitUpdateIndexes []int, pods []*v1.Pod) int {
  115.     partition := 0
  116.     if strategy.Partition != nil {
  117.         partition = int(*strategy.Partition)
  118.     }
  120.     // 保留旧版本数量
  121.     if len(waitUpdateIndexes)-partition <= 0 {
  122.         return 0
  123.     }
  124.     waitUpdateIndexes = waitUpdateIndexes[:(len(waitUpdateIndexes) - partition)]
  126.     roundUp := true
  127.     if strategy.MaxSurge != nil {
  128.         maxSurge, _ := intstrutil.GetValueFromIntOrPercent(strategy.MaxSurge, totalReplicas, true)
  129.         roundUp = maxSurge == 0
  130.     }
  131.     maxUnavailable, _ := intstrutil.GetValueFromIntOrPercent(
  132.         intstrutil.ValueOrDefault(strategy.MaxUnavailable, intstrutil.FromString(appsv1alpha1.DefaultCloneSetMaxUnavailable)), totalReplicas, roundUp)
  133.     usedSurge := len(pods) - totalReplicas
  135.     // 可升级数量理论上为：
  136.     // 已经在升级但是状态不为Ready的 ➕ 等待升级的数量
  137.     var notReadyCount, updateCount int
  138.     for _, p := range pods {
  139.         if !isPodReady(coreControl, p, minReadySeconds) {
  140.             notReadyCount++
  141.         }
  142.     }
  143.     for _, i := range waitUpdateIndexes {
  144.         if isPodReady(coreControl, pods[i], minReadySeconds) {
  145.             if notReadyCount >= (maxUnavailable + usedSurge) {
  146.                 break
  147.             } else {
  148.                 notReadyCount++
  149.             }
  150.         }
  151.         updateCount++
  152.     }
  153.     return updateCount
  154. }
  156. // pod ready要求
  157. // 1、lifecycle state为Normal
  158. // 2、pod.Status.Phase == v1.PodRunning
  159. // 3、pod是Available状态
  160. // 4、condition InPlaceUpdateReady值为True
  161. func isPodReady(coreControl clonesetcore.Control, pod *v1.Pod, minReadySeconds int32) bool {
  162.     state := lifecycle.GetPodLifecycleState(pod)
  163.     if state != "" && state != appsv1alpha1.LifecycleStateNormal {
  164.         return false
  165.     }
  166.     return coreControl.IsPodUpdateReady(pod, minReadySeconds)
  167. }

3.3.3、原地升级函数

4、整体逻辑