胡涛的个人网站
目录 标签 关于我 归档
胡涛的个人网站

目录

Kubernetes Deployment 源码分析（二）

胡涛 | Daniel Hu 收录于 源码分析 ｜ Source Code Analysis
   约 2775 字   预计阅读 6 分钟 

概述

源码版本：kubernetes-v1.22.3 / commit-id: c92036

Deployment 是最常用的 Kubernetes 原生 Workload 资源之一，我们一开始尝试使用 Kubernetes 的时候大概率就是从运行一个 Deployment 类型的工作负载开始的。

在上一篇《Kubernetes Deployment 源码分析（一）》中我们过了下 Deployment 的全部特性，主要介绍“滚动更新”和“回滚”等主要功能，从而心中有个概念，知道 Deployment 的能力边界在那里，所以今天以此为基础，我们继续从源码角度看下 Deployment 的实现。

《Kubernetes Deployment Controller 源码分析》分为两讲：

startDeploymentController 入口逻辑

DeploymentController 的初始化和启动入口是 startDeploymentController() 函数

  • cmd/kube-controller-manager/app/apps.go:72
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

func startDeploymentController(ctx ControllerContext) (http.Handler, bool, error) {
   dc, err := deployment.NewDeploymentController(
      ctx.InformerFactory.Apps().V1().Deployments(),
      ctx.InformerFactory.Apps().V1().ReplicaSets(),
      ctx.InformerFactory.Core().V1().Pods(),
      ctx.ClientBuilder.ClientOrDie("deployment-controller"),
   )
   if err != nil {
      return nil, true, fmt.Errorf("error creating Deployment controller: %v", err)
   }
   go dc.Run(int(ctx.ComponentConfig.DeploymentController.ConcurrentDeploymentSyncs), ctx.Stop)
   return nil, true, nil
}

在 startDeploymentController() 函数中先通过 NewDeploymentController() 方法初始化一个 DeploymentController 实例，这里的参数 DeploymentInformer、ReplicaSetInformer、PodInformer 和 Clientset，因而 DeploymentController 也就具备了获取Deployment、 ReplicaSet、Pod 三类资源变更事件以及 CURD apiserver 操作各种资源的能力。接着这个函数中又调用了 DeploymentController 的 Run() 方法来启动 DeploymentController，这里的参数 ConcurrentDeploymentSyncs 默认值是 5，也就是默认情况下并发调谐的 Deployment 数量是 5 个。

DeploymentController 对象

继续看下 DeploymentController 对象的定义和初始化。

DeploymentController 类型定义

接着来看 DeploymentController 类型的定义

  • pkg/controller/deployment/deployment_controller.go:68
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

type DeploymentController struct {
   // ReplicaSet 操控器
   rsControl     controller.RSControlInterface
   client        clientset.Interface
   eventRecorder record.EventRecorder

   syncHandler func(dKey string) error
   // 测试用
   enqueueDeployment func(deployment *apps.Deployment)

   // 用来从 cache 里 get/list Deployment
   dLister appslisters.DeploymentLister
   // 用来从 cache 里 get/list ReplicaSet
   rsLister appslisters.ReplicaSetLister
   // 用来从 cache 里 get/list Pod
   podLister corelisters.PodLister

   dListerSynced cache.InformerSynced
   rsListerSynced cache.InformerSynced
   podListerSynced cache.InformerSynced

   // 工作队列，限速队列实现
   queue workqueue.RateLimitingInterface
}

DeploymentController 初始化

  • pkg/controller/deployment/deployment_controller.go:101
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

func NewDeploymentController(dInformer appsinformers.DeploymentInformer, rsInformer appsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, client clientset.Interface) (*DeploymentController, error) {
   // Event 相关逻辑
   eventBroadcaster := record.NewBroadcaster()
   eventBroadcaster.StartStructuredLogging(0)
   eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})

   // ……
  
   // new dc
   dc := &DeploymentController{
      client:        client,
      eventRecorder: eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "deployment-controller"}),
      queue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "deployment"),
   }
   // 主要是 clientset
   dc.rsControl = controller.RealRSControl{
      KubeClient: client,
      Recorder:   dc.eventRecorder,
   }
   // ResourceEventHandler 配置，后面会分析
   dInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
      AddFunc:    dc.addDeployment,
      UpdateFunc: dc.updateDeployment,
      DeleteFunc: dc.deleteDeployment,
   })
   rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
      AddFunc:    dc.addReplicaSet,
      UpdateFunc: dc.updateReplicaSet,
      DeleteFunc: dc.deleteReplicaSet,
   })
   podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
      DeleteFunc: dc.deletePod,
   })
   
   // 这里有主要逻辑，后面会讲
   dc.syncHandler = dc.syncDeployment
   dc.enqueueDeployment = dc.enqueue
   // 各种 lister
   dc.dLister = dInformer.Lister()
   dc.rsLister = rsInformer.Lister()
   dc.podLister = podInformer.Lister()
   dc.dListerSynced = dInformer.Informer().HasSynced
   dc.rsListerSynced = rsInformer.Informer().HasSynced
   dc.podListerSynced = podInformer.Informer().HasSynced
   return dc, nil
}

ResourceEventHandler

上面提到了几个 ResourceEventHandler 回调函数：

  • addDeployment
  • updateDeployment
  • deleteDeployment
  • addReplicaSet
  • updateReplicaSet
  • deleteReplicaSet
  • deletePod

逐个分析下。

Deployment 变更事件

这里逻辑比较简单，三个方法一起看：

  • pkg/controller/deployment/deployment_controller.go:167
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

func (dc *DeploymentController) addDeployment(obj interface{}) {
   d := obj.(*apps.Deployment)
   klog.V(4).InfoS("Adding deployment", "deployment", klog.KObj(d))
   // 新增 Deployment 时直接 enqueue
   dc.enqueueDeployment(d)
}

func (dc *DeploymentController) updateDeployment(old, cur interface{}) {
   oldD := old.(*apps.Deployment)
   curD := cur.(*apps.Deployment)
   klog.V(4).InfoS("Updating deployment", "deployment", klog.KObj(oldD))
   // old Deployment 只用来打印一个日志，cur Deployment enqueue
   dc.enqueueDeployment(curD)
}

func (dc *DeploymentController) deleteDeployment(obj interface{}) {
   d, ok := obj.(*apps.Deployment)
   if !ok {
      // 处理 DeletedFinalStateUnknown 场景
      tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
      if !ok {
         utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
         return
      }
      d, ok = tombstone.Obj.(*apps.Deployment)
      if !ok {
         utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a Deployment %#v", obj))
         return
      }
   }
   klog.V(4).InfoS("Deleting deployment", "deployment", klog.KObj(d))
   // 入队
   dc.enqueueDeployment(d)
}

ReplicaSet 变更事件

然后来看 ReplicaSet 相关回调函数的实现。

1、Added

  • pkg/controller/deployment/deployment_controller.go:199
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

func (dc *DeploymentController) addReplicaSet(obj interface{}) {
   rs := obj.(*apps.ReplicaSet)
   // 如果是准备删除了，重启的过程会收到 Added 事件，这时候直接调用删除操作
   if rs.DeletionTimestamp != nil {
      dc.deleteReplicaSet(rs)
      return
   }
   // 查询对应的 Deployment
   if controllerRef := metav1.GetControllerOf(rs); controllerRef != nil {
      d := dc.resolveControllerRef(rs.Namespace, controllerRef)
      if d == nil {
         return
      }
      klog.V(4).InfoS("ReplicaSet added", "replicaSet", klog.KObj(rs))
      // 将这个 Deployment 加入工作队列
      dc.enqueueDeployment(d)
      return
   }

   // 如果是一个孤儿 ReplicaSet，则看是不是能找到一个 Deployment 来领养
   ds := dc.getDeploymentsForReplicaSet(rs)
   if len(ds) == 0 {
      return
   }
   klog.V(4).InfoS("Orphan ReplicaSet added", "replicaSet", klog.KObj(rs))
   // 一般只有一个 Deployment，但是也不能排出多个的情况，所以这里用的是 ds 列表，循环 enqueue
   for _, d := range ds {
      dc.enqueueDeployment(d)
   }
}

2、Updated

  • pkg/controller/deployment/deployment_controller.go:256
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

func (dc *DeploymentController) updateReplicaSet(old, cur interface{}) {
   curRS := cur.(*apps.ReplicaSet)
   oldRS := old.(*apps.ReplicaSet)
   if curRS.ResourceVersion == oldRS.ResourceVersion {
      // Resync 的时候 RV 相同，不做处理
      return
   }

   curControllerRef := metav1.GetControllerOf(curRS)
   oldControllerRef := metav1.GetControllerOf(oldRS)
   controllerRefChanged := !reflect.DeepEqual(curControllerRef, oldControllerRef)
   if controllerRefChanged && oldControllerRef != nil {
      // 如果 rs 的 ref 变更了，就需要通知老的 ref 对应的 Deployment
      if d := dc.resolveControllerRef(oldRS.Namespace, oldControllerRef); d != nil {
         dc.enqueueDeployment(d)
      }
   }

   if curControllerRef != nil {
      d := dc.resolveControllerRef(curRS.Namespace, curControllerRef)
      if d == nil {
         return
      }
      klog.V(4).InfoS("ReplicaSet updated", "replicaSet", klog.KObj(curRS))
      // 当前 rs 对应 dp 入队
      dc.enqueueDeployment(d)
      return
   }

   // 孤儿 rs 的场景，和 Added 时处理逻辑一样
   labelChanged := !reflect.DeepEqual(curRS.Labels, oldRS.Labels)
   if labelChanged || controllerRefChanged {
      ds := dc.getDeploymentsForReplicaSet(curRS)
      if len(ds) == 0 {
         return
      }
      klog.V(4).InfoS("Orphan ReplicaSet updated", "replicaSet", klog.KObj(curRS))
      for _, d := range ds {
         dc.enqueueDeployment(d)
      }
   }
}

3、Deleted

  • pkg/controller/deployment/deployment_controller.go:304
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

func (dc *DeploymentController) deleteReplicaSet(obj interface{}) {
   rs, ok := obj.(*apps.ReplicaSet)

   // 删除场景需要处理的 DeletedFinalStateUnknown 场景
   if !ok {
      tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
      if !ok {
         utilruntime.HandleError(fmt.Errorf("couldn't get object from tombstone %#v", obj))
         return
      }
      rs, ok = tombstone.Obj.(*apps.ReplicaSet)
      if !ok {
         utilruntime.HandleError(fmt.Errorf("tombstone contained object that is not a ReplicaSet %#v", obj))
         return
      }
   }

   // 孤儿 rs 被删除时没有 Deployment 需要关心
   controllerRef := metav1.GetControllerOf(rs)
   if controllerRef == nil {
      return
   }
   d := dc.resolveControllerRef(rs.Namespace, controllerRef)
   if d == nil {
      return
   }
   klog.V(4).InfoS("ReplicaSet deleted", "replicaSet", klog.KObj(rs))
   // 入队
   dc.enqueueDeployment(d)
}

DeploymentController 启动

前面看了哪些 Event 会向 workqueue 中添加 item，接着看下这些 item 是怎么被消费的。

Run()

Run() 方法本身很简洁，根据给定的并发数，也就是默认 5 并发，启动 dc.worker

  • pkg/controller/deployment/deployment_controller.go:149
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

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

   klog.InfoS("Starting controller", "controller", "deployment")
   defer klog.InfoS("Shutting down controller", "controller", "deployment")

   if !cache.WaitForNamedCacheSync("deployment", stopCh, dc.dListerSynced, dc.rsListerSynced, dc.podListerSynced) {
      return
   }

   for i := 0; i < workers; i++ {
      go wait.Until(dc.worker, time.Second, stopCh)
   }

   <-stopCh
}

继续看 worker 的内容

  • pkg/controller/deployment/deployment_controller.go:460
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func (dc *DeploymentController) worker() {
   for dc.processNextWorkItem() {
   }
}

func (dc *DeploymentController) processNextWorkItem() bool {
   key, quit := dc.queue.Get() // 从 workqueue 中获取一个 item
   if quit {
      return false
   }
   defer dc.queue.Done(key)
   // 主要逻辑
   err := dc.syncHandler(key.(string))
   dc.handleErr(err, key)

   return true
}

这里从 workqueue 里拿到一个 key 之后，通过调用 syncHandler() 方法来处理，前面强调过这行代码：

  • dc.syncHandler = dc.syncDeployment

所以接着我们继续跟 dc.syncDeployment 的实现。

syncDeployment

syncDeployment() 方法做的事情是拿着 workqueue 里出队的 key，根据这个 key 来 sync 对应的 Deployment，继续看下具体的逻辑。

  • pkg/controller/deployment/deployment_controller.go
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

func (dc *DeploymentController) syncDeployment(key string) error {
   // 从 key 中分割出 namespace 和 name
   namespace, name, err := cache.SplitMetaNamespaceKey(key)
   if err != nil {
      klog.ErrorS(err, "Failed to split meta namespace cache key", "cacheKey", key)
      return err
   }

   startTime := time.Now()
   klog.V(4).InfoS("Started syncing deployment", "deployment", klog.KRef(namespace, name), "startTime", startTime)
   defer func() {
      klog.V(4).InfoS("Finished syncing deployment", "deployment", klog.KRef(namespace, name), "duration", time.Since(startTime))
   }()
   // 根据 namespace 和 name 从 cache 中检索对应 Deployment 对象
   deployment, err := dc.dLister.Deployments(namespace).Get(name)
   if errors.IsNotFound(err) {
      klog.V(2).InfoS("Deployment has been deleted", "deployment", klog.KRef(namespace, name))
      return nil
   }
   if err != nil {
      return err
   }

   // 为了不改动这个 cache，这是一个 ThreadSafeStore
   d := deployment.DeepCopy()

   // 空 LabelSelector 会匹配到所有 pods，发一个 Warning Event，更新 .Status.ObservedGeneration 然后返回
   everything := metav1.LabelSelector{}
   if reflect.DeepEqual(d.Spec.Selector, &everything) {
      dc.eventRecorder.Eventf(d, v1.EventTypeWarning, "SelectingAll", "This deployment is selecting all pods. A non-empty selector is required.")
      if d.Status.ObservedGeneration < d.Generation {
         d.Status.ObservedGeneration = d.Generation
         dc.client.AppsV1().Deployments(d.Namespace).UpdateStatus(context.TODO(), d, metav1.UpdateOptions{})
      }
      return nil
   }

   // 获取当前 Deployment 拥有的所有 ReplicaSet，同时会更新这些 ReplicaSet 的 ControllerRef
   rsList, err := dc.getReplicaSetsForDeployment(d)
   if err != nil {
      return err
   }
   // 这个 map 是 map[types.UID][]*v1.Pod 类型，key 是 rs 的 UID，value 是对应 rs 管理的所有 pod 列表
   podMap, err := dc.getPodMapForDeployment(d, rsList)
   if err != nil {
      return err
   }

   // 已经标记要删除了，这时候只更新状态
   if d.DeletionTimestamp != nil {
      return dc.syncStatusOnly(d, rsList)
   }

   // 根据 .Spec.Pause配置看是否更新 Deployment 的 conditions
   if err = dc.checkPausedConditions(d); err != nil {
      return err
   }

   if d.Spec.Paused {
      // Pause 或 scale 时的调谐逻辑
      return dc.sync(d, rsList)
   }

   // 应该过期了，老版本的 "deprecated.deployment.rollback.to" 注解回滚逻辑
   if getRollbackTo(d) != nil {
      // 回滚到旧版本的逻辑
      return dc.rollback(d, rsList)
   }
   // 如果是 scale
   scalingEvent, err := dc.isScalingEvent(d, rsList)
   if err != nil {
      return err
   }
   if scalingEvent {
      // Pause 或 scale 时的调谐逻辑
      return dc.sync(d, rsList)
   }

   switch d.Spec.Strategy.Type {
   // 重建策略
   case apps.RecreateDeploymentStrategyType:
      return dc.rolloutRecreate(d, rsList, podMap)
   // 滚动更新策略
   case apps.RollingUpdateDeploymentStrategyType:
      return dc.rolloutRolling(d, rsList)
   }
   return fmt.Errorf("unexpected deployment strategy type: %s", d.Spec.Strategy.Type)
}

小结

syncDeployment() 方法看完之后，Deployment 控制器的逻辑就算过完一遍了。当然这个方法内部涉及到的一些小方法的调用这里只是简单介绍其功能，并没有深究所有实现细节，不过这些小方法的逻辑都不难，就暂不赘述了。

对 Deployment Controller 的代码分析在 client-go 和 Job Controller 之后，所以这里看起来感觉会很简单，对应有些描述也没有那么详细，如果大家看过前面我发的相关文章，看本文的逻辑应该也不会感觉吃力，反正我希望你先回过头看下我之前发的相关文章，最新版可以在我的博客网站 Daniel Hu’s Blog 查阅。

更新于 2021-11-03
阅读原始文档
 k8s, 源码分析, deployment
返回 | 主页