队列公平分享调试指南

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 57-74
type queueAttr struct {
    queueID api.QueueID
    name    string
    weight  int32          // 队列权重，决定公平分配比例
    share   float64        // allocated/deserved 比率，用于排序

    deserved  *api.Resource  // 公平分配目标（计算得出）
    allocated *api.Resource  // 当前已分配资源
    request   *api.Resource  // 队列中所有 Job 的资源请求总和
    elastic   *api.Resource  // 可回收部分 (allocated - minAvailable)
    inqueue   *api.Resource  // 已入队 Job 预留的资源
    capability     *api.Resource  // 用户配置的硬上限
    realCapability *api.Resource  // 实际可用上限 (考虑 guarantee)
    guarantee      *api.Resource  // 最低保障资源
}

// Line 189: 初始化 remaining 为集群总资源
remaining := pp.totalResource.Clone()
meet := map[api.QueueID]struct{}{}

for {
    // Line 192-198: 计算未满足队列的总权重
    totalWeight := int32(0)
    for _, attr := range pp.queueOpts {
        if _, found := meet[attr.queueID]; found {
            continue
        }
        totalWeight += attr.weight
    }

    // Line 201-204: 所有队列都满足或无队列，退出
    if totalWeight == 0 {
        break
    }

    oldRemaining := remaining.Clone()
    increasedDeserved := api.EmptyResource()
    decreasedDeserved := api.EmptyResource()

    for _, attr := range pp.queueOpts {
        if _, found := meet[attr.queueID]; found {
            continue
        }

        oldDeserved := attr.deserved.Clone()

        // Line 220: 按权重比例分配 remaining 资源
        attr.deserved.Add(remaining.Clone().Multi(float64(attr.weight) / float64(totalWeight)))

        // Line 222-224: 应用 realCapability 上限
        if attr.realCapability != nil {
            attr.deserved.MinDimensionResource(attr.realCapability, api.Infinity)
        }

        // Line 225: 应用 request 上限（不超过实际需求）
        attr.deserved.MinDimensionResource(attr.request, api.Zero)

        // Line 227: 应用 guarantee 下限（不低于保障值）
        attr.deserved = helpers.Max(attr.deserved, attr.guarantee)

        // Line 231-237: 检查是否满足
        if attr.request.LessEqual(attr.deserved, api.Zero) {
            meet[attr.queueID] = struct{}{}
        } else if equality.Semantic.DeepEqual(attr.deserved, oldDeserved) {
            meet[attr.queueID] = struct{}{}  // capability 限制导致无法继续增长
        }

        // Line 242-244: 计算本轮增减
        increased, decreased := attr.deserved.Diff(oldDeserved, api.Zero)
        increasedDeserved.Add(increased)
        decreasedDeserved.Add(decreased)
    }

    // Line 250: 更新 remaining
    remaining = api.ExceededPart(remaining.Clone().Add(decreasedDeserved), increasedDeserved)

    // Line 252-255: 检查终止条件
    if remaining.IsEmpty() || equality.Semantic.DeepEqual(remaining, oldRemaining) {
        break
    }
}

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 545-557
func updateQueueAttrShare(attr *queueAttr) {
    res := float64(0)
    for _, rn := range attr.deserved.ResourceNames() {
        share := helpers.Share(attr.allocated.Get(rn), attr.deserved.Get(rn))
        if share > res {
            res = share
        }
    }
    attr.share = res
}

func Share(l, r float64) float64 {
    var share float64
    if r == 0 {
        if l == 0 {
            share = 0   // 无 deserved 且无 allocated -> 0
        } else {
            share = 1   // 无 deserved 但有 allocated -> 1（已超用）
        }
    } else {
        share = l / r   // 正常计算比率
    }
    return share
}

// Line 258-276
ssn.AddQueueOrderFn(pp.Name(), func(l, r interface{}) int {
    lv := l.(*api.QueueInfo)
    rv := r.(*api.QueueInfo)

    // 优先级不同时，高优先级排前面
    if lv.Queue.Spec.Priority != rv.Queue.Spec.Priority {
        return int(rv.Queue.Spec.Priority) - int(lv.Queue.Spec.Priority)
    }

    // 优先级相同时，低 share 排前面
    if pp.queueOpts[lv.UID].share == pp.queueOpts[rv.UID].share {
        return 0
    }
    if pp.queueOpts[lv.UID].share < pp.queueOpts[rv.UID].share {
        return -1
    }
    return 1
})

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 300-312
ssn.AddOverusedFn(pp.Name(), func(obj interface{}) bool {
    queue := obj.(*api.QueueInfo)
    attr := pp.queueOpts[queue.UID]

    // 判定: allocated >= deserved（所有维度）
    overused := attr.deserved.LessEqual(attr.allocated, api.Zero)
    metrics.UpdateQueueOverused(attr.name, overused)

    if overused {
        klog.V(3).Infof("Queue <%v> is overused: deserved <%v>, allocated <%v>, share <%v>",
            queue.Name, attr.deserved, attr.allocated, attr.share)
    }
    return overused
})

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 314-333
queueAllocatable := func(queue *api.QueueInfo, candidate *api.TaskInfo) bool {
    if queue.Queue.Status.State != scheduling.QueueStateOpen {
        klog.V(3).Infof("Queue <%s> current state: %s, is not in open state", ...)
        return false
    }

    attr := pp.queueOpts[queue.UID]
    // 模拟分配后的资源使用量
    futureUsed := attr.allocated.Clone().Add(candidate.Resreq)
    // 检查: allocated + task.Resreq <= deserved
    allocatable, _ := futureUsed.LessEqualWithDimensionAndResourcesName(attr.deserved, candidate.Resreq)
    if !allocatable {
        klog.V(3).Infof("Queue <%v>: deserved <%v>, allocated <%v>; Candidate <%v>: resource request <%v>",
            queue.Name, attr.deserved, attr.allocated, candidate.Name, candidate.Resreq)
    }
    return allocatable
}

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 278-298
ssn.AddReclaimableFn(pp.Name(), func(reclaimer *api.TaskInfo, reclaimees []*api.TaskInfo) ([]*api.TaskInfo, int) {
    var victims []*api.TaskInfo
    allocations := map[api.QueueID]*api.Resource{}

    for _, reclaimee := range reclaimees {
        job := ssn.Jobs[reclaimee.Job]
        attr := pp.queueOpts[job.Queue]

        if _, found := allocations[job.Queue]; !found {
            allocations[job.Queue] = attr.allocated.Clone()
        }
        allocated := allocations[job.Queue]

        // 关键判定: allocated > deserved 时，该 Task 可被回收
        if !allocated.LessEqual(attr.deserved, api.Zero) {
            allocated.Sub(reclaimee.Resreq)
            victims = append(victims, reclaimee)
        }
    }
    return victims, util.Permit
})

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 377-415
ssn.AddJobEnqueueableFn(pp.Name(), func(obj interface{}) int {
    job := obj.(*api.JobInfo)
    attr := pp.queueOpts[job.Queue]
    queue := ssn.Queues[job.Queue]

    // 检查 1: 队列是否 Open
    if queue.Queue.Status.State != scheduling.QueueStateOpen {
        return util.Reject
    }

    // 检查 2: 无 capability 限制时直接放行
    if attr.realCapability == nil {
        return util.Permit
    }

    // 检查 3: 无 MinResources 时直接放行
    if job.PodGroup.Spec.MinResources == nil {
        return util.Permit
    }

    // 检查 4: minReq + allocated + inqueue - elastic <= realCapability
    minReq := job.GetMinResources()
    r := minReq.Clone().Add(attr.allocated).Add(attr.inqueue).Sub(attr.elastic)
    inqueue, resourceNames := r.LessEqualWithDimensionAndResourcesName(attr.realCapability, minReq)

    if inqueue {
        attr.inqueue.Add(job.DeductSchGatedResources(minReq))
        return util.Permit
    }
    // 记录事件: 队列资源配额不足
    ssn.RecordPodGroupEvent(job.PodGroup, ...)
    return util.Reject
})

// 源码: pkg/scheduler/plugins/capacity/capacity.go Line 901-917
func (cp *capacityPlugin) checkQueueAllocatableHierarchically(
    ssn *framework.Session, queue *api.QueueInfo, candidate *api.TaskInfo) bool {

    // 构建检查列表: [root, ..., parent, queue]
    list := append(cp.queueOpts[queue.UID].ancestors, queue.UID)

    // 从叶子向上逐层检查
    for i := len(list) - 1; i >= 0; i-- {
        if !cp.queueAllocatable(ssn.Queues[list[i]], candidate) {
            // V(5) 时打印所有层级信息
            if klog.V(5).Enabled() {
                for j := i - 1; j >= 0; j-- {
                    cp.queueAllocatable(ssn.Queues[list[j]], candidate)
                }
            }
            return false
        }
    }
    return true
}

// 源码: pkg/scheduler/plugins/capacity/capacity.go Line 1032-1059
func updateQueueAttrShare(attr *queueAttr) {
    res := float64(0)

    // best-effort 队列: 无 deserved -> share = 1
    // 确保 best-effort 队列总是排在有 deserved 的队列之后
    if attr.deserved.IsEmpty() {
        attr.share = 1
        return
    }

    for _, rn := range attr.deserved.ResourceNames() {
        res = max(res, helpers.Share(attr.allocated.Get(rn), attr.deserved.Get(rn)))
    }
    attr.share = res
}

// 源码: pkg/scheduler/plugins/proportion/proportion.go Line 450-473
ssn.AddEventHandler(&framework.EventHandler{
    AllocateFunc: func(event *framework.Event) {
        job := ssn.Jobs[event.Task.Job]
        attr := pp.queueOpts[job.Queue]
        attr.allocated.Add(event.Task.Resreq)       // 分配时增加
        pp.updateShare(attr)                          // 重新计算 share
    },
    DeallocateFunc: func(event *framework.Event) {
        job := ssn.Jobs[event.Task.Job]
        attr := pp.queueOpts[job.Queue]
        attr.allocated.Sub(event.Task.Resreq)        // 回收时减少
        pp.updateShare(attr)                          // 重新计算 share
    },
})

// SimulateAllocatableFn (Line 335-355): 使用 CycleState 中的快照进行检查
// SimulateAddTaskFn (Line 417-431): 模拟添加 Task
// SimulateRemoveTaskFn (Line 433-447): 模拟移除 Task