Controller Manager 架构概览

// ControllerContext 包含控制器运行所需的所有上下文
// cmd/kube-controller-manager/app/controllermanager.go

type ControllerContext struct {
    // 客户端配置
    ClientBuilder    clientbuilder.ControllerClientBuilder

    // Informer 工厂
    InformerFactory  informers.SharedInformerFactory

    // 对象元数据 Informer
    ObjectOrMetadataInformerFactory controller.InformerFactory

    // 组件配置
    ComponentConfig  kubectrlmgrconfig.KubeControllerManagerConfiguration

    // 事件记录器
    EventRecorder    record.EventRecorder

    // 云提供商
    Cloud            cloudprovider.Interface

    // 停止通道
    Stop             <-chan struct{}

    // Informer 启动状态
    InformersStarted chan struct{}

    // 需要的集群角色
    AvailableResources map[schema.GroupVersionResource]bool
}

// 控制器初始化函数类型
type InitFunc func(ctx context.Context, controllerCtx ControllerContext) (controller.Interface, bool, error)

// main 入口
// cmd/kube-controller-manager/controller-manager.go

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

// NewControllerManagerCommand 创建命令
// cmd/kube-controller-manager/app/controllermanager.go

func NewControllerManagerCommand() *cobra.Command {
    s, err := options.NewKubeControllerManagerOptions()
    if err != nil {
        klog.Fatalf("unable to initialize command options: %v", err)
    }

    cmd := &cobra.Command{
        Use: "kube-controller-manager",
        RunE: func(cmd *cobra.Command, args []string) error {
            // 完成配置
            c, err := s.Config(KnownControllers(), ControllersDisabledByDefault.List())
            if err != nil {
                return err
            }

            // 运行
            return Run(c.Complete(), wait.NeverStop)
        },
    }

    return cmd
}

// Run 启动 Controller Manager
// cmd/kube-controller-manager/app/controllermanager.go

func Run(c *config.CompletedConfig, stopCh <-chan struct{}) error {
    // 1. 创建控制器上下文
    controllerContext, err := CreateControllerContext(c, rootClientBuilder, clientBuilder, stopCh)
    if err != nil {
        return err
    }

    // 2. 设置健康检查
    var checks []healthz.HealthChecker
    var electionChecker *leaderelection.HealthzAdaptor

    // 3. Leader Election
    if c.ComponentConfig.Generic.LeaderElection.LeaderElect {
        electionChecker = leaderelection.NewLeaderHealthzAdaptor(time.Second * 20)
        checks = append(checks, electionChecker)

        go leaderelection.RunOrDie(context.TODO(), leaderelection.LeaderElectionConfig{
            Lock:          lock,
            LeaseDuration: c.ComponentConfig.Generic.LeaderElection.LeaseDuration.Duration,
            RenewDeadline: c.ComponentConfig.Generic.LeaderElection.RenewDeadline.Duration,
            RetryPeriod:   c.ComponentConfig.Generic.LeaderElection.RetryPeriod.Duration,
            Callbacks: leaderelection.LeaderCallbacks{
                OnStartedLeading: func(ctx context.Context) {
                    // 获得 Leader，启动控制器
                    run(ctx, controllerContext)
                },
                OnStoppedLeading: func() {
                    klog.Fatalf("leaderelection lost")
                },
            },
        })
    } else {
        // 不使用选举，直接启动
        run(context.TODO(), controllerContext)
    }

    <-stopCh
    return nil
}

// StartControllers 启动所有控制器
// cmd/kube-controller-manager/app/controllermanager.go

func StartControllers(ctx context.Context, controllerCtx ControllerContext,
    controllers map[string]InitFunc, unsecuredMux *mux.PathRecorderMux) error {

    // 启动所有控制器
    for controllerName, initFn := range controllers {
        if !controllerCtx.IsControllerEnabled(controllerName) {
            klog.Warningf("%q is disabled", controllerName)
            continue
        }

        time.Sleep(wait.Jitter(controllerCtx.ComponentConfig.Generic.ControllerStartInterval.Duration, ControllerStartJitter))

        klog.V(1).Infof("Starting %q", controllerName)
        ctrl, started, err := initFn(ctx, controllerCtx)
        if err != nil {
            klog.Errorf("Error starting %q: %v", controllerName, err)
            return err
        }
        if !started {
            klog.Warningf("Skipping %q", controllerName)
            continue
        }

        check := controllerhealthz.NamedPingChecker(controllerName)
        if ctrl != nil {
            if healthCheckable, ok := ctrl.(controller.Healthz); ok {
                check = controllerhealthz.NamedHealthChecker(controllerName, healthCheckable)
            }
        }
        klog.Infof("Started %q", controllerName)
    }

    return nil
}

// 控制器注册表
// cmd/kube-controller-manager/app/controllermanager.go

func NewControllerInitializers(loopMode ControllerLoopMode) map[string]InitFunc {
    controllers := map[string]InitFunc{}

    // 工作负载控制器
    controllers["deployment"] = startDeploymentController
    controllers["replicaset"] = startReplicaSetController
    controllers["statefulset"] = startStatefulSetController
    controllers["daemonset"] = startDaemonSetController
    controllers["job"] = startJobController
    controllers["cronjob"] = startCronJobController

    // Pod 相关控制器
    controllers["replicationcontroller"] = startReplicationController
    controllers["podgc"] = startPodGCController

    // 服务相关控制器
    controllers["endpoint"] = startEndpointController
    controllers["endpointslice"] = startEndpointSliceController
    controllers["endpointslicemirroring"] = startEndpointSliceMirroringController

    // 节点相关控制器
    controllers["nodelifecycle"] = startNodeLifecycleController
    controllers["nodeipam"] = startNodeIpamController
    controllers["ttl"] = startTTLController

    // 存储相关控制器
    controllers["persistentvolume-binder"] = startPersistentVolumeBinderController
    controllers["attachdetach"] = startAttachDetachController
    controllers["persistentvolume-expander"] = startVolumeExpandController
    controllers["pvc-protection"] = startPVCProtectionController
    controllers["pv-protection"] = startPVProtectionController

    // 命名空间控制器
    controllers["namespace"] = startNamespaceController

    // ServiceAccount 控制器
    controllers["serviceaccount"] = startServiceAccountController
    controllers["serviceaccount-token"] = startServiceAccountTokenController

    // 资源配额控制器
    controllers["resourcequota"] = startResourceQuotaController

    // 垃圾回收控制器
    controllers["garbagecollector"] = startGarbageCollectorController

    // TTL 控制器
    controllers["ttl-after-finished"] = startTTLAfterFinishedController

    // 证书控制器
    controllers["csrapproving"] = startCSRApprovingController
    controllers["csrcleaner"] = startCSRCleanerController
    controllers["csrsigning"] = startCSRSigningController

    // 其他控制器...

    return controllers
}

// Deployment 控制器初始化
// cmd/kube-controller-manager/app/apps.go

func startDeploymentController(ctx context.Context, controllerContext ControllerContext) (controller.Interface, bool, error) {
    dc, err := deployment.NewDeploymentController(
        controllerContext.InformerFactory.Apps().V1().Deployments(),
        controllerContext.InformerFactory.Apps().V1().ReplicaSets(),
        controllerContext.InformerFactory.Core().V1().Pods(),
        controllerContext.ClientBuilder.ClientOrDie("deployment-controller"),
    )
    if err != nil {
        return nil, true, fmt.Errorf("error creating Deployment controller: %v", err)
    }

    go dc.Run(ctx, int(controllerContext.ComponentConfig.DeploymentController.ConcurrentDeploymentSyncs))

    return nil, true, nil
}

// SharedInformerFactory 创建
// cmd/kube-controller-manager/app/controllermanager.go

func CreateControllerContext(c *config.CompletedConfig, ...) (ControllerContext, error) {
    // 创建 SharedInformerFactory
    sharedInformers := informers.NewSharedInformerFactory(
        versionedClient,
        ResyncPeriod(c)(),
    )

    // 创建 Metadata Informer (用于 GC)
    metadataInformers := metadatainformers.NewSharedInformerFactory(
        metadataClient,
        ResyncPeriod(c)(),
    )

    return ControllerContext{
        InformerFactory:  sharedInformers,
        ObjectOrMetadataInformerFactory: controller.NewInformerFactory(
            sharedInformers,
            metadataInformers,
        ),
        ...
    }, nil
}

// 启动 Informers
// cmd/kube-controller-manager/app/controllermanager.go

func (ctx ControllerContext) StartInformers(stopCh <-chan struct{}) {
    ctx.InformerFactory.Start(stopCh)
    ctx.ObjectOrMetadataInformerFactory.Start(stopCh)
}

// 等待缓存同步
func (ctx ControllerContext) WaitForInformersSync() {
    ctx.InformerFactory.WaitForCacheSync(ctx.Stop)
}

// Leader Election 配置
// staging/src/k8s.io/component-base/config/types.go

type LeaderElectionConfiguration struct {
    // 是否启用选举
    LeaderElect bool

    // 锁的持续时间
    LeaseDuration metav1.Duration

    // 续约截止时间
    RenewDeadline metav1.Duration

    // 重试间隔
    RetryPeriod metav1.Duration

    // 资源锁类型
    ResourceLock string

    // 资源名称
    ResourceName string

    // 资源命名空间
    ResourceNamespace string
}

// 默认值
var defaultLeaderElectionConfiguration = LeaderElectionConfiguration{
    LeaderElect:       true,
    LeaseDuration:     metav1.Duration{Duration: 15 * time.Second},
    RenewDeadline:     metav1.Duration{Duration: 10 * time.Second},
    RetryPeriod:       metav1.Duration{Duration: 2 * time.Second},
    ResourceLock:      "leases",
    ResourceName:      "kube-controller-manager",
    ResourceNamespace: "kube-system",
}

// 运行选举
// staging/src/k8s.io/client-go/tools/leaderelection/leaderelection.go

func RunOrDie(ctx context.Context, lec LeaderElectionConfig) {
    le, err := NewLeaderElector(lec)
    if err != nil {
        panic(err)
    }
    le.Run(ctx)
}

func (le *LeaderElector) Run(ctx context.Context) {
    defer func() {
        le.config.Callbacks.OnStoppedLeading()
    }()

    // 尝试获取锁
    if !le.acquire(ctx) {
        return
    }

    // 获取锁成功，启动回调
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()
    go le.config.Callbacks.OnStartedLeading(ctx)

    // 续约循环
    le.renew(ctx)
}

# Controller Manager 配置示例
apiVersion: kubecontroller.config.k8s.io/v1alpha1
kind: KubeControllerManagerConfiguration
generic:
  # 并发同步数
  controllers:
    - "*"  # 启用所有控制器

  # 或禁用特定控制器
  # controllers:
  # - "*"
  # - "-bootstrapsigner"

  # Leader Election
  leaderElection:
    leaderElect: true
    leaseDuration: 15s
    renewDeadline: 10s
    retryPeriod: 2s
    resourceLock: leases
    resourceName: kube-controller-manager
    resourceNamespace: kube-system

# 各控制器配置
deploymentController:
  concurrentDeploymentSyncs: 5

replicaSetController:
  concurrentRSSyncs: 5

endpointController:
  concurrentEndpointSyncs: 5

serviceAccountController:
  concurrentSATokenSyncs: 5
  serviceAccountKeyFile: /etc/kubernetes/pki/sa.key

garbageCollectorController:
  concurrentGCSyncs: 20
  enableGarbageCollector: true

nodeLifecycleController:
  podEvictionTimeout: 5m0s
  nodeMonitorGracePeriod: 40s
  nodeStartupGracePeriod: 1m0s