Volcano 调度插件开发指南

// Plugin is the interface of scheduler plugin
type Plugin interface {
    // The unique name of Plugin.
    Name() string

    OnSessionOpen(ssn *Session)
    OnSessionClose(ssn *Session)
}

type BindContextHandler interface {
    SetupBindContextExtension(state *k8sframework.CycleState, bindCtx *cache.BindContext)
}

type PluginBuilder = func(Arguments) Plugin

func New(arguments framework.Arguments) framework.Plugin {
    return &myPlugin{pluginArguments: arguments}
}

tiers:
- plugins:           # Tier 0 - 最高优先级
  - name: gang
  - name: priority
- plugins:           # Tier 1
  - name: drf
  - name: predicates
  - name: nodeorder

const (
    Permit  = 1    // 允许
    Abstain = 0    // 弃权
    Reject  = -1   // 拒绝
)

func isEnabled(enabled *bool) bool {
    return enabled != nil && *enabled
}

- name: gang
  enableJobOrder: true      # 显式启用 JobOrder
  enableJobReady: true      # 显式启用 JobReady
  enableJobPipelined: true  # 显式启用 JobPipelined
  # enablePredicate 未配置，PredicateFn 不会被调用

type Arguments map[string]interface{}

// 获取 int 值，通过指针修改目标变量
func (a Arguments) GetInt(ptr *int, key string)

// 获取 float64 值
func (a Arguments) GetFloat64(ptr *float64, key string)

// 获取 bool 值
func (a Arguments) GetBool(ptr *bool, key string)

// 获取 string 值
func (a Arguments) GetString(ptr *string, key string)

weight := 1                            // 默认值
args.GetInt(&weight, "myPlugin.weight") // key 存在则覆盖

func Get[T any](a Arguments, key string) (T, bool)

// 获取结构化配置
type MyConfig struct {
    Threshold float64 `mapstructure:"threshold"`
    Mode      string  `mapstructure:"mode"`
}

config, ok := framework.Get[MyConfig](args, "myPlugin.config")
if !ok {
    // key 不存在，使用默认配置
}

tiers:
- plugins:
  - name: binpack
    enableNodeOrder: true
    arguments:
      binpack.weight: 10
      binpack.cpu: 5
      binpack.memory: 1
      binpack.resources: nvidia.com/gpu, example.com/foo
      binpack.resources.nvidia.com/gpu: 2
      binpack.resources.example.com/foo: 3

func calculateWeight(args framework.Arguments) priorityWeight {
    weight := priorityWeight{
        BinPackingWeight:    1,  // 默认值
        BinPackingCPU:       1,
        BinPackingMemory:    1,
        BinPackingResources: make(map[v1.ResourceName]int),
    }

    args.GetInt(&weight.BinPackingWeight, BinpackWeight)
    args.GetInt(&weight.BinPackingCPU, BinpackCPU)
    args.GetInt(&weight.BinPackingMemory, BinpackMemory)
    // ... 解析扩展资源
    return weight
}

package costaware

import (
    "k8s.io/klog/v2"

    "volcano.sh/volcano/pkg/scheduler/api"
    "volcano.sh/volcano/pkg/scheduler/framework"
    "volcano.sh/volcano/pkg/scheduler/plugins/util"
)

const (
    // PluginName indicates name of volcano scheduler plugin.
    PluginName = "costaware"

    // 配置 key
    CostWeightKey = "costaware.weight"
)

type costAwarePlugin struct {
    pluginArguments framework.Arguments
    weight          float64
}

// New 是 PluginBuilder 工厂函数
func New(arguments framework.Arguments) framework.Plugin {
    weight := 1.0
    arguments.GetFloat64(&weight, CostWeightKey)

    return &costAwarePlugin{
        pluginArguments: arguments,
        weight:          weight,
    }
}

func (cp *costAwarePlugin) Name() string {
    return PluginName
}

func (cp *costAwarePlugin) OnSessionOpen(ssn *framework.Session) {
    klog.V(5).Infof("Enter costaware plugin ...")
    // 在这里注册扩展点（见 Step 3）
}

func (cp *costAwarePlugin) OnSessionClose(ssn *framework.Session) {
    klog.V(5).Infof("Leaving costaware plugin ...")
}

func (cp *costAwarePlugin) OnSessionOpen(ssn *framework.Session) {
    klog.V(5).Infof("Enter costaware plugin ...")

    // 1. Job 排序 - 高优先级 Job 优先
    ssn.AddJobOrderFn(cp.Name(), func(l, r interface{}) int {
        lJob := l.(*api.JobInfo)
        rJob := r.(*api.JobInfo)

        klog.V(4).Infof("CostAware JobOrder: <%v/%v> priority=%d, <%v/%v> priority=%d",
            lJob.Namespace, lJob.Name, lJob.Priority,
            rJob.Namespace, rJob.Name, rJob.Priority)

        if lJob.Priority > rJob.Priority {
            return -1 // l 优先
        }
        if lJob.Priority < rJob.Priority {
            return 1  // r 优先
        }
        return 0 // 相等，交给下一个 Plugin
    })

    // 2. 节点评分 - 空闲资源越少的节点评分越高（binpack 风格，降低成本）
    ssn.AddNodeOrderFn(cp.Name(), func(task *api.TaskInfo, node *api.NodeInfo) (float64, error) {
        // 计算节点的资源使用率
        cpuUsageRatio := 0.0
        if node.Allocatable.MilliCPU > 0 {
            cpuUsageRatio = node.Used.MilliCPU / node.Allocatable.MilliCPU
        }
        memUsageRatio := 0.0
        if node.Allocatable.Memory > 0 {
            memUsageRatio = node.Used.Memory / node.Allocatable.Memory
        }

        // 使用率越高越好（减少碎片，降低成本）
        score := (cpuUsageRatio + memUsageRatio) / 2 * cp.weight * 100

        klog.V(4).Infof("CostAware score for task %s/%s on node %s: %.2f",
            task.Namespace, task.Name, node.Name, score)

        return score, nil
    })

    // 3. 抢占 - 只允许抢占比自己优先级低的 Job 的 Task
    ssn.AddPreemptableFn(cp.Name(), func(preemptor *api.TaskInfo, preemptees []*api.TaskInfo) ([]*api.TaskInfo, int) {
        preemptorJob := ssn.Jobs[preemptor.Job]
        var victims []*api.TaskInfo

        for _, preemptee := range preemptees {
            preempteeJob := ssn.Jobs[preemptee.Job]
            if preempteeJob.Priority < preemptorJob.Priority {
                victims = append(victims, preemptee)
            }
        }

        klog.V(4).Infof("CostAware preemptable: victims=%d", len(victims))
        return victims, util.Permit
    })
}

import (
    // ... 现有导入
    "volcano.sh/volcano/pkg/scheduler/plugins/costaware"
)

func init() {
    // ... 现有注册
    framework.RegisterPluginBuilder(costaware.PluginName, costaware.New)
}

apiVersion: v1
kind: ConfigMap
metadata:
  name: volcano-scheduler-configmap
  namespace: volcano-system
data:
  volcano-scheduler.conf: |
    actions: "enqueue, allocate, backfill, preempt"
    tiers:
    - plugins:
      - name: gang
        enableJobOrder: true
        enableJobReady: true
        enableJobPipelined: true
      - name: costaware
        enableJobOrder: true
        enableNodeOrder: true
        enablePreemptable: true
        arguments:
          costaware.weight: 2.0
    - plugins:
      - name: predicates
        enablePredicate: true
      - name: proportion
        enabledOverused: true
        enabledAllocatable: true
        enabledJobEnqueued: true

package costaware

import (
    "testing"

    "volcano.sh/volcano/pkg/scheduler/api"
    "volcano.sh/volcano/pkg/scheduler/framework"
)

func TestCostAwareJobOrder(t *testing.T) {
    arguments := framework.Arguments{
        CostWeightKey: 1.5,
    }
    plugin := New(arguments)

    if plugin.Name() != PluginName {
        t.Errorf("expected plugin name %s, got %s", PluginName, plugin.Name())
    }

    // 验证 weight 解析
    cp := plugin.(*costAwarePlugin)
    if cp.weight != 1.5 {
        t.Errorf("expected weight 1.5, got %f", cp.weight)
    }
}

func TestCostAwareNodeOrder(t *testing.T) {
    tests := []struct {
        name           string
        allocatableCPU float64
        usedCPU        float64
        allocatableMem float64
        usedMem        float64
        wantHigher     bool // true if this node should score higher than a less utilized node
    }{
        {
            name:           "highly utilized node scores higher",
            allocatableCPU: 4000,
            usedCPU:        3600,
            allocatableMem: 8000,
            usedMem:        7200,
            wantHigher:     true,
        },
        {
            name:           "empty node scores lower",
            allocatableCPU: 4000,
            usedCPU:        0,
            allocatableMem: 8000,
            usedMem:        0,
            wantHigher:     false,
        },
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            // 构建测试节点并验证评分逻辑
            usageRatio := (tt.usedCPU/tt.allocatableCPU + tt.usedMem/tt.allocatableMem) / 2
            if tt.wantHigher && usageRatio <= 0.5 {
                t.Error("expected higher utilization for higher score")
            }
        })
    }
}

type Statement struct {
    operations []operation
    ssn        *Session
}

type operation struct {
    name   Operation  // Evict | Pipeline | Allocate
    task   *api.TaskInfo
    reason string
}

stmt := ssn.Statement()

// 尝试一组调度操作
if err := stmt.Allocate(task, node); err != nil {
    // 分配失败，回滚所有操作
    stmt.Discard()
    return
}

// 检查 Job 是否就绪
if ssn.JobReady(job) {
    stmt.Commit()  // 提交所有操作
} else {
    stmt.Discard() // 回滚所有操作
}

type Event struct {
    Task *api.TaskInfo
    Err  error
}

type EventHandler struct {
    AllocateFunc   func(event *Event)
    DeallocateFunc func(event *Event)
}

func (p *myPlugin) OnSessionOpen(ssn *framework.Session) {
    ssn.AddEventHandler(&framework.EventHandler{
        AllocateFunc: func(event *framework.Event) {
            task := event.Task
            job := ssn.Jobs[task.Job]
            // 更新 Queue 已分配资源跟踪
            klog.V(4).Infof("Task <%s/%s> allocated to node %s",
                task.Namespace, task.Name, task.NodeName)
        },
        DeallocateFunc: func(event *framework.Event) {
            task := event.Task
            // 释放 Queue 已分配资源跟踪
            klog.V(4).Infof("Task <%s/%s> deallocated from node %s",
                task.Namespace, task.Name, task.NodeName)
        },
    })
}

func (p *myPlugin) OnSessionOpen(ssn *framework.Session) {
    klog.V(5).Infof("Enter %s plugin ...", p.Name())
    defer klog.V(5).Infof("Leaving %s plugin ...", p.Name())

    ssn.AddNodeOrderFn(p.Name(), func(task *api.TaskInfo, node *api.NodeInfo) (float64, error) {
        score := calculateScore(task, node)
        klog.V(4).Infof("%s score for Task %s/%s on node %s: %.2f",
            p.Name(), task.Namespace, task.Name, node.Name, score)
        return score, nil
    })
}

func TestMyPlugin(t *testing.T) {
    // 1. 构造 Arguments
    args := framework.Arguments{
        "myKey": 42,
    }

    // 2. 创建 Plugin 实例
    plugin := New(args)

    // 3. 验证 Name
    assert.Equal(t, "myPlugin", plugin.Name())

    // 4. 如需测试扩展点，构造 mock 数据
    task := &api.TaskInfo{
        Namespace: "default",
        Name:      "test-task",
        Resreq:    api.NewResource(api.ResourceList{"cpu": "1"}),
    }
    node := &api.NodeInfo{
        Name:        "node-1",
        Allocatable: api.NewResource(api.ResourceList{"cpu": "4"}),
        Used:        api.NewResource(api.ResourceList{"cpu": "2"}),
    }

    // 5. 直接测试评分函数
    // （可以直接调用导出的函数或通过创建 Session 测试）
}

// PredicateFn - 返回 error 表示节点不可用
ssn.AddPredicateFn(p.Name(), func(task *api.TaskInfo, node *api.NodeInfo) error {
    if !checkCondition(task, node) {
        return fmt.Errorf("node %s does not meet requirement for task %s/%s",
            node.Name, task.Namespace, task.Name)
    }
    return nil // nil 表示通过
})

// NodeOrderFn - 返回 error 表示评分失败（会导致节点被跳过）
ssn.AddNodeOrderFn(p.Name(), func(task *api.TaskInfo, node *api.NodeInfo) (float64, error) {
    score, err := calculateScore(task, node)
    if err != nil {
        return 0, err // 返回 error, 该节点评分为 0
    }
    return score, nil
})

// VoteFn - 使用常量返回值
ssn.AddJobPipelinedFn(p.Name(), func(obj interface{}) int {
    job, ok := obj.(*api.JobInfo)
    if !ok {
        return util.Reject  // 类型断言失败，拒绝
    }
    if meetCondition(job) {
        return util.Permit   // 允许
    }
    return util.Reject       // 拒绝
})

func New(arguments framework.Arguments) framework.Plugin {
    weight := 1.0
    arguments.GetFloat64(&weight, "myPlugin.weight")

    // 参数验证
    if weight < 0 {
        klog.Warningf("myPlugin.weight must be non-negative, got %f, using default 1.0", weight)
        weight = 1.0
    }
    if weight > 100 {
        klog.Warningf("myPlugin.weight too large: %f, clamping to 100", weight)
        weight = 100
    }

    return &myPlugin{weight: weight}
}

// 框架自动扫描 pluginsDir 目录下的 *.so 文件
func LoadCustomPlugins(pluginsDir string) error

func New(arguments framework.Arguments) framework.Plugin

- name: myPlugin
  enableNodeOrder: true     # 必须显式设置
  enableJobOrder: true