volcano源码阅读——action/enqueue

enqueue

将pending状态的job的PodGroup状态设置为inqueue，当会话关闭的时候会更新PodGroup状态。PodGroup状态变为inqueue后，controller会为其创建Pod。

 

pkg\scheduler\actions\enqueue\enqueue.go

func (enqueue *Action) Execute(ssn *framework.Session) {
    klog.V(5).Infof("Enter Enqueue ...")
    defer klog.V(5).Infof("Leaving Enqueue ...")

    queues := util.NewPriorityQueue(ssn.QueueOrderFn) // 临时队列
    queueSet := sets.NewString()                      // 临时集合
    jobsMap := map[api.QueueID]*util.PriorityQueue{}  // 临时字典：queueID与job队列对应关系

    for _, job := range ssn.Jobs {
        // 检查调度开始时间戳，如果是零则为其设置当前时间
        if job.ScheduleStartTimestamp.IsZero() {
            ssn.Jobs[job.UID].ScheduleStartTimestamp = metav1.Time{
                Time: time.Now(),
            }
        }
        if queue, found := ssn.Queues[job.Queue]; !found {
            klog.Errorf("Failed to find Queue <%s> for Job <%s/%s>",
                job.Queue, job.Namespace, job.Name)
            continue
        } else if !queueSet.Has(string(queue.UID)) {
            klog.V(5).Infof("Added Queue <%s> for Job <%s/%s>",
                queue.Name, job.Namespace, job.Name)

            queueSet.Insert(string(queue.UID))
            queues.Push(queue)
        }

        // 检查job的状态是不是pending，如果是则将其放到jobsMap的队列里
        if job.IsPending() {
            if _, found := jobsMap[job.Queue]; !found {
                jobsMap[job.Queue] = util.NewPriorityQueue(ssn.JobOrderFn)
            }
            klog.V(5).Infof("Added Job <%s/%s> into Queue <%s>", job.Namespace, job.Name, job.Queue)
            jobsMap[job.Queue].Push(job)
        }
    }

    klog.V(3).Infof("Try to enqueue PodGroup to %d Queues", len(jobsMap))

    // 遍历PodGroup，将PodGroup状态改为inqueue，并在ssn.Jobs里记录job
    for {
        if queues.Empty() {
            break
        }

        queue := queues.Pop().(*api.QueueInfo)

        // skip the Queue that has no pending job
        jobs, found := jobsMap[queue.UID]
        if !found || jobs.Empty() { // 如果当前队列不存在或者job为空，则遍历下个队列
            continue
        }
        job := jobs.Pop().(*api.JobInfo)

        // 如果未设置最小资源或判定job可以入队，则对job进行处理：
        // 1、对job执行插件的JobEnqueue处理
        // 2、将PodGroup的状态改为inqueue
        // 3、将job记录到ssn上
        if job.PodGroup.Spec.MinResources == nil || ssn.JobEnqueueable(job) {
            ssn.JobEnqueued(job)
            job.PodGroup.Status.Phase = scheduling.PodGroupInqueue
            ssn.Jobs[job.UID] = job
        }

        // Added Queue back until no job in Queue.
        queues.Push(queue)
    }
}

 

调度周期完毕后，关闭会话时会更信PodGroup

func closeSession(ssn *Session) {
    ju := NewJobUpdater(ssn)
    ju.UpdateAll() // 更新所有

    updateQueueStatus(ssn)

    ssn.Jobs = nil
    ssn.Nodes = nil
    ssn.RevocableNodes = nil
    ssn.plugins = nil
    ssn.eventHandlers = nil
    ssn.jobOrderFns = nil
    ssn.queueOrderFns = nil
    ssn.clusterOrderFns = nil
    ssn.NodeList = nil
    ssn.TotalResource = nil

    klog.V(3).Infof("Close Session %v", ssn.UID)
}

// updateJob update specified job
func (ju *JobUpdater) updateJob(index int) {
    job := ju.jobQueue[index]
    ssn := ju.ssn

    job.PodGroup.Status = jobStatus(ssn, job)
    oldStatus, found := ssn.PodGroupOldState.Status[job.UID]
    updatePGStatus := !found || isPodGroupStatusUpdated(job.PodGroup.Status, oldStatus)
    updatePGAnnotations := ju.isJobAllocatedHyperNodeChanged(job)
    if _, err := ssn.cache.UpdateJobStatus(job, updatePGStatus, updatePGAnnotations); err != nil {
        klog.Errorf("Failed to update job <%s/%s>: %v",
            job.Namespace, job.Name, err)
    }
}

// UpdateJobStatus update the status of job and its tasks.
func (sc *SchedulerCache) UpdateJobStatus(job *schedulingapi.JobInfo, updatePGStatus, updatePGAnnotations bool) (*schedulingapi.JobInfo, error) {
    if updatePGStatus || updatePGAnnotations {
        if updatePGAnnotations {
            sc.updateJobAnnotations(job)
        }
        pg, err := sc.StatusUpdater.UpdatePodGroup(job.PodGroup)
        if err != nil {
            return nil, err
        }
        job.PodGroup = pg
    }
    sc.RecordJobStatusEvent(job, updatePGStatus)

    return job, nil
}