Session 框架

type Session struct {
    UID types.UID

    // Kubernetes 客户端
    kubeClient      kubernetes.Interface
    vcClient        vcclient.Interface
    recorder        record.EventRecorder
    cache           cache.Cache
    restConfig      *rest.Config
    informerFactory informers.SharedInformerFactory

    // 资源统计
    TotalResource  *api.Resource          // 集群总资源
    TotalGuarantee *api.Resource          // 队列总 Guarantee
    TotalDeserved  *api.Resource          // 队列总 Deserved
    PodGroupOldState *api.PodGroupOldState // PodGroup 历史状态（不可修改）
    DirtyJobs sets.Set[api.JobID]         // 需要刷新到 Cache 的 Job

    // 快照数据
    Jobs           map[api.JobID]*api.JobInfo
    Nodes          map[string]*api.NodeInfo
    CSINodesStatus map[string]*api.CSINodeStatusInfo
    RevocableNodes map[string]*api.NodeInfo
    Queues         map[api.QueueID]*api.QueueInfo
    NamespaceInfo  map[api.NamespaceName]*api.NamespaceInfo

    // K8s 兼容层
    NodeMap   map[string]fwk.NodeInfo    // K8s Framework 兼容的 NodeInfo
    PodLister *PodLister                  // Pod 列表查询器

    // 配置
    Tiers          []conf.Tier
    Configurations []conf.Configuration
    NodeList       []*api.NodeInfo

    // HyperNode 拓扑
    HyperNodes           api.HyperNodeInfoMap
    HyperNodeTierNameMap api.HyperNodeTierNameMap
    HyperNodesSetByTier  map[int]sets.Set[string]
    HyperNodesTiers      []int
    RealNodesList        map[string][]*api.NodeInfo
    RealNodesSet         map[string]sets.Set[string]
    HyperNodesReadyToSchedule bool

    // Plugin 管理
    plugins       map[string]Plugin
    eventHandlers []*EventHandler

    // 扩展点函数映射（30+ 扩展点）
    jobOrderFns, queueOrderFns, taskOrderFns, clusterOrderFns       ...CompareFn
    victimQueueOrderFns                                              ...VictimCompareFn
    predicateFns, prePredicateFns                                    ...PredicateFn
    bestNodeFns, nodeOrderFns, batchNodeOrderFns                     ...NodeOrderFn
    nodeMapFns, nodeReduceFns, hyperNodeOrderFns                     ...NodeMapFn
    preemptableFns, reclaimableFns                                   ...EvictableFn
    overusedFns, allocatableFns                                      ...ValidateFn
    preemptiveFns                                                    ...ValidateWithCandidateFn
    jobReadyFns, jobPipelinedFns, jobValidFns                        ...ValidateFn/VoteFn
    jobEnqueueableFns, jobEnqueuedFns                                ...VoteFn/JobEnqueuedFn
    targetJobFns, reservedNodesFns, victimTasksFns, jobStarvingFns   ...各类函数
    // 模拟调度扩展点
    simulateRemoveTaskFns, simulateAddTaskFns                        ...SimulateFn
    simulatePredicateFns, simulateAllocatableFns                     ...SimulateFn
    // SubJob 扩展点
    subJobReadyFns, subJobPipelinedFns, subJobOrderFns               ...各类函数
    // HyperNode 梯度扩展点
    hyperNodeGradientForJobFns, hyperNodeGradientForSubJobFns        ...GradientFn

    // CycleState 缓存
    cycleStatesMap sync.Map              // TaskID -> CycleState
    NodesInShard   sets.Set[string]      // 分片节点集合
}

ssn := openSession(cache)

snapshot := cache.Snapshot()

func (ssn *Session) JobOrderFn(l, r interface{}) bool {
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            if !isEnabled(plugin.EnabledJobOrder) { continue }
            fn, found := ssn.jobOrderFns[plugin.Name]
            if !found { continue }
            if result := fn(l, r); result != 0 {
                return result < 0  // 第一个非零结果即返回
            }
        }
    }
    // 所有 Plugin 都返回 0，使用默认排序
    return defaultJobOrder(l, r)
}

// VoteFn 返回值：-1 = Reject, 0 = Abstain, +1 = Permit
func (ssn *Session) JobPipelined(obj interface{}) bool {
    var hasFound bool
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            res := fn(obj)
            if res < 0 { return false }   // Reject 立即生效
            if res > 0 { hasFound = true } // Permit 标记
        }
        if hasFound { return true }        // Tier 级别的 Permit
    }
    return true // 默认允许
}

func (ssn *Session) Preemptable(preemptor *api.TaskInfo, preemptees []*api.TaskInfo) []*api.TaskInfo {
    var victims []*api.TaskInfo
    for _, tier := range ssn.Tiers {
        for _, plugin := range tier.Plugins {
            candidates, abstain := fn(preemptor, preemptees)
            if abstain == 0 { continue }     // 弃权
            if len(candidates) == 0 {
                victims = nil; break          // 空列表 = 否决
            }
            if victims == nil {
                victims = candidates           // 第一票
            } else {
                victims = intersect(victims, candidates) // 取交集
            }
        }
        if victims != nil { return victims }
    }
    return victims
}

type EventHandler struct {
    AllocateFunc   func(event *Event)  // Task 分配到 Node 时触发
    DeallocateFunc func(event *Event)  // Task 从 Node 释放时触发
}

// 存储：TaskID -> CycleState
cycleStatesMap sync.Map

// PreFilter 阶段：计算并存储中间结果
state := ssn.GetCycleState(taskUID)
state.Write("nodeaffinity", &NodeAffinityState{...})

// Filter 阶段：读取中间结果
data, _ := state.Read("nodeaffinity")
affinityState := data.(*NodeAffinityState)

tiers:
  - plugins:
      - name: gang        # Tier 1（最高优先级）
      - name: priority
  - plugins:
      - name: drf         # Tier 2
      - name: proportion
  - plugins:
      - name: predicates  # Tier 3
      - name: nodeorder

plugins:
  - name: gang
    enabledJobOrder: true
    enabledJobReady: true
    enabledJobPipelined: true
    enabledPreemptable: true
    # 其他扩展点默认启用