Capacity Plugin 深度解析

type capacityPlugin struct {
    rootQueue      string           // 根队列名称（层级模式下为 "root"）
    totalResource  *api.Resource    // 集群总资源
    totalGuarantee *api.Resource    // 所有队列 guarantee 之和
    totalDeserved  *api.Resource    // 所有队列 deserved 之和

    queueOpts map[api.QueueID]*queueAttr  // 队列属性映射表
    pluginArguments framework.Arguments    // 插件参数
}

type queueAttr struct {
    queueID   api.QueueID
    name      string
    share     float64                    // 主导资源使用率
    ancestors []api.QueueID              // 祖先队列列表（从 root 到 parent）
    children  map[api.QueueID]*queueAttr // 子队列映射

    deserved       *api.Resource  // 应得资源（来自 Queue Spec）
    allocated      *api.Resource  // 已分配资源
    request        *api.Resource  // 资源需求总和
    elastic        *api.Resource  // 弹性资源（job.allocated - job.minAvailable 之和）
    inqueue        *api.Resource  // 已入队 Job 的资源需求
    capability     *api.Resource  // 用户配置的容量上限
    realCapability *api.Resource  // 实际容量上限（受 guarantee 修正）
    guarantee      *api.Resource  // 最低保障资源
}

hierarchyEnabled := ssn.HierarchyEnabled(cp.Name())
if hierarchyEnabled {
    readyToSchedule = cp.buildHierarchicalQueueAttrs(ssn)
} else {
    cp.buildQueueAttrs(ssn)
}

func (cp *capacityPlugin) updateAncestors(queue *api.QueueInfo, ssn *framework.Session, visited map[api.QueueID]struct{}) error {
    // 环路检测：如果 visited 中已存在当前队列，报错
    if _, exist := visited[queue.UID]; exist {
        return fmt.Errorf("cycle detected in queue hierarchy for queue %s", queue.Name)
    }
    visited[queue.UID] = struct{}{}
    defer delete(visited, queue.UID)
    // 确定父队列（默认为 root）
    parent := cp.rootQueue
    if queue.Queue.Spec.Parent != "" {
        parent = queue.Queue.Spec.Parent
    }
    // 递归处理父队列，建立 children 和 ancestors 关系
    ...
}

func updateQueueAttrShare(attr *queueAttr) {
    // Best-effort 队列（无 deserved）：share 固定为 1
    if attr.deserved.IsEmpty() {
        attr.share = 1
        return
    }
    // 有 deserved 的队列：取所有资源维度的最大使用率
    res := float64(0)
    for _, rn := range attr.deserved.ResourceNames() {
        res = max(res, helpers.Share(attr.allocated.Get(rn), attr.deserved.Get(rn)))
    }
    attr.share = res
}

func (cp *capacityPlugin) compareShareWithDeserved(lattr, rattr *queueAttr) int {
    if lattr.share == rattr.share {
        // 打破平局：有 deserved 的队列优先于 best-effort 队列
        lHasDeserved := !lattr.deserved.IsEmpty()
        rHasDeserved := !rattr.deserved.IsEmpty()
        if lHasDeserved == rHasDeserved { return 0 }
        if lHasDeserved { return -1 }
        return 1
    }
    if lattr.share < rattr.share { return -1 }
    return 1
}

futureUsed := attr.allocated.Clone().Add(task.Resreq)
isPreemptive, resourceNames := futureUsed.LessEqualPartlyWithDimensionZeroFiltered(attr.deserved, task.Resreq)

// 三项前置检查
1. Queue 必须处于 Open 状态
2. 层级模式下 Queue 必须是叶子队列
3. checkQueueAllocatableHierarchically：从叶子到根逐层检查

attr.inqueue.Add(deductedResources)
// 层级模式下同步更新所有祖先
for _, ancestorID := range attr.ancestors {
    ancestorAttr := cp.queueOpts[ancestorID]
    ancestorAttr.inqueue.Add(deductedResources)
}

ssn.AddPrePredicateFn(cp.Name(), func(task *api.TaskInfo) error {
    state := &capacityState{
        queueAttrs: make(map[api.QueueID]*queueAttr),
    }
    for _, queue := range cp.queueOpts {
        state.queueAttrs[queue.queueID] = queue.Clone()
    }
    ssn.GetCycleState(task.UID).Write(capacityStateKey, state)
    return nil
})

// 从叶子到根逐层检查模拟状态
list := append(state.queueAttrs[queue.UID].ancestors, queue.UID)
for i := len(list) - 1; i >= 0; i-- {
    if !simulateQueueAllocatable(state, ssn.Queues[list[i]], candidate) {
        return false
    }
}

AllocateFunc: func(event *framework.Event) {
    attr.allocated.Add(event.Task.Resreq)     // 更新 allocated
    metrics.UpdateQueueAllocated(...)          // 更新指标
    cp.updateShare(attr)                       // 重新计算 share
    // 层级模式：向所有祖先传播
    for _, ancestorID := range attr.ancestors {
        ancestorAttr.allocated.Add(event.Task.Resreq)
    }
}

DeallocateFunc: func(event *framework.Event) {
    attr.allocated.Sub(event.Task.Resreq)      // 更新 allocated
    metrics.UpdateQueueAllocated(...)           // 更新指标
    cp.updateShare(attr)                        // 重新计算 share
    // 层级模式：向所有祖先传播
    for _, ancestorID := range attr.ancestors {
        ancestorAttr.allocated.Sub(event.Task.Resreq)
    }
}

type capacityState struct {
    queueAttrs map[api.QueueID]*queueAttr
}

func (s *capacityState) Clone() fwk.StateData {
    newState := &capacityState{
        queueAttrs: make(map[api.QueueID]*queueAttr, len(s.queueAttrs)),
    }
    for qID, qa := range s.queueAttrs {
        newState.queueAttrs[qID] = qa.Clone()
    }
    return newState
}

ssn.AddVictimQueueOrderFn(cp.Name(), func(l, r, preemptor interface{}) int {
    lLevel := getQueueLevel(cp.queueOpts[lv.UID], cp.queueOpts[pv.UID])
    rLevel := getQueueLevel(cp.queueOpts[rv.UID], cp.queueOpts[pv.UID])
    // level 越大（公共祖先越深），表示与 preemptor 越 "亲近"
    // 亲近的队列排前面（优先被抢占）
    if lLevel > rLevel { return -1 }
    if lLevel < rLevel { return 1 }
    return 0
})

func getQueueLevel(l *queueAttr, r *queueAttr) int {
    level := 0
    for i := 0; i < min(len(l.ancestors), len(r.ancestors)); i++ {
        if l.ancestors[i] == r.ancestors[i] {
            level = i
        } else {
            return level
        }
    }
    return level
}

apiVersion: v1
kind: ConfigMap
metadata:
  name: volcano-scheduler-configmap
  namespace: volcano-system
data:
  volcano-scheduler.conf: |
    actions: "enqueue, allocate, reclaim, backfill, preempt"
    tiers:
    - plugins:
      - name: gang
      - name: capacity               # 使用 capacity 替代 proportion
        enabledHierarchy: true        # 启用层级模式（可选）
      - name: predicates
      - name: nodeorder

apiVersion: scheduling.volcano.sh/v1beta1
kind: Queue
metadata:
  name: training-queue
spec:
  weight: 1
  deserved:                           # Capacity Plugin 使用 deserved 而非 weight 计算
    cpu: "40"
    memory: "160Gi"
  capability:                         # 资源上限
    cpu: "60"
    memory: "240Gi"
  guarantee:                          # 最低保障
    resource:
      cpu: "10"
      memory: "40Gi"

# 根队列
apiVersion: scheduling.volcano.sh/v1beta1
kind: Queue
metadata:
  name: root
spec:
  deserved:
    cpu: "100"
    memory: "400Gi"
---
# 中间队列
apiVersion: scheduling.volcano.sh/v1beta1
kind: Queue
metadata:
  name: team-a
spec:
  parent: root                        # 指定父队列
  deserved:
    cpu: "60"
    memory: "240Gi"
  capability:
    cpu: "70"
    memory: "300Gi"
  guarantee:
    resource:
      cpu: "20"
      memory: "80Gi"
---
# 叶子队列（Job 只能提交到叶子队列）
apiVersion: scheduling.volcano.sh/v1beta1
kind: Queue
metadata:
  name: training
spec:
  parent: team-a                      # 指定父队列
  deserved:
    cpu: "40"
    memory: "160Gi"
  capability:
    cpu: "50"
    memory: "200Gi"
  guarantee:
    resource:
      cpu: "10"
      memory: "40Gi"