shiftzz
k8s-vgpu-scheduler调度分析
2024-05-15
技术笔记
00
请注意，本文编写于 94 天前，最后修改于 48 天前，其中某些信息可能已经过时。

目录

总览
scheduler
scheduler 配置
1.22及更早
1.22之后
managedResources
scheduler extender
filter(predicate)
计算接收的参数中每个节点的可用vGPU资源
bind
mutate(adminssion webhook)

k8s-vgpu-scheduler是由第四范式开源的基于Kubernetes平台的GPU虚拟化项目，本人在之前的文章中曾经简单介绍过。在今年，k8s-vgpu-scheduler正式改名为HAMi, 并加入了CNCF. 本文将基于HAMi对vgpu scheduler部分进行代码分析。

总览

k8s-vgpu-scheduler会起一个自定义scheduler，并在scheduler的配置中配置extender. 这个自定义的scheduler叫做4pd-scheduler, 通过KubeSchedulerConfiguration, 该schduler与gpu相关的资源绑定，当pod申请gpu相关资源时，会由4pd-scheduler进行调度。

scheduler

官方文档 对如何运行自己的scheduler进行了介绍。在vgpu-scheduler项目中，是直接运行了一个官方的kube-scheduler实例：

yaml
      containers:
        - name: kube-scheduler
          image: {{ .Values.scheduler.kubeScheduler.image }}:{{ .Values.scheduler.kubeScheduler.imageTag }}
          imagePullPolicy: {{ .Values.scheduler.kubeScheduler.imagePullPolicy | quote }}
          command:
            - kube-scheduler
            {{- if ge (.Values.scheduler.kubeScheduler.imageTag | substr 3 5| atoi) 22}}
            {{- range .Values.scheduler.kubeScheduler.extraNewArgs }}
            - {{ . }}
            {{- end }}
            {{- else }}
            - --scheduler-name={{ .Values.schedulerName }}
            {{- range .Values.scheduler.kubeScheduler.extraArgs }}
            - {{ . }}
            {{- end }}
            {{- end }}

scheduler 配置

通过配置文件，该scheduler定制成了vgpu-scheduler. 需要注意的是，在k8s 1.22之后，scheduler的配置方式发生了变化。

1.22及更早

需要加上启动参数--scheduler-name=$scheduler_name和--leader-elect=false
通过--policy-config-file指定配置文件，格式为json, 该配置文件内容称为scheduler policy.

我们查看vgpu-scheduler的scheduler policy:

json
{
    "kind": "Policy",
    "apiVersion": "v1",
    "extenders": [
        {
            "urlPrefix": "https://127.0.0.1:443",
            "filterVerb": "filter",
            "bindVerb": "bind",
            "enableHttps": true,
            "weight": 1,
            "nodeCacheCapable": true,
            "httpTimeout": 30000000000,
            "tlsConfig": {
                "insecure": true
            },
            "managedResources": [
                {
                    "name": "{{ .Values.resourceName }}",
                    "ignoredByScheduler": true
                },
                {
                    "name": "{{ .Values.resourceMem }}",
                    "ignoredByScheduler": true
                },
                {
                    "name": "{{ .Values.resourceCores }}",
                    "ignoredByScheduler": true
                },
                {
                    "name": "{{ .Values.resourceMemPercentage }}",
                    "ignoredByScheduler": true
                },
                {
                    "name": "{{ .Values.resourcePriority }}",
                    "ignoredByScheduler": true
                },
                {
                    "name": "{{ .Values.mluResourceName }}",
                    "ignoredByScheduler": true
                },
                {
                    "name": "{{ .Values.mluResourceMem }}",
                    "ignoredByScheduler": true   
                }
            ],
            "ignoreable": false
        }
    ]
}

可以看到主要是配置了extender和managedResources. extender主要用于做调度的决策，managedResources决定了哪些pod会由该scheduler调度。后面会详解这两部分。

1.22之后

通过--config=/config/config.yaml指定配置文件，格式为yaml, 文件内容为 KubeSchedulerConfiguration

关于1.22之后的 scheduler如何配置，参考官方文档。
KubeSchedulerConfiguration 的字段，可以参考API文档。

具体到vgpu-scheduler, KubeSchedulerConfiguration 内容如下：

yaml
apiVersion: kubescheduler.config.k8s.io/v1beta2
kind: KubeSchedulerConfiguration
leaderElection:
  leaderElect: false
profiles:
- schedulerName: {{ .Values.schedulerName }}
extenders:
- urlPrefix: "https://127.0.0.1:443"
  filterVerb: filter
  bindVerb: bind
  nodeCacheCapable: true
  weight: 1
  httpTimeout: 30s
  enableHTTPS: true
  tlsConfig:
    insecure: true
  managedResources:
  - name: {{ .Values.resourceName }}
    ignoredByScheduler: true
  - name: {{ .Values.resourceMem }}
    ignoredByScheduler: true
  - name: {{ .Values.resourceCores }}
    ignoredByScheduler: true
  - name: {{ .Values.resourceMemPercentage }}
    ignoredByScheduler: true
  - name: {{ .Values.resourcePriority }}
    ignoredByScheduler: true
  - name: {{ .Values.mluResourceName }}
    ignoredByScheduler: true
  - name: {{ .Values.mluResourceMem }}
    ignoredByScheduler: true

managedResources

那么，当我们创建一个GPU的pod, k8s是如何判断要使用vgpu-scheduler呢？从API文档 中 KubeSchedulerConfiguration.externders.managedResources 的描述可以看到：

managedResources
[]ExtenderManagedResource	
managedResources 是一个由此扩展模块所管理的扩展资源的列表。
如果某 Pod 请求了此列表中的至少一个扩展资源，则 Pod 会在 filter、 prioritize 和 bind （如果扩展模块可以执行绑定操作）阶段被发送到该扩展模块。
如果某资源上设置了 ignoredByScheduler 为 true，则 kube-scheduler 会在断言阶段略过对该资源的检查。

scheduler extender

k8s 的调度包含三个阶段：predicate(filter), prioritize and bind. 在 kube-scheduler 的配置中，甚至提供了12个调度阶段的插件配置 。

scheduler extender 本质上是一个 webhook, scheduler 在调度的三个阶段（还有一个preempt, 抢占式调度，本文暂不介绍）会调用 extender 的对应接口，从而得到最终的调度结果。参考这里 文中给出了一个流程图：

       kube-scheduler                                kube-scheduler-extender(HttpAPI)
+------------|-------------+                          +--------------------------+
|            |             |                          |                          |
|    +-------↓--------+    |      Pod + []Node        |    +----------------+    |
|    |                ├───────────────────────────────────>|                |    |
|    |   predicates   |    |                          |    |   predicates   |    |
|    |                |<───────────────────────────────────┤                |    |
|    +-------┬--------+    |     []Node(filtered)     |    +----------------+    |
|            |             |                          |                          |
|            |             |                          |                          |
|    +-------↓--------+    |       Pod + []Node       |    +----------------+    |
|    |                ├───────────────────────────────────>|                |    |
|    |   prioritize   |    |                          |    |   prioritize   |    |
|    |                |<───────────────────────────────────┤                |    |
|    +-------┬--------+    |      []Node(scored)      |    +----------------+    |
|            |             |                          |                          |
|            |             |                          |                          |
|    +-------↓--------+    |        Pod + Node        |    +----------------+    |
|    |                ├───────────────────────────────────>|                |    |
|    |      bind      |    |                          |    |      bind      |    |
|    |                |<───────────────────────────────────┤                |    |
|    +-------┬--------+    |        Error/nil         |    +----------------+    |
|            |             |                          |                          |
|            |             |                          |                          |
+------------|-------------+                          +--------------------------+
             ↓

scheduler extender 的官方设计文档在 这里。注意该文档提供的配置文件格式是k8s 1.23版本之前的格式。

查看vgpu-scheduler extender的代码，发现是一个http服务，在cmd/scheduler/main.go中注册了三个handler:

go
router := httprouter.New()
router.POST("/filter", routes.PredicateRoute(sher))
router.POST("/bind", routes.Bind(sher))
router.POST("/webhook", routes.WebHookRoute())

结合上面的 KubeSchedulerConfiguration, 可以知道实现了filter(即predicate)和bind两个调度过程的接口，至于preempt和prioritize则没有实现。而webhook接口主要是使用admission webhook对pod yaml进行修改。

filter(predicate)

filter阶段主要用于过滤掉pod不能运行的节点，然而在k8s-vgpu-scheduler中把prioritize, 即优选节点的工作也做了。k8s-vgpu-scheduler的filter逻辑在pkg/scheduler/scheduler.go中实现，主要包括几步：

  1. 确认pod有请求vGPU资源
  2. 计算接收的参数中每个节点的可用vGPU资源
  3. 对所有节点进行计算打分并排序
  4. 将得分最高的节点作为调度目标节点，通过annotation注入到pod yaml中
  5. 返回调度目标节点ID作为调度结果

计算接收的参数中每个节点的可用vGPU资源

该步骤由pkg/scheduler/scheduler.go中的getNodesUsage函数完成： 创建三个映射：overallnodeMap用于存储所有节点的使用情况，cachenodeMap用于存储入参节点列表中节点的使用情况，failedNodes用于存储失败节点的信息。

go
	overallnodeMap := make(map[string]*NodeUsage)
	cachenodeMap := make(map[string]*NodeUsage)
	failedNodes := make(map[string]string)

调用Scheduler的ListNodes方法获取所有节点的信息。如果出现错误，返回overallnodeMap、failedNodes和错误。

go
	allNodes, err := s.ListNodes()
	if err != nil {
		return &overallnodeMap, failedNodes, err
	}

遍历所有节点，为每个节点创建一个NodeUsage对象，并根据节点的设备信息初始化DeviceUsageList。

go
	for _, node := range allNodes {
		nodeInfo := &NodeUsage{}

如果提供了task（Pod），则从其注解中获取GPU调度策略，否则使用默认策略。

go
		userGPUPolicy := config.GPUSchedulerPolicy
		if task != nil && task.Annotations != nil {
			if value, ok := task.Annotations[policy.GPUSchedulerPolicyAnnotationKey]; ok {
				userGPUPolicy = value
			}
		}

遍历节点的设备信息，为每个设备创建一个DeviceUsage对象，并将其添加到DeviceUsageList中。

go
		nodeInfo.Devices = policy.DeviceUsageList{
			Policy:      userGPUPolicy,
			DeviceLists: make([]*policy.DeviceListsScore, 0),
		}
		for _, d := range node.Devices {
			nodeInfo.Devices.DeviceLists = append(nodeInfo.Devices.DeviceLists, &policy.DeviceListsScore{
				Score: 0,
				Device: &util.DeviceUsage{
					ID:        d.ID,
					Index:     d.Index,
					Used:      0,
					Count:     d.Count,
					Usedmem:   0,
					Totalmem:  d.Devmem,
					Totalcore: d.Devcore,
					Usedcores: 0,
					Type:      d.Type,
					Numa:      d.Numa,
					Health:    d.Health,
				},
			})
		}
		overallnodeMap[node.ID] = nodeInfo
	}

遍历Scheduler中已分配的Pods，根据Pod的节点ID查找对应的节点使用情况。 遍历Pod的设备信息，根据设备的UUID更新对应节点设备的使用情况。 使用klog记录设备的使用情况。 更新Scheduler的overviewstatus为overallnodeMap。

go
	for _, p := range s.pods {
		node, ok := overallnodeMap[p.NodeID]
		if !ok {
			continue
		}
		for _, podsingleds := range p.Devices {
			for _, ctrdevs := range podsingleds {
				for _, udevice := range ctrdevs {
					for _, d := range node.Devices.DeviceLists {
						if d.Device.ID == udevice.UUID {
							d.Device.Used++
							d.Device.Usedmem += udevice.Usedmem
							d.Device.Usedcores += udevice.Usedcores
						}
					}
				}
			}
		}
		klog.V(5).Infof("usage: pod %v assigned %v %v", p.Name, p.NodeID, p.Devices)
	}
	s.overviewstatus = overallnodeMap

遍历提供的节点列表，获取节点信息。如果获取失败，将节点ID和错误信息添加到failedNodes中。否则，将节点的使用情况添加到cachenodeMap中。 更新Scheduler的cachedstatus为cachenodeMap。 返回cachenodeMap、failedNodes和nil。

go
	for _, nodeID := range *nodes {
		node, err := s.GetNode(nodeID)
		if err != nil {
			klog.Warningf("get node %v device error, %v", nodeID, err)
			failedNodes[nodeID] = "node unregisterd"
			continue
		}
		cachenodeMap[node.ID] = overallnodeMap[node.ID]
	}
	s.cachedstatus = cachenodeMap
	return &cachenodeMap, failedNodes, nil

bind

bind函数首先将node上锁

go
	tmppatch := make(map[string]string)
	for _, val := range device.GetDevices() {
		err = val.LockNode(node, current)
		if err != nil {
			goto RelaseNodeLocks
		}
	}

然后通过annotation给pod加上allocating状态；至于何时将pod状态更新为分配成功或者失败，则是由device plugin完成的

go
	tmppatch[util.DeviceBindPhase] = "allocating"
	tmppatch[util.BindTimeAnnotations] = strconv.FormatInt(time.Now().Unix(), 10)

	err = util.PatchPodAnnotations(current, tmppatch)
	if err != nil {
		klog.ErrorS(err, "patch pod annotation failed")
	}

最后调用Bind函数将pod与node进行绑定

go
if err = s.kubeClient.CoreV1().Pods(args.PodNamespace).Bind(context.Background(), binding, metav1.CreateOptions{}); err != nil {
		klog.ErrorS(err, "Failed to bind pod", "pod", args.PodName, "namespace", args.PodNamespace, "podUID", args.PodUID, "node", args.Node)
	}
	if err == nil {
		res = &extenderv1.ExtenderBindingResult{
			Error: "",
		}
		klog.Infoln("After Binding Process")
		return res, nil
	}

当然，最后还需要将node解锁

mutate(adminssion webhook)

webhook接口主要是在pod提交创建时对pod的yaml进行修改，由于本文主要介绍scheduler的部分，adminssion webhook只做简单介绍，详细原理计划在另外一篇文章介绍。 webhook的逻辑比较简单，主要是：

  1. 针对设置了gpucore, gpumem 和 gpumempercentage 但没有设置gpu数量的pod，加上默认gpu数量的request和limit
  2. 设置pod的scheduler name, 指定后续调度由vgpu scheduler完成

本文作者:renbear

本文链接:

版权声明:本博客所有文章除特别声明外，均采用 CC BY-NC 2.0 许可协议。转载请注明出处！