replicaset

1.简介

1.1.定义

• rs controller是kube-controller-manager组件负责rs资源对象的控制器，根据期望副本调整pod数量，实现承接业务容器的动态管理。

• // ReplicaSetController is responsible for synchronizing rs objects with actual running pods.
  type ReplicaSetController struct {
  	...
      
  	podControl controller.PodControlInterface						// pod交互控制
  	...
  
  	burstReplicas int												// 创建或删除一定数量Pod临时挂起,监听到指定事件恢复
  
  	syncHandler func(ctx context.Context, rsKey string) error		// 处理模块
  
      // 缓存对象,记录每个ReplicaSet需要创建/删除的Pod
  	// 每轮同步过程中,对于创建/删除操作失败的Pod数量,都会记录起来
  	// 等到下一轮同步时继续执行相关的操作
  	// 直到Pod数量副本达到期望状态
  	expectations *controller.UIDTrackingControllerExpectations
  
  	rsLister appslisters.ReplicaSetLister							// rs informer缓存
  	...
      
  	rsIndexer      cache.Indexer									// 索引
  
  	podLister corelisters.PodLister									// pod informer缓存
  	...
  
  	queue workqueue.RateLimitingInterface							// 工作队列
  }
  

• 

• 注意

  replicaset是控制pod数量的重要资源，deployment基于replicaset管理间接影响承载业务的pod数量

1.2.原理

• rs controller利用informer监听rs/pod，将相关的replicaset推入workqueue，由syncHandler处理rs副本同步任务。

• --- 执行过程
  	1.基于informer监听执行事件回调,将replicaset推入workqueue
  	2.异步worker执行syncHandler，实时获取workqueue的item及处理
  	3.根据item执行pod数量调整
  

• 

• 注意

  workqueue是一个先入先出的队列，由item切片、dirty map和processing map构成

2.分析

2.1.start

• controller-manager基于注册的initFunc管理不同的controller，startReplicaSetController负责实例化及启动replicaset同步。

• // NewReplicaSetController configures a replica set controller with the specified event recorder
  func NewReplicaSetController(...) *ReplicaSetController {
  	...
  	return NewBaseController(rsInformer, podInformer, kubeClient, burstReplicas...)
  }
  
  // NewBaseController is the implementation of NewReplicaSetController with additional injected.
  func NewBaseController(...) *ReplicaSetController {
  
      // 初始化rs controller
  	rsc := &ReplicaSetController{
  		GroupVersionKind: gvk,
  		...
  		burstReplicas:    burstReplicas,
  		expectations:     c.NewUIDTrackingControllerExpectations(controller.NewControllerExpectations()),
  		queue:            workqueue.NewNamedRateLimitingQueue(w.DefaultControllerRateLimiter(), queueName),
  	}
  
      // rs监听
  	rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc:    rsc.addRS,
  		UpdateFunc: rsc.updateRS,
  		DeleteFunc: rsc.deleteRS,
  	})
      
      
  	rsInformer.Informer().AddIndexers(cache.Indexers{
          // ownerRef.uid作为索引
  		controllerUIDIndex: func(obj interface{}) ([]string, error) {
  			rs, ok := obj.(*apps.ReplicaSet)
  			...
              
  			controllerRef := metav1.GetControllerOf(rs)
  			...
              
  			return []string{string(controllerRef.UID)}, nil
  		},
  	})
      
      // rs informer缓存及索引
  	rsc.rsIndexer = rsInformer.Informer().GetIndexer()
  	rsc.rsLister = rsInformer.Lister()
  	rsc.rsListerSynced = rsInformer.Informer().HasSynced
  
      // pod监听
  	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
  		AddFunc: rsc.addPod,
  		UpdateFunc: rsc.updatePod,
  		DeleteFunc: rsc.deletePod,
  	})
  	rsc.podLister = podInformer.Lister()
  	rsc.podListerSynced = podInformer.Informer().HasSynced
  
      // handler
  	rsc.syncHandler = rsc.syncReplicaSet
  
  	return rsc
  }
  
  func startReplicaSetController(...) (controller.Interface, bool, error) {
  	go replicaset.NewReplicaSetController(
  		klog.FromContext(ctx),
  		controllerContext.InformerFactory.Apps().V1().ReplicaSets(),
  		controllerContext.InformerFactory.Core().V1().Pods(),
  		controllerContext.ClientBuilder.ClientOrDie("replicaset-controller"),
  		replicaset.BurstReplicas,
  	).Run(ctx, int(controllerContext.ComponentConfig.ReplicaSetController.ConcurrentRSSyncs))
  	return nil, true, nil
  }
  

• 

• 注意

  rsInformer和podInformer会获取相关的replicaset，将rs入队供syncHandler处理

2.2.run

• rsc.Run()激活replicaset controller，基于一定数量的worker协程定义执行syncHandler，共同消费workqueue数据进行rs同步。

• // Run begins watching and syncing.
  func (rsc *ReplicaSetController) Run(ctx context.Context, workers int) {
  	...
  
  	defer rsc.queue.ShutDown()
  	...
  
      // 阻塞等待首次informer同步
  	if !cache.WaitForNamedCacheSync(rsc.Kind, ctx.Done(), rsc.podListerSynced, rsc.rsListerSynced) {
  		return
  	}
  
      // 激活一定worker执行syncHandler
  	for i := 0; i < workers; i++ {
  		go wait.UntilWithContext(ctx, rsc.worker, time.Second)
  	}
  
  	<-ctx.Done()
  }
  
  // worker runs a worker thread that just dequeues items, processes them, and marks them done.
  func (rsc *ReplicaSetController) worker(ctx context.Context) {
  	for rsc.processNextWorkItem(ctx) {
  	}
  }
  
  func (rsc *ReplicaSetController) processNextWorkItem(ctx context.Context) bool {
  	// 获取队首Item
      key, quit := rsc.queue.Get()
  	...
      
      // 成功处理的由processing map清理
  	defer rsc.queue.Done(key)
  
      // 执行同步
  	err := rsc.syncHandler(ctx, key.(string))
  	if err == nil {
          // 重置限流状态
  		rsc.queue.Forget(key)
  		return true
  	}
  
  	// 设置限流重试
  	rsc.queue.AddRateLimited(key)
  
  	return true
  }
  

• 

• 补充

  worker()会定时执行，不停消费workqueue推入的rs，执行syncHandler同步处理

2.3.syncHandler

• rsc.syncReplicaset()会控制驱动rs副本达到预期状态，实际副本数低于预期会驱动创建，高于预期会驱动销毁，执行后计算的状态会更新到rs。

• // syncReplicaSet will sync the ReplicaSet with the given key if it has had its expectations fulfilled.
  func (rsc *ReplicaSetController) syncReplicaSet(ctx context.Context, key string) error {
  	...
  	
  	namespace, name, err := cache.SplitMetaNamespaceKey(key)
  	...
      
      // 获取rs
  	rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)
  	if apierrors.IsNotFound(err) {
  		// 未找到,由预期缓存删除
  		rsc.expectations.DeleteExpectations(key)
  		return nil
  	}
  	...
  
      // 检查执行manageReplicas同步的必要性
  	rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
      
      // selector
  	selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
  	...
  
  	// 获取ns所有pod
  	allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
  	...
      
      // 过滤未走到终态(succeed/failed)及未删除的pod
  	filteredPods := controller.FilterActivePods(allPods)
  
  	// pod领养
  	filteredPods, err = rsc.claimPods(ctx, rs, selector, filteredPods)
      ...
  
  	// rs未达到预期及未删除,执行manageReplicas同步
  	if rsNeedsSync && rs.DeletionTimestamp == nil {
  		manageReplicasErr = rsc.manageReplicas(ctx, filteredPods, rs)
  	}
      
      // 计算rs最新状态
  	rs = rs.DeepCopy()
  	newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)
  
  	// 更新rs最新状态
  	updatedRS, err := updateReplicaSetStatus(ctx,rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace),rs,newStatus)
  	...
      
  	// 其它条件满足,可用副本未达到预期,rs延迟入队
  	if manageReplicasErr == nil && updatedRS.Spec.MinReadySeconds > 0 &&
  		updatedRS.Status.ReadyReplicas == *(updatedRS.Spec.Replicas) &&
  		updatedRS.Status.AvailableReplicas != *(updatedRS.Spec.Replicas) {
  		rsc.queue.AddAfter(key, time.Duration(updatedRS.Spec.MinReadySeconds)*time.Second)
  	}
      
  	return manageReplicasErr
  }
  
  // SatisfiedExpectations returns true if the required adds/dels for the given controller have been observed.
  func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {
      // exp存在
  	if exp, exists, err := r.GetExpectations(controllerKey); exists {
          // add及del<=0,rs副本数达到预期(预期返回true?)
  		if exp.Fulfilled() {
  			return true
          // 距上次同步超5min
  		} else if exp.isExpired() {
  			return true
  		} else {
  			return false
  		}
  	}
      
  	// exp不存在或出错
  	return true
  }
  

• 

• 注意

  rsc.claimPods()会基于Pod标注的ownerRef及label进行领养及弃养，梳理rs可认领的Pod

2.4.claimPods

• controller.FilterActivePods()过滤的活跃Pod会执行rsc.claimPods()进一步认领，基于ownerRef和label针对性领养和弃养。

• func (rsc *ReplicaSetController) claimPods(...) ([]*v1.Pod, error) {
  	// If any adoptions are attempted, we should first recheck for deletion with
  	// an uncached quorum read sometime after listing Pods (see #42639).
  	canAdoptFunc := controller.RecheckDeletionTimestamp(func(ctx context.Context) (metav1.Object, error) {
  		fresh, err := rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace).Get(ctx, rs.Name, metav1.GetOptions{})
  		...
          
          // rs重建,无资格认领
  		if fresh.UID != rs.UID {
  			return nil, fmt.Errorf("original %v %v/%v is gone: got uid %v, wanted %v"...)
  		}
  		return fresh, nil
  	})
      
      // 认领
  	cm := controller.NewPodControllerRefManager(rsc.podControl,rs,selector,rsc.GroupVersionKind, canAdoptFunc)
  	return cm.ClaimPods(ctx, filteredPods)
  }
  
  // ClaimPods tries to take ownership of a list of Pods.
  func (m *PodControllerRefManager) ClaimPods(...) ([]*v1.Pod, error) {
  	...
  
      // selector匹配
  	match := func(obj metav1.Object) bool {
  		pod := obj.(*v1.Pod)
  		// Check selector first so filters only run on potentially matching Pods.
  		if !m.Selector.Matches(labels.Set(pod.Labels)) {
  			return false
  		}
          
  		...
  		return true
  	}
      
      // 领养(设置pod ownerRef)
  	adopt := func(ctx context.Context, obj metav1.Object) error {
  		return m.AdoptPod(ctx, obj.(*v1.Pod))
  	}
      
      // 弃养(清理pod ownerRef)
  	release := func(ctx context.Context, obj metav1.Object) error {
  		return m.ReleasePod(ctx, obj.(*v1.Pod))
  	}
  
      // 依次认领activePod
  	for _, pod := range pods {
  		// 检查认领条件
          ok, err := m.ClaimObject(ctx, pod, match, adopt, release)
  		...
          
          // 认领
  		if ok {
  			claimed = append(claimed, pod)
  		}
  	}
  	return claimed, utilerrors.NewAggregate(errlist)
  }
  

• 

• 补充

  rsc.claimPod()基于label或owner决定pod归属的rs，进行领养或弃养，实现activPods再次过滤

2.5.manage

• rsc.manageReplicas()会计算实际副本和预期副本差值，副本不够执行slowStartBatch()并发分批创建，副本超出则销毁多余的副本。

• // manageReplicas checks and updates replicas for the given ReplicaSet.
  func (rsc *ReplicaSetController) manageReplicas(...) error {
      // 计算差值
  	diff := len(filteredPods) - int(*(rs.Spec.Replicas))
      
  	rsKey, err := controller.KeyFunc(rs)
  	...
      
      // 副本不够
  	if diff < 0 {
  		diff *= -1
          // 每轮创建不能超过500
  		if diff > rsc.burstReplicas {
  			diff = rsc.burstReplicas
  		}
  		
          // 创建一个exp对象,add值为diff
  		rsc.expectations.ExpectCreations(rsKey, diff)
  		
          // 执行slowStartBatch并发分批创建,返回成功的pod数量(步长为2,4,6,8...,diff)
  		successfulCreations, err := slowStartBatch(diff, 1, func() error {
  			err := rsc.podControl.CreatePods(ctx, rs.Namespace, &rs.Spec.Template, rs, metav1.NewControllerRef(rs, rsc.GroupVersionKind))
              / ns terminating跳过
  			if apierrors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
                  return nil
  			}
              
  			return err
  		})
  
          // exp对象扣减失败的pod(5-->3,失败2)
  		if skippedPods := diff - successfulCreations; skippedPods > 0 {
  			for i := 0; i < skippedPods; i++ {
  				// Decrement the expected number of creates because the informer won't observe this pod
  				rsc.expectations.CreationObserved(rsKey)
  			}
  		}
  		return err
      
      // 副本超出
  	} else if diff > 0 {
          // 设置每轮最大调度数量
  		if diff > rsc.burstReplicas {
  			diff = rsc.burstReplicas
  		}
  
          // 获取相关deploy的rs管理的所有pod
  		relatedPods, err := rsc.getIndirectlyRelatedPods(klog.FromContext(ctx), rs)
  		...
  
  		// 获取可以被清理的pod
  		podsToDelete := getPodsToDelete(filteredPods, relatedPods, diff)
  
  		// 覆盖exp对象,del值为diff
  		rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))
  		...
          
  		wg.Add(diff)
  		for _, pod := range podsToDelete {
  			go func(targetPod *v1.Pod) {
  				defer wg.Done()
                  // 清理可删除pod
  				if err := rsc.podControl.DeletePod(ctx, rs.Namespace, targetPod.Name, rs); err != nil {
  					...
                      // 更新exp对象(清理pod uid,del数量减1)
  					rsc.expectations.DeletionObserved(rsKey, podKey)
  					...
                      
  				}
  			}(pod)
  		}
  		wg.Wait()
  
  		select {
  		case err := <-errCh:
  			if err != nil {
  				return err
  			}
  		default:
  		}
  	}
  
  	return nil
  }
  

• 

• 注意

  rsc.manageReplicas()会限制单次最多操作500个Pod，创建或删除会设置expectations对象的add/del记录，操作一个扣减一个

3.补充

3.1.slowBatch

• slowStartBatch()用于并发分批Pod，每轮批次呈指数级增长，初始batchSize=1，之后按照2,4,8,16,32...递进，直到达到diff。

• // slowStartBatch tries to call the provided function a total of 'count' times.
  func slowStartBatch(count int, initialBatchSize int, fn func() error) (int, error) {
  	...
      
      // 分批指数级调度
  	for batchSize := integer.IntMin(remaining, initialBatchSize); batchSize > 0; batchSize = integer.IntMin(2*batchSize, remaining) {
  		...
          
  		wg.Add(batchSize)
  		// 执行创建
          for i := 0; i < batchSize; i++ {
  			go func() {
  				defer wg.Done()
  				if err := fn(); err != nil {
  					errCh <- err
  				}
  			}()
  		}
  		wg.Wait()
          
          // 统计成功的pod数量
  		curSuccesses := batchSize - len(errCh)
  		successes += curSuccesses
  		
          // 出现错误直接返回
          if len(errCh) > 0 {
  			return successes, <-errCh
  		}
          
          // 继续下一轮
  		remaining -= batchSize
  	}
  	return successes, nil
  }
  
  func (r RealPodControl) CreatePods(...) error {
  	return r.CreatePodsWithGenerateName(ctx, namespace, template, controllerObject, controllerRef, "")
  }
  
  func (r RealPodControl) CreatePodsWithGenerateName(...) error {
      // 检查ownerRef定义
  	validateControllerRef(controllerRef)
      ...
      
      // 基于rs.spec.template生成pod定义
  	pod, err := GetPodFromTemplate(template, controllerObject, controllerRef)
  	...
      
      // 覆盖pod名称
  	if len(generateName) > 0 {
  		pod.ObjectMeta.GenerateName = generateName
  	}
      
      // 执行创建
  	return r.createPods(ctx, namespace, pod, controllerObject)
  }
  

• 

• 注意

  slowStartBatch()会并发分批创建pod，只要有一个批次出现错误就直接退出

3.2.podToDelete

• getPodsToDelete()会过滤待删除的Pod，具体策略就是Pod根据质量优先级排序，低优先级的位于前面，直接截取前diff的Pod作为删除项。

• func getPodsToDelete(filteredPods, relatedPods []*v1.Pod, diff int) []*v1.Pod {
  	// 少量删除排序pods
  	if diff < len(filteredPods) {
  		podsWithRanks := getPodsRankedByRelatedPodsOnSameNode(filteredPods, relatedPods)
  		sort.Sort(podsWithRanks)
  		...
  	}
      
      // 超出或排序后直接取diff个
  	return filteredPods[:diff]
  }
  
  // getPodsRankedByRelatedPodsOnSameNode returns an ActivePodsWithRanks value that wraps podsToRank.
  func getPodsRankedByRelatedPodsOnSameNode(podsToRank, relatedPods []*v1.Pod) controller.ActivePodsWithRanks {
  	...
      
      // 构造打分对象
  	return controller.ActivePodsWithRanks{Pods: podsToRank, Rank: ranks, Now: metav1.Now()}
  }
  
  // Less compares two pods with corresponding ranks.
  func (s ActivePodsWithRanks) Less(i, j int) bool {
  	// 1.unbind node < bind node
  	if s.Pods[i].Spec.NodeName != s.Pods[j].Spec.NodeName && (len(s.Pods[i].Spec.NodeName) == 0 || len(s.Pods[j].Spec.NodeName) == 0) {
  		return len(s.Pods[i].Spec.NodeName) == 0
  	}
      
  	// 2. PodPending < PodUnknown < PodRunning
  	if podPhaseToOrdinal[s.Pods[i].Status.Phase] != podPhaseToOrdinal[s.Pods[j].Status.Phase] {
  		return podPhaseToOrdinal[s.Pods[i].Status.Phase] < podPhaseToOrdinal[s.Pods[j].Status.Phase]
  	}
      
  	// 3. Not ready < ready
  	if podutil.IsPodReady(s.Pods[i]) != podutil.IsPodReady(s.Pods[j]) {
  		return !podutil.IsPodReady(s.Pods[i])
  	}
  
      // 4. lower pod-deletion-cost < higher pod-deletion cost(低成本的优先删)
  	if utilfeature.DefaultFeatureGate.Enabled(features.PodDeletionCost) {
  		pi, _ := helper.GetDeletionCostFromPodAnnotations(s.Pods[i].Annotations)
  		pj, _ := helper.GetDeletionCostFromPodAnnotations(s.Pods[j].Annotations)
  		if pi != pj {
  			return pi < pj
  		}
  	}
  
  	// 5. more num in node < less num in node
  	if s.Rank[i] != s.Rank[j] {
  		return s.Rank[i] > s.Rank[j]
  	}
      
      // 6. Been ready for empty time < less time < more time(均为ready)
  	if podutil.IsPodReady(s.Pods[i]) && podutil.IsPodReady(s.Pods[j]) {
  		readyTime1 := podReadyTime(s.Pods[i])
  		readyTime2 := podReadyTime(s.Pods[j])
  		if !readyTime1.Equal(readyTime2) {
              // 线性对比
  			if !utilfeature.DefaultFeatureGate.Enabled(features.LogarithmicScaleDown) {
  				return afterOrZero(readyTime1, readyTime2)
              // 取对数对比
  			} else {
  				if s.Now.IsZero() || readyTime1.IsZero() || readyTime2.IsZero() {
  					return afterOrZero(readyTime1, readyTime2)
  				}
                  // log(now-r1) - log(now-r2),缓解长时间差异
  				rankDiff := logarithmicRankDiff(*readyTime1, *readyTime2, s.Now)
                  
                  // 无差异,uid小的优先删
  				if rankDiff == 0 {
  					return s.Pods[i].UID < s.Pods[j].UID
  				}
  				return rankDiff < 0
  			}
  		}
  	}
      
  	// 7. Pods higher restart counts < lower restart counts
  	if maxContainerRestarts(s.Pods[i]) != maxContainerRestarts(s.Pods[j]) {
  		return maxContainerRestarts(s.Pods[i]) > maxContainerRestarts(s.Pods[j])
  	}
      
  	// 8. Empty creation time pods < newer pods < older pods
  	if !s.Pods[i].CreationTimestamp.Equal(&s.Pods[j].CreationTimestamp) {
          // 线性对比创建时间,时间短的优先删除
  		if !utilfeature.DefaultFeatureGate.Enabled(features.LogarithmicScaleDown) {
  			return afterOrZero(&s.Pods[i].CreationTimestamp, &s.Pods[j].CreationTimestamp)
          // 取对数对比
  		} else {
  			if s.Now.IsZero() || s.Pods[i].CreationTimestamp.IsZero() || s.Pods[j].CreationTimestamp.IsZero() {
  				return afterOrZero(&s.Pods[i].CreationTimestamp, &s.Pods[j].CreationTimestamp)
  			}
              // log(now-c1) - log(now-c2),缓解长时间差异
  			rankDiff := logarithmicRankDiff(s.Pods[i].CreationTimestamp, s.Pods[j].CreationTimestamp, s.Now)
  			
              // 无差异,uid小的优先删
              if rankDiff == 0 {
  				return s.Pods[i].UID < s.Pods[j].UID
  			}
              
              // 创建时间短的优先删
  			return rankDiff < 0
  		}
  	}
  	return false
  }
  

• 

• 注意

  podToDelete过滤原则很多，基本都是选择最小的代价、最稳定的角度选出最值得删除的Pod