daemonset controller 源码分析 - 《kubernetes 源码分析》

DaemonSet 的基本操作
DaemonSetController 源码分析
总结

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

DaemonSet 的基本操作

创建

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

$ kubectl create -f nginx-ds.yaml

扩缩容

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

更新

daemonset 有两种更新策略 OnDelete 和 RollingUpdate，默认为 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 会监听 daemonsets、controllerRevision、pod 和 node 四种对象资源的变动。其中 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 中主要有 manage、rollingUpdate和updateDaemonSetStatus 三个方法，分别对应创建、更新与状态同步，下面主要来分析这三个方法。

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 方法中又调用了 getNodesToDaemonPods、podsShouldBeOnNode 和 syncNodes 三个方法，继续来看这几种方法的作用。

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、将 createDiff 和 deleteDiff 与 burstReplicas 进行比较，burstReplicas 默认值为 250 即每个 syncLoop 中创建或者删除的 pod 数最多为 250 个，若超过其值则剩余需要创建或者删除的 pod 在下一个 syncLoop 继续操作；
2、将 createDiff 和 deleteDiff 写入到 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 的方式有两种 OnDelete 和 RollingUpdate，当为 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

updateDaemonSetStatus 是 syncDaemonSet 中最后执行的方法，主要是用来计算 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 一样都有 OnDelete 和 RollingUpdate 两种， OnDelete 方式与 statefulset 相似，都需要手动删除对应的 pod，而 RollingUpdate 方式与 statefulset 和 deployment 都有点区别， RollingUpdate方式更新时不支持暂停操作并且 pod 是先删除再创建的顺序进行。版本控制方式与 statefulset 的一样都是使用 controllerRevision。最后要说的一点是在 v1.12 及以后的版本中，使用 daemonset 创建的 pod 已不再使用直接指定 .spec.nodeName的方式绕过调度器进行调度，而是走默认调度器通过 nodeAffinity 的方式调度到每一个节点上。

参考：

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