daemonset controller 源码分析

在前面的文章中已经分析过 deployment、statefulset 两个重要对象了,本文会继续分析 kubernetes 中另一个重要的对象 daemonset,在 kubernetes 中 daemonset 类似于 linux 上的守护进程会运行在每一个 node 上,在实际场景中,一般会将日志采集或者网络插件采用 daemonset 的方式部署。

DaemonSet 的基本操作

创建

daemonset 在创建后会在每个 node 上都启动一个 pod。

$ kubectl create -f nginx-ds.yaml

扩缩容

由于 daemonset 是在每个 node 上启动一个 pod,其不存在扩缩容操作,副本数量跟 node 数量保持一致。

更新

daemonset 有两种更新策略 OnDeleteRollingUpdate,默认为 RollingUpdate。滚动更新时,需要指定 .spec.updateStrategy.rollingUpdate.maxUnavailable(默认为1)和 .spec.minReadySeconds(默认为 0)。

// 更新镜像
$ kubectl set image ds/nginx-ds nginx-ds=nginx:1.16

// 查看更新状态
$ kubectl rollout status ds/nginx-ds

回滚

在 statefulset 源码分析一节已经提到过 controllerRevision 这个对象了,其主要用来保存历史版本信息,在更新以及回滚操作时使用,daemonset controller 也是使用 controllerrevision 保存历史版本信息,在回滚时会使用历史 controllerrevision 中的信息替换 daemonset 中 Spec.Template

// 查看 ds 历史版本信息
$ kubectl get controllerrevision
NAME                  CONTROLLER                REVISION   AGE
nginx-ds-5c4b75bdbb   daemonset.apps/nginx-ds   2          122m
nginx-ds-7cd7798dcd   daemonset.apps/nginx-ds   1          133m

// 回滚到版本 1
$ kubectl rollout undo daemonset nginx-ds --to-revision=1

// 查看回滚状态
$ kubectl rollout status ds/nginx-ds

暂停

daemonset 目前不支持暂停操作。

删除

daemonset 也支持两种删除操作。

// 非级联删除
$ kubectl delete ds/nginx-ds --cascade=false

// 级联删除
$ kubectl delete ds/nginx-ds

DaemonSetController 源码分析

kubernetes 版本:v1.16

首先还是看 startDaemonSetController 方法,在此方法中会初始化 DaemonSetsController 对象并调用 Run方法启动 daemonset controller,从该方法中可以看出 daemonset controller 会监听 daemonsetscontrollerRevisionpodnode 四种对象资源的变动。其中 ConcurrentDaemonSetSyncs的默认值为 2。

k8s.io/kubernetes/cmd/kube-controller-manager/app/apps.go:36

func startDaemonSetController(ctx ControllerContext) (http.Handler, bool, error) {
    if !ctx.AvailableResources[schema.GroupVersionResource{Group: "apps", Version: "v1", Resource: "daemonsets"}] {
        return nil, false, nil
    }
    dsc, err := daemon.NewDaemonSetsController(
        ctx.InformerFactory.Apps().V1().DaemonSets(),
        ctx.InformerFactory.Apps().V1().ControllerRevisions(),
        ctx.InformerFactory.Core().V1().Pods(),
        ctx.InformerFactory.Core().V1().Nodes(),
        ctx.ClientBuilder.ClientOrDie("daemon-set-controller"),
        flowcontrol.NewBackOff(1*time.Second, 15*time.Minute),
    )
    if err != nil {
        return nil, true, fmt.Errorf("error creating DaemonSets controller: %v", err)
    }
    go dsc.Run(int(ctx.ComponentConfig.DaemonSetController.ConcurrentDaemonSetSyncs), ctx.Stop)
    return nil, true, nil
}

Run 方法中会启动两个操作,一个就是 dsc.runWorker 执行的 sync 操作,另一个就是 dsc.failedPodsBackoff.GC 执行的 gc 操作,主要逻辑为:

  • 1、等待 informer 缓存同步完成;
  • 2、启动两个 goroutine 分别执行 dsc.runWorker
  • 3、启动一个 goroutine 每分钟执行一次 dsc.failedPodsBackoff.GC,从 startDaemonSetController 方法中可以看到 failedPodsBackoff 的 duration为1s,max duration为15m,failedPodsBackoff 的主要作用是当发现 daemon pod 状态为 failed 时,会定时重启该 pod;

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:263

func (dsc *DaemonSetsController) Run(workers int, stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    defer dsc.queue.ShutDown()

    defer klog.Infof("Shutting down daemon sets controller")

    if !cache.WaitForNamedCacheSync("daemon sets", stopCh, dsc.podStoreSynced, dsc.nodeStoreSynced, dsc.historyStoreSynced, dsc.dsStoreSynced) {
        return
    }

    for i := 0; i < workers; i++ {
        // sync 操作
        go wait.Until(dsc.runWorker, time.Second, stopCh)
    }
        
    // GC 操作
    go wait.Until(dsc.failedPodsBackoff.GC, BackoffGCInterval, stopCh)

    <-stopCh
}

syncDaemonSet

daemonset 中 pod 的创建与删除是与 node 相关联的,所以每次执行 sync 操作时需要遍历所有的 node 进行判断。syncDaemonSet 的主要逻辑为:

  • 1、通过 key 获取 ns 和 name;
  • 2、从 dsLister 中获取 ds 对象;
  • 3、从 nodeLister 获取所有 node;
  • 4、获取 dsKey;
  • 5、判断 ds 是否处于删除状态;
  • 6、调用 constructHistory 获取 current 和 old controllerRevision
  • 7、调用 dsc.expectations.SatisfiedExpectations 判断是否满足 expectations 机制,expectations 机制的目的就是减少不必要的 sync 操作,关于 expectations 机制的详细说明可以参考笔者以前写的 "replicaset controller 源码分析"一文;
  • 8、调用 dsc.manage 执行实际的 sync 操作;
  • 9、判断是否为更新操作,并执行对应的更新操作逻辑;
  • 10、调用 dsc.cleanupHistory 根据 spec.revisionHistoryLimit字段清理过期的 controllerrevision
  • 11、调用 dsc.updateDaemonSetStatus 更新 ds 状态;

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:1212

func (dsc *DaemonSetsController) syncDaemonSet(key string) error {
    ......

    // 1、通过 key 获取 ns 和 name
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }
    
    // 2、从 dsLister 中获取 ds 对象
    ds, err := dsc.dsLister.DaemonSets(namespace).Get(name)
    if errors.IsNotFound(err) {
        dsc.expectations.DeleteExpectations(key)
        return nil
    }
    ......

    // 3、从 nodeLister 获取所有 node
    nodeList, err := dsc.nodeLister.List(labels.Everything())
        ......

    everything := metav1.LabelSelector{}
    if reflect.DeepEqual(ds.Spec.Selector, &everything) {
        dsc.eventRecorder.Eventf(ds, v1.EventTypeWarning, SelectingAllReason, "This daemon set is selecting all pods. A non-empty selector is required. ")
        return nil
    }
    
    // 4、获取 dsKey
    dsKey, err := controller.KeyFunc(ds)
    if err != nil {
        return fmt.Errorf("couldn't get key for object %#v: %v", ds, err)
    }

    // 5、判断 ds 是否处于删除状态
    if ds.DeletionTimestamp != nil {
        return nil
    }

    // 6、获取 current 和 old controllerRevision
    cur, old, err := dsc.constructHistory(ds)
    if err != nil {
        return fmt.Errorf("failed to construct revisions of DaemonSet: %v", err)
    }
    hash := cur.Labels[apps.DefaultDaemonSetUniqueLabelKey]
    
    // 7、判断是否满足 expectations 机制
    if !dsc.expectations.SatisfiedExpectations(dsKey) {
        return dsc.updateDaemonSetStatus(ds, nodeList, hash, false)
    }

    // 8、执行实际的 sync 操作
    err = dsc.manage(ds, nodeList, hash)
    if err != nil {
        return err
    }

    // 9、判断是否为更新操作,并执行对应的更新操作
    if dsc.expectations.SatisfiedExpectations(dsKey) {
        switch ds.Spec.UpdateStrategy.Type {
        case apps.OnDeleteDaemonSetStrategyType:
        case apps.RollingUpdateDaemonSetStrategyType:
            err = dsc.rollingUpdate(ds, nodeList, hash)
        }
        if err != nil {
            return err
        }
    }
    // 10、清理过期的 controllerrevision
    err = dsc.cleanupHistory(ds, old)
    if err != nil {
        return fmt.Errorf("failed to clean up revisions of DaemonSet: %v", err)
    }

    // 11、更新 ds 状态
    return dsc.updateDaemonSetStatus(ds, nodeList, hash, true)
}

syncDaemonSet 中主要有 managerollingUpdateupdateDaemonSetStatus 三个方法,分别对应创建、更新与状态同步,下面主要来分析这三个方法。

manage

manage 主要是用来保证 ds 的 pod 数正常运行在每一个 node 上,其主要逻辑为:

  • 1、调用 dsc.getNodesToDaemonPods 获取已存在 daemon pod 与 node 的映射关系;
  • 2、遍历所有 node,调用 dsc.podsShouldBeOnNode 方法来确定在给定的节点上需要创建还是删除 daemon pod;
  • 3、判断是否启动了 ScheduleDaemonSetPodsfeature-gates 特性,若启动了则需要删除通过默认调度器已经调度到不存在 node 上的 daemon pod;
  • 4、调用 dsc.syncNodes 为对应的 node 创建 daemon pod 以及删除多余的 pods;

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:952

func (dsc *DaemonSetsController) manage(ds *apps.DaemonSet, nodeList []*v1.Node, hash string) error {
    // 1、获取已存在 daemon pod 与 node 的映射关系
    nodeToDaemonPods, err := dsc.getNodesToDaemonPods(ds)
    ......

    // 2、判断每一个 node 是否需要运行 daemon pod
    var nodesNeedingDaemonPods, podsToDelete []string
    for _, node := range nodeList {
        nodesNeedingDaemonPodsOnNode, podsToDeleteOnNode, err := dsc.podsShouldBeOnNode(
            node, nodeToDaemonPods, ds)

        if err != nil {
            continue
        }

        nodesNeedingDaemonPods = append(nodesNeedingDaemonPods, nodesNeedingDaemonPodsOnNode...)
        podsToDelete = append(podsToDelete, podsToDeleteOnNode...)
    }

    // 3、判断是否启动了 ScheduleDaemonSetPods feature-gates 特性,若启用了则对不存在 node 上的 
    // daemon pod 进行删除 
    if utilfeature.DefaultFeatureGate.Enabled(features.ScheduleDaemonSetPods) {
        podsToDelete = append(podsToDelete, getUnscheduledPodsWithoutNode(nodeList, nodeToDaemonPods)...)
    }

    // 4、为对应的 node 创建 daemon pod 以及删除多余的 pods
    if err = dsc.syncNodes(ds, podsToDelete, nodesNeedingDaemonPods, hash); err != nil {
        return err
    }

    return nil
}

manage 方法中又调用了 getNodesToDaemonPodspodsShouldBeOnNodesyncNodes 三个方法,继续来看这几种方法的作用。

getNodesToDaemonPods

getNodesToDaemonPods 是用来获取已存在 daemon pod 与 node 的映射关系,并且会通过 adopt/orphan 方法关联以及释放对应的 pod。

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:820

func (dsc *DaemonSetsController) getNodesToDaemonPods(ds *apps.DaemonSet) (map[string][]*v1.Pod, error) {
    claimedPods, err := dsc.getDaemonPods(ds)
    if err != nil {
        return nil, err
    }
    nodeToDaemonPods := make(map[string][]*v1.Pod)
    for _, pod := range claimedPods {
        nodeName, err := util.GetTargetNodeName(pod)
        if err != nil {
            klog.Warningf("Failed to get target node name of Pod %v/%v in DaemonSet %v/%v",
                pod.Namespace, pod.Name, ds.Namespace, ds.Name)
            continue
        }

        nodeToDaemonPods[nodeName] = append(nodeToDaemonPods[nodeName], pod)
    }

    return nodeToDaemonPods, nil
}
podsShouldBeOnNode

podsShouldBeOnNode 方法用来确定在给定的节点上需要创建还是删除 daemon pod,主要逻辑为:

  • 1、调用 dsc.nodeShouldRunDaemonPod 判断该 node 是否需要运行 daemon pod 以及 pod 能不能调度成功,该方法返回三个值 wantToRun, shouldSchedule, shouldContinueRunning
  • 2、通过判断 wantToRun, shouldSchedule, shouldContinueRunning 将需要创建 daemon pod 的 node 列表以及需要删除的 pod 列表获取到, wantToRun主要检查的是 selector、taints 等是否匹配,shouldSchedule 主要检查 node 上的资源是否充足,shouldContinueRunning 默认为 true;

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:866

func (dsc *DaemonSetsController) podsShouldBeOnNode(...) (nodesNeedingDaemonPods, podsToDelete []string, err error) {
    // 1、判断该 node 是否需要运行 daemon pod 以及能不能调度成功
    wantToRun, shouldSchedule, shouldContinueRunning, err := dsc.nodeShouldRunDaemonPod(node, ds)
    if err != nil {
        return
    }
    // 2、获取该节点上的指定ds的pod列表
    daemonPods, exists := nodeToDaemonPods[node.Name]
    dsKey, err := cache.MetaNamespaceKeyFunc(ds)
    if err != nil {
        utilruntime.HandleError(err)
        return
    }
        
    // 3、从 suspended list 中移除在该节点上 ds 的 pod
    dsc.removeSuspendedDaemonPods(node.Name, dsKey)

    switch {
    // 4、对于需要创建 pod 但是不能调度 pod 的 node,先把 pod 放入到 suspended 队列中
    case wantToRun && !shouldSchedule:
        dsc.addSuspendedDaemonPods(node.Name, dsKey)
    // 5、需要创建 pod 且 pod 未运行,则创建 pod
    case shouldSchedule && !exists:
        nodesNeedingDaemonPods = append(nodesNeedingDaemonPods, node.Name)
    // 6、需要 pod 一直运行
    case shouldContinueRunning:
        var daemonPodsRunning []*v1.Pod
        for _, pod := range daemonPods {
            if pod.DeletionTimestamp != nil {
                continue
            }
            // 7、如果 pod 运行状态为 failed,则删除该 pod
            if pod.Status.Phase == v1.PodFailed {
                backoffKey := failedPodsBackoffKey(ds, node.Name)

                now := dsc.failedPodsBackoff.Clock.Now()
                inBackoff := dsc.failedPodsBackoff.IsInBackOffSinceUpdate(backoffKey, now)
                if inBackoff {
                    delay := dsc.failedPodsBackoff.Get(backoffKey)
                    dsc.enqueueDaemonSetAfter(ds, delay)
                    continue
                }

                dsc.failedPodsBackoff.Next(backoffKey, now)
                podsToDelete = append(podsToDelete, pod.Name)
            } else {
                daemonPodsRunning = append(daemonPodsRunning, pod)
            }
        }
        // 8、如果节点上已经运行 daemon pod 数 > 1,保留运行时间最长的 pod,其余的删除
        if len(daemonPodsRunning) > 1 {
            sort.Sort(podByCreationTimestampAndPhase(daemonPodsRunning))
            for i := 1; i < len(daemonPodsRunning); i++ {
                podsToDelete = append(podsToDelete, daemonPodsRunning[i].Name)
            }
        }
    // 9、如果 pod 不需要继续运行但 pod 已存在则需要删除 pod
    case !shouldContinueRunning && exists:
        for _, pod := range daemonPods {
            if pod.DeletionTimestamp != nil {
                continue
            }
            podsToDelete = append(podsToDelete, pod.Name)
        }
    }

    return nodesNeedingDaemonPods, podsToDelete, nil
}

然后继续看 nodeShouldRunDaemonPod 方法的主要逻辑:

  • 1、调用 NewPod 为该 node 构建一个 daemon pod object;
  • 2、判断 ds 是否指定了 .Spec.Template.Spec.NodeName 字段;
  • 3、调用 dsc.simulate 执行 GeneralPredicates 预选算法检查该 node 是否能够调度成功;
  • 4、判断 GeneralPredicates 预选算法执行后的 reasons 确定 wantToRun, shouldSchedule, shouldContinueRunning 的值;

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:1337

func (dsc *DaemonSetsController) nodeShouldRunDaemonPod(node *v1.Node, ds *apps.DaemonSet) (wantToRun, shouldSchedule, shouldContinueRunning bool, err   error) {
    // 1、构建 daemon pod object
    newPod := NewPod(ds, node.Name)

    wantToRun, shouldSchedule, shouldContinueRunning = true, true, true
    // 2、判断 ds 是否指定了 node,若指定了且不为当前 node 直接返回 false
    if !(ds.Spec.Template.Spec.NodeName == "" || ds.Spec.Template.Spec.NodeName == node.Name) {
        return false, false, false, nil
    }

    // 3、执行 GeneralPredicates 预选算法
    reasons, nodeInfo, err := dsc.simulate(newPod, node, ds)
    if err != nil {
        ......
    }

    // 4、检查预选算法执行的结果
    var insufficientResourceErr error
    for _, r := range reasons {
        switch reason := r.(type) {
        case *predicates.InsufficientResourceError:
            insufficientResourceErr = reason
        case *predicates.PredicateFailureError:
            var emitEvent bool
            switch reason {
            case
                predicates.ErrNodeSelectorNotMatch,
                predicates.ErrPodNotMatchHostName,
                predicates.ErrNodeLabelPresenceViolated,

                predicates.ErrPodNotFitsHostPorts:
                return false, false, false, nil
            case predicates.ErrTaintsTolerationsNotMatch:
                fitsNoExecute, _, err := predicates.PodToleratesNodeNoExecuteTaints(newPod, nil, nodeInfo)
                if err != nil {
                    return false, false, false, err
                }
                if !fitsNoExecute {
                    return false, false, false, nil
                }
                wantToRun, shouldSchedule = false, false
            case
                predicates.ErrDiskConflict,
                predicates.ErrVolumeZoneConflict,
                predicates.ErrMaxVolumeCountExceeded,
                predicates.ErrNodeUnderMemoryPressure,
                predicates.ErrNodeUnderDiskPressure:
                shouldSchedule = false
                emitEvent = true
            case
                predicates.ErrPodAffinityNotMatch,
                predicates.ErrServiceAffinityViolated:
                
                return false, false, false, fmt.Errorf("unexpected reason: DaemonSet Predicates should not return reason %s", reason.GetReason())
            default:             
                wantToRun, shouldSchedule, shouldContinueRunning = false, false, false
                emitEvent = true
            }
            ......
        }
    }

    if shouldSchedule && insufficientResourceErr != nil {
        dsc.eventRecorder.Eventf(ds, v1.EventTypeWarning, FailedPlacementReason, "failed to place pod on %q: %s", node.ObjectMeta.Name,                  insufficientResourceErr.Error())
        shouldSchedule = false
    }
    return
}
syncNodes

syncNodes 方法主要是为需要 daemon pod 的 node 创建 pod 以及删除多余的 pod,其主要逻辑为:

  • 1、将 createDiffdeleteDiffburstReplicas 进行比较,burstReplicas 默认值为 250 即每个 syncLoop 中创建或者删除的 pod 数最多为 250 个,若超过其值则剩余需要创建或者删除的 pod 在下一个 syncLoop 继续操作;
  • 2、将 createDiffdeleteDiff 写入到 expectations 中;
  • 3、并发创建 pod,创建 pod 有两种方法:(1)创建的 pod 不经过默认调度器,直接指定了 pod 的运行节点(即设定pod.Spec.NodeName);(2)若启用了 ScheduleDaemonSetPods feature-gates 特性,则使用默认调度器进行创建 pod,通过 nodeAffinity来保证每个节点都运行一个 pod;
  • 4、并发删除 deleteDiff 中的所有 pod;

ScheduleDaemonSetPods 是一个 feature-gates 特性,其出现在 v1.11 中,在 v1.12 中处于 Beta 版本,v1.17 为 GA 版。最初 daemonset controller 只有一种创建 pod 的方法,即直接指定 pod 的 spec.NodeName 字段,但是目前这种方式已经暴露了许多问题,在以后的发展中社区还是希望能通过默认调度器进行调度,所以才出现了第二种方式,原因主要有以下五点:

  • 1、DaemonSet 无法感知 node 上资源的变化 (#46935, #58868):当 pod 第一次因资源不够无法创建时,若其他 pod 退出后资源足够时 DaemonSet 无法感知到;
  • 2、Daemonset 无法支持 Pod Affinity 和 Pod AntiAffinity 的功能(#29276);
  • 3、在某些功能上需要实现和 scheduler 重复的代码逻辑, 例如:critical pods (#42028), tolerant/taint;
  • 4、当 DaemonSet 的 Pod 创建失败时难以 debug,例如:资源不足时,对于 pending pod 最好能打一个 event 说明;
  • 5、多个组件同时调度时难以实现抢占机制:这也是无法通过横向扩展调度器提高调度吞吐量的一个原因;

更详细的原因可以参考社区的文档:schedule-DS-pod-by-scheduler.md

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:990

func (dsc *DaemonSetsController) syncNodes(ds *apps.DaemonSet, podsToDelete, nodesNeedingDaemonPods []string, hash string) error {
    ......

    // 1、设置 burstReplicas 
    createDiff := len(nodesNeedingDaemonPods)
    deleteDiff := len(podsToDelete)

    if createDiff > dsc.burstReplicas {
        createDiff = dsc.burstReplicas
    }
    if deleteDiff > dsc.burstReplicas {
        deleteDiff = dsc.burstReplicas
    }

    // 2、写入到 expectations 中
    dsc.expectations.SetExpectations(dsKey, createDiff, deleteDiff)

    errCh := make(chan error, createDiff+deleteDiff)
    createWait := sync.WaitGroup{}
    generation, err := util.GetTemplateGeneration(ds)
    if err != nil {
        generation = nil
    }
    template := util.CreatePodTemplate(ds.Spec.Template, generation, hash)

    // 3、并发创建 pod,创建的 pod 数依次为 1, 2, 4, 8, ...
    batchSize := integer.IntMin(createDiff, controller.SlowStartInitialBatchSize)
    for pos := 0; createDiff > pos; batchSize, pos = integer.IntMin(2*batchSize, createDiff-(pos+batchSize)), pos+batchSize {
        errorCount := len(errCh)
        createWait.Add(batchSize)
        for i := pos; i < pos+batchSize; i++ {
            go func(ix int) {
                defer createWait.Done()
                var err error

                podTemplate := template.DeepCopy()
                
                // 4、若启动了 ScheduleDaemonSetPods 功能,则通过 kube-scheduler 创建 pod
                if utilfeature.DefaultFeatureGate.Enabled(features.ScheduleDaemonSetPods) {
                    podTemplate.Spec.Affinity = util.ReplaceDaemonSetPodNodeNameNodeAffinity(
                        podTemplate.Spec.Affinity, nodesNeedingDaemonPods[ix])

                    err = dsc.podControl.CreatePodsWithControllerRef(ds.Namespace, podTemplate,
                        ds, metav1.NewControllerRef(ds, controllerKind))
                } else {
                    // 5、否则直接设置 pod 的 .spec.NodeName 创建 pod
                    err = dsc.podControl.CreatePodsOnNode(nodesNeedingDaemonPods[ix], ds.Namespace, podTemplate,
                        ds, metav1.NewControllerRef(ds, controllerKind))
                }

                // 6、创建 pod 时忽略 timeout err
                if err != nil && errors.IsTimeout(err) {
                    return
                }
                if err != nil {
                    dsc.expectations.CreationObserved(dsKey)
                    errCh <- err
                    utilruntime.HandleError(err)
                }
            }(i)
        }
        createWait.Wait()

        // 7、将创建失败的 pod 数记录到 expectations 中
        skippedPods := createDiff - (batchSize + pos)
        if errorCount < len(errCh) && skippedPods > 0 {
            dsc.expectations.LowerExpectations(dsKey, skippedPods, 0)
            break
        }
    }

    // 8、并发删除 deleteDiff 中的 pod
    deleteWait := sync.WaitGroup{}
    deleteWait.Add(deleteDiff)
    for i := 0; i < deleteDiff; i++ {
        go func(ix int) {
            defer deleteWait.Done()
            if err := dsc.podControl.DeletePod(ds.Namespace, podsToDelete[ix], ds); err != nil {
                dsc.expectations.DeletionObserved(dsKey)
                errCh <- err
                utilruntime.HandleError(err)
            }
        }(i)
    }
    deleteWait.Wait()
    errors := []error{}
    close(errCh)
    for err := range errCh {
        errors = append(errors, err)
    }
    return utilerrors.NewAggregate(errors)
}

RollingUpdate

daemonset update 的方式有两种 OnDeleteRollingUpdate,当为 OnDelete 时需要用户手动删除每一个 pod 后完成更新操作,当为 RollingUpdate 时,daemonset controller 会自动控制升级进度。

当为 RollingUpdate 时,主要逻辑为:

  • 1、获取 daemonset pod 与 node 的映射关系;
  • 2、根据 controllerrevision 的 hash 值获取所有未更新的 pods;
  • 3、获取 maxUnavailable, numUnavailable 的 pod 数值,maxUnavailable 是从 ds 的 rollingUpdate 字段中获取的默认值为 1,numUnavailable 的值是通过 daemonset pod 与 node 的映射关系计算每个 node 下是否有 available pod 得到的;
  • 4、通过 oldPods 获取 oldAvailablePods, oldUnavailablePods 的 pod 列表;
  • 5、遍历 oldUnavailablePods 列表将需要删除的 pod 追加到 oldPodsToDelete 数组中。oldUnavailablePods 列表中的 pod 分为两种,一种处于更新中,即删除状态,一种处于未更新且异常状态,处于异常状态的都需要被删除;
  • 6、遍历 oldAvailablePods 列表,此列表中的 pod 都处于正常运行状态,根据 maxUnavailable 值确定是否需要删除该 pod 并将需要删除的 pod 追加到 oldPodsToDelete 数组中;
  • 7、调用 dsc.syncNodes 删除 oldPodsToDelete 数组中的 pods,syncNodes 方法在 manage 阶段已经分析过,此处不再详述;

rollingUpdate 的结果是找出需要删除的 pods 并进行删除,被删除的 pod 在下一个 syncLoop 中会通过 manage 方法使用最新版本的 daemonset template 进行创建,整个滚动更新的过程是通过先删除再创建的方式一步步完成更新的,每次操作都是严格按照 maxUnavailable 的值确定需要删除的 pod 数。

k8s.io/kubernetes/pkg/controller/daemon/update.go:43

func (dsc *DaemonSetsController) rollingUpdate(......) error {
    // 1、获取 daemonset pod 与 node 的映射关系
    nodeToDaemonPods, err := dsc.getNodesToDaemonPods(ds)
    ......

    // 2、获取所有未更新的 pods
    _, oldPods := dsc.getAllDaemonSetPods(ds, nodeToDaemonPods, hash)
    
    // 3、计算 maxUnavailable, numUnavailable 的 pod 数值
    maxUnavailable, numUnavailable, err := dsc.getUnavailableNumbers(ds, nodeList, nodeToDaemonPods)
    if err != nil {
        return fmt.Errorf("couldn't get unavailable numbers: %v", err)
    }
    oldAvailablePods, oldUnavailablePods := util.SplitByAvailablePods(ds.Spec.MinReadySeconds, oldPods)

    // 4、将非 running 状态的 pods 加入到 oldPodsToDelete 中
    var oldPodsToDelete []string
    for _, pod := range oldUnavailablePods {
        if pod.DeletionTimestamp != nil {
            continue
        }
        oldPodsToDelete = append(oldPodsToDelete, pod.Name)
    }
    // 5、根据 maxUnavailable 值确定是否需要删除 pod
    for _, pod := range oldAvailablePods {
        if numUnavailable >= maxUnavailable {
            break
        }
        oldPodsToDelete = append(oldPodsToDelete, pod.Name)
        numUnavailable++
    }
    // 6、调用 syncNodes 方法删除 oldPodsToDelete 数组中的 pods
    return dsc.syncNodes(ds, oldPodsToDelete, []string{}, hash)
}

总结一下,manage 方法中的主要流程为:

            |->  dsc.getNodesToDaemonPods
            |
            |
manage ---- |->  dsc.podsShouldBeOnNode  ---> dsc.nodeShouldRunDaemonPod
            |
            |
            |->  dsc.syncNodes

updateDaemonSetStatus

updateDaemonSetStatussyncDaemonSet 中最后执行的方法,主要是用来计算 ds status subresource 中的值并更新其 status。status 如下所示:

status:
  currentNumberScheduled: 1  // 已经运行了 DaemonSet Pod的节点数量
  desiredNumberScheduled: 1  // 需要运行该DaemonSet Pod的节点数量
  numberMisscheduled: 0      // 不需要运行 DeamonSet Pod 但是已经运行了的节点数量
  numberReady: 0             // DaemonSet Pod状态为Ready的节点数量
  numberAvailable: 1                 // DaemonSet Pod状态为Ready且运行时间超过                                                                                                     // Spec.MinReadySeconds 的节点数量
  numberUnavailable: 0           // desiredNumberScheduled - numberAvailable 的节点数量
  observedGeneration: 3
  updatedNumberScheduled: 1  // 已经完成DaemonSet Pod更新的节点数量

updateDaemonSetStatus 主要逻辑为:

  • 1、调用 dsc.getNodesToDaemonPods 获取已存在 daemon pod 与 node 的映射关系;
  • 2、遍历所有 node,调用 dsc.nodeShouldRunDaemonPod 判断该 node 是否需要运行 daemon pod,然后计算 status 中的部分字段值;
  • 3、调用 storeDaemonSetStatus 更新 ds status subresource;
  • 4、判断 ds 是否需要 resync;

k8s.io/kubernetes/pkg/controller/daemon/daemon_controller.go:1152

func (dsc *DaemonSetsController) updateDaemonSetStatus(......) error {
    // 1、获取已存在 daemon pod 与 node 的映射关系
    nodeToDaemonPods, err := dsc.getNodesToDaemonPods(ds)
    ......

    var desiredNumberScheduled, currentNumberScheduled, numberMisscheduled, numberReady, updatedNumberScheduled, numberAvailable int
    for _, node := range nodeList {
        // 2、判断该 node 是否需要运行 daemon pod
        wantToRun, _, _, err := dsc.nodeShouldRunDaemonPod(node, ds)
        if err != nil {
            return err
        }

        scheduled := len(nodeToDaemonPods[node.Name]) > 0
        // 3、计算 status 中的字段值
        if wantToRun {
            desiredNumberScheduled++
            if scheduled {
                currentNumberScheduled++
                daemonPods, _ := nodeToDaemonPods[node.Name]
                sort.Sort(podByCreationTimestampAndPhase(daemonPods))
                pod := daemonPods[0]
                if podutil.IsPodReady(pod) {
                    numberReady++
                    if podutil.IsPodAvailable(pod, ds.Spec.MinReadySeconds, metav1.Now()) {
                        numberAvailable++
                    }
                }

                generation, err := util.GetTemplateGeneration(ds)
                if err != nil {
                    generation = nil
                }
                if util.IsPodUpdated(pod, hash, generation) {
                    updatedNumberScheduled++
                }
            }
        } else {
            if scheduled {
                numberMisscheduled++
            }
        }
    }
    numberUnavailable := desiredNumberScheduled - numberAvailable
    // 4、更新 daemonset status subresource
    err = storeDaemonSetStatus(dsc.kubeClient.AppsV1().DaemonSets(ds.Namespace), ds, desiredNumberScheduled, currentNumberScheduled, numberMisscheduled, numberReady, updatedNumberScheduled, numberAvailable, numberUnavailable, updateObservedGen)
    if err != nil {
        return fmt.Errorf("error storing status for daemon set %#v: %v", ds, err)
    }

    // 5、判断 ds 是否需要 resync
    if ds.Spec.MinReadySeconds > 0 && numberReady != numberAvailable {
        dsc.enqueueDaemonSetAfter(ds, time.Duration(ds.Spec.MinReadySeconds)*time.Second)
    }
    return nil
}

最后,再总结一下 syncDaemonSet 方法的主要流程:

                                |-> dsc.getNodesToDaemonPods
                                |
                                |
                  |-> manage -->|-> dsc.podsShouldBeOnNode  ---> dsc.nodeShouldRunDaemonPod
                  |             |
                  |             |
syncDaemonSet --> |             |-> dsc.syncNodes
                  |
                  |-> rollingUpdate
                  |
                  |
                  |-> updateDaemonSetStatus

总结

在 daemonset controller 中可以看到许多功能都是 deployment 和 statefulset 已有的。在创建 pod 的流程与 replicaset controller 创建 pod 的流程是相似的,都使用了 expectations 机制并且限制了在一个 syncLoop 中最多创建或删除的 pod 数。更新方式与 statefulset 一样都有 OnDeleteRollingUpdate 两种, OnDelete 方式与 statefulset 相似,都需要手动删除对应的 pod,而 RollingUpdate 方式与 statefulset 和 deployment 都有点区别, RollingUpdate方式更新时不支持暂停操作并且 pod 是先删除再创建的顺序进行。版本控制方式与 statefulset 的一样都是使用 controllerRevision。最后要说的一点是在 v1.12 及以后的版本中,使用 daemonset 创建的 pod 已不再使用直接指定 .spec.nodeName的方式绕过调度器进行调度,而是走默认调度器通过 nodeAffinity 的方式调度到每一个节点上。

参考:

https://yq.aliyun.com/articles/702305

最后编辑于
©著作权归作者所有,转载或内容合作请联系作者
禁止转载,如需转载请通过简信或评论联系作者。
  • 序言:七十年代末,一起剥皮案震惊了整个滨河市,随后出现的几起案子,更是在滨河造成了极大的恐慌,老刑警刘岩,带你破解...
    沈念sama阅读 203,271评论 5 476
  • 序言:滨河连续发生了三起死亡事件,死亡现场离奇诡异,居然都是意外死亡,警方通过查阅死者的电脑和手机,发现死者居然都...
    沈念sama阅读 85,275评论 2 380
  • 文/潘晓璐 我一进店门,熙熙楼的掌柜王于贵愁眉苦脸地迎上来,“玉大人,你说我怎么就摊上这事。” “怎么了?”我有些...
    开封第一讲书人阅读 150,151评论 0 336
  • 文/不坏的土叔 我叫张陵,是天一观的道长。 经常有香客问我,道长,这世上最难降的妖魔是什么? 我笑而不...
    开封第一讲书人阅读 54,550评论 1 273
  • 正文 为了忘掉前任,我火速办了婚礼,结果婚礼上,老公的妹妹穿的比我还像新娘。我一直安慰自己,他们只是感情好,可当我...
    茶点故事阅读 63,553评论 5 365
  • 文/花漫 我一把揭开白布。 她就那样静静地躺着,像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上,一...
    开封第一讲书人阅读 48,559评论 1 281
  • 那天,我揣着相机与录音,去河边找鬼。 笑死,一个胖子当着我的面吹牛,可吹牛的内容都是我干的。 我是一名探鬼主播,决...
    沈念sama阅读 37,924评论 3 395
  • 文/苍兰香墨 我猛地睁开眼,长吁一口气:“原来是场噩梦啊……” “哼!你这毒妇竟也来了?” 一声冷哼从身侧响起,我...
    开封第一讲书人阅读 36,580评论 0 257
  • 序言:老挝万荣一对情侣失踪,失踪者是张志新(化名)和其女友刘颖,没想到半个月后,有当地人在树林里发现了一具尸体,经...
    沈念sama阅读 40,826评论 1 297
  • 正文 独居荒郊野岭守林人离奇死亡,尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,578评论 2 320
  • 正文 我和宋清朗相恋三年,在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,661评论 1 329
  • 序言:一个原本活蹦乱跳的男人离奇死亡,死状恐怖,灵堂内的尸体忽然破棺而出,到底是诈尸还是另有隐情,我是刑警宁泽,带...
    沈念sama阅读 33,363评论 4 318
  • 正文 年R本政府宣布,位于F岛的核电站,受9级特大地震影响,放射性物质发生泄漏。R本人自食恶果不足惜,却给世界环境...
    茶点故事阅读 38,940评论 3 307
  • 文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹,春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,926评论 0 19
  • 文/苍兰香墨 我抬头看了看天上的太阳。三九已至,却和暖如春,着一层夹袄步出监牢的瞬间,已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,156评论 1 259
  • 我被黑心中介骗来泰国打工, 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留,地道东北人。 一个月前我还...
    沈念sama阅读 42,872评论 2 349
  • 正文 我出身青楼,却偏偏与公主长得像,于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子,可洞房花烛夜当晚...
    茶点故事阅读 42,391评论 2 342

推荐阅读更多精彩内容