实现一个 Kubernetes Scheduler Extender

0. 概述

在使用 Kubernetes 的时候，虽然默认的调度策略已经比较充足了，但是，有些时候我们有些私人的需求不得不干涉默认的调度器，这里方式有很多，本文就介绍其中一种简单且侵入性小的方式：Scheduler Extender。

在开始了解 Scheduler Extender 之前，我希望你已经了解了 Kubernetes 的 Scheduler Framework，大概知道了 Kubernetes 的调度的整体过程，这样，看这个才有意义，不然可能会出现上下文无法衔接的情况。如果之前没有了解过，那么可以看下我写过的一篇旧文：Kubernetes Scheduler Framwork。

1. 调度扩展

Kubernetes 支持在原生的调度过程中加入自定义的调度策略，支持的扩展方式也有好几种，这里我介绍的 Scheduler Extender 的方式是一种侵入性比较小的一种，可以独立于原生 Scheduler 运行，并且无需修改原生 Scheduler 的代码，只需要在运行原生 Scheduler 的时候加一个配置即可。

但是，Scheduler Extender 并不是万能的，目前只支持三个操作，分别是：

Filter：用于确定一个 Node 是否可以放置一个 Pod；
Priority：用于确定一个 Pod 与一个 Node 的匹配程度；
Bind：将一个 Pod 绑定到一个 Node 中；

如果你了解了 Scheduler Framework 的话，那么这两个概念应该都是很熟悉的了。虽然简单，但是，其实这三个操作已经可以满足很多需求了，例如让一个 Pod 不调度到一个 Node，或者让一个 Pod 优先调度到一个 Node 等等，可以弱弱地说一句，能实现的效果可能局限于我的想象力。

2. 运行方式

Scheduler Extender 的运行方式是以独立进程的形式运行，你可以放到 Kubernetes 的 Node 中用 System 来管理；也可以以一个 Pod 的形式运行在 Kubernetes 中；当然，你也可以以 Sidecar 形式与原生 Scheduler 运行在同一个 Pod 上（如果你的 Scheduler 是以 Pod 的形式运行的话）。

然后 Scheduler Extender 是以 HTTP Endpoint 的形式提供服务，通过在原生 Scheduler 的运行参数中加入一个：--config，例如：

[[email protected]]# cat /etc/systemd/system/kube-scheduler.service
... ...
ExecStart=/opt/kube/bin/kube-scheduler \
  --config=/etc/sysconfig/kube-scheduler/extender-config.yaml \
  --address=127.0.0.1 \
  --master=http://127.0.0.1:8080 \
  --leader-elect=true \
  --v=2
... ...

这里就指定了一个 Config，里面的内容为：

[[email protected]]# cat /etc/sysconfig/kube-scheduler/extender-config.yaml
apiVersion: kubescheduler.config.k8s.io/v1alpha1
kind: KubeSchedulerConfiguration
clientConnection:
  kubeconfig: "/root/.kube/config"
algorithmSource:
  policy:
    file:
      path: "/etc/sysconfig/kube-scheduler/extender-policy.json"

这里指定了一个 Algorithm Source，如果你看过我之前关于 Scheduler 源码解析的内容，那么你也就了解这个项是啥意思了，如果不想看，那么也没关系，直接理解为还需要加载另外一个配置文件就对了，另外一个配置文件就是指定 Scheduler Extender 的 HTTP Endpint：

[[email protected]]# cat /etc/sysconfig/kube-scheduler/extender-policy.json
{
    "kind" : "Policy",
    "apiVersion" : "v1",
    "extenders" : [{
        "urlPrefix": "http://localhost:8888/",
        "filterVerb": "filter",
        "enableHttps": false
    }]
}

这里只指定了一个 Filter 的操作，如果你想要指定 Priority 的操作的话，那么可以加上：

[[email protected]]# cat /etc/sysconfig/kube-scheduler/extender-policy-full.json
   ... ...
  "extenders" : [{
        "urlPrefix": "http://localhost:8888/",
        "filterVerb": "filter",
        "prioritizeVerb": "prioritize",
        "weight": 1,
        "enableHttps": false
  }
  ... ...

然后重启 Scheduler 就可以生效了。如果你想了解更多的详细配置信息，那么可以看一下 Kubernetes 的官方文档：Scheduler Extender 文档

3. Scheduler Extender 接口

在 2 中已经介绍过了，Scheduler Extender 是一个独立运行的进程，并且以 HTTP Endpoint 的形式提供服务，所以，开发 Scheduler Extender 就像开发一个普通的 HTTP 应用一样简单。你可以选择你熟悉的 HTTP 框架或者工具开发，而唯一需要关注的就是 Filter/Priority 以及 Binding 这三个操作的接口：

对于 Filter 和 Priority 这两个操作，HTTP 请求的 Body 是这么一个结构体：

// ExtenderArgs represents the arguments needed by the extender to filter/prioritize
// nodes for a pod.
type ExtenderArgs struct {
    // Pod being scheduled
    Pod   api.Pod      `json:"pod"`
    // List of candidate nodes where the pod can be scheduled
    Nodes api.NodeList `json:"nodes"`
}

你只需要在 HTTP 处理代码中反序列化成这个结构体，然后执行自己的逻辑，并且返回对应的结果就可以了，其中 Filter 返回的结构体为：

// ExtenderFilterResult represents the results of a filter call to an extender
type ExtenderFilterResult struct {
    // Filtered set of nodes where the pod can be scheduled; to be populated
    // only if ExtenderConfig.NodeCacheCapable == false
    Nodes *v1.NodeList
    // Filtered set of nodes where the pod can be scheduled; to be populated
    // only if ExtenderConfig.NodeCacheCapable == true
    NodeNames *[]string
    // Filtered out nodes where the pod can't be scheduled and the failure messages
    FailedNodes FailedNodesMap
    // Error message indicating failure
    Error string
}

Priority 返回的结构体为：

// HostPriority represents the priority of scheduling to a particular host, higher priority is better.
type HostPriority struct {
    // Name of the host
    Host string
    // Score associated with the host
    Score int64
}

// HostPriorityList declares a []HostPriority type.
type HostPriorityList []HostPriority

而 Binding 的是另外一个结构体：

// ExtenderBindingArgs represents the arguments to an extender for binding a pod to a node.
type ExtenderBindingArgs struct {
    // PodName is the name of the pod being bound
    PodName string
    // PodNamespace is the namespace of the pod being bound
    PodNamespace string
    // PodUID is the UID of the pod being bound
    PodUID types.UID
    // Node selected by the scheduler
    Node string
}

Binding 不需要返回特别的结果，只需要返回是否存在错误即可。

4. 实现示例

我这里就给出一个 Filter 的示例，用于演示一下如何开发一个 Scheduler Extender：

[[email protected]]# cat filter.go
func main() {
    router := httprouter.New()
    router.POST("/filter", Filter)
    log.Fatal(http.ListenAndServe(":8888", router))
}

func Filter(w http.ResponseWriter, r *http.Request, ps httprouter.Params) {
    var buf bytes.Buffer
    body := io.TeeReader(r.Body, &buf)
    log4go.Info("Request body: %s", buf.String())
    var extenderArgs schedulerapi.ExtenderArgs
    var extenderFilterResult *schedulerapi.ExtenderFilterResult
    if err := json.NewDecoder(body).Decode(&extenderArgs); err != nil {
        extenderFilterResult = &schedulerapi.ExtenderFilterResult{
            Error: err.Error(),
        }
    } else {
        log4go.Info("Pod: %s", extenderArgs.Pod)
        extenderFilterResult = filter(extenderArgs)
    }

    if response, err := json.Marshal(extenderFilterResult); err != nil {
        log.Fatalln(err)
    } else {
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(http.StatusOK)
        w.Write(response)
    }
}

这里我忽略了 filter 的业务逻辑，只给出了大体的框架，如果你感兴趣的话，我的完整代码已经放在公开的 Repo：Sample Scheduler Extender，你可以自行下载阅读。

5. 小结

本文介绍了一下 Scheduler Extender 的原理，以及支持的操作和数据接口是如何，最后，演示了一下实现的框架，并且给出了一个示例的应用，希望对你有所帮助。

6. Ref

Create a custom Kubernetes scheduler

格物致知

All Posts