欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  科技

如何实现自己的k8s调度器

程序员文章站 2022-04-12 21:08:59
调度器介绍 scheduler 是k8s master的一部分,作为插件存在于k8s生态体系。 自定义调度器方式 添加功能重新编译 实现自己的调度器(multi-scheduler) s...

调度器介绍

scheduler 是k8s master的一部分,作为插件存在于k8s生态体系。

自定义调度器方式

添加功能重新编译 实现自己的调度器(multi-scheduler) scheduler调用扩展程序实现最终调度(Kubernetes scheduler extender)

添加调度功能

k8s中的调度算法介绍

预选
优选

实现自己的调度器(配置多个scheduler)

scheduler以插件形式存在,集群中可以存在多个scheduler,可以显式指定scheduler

配置pod使用自己的调度器

下面pod显式指定使用my-scheduler调度器

apiVersion: v1
kind: Pod
metadata:
  name: nginx
  labels:
    app: nginx
spec:
  schedulerName: my-scheduler
  containers:
  - name: nginx
    image: nginx:1.10

官方给出的shell版本scheduler示例

#!/bin/bash
SERVER='localhost:8001'
while true;
do
    for PODNAME in $(kubectl --server $SERVER get pods -o json | jq '.items[] | select(.spec.schedulerName == "my-scheduler") | select(.spec.nodeName == null) | .metadata.name' | tr -d '"')
;
    do
        NODES=($(kubectl --server $SERVER get nodes -o json | jq '.items[].metadata.name' | tr -d '"'))
        NUMNODES=${#NODES[@]}
        CHOSEN=${NODES[$[ $RANDOM % $NUMNODES ]]}
        curl --header "Content-Type:application/json" --request POST --data '{"apiVersion":"v1", "kind": "Binding", "metadata": {"name": "'$PODNAME'"}, "target": {"apiVersion": "v1", "kind"
: "Node", "name": "'$CHOSEN'"}}' https://$SERVER/api/v1/namespaces/default/pods/$PODNAME/binding/
        echo "Assigned $PODNAME to $CHOSEN"
    done
    sleep 1
done

影响pod调度的因素

https://kubernetes.io/docs/concepts/configuration/pod-priority-preemption/

预选

过滤不符合运行条件的node

优选

对node进行打分

抢占

Kubernetes 1.8 及其以后的版本中可以指定 Pod 的优先级。优先级表明了一个 Pod 相对于其它 Pod 的重要性。
当 Pod 无法被调度时,scheduler 会尝试抢占(驱逐)低优先级的 Pod,使得这些挂起的 pod 可以被调度。
在 Kubernetes 未来的发布版本中,优先级也会影响节点上资源回收的排序。

1.9+支持pdb,优先支持PDB策略,但在无法抢占其他pod的情况下,配置pdb策略的pod依旧会被抢占

Kubernetes scheduler extender

scheduler策略配置

  {
  "kind" : "Policy",
  "apiVersion" : "v1",
  "predicates" : [
    {"name" : "PodFitsHostPorts"},
    {"name" : "PodFitsResources"},
    {"name" : "NoDiskConflict"},
    {"name" : "MatchNodeSelector"},
    {"name" : "HostName"}
    ],
  "priorities" : [
    {"name" : "LeastRequestedPriority", "weight" : 1},
    {"name" : "BalancedResourceAllocation", "weight" : 1},
    {"name" : "ServiceSpreadingPriority", "weight" : 1},
    {"name" : "EqualPriority", "weight" : 1}
    ],
  "extenders" : [
    {
          "urlPrefix": "https://localhost/scheduler",
          "apiVersion": "v1beta1",
          "filterVerb": "predicates/always_true",
          "bindVerb": "",
          "prioritizeVerb": "priorities/zero_score",
          "weight": 1,
          "enableHttps": false,
          "nodeCacheCapable": false
          "httpTimeout": 10000000000
    }
      ],
  "hardPodAffinitySymmetricWeight" : 10
  }

包含extender的配置

// ExtenderConfig保存用于与扩展器通信的参数。如果动词是未指定/空的即认为该扩展器选择不提供该扩展。
type ExtenderConfig struct {
    // 访问该extender的url前缀
    URLPrefix string `json:"urlPrefix"`
    //过滤器调用的动词,如果不支持则为空。当向扩展程序发出过滤器调用时,此谓词将附加到URLPrefix
    FilterVerb string `json:"filterVerb,omitempty"`
    //prioritize调用的动词,如果不支持则为空。当向扩展程序发出优先级调用时,此谓词被附加到URLPrefix。
    PrioritizeVerb string `json:"prioritizeVerb,omitempty"`
    //优先级调用生成的节点分数的数字乘数,权重应该是一个正整数
    Weight int `json:"weight,omitempty"`
    //绑定调用的动词,如果不支持则为空。在向扩展器发出绑定调用时,此谓词会附加到URLPrefix。
    //如果此方法由扩展器实现,则将pod绑定动作将由扩展器返回给apiserver。只有一个扩展可以实现这个功能
    BindVerb string
    // EnableHTTPS指定是否应使用https与扩展器进行通信
    EnableHTTPS bool `json:"enableHttps,omitempty"`
    // TLSConfig指定传输层安全配置
    TLSConfig *restclient.TLSClientConfig `json:"tlsConfig,omitempty"`
    // HTTPTimeout指定对扩展器的调用的超时持续时间,过滤器超时无法调度pod。Prioritize超时被忽略
    //k8s或其他扩展器优先级被用来选择节点
    HTTPTimeout time.Duration `json:"httpTimeout,omitempty"`
    //NodeCacheCapable指定扩展器能够缓存节点信息
    //所以调度器应该只发送关于合格节点的最少信息
    //假定扩展器已经缓存了群集中所有节点的完整详细信息
    NodeCacheCapable bool `json:"nodeCacheCapable,omitempty"`
    // ManagedResources是由扩展器管理的扩展资源列表.
    // - 如果pod请求此列表中的至少一个扩展资源,则将在Filter,Prioritize和Bind(如果扩展程序是活页夹)
    //阶段将一个窗格发送到扩展程序。如果空或未指定,所有pod将被发送到这个扩展器。
    // 如果pod请求此列表中的至少一个扩展资源,则将在Filter,Prioritize和Bind(如果扩展程序是活页夹)阶段将一个pod发送到扩展程序。如果空或未指定,所有pod将被发送到这个扩展器。
    ManagedResources []ExtenderManagedResource `json:"managedResources,omitempty"`
}

通过k8s predicates和pod过滤的节点集传递给扩展器上的FilterVerb端点的参数。
通过k8s predicates和扩展predicates以及pod过滤的节点集传递给扩展器上的PrioritizeVerb端点的参数。

// ExtenderArgs代表被扩展器用于为pod filter/prioritize node所需要的参数
type ExtenderArgs struct {
    // 被调度的pod
    Pod   api.Pod      `json:"pod"`
    // 可被调度的候选列表
    Nodes api.NodeList `json:"nodes"`
}

“filter”被调用时返回节点列表(schedulerapi.ExtenderFilterResult),
“prioritize”返回节点的优先级(schedulerapi.HostPriorityList).

“filter”可以根据对应动作对节点列表进行剪裁,”prioritize”返回的分数将添加到k8s最终分数(通过其优先函数进行计算),用于最终宿主选择。

“bind”调用用于将pod绑定到节点的代理绑定到扩展器。它可以选择由扩展器实现。当它被实现时,
它是向apiserver发出绑定调用的扩展器的响应。 Pod名称,名称空间,UID和节点名称被传递给扩展器

ExtenderBindingArgs表示将pod绑定到节点的扩展器的参数

type ExtenderBindingArgs struct {
    // 将被绑定的pod
    PodName string
    // 将被绑定的namespace
    PodNamespace string
    // poduid
    PodUID types.UID
    // 最终调度到的pod
    Node string
}

实现

package main

import (
    "bytes"
    "encoding/json"
    "io"
    "k8s.io/api/core/v1"
    metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
    "k8s.io/client-go/kubernetes"
    "k8s.io/client-go/tools/clientcmd"
    schedulerapi "k8s.io/kubernetes/pkg/scheduler/api/v1"
    "log"
    "net/http"
)

var (
    kubeconfig string = "xxx"
)

func main() {
    http.HandleFunc("/", func(w http.ResponseWriter, _ *http.Request) {
        w.Write([]byte("hellowrold"))
    })
    http.HandleFunc("/predicates/test", testPredicateHandler)
    http.HandleFunc("/prioritize/test", testPrioritizeHandler)
    http.HandleFunc("/bind/test", BindHandler)
    http.ListenAndServe(":8880", nil)
}

func testPredicateHandler(w http.ResponseWriter, r *http.Request) {
    var buf bytes.Buffer
    body := io.TeeReader(r.Body, &buf)
    log.Println(buf.String())

    var extenderArgs schedulerapi.ExtenderArgs
    var extenderFilterResult *schedulerapi.ExtenderFilterResult

    if err := json.NewDecoder(body).Decode(&extenderArgs); err != nil {
        extenderFilterResult = &schedulerapi.ExtenderFilterResult{
            Nodes:       nil,
            FailedNodes: nil,
            Error:       err.Error(),
        }
    } else {
        extenderFilterResult = predicateFunc(extenderArgs)
    }
    if resultBody, err := json.Marshal(extenderFilterResult); err != nil {
        panic(err)
    } else {
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(http.StatusOK)
        w.Write(resultBody)
    }

}

func testPrioritizeHandler(w http.ResponseWriter, r *http.Request) {
    var buf bytes.Buffer
    body := io.TeeReader(r.Body, &buf)
    var extenderArgs schedulerapi.ExtenderArgs
    var hostPriorityList *schedulerapi.HostPriorityList
    if err := json.NewDecoder(body).Decode(&extenderArgs); err != nil {
        panic(err)
    }
    if list, err := prioritizeFunc(extenderArgs); err != nil {
        panic(err)
    } else {
        hostPriorityList = list
    }
    if resultBody, err := json.Marshal(hostPriorityList); err != nil {
        panic(err)
    } else {
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(http.StatusOK)
        w.Write(resultBody)
    }
}

func predicateFunc(args schedulerapi.ExtenderArgs) *schedulerapi.ExtenderFilterResult {
    pod := args.Pod
    canSchedule := make([]v1.Node, 0, len(args.Nodes.Items))
    canNotSchedule := make(map[string]string)
    for _, node := range args.Nodes.Items {
        result, err := func(pod v1.Pod, node v1.Node) (bool, error) {
            return true, nil
        }(pod, node)
        if err != nil {
            canNotSchedule[node.Name] = err.Error()
        } else {
            if result {
                canSchedule = append(canSchedule, node)
            }
        }
    }
    result := schedulerapi.ExtenderFilterResult{
        Nodes: &v1.NodeList{
            Items: canSchedule,
        },
        FailedNodes: canNotSchedule,
        Error:       "",
    }
    return &result
}

func prioritizeFunc(args schedulerapi.ExtenderArgs) (*schedulerapi.HostPriorityList, error) {
    nodes := args.Nodes.Items
    var priorityList schedulerapi.HostPriorityList
    priorityList = make([]schedulerapi.HostPriority, len(nodes))
    for i, node := range nodes {
        priorityList[i] = schedulerapi.HostPriority{
            Host:  node.Name,
            Score: 0,
        }
    }
    return &priorityList, nil
}

func BindHandler(w http.ResponseWriter, r *http.Request) {
    var buf bytes.Buffer
    body := io.TeeReader(r.Body, &buf)
    var extenderBindingArgs schedulerapi.ExtenderBindingArgs
    if err := json.NewDecoder(body).Decode(&extenderBindingArgs); err != nil {
        panic(err)
    }
    b := &v1.Binding{
        ObjectMeta: metav1.ObjectMeta{Namespace: extenderBindingArgs.PodNamespace, Name: extenderBindingArgs.PodName, UID: extenderBindingArgs.PodUID},
        Target: v1.ObjectReference{
            Kind: "Node",
            Name: extenderBindingArgs.Node,
        },
    }
    bind(b)

}

func bind(b *v1.Binding) error {
    config, err := clientcmd.BuildConfigFromFlags("", kubeconfig)
    if err != nil {
        panic(err)
    }

    clientset, err := kubernetes.NewForConfig(config)
    if err != nil {
        panic(err)
    }
    return clientset.CoreV1().Pods(b.Namespace).Bind(b)
}