Job 提交到调度的端到端链路

// pkg/controllers/job/job_controller_actions.go:285
func (cc *jobcontroller) initiateJob(job *batch.Job) (*batch.Job, error) {
    klog.V(3).Infof("Starting to initiate Job <%s/%s>", job.Namespace, job.Name)
    jobInstance, err := cc.initJobStatus(job)
    if err != nil {
        cc.recorder.Event(job, v1.EventTypeWarning, string(batch.JobStatusError),
            fmt.Sprintf("Failed to initialize job status, err: %v", err))
        return nil, err
    }

    if err := cc.pluginOnJobAdd(jobInstance); err != nil {
        // ...
    }

    newJob, err := cc.createJobIOIfNotExist(jobInstance)
    // ...

    if err := cc.createOrUpdatePodGroup(newJob); err != nil {
        // ...
    }

    return newJob, nil
}

// pkg/controllers/job/job_controller_actions.go:343
func (cc *jobcontroller) syncJob(jobInfo *apis.JobInfo, updateStatus state.UpdateStatusFn) error {
    job := jobInfo.Job
    klog.V(3).Infof("Starting to sync up Job <%s/%s>, current version %d",
        job.Namespace, job.Name, job.Status.Version)

    // ... 初始化检查 ...

    if !isInitiated(job) {
        if job, err = cc.initiateJob(job); err != nil {
            return err
        }
    }
    // ... 创建 Pod 并同步状态 ...
}

// pkg/scheduler/cache/event_handlers.go:220
func (sc *SchedulerCache) addTask(pi *schedulingapi.TaskInfo) error {
    if len(pi.NodeName) != 0 {
        if _, found := sc.Nodes[pi.NodeName]; !found {
            sc.Nodes[pi.NodeName] = schedulingapi.NewNodeInfo(nil)
            sc.Nodes[pi.NodeName].Name = pi.NodeName
        }

        node := sc.Nodes[pi.NodeName]
        if !isTerminated(pi.Status) {
            if err := node.AddTask(pi); err != nil {
                return err
            }
        }
    }

    job := sc.getOrCreateJob(pi)
    if job != nil {
        job.AddTaskInfo(pi)
    }

    return nil
}

// pkg/scheduler/cache/event_handlers.go:391
func (sc *SchedulerCache) AddPod(obj interface{}) {
    pod, ok := obj.(*v1.Pod)
    if !ok {
        klog.Errorf("Cannot convert to *v1.Pod: %v", obj)
        return
    }

    sc.Mutex.Lock()
    defer sc.Mutex.Unlock()

    err := sc.addPod(pod)
    if err != nil {
        klog.Errorf("Failed to add pod <%s/%s> into cache: %v",
            pod.Namespace, pod.Name, err)
        return
    }
    klog.V(3).Infof("Added pod <%s/%v> into cache.", pod.Namespace, pod.Name)
}

// pkg/scheduler/cache/event_handlers.go:70
func (sc *SchedulerCache) getOrCreateJob(pi *schedulingapi.TaskInfo) *schedulingapi.JobInfo {
    if len(pi.Job) == 0 {
        if !slices.Contains(sc.schedulerNames, pi.Pod.Spec.SchedulerName) {
            klog.V(4).Infof("Pod %s/%s is not scheduled by %s, skip creating PodGroup and Job for it in cache.",
                pi.Pod.Namespace, pi.Pod.Name, strings.Join(sc.schedulerNames, ","))
        }
        return nil
    }

    if _, found := sc.Jobs[pi.Job]; !found {
        sc.Jobs[pi.Job] = schedulingapi.NewJobInfo(pi.Job)
    }

    return sc.Jobs[pi.Job]
}

// pkg/scheduler/scheduler.go:91
func (pc *Scheduler) Run(stopCh <-chan struct{}) {
    pc.loadSchedulerConf()
    go pc.watchSchedulerConf(stopCh)
    pc.cache.SetMetricsConf(pc.metricsConf)
    pc.cache.Run(stopCh)
    klog.V(2).Infof("Scheduler completes Initialization and start to run")
    go wait.Until(pc.runOnce, pc.schedulePeriod, stopCh)
    // ...
}

// pkg/scheduler/scheduler.go:107
func (pc *Scheduler) runOnce() {
    klog.V(4).Infof("Start scheduling ...")
    scheduleStartTime := time.Now()
    defer klog.V(4).Infof("End scheduling ...")

    pc.mutex.Lock()
    actions := pc.actions
    plugins := pc.plugins
    configurations := pc.configurations
    pc.mutex.Unlock()

    conf.EnabledActionMap = make(map[string]bool)
    for _, action := range actions {
        conf.EnabledActionMap[action.Name()] = true
    }

    ssn := framework.OpenSession(pc.cache, plugins, configurations)
    defer func() {
        framework.CloseSession(ssn)
        metrics.UpdateE2eDuration(metrics.Duration(scheduleStartTime))
    }()

    for _, action := range actions {
        actionStartTime := time.Now()
        action.Execute(ssn)
        metrics.UpdateActionDuration(action.Name(), metrics.Duration(actionStartTime))
    }
}

// pkg/scheduler/actions/factory.go:33
func init() {
    framework.RegisterAction(reclaim.New())
    framework.RegisterAction(allocate.New())
    framework.RegisterAction(backfill.New())
    framework.RegisterAction(preempt.New())
    framework.RegisterAction(enqueue.New())
    framework.RegisterAction(shuffle.New())
}

// 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 {
                if e := s.unallocate(op.task); e != nil {
                    klog.Errorf("Failed to unallocate task <%v/%v>: %v.",
                        op.task.Namespace, op.task.Name, e)
                }
            }
        }
    }
}

// pkg/scheduler/cache/cache.go:1286
func (sc *SchedulerCache) AddBindTask(bindContext *BindContext) error {
    klog.V(5).Infof("add bind task %v/%v",
        bindContext.TaskInfo.Namespace, bindContext.TaskInfo.Name)
    sc.Mutex.Lock()
    defer sc.Mutex.Unlock()
    job, task, err := sc.findJobAndTask(bindContext.TaskInfo)
    // ...
    originalStatus := task.Status
    if err := job.UpdateTaskStatus(task, schedulingapi.Binding); err != nil {
        return err
    }
    // ...
    if err := node.AddTask(task); err != nil {
        // rollback ...
    }
    sc.BindFlowChannel <- bindContext
    return nil
}

// pkg/scheduler/cache/cache.go:1330
func (sc *SchedulerCache) processBindTask() {
    for {
        select {
        case bindContext, ok := <-sc.BindFlowChannel:
            if !ok {
                return
            }
            sc.bindCache = append(sc.bindCache, bindContext)
            if len(sc.bindCache) == sc.batchNum {
                sc.BindTask()
            }
        default:
        }
        if len(sc.BindFlowChannel) == 0 {
            break
        }
    }
    if len(sc.bindCache) == 0 {
        return
    }
    sc.BindTask()
}

// pkg/scheduler/cache/cache.go:218
func (db *DefaultBinder) Bind(kubeClient kubernetes.Interface, tasks []*schedulingapi.TaskInfo) map[schedulingapi.TaskID]string {
    errMsg := make(map[schedulingapi.TaskID]string)
    for _, task := range tasks {
        p := task.Pod
        if err := db.kubeclient.CoreV1().Pods(p.Namespace).Bind(context.TODO(),
            &v1.Binding{
                ObjectMeta: metav1.ObjectMeta{
                    Namespace: p.Namespace, Name: p.Name, UID: p.UID,
                    Annotations: p.Annotations,
                },
                Target: v1.ObjectReference{Kind: "Node", Name: task.NodeName},
            },
            metav1.CreateOptions{}); err != nil {
            klog.Errorf("Failed to bind pod <%v/%v> to node %s : %#v",
                p.Namespace, p.Name, task.NodeName, err)
            errMsg[task.UID] = err.Error()
        } else {
            metrics.UpdateTaskScheduleDuration(metrics.Duration(p.CreationTimestamp.Time))
        }
    }
    return errMsg
}

# volcano-scheduler deployment
spec:
  containers:
  - name: volcano-scheduler
    args:
    - --logtostderr
    - --v=4                    # 推荐 debug 时使用 V4
    - -2>&1