Allocate Action 调试指南

// pkg/scheduler/actions/allocate/allocate.go:119
func (alloc *Action) Execute(ssn *framework.Session) {
    klog.V(5).Infof("Enter Allocate ...")
    defer klog.V(5).Infof("Leaving Allocate ...")

    alloc.parseArguments(ssn)

    // 注释说明了分配的 5 个步骤:
    // 1. pick a queue named Q (using ssn.QueueOrderFn)
    // 2. pick a job named J from Q (using ssn.JobOrderFn)
    // 3. pick a task T from J (using ssn.TaskOrderFn)
    // 4. use predicateFn to filter out node that T can not be allocated on.
    // 5. use ssn.NodeOrderFn to judge the best node and assign it to T

    alloc.session = ssn
    alloc.recorder = NewRecorder()
    actx := alloc.buildAllocateContext()
    klog.V(3).Infof("Try to allocate resource to %d Queues", actx.queues.Len())
    alloc.allocateResources(actx)
}

// pkg/scheduler/actions/allocate/allocate.go:288
if ssn.Overused(queue) {
    klog.V(3).Infof("Queue <%s> is overused, ignore it.", queue.Name)
    continue
}

// pkg/scheduler/actions/allocate/allocate.go:339
// Put back the queue to priority queue after job's resource allocating finished
queues.Push(queue)

// pkg/scheduler/actions/allocate/allocate.go:139
func (alloc *Action) buildAllocateContext() *allocateContext {
    ssn := alloc.session
    actx := &allocateContext{
        queues:      util.NewPriorityQueue(ssn.QueueOrderFn),
        jobsByQueue: make(map[api.QueueID]*util.PriorityQueue),
        // ...
    }

    for _, job := range ssn.Jobs {
        // 过滤 Pending Job（如果 enqueue 启用）
        if job.IsPending() {
            if conf.EnabledActionMap["enqueue"] {
                continue
            } else {
                job.PodGroup.Status.Phase = scheduling.PodGroupInqueue
            }
        }

        // JobValid 校验
        if vr := ssn.JobValid(job); vr != nil && !vr.Pass {
            continue
        }

        // Queue 存在性检查
        if _, found := ssn.Queues[job.Queue]; !found {
            continue
        }

        // HyperNode 就绪检查
        if !ssn.HyperNodesReadyToSchedule && job.ContainsNetworkTopology() {
            continue
        }

        // 组织 Job Worksheet
        worksheet := alloc.organizeJobWorksheet(job)
        if worksheet.Empty() { continue }

        // 加入 Queue 对应的 Job 优先队列
        actx.jobsByQueue[job.Queue].Push(job)
    }
    return actx
}

// pkg/scheduler/actions/allocate/allocate.go:205
func (alloc *Action) organizeJobWorksheet(job *api.JobInfo) *JobWorksheet {
    // 1. 过滤已 Ready 的 SubJob
    // 2. 按 SubJobOrderFn 排序
    // 3. 找出满足最小要求的 SubJob 集合
    // 4. 将 Pending Task 加入 SubJobWorksheet
    //    - 跳过 SchGated 的 Task
    //    - 跳过 BestEffort Task (资源请求为空)
}

// pkg/scheduler/actions/allocate/allocate.go:586
if err := ssn.PrePredicateFn(task); err != nil {
    klog.V(3).Infof("PrePredicate for task %s/%s failed for: %v",
        task.Namespace, task.Name, err)
    fitErrors := api.NewFitErrors()
    for _, ni := range nodes {
        fitErrors.SetNodeError(ni.Name, err)
    }
    job.NodesFitErrors[task.UID] = fitErrors
    break
}

// pkg/scheduler/actions/allocate/allocate.go:815
func (alloc *Action) predicate(task *api.TaskInfo, node *api.NodeInfo) error {
    // 资源检查
    var statusSets api.StatusSets
    if ok, resources := task.InitResreq.LessEqualWithResourcesName(
        node.FutureIdle(), api.Zero); !ok {
        statusSets = append(statusSets,
            &api.Status{Code: api.Unschedulable,
                Reason: api.WrapInsufficientResourceReason(resources)})
        return api.NewFitErrWithStatus(task, node, statusSets...)
    }
    return alloc.session.PredicateForAllocateAction(task, node)
}

// pkg/scheduler/framework/session.go:698
func (ssn *Session) PredicateForAllocateAction(task *api.TaskInfo, node *api.NodeInfo) error {
    err := ssn.PredicateFn(task, node)
    if err == nil {
        return nil
    }

    fitError, ok := err.(*api.FitError)
    if !ok {
        return api.NewFitError(task, node, err.Error())
    }

    statusSets := fitError.Status
    if statusSets.ContainsUnschedulable() ||
       statusSets.ContainsUnschedulableAndUnresolvable() ||
       statusSets.ContainsErrorSkipOrWait() {
        return fitError
    }
    return nil
}

// pkg/scheduler/framework/statement.go:262
func (s *Statement) Allocate(task *api.TaskInfo, nodeInfo *api.NodeInfo) (err error) {
    hostname := nodeInfo.Name
    task.Pod.Spec.NodeName = hostname

    // 1. 更新 Session 中 task 状态为 Allocated
    job, found := s.ssn.Jobs[task.Job]
    if found {
        if err := job.UpdateTaskStatus(task, api.Allocated); err != nil {
            // ...
        }
    }

    // 2. 将 task 添加到 Node
    task.NodeName = hostname
    if node, found := s.ssn.Nodes[hostname]; found {
        if err := node.AddTask(task); err != nil {
            // ...
        }
    }

    // 3. 触发 AllocateFunc 回调
    for _, eh := range s.ssn.eventHandlers {
        if eh.AllocateFunc != nil {
            eh.AllocateFunc(&Event{Task: task})
        }
    }

    // 4. 记录 operation
    s.operations = append(s.operations, operation{
        name: Allocate,
        task: task,
    })
    return nil
}

// pkg/scheduler/framework/statement.go:418
func (s *Statement) Commit() {
    klog.V(3).Info("Committing operations ...")
    for _, op := range s.operations {
        op.task.ClearLastTxContext()
        switch op.name {
        case Evict:
            err := s.evict(op.task, op.reason)
        case Pipeline:
            s.pipeline(op.task)
        case Allocate:
            err := s.allocate(op.task)
            if err != nil {
                // 失败则回滚
                s.unallocate(op.task)
            }
        }
    }
}

// pkg/scheduler/framework/statement.go:392
func (s *Statement) Discard() {
    klog.V(3).Info("Discarding operations ...")
    // 逆序回滚所有操作
    for i := len(s.operations) - 1; i >= 0; i-- {
        op := s.operations[i]
        op.task.GenerateLastTxContext()
        switch op.name {
        case Evict:
            s.unevict(op.task)
        case Pipeline:
            s.UnPipeline(op.task)
        case Allocate:
            s.unallocate(op.task)
        }
    }
}