K8s 自定义调度器 Part2：通过 Scheduler Framework 实现自定义调度逻辑

意琦行收录于 Kubernetes

2024-12-03 22:00:00 约 7300 字预计阅读 15 分钟 - 次阅读

本文主要分享如何通过 Scheduler Framework 实现自定义调度策略。

1. 为什么需要自定义调度逻辑

什么是所谓的调度?

所谓调度就是指给 Pod 对象的 spec.nodeName 赋值
待调度对象则是所有 spec.nodeName 为空的 Pod
调度过程则是从集群现有的 Node 中为当前 Pod 选择一个最合适的

实际上 Pod 上还有一个平时比较少关注的属性： spec.schedulerName，用于指定该 Pod 要交给哪个调度器进行调度。

那么问题来了，平时用的时候也没给 spec.schedulerName 赋值过，怎么也能调度呢？

因为默认的 kube-scheduler 可以兼容 spec.schedulerName 为空或者为 default 的 Pod。

为什么需要自定义调度逻辑

自定义调度逻辑可以解决特定应用场景和需求，使集群资源使用更高效，适应特殊的调度策略。

比如：

不同的工作负载可能有特定的资源需求，比如 GPU 或 NPU，需要确保 Pod 只能调度到满足这些资源条件的节点上。
某些集群可能需要均衡资源消耗，避免将多个负载集中到某些节点上。
为了降低延迟，可能需要将Pod调度到特定地理位置的节点上。自定义调度器可以根据节点的地理位置标签进行调度决策。
某些应用需要与其他应用隔离运行，以避免资源争抢。通过自定义调度器，可以将特定类型的任务或工作负载隔离到专用的节点上。
…

总之就是业务上有各种特殊的调度需求，因此我们需要通过实现自定义调度器来满足这些需求。

通过实现自定义调度器，可以根据具体的业务需求和集群环境，实现更灵活、更高效的资源管理和调度策略。

2.如何增加自定义调度逻辑

自定义调度器的几种方法

要增加自定义调度逻辑也并不复杂，K8s 整个调度流程都已经插件化了，我们并不需要重头开始实现一个调度器，而只需要实现一个调度插件，通过在调度过程中各个阶段加入我们的自定义逻辑，来控制最终的调度结果。

总体来说可以分为以下几个方向：

1）新增一个调度器

~~直接修改 kube-scheduler 源码，编译替换~~
使用调度框架（Scheduling Framework），我们可以使用 scheduler-plugins 作为模板，简化自定义调度器的开发流程。
- Kubernetes v1.15 版本中引入了可插拔架构的调度框架，使得定制调度器这个任务变得更加的容易。调库框架向现有的调度器中添加了一组插件化的 API，该 API 在保持调度程序“核心”简单且易于维护的同时，使得大部分的调度功能以插件的形式存在。

2）扩展原有调度器

通过 Scheduler Extender 可以实现对已有调度器进行扩展。单独创建一个 HTTP 服务并实现对应接口，后续就可以将该服务作为外置调度器使用。通过配置 KubeSchedulerConfiguration原 Scheduler 会以 HTTP 调用方式和外置调度器交互，实现在不改动原有调度器基础上增加自定义逻辑。

3）其他非主流方案

自定义 Webhook 直接修改未调度 Pod 的 spec.nodeName 字段，有点离谱但理论可行哈哈

二者都有自己的优缺点

	优点	缺点
新增调度器	性能好：由于不依赖外部插件或 HTTP 调用，调度流程的延迟相对较低，适合对性能要求较高的场景复用性高：可复用现有的调度插件，如 `scheduler-plugins`，大大降低了开发难度，提升了开发效率。	多个调度器可能会冲突：比如多个调度器同时调度了一个 Pod 到节点上，先启动的 Pod 把资源占用了，后续的 Pod 无法启动。
扩展调度器	实现简单：无需重新编译调度器，通过配置 `KubeSchedulerConfiguration` 创建一个外部 HTTP 服务来实现自定义逻辑。零侵入性：不需要修改或重构调度器的核心代码，可快速上线新的调度逻辑。灵活性较高：原有调度器和自定义逻辑相对独立，方便维护与测试。	性能差：调度请求需要经过 HTTP 调用，增加了调用延迟，对性能可能有影响。

一般在我们要改动的逻辑不多时，直接使用 Scheduler Extender 是比较简单的。

如果要实现多种调度策略，那么 Scheduler Framework 则是更好的选择。

Scheduler 配置

调度器的配置有一个单独的对象：KubeSchedulerConfiguration，不过并没有以 CRD 形式存在，而是存放到 Configmap 中的。

以下是一个完整 KubeSchedulerConfiguration 的 yaml：

 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
26
apiVersion: v1
data:
  config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1beta2
    kind: KubeSchedulerConfiguration
    leaderElection:
      leaderElect: false
    profiles:
    - schedulerName: hami-scheduler
    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: nvidia.com/gpu
        ignoredByScheduler: true
      - name: nvidia.com/gpumem
        ignoredByScheduler: true
      - name: nvidia.com/gpucores
        ignoredByScheduler: true

一般分为基础配置和 extenders 配置两部分。

基础配置

基础配置一般就是配置调度器的名称，Demo 如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
apiVersion: v1
kind: ConfigMap
metadata:
  name: my-scheduler-config
  namespace: kube-system
data:
  my-scheduler-config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1beta2
    kind: KubeSchedulerConfiguration
    profiles:
      - schedulerName: my-scheduler
    leaderElection:
      leaderElect: false   

通过 schedulerName 来指定该调度器的名称，比如这里就是 my-scheduler 。

创建 Pod 时除非手动指定 spec.schedulerName 为 my-scheduler，否则不会由该调度器进行调度。

扩展调度器：extenders 配置

extenders 部分通过额外指定一个 http 服务器来实现外置的自定义的调度逻辑。

一个简单的 Scheduler Extender 配置如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
apiVersion: v1
kind: ConfigMap
metadata:
  name: i-scheduler-extender
  namespace: kube-system
data:
  i-scheduler-extender.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1
    kind: KubeSchedulerConfiguration
    profiles:
      - schedulerName: i-scheduler-extender
    leaderElection:
      leaderElect: false
    extenders:
    - urlPrefix: "http://localhost:8080"
      enableHTTPS: false
      filterVerb: "filter"
      prioritizeVerb: "prioritize"
      bindVerb: "bind"
      weight: 1
      nodeCacheCapable: true

核心部分为

1
2
3
4
5
6
7
8
    extenders:
    - urlPrefix: "http://localhost:8080"
      enableHTTPS: false
      filterVerb: "filter"
      prioritizeVerb: "prioritize"
      bindVerb: "bind"
      weight: 1
      nodeCacheCapable: true

几个核心参数含义如下：

urlPrefix: http://127.0.0.1:8080 用于指定外置的调度服务访问地址
filterVerb: "filter"：表示 Filter 接口在外置服务中的访问地址为 filter，即完整地址为 http://127.0.0.1:8080/filter
prioritizeVerb: "prioritize"：同上，Prioritize(Score) 接口的地址为 prioritize
bindVerb: "bind"：同上，Bind 接口的地址为 bind

这样该调度器在执行 Filter 接口逻辑时，除了内置的调度器插件之外，还会通过 HTTP 方式调用外置的调度器。

这样我们只需要创建一个 HTTP 服务，实现对应接口即可实现自定义的调度逻辑，而不需要重头实现一个调度器。

ManagedResources 配置

在之前的配置中是所有 Pod 都会走 Extender 的调度逻辑，实际上 Extender 还有一个 ManagedResources 配置，用于限制只有申请使用指定资源的 Pod 才会走 Extender 调度逻辑，这样可以减少无意义的调度。

一个带 managedResources 的 KubeSchedulerConfiguration 内容如下

 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
26
apiVersion: v1
data:
  config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1
    kind: KubeSchedulerConfiguration
    leaderElection:
      leaderElect: false
    profiles:
    - schedulerName: hami-scheduler
    extenders:
    - urlPrefix: "https://127.0.0.1:443"
      filterVerb: filter
      bindVerb: bind
      nodeCacheCapable: false
      enableHTTPS: false
      managedResources:
      - name: nvidia.com/gpu
        ignoredByScheduler: true
      - name: nvidia.com/gpumem
        ignoredByScheduler: true
      - name: nvidia.com/gpucores
        ignoredByScheduler: true
      - name: nvidia.com/gpumem-percentage
        ignoredByScheduler: true
      - name: nvidia.com/priority
        ignoredByScheduler: true

在原来的基础上增加了 managedResources 部分的配置

1
2
3
4
5
6
7
8
9
      managedResources:
      - name: nvidia.com/gpu
        ignoredByScheduler: true
      - name: nvidia.com/gpumem
        ignoredByScheduler: true
      - name: nvidia.com/gpucores
        ignoredByScheduler: true
      - name: nvidia.com/gpumem-percentage
        ignoredByScheduler: true

只有 Pod 申请这些特殊资源时才走 Extender 调度逻辑，否则使用原生的 Scheduler 调度即可。

ignoredByScheduler: true 的作用是告诉调度器忽略指定资源，避免将它们作为调度决策的依据。也就是说，虽然这些资源（如 GPU 或其他加速硬件）会被 Pod 请求，但调度器不会在选择节点时基于这些资源的可用性做出决定。

ps：因为这些资源可能是虚拟的，并不会真正的出现在 Node 上，因此调度时需要忽略掉，否则就没有任何节点满足条件了,但是这些虚拟资源则是我们的自定义调度逻辑需要考虑的事情。

Scheduler 中的判断逻辑如下，和之前说的一样，只有当 Pod 申请了这些指定的资源时，Scheduler 才会调用 Extender。

 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
26
27
28
29
30
31
// IsInterested returns true if at least one extended resource requested by
// this pod is managed by this extender.
func (h *HTTPExtender) IsInterested(pod *v1.Pod) bool {
    if h.managedResources.Len() == 0 {
       return true
    }
    if h.hasManagedResources(pod.Spec.Containers) {
       return true
    }
    if h.hasManagedResources(pod.Spec.InitContainers) {
       return true
    }
    return false
}

func (h *HTTPExtender) hasManagedResources(containers []v1.Container) bool {
    for i := range containers {
       container := &containers[i]
       for resourceName := range container.Resources.Requests {
          if h.managedResources.Has(string(resourceName)) {
             return true
          }
       }
       for resourceName := range container.Resources.Limits {
          if h.managedResources.Has(string(resourceName)) {
             return true
          }
       }
    }
    return false
}

3. scheduler framework 扩展点

如下图所示，K8s 调度框架定义了一些扩展点（extension points）:

图源：scheduler-framework-extension-points

扩展点根据是否影响调度决策，可以分为两类：

影响调度决策的扩展点
- 这些插件实现的返回值中包括一个成功/失败字段，决定了是允许还是拒绝这个 pod 进入下一处理阶段；
- 任何一个扩展点失败了，这个 pod 的调度就失败了；
不影响调度决策的扩展点
- 这些插件的实现没有返回值，一般只用于打印信息，更新 Pod、Node 信息，或执行清理操作

每一个扩展点都可以对应多个插件，各自可以有不同的实现，每个扩展点的作用如下：

PreEnqueue：这些插件在将 Pod 被添加到内部活动队列之前被调用，在此队列中 Pod 被标记为准备好进行调度。
- 一般用于避免不满足调度要求的 Pod 进入调度阶段，从而降低调度工作负担。
- 该插件如果返回失败，那么 Pod 不会进入活动队列，也就不会被调度。一般用来做一些预检测，当准备好时才让 Pod 进入调度队列，否则肯定会因为资源未准备好而调度失败，也就没必要进行调度了。
- 例如：如果调度所需的 GPU 资源尚未准备好，可以通过 PreEnqueue 插件阻止相关 Pod 进入队列。
PreFilter：这些插件用于预处理 Pod 的相关信息，或者检查集群或 Pod 必须满足的某些条件。如果 PreFilter 插件返回错误，则调度周期将终止。
Filter：这些插件用于过滤出不能运行该 Pod 的节点。对于每个节点，调度器将按照其配置顺序调用这些过滤插件。如果任何过滤插件将节点标记为不可行，则不会为该节点调用剩下的过滤插件。节点可以被同时进行评估。
- 其主要作用是过滤掉不符合要求的节点，从而缩小可调度节点的范围
PostFilter：这些插件在 Filter 阶段后调用，但仅在该 Pod 没有可行的节点时调用。
- 只有在 Filter 插件没有返回可用的节点时才会调用 PostFilter 用于补救，以便为当前 Pod 找到或创造一个可行的节点。
- 典型实现是 preemption 插件，它试图通过抢占其他 Pod 来使 Pod 可访问。
PreScore：这些插件用于执行“前置评分（pre-scoring）”工作，即生成一个可共享状态供 Score 插件使用。如果 PreScore 插件返回错误，则调度周期将终止。
Score：这些插件用于对通过过滤阶段的节点打分并进行排序，帮助调度器选择最适合的节点，而不仅仅是选一个满足条件的节点。
- 调度器将为每个节点调用每个评分插件。将有一个定义明确的整数范围，代表最小和最大分数。在 NormalizeScore 阶段之后，调度器将根据配置的插件权重合并所有插件的节点分数。
- 例如：负载均衡：根据 CPU 或内存利用率进行评分，确保 Pod 分布均匀，避免资源集中于少数节点，或者按网络延迟、区域分布，将 Pod 调度到与数据位置接近的节点。
NormalizeScore：这些插件用于在调度器计算 Node 排名之前修改分数，对 Score 进行归一化处理，将得分映射到调度器规定的 0~100 范围内，便于对比。
Reserve：Reserve 阶段是 Kubernetes 调度框架中的一个关键阶段，主要用于在将 Pod 实际绑定到节点之前，临时为其保留必要的资源。这一设计旨在解决调度器可能遇到的资源竞争问题，保证资源在绑定之前不会被其他 Pod 抢占。
- 实现了 Reserve 接口的插件，拥有两个方法，即 Reserve 和 Unreserve，分别用于保留资源和释放资源。
- 如果任意一个 Reserve 方法调用失败，后面的插件就不会被执行，Reserve 阶段被认为失败。如果所有插件的 Reserve 方法都成功了，Reserve 阶段就被认为是成功的，剩下的调度周期和绑定周期就会被执行。
- 如果 Reserve 阶段或后续阶段失败了，则触发 Unreserve 阶段。发生这种情况时，所有 Reserve 插件的 Unreserve 方法将按照 Reserve 方法调用的相反顺序执行。这个阶段的存在是为了清理与保留的 Pod 相关的状态。
Permit：Permit 插件能够决定一个 Pod 是否可以被绑定到指定节点，这种控制是在调度周期的最后阶段进行的。调度器在此阶段会调用所有注册的 Permit 插件，以确认 Pod 的绑定是否符合某些条件，一般用于阻止或者延迟 Pod 的调度，每个 Permit 插件都可以做以下三件事：
- 批准：一旦所有 Permit 插件批准 Pod 后，该 Pod 将被发送以进行绑定。
- 拒绝：如果任何 Permit 插件拒绝 Pod，则该 Pod 将被返回到调度队列。这将触发 Reserve 插件中的 Unreserve 阶段。
- 等待（带有超时）：如果一个 Permit 插件返回“等待”结果，则 Pod 将保持在一个内部的“等待中” 的 Pod 列表，同时该 Pod 的绑定周期启动时即直接阻塞直到得到批准。如果超时发生，等待变成拒绝，并且 Pod 将返回调度队列，从而触发 Reserve 插件中的 Unreserve 阶段。
PreBind：这些插件用于执行 Pod 绑定前所需的所有工作。例如，一个 PreBind 插件可能需要制备网络卷并且在允许 Pod 运行在该节点之前将其挂载到目标节点上。如果任何 PreBind 插件返回错误，则 Pod 将被拒绝并且退回到调度队列中。
Bind：Bind 插件用于将 Pod 绑定到节点上。直到所有的 PreBind 插件都完成，Bind 插件才会被调用。各 Bind 插件按照配置顺序被调用。Bind 插件可以选择是否处理指定的 Pod。如果某 Bind 插件选择处理某 Pod，剩余的 Bind 插件将被跳过。
- 真正执行调度的一个插件，更新 Pod 的 spec.nodeName 字段
PostBind：这是个信息传递性质的接口。 PostBind 插件在 Pod 成功绑定后被调用。这是绑定周期的结尾，可用于清理相关的资源。

4. 使用 Scheduler Framework 开发调度器

基于scheduler framework开发插件，本质上是实现接口。

我们实现一个简单的调度器，将 Pod 调度到带有指定 label 的节点上。

完整代码见：lixd/i-scheduler

初始化项目

1
2
3
4
5
6
mkdri i-scheduler
cd i-scheduler
#go mod init github.com/lixd96/i-scheduler

git clone -b release-1.29 https://github.com/kubernetes-sigs/scheduler-plugins.git
cd scheduler-plugins

创建自定义调度器

scheduler-plugins/pkg 目录下每一个目录都是一个调度插件，我们也创建一个 priority 目录来实现自己的调度逻辑。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
/i-scheduler/scheduler-plugins# tree pkg -L 1     
├── capacityscheduling
├── controllers
├── coscheduling
├── crossnodepreemption
├── generated
├── networkaware
├── noderesources
├── noderesourcetopology
├── podstate
├── preemptiontoleration
├── pripority
├── qos
├── sysched
├── trimaran
└── util

pripority 目录下创建文件 pripority.go，内容如下：

 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
package priority

import (
    "context"
    "fmt"
    "k8s.io/kubernetes/pkg/scheduler/framework/plugins/helper"
    "log"
    "strconv"

    v1 "k8s.io/api/core/v1"
    "k8s.io/apimachinery/pkg/runtime"
    "k8s.io/kubernetes/pkg/scheduler/framework"
)

// Name is the name of the plugin used in the plugin registry and configurations.
const Name = "priority"

const Label = "priority.lixueduan.com"

type Priority struct {
    handle framework.Handle
}

var _ framework.FilterPlugin = &Priority{}
var _ framework.ScorePlugin = &Priority{}

// New initializes a new plugin and returns it.
func New(_ context.Context, _ runtime.Object, h framework.Handle) (framework.Plugin, error) {
    return &Priority{handle: h}, nil
}

// Name returns name of the plugin.
func (pl *Priority) Name() string {
    return Name
}

func (pl *Priority) Filter(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
    log.Printf("filter pod: %v, node: %v", pod.Name, nodeInfo)
    log.Println(state)

    // 只调度到携带指定 Label 的节点上
    if _, ok := nodeInfo.Node().Labels[Label]; !ok {
       return framework.NewStatus(framework.Unschedulable, fmt.Sprintf("Node:%s does not have label %s", "Node: "+nodeInfo.Node().Name, Label))
    }
    return framework.NewStatus(framework.Success, "Node: "+nodeInfo.Node().Name)
}

// Score invoked at the score extension point.
func (pl *Priority) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    if err != nil {
       return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
    }

    // 获取 Node 上的 Label 作为分数
    priorityStr, ok := nodeInfo.Node().Labels[Label]
    if !ok {
       return 0, framework.NewStatus(framework.Error, fmt.Sprintf("node %q does not have label %s", nodeName, Label))
    }

    priority, err := strconv.Atoi(priorityStr)
    if err != nil {
       return 0, framework.NewStatus(framework.Error, fmt.Sprintf("node %q has priority %s are invalid", nodeName, priorityStr))
    }

    return int64(priority), framework.NewStatus(framework.Success, "")
}

// ScoreExtensions of the Score plugin.
func (pl *Priority) ScoreExtensions() framework.ScoreExtensions {
    return pl
}

// NormalizeScore invoked after scoring all nodes.
func (pl *Priority) NormalizeScore(ctx context.Context, state *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
    return helper.DefaultNormalizeScore(framework.MaxNodeScore, false, scores)
}

代码不多，实现也很简单，实现了framework.FilterPlugin 和framework.ScorePlugin 接口，也就是实现了 Filter 和 Score 两个扩展点。

Filter 逻辑也是简单粗暴：

1
2
3
4
5
6
7
8
9
func (pl *Priority) Filter(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
    log.Printf("filter pod: %v, node: %v", pod.Name, nodeInfo)
    log.Println(state)

    // 只调度到携带指定 Label 的节点上
    if _, ok := nodeInfo.Node().Labels[Label]; !ok {
       return framework.NewStatus(framework.Unschedulable, fmt.Sprintf("Node:%s does not have label %s", "Node: "+nodeInfo.Node().Name, Label))
    }
    return framework.NewStatus(framework.Success, "Node: "+nodeInfo.Node().Name)

如果 Node 上没有指定 Label 就返回 Unschedulable，有则返回 Success。

Score 逻辑同样很简单,取 Node 上的 Label 作为得分：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
func (pl *Priority) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
    nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
    if err != nil {
       return 0, framework.NewStatus(framework.Error, fmt.Sprintf("getting node %q from Snapshot: %v", nodeName, err))
    }

    // 获取 Node 上的 Label 作为分数
    priorityStr, ok := nodeInfo.Node().Labels[Label]
    if !ok {
       return 0, framework.NewStatus(framework.Error, fmt.Sprintf("node %q does not have label %s", nodeName, Label))
    }

    priority, err := strconv.Atoi(priorityStr)
    if err != nil {
       return 0, framework.NewStatus(framework.Error, fmt.Sprintf("node %q has priority %s are invalid", nodeName, priorityStr))
    }

    return int64(priority), framework.NewStatus(framework.Success, "")
}

注册自定义调度器

编辑 cmd/scheduler/main.go，将上面创建的调度器插件也注册上：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
command := app.NewSchedulerCommand(
    app.WithPlugin(capacityscheduling.Name, capacityscheduling.New),
    app.WithPlugin(coscheduling.Name, coscheduling.New),
    app.WithPlugin(loadvariationriskbalancing.Name, loadvariationriskbalancing.New),
    app.WithPlugin(networkoverhead.Name, networkoverhead.New),
    app.WithPlugin(topologicalsort.Name, topologicalsort.New),
    app.WithPlugin(noderesources.AllocatableName, noderesources.NewAllocatable),
    app.WithPlugin(noderesourcetopology.Name, noderesourcetopology.New),
    app.WithPlugin(preemptiontoleration.Name, preemptiontoleration.New),
    app.WithPlugin(targetloadpacking.Name, targetloadpacking.New),
    app.WithPlugin(lowriskovercommitment.Name, lowriskovercommitment.New),
    app.WithPlugin(sysched.Name, sysched.New),
    // Sample plugins below.
    // app.WithPlugin(crossnodepreemption.Name, crossnodepreemption.New),
    app.WithPlugin(podstate.Name, podstate.New),
    app.WithPlugin(qos.Name, qos.New),
    app.WithPlugin(priority.Name, priority.New)
)

就是加了这一句：

1
app.WithPlugin(priority.Name, priority.New)

至此，我们的自定义调度器就开发完成了。

5.验证

部署

构建镜像

将自定义调度器打包成镜像

1
REGISTRY=docker.io/lixd96 RELEASE_VERSION=pripority PLATFORMS="linux/amd64,linux/arm64" DISTROLESS_BASE_IMAGE=busybox:1.36 make push-images

部署到集群

scheduler-plugins 项目下 manifest/install 目录中有部署的 chart，我们只需要修改下 image 参数就行了。

安装命令如下：

1
2
3
cd manifest/install 

helm upgrade --install i-scheduler charts/as-a-second-scheduler --namespace kube-system -f charts/as-a-second-scheduler/values-demo.yaml

完整 values-demo.yaml 内容如下：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
# Default values for scheduler-plugins-as-a-second-scheduler.
# This is a YAML-formatted file.
# Declare variables to be passed into your templates.

scheduler:
  name: i-scheduler
  image: lixd96/kube-scheduler:pripority

controller:
  replicaCount: 0

# LoadVariationRiskBalancing and TargetLoadPacking are not enabled by default
# as they need extra RBAC privileges on metrics.k8s.io.

plugins:
  enabled: ["Priority","Coscheduling","CapacityScheduling","NodeResourceTopologyMatch","NodeResourcesAllocatable"]

deploy.yaml 完整内容如下：

 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
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
apiVersion: v1
kind: ServiceAccount
metadata:
  name: i-scheduler
  namespace: kube-system
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: i-scheduler-clusterrolebinding
  namespace: kube-system
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: cluster-admin
subjects:
  - kind: ServiceAccount
    name: i-scheduler
    namespace: kube-system
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: scheduler-config
  namespace: kube-system
data:
  scheduler-config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1
    kind: KubeSchedulerConfiguration
    leaderElection:
      leaderElect: false
    profiles:
    - schedulerName: i-scheduler
      plugins:
        filter:
          enabled:
          - name: Priority
        score:
          enabled:
            - name: Priority
          disabled:
            - name: "*"
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: i-scheduler
  namespace: kube-system
spec:
  replicas: 1
  selector:
    matchLabels:
      component: i-scheduler
  template:
    metadata:
      labels:
        component: i-scheduler
    spec:
      serviceAccount: i-scheduler
      priorityClassName: system-cluster-critical
      volumes:
        - name: scheduler-config
          configMap:
            name: scheduler-config
      containers:
        - name: i-scheduler
          image: lixd96/kube-scheduler:pripority
          args:
            - --config=/etc/kubernetes/scheduler-config.yaml
            - --v=3
          volumeMounts:
            - name: scheduler-config
              mountPath: /etc/kubernetes

部署到集群

1
kubectl apply -f deploy.yaml 

确认已经跑起来了

1
2
[root@scheduler-1 install]# kubectl -n kube-system get po|grep i-scheduler
i-scheduler-569bbd89bc-7p5v2                  1/1     Running            0          3m44s

测试

创建 Pod

创建一个 Deployment 并指定使用上一步中部署的 Scheduler，然后测试会调度到哪个节点上。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
apiVersion: apps/v1
kind: Deployment
metadata:
  name: test
spec:
  replicas: 1
  selector:
    matchLabels:
      app: test
  template:
    metadata:
      labels:
        app: test
    spec:
      schedulerName: i-scheduler
      containers:
      - image: busybox:1.36
        name: nginx
        command: ["sleep"]         
        args: ["99999"]

创建之后 Pod 会一直处于 Pending 状态

1
2
3
[root@scheduler-1 install]# k get po
NAME                    READY   STATUS    RESTARTS   AGE
test-7f7bb8f449-w6wvv   0/1     Pending   0          4s

查看具体情况

1
2
3
4
5
6
[root@scheduler-1 install]# kubectl describe po test-7f7bb8f449-w6wvv

Events:
  Type     Reason            Age   From         Message
  ----     ------            ----  ----         -------
  Warning  FailedScheduling  8s    i-scheduler  0/2 nodes are available: 1 Node:Node: scheduler-1 does not have label priority.lixueduan.com, 1 Node:Node: scheduler-2 does not have label priority.lixueduan.com. preemption: 0/2 nodes are available: 2 No preemption victims found for incoming pod.

可以看到，是因为 Node 上没有我们定义的 Label，因此都不满足条件，最终 Pod 就一直 Pending 了。

添加 Label

由于我们实现的 Filter 逻辑是需要 Node 上有priority.lixueduan.com 才会用来调度，否则直接会忽略。

理论上，只要给任意一个 Node 打上 Label 就可以了。

1
2
3
4
5
6
[root@scheduler-1 install]# k get node
NAME          STATUS   ROLES           AGE     VERSION
scheduler-1   Ready    control-plane   4h34m   v1.27.4
scheduler-2   Ready    <none>          4h33m   v1.27.4
[root@scheduler-1 install]# k label node scheduler-1 priority.lixueduan.com=10
node/scheduler-1 labeled

再次查看 Pod 状态

1
2
3
[root@scheduler-1 install]# k get po -owide
NAME                    READY   STATUS    RESTARTS   AGE     IP               NODE          NOMINATED NODE   READINESS GATES
test-7f7bb8f449-w6wvv   1/1     Running   0          4m20s   172.25.123.199   scheduler-1   <none>           <none>

已经被调度到 node1 上了，查看详细日志

1
2
3
4
5
6
[root@scheduler-1 install]# k describe po test-7f7bb8f449-w6wvv
Events:
  Type     Reason            Age   From         Message
  ----     ------            ----  ----         -------
  Warning  FailedScheduling  4m8s  i-scheduler  0/2 nodes are available: 1 Node:Node: scheduler-1 does not have label priority.lixueduan.com, 1 Node:Node: scheduler-2 does not have label priority.lixueduan.com. preemption: 0/2 nodes are available: 2 No preemption victims found for incoming pod.
  Normal   Scheduled         33s   i-scheduler  Successfully assigned default/test-7f7bb8f449-w6wvv to scheduler-1

可以看到，也是 i-scheduler 在处理，调度到了 node1.

多节点排序

我们实现的 Score 是根据 Node 上的 priority.lixueduan.com 对应的 Value 作为得分的，因此肯定会调度到 Value 比较大的一个节点。

给 node2 也打上 label，value 设置为 20

1
2
3
4
5
6
[root@scheduler-1 install]# k get node
NAME          STATUS   ROLES           AGE     VERSION
scheduler-1   Ready    control-plane   4h34m   v1.27.4
scheduler-2   Ready    <none>          4h33m   v1.27.4
[root@scheduler-1 install]# k label node scheduler-2 priority.lixueduan.com=20
node/scheduler-2 labeled

然后更新 Deployment ，触发创建新 Pod ，测试调度逻辑。

因为 Node2 上的 priority 为 20，node1 上为 10，那么肯定会调度到 node2 上。

1
2
3
[root@scheduler-1 install]# k get po -owide
NAME                    READY   STATUS    RESTARTS   AGE   IP             NODE          NOMINATED NODE   READINESS GATES
test-7f7bb8f449-krvqj   1/1     Running   0          58s   172.25.0.150   scheduler-2   <none>           <none>

果然，被调度到了 Node2。

现在我们更新 Node1 的 label，改成 30

1
k label node scheduler-1 priority.lixueduan.com=30 --overwrite

再次更新 Deployment 触发调度

1
2
[root@scheduler-1 install]# k rollout restart deploy test
deployment.apps/test restarted

这样应该是调度到 node1 了，确认一下

1
2
3
[root@scheduler-1 install]# k get po -owide
NAME                   READY   STATUS    RESTARTS   AGE   IP               NODE          NOMINATED NODE   READINESS GATES
test-f7b597544-bbcb8   1/1     Running   0          65s   172.25.123.200   scheduler-1   <none>           <none>

果然在 node1，说明我们的 Scheduler 是能够正常工作的。

6. 小结

常见自定义调度器方式有两种：

Scheduler Extender
Scheduler Framework

本文主要分享了如何通过 Scheduler Framework 实现自定义调度逻辑。

K8s 调度框架定义了一些扩展点（extension points），我们只需要实现自己的调度插件，在具体扩展点接入即可。

克隆 scheduler-plugins 项目
实现对应的 framework plugin 接口并注册插件，例如framework.FilterPlugin 和framework.ScorePlugin
部署 Scheduler 时记得开启上一步中实现的 Plugin

完整代码见：lixd/i-scheduler

K8s 自定义调度器 Part2：通过 Scheduler Framework 实现自定义调度逻辑

1. 为什么需要自定义调度逻辑

什么是所谓的调度?

为什么需要自定义调度逻辑

2.如何增加自定义调度逻辑

自定义调度器的几种方法

Scheduler 配置

基础配置

扩展调度器：extenders 配置

ManagedResources 配置

3. scheduler framework 扩展点

4. 使用 Scheduler Framework 开发调度器

初始化项目

创建自定义调度器

注册自定义调度器

5.验证

部署

构建镜像

部署到集群

测试

创建 Pod

添加 Label

多节点排序

6. 小结

7. 参考