replicaset controller 的实现原理

func startReplicaSetController(ctx context.Context, controllerContext ControllerContext) (controller.Interface, bool, error) {
	go replicaset.NewReplicaSetController(
		controllerContext.InformerFactory.Apps().V1().ReplicaSets(),
		controllerContext.InformerFactory.Core().V1().Pods(),
	).Run(ctx, int(controllerContext.ComponentConfig.ReplicaSetController.ConcurrentRSSyncs))
	return nil, true, nil
}

func NewReplicaSetController(rsInformer appsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, kubeClient clientset.Interface, burstReplicas int) *ReplicaSetController {
	return NewBaseController(rsInformer, podInformer, kubeClient, burstReplicas,
		...
	)
}

func NewBaseController(rsInformer appsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, kubeClient clientset.Interface, burstReplicas int,
	gvk schema.GroupVersionKind, metricOwnerName, queueName string, podControl controller.PodControlInterface, eventBroadcaster record.EventBroadcaster) *ReplicaSetController {

	rsc := &ReplicaSetController{}

	rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc:    rsc.addRS,
		UpdateFunc: rsc.updateRS,
		DeleteFunc: rsc.deleteRS,
	})

	podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
		AddFunc: rsc.addPod,
		UpdateFunc: rsc.updatePod,
		DeleteFunc: rsc.deletePod,
	})

	rsc.syncHandler = rsc.syncReplicaSet

	return rsc
}

func (rsc *ReplicaSetController) Run(ctx context.Context, workers int) {
	// 等待 informer 完成同步缓存
	if !cache.WaitForNamedCacheSync(rsc.Kind, ctx.Done(), rsc.podListerSynced, rsc.rsListerSynced) {
		return
	}

	for i := 0; i < workers; i++ {
		go wait.UntilWithContext(ctx, rsc.worker, time.Second)
	}

	<-ctx.Done()
}

func (rsc *ReplicaSetController) worker(ctx context.Context) {
	for rsc.processNextWorkItem(ctx) {
	}
}

func (rsc *ReplicaSetController) processNextWorkItem(ctx context.Context) bool {
	// 从队列获取数据
	key, quit := rsc.queue.Get()
	if quit {
		return false
	}
	defer rsc.queue.Done(key)

	// syncHandler 核心处理函数
	err := rsc.syncHandler(ctx, key.(string))
	if err == nil {
		rsc.queue.Forget(key)
		return true
	}

	// 如果增删 pod 有异常, 则重新推到队列中, 等待下次重试.
	rsc.queue.AddRateLimited(key)

	return true
}

func (rsc *ReplicaSetController) addRS(obj interface{}) {
	// 直接入队列
	rs := obj.(*apps.ReplicaSet)
	rsc.enqueueRS(rs)
}

func (rsc *ReplicaSetController) updateRS(old, cur interface{}) {
	oldRS := old.(*apps.ReplicaSet)
	curRS := cur.(*apps.ReplicaSet)
	...

	// 如果新老配置的 replicas 副本数发生变更, 则打印出来.
	if *(oldRS.Spec.Replicas) != *(curRS.Spec.Replicas) {
		klog.V(4).Infof("%v %v updated. Desired pod count change: %d->%d", rsc.Kind, curRS.Name, *(oldRS.Spec.Replicas), *(curRS.Spec.Replicas))
	}
	// 入队
	rsc.enqueueRS(curRS)
}

func (rsc *ReplicaSetController) deleteRS(obj interface{}) {
	rs, ok := obj.(*apps.ReplicaSet)
	key, err := controller.KeyFunc(rs)

	// 在 expectations 里删除.
	rsc.expectations.DeleteExpectations(key)

	// 入队
	rsc.queue.Add(key)
}

func (rsc *ReplicaSetController) enqueueRS(rs *apps.ReplicaSet) {
	key, err := controller.KeyFunc(rs)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("couldn't get key for object %#v: %v", rs, err))
		return
	}

	rsc.queue.Add(key)
}

func (rsc *ReplicaSetController) addPod(obj interface{}) {
	pod := obj.(*v1.Pod)

	// 如果 pod 字段不为空，则删除.
	if pod.DeletionTimestamp != nil {
		rsc.deletePod(pod)
		return
	}

	if controllerRef := metav1.GetControllerOf(pod); controllerRef != nil {
		rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)
		rsKey, err := controller.KeyFunc(rs)

		// 在 expectations 里把 add 字段加一.
		rsc.expectations.CreationObserved(rsKey)

		// 把 rskey 推到队列里
		rsc.queue.Add(rsKey)
		return
	}

	// 走到这里说明是孤儿 pod, 获取 pod 所属的 rs 集合
	rss := rsc.getPodReplicaSets(pod)
	if len(rss) == 0 {
		return // 没找到对应 rs 则忽略.
	}
	for _, rs := range rss {
		// 把 rs 扔到队列里, 让后面的同步流程去清理
		rsc.enqueueRS(rs)
	}
}

func (rsc *ReplicaSetController) deletePod(obj interface{}) {
	pod, ok := obj.(*v1.Pod)

	controllerRef := metav1.GetControllerOf(pod)
	rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)

	// 在 expectations 里找到 rskey 对应的 exp, 在 del 字段 -1
	rsKey, err := controller.KeyFunc(rs)
	rsc.expectations.DeletionObserved(rsKey, controller.PodKey(pod))

	// 把 rskey 插入到队列里
	rsc.queue.Add(rsKey)
}

func (rsc *ReplicaSetController) updatePod(old, cur interface{}) {
	curPod := cur.(*v1.Pod)
	oldPod := old.(*v1.Pod)
	if curPod.ResourceVersion == oldPod.ResourceVersion {
		return
	}

	// 如果 pod label 改变或者处于删除状态，则直接删除
	labelChanged := !reflect.DeepEqual(curPod.Labels, oldPod.Labels)
	if curPod.DeletionTimestamp != nil {
		rsc.deletePod(curPod)
		if labelChanged {
			rsc.deletePod(oldPod)
		}
		return
	}

	// 如果 pod 的 OwnerReference 发生改变，将 oldRS 入队
	curControllerRef := metav1.GetControllerOf(curPod)
	oldControllerRef := metav1.GetControllerOf(oldPod)
	controllerRefChanged := !reflect.DeepEqual(curControllerRef, oldControllerRef)
	if controllerRefChanged && oldControllerRef != nil {
		if rs := rsc.resolveControllerRef(oldPod.Namespace, oldControllerRef); rs != nil {
			rsc.enqueueRS(rs)
		}
	}

	// 获取 pod 关联的 rs, 入队 rs
	if curControllerRef != nil {
		rs := rsc.resolveControllerRef(curPod.Namespace, curControllerRef)
		if rs == nil {
			return
		}
		rsc.enqueueRS(rs)
		if !podutil.IsPodReady(oldPod) && podutil.IsPodReady(curPod) && rs.Spec.MinReadySeconds > 0 {
			rsc.enqueueRSAfter(rs, (time.Duration(rs.Spec.MinReadySeconds)*time.Second)+time.Second)
		}
		return
	}

	// 查找匹配的 rs
	if labelChanged || controllerRefChanged {
		rss := rsc.getPodReplicaSets(curPod)
		if len(rss) == 0 {
			return
		}
		for _, rs := range rss {
			rsc.enqueueRS(rs)
		}
	}
}

func (rsc *ReplicaSetController) syncReplicaSet(ctx context.Context, key string) error {
	// 从 key 获取 namespace 命令空间 和 name 集合名.
	namespace, name, err := cache.SplitMetaNamespaceKey(key)
	if err != nil {
		return err
	}

	// 通过 namespace 和 name 从 informer cache 里获取 rs 对象.
	rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)

	// 如果 rs 已经被删除，则尝试在 expectations 清空对应的记录.
	if apierrors.IsNotFound(err) {
		rsc.expectations.DeleteExpectations(key)
		return nil
	}
	if err != nil {
		return err
	}

	// 判断下是否需要 manageReplicas 同步, 后面有详细描述.
	rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
	selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
	if err != nil {
		return nil
	}

	// 获取相关的 pods 集合.
	allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
	if err != nil {
		return err
	}

	// 过滤激活状态的 pods.
	filteredPods := controller.FilterActivePods(allPods)
	filteredPods, err = rsc.claimPods(ctx, rs, selector, filteredPods)
	if err != nil {
		return err
	}

	var manageReplicasErr error
	if rsNeedsSync && rs.DeletionTimestamp == nil {
		manageReplicasErr = rsc.manageReplicas(ctx, filteredPods, rs)
	}

	// 计算 rs 的状态.
	newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)

	// 向 apiserver 更新新的 rs 的状态.
	updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)
	if err != nil {
		return err
	}

	return manageReplicasErr
}

func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {
	// 如果有 exp 对象, 那么判断是否满足条件或者超过同步周期, 则返回 true
	if exp, exists, err := r.GetExpectations(controllerKey); exists {
		if exp.Fulfilled() {
			return true
		} else if exp.isExpired() {
			return true
		} else {
			// 其他情况则返回 false
			return false
		}
	} else if err != nil {
		// 预期检查失败
	} else {
		// 如果没有 exp 对象，也就是说从没创建过 rs, 则返回 true
	}
	return true
}

// 距离上次时间超过阈值为 5 分钟为超时
func (exp *ControlleeExpectations) isExpired() bool {
	return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout
}

// 当 add 和 del 都 <= 0 时, 当前集群已经达到了预期状态.
func (e *ControlleeExpectations) Fulfilled() bool {
	// TODO: think about why this line being atomic doesn't matter
	return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0
}

func (rsc *ReplicaSetController) manageReplicas(ctx context.Context, filteredPods []*v1.Pod, rs *apps.ReplicaSet) error {
	// 当前的副本数 - 配置里预期的副本数
	diff := len(filteredPods) - int(*(rs.Spec.Replicas))
	rsKey, err := controller.KeyFunc(rs)
	if err != nil {
		return nil
	}
	// 如果 < 0, 说明还有副本需要创建
	if diff < 0 {
		diff *= -1 // 负数转成正数
		if diff > rsc.burstReplicas { // 最大不能超过 500, 单次创建不能超过 500.
			diff = rsc.burstReplicas
		}

		// 创建一个 expections 对象, 其 add 值为 diff
		rsc.expectations.ExpectCreations(rsKey, diff)

		// 调用 slowStartBatch 来并发分批的创建 pods, 返回成功实例化 pods 的数量
		successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {
			err := rsc.podControl.CreatePods(ctx, rs.Namespace, &rs.Spec.Template, rs, metav1.NewControllerRef(rs, rsc.GroupVersionKind))
			if err != nil {
				...
			}
			return err
		})

		// diff - 成功的数量等于 skippedPods
		if skippedPods := diff - successfulCreations; skippedPods > 0 {
			for i := 0; i < skippedPods; i++ {
				// 给 exp 结构的 add 字段 atomic add 加一
				rsc.expectations.CreationObserved(rsKey)
			}
		}
		return err
	// 如果 > 0 ，说明当前副本数超过了预期, 需要删除多出来的副本.
	} else if diff > 0 {
		if diff > rsc.burstReplicas { // 一次对多可以处理 500
			diff = rsc.burstReplicas
		}

		relatedPods, err := rsc.getIndirectlyRelatedPods(rs)

		// 获取可以被清理的 pods
		podsToDelete := getPodsToDelete(filteredPods, relatedPods, diff)

		// 在发起删除操作之前, 先在 expectations 记录 del 字段
		rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))

		// 并发去删除 pod, 如果删除成功，则在 expectations 里 del 字段减一.
		var wg sync.WaitGroup
		wg.Add(diff)
		// 删除就没有创建来的优雅，直接遍历开协程去删除.
		for _, pod := range podsToDelete {
			go func(targetPod *v1.Pod) {
				defer wg.Done()
				if err := rsc.podControl.DeletePod(ctx, rs.Namespace, targetPod.Name, rs); err != nil {
					podKey := controller.PodKey(targetPod)
					rsc.expectations.DeletionObserved(rsKey, podKey)
					...
				}
			}(pod)
		}
		wg.Wait()
	}

	return nil
}

func slowStartBatch(count int, initialBatchSize int, fn func() error) (int, error) {
	remaining := count
	successes := 0
	for batchSize := integer.IntMin(remaining, initialBatchSize); batchSize > 0; batchSize = integer.IntMin(2*batchSize, remaining) {
		errCh := make(chan error, batchSize)
		var wg sync.WaitGroup
		wg.Add(batchSize)
		for i := 0; i < batchSize; i++ {
			go func() {
				defer wg.Done()
				if err := fn(); err != nil {
					errCh <- err
				}
			}()
		}
		wg.Wait()
		curSuccesses := batchSize - len(errCh)
		successes += curSuccesses
		if len(errCh) > 0 {
			return successes, <-errCh
		}
		remaining -= batchSize
	}
	return successes, nil
}

func (r RealPodControl) CreatePodsWithGenerateName(ctx context.Context, namespace string, template *v1.PodTemplateSpec, controllerObject runtime.Object, controllerRef *metav1.OwnerReference, generateName string) error {
	// 通过模板创建 pod 对应
	pod, err := GetPodFromTemplate(template, controllerObject, controllerRef)
	if err != nil {
		return err
	}
	return r.createPods(ctx, namespace, pod, controllerObject)
}

func (r RealPodControl) createPods(ctx context.Context, namespace string, pod *v1.Pod, object runtime.Object) error {
	// 通过调用 clientset 的 create 创建 pod.
	newPod, err := r.KubeClient.CoreV1().Pods(namespace).Create(ctx, pod, metav1.CreateOptions{})
	if err != nil {
		...
		return err
	}
	...
	return nil
}

type ActivePodsWithRanks struct {
	Pods []*v1.Pod
	Rank []int
}

// 实现 less 排序比较接口，调用方可以使用 sort.Sork 进行排序.
func (s ActivePodsWithRanks) Less(i, j int) bool {
	// 1. Unassigned < assigned
	// If only one of the pods is unassigned, the unassigned one is smaller
	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 only one of the pods is not ready, the not ready one is smaller
	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) {
	}

	// 5. Doubled up < not doubled up
	// If one of the two pods is on the same node as one or more additional
	// ready pods that belong to the same replicaset, whichever pod has more
	// colocated ready pods is less
	if s.Rank[i] != s.Rank[j] {
		return s.Rank[i] > s.Rank[j]
	}
	// 6. Been ready for empty time < less time < more time
	// If both pods are ready, the latest ready one is smaller
	if podutil.IsPodReady(s.Pods[i]) && podutil.IsPodReady(s.Pods[j]) {
	}

	// 7. Pods with containers with 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) {
	}
	return false
}

func getPodsToDelete(filteredPods, relatedPods []*v1.Pod, diff int) []*v1.Pod {
	if diff < len(filteredPods) {
		// 对同一个节点下 pods 进行 rank 计分 
		podsWithRanks := getPodsRankedByRelatedPodsOnSameNode(filteredPods, relatedPods)
		// 对已实现了排序接口的对象进行排序
		sort.Sort(podsWithRanks)
	}
	return filteredPods[:diff]
}

func (r RealPodControl) DeletePod(ctx context.Context, namespace string, podID string, object runtime.Object) error {
	if err := r.KubeClient.CoreV1().Pods(namespace).Delete(ctx, podID, metav1.DeleteOptions{}); err != nil {
		return fmt.Errorf("unable to delete pods: %v", err)
	}
	...
	return nil
}

// 字段 add 期望值需要减少一个 pod
CreationObserved(controllerKey string)

// 字段 del 期望值需要减少一个 pod
DeletionObserved(controllerKey string)

// 写入 key 需要 add 的 pod 数量
ExpectCreations(controllerKey string, adds int) error

// 写入 key 需要 del 的 pod 数量
ExpectDeletions(controllerKey string, dels int) error

// 删除该 key
DeleteExpectations(controllerKey string)