HAMi 调度器架构深度解析

// cmd/scheduler/main.go - start() 函数
func start() error {
    nodelock.NodeLockTimeout = config.NodeLockTimeout  // 设置节点锁超时
    client.InitGlobalClient(...)                        // 初始化 K8s 客户端
    config.InitDevices()                                // 注册设备插件
    sher = scheduler.NewScheduler()                     // 创建调度器实例
    go sher.RegisterFromNodeAnnotations()               // 启动设备注册循环（goroutine）
    sher.Start()                                        // 启动 Informer 缓存同步
    go initMetrics(config.MetricsBindAddress)            // 启动 Prometheus 指标服务
    // 注册 HTTP 路由并启动 HTTP 服务
    router.POST("/filter", routes.PredicateRoute(sher))
    router.POST("/bind", routes.Bind(sher))
    router.POST("/webhook", routes.WebHookRoute())
    router.GET("/healthz", routes.HealthzRoute())
    router.GET("/readyz", routes.ReadyzRoute(sher))
    http.ListenAndServe(config.HTTPBind, router)
}

type Scheduler struct {
    *nodeManager                           // 嵌入节点管理器，维护节点及其设备信息
    podManager    *device.PodManager       // Pod 管理器，追踪已分配设备的 Pod
    quotaManager  *device.QuotaManager     // 配额管理器，管理 ResourceQuota 中的设备配额
    leaderManager leaderelection.LeaderManager // Leader 选举管理器

    stopCh       chan struct{}             // 停止信号通道
    nodeNotify   chan struct{}             // 节点变更通知通道（带缓冲，容量 1）
    leaderNotify chan struct{}             // Leader 选举通知通道（带缓冲，容量 1）

    kubeClient  kubernetes.Interface       // Kubernetes API 客户端
    podLister   listerscorev1.PodLister    // Pod Informer Lister
    nodeLister  listerscorev1.NodeLister   // Node Informer Lister
    quotaLister listerscorev1.ResourceQuotaLister // ResourceQuota Informer Lister
    leaseLister coordinationv1.LeaseLister // Lease Informer Lister（Leader 选举用）

    cachedstatus   map[string]*NodeUsage   // Filter 请求中返回的节点状态缓存
    overviewstatus map[string]*NodeUsage   // 全局节点状态概览（用于指标监控）
    eventRecorder  record.EventRecorder    // Kubernetes Event 记录器
    started        uint32                  // 原子标志位，0=未启动，1=已启动

    lock   sync.RWMutex                   // 读写锁，保护 synced 状态
    synced bool                           // 缓存是否已同步完成
}

func (s *Scheduler) Start() error {
    informerFactory := informers.NewSharedInformerFactoryWithOptions(s.kubeClient, defaultResync)

    // Pod Informer - 追踪 Pod 的设备分配状态
    informerFactory.Core().V1().Pods().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    s.onAddPod,
        UpdateFunc: s.onUpdatePod,
        DeleteFunc: s.onDelPod,
    })

    // Node Informer - 感知节点增删
    informerFactory.Core().V1().Nodes().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    func(_ any) { s.doNodeNotify() },
        DeleteFunc: s.onDelNode,
    })

    // ResourceQuota Informer - 追踪配额变化
    informerFactory.Core().V1().ResourceQuotas().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
        AddFunc:    s.onAddQuota,
        UpdateFunc: s.onUpdateQuota,
        DeleteFunc: s.onDelQuota,
    })

    informerFactory.Start(s.stopCh)
    informerFactory.WaitForCacheSync(s.stopCh)
    // ...
}

func (s *Scheduler) RegisterFromNodeAnnotations() {
    ticker := time.NewTicker(time.Second * 15)
    for {
        select {
        case <-s.nodeNotify:    // 节点增删事件触发
        case <-s.leaderNotify:  // Leader 选举变更触发
        case <-ticker.C:        // 每 15 秒定时触发
        case <-s.stopCh:        // 停止信号
            return
        }
        if atomic.LoadUint32(&s.started) == 0 {
            continue  // Informer 未就绪，跳过
        }
        s.register(labelSelector, printedLog)
    }
}

func (m *leaderManager) IsLeader() bool {
    m.leaseLock.RLock()
    defer m.leaseLock.RUnlock()
    return m.isHolderOf(m.observedLease) && m.isLeaseValid(time.Now())
}

func ReadyzRoute(s *scheduler.Scheduler) httprouter.Handle {
    return func(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
        if !s.GetLeaderManager().IsLeader() {
            w.WriteHeader(http.StatusServiceUnavailable) // 503
            return
        }
        w.WriteHeader(http.StatusOK) // 200
    }
}