专栏首页田飞雨的专栏replicaset controller 源码分析
原创

replicaset controller 源码分析

在前面的文章中已经介绍了 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

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

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

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

func NewReplicaSetController(......) *ReplicaSetController {
    ......
    // 1、此处调用 NewBaseController
    return NewBaseController(rsInformer, podInformer, kubeClient, burstReplicas,
        apps.SchemeGroupVersion.WithKind("ReplicaSet"),
        "replicaset_controller",
        "replicaset",
        controller.RealPodControl{
            KubeClient: kubeClient,
            Recorder:   eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "replicaset-controller"}),
        },
    )
}

func NewBaseController(......) *ReplicaSetController {
    ......
    // 2、ReplicaSetController 初始化
    rsc := &ReplicaSetController{
        GroupVersionKind: gvk,
        kubeClient:       kubeClient,
        podControl:       podControl,
        burstReplicas:    burstReplicas,
        // 3、expectations 的初始化
        expectations:     controller.NewUIDTrackingControllerExpectations(controller.NewControllerExpectations()),
        queue:            workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), queueName),
    }

    // 4、rsInformer 中注册的 EventHandler
    rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    rsc.enqueueReplicaSet,
        UpdateFunc: rsc.updateRS,
        DeleteFunc: rsc.enqueueReplicaSet,
    })
    ......

    // 5、podInformer 中注册的 EventHandler
    podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: rsc.addPod,
        UpdateFunc: rsc.updatePod,
        DeleteFunc: rsc.deletePod,
    })
    ......

    return rsc
}

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

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

func (rsc *ReplicaSetController) Run(workers int, stopCh <-chan struct{}) {
    ......

    // 1、等待 informer 同步缓存
    if !cache.WaitForNamedCacheSync(rsc.Kind, stopCh, rsc.podListerSynced, rsc.rsListerSynced) {
        return
    }

    // 2、启动 5 个 goroutine 执行 worker 方法
    for i := 0; i < workers; i++ {
        go wait.Until(rsc.worker, time.Second, stopCh)
    }

    <-stopCh
}

// 3、worker 方法中调用 rocessNextWorkItem
func (rsc *ReplicaSetController) worker() {
    for rsc.processNextWorkItem() {
    }
}

func (rsc *ReplicaSetController) processNextWorkItem() bool {
    // 4、从队列中取出对象
    key, quit := rsc.queue.Get()
    if quit {
        return false
    }
    defer rsc.queue.Done(key)

    // 5、执行 sync 操作
    err := rsc.syncHandler(key.(string))
    ......

    return true
}

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

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

    if pod.DeletionTimestamp != nil {
        rsc.deletePod(pod)
        return
    }

    // 1、获取 pod 所关联的 rs
    if controllerRef := metav1.GetControllerOf(pod); controllerRef != nil {
        rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)
        if rs == nil {
            return
        }
        rsKey, err := controller.KeyFunc(rs)
        if err != nil {
            return
        }
        // 2、更新 expectations,rsKey 的 add - 1
        rsc.expectations.CreationObserved(rsKey)
        rsc.enqueueReplicaSet(rs)
        return
    }


    rss := rsc.getPodReplicaSets(pod)
    if len(rss) == 0 {
        return
    }

    for _, rs := range rss {
        rsc.enqueueReplicaSet(rs)
    }
}
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

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

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

    // 2、如果 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.enqueueReplicaSet(rs)
        }
    }

    // 3、获取 pod 关联的 rs,入队 rs
    if curControllerRef != nil {
        rs := rsc.resolveControllerRef(curPod.Namespace, curControllerRef)
        if rs == nil {
            return
        }

        rsc.enqueueReplicaSet(rs)
        if !podutil.IsPodReady(oldPod) && podutil.IsPodReady(curPod) && rs.Spec.MinReadySeconds > 0 {
            rsc.enqueueReplicaSetAfter(rs, (time.Duration(rs.Spec.MinReadySeconds)*time.Second)+time.Second)
        }
        return
    }


    // 4、查找匹配的 rs
    if labelChanged || controllerRefChanged {
        rss := rsc.getPodReplicaSets(curPod)
        if len(rss) == 0 {
            return
        }
        for _, rs := range rss {
            rsc.enqueueReplicaSet(rs)
        }
    }
}
deletePod
  • 1、确认该对象是否为 pod;
  • 2、判断是否为孤儿 pod;
  • 3、获取其对应的 rs 以及 rsKey;
  • 4、更新 expectations 中 rsKey 的 del 值;
  • 5、将 rs 入队;

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

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

    if !ok {
        ......
    }

    controllerRef := metav1.GetControllerOf(pod)
    if controllerRef == nil {
        return
    }
    rs := rsc.resolveControllerRef(pod.Namespace, controllerRef)
    if rs == nil {
        return
    }
    rsKey, err := controller.KeyFunc(rs)
    if err != nil {
        return
    }
    // 更新 expectations,该 rsKey 的 del - 1
    rsc.expectations.DeletionObserved(rsKey, controller.PodKey(pod))
    rsc.enqueueReplicaSet(rs)
}
AddRS 和 DeleteRS

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

#####UpdateRS

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

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

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

    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.enqueueReplicaSet(cur)
}

至于 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

func (rsc *ReplicaSetController) syncReplicaSet(key string) error {
		......

    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }

    // 1、根据 ns/name 从 informer cache 中获取 rs 对象,
    // 若 rs 已经被删除则直接删除 expectations 中的对象
    rs, err := rsc.rsLister.ReplicaSets(namespace).Get(name)
    if errors.IsNotFound(err) {
        rsc.expectations.DeleteExpectations(key)
        return nil
    }
    ......

    // 2、判断该 rs 是否需要执行 sync 操作
    rsNeedsSync := rsc.expectations.SatisfiedExpectations(key)
    selector, err := metav1.LabelSelectorAsSelector(rs.Spec.Selector)
    if err != nil {
        ......
    }

    // 3、获取所有 pod list
    allPods, err := rsc.podLister.Pods(rs.Namespace).List(labels.Everything())
		......

    // 4、过滤掉异常 pod,处于删除状态或者 failed 状态的 pod 都为非 active 状态
    filteredPods := controller.FilterActivePods(allPods)

    // 5、检查所有 pod,根据 pod 并进行 adopt 与 release 操作,最后获取与该 rs 关联的 pod list
    filteredPods, err = rsc.claimPods(rs, selector, filteredPods)
    ......

    // 6、若需要 sync 则执行 manageReplicas 创建/删除 pod
    var manageReplicasErr error
    if rsNeedsSync && rs.DeletionTimestamp == nil {
        manageReplicasErr = rsc.manageReplicas(filteredPods, rs)
    }
    rs = rs.DeepCopy()
    // 7、计算 rs 当前的 status
    newStatus := calculateStatus(rs, filteredPods, manageReplicasErr)

    // 8、更新 rs status
    updatedRS, err := updateReplicaSetStatus(rsc.kubeClient.AppsV1().ReplicaSets(rs.Namespace), rs, newStatus)


    // 9、判断是否需要将 rs 加入到延迟队列中
    if manageReplicasErr == nil && updatedRS.Spec.MinReadySeconds > 0 &&
        updatedRS.Status.ReadyReplicas == *(updatedRS.Spec.Replicas) &&
        updatedRS.Status.AvailableReplicas != *(updatedRS.Spec.Replicas) {
        rsc.enqueueReplicaSetAfter(updatedRS, time.Duration(updatedRS.Spec.MinReadySeconds)*time.Second)
    }
    return manageReplicasErr
}

syncReplicaSet 方法中有几个重要的操作分别为:rsc.expectations.SatisfiedExpectationsrsc.manageReplicascalculateStatus,下面一一进行分析。

SatisfiedExpectations

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

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

func (r *ControllerExpectations) SatisfiedExpectations(controllerKey string) bool {
    // 1、若该 key 存在时,判断是否满足条件或者是否超过同步周期
    if exp, exists, err := r.GetExpectations(controllerKey); exists {
        if exp.Fulfilled() {
            return true
        } else if exp.isExpired() {
            return true
        } else {
            return false
        }
    } else if err != nil {
		......
    } else {
        // 2、该 rs 可能为新创建的,需要进行 sync
        ......
    }
    return true
}

// 3、若 add <= 0 且 del <= 0 说明本地观察到的状态已经为期望状态了
func (e *ControlleeExpectations) Fulfilled() bool {
    return atomic.LoadInt64(&e.add) <= 0 && atomic.LoadInt64(&e.del) <= 0
}

// 4、判断 key 是否过期,ExpectationsTimeout 默认值为 5 * time.Minute
func (exp *ControlleeExpectations) isExpired() bool {
    return clock.RealClock{}.Since(exp.timestamp) > ExpectationsTimeout
}
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

func (rsc *ReplicaSetController) manageReplicas(......) error {
    // 1、计算已存在 pod 数与期望数的差异
    diff := len(filteredPods) - int(*(rs.Spec.Replicas))
    rsKey, err := controller.KeyFunc(rs)
    if err != nil {
        ......
    }
    2、如果 <0,则需要创建 pod
    if diff < 0 {
        diff *= -1
        3、判断需要创建的 pod 数是否超过单次 sync 上限值 500
        if diff > rsc.burstReplicas {
            diff = rsc.burstReplicas
        }

        4、在 expectations 中进行记录,若该 key 已经存在会进行覆盖
        rsc.expectations.ExpectCreations(rsKey, diff)

        5、调用 slowStartBatch 创建所需要的 pod
        successfulCreations, err := slowStartBatch(diff, controller.SlowStartInitialBatchSize, func() error {
            err := rsc.podControl.CreatePodsWithControllerRef(rs.Namespace, &rs.Spec.Template, rs, metav1.NewControllerRef(rs, rsc.GroupVersionKind))
            // 6、若为 timeout err 则忽略
            if err != nil && errors.IsTimeout(err) {
                return nil
            }
            return err
        })

        // 7、计算未创建的 pod 数,并记录在 expectations 中
		// 若 pod 创建成功,informer watch 到事件后会在 addPod handler 中更新 expectations
        if skippedPods := diff - successfulCreations; skippedPods > 0 {
            for i := 0; i < skippedPods; i++ {
                rsc.expectations.CreationObserved(rsKey)
            }
        }
        return err
    } else if diff > 0 {
    	// 8、若 diff >0 说明需要删除多创建的 pod
        if diff > rsc.burstReplicas {
            diff = rsc.burstReplicas
        }

		// 9、getPodsToDelete 会按照一定的策略找出需要删除的 pod 列表
        podsToDelete := getPodsToDelete(filteredPods, diff)

		// 10、在 expectations 中进行记录,若该 key 已经存在会进行覆盖
        rsc.expectations.ExpectDeletions(rsKey, getPodKeys(podsToDelete))

        // 11、进行并发删除的操作
        errCh := make(chan error, diff)
        var wg sync.WaitGroup
        wg.Add(diff)
        for _, pod := range podsToDelete {
            go func(targetPod *v1.Pod) {
                defer wg.Done()
                if err := rsc.podControl.DeletePod(rs.Namespace, targetPod.Name, rs); err != nil {
                    podKey := controller.PodKey(targetPod)
					// 12、某次删除操作若失败会记录在 expectations 中
                    rsc.expectations.DeletionObserved(rsKey, podKey)
                    errCh <- err
                }
            }(pod)
        }
        wg.Wait()

		// 13、返回其中一条 err
        select {
        case err := <-errCh:
            if err != nil {
                return err
            }
        default:
        }
    }

    return nil
}

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

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

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
}

若 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

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

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

type ActivePods []*v1.Pod

func (s ActivePods) Len() int      { return len(s) }
func (s ActivePods) Swap(i, j int) { s[i], s[j] = s[j], s[i] }

func (s ActivePods) Less(i, j int) bool {
    // 1. Unassigned < assigned
    if s[i].Spec.NodeName != s[j].Spec.NodeName && (len(s[i].Spec.NodeName) == 0 || len(s[j].Spec.NodeName) == 0) {
        return len(s[i].Spec.NodeName) == 0
    }

    // 2. PodPending < PodUnknown < PodRunning
    m := map[v1.PodPhase]int{v1.PodPending: 0, v1.PodUnknown: 1, v1.PodRunning: 2}
    if m[s[i].Status.Phase] != m[s[j].Status.Phase] {
        return m[s[i].Status.Phase] < m[s[j].Status.Phase]
    }

    // 3. Not ready < ready
    if podutil.IsPodReady(s[i]) != podutil.IsPodReady(s[j]) {
        return !podutil.IsPodReady(s[i])
    }

    // 4. Been ready for empty time < less time < more time
    if podutil.IsPodReady(s[i]) && podutil.IsPodReady(s[j]) && !podReadyTime(s[i]).Equal(podReadyTime(s[j])) {
        return afterOrZero(podReadyTime(s[i]), podReadyTime(s[j]))
    }

    // 5. Pods with containers with higher restart counts < lower restart counts
    if maxContainerRestarts(s[i]) != maxContainerRestarts(s[j]) {
        return maxContainerRestarts(s[i]) > maxContainerRestarts(s[j])
    }

    // 6. Empty creation time pods < newer pods < older pods
    if !s[i].CreationTimestamp.Equal(&s[j].CreationTimestamp) {
        return afterOrZero(&s[i].CreationTimestamp, &s[j].CreationTimestamp)
    }
    return false
}
calculateStatus

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

status:
  availableReplicas: 10
  fullyLabeledReplicas: 10
  observedGeneration: 1
  readyReplicas: 10
  replicas: 10

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

func calculateStatus(......) apps.ReplicaSetStatus {
    newStatus := rs.Status
    fullyLabeledReplicasCount := 0
    readyReplicasCount := 0
    availableReplicasCount := 0
    templateLabel := labels.Set(rs.Spec.Template.Labels).AsSelectorPreValidated()
    for _, pod := range filteredPods {
        if templateLabel.Matches(labels.Set(pod.Labels)) {
            fullyLabeledReplicasCount++
        }
        if podutil.IsPodReady(pod) {
            readyReplicasCount++
            if podutil.IsPodAvailable(pod, rs.Spec.MinReadySeconds, metav1.Now()) {
                availableReplicasCount++
            }
        }
    }

    failureCond := GetCondition(rs.Status, apps.ReplicaSetReplicaFailure)
    if manageReplicasErr != nil && failureCond == nil {
        var reason string
        if diff := len(filteredPods) - int(*(rs.Spec.Replicas)); diff < 0 {
            reason = "FailedCreate"
        } else if diff > 0 {
            reason = "FailedDelete"
        }
        cond := NewReplicaSetCondition(apps.ReplicaSetReplicaFailure, v1.ConditionTrue, reason, manageReplicasErr.Error())
        SetCondition(&newStatus, cond)
    } else if manageReplicasErr == nil && failureCond != nil {
        RemoveCondition(&newStatus, apps.ReplicaSetReplicaFailure)
    }

    newStatus.Replicas = int32(len(filteredPods))
    newStatus.FullyLabeledReplicas = int32(fullyLabeledReplicasCount)
    newStatus.ReadyReplicas = int32(readyReplicasCount)
    newStatus.AvailableReplicas = int32(availableReplicasCount)
    return newStatus
}

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 中用到的几个方法:

 // 创建了一个 pod 说明 expectations 中对应的 key add 期望值需要减少一个 pod, add -1
 CreationObserved(controllerKey string)

 // 删除了一个 pod 说明 expectations 中对应的 key del 期望值需要减少一个 pod, del - 1
 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)

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

总结

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

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

                                    SatisfiedExpectations
                                    (expectations 中不存在
                                     rsKey,rsNeedsSync
                                     为 true)
                                              |              判断 add/del pod
                                              |                     |
                                              |                     ∨
                                              |             创建 expectations 对象,
                                              |             并设置 add/del 值
                                              ∨                     |
create rs --> syncReplicaSet -->       manageReplicas  -->          ∨
                                       (为 rs 创建 pod)       调用 slowStartBatch 批量创建 pod/
                                              |               删除筛选出的多余 pod
                                              |                     |
                                              |                     ∨
                                              |               更新 expectations 对象
                                              ∨
                                    updateReplicaSetStatus
                                    (更新 rs 的 status
                                    subResource)

参考:

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

原创声明,本文系作者授权云+社区发表,未经许可,不得转载。

如有侵权,请联系 yunjia_community@tencent.com 删除。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 干货满满的 Go Modules 和 goproxy.cn

    大家好,我是一只普通的煎鱼,周四晚上很有幸邀请到 goproxy.cn 的作者 @盛傲飞(@aofei) 到 Go 夜读给我们进行第 61 期 《Go Modu...

    madneal
  • GShark-监测你的 Github 敏感信息泄露

    近几年由于 Github 信息泄露导致的信息安全事件屡见不鲜,且规模越来越大。就前段时间华住集团旗下酒店开房记录疑似泄露,涉及近5亿个人信息。后面调查发现疑似是...

    madneal
  • PHP 闭包及Closure类

    上面就是一个简单的匿名函数,定义一个函数体,将函数体赋值给一个变量(php5.3之后支持该写法)。

    猿哥
  • 你再也不用设置 GOROOT 了

    为什么不再需要设置 GOROOT 呢?推荐读两篇英文文章,我意译了下,将它们放在了一篇里。

    波罗学
  • 真香系列之 Golang 升级

    Golang 以前的依赖管理一直饱受诟病,社区的方案也层出不穷,比如 vendor, glide, godep 等。之前的依赖管理一直是依靠 GOPATH 或者...

    madneal
  • 比原链CTO James | Go语言成为区块链主流开发语言的四点理由

    11月24日,比原链CTO James参加了Go中国举办的Gopher Meetup杭州站活动,与来自阿里、网易的技术专家带来Kubernetes、区块链、日志...

    比原链Bytom
  • 如何快速理解go的并发?【Golang 入门系列十五】

    Go是并发语言,而不是并行语言。所以我们在讨论,我们首先必须了解什么是并发,以及它与并行性有什么不同。

    章为忠
  • mac 安装 selenium+chromedriver mac64位自动脚本下载安装

    在网络上折腾了半天都没有找到好办法,最后使用了下列脚本运行后成功安装好mac下的selenium+chrome环境。实验时间–2016-12-20.附mac下c...

    十四君
  • DeepMind的机器学习模型MuZero是如何学会下棋的

    DeepMind在去年年底发表在《科学》杂志上的一篇论文中,详细介绍了AlphaZero,这是一种人工智能系统,可以自学如何掌握国际象棋、日本象棋“将棋”的变体...

    AiTechYun
  • gRPC- Go和Java的一次HelloWorld

    都说grpc是跨语言的一个rpc框架,当团队内部有多种流行编程语言时,那么grpc可以为他们提供通信,今天我们就通过一个Hello World来看看Java和G...

    阿伟

扫码关注云+社区

领取腾讯云代金券