schedextender

1.简介

1.1.扩展

• kube-scheduler基于plugin模式进行功能扩展，plugin可以在多个扩展点进行注册，调度器启动会加载注册插件到registry，pod调度就是根据plugins过滤出适合的节点集合，通过score plugin打分选出分值最高的节点。
• 

• 注意

  plugin对应三种扩展形式：inTree、outTree和extender

1.2.extention

• kube-scheduler支持四种调度扩展方式，以影响pod调度的预选、优选及绑定，标准扩展推荐的是scheduler framework形式。

• --- default-scheduler
  	直接修改kube-scheduler开发inTree插件,扩展后重新编译
  
  --- custom-scheduler
  	重新实现一个与kube-scheduler并行的调度器,单独或配合kube-scheduler一起运行
  
  --- scheduler-extender
  	实现extender接口注册到kube-scheduler,kube-scheduler基于http/https调用extender作为补充
  
  --- scheduler-framework
  	基于framework标准化实现outTree插件,重新编译kube-scheduler
  

• 

• 注意

  inTree plugin是kube-scheduler内置的常用插件，extender和outTree是扩展实现，extender由于网络开销已不被推荐

1.3.framework

• scheduler framework基于scheduler core基础进行改造和提取，关键路径均提供plugin扩展，plugin实现及注册后与core重新编译。

• type Framework interface {
      // 数据维护相关
  	Handle
  
  	// preEnqueue插件获取接口
  	PreEnqueuePlugins() []PreEnqueuePlugin
  
  	// 待调度队列的pod排序函数
  	QueueSortFunc() LessFunc
  
  	// 执行预检查插件
  	RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (*PreFilterResult, *Status)
  
  	// 执行后置检查插件
  	RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)
  
  	// 执行预绑定插件
  	RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
  
  	// 执行后置绑定插件
  	RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
  
  	// 执行资源锁定插件
  	RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
  
  	// 执行资源释放插件
  	RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
  
  	// 执行许可检查插件
  	RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
  
  	// 许可唤醒检查
  	WaitOnPermit(ctx context.Context, pod *v1.Pod) *Status
  
  	// 执行绑定插件
  	RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
  
  	// 执行过滤插件
  	HasFilterPlugins() bool
  
  	// 执行抢占调度插件
  	HasPostFilterPlugins() bool
  
  	// 执行打分插件
  	HasScorePlugins() bool
  
  	// 获取插件列表
  	ListPlugins() *config.Plugins
  
  	// 调度框架对应名字
  	ProfileName() string
  
  	// 参与预选的节点百分比
  	PercentageOfNodesToScore() *int32
  
  	// 设置pod提名器
  	SetPodNominator(nominator PodNominator)
  }
  

• 

• 注意

  outTree实现需引入scheduler core包，plugin注册到registry，类似的实现有koordinator和scheduler-plugins

2.插件

2.1.inTree插件

• kube-scheduler内置了很多常用插件，包括NodeName、TaintToleration、NodeAffinity及NodeResourcesFit，以支持pod调度。

• // NewInTreeRegistry builds the registry with all the in-tree plugins.
  func NewInTreeRegistry() runtime.Registry {
  	...
  
      // 部分内置插件为实验特性
  	registry := runtime.Registry{
  		dynamicresources.Name:                runtime.FactoryAdapter(fts, dynamicresources.New),
  		selectorspread.Name:                  selectorspread.New,
  		imagelocality.Name:                   imagelocality.New,
  		tainttoleration.Name:                 tainttoleration.New,
  		nodename.Name:                        nodename.New,
  		nodeports.Name:                       nodeports.New,
  		nodeaffinity.Name:                    nodeaffinity.New,
  		podtopologyspread.Name:               runtime.FactoryAdapter(fts, podtopologyspread.New),
  		nodeunschedulable.Name:               nodeunschedulable.New,
  		noderesources.Name:                   runtime.FactoryAdapter(fts, noderesources.NewFit),
  		noderesources.BalancedAllocationName: runtime.FactoryAdapter(fts, noderesources.NewBalancedAllocation),
  		volumebinding.Name:                   runtime.FactoryAdapter(fts, volumebinding.New),
  		volumerestrictions.Name:              runtime.FactoryAdapter(fts, volumerestrictions.New),
  		volumezone.Name:                      volumezone.New,
  		nodevolumelimits.CSIName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewCSI),
  		nodevolumelimits.EBSName:             runtime.FactoryAdapter(fts, nodevolumelimits.NewEBS),
  		nodevolumelimits.GCEPDName:           runtime.FactoryAdapter(fts, nodevolumelimits.NewGCEPD),
  		nodevolumelimits.AzureDiskName:       runtime.FactoryAdapter(fts, nodevolumelimits.NewAzureDisk),
  		nodevolumelimits.CinderName:          runtime.FactoryAdapter(fts, nodevolumelimits.NewCinder),
  		interpodaffinity.Name:                interpodaffinity.New,
  		queuesort.Name:                       queuesort.New,
  		defaultbinder.Name:                   defaultbinder.New,
  		defaultpreemption.Name:               runtime.FactoryAdapter(fts, defaultpreemption.New),
  		schedulinggates.Name:                 runtime.FactoryAdapter(fts, schedulinggates.New),
  	}
  
  	return registry
  }
  

• 

• 注意

  kube-scheduler会根据profile配置及inTree插件选出启用的plugin和extender，profile相关配置未定义会采用默认配置

2.2.outTree插件

• outTree plugin基于scheduler framework扩展，实现时需引入scheduler core包及将plugin注册到schedulerCommand。

• func main() {
  	// 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(
          // 实现的插件注册在这里,沿用的还是scheduler core的调度能力
  		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),
  		// Sample plugins below.
  		// app.WithPlugin(crossnodepreemption.Name, crossnodepreemption.New),
  		app.WithPlugin(podstate.Name, podstate.New),
  		app.WithPlugin(qos.Name, qos.New),
  	)
  
  	code := cli.Run(command)
  	os.Exit(code)
  }
  

• 注意

  outTree是推荐的实现形式，可以作到无侵入、无网络开销，共享scheduler cache缓存

2.3.registry

• setup()会调用sched.New()初始化调度器，其中会基于回调完成scheduler conf渲染，外部未定义则采用scheme注册的默认配置。

• // New returns a Scheduler
  func New(...) (*Scheduler, error) {
  	...
  
  	options := defaultSchedulerOptions
      // callback渲染器
  	for _, opt := range opts {
          // 执行回调(包括渲染插件配置)
  		opt(&options)
  	}
  
      // 外部未定义配置,基于schema注册的默认配置渲染
  	if options.applyDefaultProfile {
  		...
          // 渲染
  		scheme.Scheme.Default(&versionedCfg)
  		...
          // 标准化
  		scheme.Scheme.Convert(&versionedCfg, &cfg, nil)
          ...
          // 回填
  		options.profiles = cfg.Profiles
  	}
  	...
  
      // 基于profiles配置及插件渲染调度器
  	profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh...)
  	...
  
  	sched := &Scheduler{...}
  	...
  
  	return sched, nil
  }
  
  // Default sets defaults on the provided Object.
  func (s *Scheme) Default(src Object) {
      // scheme渲染回调
  	if fn, ok := s.defaulterFuncs[reflect.TypeOf(src)]; ok {
  		fn(src)
  	}
  }
  
  // defaulterFuncs注册
  func (s *Scheme) AddTypeDefaultingFunc(srcType Object, fn func(interface{})) {
  	s.defaulterFuncs[reflect.TypeOf(srcType)] = fn
  }
  
  // RegisterDefaults adds defaulters functions to the given scheme.
  func RegisterDefaults(scheme *runtime.Scheme) error {
  	...
      scheme.AddTypeDefaultingFunc(&v1.KubeSchedulerConfiguration{}, func(obj interface{}) {
          // 注册默认配置回调至defaulterFuncs
  		SetObjectDefaults_KubeSchedulerConfiguration(obj.(*v1.KubeSchedulerConfiguration))
  	})
  	...
  	return nil
  }
  
  // SetDefaults_KubeSchedulerConfiguration sets additional defaults
  func SetDefaults_KubeSchedulerConfiguration(obj *configv1.KubeSchedulerConfiguration) {
  	...
  	// 未指定外部配置
  	if len(obj.Profiles) == 0 {
  		obj.Profiles = append(obj.Profiles, configv1.KubeSchedulerProfile{})
  	}
  	// Only apply a default scheduler name when there is a single profile.
  	if len(obj.Profiles) == 1 && obj.Profiles[0].SchedulerName == nil {
  		obj.Profiles[0].SchedulerName = pointer.String(v1.DefaultSchedulerName)
  	}
  
  	// Add the default set of plugins and apply the configuration.
  	for i := range obj.Profiles {
  		prof := &obj.Profiles[i]
          // 渲染配置
  		setDefaults_KubeSchedulerProfile(logger, prof)
  	}
  	...
  }
  
  func setDefaults_KubeSchedulerProfile(logger klog.Logger, prof *configv1.KubeSchedulerProfile) {
  	// 渲染默认插件列表
  	prof.Plugins = mergePlugins(logger, getDefaultPlugins(), prof.Plugins)
  	...
      
      // 外部插件配置渲染
  	for j := range prof.PluginConfig {
  		existingConfigs.Insert(prof.PluginConfig[j].Name)
  		args := prof.PluginConfig[j].Args.Object
  		...
  		scheme.Default(args)
  	}
  
  	// 默认插件配置渲染
  	for _, name := range pluginsNames(prof.Plugins) {
  		if existingConfigs.Has(name) {
  			continue
  		}
  		...
  		scheme.Default(args)
  		...
          // 记录
  		prof.PluginConfig = append(prof.PluginConfig, configv1.PluginConfig{
  			Name: name,
  			Args: runtime.RawExtension{Object: args},
  		})
  	}
  }
  
  // getDefaultPlugins returns the default set of plugins.
  func getDefaultPlugins() *v1.Plugins {
  	plugins := &v1.Plugins{
  		MultiPoint: v1.PluginSet{
  			Enabled: []v1.Plugin{
  				{Name: names.PrioritySort},
  				{Name: names.NodeUnschedulable},
  				{Name: names.NodeName},
  				{Name: names.TaintToleration, Weight: pointer.Int32(3)},
  				{Name: names.NodeAffinity, Weight: pointer.Int32(2)},
  				{Name: names.NodePorts},
  				{Name: names.NodeResourcesFit, Weight: pointer.Int32(1)},
  				{Name: names.VolumeRestrictions},
  				{Name: names.EBSLimits},
  				{Name: names.GCEPDLimits},
  				{Name: names.NodeVolumeLimits},
  				{Name: names.AzureDiskLimits},
  				{Name: names.VolumeBinding},
  				{Name: names.VolumeZone},
  				{Name: names.PodTopologySpread, Weight: pointer.Int32(2)},
  				{Name: names.InterPodAffinity, Weight: pointer.Int32(2)},
  				{Name: names.DefaultPreemption},
  				{Name: names.NodeResourcesBalancedAllocation, Weight: pointer.Int32(1)},
  				{Name: names.ImageLocality, Weight: pointer.Int32(1)},
  				{Name: names.DefaultBinder},
  			},
  		},
  	}
  	...
  
  	return plugins
  }
  

• 

• 注意

  profiles配置对应一个scheduler及关联插件和配置，优先用外部定义的调度配置，否则启用default-scheduler调度配置

3.原理

3.1.nodeName

• nodeName plugin是一个预选插件，实现了Filter方法，筛选与pod相同nodeName的node节点对象。

• type NodeName struct{}
  
  func (pl *NodeName) Name() string {
  	return Name
  }
  
  func (pl *NodeName) Filter(...) *framework.Status {
  	if nodeInfo.Node() == nil {
  		return framework.NewStatus(framework.Error, "node not found")
  	}
      // pod与节点未匹配
  	if !Fits(pod, nodeInfo) {
  		return framework.NewStatus(framework.UnschedulableAndUnresolvable, ErrReason)
  	}
  	return nil
  }
  
  func Fits(pod *v1.Pod, nodeInfo *framework.NodeInfo) bool {
  	// pod未调度或调度的节点与当前节点一致
  	return len(pod.Spec.NodeName) == 0 || pod.Spec.NodeName == nodeInfo.Node().Name
  }
  

• 注意

  nodeName插件会进行节点检查，未调度的pod会准入所有节点，显示指定nodeName的pod仅会匹配对应节点

3.2.imageLocality

• imageLocality plugin实现score接口，用于计算pod关联的镜像位于node的状态，经过校验及规则计算node得分以考虑镜像拉取成本。

• type ImageLocality struct {
  	handle framework.Handle
  }
  
  func (pl *ImageLocality) Score(...) (int64, *framework.Status) {
  	// 通过nodeName获取nodeInfo对象
  	nodeInfo, err := pl.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  	...
  
  	// 获取节点列表
  	nodeInfos, err := pl.handle.SnapshotSharedLister().NodeInfos().List()
  	...
  
  	// 当前有多少node
  	totalNumNodes := len(nodeInfos)
  
  	// 计算出node对应的分值
  	score := calculatePriority(sumImageScores(nodeInfo,pod.Spec.Containers,totalNumNodes), len(pod.Spec.Containers))
  
  	return score, nil
  }
  
  // 分值标准化
  func calculatePriority(sumScores int64, numContainers int) int64 {
      // 分值上限
  	maxThreshold := maxContainerThreshold * int64(numContainers)
  	if sumScores < minThreshold {
  		sumScores = minThreshold
  	} else if sumScores > maxThreshold {
  		sumScores = maxThreshold
  	}
  
      // 计算分值——100*(sum-min)/(max-min)
  	return int64(framework.MaxNodeScore) * (sumScores - minThreshold) / (maxThreshold - minThreshold)
  }
  
  // 总分计算
  func sumImageScores(nodeInfo *framework.NodeInfo, containers []v1.Container, totalNumNodes int) int64 {
  	...
      // 遍历pod容器
  	for _, container := range containers {
  		// node已拉取容器镜像
  		if state, ok := nodeInfo.ImageStates[normalizedImageName(container.Image)]; ok {
              // 累加分值
  			sum += scaledImageScore(state, totalNumNodes)
  		}
  	}
  	return sum
  }
  
  // scaledImageScore returns an adaptively scaled score for the given state of an image.
  func scaledImageScore(imageState *framework.ImageStateSummary, totalNumNodes int) int64 {
      // 计算镜像对当前节点得分贡献
  	spread := float64(imageState.NumNodes) / float64(totalNumNodes)
      // size*spread计算镜像得分
  	return int64(float64(imageState.Size) * spread)
  }
  

• 

• 注意

  pod containers的镜像对节点的贡献最终会转为分值参与调度

3.3.nodeResourcesFit

• nodeResourcesFit plugin实现了preFilter、Filter、preScore和Score插件，用于资源检查、节点过滤及可选节点打分。

• type Fit struct {
  	...
  	handle                          framework.Handle
  	resourceAllocationScorer
  }
  
  // preFilter实现
  func (f *Fit) PreFilter(...) (*framework.PreFilterResult, *framework.Status) {
      // 记录pod申请的资源
  	cycleState.Write(preFilterStateKey, computePodResourceRequest(pod))
  	return nil, nil
  }
  
  // filter相关
  func (f *Fit) Filter(...) *framework.Status {
      // 获取记录的申请资源
  	s, err := getPreFilterState(cycleState)
  	...
  	
      // 检查node资源是否满足pod申请需求
  	insufficientResources := fitsRequest(s, nodeInfo, f.ignoredResources, f.ignoredResourceGroups)
  	...
  	return nil
  }
  
  // preScore相关
  func (f *Fit) PreScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodes []*v1.Node) *framework.Status {
  	state := &preScoreState{
          // 计算参与打分的pod申请资源
  		podRequests: f.calculatePodResourceRequestList(pod, f.resources),
  	}
      
      // 记录到调度周期
  	cycleState.Write(preScoreStateKey, state)
  	return nil
  }
  
  // score相关
  func (f *Fit) Score(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
      // 获取节点
  	nodeInfo, err := f.handle.SnapshotSharedLister().NodeInfos().Get(nodeName)
  	...
  
      // 获取参与打分的pod申请资源
  	s, err := getPreScoreState(state)
  	...
  
      // 打分
  	return f.score(pod, nodeInfo, s.podRequests)
  }
  
  // score will use `scorer` function to calculate the score.
  func (r *resourceAllocationScorer) score(...) (int64, *framework.Status) {
  	node := nodeInfo.Node()
  	...
  
      // 参与打分的资源
  	for i := range r.resources {
          // 计算node可分配及pod申请资源
  		alloc, req := r.calculateResourceAllocatableRequest(nodeInfo, v1.ResourceName(r.resources[i].Name), podRequests[i])
  		// 扩展资源不参与分值计算
  		if alloc == 0 {
  			continue
  		}
          
  		allocatable[i] = alloc
  		requested[i] = req
  	}
  
      // 基于公式打分
  	score := r.scorer(requested, allocatable)
  	...
  
  	return score, nil
  }
  

• 

• 注意

  kube-scheduler内置的score插件不好理解，无法推断打分公式基于什么场景得来的

4.4.nodeAffinity

• NodeAffinity plugin实现了preFilter、Filter、preScore和Score插件，用于pod调度期间的节点亲和性计算。

• type NodeAffinity struct {
  	handle              framework.Handle
  	addedNodeSelector   *nodeaffinity.NodeSelector
  	addedPrefSchedTerms *nodeaffinity.PreferredSchedulingTerms
  }
  
  // 硬亲和性检查
  func (pl *NodeAffinity) PreFilter(...) (*framework.PreFilterResult, *framework.Status) {
      // pod定义的亲和性
  	affinity := pod.Spec.Affinity
  	...
  
      // 硬亲和(nodeSelector和requiredNodeAffinity)
  	state := &preFilterState{requiredNodeSelectorAndAffinity: nodeaffinity.GetRequiredNodeAffinity(pod)}
      
      // 记录到调度周期
  	cycleState.Write(preFilterStateKey, state)
  	...
      
      // 筛选matchFields的节点名作为预选
      if len(nodeNames) > 0 {
  		return &framework.PreFilterResult{NodeNames: nodeNames}, nil
  	}
  	return nil, nil
  
  }
  
  // node亲和条件检查
  func (pl *NodeAffinity) Filter(...) *framework.Status {
  	node := nodeInfo.Node()
  	...
      
      // 插件硬亲和限制
      if pl.addedNodeSelector != nil && !pl.addedNodeSelector.Match(node) {
  		return framework.NewStatus(framework.UnschedulableAndUnresolvable, errReasonEnforced)
  	}
  
  	// 获取pod硬亲和条件
  	s, err := getPreFilterState(state)
  	if err != nil {
          // 未找到直接由pod再次获取
  		s = &preFilterState{requiredNodeSelectorAndAffinity: nodeaffinity.GetRequiredNodeAffinity(pod)}
  	}
  
  	// 匹配硬亲和和node
  	match, _ := s.requiredNodeSelectorAndAffinity.Match(node)
  	if !match {
  		// 不满足直接退出
  		return framework.NewStatus(framework.UnschedulableAndUnresolvable, ErrReasonPod)
  	}
  
  	return nil
  }
  
  // 软亲和性检查
  func (pl *NodeAffinity) PreScore(...) *framework.Status {
  	...
      
      // 获取pod软亲和
  	preferredNodeAffinity, err := getPodPreferredNodeAffinity(pod)
  	...
      
  	state := &preScoreState{ preferredNodeAffinity: preferredNodeAffinity }
      
      // 记录到调度周期
  	cycleState.Write(preScoreStateKey, state)
  	return nil
  }
  
  // node亲和打分
  func (pl *NodeAffinity) Score(...) (int64, *framework.Status) {
  	...
  
  	// 插件软亲和得分
  	if pl.addedPrefSchedTerms != nil {
  		count += pl.addedPrefSchedTerms.Score(node)
  	}
  
  	// 获取pod的软亲和条件
  	s, err := getPreScoreState(state)
      // 未成功
  	if err != nil {
  		// 尝试由pod再次获取
  		preferredNodeAffinity, err := getPodPreferredNodeAffinity(pod)
  		...
  		s = &preScoreState{
  			preferredNodeAffinity: preferredNodeAffinity,
  		}
  	}
  
  	// 根据pod的软亲和条件对节点打分
  	if s.preferredNodeAffinity != nil {
  		count += s.preferredNodeAffinity.Score(node)
  	}
  
  	return count, nil
  }
  

• 

• 注意

  这里介绍的都是预选和优选节点插件，绑定阶段类似的，例如volumeBinder实现的preBind plugin，主要触发动态PVC对应PV创建及绑定