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

欢迎访问我的GitHub

本篇概览

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

1. 查看启动命令：了解真实kubernetes环境下controll-manager的启动命令，知道会传入那些参数
2. 入口：搞清楚代码从哪里开始看，然后分析如何处理输入的参数
3. 初始化配置：自身有大量默认参数，还有启动时输入的大量参数，使用这些信息进行初始化配置
4. 深入理解启动逻辑：先学习启动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

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

1. 没有用到子命令
2. 有多个标志（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的启动代码，尽管已经添加了详细的注释，仍有两处重点需要展开说明：

1. s.Config(KnownControllers(), ControllersDisabledByDefault.List()) ：这里面涉及到一些重要的初始化逻辑和数据结构
2. 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.Config
	kubeconfig, err := clientcmd.BuildConfigFromFlags(s.Master, s.Generic.ClientConnection.Kubeconfig)
	if err != nil {
		return nil, err
	}
	kubeconfig.DisableCompression = true
	kubeconfig.ContentConfig.AcceptContentTypes = s.Generic.ClientConnection.AcceptContentTypes
	kubeconfig.ContentConfig.ContentType = s.Generic.ClientConnection.ContentType
	kubeconfig.QPS = s.Generic.ClientConnection.QPS
	kubeconfig.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 := NewControllerInitializers
			if 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)
		}
		
		// 启动所有informer
		controllerContext.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-server
			ok, 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 instead
			return 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/" + controllerName
					unsecuredMux.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...)

• 至此，controller-manager的源码分析就完成了，可见主要工作是准备配置，确保每个controller完成启动，接下来的文章，咱们再深入一个典型的controller，了解kubernetes自己的controller是如何启动的

你不孤单，欣宸原创一路相伴

1. Java系列
2. Spring系列
3. Docker系列
4. kubernetes系列
5. 数据库+中间件系列
6. DevOps系列