Volcano

1
2
3
4
5
6
7
8

type SchedulerCache struct {
   xxInformer	...	    // 各种informer
   ...
   Jobs                 map[schedulingapi.JobID]*schedulingapi.JobInfo
   Nodes                map[string]*schedulingapi.NodeInfo
   Queues               map[schedulingapi.QueueID]*schedulingapi.QueueInfo
   PriorityClasses      map[string]*v1beta1.PriorityClass
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

// kube-batch/pkg/scheduler/framework/session.go
type Session struct {
	UID types.UID
	cache cache.Cache
	Jobs    map[api.JobID]*api.JobInfo
	Nodes   map[string]*api.NodeInfo
	Queues  map[api.QueueID]*api.QueueInfo
	Backlog []*api.JobInfo
	Tiers   []conf.Tier
	plugins          map[string]Plugin
	eventHandlers    []*EventHandler
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

type Session struct {
	jobOrderFns      map[string]api.CompareFn	// 决定哪个训练任务优先被处理（调度、回收、抢占）。
	queueOrderFns    map[string]api.CompareFn	// 决定哪个训练队列优先被处理。
	taskOrderFns     map[string]api.CompareFn	// 决定任务中哪个容器优先被处理。
	predicateFns     map[string]api.PredicateFn		// 判断某个节点是否满足容器的基本调度要求。比如容器中指定的节点的标签。
	nodePrioritizers map[string][]priorities.PriorityConfig	//  当多个节点满足容器的调度要求时，优先选择哪个节点。
	preemptableFns   map[string]api.EvictableFn		//  决定某个容器是否可以被抢占
	reclaimableFns   map[string]api.EvictableFn		// 决定某个容器是否可以被回收
	overusedFns      map[string]api.ValidateFn		// 决定某个队列使用的资源是否超过限额，是的话不再调度对队列中的任务
	jobReadyFns      map[string]api.ValidateFn		// 判断某个任务是否已经准备好，可以调用 API Server 的接口将任务的容器调度到节点。
	jobPipelinedFns  map[string]api.ValidateFn		// 判断某个任务是否处于 Pipelined 状态
	jobValidFns      map[string]api.ValidateExFn	// 判断某个任务是否有效

}
// kube-batch/pkg/scheduler/framework/session_plugins.go
func (ssn *Session) AddJobReadyFn(name string, vf api.ValidateFn) {...}
func (ssn *Session) JobReady(obj interface{}) bool {...jobValidFns...}
func (ssn *Session) AddJobValidFn(name string, fn api.ValidateExFn) {...}
func (ssn *Session) JobValid(obj interface{}) *api.ValidateResult {...jobReadyFns...}
// kube-batch/pkg/scheduler/framework/session.go
func (ssn *Session) Allocate(task *api.TaskInfo, hostname string) error {...}
func (ssn *Session) Pipeline(task *api.TaskInfo, hostname string) error {...}
func (ssn *Session) Evict(reclaimee *api.TaskInfo, reason string) error {...}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// pkg/controllers/job/job_controller.go
// jobcontroller the Job jobcontroller type.
type jobcontroller struct {
	// queueList是所有watch到的对象，注意这里之所以是slice是为了多worker协同，每个worker一个queue
    // controller根据job的namecepace-name来进行hash后随机分配到各个queue中，其实这可能存在一定的
	queueList    []workqueue.RateLimitingInterface
    // cache是所有list到的job对象的缓存
	cache        jobcache.Cache
	// 其他成员略
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

// pkg/controllers/apis/request.go
// Request struct.
type Request struct {
	Namespace string    // job的namespace
	JobName   string    // job的name
	TaskName  string    // task的name
	QueueName string    // 分配到的Queue的name

	Event      v1alpha1.Event
	ExitCode   int32
	Action     v1alpha1.Action
	JobVersion int32
}

1
2
3
4
5
6
7
8

// pkg/controllers/cache/cache.go  
type jobCache struct {  
	sync.Mutex  
    // 保存所有的job资源，key是namespace/name  
	jobs        map[string]*apis.JobInfo  
    // 待清理的job  
	deletedJobs workqueue.RateLimitingInterface  
}  

1
2
3
4
5
6

type JobInfo struct {  
	Namespace string  
	Name      string  
	Job  *batch.Job    // 保存的Job实体  
	Pods map[string]map[string]*v1.Pod  //   
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// pkg/controllers/job/job_controller.go
func (cc *jobcontroller) processNextReq(count uint32) bool {
    // 前面部分代码略，主要是从队列中获取key信息
    // 然后从cache中获取job实体
	jobInfo, err := cc.cache.Get(jobcache.JobKeyByReq(&req))
    // 1、将job根据其job.Status.State.Phase封装成state
	st := state.NewState(jobInfo)
    // 2、根据job的policy等信息获取action
	action := applyPolicies(jobInfo.Job, &req)
    // 3、执行对应state的逻辑，里面主要是KillJob或者SyncJob
	if err := st.Execute(action); err != nil {
		// 错误处理，如果执行失败则重试，如果失败次数超过最大重试次数则重新入队列
	}
	// 处理成功则从队列中移除
	queue.Forget(req)
	return true
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Scheduler watches for new unscheduled pods for volcano. It attempts to find
// nodes that they fit on and writes bindings back to the api server.
type Scheduler struct {
	cache          schedcache.Cache   // Cache是一个接口，实现了这个接口的是SchedulerCache，所有调度资源的缓存
	actions        []framework.Action // 调度器的配置文件中设置的每一轮调度的处理逻辑，有enqueue, allocate, backfill等
	plugins        []conf.Tier        // 调度器的配置文件中设置的调度算法的各个层级的算法集合
	configurations []conf.Configuration  // 调度器每个Action对应的参数
	schedulerConf  string             // 调度器配置文件路径
	schedulePeriod time.Duration      // 调度器的调度周期，默认1s
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

actions: "enqueue, allocate, backfill"
tiers:
- plugins:
  - name: priority
  - name: gang
- plugins:
  - name: drf
  - name: predicates
  - name: proportion
  - name: nodeorder

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

// pkg/scheduler/scheduler.go
// Run runs the Scheduler
func (pc *Scheduler) Run(stopCh <-chan struct{}) {
	// list-watch调度器所需要的资源
	go pc.cache.Run(stopCh)
	pc.cache.WaitForCacheSync(stopCh)
    // 周期性执行runOnce
	go wait.Until(pc.runOnce, pc.schedulePeriod, stopCh)
}

func (pc *Scheduler) runOnce() {
	// metric代码略
    
    // 注意每次进行调度的时候都会load一次配置文件（目的是为了热更新configmap？）
	pc.loadSchedulerConf()
    // 创建一个新的Session，将cache、配置等复制到其中，并为每一个调度器注册其对应的调度算法函数
	ssn := framework.OpenSession(pc.cache, pc.plugins, pc.configurations)
	defer framework.CloseSession(ssn)
    // 根据配置文件中的action顺序执行调度操作
	for _, action := range pc.actions {
		actionStartTime := time.Now()
		action.Execute(ssn)
		metrics.UpdateActionDuration(action.Name(), metrics.Duration(actionStartTime))
	}
}