创建

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

扩缩容

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

更新

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

  1. // 更新镜像
  2. $ kubectl set image ds/nginx-ds nginx-ds=nginx:1.16
  4. // 查看更新状态
  5. $ kubectl rollout status ds/nginx-ds

回滚

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

  1. // 查看 ds 历史版本信息
  2. $ kubectl get controllerrevision
  3. NAME                  CONTROLLER                REVISION   AGE
  4. nginx-ds-5c4b75bdbb   daemonset.apps/nginx-ds   2          122m
  5. nginx-ds-7cd7798dcd   daemonset.apps/nginx-ds   1          133m
  7. // 回滚到版本 1
  8. $ kubectl rollout undo daemonset nginx-ds --to-revision=1
  10. // 查看回滚状态
  11. $ kubectl rollout status ds/nginx-ds

暂停

daemonset 目前不支持暂停操作。

删除

daemonset 也支持两种删除操作。

  1. // 非级联删除
  2. $ kubectl delete ds/nginx-ds --cascade=false
  4. // 级联删除
  5. $ kubectl delete ds/nginx-ds

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

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

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

在 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

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 状态； 
  1. func (dsc *DaemonSetsController) syncDaemonSet(key string) error {
  2.     ......
  4.     // 1、通过 key 获取 ns 和 name
  5.     namespace, name, err := cache.SplitMetaNamespaceKey(key)
  6.     if err != nil {
  7.         return err
  8.     }
  10.     // 2、从 dsLister 中获取 ds 对象
  11.     ds, err := dsc.dsLister.DaemonSets(namespace).Get(name)
  12.     if errors.IsNotFound(err) {
  13.         dsc.expectations.DeleteExpectations(key)
  14.         return nil
  15.     }
  16.     ......
  18.     // 3、从 nodeLister 获取所有 node
  19.     nodeList, err := dsc.nodeLister.List(labels.Everything())
  20.     ......
  22.     everything := metav1.LabelSelector{}
  23.     if reflect.DeepEqual(ds.Spec.Selector, &everything) {
  24.         dsc.eventRecorder.Eventf(ds, v1.EventTypeWarning, SelectingAllReason, "This daemon set is selecting all pods. A non-empty selector is required. ")
  25.         return nil
  26.     }
  28.     // 4、获取 dsKey
  29.     dsKey, err := controller.KeyFunc(ds)
  30.     if err != nil {
  31.         return fmt.Errorf("couldn't get key for object %#v: %v", ds, err)
  32.     }
  34.     // 5、判断 ds 是否处于删除状态
  35.     if ds.DeletionTimestamp != nil {
  36.         return nil
  37.     }
  39.     // 6、获取 current 和 old controllerRevision
  40.     cur, old, err := dsc.constructHistory(ds)
  41.     if err != nil {
  42.         return fmt.Errorf("failed to construct revisions of DaemonSet: %v", err)
  43.     }
  44.     hash := cur.Labels[apps.DefaultDaemonSetUniqueLabelKey]
  46.     // 7、判断是否满足 expectations 机制
  47.     if !dsc.expectations.SatisfiedExpectations(dsKey) {
  48.         return dsc.updateDaemonSetStatus(ds, nodeList, hash, false)
  49.     }
  51.     // 8、执行实际的 sync 操作
  52.     err = dsc.manage(ds, nodeList, hash)
  53.     if err != nil {
  54.         return err
  55.     }
  57.     // 9、判断是否为更新操作，并执行对应的更新操作
  58.     if dsc.expectations.SatisfiedExpectations(dsKey) {
  59.         switch ds.Spec.UpdateStrategy.Type {
  60.         case apps.OnDeleteDaemonSetStrategyType:
  61.         case apps.RollingUpdateDaemonSetStrategyType:
  62.             err = dsc.rollingUpdate(ds, nodeList, hash)
  63.         }
  64.         if err != nil {
  65.             return err
  66.         }
  67.     }
  68.     // 10、清理过期的 controllerrevision
  69.     err = dsc.cleanupHistory(ds, old)
  70.     if err != nil {
  71.         return fmt.Errorf("failed to clean up revisions of DaemonSet: %v", err)
  72.     }
  74.     // 11、更新 ds 状态
  75.     return dsc.updateDaemonSetStatus(ds, nodeList, hash, true)
  76. }

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

  1.     // 1、获取已存在 daemon pod 与 node 的映射关系
  2.     nodeToDaemonPods, err := dsc.getNodesToDaemonPods(ds)
  3.     ......
  5.     // 2、判断每一个 node 是否需要运行 daemon pod
  6.     var nodesNeedingDaemonPods, podsToDelete []string
  7.     for _, node := range nodeList {
  8.         nodesNeedingDaemonPodsOnNode, podsToDeleteOnNode, err := dsc.podsShouldBeOnNode(
  9.             node, nodeToDaemonPods, ds)
  12.             continue
  13.         }
  15.         nodesNeedingDaemonPods = append(nodesNeedingDaemonPods, nodesNeedingDaemonPodsOnNode...)
  16.         podsToDelete = append(podsToDelete, podsToDeleteOnNode...)
  17.     }
  19.     // 3、判断是否启动了 ScheduleDaemonSetPods feature-gates 特性，若启用了则对不存在 node 上的 
  20.     // daemon pod 进行删除 
  21.     if utilfeature.DefaultFeatureGate.Enabled(features.ScheduleDaemonSetPods) {
  22.         podsToDelete = append(podsToDelete, getUnscheduledPodsWithoutNode(nodeList, nodeToDaemonPods)...)
  23.     }
  25.     // 4、为对应的 node 创建 daemon pod 以及删除多余的 pods
  26.     if err = dsc.syncNodes(ds, podsToDelete, nodesNeedingDaemonPods, hash); err != nil {
  27.         return err
  28.     }
  30.     return nil
  31. }

在 manage 方法中又调用了 getNodesToDaemonPods、podsShouldBeOnNode 和 syncNodes 三个方法，继续来看这几种方法的作用。

getNodesToDaemonPods

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

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

  1. func (dsc *DaemonSetsController) getNodesToDaemonPods(ds *apps.DaemonSet) (map[string][]*v1.Pod, error) {
  2.     claimedPods, err := dsc.getDaemonPods(ds)
  3.     if err != nil {
  4.         return nil, err
  5.     }
  6.     nodeToDaemonPods := make(map[string][]*v1.Pod)
  7.     for _, pod := range claimedPods {
  8.         nodeName, err := util.GetTargetNodeName(pod)
  9.         if err != nil {
  10.             klog.Warningf("Failed to get target node name of Pod %v/%v in DaemonSet %v/%v",
  11.                 pod.Namespace, pod.Name, ds.Namespace, ds.Name)
  12.             continue
  13.         }
  15.         nodeToDaemonPods[nodeName] = append(nodeToDaemonPods[nodeName], pod)
  16.     }
  18.     return nodeToDaemonPods, nil
  19. }
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

  1. func (dsc *DaemonSetsController) podsShouldBeOnNode(...) (nodesNeedingDaemonPods, podsToDelete []string, err error) {
  2.     // 1、判断该 node 是否需要运行 daemon pod 以及能不能调度成功
  3.     wantToRun, shouldSchedule, shouldContinueRunning, err := dsc.nodeShouldRunDaemonPod(node, ds)
  4.     if err != nil {
  5.         return
  6.     }
  7.     // 2、获取该节点上的指定ds的pod列表
  8.     daemonPods, exists := nodeToDaemonPods[node.Name]
  9.     dsKey, err := cache.MetaNamespaceKeyFunc(ds)
  10.     if err != nil {
  11.         utilruntime.HandleError(err)
  12.         return
  13.     }
  15.     // 3、从 suspended list 中移除在该节点上 ds 的 pod
  16.     dsc.removeSuspendedDaemonPods(node.Name, dsKey)
  18.     switch {
  19.     // 4、对于需要创建 pod 但是不能调度 pod 的 node，先把 pod 放入到 suspended 队列中
  20.     case wantToRun && !shouldSchedule:
  21.         dsc.addSuspendedDaemonPods(node.Name, dsKey)
  22.     // 5、需要创建 pod 且 pod 未运行，则创建 pod
  23.     case shouldSchedule && !exists:
  24.         nodesNeedingDaemonPods = append(nodesNeedingDaemonPods, node.Name)
  25.     // 6、需要 pod 一直运行
  26.     case shouldContinueRunning:
  27.         var daemonPodsRunning []*v1.Pod
  28.         for _, pod := range daemonPods {
  29.             if pod.DeletionTimestamp != nil {
  30.                 continue
  31.             }
  32.             // 7、如果 pod 运行状态为 failed，则删除该 pod
  33.             if pod.Status.Phase == v1.PodFailed {
  34.                 backoffKey := failedPodsBackoffKey(ds, node.Name)
  36.                 now := dsc.failedPodsBackoff.Clock.Now()
  37.                 inBackoff := dsc.failedPodsBackoff.IsInBackOffSinceUpdate(backoffKey, now)
  38.                 if inBackoff {
  39.                     delay := dsc.failedPodsBackoff.Get(backoffKey)
  40.                     dsc.enqueueDaemonSetAfter(ds, delay)
  41.                     continue
  42.                 }
  44.                 dsc.failedPodsBackoff.Next(backoffKey, now)
  45.                 podsToDelete = append(podsToDelete, pod.Name)
  46.             } else {
  47.                 daemonPodsRunning = append(daemonPodsRunning, pod)
  48.             }
  49.         }
  50.         // 8、如果节点上已经运行 daemon pod 数 > 1，保留运行时间最长的 pod，其余的删除
  51.         if len(daemonPodsRunning) > 1 {
  52.             sort.Sort(podByCreationTimestampAndPhase(daemonPodsRunning))
  53.             for i := 1; i < len(daemonPodsRunning); i++ {
  54.                 podsToDelete = append(podsToDelete, daemonPodsRunning[i].Name)
  55.             }
  56.         }
  57.     // 9、如果 pod 不需要继续运行但 pod 已存在则需要删除 pod
  58.     case !shouldContinueRunning && exists:
  59.         for _, pod := range daemonPods {
  60.             if pod.DeletionTimestamp != nil {
  61.                 continue
  62.             }
  63.             podsToDelete = append(podsToDelete, pod.Name)
  64.         }
  65.     }
  67.     return nodesNeedingDaemonPods, podsToDelete, nil
  68. }

然后继续看 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

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

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

  1. func (dsc *DaemonSetsController) syncNodes(ds *apps.DaemonSet, podsToDelete, nodesNeedingDaemonPods []string, hash string) error {
  2.     ......
  4.     // 1、设置 burstReplicas 
  5.     createDiff := len(nodesNeedingDaemonPods)
  6.     deleteDiff := len(podsToDelete)
  8.     if createDiff > dsc.burstReplicas {
  9.         createDiff = dsc.burstReplicas
  10.     }
  11.     if deleteDiff > dsc.burstReplicas {
  12.         deleteDiff = dsc.burstReplicas
  13.     }
  15.     // 2、写入到 expectations 中
  17.     errCh := make(chan error, createDiff+deleteDiff)
  18.     createWait := sync.WaitGroup{}
  19.     generation, err := util.GetTemplateGeneration(ds)
  20.     if err != nil {
  21.         generation = nil
  22.     }
  25.     // 3、并发创建 pod，创建的 pod 数依次为 1, 2, 4, 8, ...
  26.     batchSize := integer.IntMin(createDiff, controller.SlowStartInitialBatchSize)
  27.     for pos := 0; createDiff > pos; batchSize, pos = integer.IntMin(2*batchSize, createDiff-(pos+batchSize)), pos+batchSize {
  28.         errorCount := len(errCh)
  29.         createWait.Add(batchSize)
  30.         for i := pos; i < pos+batchSize; i++ {
  31.             go func(ix int) {
  32.                 defer createWait.Done()
  33.                 var err error
  35.                 podTemplate := template.DeepCopy()
  37.                 // 4、若启动了 ScheduleDaemonSetPods 功能，则通过 kube-scheduler 创建 pod
  38.                 if utilfeature.DefaultFeatureGate.Enabled(features.ScheduleDaemonSetPods) {
  39.                     podTemplate.Spec.Affinity = util.ReplaceDaemonSetPodNodeNameNodeAffinity(
  40.                         podTemplate.Spec.Affinity, nodesNeedingDaemonPods[ix])
  42.                     err = dsc.podControl.CreatePodsWithControllerRef(ds.Namespace, podTemplate,
  43.                         ds, metav1.NewControllerRef(ds, controllerKind))
  44.                 } else {
  45.                     // 5、否则直接设置 pod 的 .spec.NodeName 创建 pod
  46.                     err = dsc.podControl.CreatePodsOnNode(nodesNeedingDaemonPods[ix], ds.Namespace, podTemplate,
  47.                         ds, metav1.NewControllerRef(ds, controllerKind))
  48.                 }
  50.                 // 6、创建 pod 时忽略 timeout err
  51.                 if err != nil && errors.IsTimeout(err) {
  52.                     return
  53.                 }
  54.                 if err != nil {
  55.                     dsc.expectations.CreationObserved(dsKey)
  56.                     errCh <- err
  57.                     utilruntime.HandleError(err)
  58.                 }
  59.             }(i)
  60.         }
  61.         createWait.Wait()
  63.         // 7、将创建失败的 pod 数记录到 expectations 中
  64.         skippedPods := createDiff - (batchSize + pos)
  65.         if errorCount < len(errCh) && skippedPods > 0 {
  66.             dsc.expectations.LowerExpectations(dsKey, skippedPods, 0)
  67.             break
  68.         }
  69.     }
  71.     // 8、并发删除 deleteDiff 中的 pod
  72.     deleteWait := sync.WaitGroup{}
  73.     deleteWait.Add(deleteDiff)
  74.     for i := 0; i < deleteDiff; i++ {
  75.         go func(ix int) {
  76.             defer deleteWait.Done()
  77.             if err := dsc.podControl.DeletePod(ds.Namespace, podsToDelete[ix], ds); err != nil {
  78.                 dsc.expectations.DeletionObserved(dsKey)
  79.                 errCh <- err
  80.                 utilruntime.HandleError(err)
  81.             }
  82.         }(i)
  83.     }
  84.     deleteWait.Wait()
  85.     errors := []error{}
  86.     close(errCh)
  87.     for err := range errCh {
  88.         errors = append(errors, err)
  89.     }
  90.     return utilerrors.NewAggregate(errors)
  91. }

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

  1. func (dsc *DaemonSetsController) rollingUpdate(......) error {
  2.     // 1、获取 daemonset pod 与 node 的映射关系
  3.     nodeToDaemonPods, err := dsc.getNodesToDaemonPods(ds)
  4.     ......
  6.     // 2、获取所有未更新的 pods
  7.     _, oldPods := dsc.getAllDaemonSetPods(ds, nodeToDaemonPods, hash)
  9.     // 3、计算 maxUnavailable, numUnavailable 的 pod 数值
  10.     maxUnavailable, numUnavailable, err := dsc.getUnavailableNumbers(ds, nodeList, nodeToDaemonPods)
  11.     if err != nil {
  12.         return fmt.Errorf("couldn't get unavailable numbers: %v", err)
  13.     }
  14.     oldAvailablePods, oldUnavailablePods := util.SplitByAvailablePods(ds.Spec.MinReadySeconds, oldPods)
  16.     // 4、将非 running 状态的 pods 加入到 oldPodsToDelete 中
  17.     var oldPodsToDelete []string
  18.     for _, pod := range oldUnavailablePods {
  19.         if pod.DeletionTimestamp != nil {
  20.             continue
  21.         }
  22.         oldPodsToDelete = append(oldPodsToDelete, pod.Name)
  23.     }
  24.     // 5、根据 maxUnavailable 值确定是否需要删除 pod
  25.     for _, pod := range oldAvailablePods {
  26.         if numUnavailable >= maxUnavailable {
  27.             break
  28.         }
  29.         oldPodsToDelete = append(oldPodsToDelete, pod.Name)
  30.         numUnavailable++
  31.     }
  32.     // 6、调用 syncNodes 方法删除 oldPodsToDelete 数组中的 pods
  33.     return dsc.syncNodes(ds, oldPodsToDelete, []string{}, hash)
  34. }

总结一下，manage 方法中的主要流程为：

  1.             |->  dsc.getNodesToDaemonPods
  2.             |
  3.             |
  4. manage ---- |->  dsc.podsShouldBeOnNode  ---> dsc.nodeShouldRunDaemonPod
  5.             |
  6.             |
  7.             |->  dsc.syncNodes

updateDaemonSetStatus

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

  1. status:
  2.   currentNumberScheduled: 1  // 已经运行了 DaemonSet Pod的节点数量
  3.   desiredNumberScheduled: 1  // 需要运行该DaemonSet Pod的节点数量
  4.   numberMisscheduled: 0      // 不需要运行 DeamonSet Pod 但是已经运行了的节点数量
  5.   numberReady: 0             // DaemonSet Pod状态为Ready的节点数量
  6.   numberAvailable: 1          // DaemonSet Pod状态为Ready且运行时间超过                                                                                                      // Spec.MinReadySeconds 的节点数量
  7.   numberUnavailable: 0             // desiredNumberScheduled - numberAvailable 的节点数量
  8.   observedGeneration: 3
  9.   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

最后，再总结一下 syncDaemonSet 方法的主要流程：

  1.                                 |-> dsc.getNodesToDaemonPods
  2.                                 |
  3.                                 |
  4.                   |-> manage -->|-> dsc.podsShouldBeOnNode  ---> dsc.nodeShouldRunDaemonPod
  5.                   |             |
  6.                   |             |
  7. syncDaemonSet --> |             |-> dsc.syncNodes
  8.                   |
  9.                   |-> rollingUpdate
  10.                   |
  11.                   |

在 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 的方式调度到每一个节点上。

参考：