controller-manager学习三部曲之二：源码学习

欢迎访问我的GitHub

这里分类和汇总了欣宸的全部原创(含配套源码)：https://github.com/zq2599/blog_demos

本篇概览

作为《controller-manager学习三部曲》系列的第二篇，前面通过shell脚本找到了程序的入口，接下来咱们来学习controller-manager的源码
学习源码虽然收获巨大，然而耗时耗力，为了让学习变得轻松，这里提前梳理一下，以确保主方向正确

查看启动命令：了解真实kubernetes环境下controll-manager的启动命令，知道会传入那些参数
入口：搞清楚代码从哪里开始看，然后分析如何处理输入的参数
初始化配置：自身有大量默认参数，还有启动时输入的大量参数，使用这些信息进行初始化配置
深入理解启动逻辑：先学习启动controller的大框架，然后重点分析两个重要方法，创建上下文的CreateControllerContext，调用每个controller启动的StartControllers

总的来说，复杂的controller-manager就是由这些部分组成

查看启动命令

找个现成的kubernetes系统，看一下真正运行的controller-manager的启动命令是啥样的
以我自己测试用的kubernetes环境为例，先查看pod名

kubectl get pods -n kube-system
NAME                           READY   STATUS    RESTARTS       AGE
coredns-78fcd69978-jztff       1/1     Running   6 (35d ago)    125d
coredns-78fcd69978-ts7gq       1/1     Running   6 (35d ago)    125d
etcd-hedy                      1/1     Running   6 (35d ago)    125d
kube-apiserver-hedy            1/1     Running   7 (35d ago)    125d
kube-controller-manager-hedy   1/1     Running   11 (30h ago)   125d
kube-proxy-2qx6k               1/1     Running   6              125d
kube-scheduler-hedy            1/1     Running   11 (30h ago)   125d

可见controller-manager的pod名是kube-controller-manager-hedy，执行以下命令即可查看看pod的详细信息

kubectl describe pod kube-controller-manager-hedy -n kube-system

上述命令会输出大量信息，这里只展示我们最关心的内容，即controller-manager的启动命令

    Command:kube-controller-manager--allocate-node-cidrs=true--authentication-kubeconfig=/etc/kubernetes/controller-manager.conf--authorization-kubeconfig=/etc/kubernetes/controller-manager.conf--bind-address=0.0.0.0--client-ca-file=/etc/kubernetes/pki/ca.crt--cluster-cidr=100.64.0.0/10--cluster-name=kubernetes--cluster-signing-cert-file=/etc/kubernetes/pki/ca.crt--cluster-signing-key-file=/etc/kubernetes/pki/ca.key--controllers=*,bootstrapsigner,tokencleaner--experimental-cluster-signing-duration=876000h--feature-gates=TTLAfterFinished=true,EphemeralContainers=true--kubeconfig=/etc/kubernetes/controller-manager.conf--leader-elect=true--port=0--requestheader-client-ca-file=/etc/kubernetes/pki/front-proxy-ca.crt--root-ca-file=/etc/kubernetes/pki/ca.crt--service-account-private-key-file=/etc/kubernetes/pki/sa.key--service-cluster-ip-range=10.96.0.0/22--use-service-account-credentials=true

分析上述信息得到以下结论：

没有用到子命令
有多个标志（flag），所有参数都是flag参数，没有命令参数（命令、子命令名、标志、命令参数、标志参数这些都是cobra的概念）

接下来可以看代码了

入口

看过启动命令后，寻找入口代码也就简单了：找到cobra的命令function定义即可
先看main函数，在kubernetes/cmd/kube-controller-manager/controller-manager.go文件中

func main() {command := app.NewControllerManagerCommand()code := cli.Run(command)os.Exit(code)
}

进入NewControllerManagerCommand方法中，整个方法的代码如下图，我将其分为三部分，请按照序号来阅读（只要是2和3的顺序不要弄错了）
第一部分就是第一行代码，如下，为每个内置controller创建默认配置文件

s, err := options.NewKubeControllerManagerOptions()

这个NewKubeControllerManagerOptions方法也值得一看，如下图，就是为所有controller创建了保存配置信息的对象，都放在数据结构KubeControllerManagerOptions中返回
再回到NewControllerManagerCommand，该看第二部分了，这里的namedFlagSets需要注意，可以这么理解：每个controller都有一个flag集合（里面是多个flag），所以多个controller就有多个flag集合，全部存放在namedFlagSets中，通过controller的name来存取

// 准备一个flag集合，注意，flag是cobra中的概念
fs := cmd.Flags()
// 每个controller都有一个flag集合（里面是多个flag），所以多个controller就有多个flag集合，全部存放在namedFlagSets中，通过controller的name来存取
namedFlagSets := s.Flags(KnownControllers(), ControllersDisabledByDefault.List())
verflag.AddFlags(namedFlagSets.FlagSet("global"))
globalflag.AddGlobalFlags(namedFlagSets.FlagSet("global"), cmd.Name(), logs.SkipLoggingConfigurationFlags())
registerLegacyGlobalFlags(namedFlagSets)
for _, f := range namedFlagSets.FlagSets {fs.AddFlagSet(f)
}

上面的代码还有一处需要注意，就是s.Flags方法中为每个controller都创建了里面创建了那么多flag，到底用在哪里了？如下图，在fs.AddFlagSet方法中，根据实际输入的controller-name取出s.Flags中对应的controller对象，放入cmd.fs中，然后就成为cmd的启动参数的一部分
上诉代码完成了输入flag和数据结构对象的关联
接下来就是第三部分：创建cobra的Command对象，了解cobra的读者应该清楚，命令响应逻辑就在这个Command的参数中，这里亦是如此，通过RunE参数传入了整个应用的启动方法，也就是说进程启动后，就会运行RunE定义的方法

RunE: func(cmd *cobra.Command, args []string) error {// 如果启动命令传入了"--version"，就打印版本信息然后退出进程verflag.PrintAndExitIfRequested()// 验证日志服务的设置并使之生效(如格式、文件目录等)if err := logsapi.ValidateAndApply(s.Logs, utilfeature.DefaultFeatureGate); err != nil {return err}// 打印所有flag的名字和值(这里的flag是cobra中的概念)cliflag.PrintFlags(cmd.Flags())// 配置初始化：注册controller，校验配置，生成配置对象，生成客户端对象clientSet// 这里有些重要的逻辑，稍后会详细说明c, err := s.Config(KnownControllers(), ControllersDisabledByDefault.List())if err != nil {return err}// 监控指标配置utilfeature.DefaultMutableFeatureGate.AddMetrics()// controller-manager的业务逻辑启动return Run(context.Background(), c.Complete())
}

上述就是controller-manager的启动代码，尽管已经添加了详细的注释，仍有两处重点需要展开说明：

s.Config(KnownControllers(), ControllersDisabledByDefault.List()) ：这里面涉及到一些重要的初始化逻辑和数据结构
return Run(context.Background(), c.Complete())：具体的启动逻辑在这里面，会启动各controller

重要：初始化配置

接下来展开第一个重点，也就是下图黄色箭头所指的这行
先看KnownControllers方法，这里面调用了NewControllerInitializers方法，这个NewControllerInitializers很重要，也是咱们开发controller时非常值得借鉴的代码，它返回了一个map，key是一个controller的名字，value是这个controller的初始化方法，例如deployment的controller，其初始化方法是startDeploymentController，这里只用到了key，所有的key代表所有的controller名

func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {controllers := map[string]InitFunc{}// All of the controllers must have unique names, or else we will explode.register := func(name string, fn InitFunc) {if _, found := controllers[name]; found {panic(fmt.Sprintf("controller name %q was registered twice", name))}controllers[name] = fn}register("endpoint", startEndpointController)register("endpointslice", startEndpointSliceController)register("endpointslicemirroring", startEndpointSliceMirroringController)register("replicationcontroller", startReplicationController)register("podgc", startPodGCController)register("resourcequota", startResourceQuotaController)register("namespace", startNamespaceController)register("serviceaccount", startServiceAccountController)register("garbagecollector", startGarbageCollectorController)register("daemonset", startDaemonSetController)register("job", startJobController)register("deployment", startDeploymentController)register("replicaset", startReplicaSetController)register("horizontalpodautoscaling", startHPAController)register("disruption", startDisruptionController)register("statefulset", startStatefulSetController)register("cronjob", startCronJobController)register("csrsigning", startCSRSigningController)register("csrapproving", startCSRApprovingController)register("csrcleaner", startCSRCleanerController)register("ttl", startTTLController)register("bootstrapsigner", startBootstrapSignerController)register("tokencleaner", startTokenCleanerController)register("nodeipam", startNodeIpamController)register("nodelifecycle", startNodeLifecycleController)if loopMode == IncludeCloudLoops {register("service", startServiceController)register("route", startRouteController)register("cloud-node-lifecycle", startCloudNodeLifecycleController)// TODO: volume controller into the IncludeCloudLoops only set.}register("persistentvolume-binder", startPersistentVolumeBinderController)register("attachdetach", startAttachDetachController)register("persistentvolume-expander", startVolumeExpandController)register("clusterrole-aggregation", startClusterRoleAggregrationController)register("pvc-protection", startPVCProtectionController)register("pv-protection", startPVProtectionController)register("ttl-after-finished", startTTLAfterFinishedController)register("root-ca-cert-publisher", startRootCACertPublisher)register("ephemeral-volume", startEphemeralVolumeController)if utilfeature.DefaultFeatureGate.Enabled(genericfeatures.APIServerIdentity) &&utilfeature.DefaultFeatureGate.Enabled(genericfeatures.StorageVersionAPI) {register("storage-version-gc", startStorageVersionGCController)}if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.DynamicResourceAllocation) {register("resource-claim-controller", startResourceClaimController)}if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.LegacyServiceAccountTokenCleanUp) {register("legacy-service-account-token-cleaner", startLegacySATokenCleaner)}return controllers
}

这个NewControllerInitializers方法看起来就很重要，刚才的调用只用到了返回值的key，也就是所有controller的名字，稍后还会用到这个方法
现在展开上图黄色箭头所指的s.Config方法内部，如下所示，最终得到了数据结构kubecontrollerconfig.Config的实例，这里面有各controller的配置信息，以及client-go的客户端对象

func (s KubeControllerManagerOptions) Config(allControllers []string, disabledByDefaultControllers []string) (*kubecontrollerconfig.Config, error) {// 对每个controller的配置进行校验if err := s.Validate(allControllers, disabledByDefaultControllers); err != nil {return nil, err}// 如果有必要就创建自签证书if err := s.SecureServing.MaybeDefaultWithSelfSignedCerts("localhost", nil, []net.IP{netutils.ParseIPSloppy("127.0.0.1")}); err != nil {return nil, fmt.Errorf("error creating self-signed certificates: %v", err)}// 为创建client-go的客户端对象做准备：创建restclient.Configkubeconfig, err := clientcmd.BuildConfigFromFlags(s.Master, s.Generic.ClientConnection.Kubeconfig)if err != nil {return nil, err}kubeconfig.DisableCompression = truekubeconfig.ContentConfig.AcceptContentTypes = s.Generic.ClientConnection.AcceptContentTypeskubeconfig.ContentConfig.ContentType = s.Generic.ClientConnection.ContentTypekubeconfig.QPS = s.Generic.ClientConnection.QPSkubeconfig.Burst = int(s.Generic.ClientConnection.Burst)// 创建cliet-go库的客户端对象，有了它，就能对kubernetes的资源进行读写和监听了，非常重要client, err := clientset.NewForConfig(restclient.AddUserAgent(kubeconfig, KubeControllerManagerUserAgent))if err != nil {return nil, err}// 事件广播对象eventBroadcaster := record.NewBroadcaster()eventRecorder := eventBroadcaster.NewRecorder(clientgokubescheme.Scheme, v1.EventSource{Component: KubeControllerManagerUserAgent})// 创建配置对象c := &kubecontrollerconfig.Config{Client:           client,Kubeconfig:       kubeconfig,EventBroadcaster: eventBroadcaster,EventRecorder:    eventRecorder,}// 更新配置对象的信息（s的配置设置到c）if err := s.ApplyTo(c); err != nil {return nil, err}s.Metrics.Apply()return c, nil
}

至此，配置相关的算是过了一遍，接下来该看启动的代码了，也就是下图黄色箭头所示

启动逻辑分析

整个Run方法的内容很多，除了启动controller，还有安全处理，以及leader迁移等逻辑，这里咱们只聚焦重点：选主和controller启动
首先要看的是选主逻辑，注意下面的中文注释

// 如果无需选主（例如固定一个实例），这里直接调用run启动controller了，并提前返回
// 不过从启动命令的参数中有leader-elect=true，表示需要选主
if !c.ComponentConfig.Generic.LeaderElection.LeaderElect {run(ctx, saTokenControllerInitFunc, NewControllerInitializers)return nil
}
// 如果涉及到选主，也就是几个controller-manager进程，只有一个启动controller，就要准备一个独一无二的身份，这里是主机名+UUID
id, err := os.Hostname()
if err != nil {return err
}id = id + "_" + string(uuid.NewUUID())

身份确定后就是选主逻辑，注意是在一个新的协程中执行的选主，当前协程并未阻塞，如果选主成功，会执行run方法来启动controller

go leaderElectAndRun(ctx, c, id, electionChecker,c.ComponentConfig.Generic.LeaderElection.ResourceLock,c.ComponentConfig.Generic.LeaderElection.ResourceName,leaderelection.LeaderCallbacks{// OnStartedLeading会在选主成功后执行OnStartedLeading: func(ctx context.Context) {// NewControllerInitializers在前面分析过，会生成一个map，// key是controller名，value是controller的初始化方法initializersFunc := NewControllerInitializersif leaderMigrator != nil {// If leader migration is enabled, we should start only non-migrated controllers//  for the main lock.initializersFunc = createInitializersFunc(leaderMigrator.FilterFunc, leadermigration.ControllerNonMigrated)logger.Info("leader migration: starting main controllers.")}run(ctx, startSATokenController, initializersFunc)},// OnStoppedLeading会在失去leader身份时执行，klog.FlushAndExit内部会结束进程OnStoppedLeading: func() {logger.Error(nil, "leaderelection lost")klog.FlushAndExit(klog.ExitFlushTimeout, 1)},})

又见到了熟悉的NewControllerInitializers方法，这一次它被作为第三个参数传入run（如果leaderMigrator非空，就会调用createInitializersFunc，里面还是调用了NewControllerInitializers）
也就是说run方法通过第三个参数，知道了所有controller应该如何初始化
接下来就是最核心的启动逻辑了

启动所有controller

负责启动controller的是run方法，如下所示，逻辑上很简单：先准备一个通用的context，再得到所有的controller的初始化方法，逐一执行即可完成启动，这里面有两个重点，CreateControllerContext和StartControllers，后面会重点讲到

	run := func(ctx context.Context, startSATokenController InitFunc, initializersFunc ControllerInitializersFunc) {// 为启动controller准备context，里面存了多中公共对象，给各个controller用controllerContext, err := CreateControllerContext(logger, c, rootClientBuilder, clientBuilder, ctx.Done())if err != nil {logger.Error(err, "Error building controller context")klog.FlushAndExit(klog.ExitFlushTimeout, 1)}// initializersFunc即NewControllerInitializers方法，// 可以返回所有controller及其初始化方法controllerInitializers := initializersFunc(controllerContext.LoopMode)// StartControllers中会遍历controllerInitializers的返回值，对每个controller执行初始化和启动if err := StartControllers(ctx, controllerContext, startSATokenController, controllerInitializers, unsecuredMux, healthzHandler); err != nil {logger.Error(err, "Error starting controllers")klog.FlushAndExit(klog.ExitFlushTimeout, 1)}// 启动所有informercontrollerContext.InformerFactory.Start(stopCh)controllerContext.ObjectOrMetadataInformerFactory.Start(stopCh)close(controllerContext.InformersStarted)<-ctx.Done()}

主方法看过，算是对启动有了大致了解，再来看看细节，首先是创建通用上下文的CreateControllerContext，如下，生成了各类informer，client，都封装在context中，后面的controller可以使用

func CreateControllerContext(logger klog.Logger, s *config.CompletedConfig, rootClientBuilder, clientBuilder clientbuilder.ControllerClientBuilder, stop <-chan struct{}) (ControllerContext, error) {versionedClient := rootClientBuilder.ClientOrDie("shared-informers")sharedInformers := informers.NewSharedInformerFactory(versionedClient, ResyncPeriod(s)())metadataClient := metadata.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("metadata-informers"))metadataInformers := metadatainformer.NewSharedInformerFactory(metadataClient, ResyncPeriod(s)())// If apiserver is not running we should wait for some time and fail only then. This is particularly// important when we start apiserver and controller manager at the same time.if err := genericcontrollermanager.WaitForAPIServer(versionedClient, 10*time.Second); err != nil {return ControllerContext{}, fmt.Errorf("failed to wait for apiserver being healthy: %v", err)}// Use a discovery client capable of being refreshed.discoveryClient := rootClientBuilder.DiscoveryClientOrDie("controller-discovery")cachedClient := cacheddiscovery.NewMemCacheClient(discoveryClient)restMapper := restmapper.NewDeferredDiscoveryRESTMapper(cachedClient)go wait.Until(func() {restMapper.Reset()}, 30*time.Second, stop)availableResources, err := GetAvailableResources(rootClientBuilder)if err != nil {return ControllerContext{}, err}cloud, loopMode, err := createCloudProvider(logger, s.ComponentConfig.KubeCloudShared.CloudProvider.Name, s.ComponentConfig.KubeCloudShared.ExternalCloudVolumePlugin,s.ComponentConfig.KubeCloudShared.CloudProvider.CloudConfigFile, s.ComponentConfig.KubeCloudShared.AllowUntaggedCloud, sharedInformers)if err != nil {return ControllerContext{}, err}ctx := ControllerContext{ClientBuilder:                   clientBuilder,InformerFactory:                 sharedInformers,ObjectOrMetadataInformerFactory: informerfactory.NewInformerFactory(sharedInformers, metadataInformers),ComponentConfig:                 s.ComponentConfig,RESTMapper:                      restMapper,AvailableResources:              availableResources,Cloud:                           cloud,LoopMode:                        loopMode,InformersStarted:                make(chan struct{}),ResyncPeriod:                    ResyncPeriod(s),ControllerManagerMetrics:        controllersmetrics.NewControllerManagerMetrics("kube-controller-manager"),}controllersmetrics.Register()return ctx, nil
}

上述代码中，有一小段值得注意，注释也值得一看，就是同步等待的逻辑，这里面又会引出一个重要的知识点：用channel实现轮询同步等待，学习并发的读者可以输入研究，一定受益匪浅

	// If apiserver is not running we should wait for some time and fail only then. This is particularly// important when we start apiserver and controller manager at the same time.if err := genericcontrollermanager.WaitForAPIServer(versionedClient, 10*time.Second); err != nil {return ControllerContext{}, fmt.Errorf("failed to wait for apiserver being healthy: %v", err)}

关于同步等待并非本文主题，就不多说了，上面的WaitForAPIServer经过层层展开到了waitForWithContext，简单的分析一下

// 第二个入参wait，返回值是个channel，wait方法里面有定时器，每一秒向返回的channel写数据，超时了就关闭
func waitForWithContext(ctx context.Context, wait waitWithContextFunc, fn ConditionWithContextFunc) error {waitCtx, cancel := context.WithCancel(context.Background())defer cancel()c := wait(waitCtx)for {select {// 由于c在wait方法中每秒被写一次，所以下面这个case每秒执行一次case _, open := <-c:// 这里的fn在外面传入的是远程请求api-serverok, err := runConditionWithCrashProtectionWithContext(ctx, fn)if err != nil {return err}// 这表示api-server能正常响应，也就是在超时时间内拿到了想要的结果if ok {return nil}// 如果c被关闭，就证明已经超时了if !open {return ErrWaitTimeout}case <-ctx.Done():// returning ctx.Err() will break backward compatibility, use new PollUntilContext*// methods insteadreturn ErrWaitTimeout}}
}

StartControllers分析

前面说到run方法中最重要的是CreateControllerContext和StartControllers，现在该看这个StartControllers了，注意它的第四个参数controllers就是咱们的老朋友NewControllerInitializers，这里面有所有controller的初始化方法
StartControllers是本篇最重要的内容，此方法代码也挺多，咱们只看最关键的，也就是遍历controller的初始化方法集合的处理逻辑，似乎很容易，因为咱们对controllers已经了如指掌，下面这个循环其实就是将NewControllerInitializers返回的所有controller初始化方法都执行一遍（可能会条件过滤掉一些）
TODO：下面的代码应该再加上一些注释

	// 遍历初始化方法集合for controllerName, initFn := range controllers {if !controllerCtx.IsControllerEnabled(controllerName) {logger.Info("Warning: controller is disabled", "controller", controllerName)continue}time.Sleep(wait.Jitter(controllerCtx.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))logger.V(1).Info("Starting controller", "controller", controllerName)// 执行initFn，就完成了controller的初始化和启动，initFn具体是什么呢？那要看NewControllerInitializers方法中的内容ctrl, started, err := initFn(klog.NewContext(ctx, klog.LoggerWithName(logger, controllerName)), controllerCtx)if err != nil {logger.Error(err, "Error starting controller", "controller", controllerName)return err}// 启动失败则继续下一个if !started {logger.Info("Warning: skipping controller", "controller", controllerName)continue}check := controllerhealthz.NamedPingChecker(controllerName)if ctrl != nil {// check if the controller supports and requests a debugHandler// and it needs the unsecuredMux to mount the handler onto.if debuggable, ok := ctrl.(controller.Debuggable); ok && unsecuredMux != nil {if debugHandler := debuggable.DebuggingHandler(); debugHandler != nil {basePath := "/debug/controllers/" + controllerNameunsecuredMux.UnlistedHandle(basePath, http.StripPrefix(basePath, debugHandler))unsecuredMux.UnlistedHandlePrefix(basePath+"/", http.StripPrefix(basePath, debugHandler))}}// 如果当前controller支持健康检查，就放入check切片，后面统一注册if healthCheckable, ok := ctrl.(controller.HealthCheckable); ok {if realCheck := healthCheckable.HealthChecker(); realCheck != nil {check = controllerhealthz.NamedHealthChecker(controllerName, realCheck)}}}controllerChecks = append(controllerChecks, check)logger.Info("Started controller", "controller", controllerName)}// 注册健康检查，类似gin注册路由，每个path对应一个controller的健康检查路径，这样外部就能通过这个path来确定controller是否健康healthzHandler.AddHealthChecker(controllerChecks...)