深入kube-controller-manager：Kubernetes控制器管理器源码剖析

 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

// ControllerManager 控制器管理器主结构体
// 文件路径: cmd/kube-controller-manager/app/controllermanager.go
type ControllerManager struct {
    // GenericControllerManager 通用控制器管理器
    *controllermanager.GenericControllerManager
    
    // SecureServing 安全服务配置
    SecureServing *apiserver.SecureServingInfo
    
    // InsecureServing 非安全服务配置（已废弃）
    InsecureServing *controllermanager.InsecureServingInfo
    
    // Authentication 认证配置
    Authentication apiserver.AuthenticationInfo
    
    // Authorization 授权配置
    Authorization apiserver.AuthorizationInfo
    
    // KubeCloudShared 云平台共享配置
    KubeCloudShared *cpconfig.KubeCloudSharedConfiguration
    
    // ServiceController 服务控制器配置
    ServiceController *serviceconfig.ServiceControllerConfiguration
    
    // AttachDetachController 附加分离控制器配置
    AttachDetachController *attachdetach.AttachDetachControllerConfiguration
    
    // CSRSigningController 证书签名控制器配置
    CSRSigningController *csrsigningconfig.CSRSigningControllerConfiguration
    
    // Extra 额外配置
    Extra ExtraConfig
}

// ExtraConfig 额外配置结构体
type ExtraConfig struct {
    // ClientBuilder 客户端构建器
    ClientBuilder clientbuilder.ControllerClientBuilder
    
    // EventRecorder 事件记录器
    EventRecorder events.EventRecorder
    
    // LeaderElection 领导者选举配置
    LeaderElection *leaderelection.LeaderElectionConfig
    
    // LeaderElectionResourceLock 领导者选举资源锁类型
    LeaderElectionResourceLock string
}

// ControllerContext 控制器上下文
type ControllerContext struct {
    // ClientBuilder 客户端构建器
    ClientBuilder clientbuilder.ControllerClientBuilder
    
    // InformerFactory Informer工厂
    InformerFactory informers.SharedInformerFactory
    
    // ComponentConfig 组件配置
    ComponentConfig kubectrlmgrconfig.KubeControllerManagerConfiguration
    
    // DeferredDiscoveryRESTMapper 延迟发现REST映射器
    DeferredDiscoveryRESTMapper *restmapper.DeferredDiscoveryRESTMapper
    
    // AvailableResources 可用资源映射
    AvailableResources map[schema.GroupVersionResource]bool
    
    // Cloud 云平台接口
    Cloud cloudprovider.Interface
    
    // LoopMode 循环模式
    LoopMode ControllerManagerLoopMode
    
    // Stop 停止信号通道
    Stop <-chan struct{}
    
    // InformersStarted Informers启动信号通道
    InformersStarted chan struct{}
    
    // ResyncPeriod 重新同步周期
    ResyncPeriod func() time.Duration
    
    // ControllerManagerMetrics 控制器管理器指标
    ControllerManagerMetrics *controllermanager.ControllerManagerMetrics
}

 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

// Controller 控制器接口
// 文件路径: pkg/controller/controller_utils.go
type Controller interface {
    // Run 运行控制器
    // workers: 工作协程数量
    // stopCh: 停止信号通道
    Run(workers int, stopCh <-chan struct{})
    
    // HasSynced 检查控制器是否已同步
    HasSynced() bool
    
    // LastSyncResourceVersion 返回最后同步的资源版本
    LastSyncResourceVersion() string
}

// ControllerInitFunc 控制器初始化函数类型
type ControllerInitFunc func(ctx ControllerContext) (controller Controller, enabled bool, err error)

// InitFunc 控制器初始化函数映射
var InitFuncs = map[string]ControllerInitFunc{
    "deployment":                   startDeploymentController,
    "replicaset":                   startReplicaSetController,
    "node":                         startNodeController,
    "service":                      startServiceController,
    "namespace":                    startNamespaceController,
    "persistentvolume-binder":      startPersistentVolumeBinderController,
    "serviceaccount":               startServiceAccountController,
    "endpoints":                    startEndpointController,
    "resourcequota":                startResourceQuotaController,
    "job":                          startJobController,
    "cronjob":                      startCronJobController,
    "daemonset":                    startDaemonSetController,
    "statefulset":                  startStatefulSetController,
    "certificatesigningrequest":    startCSRSigningController,
    "disruption":                   startDisruptionController,
    // ... 更多控制器
}

// NewControllerInitializers 创建控制器初始化器映射
func NewControllerInitializers(loopMode ControllerManagerLoopMode) map[string]ControllerInitFunc {
    controllers := map[string]ControllerInitFunc{}
    
    // 注册所有控制器
    for name, initFunc := range InitFuncs {
        controllers[name] = initFunc
    }
    
    return controllers
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// main kube-controller-manager主函数入口
// 文件路径: cmd/kube-controller-manager/controller-manager.go
func main() {
    // 创建kube-controller-manager命令对象
    command := app.NewControllerManagerCommand()
    
    // 使用component-base的CLI运行器执行命令
    code := cli.Run(command)
    
    // 以返回的退出码结束程序
    os.Exit(code)
}

  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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169

// NewControllerManagerCommand 创建控制器管理器命令
// 文件路径: cmd/kube-controller-manager/app/controllermanager.go
func NewControllerManagerCommand() *cobra.Command {
    // 创建默认配置
    s, err := options.NewKubeControllerManagerOptions()
    if err != nil {
        klog.ErrorS(err, "Unable to initialize command options")
        klog.FlushAndExit(klog.ExitFlushTimeout, 1)
    }

    cmd := &cobra.Command{
        Use: "kube-controller-manager",
        Long: `The Kubernetes controller manager is a daemon that embeds
the core control loops shipped with Kubernetes. In applications of robotics and
automation, a control loop is a non-terminating loop that regulates the state of
the system. In Kubernetes, a controller is a control loop that watches the shared
state of the cluster through the apiserver and makes changes attempting to move the
current state towards the desired state. Examples of controllers that ship with
Kubernetes today are the replication controller, endpoints controller, namespace
controller, and serviceaccounts controller.`,

        // 禁用在命令错误时打印使用说明
        SilenceUsage: true,
        
        // 预运行钩子
        PersistentPreRunE: func(*cobra.Command, []string) error {
            // 设置默认的klog标志
            if err := s.Complete(); err != nil {
                return err
            }
            
            if err := s.Validate(); err != nil {
                return err
            }
            
            return nil
        },
        
        // 主运行函数
        RunE: func(cmd *cobra.Command, args []string) error {
            verflag.PrintAndExitIfRequested()
            
            // 应用配置
            c, err := s.Config(KnownControllers(), ControllersDisabledByDefault.List())
            if err != nil {
                return err
            }
            
            // 运行控制器管理器
            return Run(c.Complete(), setupSignalHandler())
        },
        
        Args: func(cmd *cobra.Command, args []string) error {
            for _, arg := range args {
                if len(arg) > 0 {
                    return fmt.Errorf("%q does not take any arguments, got %q", cmd.CommandPath(), args)
                }
            }
            return nil
        },
    }

    // 添加标志
    fs := cmd.Flags()
    namedFlagSets := s.Flags(KnownControllers(), ControllersDisabledByDefault.List())
    verflag.AddFlags(namedFlagSets.FlagSet("global"))
    globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name(), logs.SkipLoggingConfigurationFlags())
    
    for _, f := range namedFlagSets.FlagSets {
        fs.AddFlagSet(f)
    }

    cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
    cliflag.SetUsageAndHelpFunc(cmd, namedFlagSets, cols)

    return cmd
}

// Run 运行控制器管理器
func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
    // 运行控制器管理器
    run := func(ctx context.Context, startSATokenController InitFunc, initializersFunc ControllerInitializersFunc) {
        // 创建控制器上下文
        controllerContext, err := CreateControllerContext(c, rootClientBuilder, clientBuilder, ctx.Done())
        if err != nil {
            klog.ErrorS(err, "Error building controller context")
            return
        }
        
        // 启动SAToken控制器
        saTokenControllerInitFunc := startSATokenController(controllerContext)
        if saTokenControllerInitFunc != nil {
            go saTokenControllerInitFunc(ctx, controllerContext)
        }
        
        // 启动所有控制器
        if err := StartControllers(ctx, controllerContext, initializersFunc, unsecuredMux, healthzHandler); err != nil {
            klog.ErrorS(err, "Error starting controllers")
            return
        }
        
        // 启动Informers
        controllerContext.InformerFactory.Start(ctx.Done())
        controllerContext.ObjectOrMetadataInformerFactory.Start(ctx.Done())
        close(controllerContext.InformersStarted)
        
        // 等待停止信号
        <-ctx.Done()
    }
    
    // 如果启用了领导者选举，则在选举成功后运行
    if c.ComponentConfig.LeaderElection.LeaderElect {
        id, err := os.Hostname()
        if err != nil {
            return err
        }
        
        // 添加唯一标识符避免冲突
        id = id + "_" + string(uuid.NewUUID())
        
        // 配置领导者选举
        rl, err := resourcelock.New(c.ComponentConfig.LeaderElection.ResourceLock,
            c.ComponentConfig.LeaderElection.ResourceNamespace,
            c.ComponentConfig.LeaderElection.ResourceName,
            c.LeaderElectionClient.CoreV1(),
            c.LeaderElectionClient.CoordinationV1(),
            resourcelock.ResourceLockConfig{
                Identity:      id,
                EventRecorder: c.EventRecorder,
            })
        if err != nil {
            return fmt.Errorf("couldn't create resource lock: %v", err)
        }
        
        // 启动领导者选举
        leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
            Lock:          rl,
            LeaseDuration: c.ComponentConfig.LeaderElection.LeaseDuration.Duration,
            RenewDeadline: c.ComponentConfig.LeaderElection.RenewDeadline.Duration,
            RetryPeriod:   c.ComponentConfig.LeaderElection.RetryPeriod.Duration,
            Callbacks: leaderelection.LeaderCallbacks{
                OnStartedLeading: func(ctx context.Context) {
                    initializersFunc := NewControllerInitializers
                    run(ctx, startSATokenController, initializersFunc)
                },
                OnStoppedLeading: func() {
                    klog.ErrorS(nil, "Leaderelection lost")
                    klog.FlushAndExit(klog.ExitFlushTimeout, 1)
                },
            },
            WatchDog: electionChecker,
            Name:     "kube-controller-manager",
        })
        
        panic("unreachable")
    } else {
        // 直接运行，不进行领导者选举
        ctx, cancel := context.WithCancel(context.TODO())
        defer cancel()
        go func() {
            <-stopCh
            cancel()
        }()
        
        run(ctx, startSATokenController, NewControllerInitializers)
    }
    
    return nil
}

 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

// StartControllers 启动所有控制器
// 文件路径: cmd/kube-controller-manager/app/controllermanager.go
func StartControllers(ctx context.Context, controllerCtx ControllerContext, controllers map[string]ControllerInitFunc, 
    unsecuredMux *mux.PathRecorderMux, healthzHandler *controllerhealthz.MutableHealthzHandler) error {
    
    // 启动每个控制器
    for controllerName, initFn := range controllers {
        if !controllerCtx.IsControllerEnabled(controllerName) {
            klog.InfoS("Controller is disabled", "controller", controllerName)
            continue
        }
        
        klog.V(1).InfoS("Starting controller", "controller", controllerName)
        
        // 调用控制器初始化函数
        ctrl, started, err := initFn(controllerCtx)
        if err != nil {
            klog.ErrorS(err, "Error starting controller", "controller", controllerName)
            return err
        }
        if !started {
            klog.InfoS("Controller not started", "controller", controllerName)
            continue
        }
        
        // 添加健康检查
        check := controllerhealthz.NamedCheck(controllerName, func(req *http.Request) error {
            return ctrl.Healthz()
        })
        healthzHandler.AddHealthChecker(check)
        
        klog.InfoS("Started controller", "controller", controllerName)
    }
    
    return nil
}

// CreateControllerContext 创建控制器上下文
func CreateControllerContext(c *config.CompletedConfig, rootClientBuilder, clientBuilder clientbuilder.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {
    // 创建版本化Informer工厂
    versionedClient := rootClientBuilder.ClientOrDie("shared-informers")
    sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(c)())
    
    // 创建元数据Informer工厂
    metadataClient := metadata.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("metadata-informers"))
    metadataInformers := metadatainformer.NewSharedInformerFactory(metadataClient, ResyncPeriod(c)())
    
    // 如果启用了动态资源分配，创建动态Informer工厂
    var dynamicInformers dynamicinformer.DynamicSharedInformerFactory
    if utilfeature.DefaultFeatureGate.Enabled(features.DynamicResourceAllocation) {
        dynamicClient := dynamic.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("dynamic-resource-allocation-informers"))
        dynamicInformers = dynamicinformer.NewFilteredDynamicSharedInformerFactory(dynamicClient, ResyncPeriod(c)(), metav1.NamespaceAll, nil)
    }
    
    // 创建可用资源映射
    availableResources, err := GetAvailableResources(rootClientBuilder)
    if err != nil {
        return ControllerContext{}, err
    }
    
    // 创建云平台接口
    cloud, loopMode, err := createCloudProvider(c.ComponentConfig.KubeCloudShared.CloudProvider.Name, c.ComponentConfig.KubeCloudShared.ExternalCloudVolumePlugin,
        c.ComponentConfig.KubeCloudShared.CloudProvider.CloudConfigFile, c.ComponentConfig.KubeCloudShared.AllowUntaggedCloud, sharedInformers)
    if err != nil {
        return ControllerContext{}, err
    }
    
    // 返回控制器上下文
    return ControllerContext{
        ClientBuilder:                   clientBuilder,
        InformerFactory:                sharedInformers,
        ObjectOrMetadataInformerFactory: informerfactory.NewInformerFactory(sharedInformers, metadataInformers),
        DynamicInformerFactory:         dynamicInformers,
        ComponentConfig:                c.ComponentConfig,
        RESTMapper:                     restMapper,
        AvailableResources:             availableResources,
        Cloud:                          cloud,
        LoopMode:                       loopMode,
        Stop:                           stop,
        InformersStarted:               make(chan struct{}),
        ResyncPeriod:                   ResyncPeriod(c),
        ControllerManagerMetrics:       controllermanager.NewControllerManagerMetrics("kube-controller-manager"),
    }, nil
}

  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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155

// Controller 通用控制器结构体
// 文件路径: pkg/controller/controller_utils.go
type Controller struct {
    // 名称
    name string
    
    // Kubernetes客户端
    kubeClient clientset.Interface
    
    // 工作队列
    queue workqueue.RateLimitingInterface
    
    // Informer和Lister
    informer cache.SharedIndexInformer
    lister   cache.GenericLister
    
    // 同步函数
    syncFunc func(key string) error
    
    // 事件记录器
    recorder record.EventRecorder
    
    // 工作协程数
    workers int
    
    // 同步状态
    hasSynced func() bool
}

// NewController 创建通用控制器
func NewController(name string, kubeClient clientset.Interface, informer cache.SharedIndexInformer, 
    syncFunc func(string) error) *Controller {
    
    // 创建工作队列
    queue := workqueue.NewNamedRateLimitingQueue(
        workqueue.DefaultControllerRateLimiter(), name)
    
    // 创建事件记录器
    eventBroadcaster := record.NewBroadcaster()
    eventBroadcaster.StartStructuredLogging(0)
    eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{
        Interface: kubeClient.CoreV1().Events(""),
    })
    recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: name})
    
    controller := &Controller{
        name:       name,
        kubeClient: kubeClient,
        queue:      queue,
        informer:   informer,
        lister:     listers.NewGenericLister(informer.GetIndexer(), resource),
        syncFunc:   syncFunc,
        recorder:   recorder,
        workers:    1,
        hasSynced:  informer.HasSynced,
    }
    
    // 设置事件处理器
    informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            controller.enqueue(obj)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            controller.enqueue(newObj)
        },
        DeleteFunc: func(obj interface{}) {
            controller.enqueue(obj)
        },
    })
    
    return controller
}

// Run 运行控制器
func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    defer c.queue.ShutDown()
    
    klog.InfoS("Starting controller", "controller", c.name)
    defer klog.InfoS("Shutting down controller", "controller", c.name)
    
    // 等待缓存同步
    if !cache.WaitForCacheSync(stopCh, c.hasSynced) {
        utilruntime.HandleError(fmt.Errorf("failed to wait for %s caches to sync", c.name))
        return
    }
    
    // 启动工作协程
    for i := 0; i < workers; i++ {
        go wait.Until(c.runWorker, time.Second, stopCh)
    }
    
    <-stopCh
}

// runWorker 运行工作协程
func (c *Controller) runWorker() {
    for c.processNextWorkItem() {
    }
}

// processNextWorkItem 处理下一个工作项
func (c *Controller) processNextWorkItem() bool {
    obj, shutdown := c.queue.Get()
    if shutdown {
        return false
    }
    
    // 处理工作项
    err := func(obj interface{}) error {
        defer c.queue.Done(obj)
        
        var key string
        var ok bool
        
        if key, ok = obj.(string); !ok {
            // 无效的工作项，丢弃
            c.queue.Forget(obj)
            utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
            return nil
        }
        
        // 调用同步函数处理
        if err := c.syncFunc(key); err != nil {
            // 同步失败，重新入队
            c.queue.AddRateLimited(key)
            return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
        }
        
        // 同步成功，从队列中移除
        c.queue.Forget(obj)
        klog.V(4).InfoS("Successfully synced", "controller", c.name, "key", key)
        return nil
    }(obj)
    
    if err != nil {
        utilruntime.HandleError(err)
        return true
    }
    
    return true
}

// enqueue 将对象加入工作队列
func (c *Controller) enqueue(obj interface{}) {
    var key string
    var err error
    
    if key, err = cache.DeletionHandlingMetaNamespaceKeyFunc(obj); err != nil {
        utilruntime.HandleError(err)
        return
    }
    
    c.queue.Add(key)
}

  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
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368

// DeploymentController Deployment控制器结构体
// 文件路径: pkg/controller/deployment/deployment_controller.go
type DeploymentController struct {
    // rsControl ReplicaSet控制接口
    rsControl controller.RSControlInterface
    
    // client Kubernetes客户端
    client clientset.Interface
    
    // eventRecorder 事件记录器
    eventRecorder record.EventRecorder
    
    // syncHandler 同步处理函数
    syncHandler func(ctx context.Context, dKey string) error
    
    // enqueueDeployment 入队函数
    enqueueDeployment func(deployment *apps.Deployment)
    
    // Deployment相关
    dLister appslisters.DeploymentLister
    dListerSynced cache.InformerSynced
    
    // ReplicaSet相关
    rsLister appslisters.ReplicaSetLister
    rsListerSynced cache.InformerSynced
    
    // Pod相关
    podLister corelisters.PodLister
    podListerSynced cache.InformerSynced
    
    // 工作队列
    queue workqueue.RateLimitingInterface
}

// NewDeploymentController 创建Deployment控制器
func NewDeploymentController(ctx context.Context, dInformer appsinformers.DeploymentInformer, rsInformer appsinformers.ReplicaSetInformer, podInformer coreinformers.PodInformer, client clientset.Interface) (*DeploymentController, error) {
    eventBroadcaster := record.NewBroadcaster()
    eventBroadcaster.StartStructuredLogging(0)
    eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: client.CoreV1().Events("")})
    
    dc := &DeploymentController{
        client:        client,
        eventRecorder: eventBroadcaster.NewRecorder(scheme.Scheme, v1.EventSource{Component: "deployment-controller"}),
        queue:         workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "deployment"),
    }
    
    dc.rsControl = controller.RealRSControl{
        KubeClient: client,
        Recorder:   dc.eventRecorder,
    }
    
    dc.syncHandler = dc.syncDeployment
    dc.enqueueDeployment = dc.enqueue
    
    // 设置Deployment事件处理器
    dInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            dc.addDeployment(logger, obj)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            dc.updateDeployment(logger, oldObj, newObj)
        },
        DeleteFunc: func(obj interface{}) {
            dc.deleteDeployment(logger, obj)
        },
    })
    dc.dLister = dInformer.Lister()
    dc.dListerSynced = dInformer.Informer().HasSynced
    
    // 设置ReplicaSet事件处理器
    rsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc: func(obj interface{}) {
            dc.addReplicaSet(logger, obj)
        },
        UpdateFunc: func(oldObj, newObj interface{}) {
            dc.updateReplicaSet(logger, oldObj, newObj)
        },
        DeleteFunc: func(obj interface{}) {
            dc.deleteReplicaSet(logger, obj)
        },
    })
    dc.rsLister = rsInformer.Lister()
    dc.rsListerSynced = rsInformer.Informer().HasSynced
    
    // 设置Pod事件处理器
    podInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        DeleteFunc: func(obj interface{}) {
            dc.deletePod(logger, obj)
        },
    })
    dc.podLister = podInformer.Lister()
    dc.podListerSynced = podInformer.Informer().HasSynced
    
    return dc, nil
}

// syncDeployment 同步Deployment的核心函数
func (dc *DeploymentController) syncDeployment(ctx context.Context, key string) error {
    logger := klog.FromContext(ctx)
    startTime := time.Now()
    logger.V(4).Info("Started syncing deployment", "key", key, "startTime", startTime)
    defer func() {
        logger.V(4).Info("Finished syncing deployment", "key", key, "duration", time.Since(startTime))
    }()
    
    // 解析键获取命名空间和名称
    namespace, name, err := cache.SplitMetaNamespaceKey(key)
    if err != nil {
        return err
    }
    
    // 从缓存中获取Deployment
    deployment, err := dc.dLister.Deployments(namespace).Get(name)
    if errors.IsNotFound(err) {
        logger.V(2).Info("Deployment has been deleted", "deployment", klog.KRef(namespace, name))
        return nil
    }
    if err != nil {
        return err
    }
    
    // 深拷贝Deployment，避免修改缓存中的对象
    d := deployment.DeepCopy()
    
    // 检查Deployment是否暂停
    if d.Spec.Paused {
        return dc.sync(ctx, d, nil)
    }
    
    // 获取该Deployment的所有ReplicaSets
    rsList, err := dc.getReplicaSetsForDeployment(ctx, d)
    if err != nil {
        return err
    }
    
    // 获取该Deployment的所有Pods
    podMap, err := dc.getPodMapForDeployment(d, rsList)
    if err != nil {
        return err
    }
    
    // 检查是否需要回滚
    if getRollbackTo(d) != nil {
        return dc.rollback(ctx, d, rsList)
    }
    
    // 检查是否需要扩缩容
    scalingEvent, err := dc.isScalingEvent(ctx, d, rsList)
    if err != nil {
        return err
    }
    if scalingEvent {
        return dc.sync(ctx, d, rsList)
    }
    
    // 根据部署策略进行更新
    switch d.Spec.Strategy.Type {
    case apps.RecreateDeploymentStrategyType:
        return dc.rolloutRecreate(ctx, d, rsList, podMap)
    case apps.RollingUpdateDeploymentStrategyType:
        return dc.rolloutRolling(ctx, d, rsList)
    }
    
    return fmt.Errorf("unexpected deployment strategy type: %s", d.Spec.Strategy.Type)
}

// rolloutRolling 执行滚动更新
func (dc *DeploymentController) rolloutRolling(ctx context.Context, d *apps.Deployment, rsList []*apps.ReplicaSet) error {
    // 找到新的和旧的ReplicaSet
    newRS, oldRSs, err := dc.getAllReplicaSetsAndSyncRevision(ctx, d, rsList, true)
    if err != nil {
        return err
    }
    
    // 获取所有活跃的ReplicaSets
    allRSs := append(oldRSs, newRS)
    
    // 扩容新的ReplicaSet
    scaledUp, err := dc.reconcileNewReplicaSet(ctx, allRSs, newRS, d)
    if err != nil {
        return err
    }
    if scaledUp {
        // 如果进行了扩容，等待扩容完成再继续
        return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
    }
    
    // 缩容旧的ReplicaSets
    scaledDown, err := dc.reconcileOldReplicaSets(ctx, allRSs, controller.FilterActiveReplicaSets(oldRSs), newRS, d)
    if err != nil {
        return err
    }
    if scaledDown {
        // 如果进行了缩容，等待缩容完成再继续
        return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
    }
    
    // 检查是否需要清理旧的ReplicaSets
    if deploymentutil.DeploymentComplete(d, &d.Status) {
        if err := dc.cleanupDeployment(ctx, oldRSs, d); err != nil {
            return err
        }
    }
    
    // 同步部署状态
    return dc.syncRolloutStatus(ctx, allRSs, newRS, d)
}

// reconcileNewReplicaSet 协调新的ReplicaSet
func (dc *DeploymentController) reconcileNewReplicaSet(ctx context.Context, allRSs []*apps.ReplicaSet, newRS *apps.ReplicaSet, deployment *apps.Deployment) (bool, error) {
    if newRS == nil {
        return false, nil
    }
    
    // 计算期望的副本数
    desiredReplicas := *(deployment.Spec.Replicas)
    currentReplicas := newRS.Status.Replicas
    
    // 计算滚动更新参数
    maxSurge := deploymentutil.MaxSurge(*deployment)
    
    // 计算新ReplicaSet的最大副本数
    // 不能超过期望副本数 + maxSurge
    allPodsCount := deploymentutil.GetReplicaCountForReplicaSets(allRSs)
    maxTotalPods := desiredReplicas + maxSurge
    if allPodsCount >= maxTotalPods {
        // 已经达到最大Pod数，不能再扩容
        return false, nil
    }
    
    // 计算可以扩容的副本数
    scaleUpCount := maxTotalPods - allPodsCount
    scaleUpCount = integer.IntMin(scaleUpCount, desiredReplicas-currentReplicas)
    
    if scaleUpCount <= 0 {
        return false, nil
    }
    
    // 执行扩容
    newReplicasCount := currentReplicas + scaleUpCount
    scaledRS, err := dc.scaleReplicaSetAndRecordEvent(ctx, newRS, newReplicasCount, deployment)
    if err != nil {
        return false, err
    }
    
    if scaledRS.Spec.Replicas != scaledRS.Status.Replicas {
        // 扩容正在进行中
        return true, nil
    }
    
    return false, nil
}

// reconcileOldReplicaSets 协调旧的ReplicaSets
func (dc *DeploymentController) reconcileOldReplicaSets(ctx context.Context, allRSs, oldRSs []*apps.ReplicaSet, newRS *apps.ReplicaSet, deployment *apps.Deployment) (bool, error) {
    // 计算需要缩容的副本数
    oldPodsCount := deploymentutil.GetReplicaCountForReplicaSets(oldRSs)
    if oldPodsCount == 0 {
        // 没有旧Pod，无需缩容
        return false, nil
    }
    
    // 计算滚动更新参数
    maxUnavailable := deploymentutil.MaxUnavailable(*deployment)
    desiredReplicas := *(deployment.Spec.Replicas)
    
    // 计算可用的Pod数量
    totalPodsCount := deploymentutil.GetReplicaCountForReplicaSets(allRSs)
    availablePodCount := deploymentutil.GetAvailableReplicaCountForReplicaSets(allRSs)
    
    // 计算可以缩容的副本数
    minAvailable := desiredReplicas - maxUnavailable
    newRSUnavailablePodCount := newRS.Status.Replicas - newRS.Status.AvailableReplicas
    maxScaledDown := availablePodCount - minAvailable - newRSUnavailablePodCount
    if maxScaledDown <= 0 {
        return false, nil
    }
    
    // 计算实际要缩容的副本数
    totalScaleDownCount := integer.IntMin(maxScaledDown, oldPodsCount)
    targetRS := deploymentutil.FindOldReplicaSets(deployment, allRSs)
    
    // 执行缩容
    totalScaledDown := int32(0)
    for _, targetRS := range targetRS {
        if totalScaledDown >= totalScaleDownCount {
            break
        }
        
        if targetRS.Spec.Replicas == 0 {
            // 已经缩容到0，跳过
            continue
        }
        
        // 计算这个ReplicaSet要缩容的数量
        scaleDownCount := integer.IntMin(targetRS.Spec.Replicas, totalScaleDownCount-totalScaledDown)
        newReplicasCount := targetRS.Spec.Replicas - scaleDownCount
        
        // 执行缩容
        _, err := dc.scaleReplicaSetAndRecordEvent(ctx, targetRS, newReplicasCount, deployment)
        if err != nil {
            return false, err
        }
        
        totalScaledDown += scaleDownCount
    }
    
    return totalScaledDown > 0, nil
}

// scaleReplicaSetAndRecordEvent 扩缩容ReplicaSet并记录事件
func (dc *DeploymentController) scaleReplicaSetAndRecordEvent(ctx context.Context, rs *apps.ReplicaSet, newScale int32, deployment *apps.Deployment) (*apps.ReplicaSet, error) {
    // 如果副本数没有变化，直接返回
    if rs.Spec.Replicas == newScale {
        return rs, nil
    }
    
    var scalingOperation string
    if rs.Spec.Replicas < newScale {
        scalingOperation = "up"
    } else {
        scalingOperation = "down"
    }
    
    // 更新ReplicaSet的副本数
    scaled, err := dc.scaleReplicaSet(ctx, rs, newScale, deployment, scalingOperation)
    if err != nil {
        dc.eventRecorder.Eventf(deployment, v1.EventTypeWarning, "FailedScale", "Failed to scale %s %s/%s: %v", rs.Kind, rs.Namespace, rs.Name, err)
        return nil, err
    }
    
    // 记录扩缩容事件
    dc.eventRecorder.Eventf(deployment, v1.EventTypeNormal, "ScalingReplicaSet", "Scaled %s replica set %s to %d", scalingOperation, rs.Name, newScale)
    
    return scaled, nil
}

// scaleReplicaSet 扩缩容ReplicaSet
func (dc *DeploymentController) scaleReplicaSet(ctx context.Context, rs *apps.ReplicaSet, newScale int32, deployment *apps.Deployment, scalingOperation string) (*apps.ReplicaSet, error) {
    // 计算重试次数
    size := *(rs.Spec.Replicas)
    name := rs.Name
    
    // 使用指数退避重试
    err := retry.RetryOnConflict(retry.DefaultBackoff, func() error {
        // 重新获取ReplicaSet以获取最新版本
        rs.Spec.Replicas = &newScale
        _, updateErr := dc.client.AppsV1().ReplicaSets(rs.Namespace).Update(ctx, rs, metav1.UpdateOptions{})
        if updateErr == nil {
            return nil
        }
        
        // 如果是冲突错误，重新获取并重试
        if errors.IsConflict(updateErr) {
            updated, err := dc.rsLister.ReplicaSets(rs.Namespace).Get(rs.Name)
            if err == nil {
                rs = updated.DeepCopy()
            }
        }
        return updateErr
    })
    
    if err != nil {
        return nil, fmt.Errorf("failed to scale ReplicaSet %s/%s to %d: %v", rs.Namespace, rs.Name, newScale, err)
    }
    
    return rs, nil
}