哈啰Kubernetes基于水位的自定义调度器落地之路

allNode = K8S所有的node节点
for PreFilter in (plugin1, plugin2, ...):  IsSuccess = PreFilter(state *CycleState, pod *v1.Pod)  if IsSuccess is False:    return // 调度周期结束  feasibleNodes = []       // 存储Filter阶段符合条件的Nodefor nodeInfo in allNode:  IsSuccess = False  for Filter in (plugin1, plugin2, ...):    IsSuccess = Filter(state *CycleState, pod *v1.Pod, nodeInfo *NodeInfo)    // 如果任意一个插件返回了False，说明Node不符合调度条件    if IsSuccess is False:         break  if IsSuccess = True:    feasibleNodes.append(nodeInfo)
// 如果只有一个Node通过了Filter阶段的检查，该Node会直接进入绑定阶段，跳过打分阶段if len(feasibleNodes) == 1:  return feasibleNodes[0]
for PreScore in (plugin1, plugin2, ...):  PreScore(state *CycleState, pod *v1.Pod)  
NodeScores = { }// NodeScores数据结构: {"plugin1": [node1_score, node2_score], "plugin2": [...], ... }// 每个插件对每个Node，都会进行一次打分，总共会有(Node数量*插件数)个分数for index, nodeInfo in feasibleNodes:  for Score in (plugin1, plugin2, ...):    score = Score(state *CycleState, pod *v1.Pod, nodeInfo *NodeInfo)    NodeScores[插件名][index] = score // 归一化， 这个阶段处理过，分数都在[1-100]之间for NormalizeScore in (plugin1, plugin2, ...):  nodeScoreList = NodeScores[插件名]  NormalizeScore(state *CycleState, p *v1.Pod, scores NodeScoreList) // 加上插件权重因子for pluginName, nodeScoreList in NodeScores:  for nodeScore in nodeScoreList:    nodeScoreList[i].Score = nodeScore.Score * int64(pluginWeight)
// 计算每个Node的总分result = []for nodeIndex, nodeName in feasibleNodes {  _result = {Name: nodeName, Score: 0})  for pluginName, _ in NodeScores {    _result.Score += NodeScores[pluginName][nodeIndex].Score  }  result.append(_result)}
// result 结果为 [{Name: node1, Score: 200}, {Name: node2, Score: 100}, ...]
// selectHost 找到得分最高的Node进入绑定阶段Node = selectHost(result)return Node

apiVersion: kubescheduler.config.k8s.io/v1alpha2kind: KubeSchedulerConfigurationprofiles:- schedulerName: default-scheduler  plugins:    preFilter:      enabled:        - name: HheWaterLevelBalance    filter:      enabled:        - name: HheWaterLevelBalance        - name: HkePodTopologySpread    preScore:      enabled:        - name: HkePodTopologySpread        - name: HheWaterLevelBalance    score:      enabled:        - name: HkePodTopologySpread         // 启用自定义插件        - name: HheWaterLevelBalance      disabled:        - name: ImageLocality             // 禁用默认插件        - name: InterPodAffinity    postBind:      enabled:        - name: HheWaterLevelBalance  pluginConfig:      // 插件配置  - name: HheWaterLevelBalance    args:      clusterCpuMinNodeWeight: 0.2

metric(cpu_usage_avg_5m、cpu_usage_max_avg_1h、cpu_usage_max_avg_1d、mem_usage_avg_5m、mem_usage_max _avg_1h、mem_usage_max_avg_1d)

cluster_cpu = 预设理想值target_cpu = node_cpu + pod_cpuif target_cpu <= cluster_cpu:  score = (100 - cluster_cpu)target_cpu/cluster_cpu+ cluster_cpu else if cluster_cpu < target_cpu <= 100:  score = cluster_cpu(100 - target_cpu)/(100 - cluster_cpu)else:  score = 0

Na = 0Nb = 4Nc = 24Nd = 49Ne = 98

Ta = 1Tb = 5Tc = 25Td = 50Te = 99

Sa = (100 - 20) * 1 / 20 + 20 = 24Sb = (100 - 20) * 5 / 20 + 20 = 40Sc = 20 * (100 - 25) / (100 - 20) = 19Sd = 20 * (100 - 50) / (100 - 20) = 13Se = 20 * (100 - 90) / (100 - 20) = 3

{  "node1": [    {      "Timestamp": "unixTime",      "PodName": "podName",      "PodUtil": { "cpu": 100, "mem": 500 }    },    {      "Timestamp": "unixTime2",      "PodName": "podName2",      "PodUtil": { "cpu": 100, "mem": 500 }    }  ],  "nodeName2": {    "Timestamp": "unixTime",    "PodName": "podName",    "PodUtil": { "cpu": 100, "mem": 500 }  }}

import (  "math/rand"  "os"  "time"
  "k8s.io/component-base/logs"  "k8s.io/kubernetes/cmd/kube-scheduler/app"
  "pkg/hhewaterlevelbalance"    "pkg/pugin2")
func main() {  rand.Seed(time.Now().UnixNano())  // Register custom plugins to the scheduler framework.  // Later they can consist of scheduler profile(s) and hence  // used by various kinds of workloads.  command := app.NewSchedulerCommand(    app.WithPlugin(hhewaterlevelbalance.Name, HheWaterLevelBalance.New),  // hhewaterlevelbalance.Name为插件名字    app.WithPlugin(pugin2.Name, pugin2.New),  )
  logs.InitLogs()  defer logs.FlushLogs()
  if err := command.Execute(); err != nil {    os.Exit(1)  }}

VERSION := $(shell git describe --tags --match "v*" | awk -F - '{print $$1}' 2>/dev/null || (printf "v0.0.0"))COMMIT := $(shell git rev-parse --short HEAD)RELEASE_DATE :=$(shell date +%Y%m%d)
LDFLAGS=-ldflags "-X k8s.io/component-base/version.gitVersion=$(VERSION)-$(COMMIT)-${RELEASE_DATE}-hellobike -w"
build:  go build $(LDFLAGS) -o bin/kube-scheduler cmd/main.g

// 插件名称const Name = "HheWaterLevelBalance"
type HheWaterLevelBalanceArgs struct {  ClusterCpuMinNodeWeight float64}
type HheWaterLevelBalance struct {  args   *HheWaterLevelBalanceArgs  handle framework.FrameworkHandle}
func (h *HheWaterLevelBalance) Name() string {  return Name}
func (h *HheWaterLevelBalance) PreFilter(pc *framework.PluginContext, pod *v1.Pod) *framework.Status {  klog.V(3).Infof("prefilter pod: %v", pod.Name)  return framework.NewStatus(framework.Success, "")}
func (h *HheWaterLevelBalance) Filter(pc *framework.PluginContext, pod *v1.Pod, nodeName string) *framework.Status {  klog.V(3).Infof("filter pod: %v, node: %v", pod.Name, nodeName)  return framework.NewStatus(framework.Success, "")}
func (h *HheWaterLevelBalance) PreScore(  pc *framework.PluginContext,  cycleState *framework.CycleState,  pod *v1.Pod,  filteredNodes []*v1.Node,) *framework.Status {  klog.V(3).Infof("prescore pod: %v", pod.Name)  return framework.NewStatus(framework.Success, "")}
func (h *HheWaterLevelBalance) Score(pc *framework.PluginContext, cycleState *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {    klog.V(3).Infof("score pod: %v, node: %v", pod.Name, nodeName)  return score, framework.NewStatus(framework.Success, "")}
func New(config *runtime.Unknown, f framework.FrameworkHandle) (framework.Plugin, error) {  args := &HheWaterLevelBalanceArgs{}  if err := framework.DecodeInto(config, args); err != nil {    return nil, err  }  klog.V(3).Infof("get plugin config args: %+v", args)  return &HheWaterLevelBalance{    args: args,    handle: f,  }, nil}