Kubernetes 开发【5】—— scheduling framework 构建自定义调度插件

目录

环境信息

背景

前言

主体框架

插件获取 clientSet/informer

设置插件调度参数

总结

---

环境信息

kubernetes:1.22

centos:7.9

apiVersion: kubescheduler.config.k8s.io/v1beta2(即将在1.25中被废弃)

背景

对于创建一个pod，我们知道有以下流程

在调度/绑定的不同阶段，我们都可以注入我们自定义的插件，来对调度流程进行一定的干预。

调度框架Scheduling Framework正是实现这一功能的工具手段。

由于调度器涉及到调度算法，本文在这方面不进行深究，仅记录Scheduling Framework的搭建和简单使用

前言

对于不同插件的基本描述，摘自官方文档

PreFilter

这些插件用于预处理 Pod 的相关信息，或者检查集群或 Pod 必须满足的某些条件。 如果 PreFilter 插件返回错误，则调度周期将终止。

Filter

这些插件用于过滤出不能运行该 Pod 的节点。对于每个节点， 调度器将按照其配置顺序调用这些过滤插件。如果任何过滤插件将节点标记为不可行， 则不会为该节点调用剩下的过滤插件。节点可以被同时进行评估。

PostFilter

这些插件在 Filter 阶段后调用，但仅在该 Pod 没有可行的节点时调用。 插件按其配置的顺序调用。如果任何 PostFilter 插件标记节点为“Schedulable”， 则其余的插件不会调用。典型的 PostFilter 实现是抢占，试图通过抢占其他 Pod 的资源使该 Pod 可以调度。

PreScore

这些插件用于执行 “前置评分（pre-scoring）” 工作，即生成一个可共享状态供 Score 插件使用。 如果 PreScore 插件返回错误，则调度周期将终止。

Score

这些插件用于对通过过滤阶段的节点进行排序。调度器将为每个节点调用每个评分插件。 将有一个定义明确的整数范围，代表最小和最大分数。 在标准化评分阶段之后，调度器将根据配置的插件权重 合并所有插件的节点分数。

主体框架

首先参考官方的示例代码

GitHub - kubernetes-sigs/scheduler-plugins: Repository for out-of-tree scheduler plugins based on scheduler framework.

我们构建出main.go

package main
 
import (
	"fmt"
	"k8s.io/component-base/logs"
	"k8s.io/kubernetes/cmd/kube-scheduler/app"
	"myscheduler/lib"
	"os"
)
 
func main() {
	//来自/blob/master/cmd/scheduler/main.go
	command := app.NewSchedulerCommand(
        //可变参数——需要注入的插件列表
		app.WithPlugin(lib.TestSchedulingName, lib.NewTestScheduling),
	)
	logs.InitLogs()
	defer logs.FlushLogs()
 
	if err := command.Execute(); err != nil {
		_, _ = fmt.Fprintf(os.Stderr, "%v\n", err)
		os.Exit(1)
	}
}

其中，需要我们定义好插件的名称和相关实现，因此引出如下的代码框架

//出自/pkg/capacityscheduling 只留了主体框架，简化了大部分
 
const TestSchedulingName = "test-scheduling" //记住这个调度器名称
 
type TestScheduling struct {}
 
func (*TestScheduling) Name() string { //实现framework.Plugin的接口方法
	return TestSchedulingName
}
 
func NewTestScheduling(configuration runtime.Object, f framework.Handle) (framework.Plugin, error) {
	return &TestScheduling{}, nil
}

插件方法

不同阶段的插件在framework包里其实就是不同的接口，我们要注入对应阶段的插件，那么必须实现相应的接口方法，这里参照官方的方法，通过goland快速生成preFilter的接口方法。

var _ framework.PreFilterPlugin = &TestScheduling{}

生成了两个接口方法

//业务方法
func PreFilter(ctx context.Context, state *framework.CycleState, p *v1.Pod) *framework.Status
//这个方法是在生成pod或删除pod时产生一些需要评估的内容，返回值同样是个接口，返回自身并快速生成接口方法即可
func PreFilterExtensions() framework.PreFilterExtensions

在其中实现我们所要进行的干预即可。

注册调度器

代码编译，打包镜像的步骤省略～

因为调度器pod需要访问apiserver，所以需要指定serviceaccount并且绑定权限

kind: ClusterRole
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: test-scheduling-clusterrole
rules:
  - apiGroups:
      - ""
    resources:
      - endpoints
      - events
    verbs:
      - create
      - get
      - update
  - apiGroups:
      - ""
    resources:
      - nodes
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - ""
    resources:
      - pods
    verbs:
      - delete
      - get
      - list
      - watch
      - update
  - apiGroups:
      - ""
    resources:
      - bindings
      - pods/binding
    verbs:
      - create
  - apiGroups:
      - ""
    resources:
      - pods/status
    verbs:
      - patch
      - update
  - apiGroups:
      - ""
    resources:
      - replicationcontrollers
      - services
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - apps
      - extensions
    resources:
      - replicasets
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - apps
    resources:
      - statefulsets
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - policy
    resources:
      - poddisruptionbudgets
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - ""
    resources:
      - persistentvolumeclaims
      - persistentvolumes
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - ""
    resources:
      - namespaces
      - configmaps
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - "storage.k8s.io"
    resources: ['*']
    verbs:
      - get
      - list
      - watch
  - apiGroups:
      - "coordination.k8s.io"
    resources:
      - leases
    verbs:
      - create
      - get
      - list
      - update
  - apiGroups:
      - "events.k8s.io"
    resources:
      - events
    verbs:
      - create
      - patch
      - update
 
---
apiVersion: v1
kind: ServiceAccount
metadata:
  name: test-scheduling-sa
  namespace: kube-system
---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: test-scheduling-clusterrolebinding
  namespace: kube-system
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: test-scheduling-clusterrole
subjects:
  - kind: ServiceAccount
    name: test-scheduling-sa
    namespace: kube-system

调度器启动时，需要引用相应的配置文件来进行插件类型的注册，参数的设置等，我们通过configMap来将配置挂载进容器

apiVersion: v1
kind: ConfigMap
metadata:
  name: test-scheduling-config
  namespace: kube-system
data:
   config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1beta2
    kind: KubeSchedulerConfiguration
    leaderElection:
      leaderElect: false
    profiles:
      - schedulerName: test-scheduling
        plugins:
          preFilter:
            enabled:
            - name: "test-scheduling"

接下来是对调度器的定义，以固定节点并将可执行文件挂载的方式运行（仅用于测试）

apiVersion: apps/v1
kind: Deployment
metadata:
  name: test-scheduling
  namespace: kube-system
spec:
  replicas: 1
  selector:
    matchLabels:
      app: test-scheduling
  template:
    metadata:
      labels:
        app: test-scheduling
    spec:
      nodeName: master-01
      serviceAccount: test-scheduling-sa
      containers:
        - name: tests-cheduling
          image: alpine:3.12
          imagePullPolicy: IfNotPresent
          command: ["/app/test-scheduling"]
          args:
            - --config=/etc/kubernetes/config.yaml
            - --v=3
          volumeMounts:
            - name: config
              mountPath: /etc/kubernetes
            - name: app
              mountPath: /app
      volumes:
        - name: config
          configMap:
            name: test-scheduling-config
        - name: app
          hostPath:
             path: /root/schedular

 检查调度器状态，设置为Running后，即可以在创建负载时在Pod的配置中的schedulerName项指定调度器。

apiVersion: apps/v1
kind: Deployment
metadata:
  name: testngx
  namespace: default
spec:
  replicas: 1
  selector:
    matchLabels:
      app: testngx
  template:
    metadata:
      labels:
        app: testngx
    spec:
      schedulerName: test-scheduling
      containers:
        - image: nginx:1.18-alpine
          imagePullPolicy: IfNotPresent
          name: testngx
          ports:
            - containerPort: 80

观察调度器的日志，可以发现，成功进行了干预

kubectl logs test-scheduling-54fd7c585f-gmbb6 -n kube-system -f

I1117 08:51:46.567953       1 eventhandlers.go:123] "Add event for unscheduled pod" pod="default/testngx-7cd55446f7-4cmgv"

I1117 08:51:46.568030       1 scheduler.go:516] "Attempting to schedule pod" pod="default/testngx-7cd55446f7-4cmgv"

I1117 08:51:46.568094       1 test-scheduling.go:57] 预过滤

插件获取 clientSet/informer

从本节开始，介绍几种插件中的常规做法；

 首先是go-client的获取，示例场景：在filter插件中，过滤带有xx标签的节点。

我们观察到，插件结构体的构造函数中，有这么一个入参：

f framework.Handle

这个Handle类型，可以帮助我们获取clientSet或者是informer；这里以获取informer为例。

那么，就有新的成员变量以及构造函数

type TestScheduling struct {
	fac  informers.SharedInformerFactory
}

func NewTestScheduling(configuration runtime.Object, f framework.Handle) (framework.Plugin, error) {
	return &TestScheduling{
		fac:  f.SharedInformerFactory(),
	}, nil //注入informer工厂
}

 那么就可以在插件接口方法里去实现这个逻辑了

func (s *TestScheduling) Filter(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
	klog.V(3).Infof("过滤节点")
	for k, v := range nodeInfo.Node().Labels {
		if k == "scheduling" && v != "true" {
			return framework.NewStatus(framework.Unschedulable, "设置了不可调度的标签")
		}
	}
	return framework.NewStatus(framework.Success)
}

同样需要在调度器的配置文件中开启这个插件

apiVersion: v1
kind: ConfigMap
metadata:
  name: test-scheduling-config
  namespace: kube-system
data:
   config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1beta2
    kind: KubeSchedulerConfiguration
    leaderElection:
      leaderElect: false
    profiles:
      - schedulerName: test-scheduling
        plugins:
          preFilter:
            enabled:
            - name: "test-scheduling"
          filter:
            enabled:
            - name: "test-scheduling"

为集群中的一个节点打上相应的标签

kubectl label node node-01 scheduling=false

然后创建负载，观察到pod为pending状态 

testngx-677b6896b-nqsk8   0/1     Pending   0          5s

查看pod事件，观察到节点因为设置了不可调度的标签而被过滤掉了

Events:

  Type     Reason            Age   From             Message

  ----     ------            ----  ----             -------

  Warning  FailedScheduling  28s   test-scheduling  0/3 nodes are available: 1 node(s) didn't match Pod's node affinity/selector, 1 node(s) had taint {node-role.kubernetes.io/master: }, that the pod didn't tolerate, 1 设置了不可调度的标签.

至此，调度器中clientset/informer的获取和调度失败的示例演示就完成了。

值得一提的是，如果负载yaml通过nodeName固定了节点，那么shcedulerName配置的调度器将不作用于Pod的调度中，即使是filter插件中的逻辑过滤了所要固定的节点时。

设置插件调度参数

本节演示通过设置配置文件的参数，来让调度器动态读取。

示例场景是如果创建负载所在的命名空间的Pod数量超过了n，则返回调度失败。因为不涉及节点筛选阶段，所以最好的实践是在prefilter插件中实现。

调度器配置文件中有如下配置：

apiVersion: v1
kind: ConfigMap
metadata:
  name: test-scheduling-config
  namespace: kube-system
data:
   config.yaml: |
    apiVersion: kubescheduler.config.k8s.io/v1beta2
    kind: KubeSchedulerConfiguration
    leaderElection:
      leaderElect: false
    profiles:
      - schedulerName: test-scheduling
        plugins:
          preFilter:
            enabled:
            - name: "test-scheduling"
          filter:
            enabled:
            - name: "test-scheduling"
        pluginConfig:
          - name: test-scheduling
            args:
              maxPods: 5

其中设定了最大 Pod数量为5。

将这个参数作为成员变量加入到调度器结构体中

type TestScheduling struct {
	fac  informers.SharedInformerFactory
	args *Args
}
 
type Args struct {
	MaxPods int `json:"maxPods,omitempty"`
}

在构造函数中，有这样一个入参：

configuration runtime.Object

这其中包含了我们在配置文件中的配置，将其反解到我们的配置结构体中，并在构造函数中赋值

func NewTestScheduling(configuration runtime.Object, f framework.Handle) (framework.Plugin, error) {
	args := &Args{}
	if err := frameworkruntime.DecodeInto(configuration, args); err != nil { //由配置文件注入参数，并通过configuration获取
		return nil, err
	}
	return &TestScheduling{
		fac:  f.SharedInformerFactory(),
		args: args,
	}, nil //注入informer工厂
}

在filter插件的接口函数中直接取即可。

func (s *TestScheduling) PreFilter(ctx context.Context, state *framework.CycleState, p *v1.Pod) *framework.Status {
	klog.V(3).Infof("预过滤")
	pods, err := s.fac.Core().V1().Pods().Lister().Pods(p.Namespace).List(labels.Everything())
	if err != nil {
		return framework.NewStatus(framework.Error, err.Error())
	}
	if len(pods) > s.args.MaxPods { 
		return framework.NewStatus(framework.Unschedulable, "pod数量超过了最大限制")
	}
	return framework.NewStatus(framework.Success)
}

调度失败的观察结果与上一节类似，这里省略了。

总结

 至此，我们完成硬性过滤插件中一些简单的玩法。后面我们将演示软性打分插件prescore/score插件中的一些基操，包括presocre预打分插件中抓取和存入原始数据，这份数据如果和prescore和score插件中传递，以及如何通过Normalize归一化处理来使多个插件协同打分，使最终得分落在一个标定的区间。