拓扑感知调度调试指南

// 源码: pkg/scheduler/framework/session.go Line 101-112
type Session struct {
    // ...
    // HyperNodes 存储每个 HyperNode 的信息
    HyperNodes           api.HyperNodeInfoMap          // map[string]*HyperNodeInfo
    HyperNodeTierNameMap api.HyperNodeTierNameMap
    // HyperNodesSetByTier 按 Tier 分组的 HyperNode 集合（从低到高）
    HyperNodesSetByTier  map[int]sets.Set[string]
    HyperNodesTiers      []int                          // 排序后的 Tier 列表
    // RealNodesList 每个 HyperNode 下的真实节点列表
    RealNodesList        map[string][]*api.NodeInfo
    RealNodesSet         map[string]sets.Set[string]
    // HyperNodesReadyToSchedule HyperNode 是否就绪
    HyperNodesReadyToSchedule bool
}

// 源码: pkg/scheduler/plugins/network-topology-aware/network_topology_aware.go Line 64-96
type networkTopologyAwarePlugin struct {
    pluginArguments        framework.Arguments
    weight                 *priorityWeight       // 评分权重配置
    *normalPodConfig                              // 普通 Pod BinPack 配置
    *hyperNodesTier                               // Tier 范围 (min/max)
    hyperNodeResourceCache map[string]*resourceStatus  // HyperNode 资源缓存
}

type resourceStatus struct {
    allocatable *api.Resource   // HyperNode 可分配资源总量
    used        *api.Resource   // HyperNode 已使用资源
    idle        *api.Resource   // HyperNode 空闲资源
    futureIdle  *api.Resource   // HyperNode 未来可用资源
}

// 源码: pkg/scheduler/cache/event_handlers.go Line 1356-1394
// AddHyperNode 将 HyperNode 名称加入处理队列
func (sc *SchedulerCache) AddHyperNode(obj interface{}) {
    hn, ok := obj.(*topologyv1alpha1.HyperNode)
    if !ok {
        klog.ErrorS(nil, "Cannot convert to *topologyv1alpha1.HyperNode", ...)
        return
    }
    sc.hyperNodesQueue.Add(string(hyperNodeEventSourceHyperNode) + "/" + hn.Name)
}

// UpdateHyperNode 处理 HyperNode 更新事件
func (sc *SchedulerCache) UpdateHyperNode(oldObj, newObj interface{}) {
    newHyperNode, ok := newObj.(*topologyv1alpha1.HyperNode)
    // ...
    sc.hyperNodesQueue.Add(string(hyperNodeEventSourceHyperNode) + "/" + newHyperNode.Name)
}

// DeleteHyperNode 处理 HyperNode 删除事件
func (sc *SchedulerCache) DeleteHyperNode(obj interface{}) {
    // ...
    sc.hyperNodesQueue.Add(string(hyperNodeEventSourceHyperNode) + "/" + hn.Name)
}

// 源码: pkg/scheduler/cache/event_handlers.go Line 770-782
func (sc *SchedulerCache) processHyperNode(name string) error {
    sc.HyperNodesInfo.Lock()
    defer sc.HyperNodesInfo.Unlock()

    hn, err := sc.hyperNodeInformer.Lister().Get(name)
    if err != nil {
        if !errors.IsNotFound(err) {
            return fmt.Errorf("failed to get hyperNode <%s>: %v", name, err)
        }
        return sc.deleteHyperNode(name)  // 已删除，从缓存移除
    }
    return sc.updateHyperNode(hn)  // 更新缓存
}

// 源码: pkg/scheduler/cache/event_handlers.go Line 737-768
func (sc *SchedulerCache) triggerUpdateHyperNode(name string) error {
    sc.HyperNodesInfo.Lock()
    defer sc.HyperNodesInfo.Unlock()

    leafNodes := sc.HyperNodesInfo.GetRegexOrLabelMatchLeafHyperNodes()
    for leafNode := range leafNodes {
        match, err := sc.HyperNodesInfo.NodeRegexOrLabelMatchLeafHyperNode(name, hn.Name)
        if match {
            // 更新匹配的叶子 HyperNode 及其祖先
            sc.updateHyperNode(hn)
        }
    }
    return nil
}

// 源码: pkg/scheduler/framework/session.go Line 240-258
ssn.HyperNodes = snapshot.HyperNodes
ssn.HyperNodesSetByTier = snapshot.HyperNodesSetByTier
ssn.HyperNodeTierNameMap = snapshot.HyperNodeTierNameMap
ssn.RealNodesList, ssn.RealNodesSet = util.GetRealNodesByHyperNode(snapshot.RealNodesSet, snapshot.Nodes)
ssn.HyperNodesReadyToSchedule = snapshot.HyperNodesReadyToSchedule

// 添加虚拟顶层 HyperNode
ssn.addClusterTopHyperNode(ssn.NodeList)
// 解析 Tier 层级顺序
ssn.parseHyperNodesTiers()

if ssn.HyperNodesReadyToSchedule {
    hyperNodeSet := sets.New[string]()
    for hn := range ssn.HyperNodes {
        hyperNodeSet.Insert(hn)
    }
    for _, job := range ssn.Jobs {
        // 移除无效的 AllocatedHyperNode
        ssn.removeInvalidAllocatedHyperNode(job, ssn.HyperNodes)
        // 恢复 AllocatedHyperNode（调度器重启恢复）
        ssn.recoverAllocatedHyperNode(job, hyperNodeSet, ssn.HyperNodes, ssn.RealNodesSet)
    }
}

// 源码: pkg/scheduler/framework/session.go Line 284-312
func (ssn *Session) addClusterTopHyperNode(nodes []*api.NodeInfo) {
    topTier := 1
    for tier := range ssn.HyperNodesSetByTier {
        if tier >= topTier {
            topTier = tier + 1  // 比最高真实 Tier 大 1
        }
    }

    topHn := &topologyv1alpha1.HyperNode{}
    topHn.Name = ClusterTopHyperNode  // "<cluster-top-hypernode>"
    topHn.Spec.Tier = topTier
    topHni := api.NewHyperNodeInfo(topHn)

    // 将所有无父节点的 HyperNode 作为 children
    for _, hni := range ssn.HyperNodes {
        if hni.Parent != "" {
            continue
        }
        topHni.Children.Insert(hni.Name)
    }

    // 注册到 Session
    ssn.HyperNodes[topHni.Name] = topHni
    ssn.HyperNodesSetByTier[topHni.Tier()] = sets.New(topHni.Name)
    ssn.RealNodesList[topHni.Name] = nodes  // 包含所有节点
}

// 源码: pkg/scheduler/plugins/network-topology-aware/network_topology_aware.go Line 144-154
func New(arguments framework.Arguments) framework.Plugin {
    plugin := networkTopologyAwarePlugin{
        pluginArguments:        arguments,
        weight:                 getPriorityWeight(arguments),    // 解析评分权重
        normalPodConfig:        getNormalPodConfig(arguments),   // 普通 Pod 配置
        hyperNodesTier:         &hyperNodesTier{},
        hyperNodeResourceCache: make(map[string]*resourceStatus),
    }
    return &plugin
}

// 源码: network_topology_aware.go Line 262-339
func (nta *networkTopologyAwarePlugin) OnSessionOpen(ssn *framework.Session) {
    // 初始化 Tier 范围
    nta.hyperNodesTier.init(ssn.HyperNodesTiers)
    // 初始化 HyperNode 资源缓存
    nta.initHyperNodeResourceCache(ssn)

    // 注册 HyperNodeOrderFn: HyperNode 评分
    ssn.AddHyperNodeOrderFn(nta.Name(), ...)

    // 注册 BatchNodeOrderFn: 节点评分
    ssn.AddBatchNodeOrderFn(nta.Name(), ...)

    // 注册 HyperNodeGradientForJobFn: Job 级梯度搜索
    ssn.AddHyperNodeGradientForJobFn(nta.Name(), ...)

    // 注册 HyperNodeGradientForSubJobFn: SubJob 级梯度搜索
    ssn.AddHyperNodeGradientForSubJobFn(nta.Name(), ...)

    // 注册 EventHandler: 资源缓存更新
    ssn.AddEventHandler(...)
}

// 源码: network_topology_aware.go Line 278-290
ssn.AddHyperNodeGradientForJobFn(nta.Name(), func(job *api.JobInfo, hyperNode *api.HyperNodeInfo) [][]*api.HyperNodeInfo {
    if hardMode, highestAllowedTier := job.IsHardTopologyMode(); hardMode {
        jobMinResource := job.GetMinResources()
        result, err := nta.hyperNodeGradientFn(ssn, hyperNode, highestAllowedTier,
            job.AllocatedHyperNode, jobMinResource)
        if err != nil {
            klog.ErrorS(err, "build hyperNode gradient fail", "job", job.UID, ...)
            return nil
        }
        return result
    }
    // Soft Mode: 直接返回输入的 HyperNode
    return [][]*api.HyperNodeInfo{{hyperNode}}
})

// 源码: network_topology_aware.go Line 564-609
func (nta *networkTopologyAwarePlugin) hyperNodeGradientFn(
    ssn *framework.Session,
    hyperNode *api.HyperNodeInfo,
    highestAllowedTier int,
    allocatedHyperNode string,
    minResource *api.Resource,
) ([][]*api.HyperNodeInfo, error) {

    enqueued := set.New[string]()
    var processQueue []*api.HyperNodeInfo

    // 确定搜索根节点
    searchRoot, err := getSearchRoot(ssn.HyperNodes, hyperNode,
        highestAllowedTier, allocatedHyperNode)
    processQueue = append(processQueue, searchRoot)
    enqueued.Insert(searchRoot.Name)

    eligibleHyperNodes := make(map[int][]*api.HyperNodeInfo)

    // BFS 遍历
    for len(processQueue) > 0 {
        current := processQueue[0]
        processQueue = processQueue[1:]

        // 检查 HyperNode 是否满足条件
        if nta.isEligibleHyperNode(current, highestAllowedTier,
            allocatedHyperNode, minResource) {
            eligibleHyperNodes[current.Tier()] = append(
                eligibleHyperNodes[current.Tier()], current)
        }

        // 将子 HyperNode 加入队列
        for child := range current.Children {
            if enqueued.Has(child) {
                continue
            }
            processQueue = append(processQueue, ssn.HyperNodes[child])
            enqueued.Insert(child)
        }
    }

    // 按 Tier 升序组织结果
    var tiers []int
    for tier := range eligibleHyperNodes {
        tiers = append(tiers, tier)
    }
    sort.Ints(tiers)

    var result [][]*api.HyperNodeInfo
    for _, tier := range tiers {
        result = append(result, eligibleHyperNodes[tier])
    }
    return result, nil
}

// 源码: network_topology_aware.go Line 611-629
func (nta *networkTopologyAwarePlugin) isEligibleHyperNode(
    hn *api.HyperNodeInfo, highestAllowedTier int,
    allocatedHyperNode string, minResource *api.Resource) bool {

    // 条件 1: Tier 不超过允许的最高层级
    if hn.Tier() > highestAllowedTier {
        return false
    }

    // 条件 2: 部分运行场景跳过资源预检
    if allocatedHyperNode != "" {
        return true
    }

    // 条件 3: 资源满足检查
    hnResourceStatus, found := nta.hyperNodeResourceCache[hn.Name]
    if !found {
        return true  // 缓存不存在时跳过预检
    }

    // idle 或 futureIdle 满足 minResource 即可
    if minResource.LessEqual(hnResourceStatus.idle, api.Zero) ||
       minResource.LessEqual(hnResourceStatus.futureIdle, api.Zero) {
        return true
    }
    return false
}

// 源码: network_topology_aware.go Line 635-660
func getSearchRoot(hyperNodes api.HyperNodeInfoMap,
    hyperNodeAvailable *api.HyperNodeInfo,
    highestAllowedTier int,
    allocatedHyperNode string) (*api.HyperNodeInfo, error) {

    if allocatedHyperNode == "" {
        return hyperNodeAvailable, nil  // 初次调度，直接使用可用 HyperNode
    }

    // 找到 allocatedHyperNode 在 highestAllowedTier 范围内的最高祖先
    hyperNodeHighestAllowed, err := getHighestAllowedHyperNode(
        hyperNodes, highestAllowedTier, allocatedHyperNode)

    // 计算 hyperNodeAvailable 和 hyperNodeHighestAllowed 的交集
    lca := hyperNodes.GetLCAHyperNode(hyperNodeAvailable.Name, hyperNodeHighestAllowed)
    // ... 返回交集根节点
}

// 源码: network_topology_aware.go Line 341-365
func (nta *networkTopologyAwarePlugin) HyperNodeOrderFn(
    ssn *framework.Session,
    subJob *api.SubJobInfo,
    hyperNodes map[string][]*api.NodeInfo) (map[string]float64, error) {

    // Step 1: BinPacking 评分
    hyperNodeScores := nta.getSubJobHyperNodeBinPackingScore(subJob, hyperNodes)

    // Step 2: 如果最高分有多个候选，用 Task 数量打破平局
    scoreToHyperNodes := map[float64][]string{}
    var maxScore float64 = -1
    for hyperNode, score := range hyperNodeScores {
        if score >= maxScore {
            maxScore = score
            scoreToHyperNodes[maxScore] = append(scoreToHyperNodes[maxScore], hyperNode)
        }
    }

    if len(scoreToHyperNodes[maxScore]) > 1 {
        candidateHyperNodes := scoreToHyperNodes[maxScore]
        for _, hyperNode := range candidateHyperNodes {
            taskNumScore := nta.scoreWithTaskNum(hyperNode, subJob.Tasks, ssn.RealNodesList)
            hyperNodeScores[hyperNode] += taskNumScore
        }
    }

    // Step 3: 缩放分数到最终范围
    hyperNodeScores = nta.scaleFinalScore(hyperNodeScores)
    return hyperNodeScores, nil
}

// 源码: network_topology_aware.go Line 367-420
func (nta *networkTopologyAwarePlugin) getSubJobHyperNodeBinPackingScore(
    subJob *api.SubJobInfo,
    hyperNodes map[string][]*api.NodeInfo) map[string]float64 {

    // 计算 SubJob 的总资源请求
    tasksRequest := make(map[corev1.ResourceName]float64)
    for _, task := range subJob.Tasks {
        for _, resourceName := range task.Resreq.ResourceNames() {
            if _, ok := nta.weight.getBinPackWeight(resourceName); !ok {
                continue
            }
            tasksRequest[resourceName] += task.Resreq.Get(resourceName)
        }
    }

    // 对每个 HyperNode 计算 BinPacking 分数
    for hyperNode := range hyperNodes {
        for resourceName, request := range tasksRequest {
            status := nta.hyperNodeResourceCache[hyperNode]
            allocatable := status.allocatable.Get(resourceName)
            used := status.used.Get(resourceName)

            if used+request > allocatable {
                // 资源不足，得分为 0
                overused = true
                break
            }
            // BinPacking 核心: (used + request) / allocatable
            score := (used + request) / allocatable
            totalScore += float64(weight) * score
            totalWeight += weight
        }
        // 加权平均
        hyperNodeBinPackingScores[hyperNode] = totalScore / float64(totalWeight)
    }
}

// 源码: network_topology_aware.go Line 422-440
func (nta *networkTopologyAwarePlugin) batchNodeOrderFn(
    ssn *framework.Session, task *api.TaskInfo, nodes []*api.NodeInfo) (map[string]float64, error) {

    job := ssn.Jobs[task.Job]
    subJob := job.SubJobs[job.TaskToSubJob[task.UID]]

    if subJob.WithNetworkTopology() {
        // 有拓扑需求: 基于 LCA Tier 评分
        nodeScores, err = nta.batchNodeOrderFnForNetworkAwarePods(ssn, task, subJob, nodes)
    } else {
        // 无拓扑需求: 按 Tier 衰减的 BinPacking 评分
        nodeScores, err = nta.batchNodeOrderFnForNormalPods(ssn, task, nodes)
    }

    nodeScores = nta.scaleFinalScore(nodeScores)
    return nodeScores, nil
}

// 源码: network_topology_aware.go Line 522-552
func (nta *networkTopologyAwarePlugin) batchNodeOrderFnForNetworkAwarePods(
    ssn *framework.Session, task *api.TaskInfo, subJob *api.SubJobInfo,
    nodes []*api.NodeInfo) (map[string]float64, error) {

    allocatedHyperNode := task.JobAllocatedHyperNode
    if allocatedHyperNode == "" {
        return nodeScores, nil
    }

    for _, node := range nodes {
        // 找到节点所属的 HyperNode
        hyperNode := util.FindHyperNodeForNode(node.Name, ...)
        // 计算拓扑评分: LCA Tier 越低 -> 分数越高
        score := nta.networkTopologyAwareScore(hyperNode, allocatedHyperNode, ssn.HyperNodes)
        nodeScores[node.Name] = score
    }

    // 平局打破: 已分配 Task 数
    if len(scoreToNodes[maxScore]) > 1 {
        for _, node := range candidateNodes {
            hyperNode := util.FindHyperNodeForNode(node, ...)
            taskNumScore := nta.scoreWithTaskNum(hyperNode, subJob.Tasks, ssn.RealNodesList)
            nodeScores[node] += taskNumScore
        }
    }
    return nodeScores, nil
}

// 源码: network_topology_aware.go Line 691-706
func (nta *networkTopologyAwarePlugin) networkTopologyAwareScore(
    hyperNodeName, jobAllocatedHyperNode string,
    hyperNodeMap api.HyperNodeInfoMap) float64 {

    if hyperNodeName == jobAllocatedHyperNode {
        return FullScore  // 1.0 - 同一 HyperNode
    }

    // 找到两个 HyperNode 的最近公共祖先 (LCA)
    LCAHyperNode := hyperNodeMap.GetLCAHyperNode(hyperNodeName, jobAllocatedHyperNode)
    hyperNodeInfo := hyperNodeMap[LCAHyperNode]

    // 评分: (maxTier - LCA.Tier) / (maxTier - minTier)
    // LCA Tier 越低 -> 网络距离越近 -> 分数越高
    return nta.scoreHyperNodeWithTier(hyperNodeInfo.Tier())
}

// 源码: network_topology_aware.go Line 442-479
func (nta *networkTopologyAwarePlugin) batchNodeOrderFnForNormalPods(
    ssn *framework.Session, task *api.TaskInfo, nodes []*api.NodeInfo) (map[string]float64, error) {

    // 计算每个 Tier 的权重 (衰减因子: fading^(tier-1))
    for tier := nta.minTier; tier <= nta.maxTier; tier++ {
        tierWeight := math.Pow(nta.hyperNodeBinPackingFading, float64(tier-1))
        tierWeights[tier] = tierWeight
    }

    for _, node := range nodes {
        totalScore := 0.0
        for tier := nta.minTier; tier <= nta.maxTier; tier++ {
            tierScore := FullScore  // 默认满分（不属于任何 HyperNode）
            for hyperNodeName := range ssn.HyperNodesSetByTier[tier] {
                if ssn.RealNodesSet[hyperNodeName].Has(node.Name) {
                    tierScore = nta.getPodHyperNodeBinPackingScore(task, hyperNodeName)
                    break
                }
            }
            totalScore += tierWeights[tier] * tierScore
        }
        nodeScores[node.Name] = totalScore / totalTierWeight
    }
}

// 源码: network_topology_aware.go Line 106-125
func (nta *networkTopologyAwarePlugin) initHyperNodeResourceCache(ssn *framework.Session) {
    nta.hyperNodeResourceCache = make(map[string]*resourceStatus)

    for hyperNode := range ssn.HyperNodes {
        nta.hyperNodeResourceCache[hyperNode] = &resourceStatus{
            allocatable: api.EmptyResource(),
            used:        api.EmptyResource(),
            idle:        api.EmptyResource(),
            futureIdle:  api.EmptyResource(),
        }
        // 聚合该 HyperNode 下所有真实节点的资源
        for node := range ssn.RealNodesSet[hyperNode] {
            nta.hyperNodeResourceCache[hyperNode].allocatable.Add(ssn.Nodes[node].Allocatable)
            nta.hyperNodeResourceCache[hyperNode].used.Add(ssn.Nodes[node].Used)
            nta.hyperNodeResourceCache[hyperNode].idle.Add(ssn.Nodes[node].Idle)
            nta.hyperNodeResourceCache[hyperNode].futureIdle.Add(ssn.Nodes[node].FutureIdle())
        }
    }
}

// 源码: network_topology_aware.go Line 309-338
ssn.AddEventHandler(&framework.EventHandler{
    AllocateFunc: func(event *framework.Event) {
        task := event.Task
        node := task.NodeName
        for hyperNode := range ssn.HyperNodes {
            if ssn.RealNodesSet[hyperNode].Has(node) {
                status := nta.hyperNodeResourceCache[hyperNode]
                status.used.Add(task.Resreq)  // 增加使用量
            }
        }
    },
    DeallocateFunc: func(event *framework.Event) {
        task := event.Task
        node := task.NodeName
        for hyperNode := range ssn.HyperNodes {
            if ssn.RealNodesSet[hyperNode].Has(node) {
                status := nta.hyperNodeResourceCache[hyperNode]
                status.used.Sub(task.Resreq)  // 减少使用量
            }
        }
    },
})

// 源码: pkg/scheduler/framework/session.go Line 362-445
func (ssn *Session) recoverAllocatedHyperNode(
    job *api.JobInfo, hyperNodeSet sets.Set[string],
    hyperNodes api.HyperNodeInfoMap,
    nodesByHyperNode map[string]sets.Set[string]) {

    // Step 1: 遍历每个 SubJob 的已分配 Task
    for _, subJob := range job.SubJobs {
        // Step 2: 根据 Task 所在节点找到 HyperNode 交集
        var subJobAllocatedHyperNode sets.Set[string]
        for _, task := range allocatedTasks {
            taskAllocatedHyperNode := sets.New[string]()
            for hn := range search {
                if nodes.Has(task.NodeName) {
                    taskAllocatedHyperNode.Insert(hn)
                }
            }
            subJobAllocatedHyperNode = taskAllocatedHyperNode
        }
        // Step 3: 选择最低 Tier 的 HyperNode
        minimumHyperNode := getLowestTierHyperNode(subJobAllocatedHyperNode, hyperNodes)
        subJob.AllocatedHyperNode = minimumHyperNode
    }

    // Step 4: 根据 SubJob 的 AllocatedHyperNode 计算 Job 的 LCA
    var lca string
    for _, subJob := range job.SubJobs {
        lca = hyperNodes.GetLCAHyperNode(lca, subJob.AllocatedHyperNode)
    }
    job.AllocatedHyperNode = lca
}

tiers:
- plugins:
  - name: network-topology-aware
    arguments:
      weight: 10                              # 全局评分权重
      hypernode.binpack.cpu: 5                # CPU BinPack 权重
      hypernode.binpack.memory: 1             # Memory BinPack 权重
      hypernode.binpack.resources: nvidia.com/gpu, example.com/foo  # 额外资源
      hypernode.binpack.resources.nvidia.com/gpu: 2                 # GPU 权重
      hypernode.binpack.resources.example.com/foo: 3                # 自定义资源权重
      hypernode.binpack.normal-pod.enable: true                     # 普通 Pod BinPack
      hypernode.binpack.normal-pod.fading: 0.8                      # Tier 衰减因子