调度周期 runOnce() 完整追踪

// pkg/scheduler/scheduler.go:107
func (pc *Scheduler) runOnce() {
    klog.V(4).Infof("Start scheduling ...")
    scheduleStartTime := time.Now()
    defer klog.V(4).Infof("End scheduling ...")

    pc.mutex.Lock()
    actions := pc.actions
    plugins := pc.plugins
    configurations := pc.configurations
    pc.mutex.Unlock()

    // Load ConfigMap to check which action is enabled.
    conf.EnabledActionMap = make(map[string]bool)
    for _, action := range actions {
        conf.EnabledActionMap[action.Name()] = true
    }
    // ...
}

// pkg/scheduler/actions/allocate/allocate.go:152
if conf.EnabledActionMap["enqueue"] {
    klog.V(4).Infof("Job <%s/%s> Queue <%s> skip allocate, reason: job status is pending.",
        job.Namespace, job.Name, job.Queue)
    continue
}

// pkg/scheduler/framework/framework.go:34
func OpenSession(cache cache.Cache, tiers []conf.Tier, configurations []conf.Configuration) *Session {
    ssn := openSession(cache)
    ssn.Tiers = tiers
    ssn.Configurations = configurations
    ssn.NodeMap = GenerateNodeMapAndSlice(ssn.Nodes)
    ssn.PodLister = NewPodLister(ssn)

    for _, tier := range tiers {
        for _, plugin := range tier.Plugins {
            if pb, found := GetPluginBuilder(plugin.Name); !found {
                klog.Errorf("Failed to get plugin %s.", plugin.Name)
            } else {
                plugin := pb(plugin.Arguments)
                ssn.plugins[plugin.Name()] = plugin
                onSessionOpenStart := time.Now()
                plugin.OnSessionOpen(ssn)
                metrics.UpdatePluginDuration(plugin.Name(), metrics.OnSessionOpen,
                    metrics.Duration(onSessionOpenStart))
            }
        }
    }

    ssn.InitCycleState()
    return ssn
}

// pkg/scheduler/cache/cache.go:1423
func (sc *SchedulerCache) Snapshot() *schedulingapi.ClusterInfo {
    sc.Mutex.Lock()
    defer sc.Mutex.Unlock()

    snapshot := &schedulingapi.ClusterInfo{
        Nodes:   make(map[string]*schedulingapi.NodeInfo),
        Jobs:    make(map[schedulingapi.JobID]*schedulingapi.JobInfo),
        Queues:  make(map[schedulingapi.QueueID]*schedulingapi.QueueInfo),
        // ... 其他字段初始化 ...
    }

    // 1. 复制 NodeList
    copy(snapshot.NodeList, sc.NodeList)

    // 2. 复制 Nodes（仅 Ready 节点）
    for _, value := range sc.Nodes {
        if !value.Ready() { continue }
        snapshot.Nodes[value.Name] = value.Clone()
    }

    // 3. 快照 HyperNodes（单独加锁）
    sc.HyperNodesInfo.Lock()
    snapshot.HyperNodes = sc.HyperNodesInfo.HyperNodes()
    // ...
    sc.HyperNodesInfo.Unlock()

    // 4. 复制 Queues
    for _, value := range sc.Queues {
        snapshot.Queues[value.UID] = value.Clone()
    }

    // 5. 并行克隆 Jobs
    var cloneJobLock sync.Mutex
    var wg sync.WaitGroup
    cloneJob := func(value *schedulingapi.JobInfo) {
        defer wg.Done()
        clonedJob := value.Clone()
        cloneJobLock.Lock()
        snapshot.Jobs[value.UID] = clonedJob
        cloneJobLock.Unlock()
    }
    for _, value := range sc.Jobs {
        if value.PodGroup == nil { continue }
        if _, found := snapshot.Queues[value.Queue]; !found { continue }
        wg.Add(1)
        go cloneJob(value)
    }
    wg.Wait()

    return snapshot
}

// pkg/scheduler/scheduler.go:130
for _, action := range actions {
    actionStartTime := time.Now()
    action.Execute(ssn)
    metrics.UpdateActionDuration(action.Name(), metrics.Duration(actionStartTime))
}

metrics.UpdateActionDuration(action.Name(), metrics.Duration(actionStartTime))

// pkg/scheduler/framework/framework.go:61
func CloseSession(ssn *Session) {
    for _, plugin := range ssn.plugins {
        onSessionCloseStart := time.Now()
        plugin.OnSessionClose(ssn)
        metrics.UpdatePluginDuration(plugin.Name(), metrics.OnSessionClose,
            metrics.Duration(onSessionCloseStart))
    }

    closeSession(ssn)
    ssn.cache.OnSessionClose()
}

// pkg/scheduler/framework/session.go:597
func closeSession(ssn *Session) {
    ju := NewJobUpdater(ssn)
    ju.UpdateAll()

    updateQueueStatus(ssn)

    ssn.Jobs = nil
    ssn.Nodes = nil
    ssn.RevocableNodes = nil
    ssn.plugins = nil
    ssn.eventHandlers = nil
    ssn.jobOrderFns = nil
    ssn.queueOrderFns = nil
    ssn.clusterOrderFns = nil
    ssn.NodeList = nil
    ssn.TotalResource = nil

    ssn.cache.OnSessionClose()

    klog.V(3).Infof("Close Session %v", ssn.UID)
}

// 伪代码 - updateQueueStatus 核心逻辑
func updateQueueStatus(ssn *Session) {
    for _, queue := range ssn.Queues {
        allocated := api.EmptyResource()
        for _, job := range ssn.Jobs {
            if job.Queue != queue.UID { continue }
            for status, tasks := range job.TaskStatusIndex {
                if api.AllocatedStatus(status) {
                    for _, task := range tasks {
                        allocated.Add(task.Resreq)
                    }
                }
            }
        }
        // 更新 Queue 的 allocated 字段
        ssn.cache.UpdateQueueStatus(queue, allocated)
    }
}