Controller Manager 与 Controller 框架深度解析

// 通过 blank import 强制触发 init() 注册
_ "volcano.sh/volcano/pkg/controllers/colocationconfig"
_ "volcano.sh/volcano/pkg/controllers/cronjob"
_ "volcano.sh/volcano/pkg/controllers/garbagecollector"
_ "volcano.sh/volcano/pkg/controllers/hypernode"
_ "volcano.sh/volcano/pkg/controllers/job"
_ "volcano.sh/volcano/pkg/controllers/jobflow"
_ "volcano.sh/volcano/pkg/controllers/jobtemplate"
_ "volcano.sh/volcano/pkg/controllers/podgroup"
_ "volcano.sh/volcano/pkg/controllers/queue"
_ "volcano.sh/volcano/pkg/controllers/sharding"

var knownControllers = func() []string {
    controllerNames := []string{}
    fn := func(controller framework.Controller) {
        controllerNames = append(controllerNames, controller.Name())
    }
    framework.ForeachController(fn)
    sort.Strings(controllerNames)
    return controllerNames
}
s.AddFlags(fs, knownControllers())

type Controller interface {
    Name() string
    Initialize(opt *ControllerOption) error
    Run(stopCh <-chan struct{})
}

type ControllerOption struct {
    KubeClient              kubernetes.Interface          // Kubernetes 原生客户端
    VolcanoClient           vcclientset.Interface          // Volcano CRD 客户端
    SharedInformerFactory   informers.SharedInformerFactory // Kube 资源 Informer 工厂
    VCSharedInformerFactory vcinformer.SharedInformerFactory // VC 资源 Informer 工厂
    SchedulerNames          []string                       // 关联的调度器名称列表
    WorkerNum               uint32                         // Job Controller worker 数
    CronJobWorkerNum        uint32                         // CronJob worker 数
    MaxRequeueNum           int                            // 最大重试次数
    InheritOwnerAnnotations bool                           // 是否继承 Owner 注解
    WorkerThreadsForPG      uint32                         // PodGroup worker 数
    WorkerThreadsForQueue   uint32                         // Queue worker 数
    WorkerThreadsForGC      uint32                         // GC worker 数
    Config                  *rest.Config                   // REST 客户端配置
}

var controllers = map[string]Controller{}

func RegisterController(ctrl Controller) error {
    if ctrl == nil {
        return fmt.Errorf("controller is nil")
    }
    if _, found := controllers[ctrl.Name()]; found {
        return fmt.Errorf("duplicated controller")
    }
    klog.V(3).Infof("Controller <%s> is registered.", ctrl.Name())
    controllers[ctrl.Name()] = ctrl
    return nil
}

func ForeachController(fn func(controller Controller)) {
    for _, ctrl := range controllers {
        fn(ctrl)
    }
}

func Run(opt *options.ServerOption) error {
    // 1. 构建 REST 配置
    config, err := kube.BuildConfig(opt.KubeClientOptions)

    // 2. 启动健康检查 (可选)
    if opt.EnableHealthz {
        helpers.StartHealthz(opt.HealthzBindAddress, ...)
    }

    // 3. 启动 Metrics 服务 (可选)
    if opt.EnableMetrics {
        go func() { /* Prometheus HTTP Server on :8081 */ }()
    }

    // 4. 构建 startControllers 闭包
    run := startControllers(config, opt)

    // 5. Leader Election (或直接运行)
    if !opt.LeaderElection.LeaderElect {
        run(ctx)
        return ...
    }
    leaderelection.RunOrDie(ctx, leaderelection.LeaderElectionConfig{
        Callbacks: leaderelection.LeaderCallbacks{
            OnStartedLeading: run,
            OnStoppedLeading: func() { klog.Fatalf("leaderelection lost") },
        },
    })
}

func startControllers(config *rest.Config, opt *options.ServerOption) func(ctx context.Context) {
    controllerOpt := &framework.ControllerOption{}
    // 填充共享配置...
    controllerOpt.KubeClient = kubeclientset.NewForConfigOrDie(config)
    controllerOpt.VolcanoClient = vcclientset.NewForConfigOrDie(config)
    controllerOpt.SharedInformerFactory = informers.NewSharedInformerFactory(...)
    controllerOpt.VCSharedInformerFactory = informerfactory.NewSharedInformerFactory(...)
    // ...

    return func(ctx context.Context) {
        framework.ForeachController(func(c framework.Controller) {
            if !isControllerEnabled(c.Name(), opt.Controllers) {
                return  // 跳过未启用的 Controller
            }
            if err := c.Initialize(controllerOpt); err != nil {
                return  // 初始化失败则跳过
            }
            go c.Run(ctx.Done())  // 在独立 goroutine 中运行
        })
        <-ctx.Done()
    }
}

func isControllerEnabled(name string, controllers []string) bool {
    // 空列表 -> 全部启用
    // "+name" 或 "name" -> 显式启用
    // "-name" -> 显式禁用
    // "*" -> 通配符启用所有
}

func NewState(jobInfo *apis.JobInfo) State {
    job := jobInfo.Job
    switch job.Status.State.Phase {
    case vcbatch.Pending:
        return &pendingState{job: jobInfo}
    case vcbatch.Running:
        return &runningState{job: jobInfo}
    case vcbatch.Restarting:
        return &restartingState{job: jobInfo}
    case vcbatch.Terminated, vcbatch.Completed, vcbatch.Failed:
        return &finishedState{job: jobInfo}
    case vcbatch.Terminating:
        return &terminatingState{job: jobInfo}
    case vcbatch.Aborting:
        return &abortingState{job: jobInfo}
    case vcbatch.Aborted:
        return &abortedState{job: jobInfo}
    case vcbatch.Completing:
        return &completingState{job: jobInfo}
    }
    return &pendingState{job: jobInfo}  // 默认为 Pending
}

// 创建多个 worker queue
cc.queueList = make([]workqueue.TypedRateLimitingInterface[any], workers)
for i = 0; i < workers; i++ {
    cc.queueList[i] = workqueue.NewTypedRateLimitingQueue(...)
}

// 通过 FNV hash 将 Job 分配到固定的 worker
func (cc *jobcontroller) getWorkerQueue(key string) workqueue.TypedRateLimitingInterface[any] {
    val := cc.genHash(key)
    queue := cc.queueList[val%cc.workers]
    return queue
}

func (cc *jobcontroller) genHash(key string) uint32 {
    hashVal := fnv.New32()
    hashVal.Write([]byte(key))
    return hashVal.Sum32()
}

// Kubernetes 原生资源的 Informer 工厂
controllerOpt.SharedInformerFactory = informers.NewSharedInformerFactory(kubeClient, 0)

// Volcano CRD 资源的 Informer 工厂
controllerOpt.VCSharedInformerFactory = informerfactory.NewSharedInformerFactory(vcClient, 0)

eventBroadcaster := record.NewBroadcaster()
eventBroadcaster.StartLogging(klog.Infof)
eventBroadcaster.StartRecordingToSink(&corev1.EventSinkImpl{
    Interface: kubeClient.CoreV1().Events(""),
})
recorder := eventBroadcaster.NewRecorder(scheme, v1.EventSource{
    Component: "vc-controller-manager",
})

queue := workqueue.NewTypedRateLimitingQueue(
    workqueue.DefaultTypedControllerRateLimiter[string](),
)