彻底搞懂kubernetes调度框架与插件
调度框架 kubernetes 1.24 进行分析
调度框架（Scheduling Framework）是Kubernetes 的调度器 kube-scheduler 设计的的可插拔架构，将插件（调度算法）嵌入到调度上下文的每个扩展点中，并编译为 kube-scheduler
在 kube-scheduler 1.22 之后，在 pkg/scheduler/framework/interface.go 中定义了一个 Plugin 的 interface，这个 interface 作为了所有插件的父级。而每个未调度的 Pod，Kubernetes 调度器会根据一组规则尝试在集群中寻找一个节点。
type Plugin interface {
Name() string
}
下面会对每个算法是如何实现的进行分析
在初始化 scheduler 时，会创建一个 profile，profile是关于 scheduler 调度配置相关的定义
func New(client clientset.Interface,
...
profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
frameworkruntime.WithClientSet(client),
frameworkruntime.WithKubeConfig(options.kubeConfig),
frameworkruntime.WithInformerFactory(informerFactory),
frameworkruntime.WithSnapshotSharedLister(snapshot),
frameworkruntime.WithPodNominator(nominator),
frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
frameworkruntime.WithClusterEventMap(clusterEventMap),
frameworkruntime.WithParallelism(int(options.parallelism)),
frameworkruntime.WithExtenders(extenders),
)
if err != nil {
return nil, fmt.Errorf("initializing profiles: %v", err)
}
if len(profiles) == 0 {
return nil, errors.New("at least one profile is required")
}
....
}
关于 profile 的实现，则为 KubeSchedulerProfile，也是作为 yaml生成时传入的配置
// KubeSchedulerProfile 是一个 scheduling profile.
type KubeSchedulerProfile struct {
// SchedulerName 是与此配置文件关联的调度程序的名称。
// 如果 SchedulerName 与 pod “spec.schedulerName”匹配，则使用此配置文件调度 pod。
SchedulerName string
// Plugins指定应该启用或禁用的插件集。
// 启用的插件是除了默认插件之外应该启用的插件。禁用插件应是禁用的任何默认插件。
// 当没有为扩展点指定启用或禁用插件时，将使用该扩展点的默认插件（如果有）。
// 如果指定了 QueueSort 插件，
/// 则必须为所有配置文件指定相同的 QueueSort Plugin 和 PluginConfig。
// 这个Plugins展现的形式则是调度上下文中的所有扩展点(这是抽象)，实际中会表现为多个扩展点
Plugins *Plugins
// PluginConfig 是每个插件的一组可选的自定义插件参数。
// 如果省略PluginConfig参数等同于使用该插件的默认配置。
PluginConfig []PluginConfig
}
对于 profile.NewMap 就是根据给定的配置来构建这个framework，因为配置可能是存在多个的。而 Registry 则是所有可用插件的集合，内部构造则是 PluginFactory ,通过函数来构建出对应的 plugin
func NewMap(cfgs []config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (Map, error) {
m := make(Map)
v := cfgValidator{m: m}
for _, cfg := range cfgs {
p, err := newProfile(cfg, r, recorderFact, stopCh, opts...)
if err != nil {
return nil, fmt.Errorf("creating profile for scheduler name %s: %v", cfg.SchedulerName, err)
}
if err := v.validate(cfg, p); err != nil {
return nil, err
}
m[cfg.SchedulerName] = p
}
return m, nil
}
// newProfile 给的配置构建出一个profile
func newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
recorder := recorderFact(cfg.SchedulerName)
opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}
可以看到最终返回的是一个 Framework 。那么来看下这个 Framework
Framework 是一个抽象，管理着调度过程中所使用的所有插件，并在调度上下文中适当的位置去运行对应的插件
type Framework interface {
Handle
// QueueSortFunc 返回对调度队列中的 Pod 进行排序的函数
// 也就是less，在Sort打分阶段的打分函数
QueueSortFunc() LessFunc
// RunPreFilterPlugins 运行配置的一组PreFilter插件。
// 如果这组插件中，任何一个插件失败，则返回 *Status 并设置为non-success。
// 如果返回状态为non-success，则调度周期中止。
// 它还返回一个 PreFilterResult，它可能会影响到要评估下游的节点。
RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (*PreFilterResult, *Status)
// RunPostFilterPlugins 运行配置的一组PostFilter插件。
// PostFilter 插件是通知性插件，在这种情况下应配置为先执行并返回 Unschedulable 状态，
// 或者尝试更改集群状态以使 pod 在未来的调度周期中可能会被调度。
RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)
// RunPreBindPlugins 运行配置的一组 PreBind 插件。
// 如果任何一个插件返回错误，则返回 *Status 并且code设置为non-success。
// 如果code为“Unschedulable”，则调度检查失败，
// 则认为是内部错误。在任何一种情况下，Pod都不会被bound。
RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// RunPostBindPlugins 运行配置的一组PostBind插件
RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// RunReservePluginsReserve运行配置的一组Reserve插件的Reserve方法。
// 如果在这组调用中的任何一个插件返回错误，则不会继续运行剩余调用的插件并返回错误。
// 在这种情况下，pod将不能被调度。
RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// RunReservePluginsUnreserve运行配置的一组Reserve插件的Unreserve方法。
RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// RunPermitPlugins运行配置的一组Permit插件。
// 如果这些插件中的任何一个返回“Success”或“Wait”之外的状态，则它不会继续运行其余插件并返回错误。
// 否则，如果任何插件返回 “Wait”，则此函数将创建等待pod并将其添加到当前等待pod的map中，
// 并使用“Wait” code返回状态。 Pod将在Permit插件返回的最短持续时间内保持等待pod。
RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// 如果pod是waiting pod，WaitOnPermit 将阻塞，直到等待的pod被允许或拒绝。
WaitOnPermit(ctx context.Context, pod *v1.Pod) *Status
// RunBindPlugins运行配置的一组bind插件。 Bind插件可以选择是否处理Pod。
// 如果 Bind 插件选择跳过binding，它应该返回 code=5("skip")状态。
// 否则，它应该返回“Error”或“Success”。
// 如果没有插件处理绑定，则RunBindPlugins返回code=5("skip")的状态。
RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// 如果至少定义了一个filter插件，则HasFilterPlugins返回true
HasFilterPlugins() bool
// 如果至少定义了一个PostFilter插件，则HasPostFilterPlugins返回 true。
HasPostFilterPlugins() bool
// 如果至少定义了一个Score插件，则HasScorePlugins返回 true。
HasScorePlugins() bool
// ListPlugins将返回map。key为扩展点名称，value则是配置的插件列表。
ListPlugins() *config.Plugins
// ProfileName则是与profile name关联的framework
ProfileName() string
}
而实现这个抽象的则是 frameworkImpl；frameworkImpl 是初始化与运行 scheduler plugins 的组件，并在调度上下文中会运行这些扩展点
type frameworkImpl struct {
registry Registry
snapshotSharedLister framework.SharedLister
waitingPods *waitingPodsMap
scorePluginWeight map[string]int
queueSortPlugins []framework.QueueSortPlugin
preFilterPlugins []framework.PreFilterPlugin
filterPlugins []framework.FilterPlugin
postFilterPlugins []framework.PostFilterPlugin
preScorePlugins []framework.PreScorePlugin
scorePlugins []framework.ScorePlugin
reservePlugins []framework.ReservePlugin
preBindPlugins []framework.PreBindPlugin
bindPlugins []framework.BindPlugin
postBindPlugins []framework.PostBindPlugin
permitPlugins []framework.PermitPlugin
clientSet clientset.Interface
kubeConfig *restclient.Config
eventRecorder events.EventRecorder
informerFactory informers.SharedInformerFactory
metricsRecorder *metricsRecorder
profileName string
extenders []framework.Extender
framework.PodNominator
parallelizer parallelize.Parallelizer
}
那么来看下 Registry ，Registry 是作为一个可用插件的集合。framework 使用 registry 来启用和对插件配置的初始化。在初始化框架之前，所有插件都必须在注册表中。表现形式就是一个 map[]；key 是插件的名称，value是 PluginFactory 。
type Registry map[string]PluginFactory
而在 pkg\scheduler\framework\plugins\registry.go 中会将所有的 in-tree plugin 注册进来。通过 NewInTreeRegistry 。后续如果还有插件要注册，可以通过 WithFrameworkOutOfTreeRegistry 来注册其他的插件。
func NewInTreeRegistry() runtime.Registry {
fts := plfeature.Features{
EnableReadWriteOncePod: feature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
EnableVolumeCapacityPriority: feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority),
EnableMinDomainsInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.MinDomainsInPodTopologySpread),
EnableNodeInclusionPolicyInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.NodeInclusionPolicyInPodTopologySpread),
}
return runtime.Registry{
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),
}
}
这里插入一个题外话，关于 in-tree plugin
在这里没有找到关于，kube-scheduler ，只是找到有关的概念，大概可以解释为，in-tree表示为随kubernetes官方提供的二进制构建的 plugin 则为 in-tree，而独立于kubernetes代码库之外的为 out-of-tree scheduleOne 中可以很好的看出，功能都是 framework 提供的。
func (sched *Scheduler) scheduleOne(ctx context.Context) {
...
scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod)
...
// Run the Reserve method of reserve plugins.
if sts := fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
}
...
// Run "permit" plugins.
runPermitStatus := fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
// One of the plugins returned status different than success or wait.
fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
go func() {
...
waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
if !waitOnPermitStatus.IsSuccess() {
...
// trigger un-reserve plugins to clean up state associated with the reserved Pod
fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
}
// Run "prebind" plugins.
preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
// trigger un-reserve plugins to clean up state associated with the reserved Pod
fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
...
// trigger un-reserve plugins to clean up state associated with the reserved Pod
fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
// Run "postbind" plugins.
fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
}
插件 kubernetes/cmd/kube-scheduler/app/server.go
profile.NewMap：kubernetes/pkg/scheduler/scheduler.go
newProfile：kubernetes/pkg/scheduler/scheduler.go
frameworkruntime.NewFramework：kubernetes/pkg/scheduler/framework/runtime/framework.go
pluginsNeeded：kubernetes/pkg/scheduler/framework/runtime/framework.go
NewScheduler
我们了解如何 New 一个 scheduler 即为 Setup 中去配置这些参数，
func Setup(ctx context.Context, opts *options.Options, outOfTreeRegistryOptions ...Option) (*schedulerserverconfig.CompletedConfig, *scheduler.Scheduler, error) {
...
// Create the scheduler.
sched, err := scheduler.New(cc.Client,
cc.InformerFactory,
cc.DynInformerFactory,
recorderFactory,
ctx.Done(),
scheduler.WithComponentConfigVersion(cc.ComponentConfig.TypeMeta.APIVersion),
scheduler.WithKubeConfig(cc.KubeConfig),
scheduler.WithProfiles(cc.ComponentConfig.Profiles...),
scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
scheduler.WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),
scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),
scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),
scheduler.WithPodMaxInUnschedulablePodsDuration(cc.PodMaxInUnschedulablePodsDuration),
scheduler.WithExtenders(cc.ComponentConfig.Extenders...),
scheduler.WithParallelism(cc.ComponentConfig.Parallelism),
scheduler.WithBuildFrameworkCapturer(func(profile kubeschedulerconfig.KubeSchedulerProfile) {
// Profiles are processed during Framework instantiation to set default plugins and configurations. Capturing them for logging
completedProfiles = append(completedProfiles, profile)
}),
)
...
}
profile.NewMap
在 scheduler.New 中，会根据配置生成profile，而 profile.NewMap 会完成这一步
func New(client clientset.Interface,
...
clusterEventMap := make(map[framework.ClusterEvent]sets.String)
profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
frameworkruntime.WithClientSet(client),
frameworkruntime.WithKubeConfig(options.kubeConfig),
frameworkruntime.WithInformerFactory(informerFactory),
frameworkruntime.WithSnapshotSharedLister(snapshot),
frameworkruntime.WithPodNominator(nominator),
frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
frameworkruntime.WithClusterEventMap(clusterEventMap),
frameworkruntime.WithParallelism(int(options.parallelism)),
frameworkruntime.WithExtenders(extenders),
)
...
}
NewFramework
newProfile 返回的则是一个创建好的 framework
func newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
recorder := recorderFact(cfg.SchedulerName)
opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}
最终会走到 pluginsNeeded，这里会根据配置中开启的插件而返回一个插件集，这个就是最终在每个扩展点中药执行的插件。
func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) sets.String {
pgSet := sets.String{}
if plugins == nil {
return pgSet
}
find := func(pgs *config.PluginSet) {
for _, pg := range pgs.Enabled {
pgSet.Insert(pg.Name)
}
}
// 获取到所有的扩展点，找到为Enabled的插件加入到pgSet
for _, e := range f.getExtensionPoints(plugins) {
find(e.plugins)
}
// Parse MultiPoint separately since they are not returned by f.getExtensionPoints()
find(&plugins.MultiPoint)
return pgSet
}
插件的执行
在对插件源码部分分析，会找几个典型的插件进行分析，而不会对全部的进行分析，因为总的来说是大同小异，分析的插件有 NodePorts，NodeResourcesFit，podtopologyspread
NodePorts
这里以一个简单的插件来分析；NodePorts 插件用于检查Pod请求的端口，在节点上是否为空闲端口。
NodePorts 实现了 FilterPlugin 和 PreFilterPlugin
PreFilter 将会被 framework 中 PreFilter 扩展点被调用。
func (pl *NodePorts) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s := getContainerPorts(pod) // 或得Pod得端口
// 写入状态
cycleState.Write(preFilterStateKey, preFilterState(s))
return nil, nil
}
Filter 将会被 framework 中 Filter 扩展点被调用。
// Filter invoked at the filter extension point.
func (pl *NodePorts) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
wantPorts, err := getPreFilterState(cycleState)
if err != nil {
return framework.AsStatus(err)
}
fits := fitsPorts(wantPorts, nodeInfo)
if !fits {
return framework.NewStatus(framework.Unschedulable, ErrReason)
}
return nil
}
func fitsPorts(wantPorts []*v1.ContainerPort, nodeInfo *framework.NodeInfo) bool {
// 对比existingPorts 和 wantPorts是否冲突，冲突则调度失败
existingPorts := nodeInfo.UsedPorts
for _, cp := range wantPorts {
if existingPorts.CheckConflict(cp.HostIP, string(cp.Protocol), cp.HostPort) {
return false
}
}
return true
}
New ，初始化新插件，在 register 中注册得
func New(_ runtime.Object, _ framework.Handle) (framework.Plugin, error) {
return &NodePorts{}, nil
}
在调用中，如果有任何一个插件返回错误，则跳过该扩展点注册得其他插件，返回失败。
func (f *frameworkImpl) RunFilterPlugins(
ctx context.Context,
state *framework.CycleState,
pod *v1.Pod,
nodeInfo *framework.NodeInfo,
) framework.PluginToStatus {
statuses := make(framework.PluginToStatus)
for _, pl := range f.filterPlugins {
pluginStatus := f.runFilterPlugin(ctx, pl, state, pod, nodeInfo)
if !pluginStatus.IsSuccess() {
if !pluginStatus.IsUnschedulable()
errStatus := framework.AsStatus(fmt.Errorf("running %q filter plugin: %w", pl.Name(), pluginStatus.AsError())).WithFailedPlugin(pl.Name())
return map[string]*framework.Status{pl.Name(): errStatus}
}
pluginStatus.SetFailedPlugin(pl.Name())
statuses[pl.Name()] = pluginStatus
}
}
return statuses
}
返回得状态是一个 Status 结构体，该结构体表示了插件运行的结果。由 Code、reasons、（可选）err 和 failedPlugin （失败的那个插件名）组成。当 code 不是 Success 时，应说明原因。而且，当 code 为 Success 时，其他所有字段都应为空。nil 状态也被视为成功。
type Status struct {
code Code
reasons []string
err error
// failedPlugin is an optional field that records the plugin name a Pod failed by.
// It's set by the framework when code is Error, Unschedulable or UnschedulableAndUnresolvable.
failedPlugin string
}
NodeResourcesFit computePodResourceRequest 这里有一个注释，总体解释起来是这样得：computePodResourceRequest ，返回值（ framework.Resource）覆盖了每一个维度中资源的最大宽度。因为将按照 init-containers , containers 得顺序运行，会通过迭代方式收集每个维度中的最大值。计算时会对常规容器的资源向量求和，因为containers 运行会同时运行多个容器。计算示例为：
Pod:
InitContainers
IC1:
CPU: 2
Memory: 1G
IC2:
CPU: 2
Memory: 3G
Containers
C1:
CPU: 2
Memory: 1G
C2:
CPU: 1
Memory: 1G
在维度1中（InitContainers）所需资源最大值时，CPU=2, Memory=3G；而维度2（Containers）所需资源最大值为：CPU=2, Memory=1G；那么最终结果为 CPU=3, Memory=3G，因为在维度1，最大资源时Memory=3G；而维度2最大资源是CPU=1+2, Memory=1+1，取每个维度中最大资源最大宽度即为 CPU=3, Memory=3G。
下面则看下代码得实现
func computePodResourceRequest(pod *v1.Pod) *preFilterState {
result := &preFilterState{}
for _, container := range pod.Spec.Containers {
result.Add(container.Resources.Requests)
}
// 取最大得资源
for _, container := range pod.Spec.InitContainers {
result.SetMaxResource(container.Resources.Requests)
}
// 如果Overhead正在使用，需要将其计算到总资源中
if pod.Spec.Overhead != nil {
result.Add(pod.Spec.Overhead)
}
return result
}
// SetMaxResource 是比较ResourceList并为每个资源取最大值。
func (r *Resource) SetMaxResource(rl v1.ResourceList) {
if r == nil {
return
}
for rName, rQuantity := range rl {
switch rName {
case v1.ResourceMemory:
r.Memory = max(r.Memory, rQuantity.Value())
case v1.ResourceCPU:
r.MilliCPU = max(r.MilliCPU, rQuantity.MilliValue())
case v1.ResourceEphemeralStorage:
if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
r.EphemeralStorage = max(r.EphemeralStorage, rQuantity.Value())
}
default:
if schedutil.IsScalarResourceName(rName) {
r.SetScalar(rName, max(r.ScalarResources[rName], rQuantity.Value()))
}
}
}
}
leastAllocate
LeastAllocated 是 NodeResourcesFit 的打分策略 ，LeastAllocated 打分的标准是更偏向于请求资源较少的Node。将会先计算出Node上调度的pod请求的内存、CPU与其他资源的百分比，然后并根据请求的比例与容量的平均值的最小值进行优先级排序。
计算公式是这样的：\(\frac{\frac{cpu((capacity-requested) \times MaxNodeScore \times cpuWeight)}{capacity} + \frac{memory((capacity-requested) \times MaxNodeScore \times memoryWeight}{capacity}) + ...}{weightSum}\)
下面来看下实现
func leastResourceScorer(resToWeightMap resourceToWeightMap) func(resourceToValueMap, resourceToValueMap) int64 {
return func(requested, allocable resourceToValueMap) int64 {
var nodeScore, weightSum int64
for resource := range requested {
weight := resToWeightMap[resource]
// 计算出的资源分数乘weight
resourceScore := leastRequestedScore(requested[resource], allocable[resource])
nodeScore += resourceScore * weight
weightSum += weight
}
if weightSum == 0 {
return 0
}
// 最终除weightSum
return nodeScore / weightSum
}
}
leastRequestedScore 计算标准为未使用容量的计算范围为 0~MaxNodeScore，0 为最低优先级，MaxNodeScore 为最高优先级。未使用的资源越多，得分越高。
func leastRequestedScore(requested, capacity int64) int64 {
if capacity == 0 {
return 0
}
if requested > capacity {
return 0
}
// 容量 - 请求的 x 预期值（100）/ 容量
return ((capacity - requested) * int64(framework.MaxNodeScore)) / capacity
}
Topology pkg\scheduler\framework\plugins\podtopologyspread\plugin.go
var systemDefaultConstraints = []v1.TopologySpreadConstraint{
{
TopologyKey: v1.LabelHostname,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 3,
},
{
TopologyKey: v1.LabelTopologyZone,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 5,
},
}
可以通过在配置文件中留空，来禁用默认配置
defaultConstraints: []defaultingType: List
apiVersion: kubescheduler.config.k8s.io/v1beta3
kind: KubeSchedulerConfiguration
profiles:
- schedulerName: default-scheduler
pluginConfig:
- name: PodTopologySpread
args:
defaultConstraints: []
defaultingType: List
通过源码学习Topology
podtopologyspread 实现了4种扩展点方法，包含 filter 和 score
PreFilter
可以看到 PreFilter 的核心为 calPreFilterState
func (pl *PodTopologySpread) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s, err := pl.calPreFilterState(ctx, pod)
if err != nil {
return nil, framework.AsStatus(err)
}
cycleState.Write(preFilterStateKey, s)
return nil, nil
}
calPreFilterState 主要功能是用在计算如何在拓扑域中分布Pod，首先看段代码时，需要掌握下属几个概念
- update 函数实际上时用于计算 preFilter 中 最后的计算结果会保存在
CycleState 中
cycleState.Write(preFilterStateKey, s)
Filter 主要是从 PreFilter 处理的过程中拿到状态 preFilterState，然后看下每个拓扑约束中的 MaxSkew 是否合法，具体的计算公式为：\(matchNum + selfMatchNum - minMatchNum\)
matchNum：Prefilter 中计算出的对应的拓扑分布数量，可以在Framework 中会运行 ScoreExtension ，即 NormalizeScore
// Run NormalizeScore method for each ScorePlugin in parallel.
f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
nodeScoreList := pluginToNodeScores[pl.Name()]
if pl.ScoreExtensions() == nil {
return
}
status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
if !status.IsSuccess() {
err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
errCh.SendErrorWithCancel(err, cancel)
return
}
})
if err := errCh.ReceiveError(); err != nil {
return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
}
NormalizeScore 会为所有的node根据之前计算出的权重进行打分
func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
s, err := getPreScoreState(cycleState)
if err != nil {
return framework.AsStatus(err)
}
if s == nil {
return nil
}
// 计算 和
var minScore int64 = math.MaxInt64
var maxScore int64
for i, score := range scores {
// it's mandatory to check if is present in m.IgnoredNodes
if s.IgnoredNodes.Has(score.Name) {
scores[i].Score = invalidScore
continue
}
if score.Score < minScore {
minScore = score.Score
}
if score.Score > maxScore {
maxScore = score.Score
}
}
for i := range scores {
if scores[i].Score == invalidScore {
scores[i].Score = 0
continue
}
if maxScore == 0 {
scores[i].Score = framework.MaxNodeScore
continue
}
s := scores[i].Score
scores[i].Score = framework.MaxNodeScore * (maxScore + minScore - s) / maxScore
}
return nil
}
到此，对于pod拓扑插件功能大概可以明了了，
- Filter 部分（
PreFilter，Filter）完成拓扑对(Topology Pair)划分
- Score部分（
PreScore, Score , NormalizeScore ）主要是对拓扑对（可以理解为拓扑结构划分）来选择一个最适合的pod的节点（即分数最优的节点）
而在 scoring_test.go 给了很多用例，可以更深入的了解这部分算法
Reference
[1] scheduling code hierarchy
[2] scheduler algorithm
[3] in tree VS out of tree volume plugins
[4] scheduler_framework_plugins
[5] scheduling config
[6] topology spread constraints
Scheduling Framework）是Kubernetes 的调度器 kube-scheduler 设计的的可插拔架构，将插件（调度算法）嵌入到调度上下文的每个扩展点中，并编译为 kube-schedulerkube-scheduler 1.22 之后，在 pkg/scheduler/framework/interface.go 中定义了一个 Plugin 的 interface，这个 interface 作为了所有插件的父级。而每个未调度的 Pod，Kubernetes 调度器会根据一组规则尝试在集群中寻找一个节点。type Plugin interface {
Name() string
}
profile，profile是关于 scheduler 调度配置相关的定义func New(client clientset.Interface,
...
profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
frameworkruntime.WithClientSet(client),
frameworkruntime.WithKubeConfig(options.kubeConfig),
frameworkruntime.WithInformerFactory(informerFactory),
frameworkruntime.WithSnapshotSharedLister(snapshot),
frameworkruntime.WithPodNominator(nominator),
frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
frameworkruntime.WithClusterEventMap(clusterEventMap),
frameworkruntime.WithParallelism(int(options.parallelism)),
frameworkruntime.WithExtenders(extenders),
)
if err != nil {
return nil, fmt.Errorf("initializing profiles: %v", err)
}
if len(profiles) == 0 {
return nil, errors.New("at least one profile is required")
}
....
}
profile 的实现，则为 KubeSchedulerProfile，也是作为 yaml生成时传入的配置// KubeSchedulerProfile 是一个 scheduling profile.
type KubeSchedulerProfile struct {
// SchedulerName 是与此配置文件关联的调度程序的名称。
// 如果 SchedulerName 与 pod “spec.schedulerName”匹配，则使用此配置文件调度 pod。
SchedulerName string
// Plugins指定应该启用或禁用的插件集。
// 启用的插件是除了默认插件之外应该启用的插件。禁用插件应是禁用的任何默认插件。
// 当没有为扩展点指定启用或禁用插件时，将使用该扩展点的默认插件（如果有）。
// 如果指定了 QueueSort 插件，
/// 则必须为所有配置文件指定相同的 QueueSort Plugin 和 PluginConfig。
// 这个Plugins展现的形式则是调度上下文中的所有扩展点(这是抽象)，实际中会表现为多个扩展点
Plugins *Plugins
// PluginConfig 是每个插件的一组可选的自定义插件参数。
// 如果省略PluginConfig参数等同于使用该插件的默认配置。
PluginConfig []PluginConfig
}
profile.NewMap 就是根据给定的配置来构建这个framework，因为配置可能是存在多个的。而 Registry 则是所有可用插件的集合，内部构造则是 PluginFactory ,通过函数来构建出对应的 pluginfunc NewMap(cfgs []config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (Map, error) {
m := make(Map)
v := cfgValidator{m: m}
for _, cfg := range cfgs {
p, err := newProfile(cfg, r, recorderFact, stopCh, opts...)
if err != nil {
return nil, fmt.Errorf("creating profile for scheduler name %s: %v", cfg.SchedulerName, err)
}
if err := v.validate(cfg, p); err != nil {
return nil, err
}
m[cfg.SchedulerName] = p
}
return m, nil
}
// newProfile 给的配置构建出一个profile
func newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
recorder := recorderFact(cfg.SchedulerName)
opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}
Framework 。那么来看下这个 Frameworktype Framework interface {
Handle
// QueueSortFunc 返回对调度队列中的 Pod 进行排序的函数
// 也就是less，在Sort打分阶段的打分函数
QueueSortFunc() LessFunc
// RunPreFilterPlugins 运行配置的一组PreFilter插件。
// 如果这组插件中，任何一个插件失败，则返回 *Status 并设置为non-success。
// 如果返回状态为non-success，则调度周期中止。
// 它还返回一个 PreFilterResult，它可能会影响到要评估下游的节点。
RunPreFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod) (*PreFilterResult, *Status)
// RunPostFilterPlugins 运行配置的一组PostFilter插件。
// PostFilter 插件是通知性插件，在这种情况下应配置为先执行并返回 Unschedulable 状态，
// 或者尝试更改集群状态以使 pod 在未来的调度周期中可能会被调度。
RunPostFilterPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, filteredNodeStatusMap NodeToStatusMap) (*PostFilterResult, *Status)
// RunPreBindPlugins 运行配置的一组 PreBind 插件。
// 如果任何一个插件返回错误，则返回 *Status 并且code设置为non-success。
// 如果code为“Unschedulable”，则调度检查失败，
// 则认为是内部错误。在任何一种情况下，Pod都不会被bound。
RunPreBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// RunPostBindPlugins 运行配置的一组PostBind插件
RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// RunReservePluginsReserve运行配置的一组Reserve插件的Reserve方法。
// 如果在这组调用中的任何一个插件返回错误，则不会继续运行剩余调用的插件并返回错误。
// 在这种情况下，pod将不能被调度。
RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// RunReservePluginsUnreserve运行配置的一组Reserve插件的Unreserve方法。
RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// RunPermitPlugins运行配置的一组Permit插件。
// 如果这些插件中的任何一个返回“Success”或“Wait”之外的状态，则它不会继续运行其余插件并返回错误。
// 否则，如果任何插件返回 “Wait”，则此函数将创建等待pod并将其添加到当前等待pod的map中，
// 并使用“Wait” code返回状态。 Pod将在Permit插件返回的最短持续时间内保持等待pod。
RunPermitPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// 如果pod是waiting pod，WaitOnPermit 将阻塞，直到等待的pod被允许或拒绝。
WaitOnPermit(ctx context.Context, pod *v1.Pod) *Status
// RunBindPlugins运行配置的一组bind插件。 Bind插件可以选择是否处理Pod。
// 如果 Bind 插件选择跳过binding，它应该返回 code=5("skip")状态。
// 否则，它应该返回“Error”或“Success”。
// 如果没有插件处理绑定，则RunBindPlugins返回code=5("skip")的状态。
RunBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// 如果至少定义了一个filter插件，则HasFilterPlugins返回true
HasFilterPlugins() bool
// 如果至少定义了一个PostFilter插件，则HasPostFilterPlugins返回 true。
HasPostFilterPlugins() bool
// 如果至少定义了一个Score插件，则HasScorePlugins返回 true。
HasScorePlugins() bool
// ListPlugins将返回map。key为扩展点名称，value则是配置的插件列表。
ListPlugins() *config.Plugins
// ProfileName则是与profile name关联的framework
ProfileName() string
}
frameworkImpl 是初始化与运行 scheduler plugins 的组件，并在调度上下文中会运行这些扩展点type frameworkImpl struct {
registry Registry
snapshotSharedLister framework.SharedLister
waitingPods *waitingPodsMap
scorePluginWeight map[string]int
queueSortPlugins []framework.QueueSortPlugin
preFilterPlugins []framework.PreFilterPlugin
filterPlugins []framework.FilterPlugin
postFilterPlugins []framework.PostFilterPlugin
preScorePlugins []framework.PreScorePlugin
scorePlugins []framework.ScorePlugin
reservePlugins []framework.ReservePlugin
preBindPlugins []framework.PreBindPlugin
bindPlugins []framework.BindPlugin
postBindPlugins []framework.PostBindPlugin
permitPlugins []framework.PermitPlugin
clientSet clientset.Interface
kubeConfig *restclient.Config
eventRecorder events.EventRecorder
informerFactory informers.SharedInformerFactory
metricsRecorder *metricsRecorder
profileName string
extenders []framework.Extender
framework.PodNominator
parallelizer parallelize.Parallelizer
}
Registry 是作为一个可用插件的集合。framework 使用 registry 来启用和对插件配置的初始化。在初始化框架之前，所有插件都必须在注册表中。表现形式就是一个 map[]；key 是插件的名称，value是 PluginFactory 。type Registry map[string]PluginFactory
in-tree plugin 注册进来。通过 NewInTreeRegistry 。后续如果还有插件要注册，可以通过 WithFrameworkOutOfTreeRegistry 来注册其他的插件。func NewInTreeRegistry() runtime.Registry {
fts := plfeature.Features{
EnableReadWriteOncePod: feature.DefaultFeatureGate.Enabled(features.ReadWriteOncePod),
EnableVolumeCapacityPriority: feature.DefaultFeatureGate.Enabled(features.VolumeCapacityPriority),
EnableMinDomainsInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.MinDomainsInPodTopologySpread),
EnableNodeInclusionPolicyInPodTopologySpread: feature.DefaultFeatureGate.Enabled(features.NodeInclusionPolicyInPodTopologySpread),
}
return runtime.Registry{
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),
}
}
这里插入一个题外话，关于 in-tree plugin
在这里没有找到关于，kube-scheduler ，只是找到有关的概念，大概可以解释为，in-tree表示为随kubernetes官方提供的二进制构建的 plugin 则为 in-tree，而独立于kubernetes代码库之外的为 out-of-tree scheduleOne 中可以很好的看出，功能都是 framework 提供的。
func (sched *Scheduler) scheduleOne(ctx context.Context) {
...
scheduleResult, err := sched.SchedulePod(schedulingCycleCtx, fwk, state, pod)
...
// Run the Reserve method of reserve plugins.
if sts := fwk.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
}
...
// Run "permit" plugins.
runPermitStatus := fwk.RunPermitPlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
// One of the plugins returned status different than success or wait.
fwk.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
// bind the pod to its host asynchronously (we can do this b/c of the assumption step above).
go func() {
...
waitOnPermitStatus := fwk.WaitOnPermit(bindingCycleCtx, assumedPod)
if !waitOnPermitStatus.IsSuccess() {
...
// trigger un-reserve plugins to clean up state associated with the reserved Pod
fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
}
// Run "prebind" plugins.
preBindStatus := fwk.RunPreBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
// trigger un-reserve plugins to clean up state associated with the reserved Pod
fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
...
// trigger un-reserve plugins to clean up state associated with the reserved Pod
fwk.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
// Run "postbind" plugins.
fwk.RunPostBindPlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
...
}
插件 kubernetes/cmd/kube-scheduler/app/server.go
profile.NewMap：kubernetes/pkg/scheduler/scheduler.go
newProfile：kubernetes/pkg/scheduler/scheduler.go
frameworkruntime.NewFramework：kubernetes/pkg/scheduler/framework/runtime/framework.go
pluginsNeeded：kubernetes/pkg/scheduler/framework/runtime/framework.go
NewScheduler
我们了解如何 New 一个 scheduler 即为 Setup 中去配置这些参数，
func Setup(ctx context.Context, opts *options.Options, outOfTreeRegistryOptions ...Option) (*schedulerserverconfig.CompletedConfig, *scheduler.Scheduler, error) {
...
// Create the scheduler.
sched, err := scheduler.New(cc.Client,
cc.InformerFactory,
cc.DynInformerFactory,
recorderFactory,
ctx.Done(),
scheduler.WithComponentConfigVersion(cc.ComponentConfig.TypeMeta.APIVersion),
scheduler.WithKubeConfig(cc.KubeConfig),
scheduler.WithProfiles(cc.ComponentConfig.Profiles...),
scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
scheduler.WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),
scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),
scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),
scheduler.WithPodMaxInUnschedulablePodsDuration(cc.PodMaxInUnschedulablePodsDuration),
scheduler.WithExtenders(cc.ComponentConfig.Extenders...),
scheduler.WithParallelism(cc.ComponentConfig.Parallelism),
scheduler.WithBuildFrameworkCapturer(func(profile kubeschedulerconfig.KubeSchedulerProfile) {
// Profiles are processed during Framework instantiation to set default plugins and configurations. Capturing them for logging
completedProfiles = append(completedProfiles, profile)
}),
)
...
}
profile.NewMap
在 scheduler.New 中，会根据配置生成profile，而 profile.NewMap 会完成这一步
func New(client clientset.Interface,
...
clusterEventMap := make(map[framework.ClusterEvent]sets.String)
profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
frameworkruntime.WithClientSet(client),
frameworkruntime.WithKubeConfig(options.kubeConfig),
frameworkruntime.WithInformerFactory(informerFactory),
frameworkruntime.WithSnapshotSharedLister(snapshot),
frameworkruntime.WithPodNominator(nominator),
frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
frameworkruntime.WithClusterEventMap(clusterEventMap),
frameworkruntime.WithParallelism(int(options.parallelism)),
frameworkruntime.WithExtenders(extenders),
)
...
}
NewFramework
newProfile 返回的则是一个创建好的 framework
func newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
recorder := recorderFact(cfg.SchedulerName)
opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}
最终会走到 pluginsNeeded，这里会根据配置中开启的插件而返回一个插件集，这个就是最终在每个扩展点中药执行的插件。
func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) sets.String {
pgSet := sets.String{}
if plugins == nil {
return pgSet
}
find := func(pgs *config.PluginSet) {
for _, pg := range pgs.Enabled {
pgSet.Insert(pg.Name)
}
}
// 获取到所有的扩展点，找到为Enabled的插件加入到pgSet
for _, e := range f.getExtensionPoints(plugins) {
find(e.plugins)
}
// Parse MultiPoint separately since they are not returned by f.getExtensionPoints()
find(&plugins.MultiPoint)
return pgSet
}
插件的执行
在对插件源码部分分析，会找几个典型的插件进行分析，而不会对全部的进行分析，因为总的来说是大同小异，分析的插件有 NodePorts，NodeResourcesFit，podtopologyspread
NodePorts
这里以一个简单的插件来分析；NodePorts 插件用于检查Pod请求的端口，在节点上是否为空闲端口。
NodePorts 实现了 FilterPlugin 和 PreFilterPlugin
PreFilter 将会被 framework 中 PreFilter 扩展点被调用。
func (pl *NodePorts) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s := getContainerPorts(pod) // 或得Pod得端口
// 写入状态
cycleState.Write(preFilterStateKey, preFilterState(s))
return nil, nil
}
Filter 将会被 framework 中 Filter 扩展点被调用。
// Filter invoked at the filter extension point.
func (pl *NodePorts) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
wantPorts, err := getPreFilterState(cycleState)
if err != nil {
return framework.AsStatus(err)
}
fits := fitsPorts(wantPorts, nodeInfo)
if !fits {
return framework.NewStatus(framework.Unschedulable, ErrReason)
}
return nil
}
func fitsPorts(wantPorts []*v1.ContainerPort, nodeInfo *framework.NodeInfo) bool {
// 对比existingPorts 和 wantPorts是否冲突，冲突则调度失败
existingPorts := nodeInfo.UsedPorts
for _, cp := range wantPorts {
if existingPorts.CheckConflict(cp.HostIP, string(cp.Protocol), cp.HostPort) {
return false
}
}
return true
}
New ，初始化新插件，在 register 中注册得
func New(_ runtime.Object, _ framework.Handle) (framework.Plugin, error) {
return &NodePorts{}, nil
}
在调用中，如果有任何一个插件返回错误，则跳过该扩展点注册得其他插件，返回失败。
func (f *frameworkImpl) RunFilterPlugins(
ctx context.Context,
state *framework.CycleState,
pod *v1.Pod,
nodeInfo *framework.NodeInfo,
) framework.PluginToStatus {
statuses := make(framework.PluginToStatus)
for _, pl := range f.filterPlugins {
pluginStatus := f.runFilterPlugin(ctx, pl, state, pod, nodeInfo)
if !pluginStatus.IsSuccess() {
if !pluginStatus.IsUnschedulable()
errStatus := framework.AsStatus(fmt.Errorf("running %q filter plugin: %w", pl.Name(), pluginStatus.AsError())).WithFailedPlugin(pl.Name())
return map[string]*framework.Status{pl.Name(): errStatus}
}
pluginStatus.SetFailedPlugin(pl.Name())
statuses[pl.Name()] = pluginStatus
}
}
return statuses
}
返回得状态是一个 Status 结构体，该结构体表示了插件运行的结果。由 Code、reasons、（可选）err 和 failedPlugin （失败的那个插件名）组成。当 code 不是 Success 时，应说明原因。而且，当 code 为 Success 时，其他所有字段都应为空。nil 状态也被视为成功。
type Status struct {
code Code
reasons []string
err error
// failedPlugin is an optional field that records the plugin name a Pod failed by.
// It's set by the framework when code is Error, Unschedulable or UnschedulableAndUnresolvable.
failedPlugin string
}
NodeResourcesFit computePodResourceRequest 这里有一个注释，总体解释起来是这样得：computePodResourceRequest ，返回值（ framework.Resource）覆盖了每一个维度中资源的最大宽度。因为将按照 init-containers , containers 得顺序运行，会通过迭代方式收集每个维度中的最大值。计算时会对常规容器的资源向量求和，因为containers 运行会同时运行多个容器。计算示例为：
Pod:
InitContainers
IC1:
CPU: 2
Memory: 1G
IC2:
CPU: 2
Memory: 3G
Containers
C1:
CPU: 2
Memory: 1G
C2:
CPU: 1
Memory: 1G
在维度1中（InitContainers）所需资源最大值时，CPU=2, Memory=3G；而维度2（Containers）所需资源最大值为：CPU=2, Memory=1G；那么最终结果为 CPU=3, Memory=3G，因为在维度1，最大资源时Memory=3G；而维度2最大资源是CPU=1+2, Memory=1+1，取每个维度中最大资源最大宽度即为 CPU=3, Memory=3G。
下面则看下代码得实现
func computePodResourceRequest(pod *v1.Pod) *preFilterState {
result := &preFilterState{}
for _, container := range pod.Spec.Containers {
result.Add(container.Resources.Requests)
}
// 取最大得资源
for _, container := range pod.Spec.InitContainers {
result.SetMaxResource(container.Resources.Requests)
}
// 如果Overhead正在使用，需要将其计算到总资源中
if pod.Spec.Overhead != nil {
result.Add(pod.Spec.Overhead)
}
return result
}
// SetMaxResource 是比较ResourceList并为每个资源取最大值。
func (r *Resource) SetMaxResource(rl v1.ResourceList) {
if r == nil {
return
}
for rName, rQuantity := range rl {
switch rName {
case v1.ResourceMemory:
r.Memory = max(r.Memory, rQuantity.Value())
case v1.ResourceCPU:
r.MilliCPU = max(r.MilliCPU, rQuantity.MilliValue())
case v1.ResourceEphemeralStorage:
if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
r.EphemeralStorage = max(r.EphemeralStorage, rQuantity.Value())
}
default:
if schedutil.IsScalarResourceName(rName) {
r.SetScalar(rName, max(r.ScalarResources[rName], rQuantity.Value()))
}
}
}
}
leastAllocate
LeastAllocated 是 NodeResourcesFit 的打分策略 ，LeastAllocated 打分的标准是更偏向于请求资源较少的Node。将会先计算出Node上调度的pod请求的内存、CPU与其他资源的百分比，然后并根据请求的比例与容量的平均值的最小值进行优先级排序。
计算公式是这样的：\(\frac{\frac{cpu((capacity-requested) \times MaxNodeScore \times cpuWeight)}{capacity} + \frac{memory((capacity-requested) \times MaxNodeScore \times memoryWeight}{capacity}) + ...}{weightSum}\)
下面来看下实现
func leastResourceScorer(resToWeightMap resourceToWeightMap) func(resourceToValueMap, resourceToValueMap) int64 {
return func(requested, allocable resourceToValueMap) int64 {
var nodeScore, weightSum int64
for resource := range requested {
weight := resToWeightMap[resource]
// 计算出的资源分数乘weight
resourceScore := leastRequestedScore(requested[resource], allocable[resource])
nodeScore += resourceScore * weight
weightSum += weight
}
if weightSum == 0 {
return 0
}
// 最终除weightSum
return nodeScore / weightSum
}
}
leastRequestedScore 计算标准为未使用容量的计算范围为 0~MaxNodeScore，0 为最低优先级，MaxNodeScore 为最高优先级。未使用的资源越多，得分越高。
func leastRequestedScore(requested, capacity int64) int64 {
if capacity == 0 {
return 0
}
if requested > capacity {
return 0
}
// 容量 - 请求的 x 预期值（100）/ 容量
return ((capacity - requested) * int64(framework.MaxNodeScore)) / capacity
}
Topology pkg\scheduler\framework\plugins\podtopologyspread\plugin.go
var systemDefaultConstraints = []v1.TopologySpreadConstraint{
{
TopologyKey: v1.LabelHostname,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 3,
},
{
TopologyKey: v1.LabelTopologyZone,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 5,
},
}
可以通过在配置文件中留空，来禁用默认配置
defaultConstraints: []defaultingType: List
apiVersion: kubescheduler.config.k8s.io/v1beta3
kind: KubeSchedulerConfiguration
profiles:
- schedulerName: default-scheduler
pluginConfig:
- name: PodTopologySpread
args:
defaultConstraints: []
defaultingType: List
通过源码学习Topology
podtopologyspread 实现了4种扩展点方法，包含 filter 和 score
PreFilter
可以看到 PreFilter 的核心为 calPreFilterState
func (pl *PodTopologySpread) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s, err := pl.calPreFilterState(ctx, pod)
if err != nil {
return nil, framework.AsStatus(err)
}
cycleState.Write(preFilterStateKey, s)
return nil, nil
}
calPreFilterState 主要功能是用在计算如何在拓扑域中分布Pod，首先看段代码时，需要掌握下属几个概念
- update 函数实际上时用于计算 preFilter 中 最后的计算结果会保存在
CycleState 中
cycleState.Write(preFilterStateKey, s)
Filter 主要是从 PreFilter 处理的过程中拿到状态 preFilterState，然后看下每个拓扑约束中的 MaxSkew 是否合法，具体的计算公式为：\(matchNum + selfMatchNum - minMatchNum\)
matchNum：Prefilter 中计算出的对应的拓扑分布数量，可以在Framework 中会运行 ScoreExtension ，即 NormalizeScore
// Run NormalizeScore method for each ScorePlugin in parallel.
f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
nodeScoreList := pluginToNodeScores[pl.Name()]
if pl.ScoreExtensions() == nil {
return
}
status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
if !status.IsSuccess() {
err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
errCh.SendErrorWithCancel(err, cancel)
return
}
})
if err := errCh.ReceiveError(); err != nil {
return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
}
NormalizeScore 会为所有的node根据之前计算出的权重进行打分
func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
s, err := getPreScoreState(cycleState)
if err != nil {
return framework.AsStatus(err)
}
if s == nil {
return nil
}
// 计算 和
var minScore int64 = math.MaxInt64
var maxScore int64
for i, score := range scores {
// it's mandatory to check if is present in m.IgnoredNodes
if s.IgnoredNodes.Has(score.Name) {
scores[i].Score = invalidScore
continue
}
if score.Score < minScore {
minScore = score.Score
}
if score.Score > maxScore {
maxScore = score.Score
}
}
for i := range scores {
if scores[i].Score == invalidScore {
scores[i].Score = 0
continue
}
if maxScore == 0 {
scores[i].Score = framework.MaxNodeScore
continue
}
s := scores[i].Score
scores[i].Score = framework.MaxNodeScore * (maxScore + minScore - s) / maxScore
}
return nil
}
到此，对于pod拓扑插件功能大概可以明了了，
- Filter 部分（
PreFilter，Filter）完成拓扑对(Topology Pair)划分
- Score部分（
PreScore, Score , NormalizeScore ）主要是对拓扑对（可以理解为拓扑结构划分）来选择一个最适合的pod的节点（即分数最优的节点）
而在 scoring_test.go 给了很多用例，可以更深入的了解这部分算法
Reference
[1] scheduling code hierarchy
[2] scheduler algorithm
[3] in tree VS out of tree volume plugins
[4] scheduler_framework_plugins
[5] scheduling config
[6] topology spread constraints
profile.NewMap：kubernetes/pkg/scheduler/scheduler.go
newProfile：kubernetes/pkg/scheduler/scheduler.go
frameworkruntime.NewFramework：kubernetes/pkg/scheduler/framework/runtime/framework.go
pluginsNeeded：kubernetes/pkg/scheduler/framework/runtime/framework.go
Setup 中去配置这些参数，func Setup(ctx context.Context, opts *options.Options, outOfTreeRegistryOptions ...Option) (*schedulerserverconfig.CompletedConfig, *scheduler.Scheduler, error) {
...
// Create the scheduler.
sched, err := scheduler.New(cc.Client,
cc.InformerFactory,
cc.DynInformerFactory,
recorderFactory,
ctx.Done(),
scheduler.WithComponentConfigVersion(cc.ComponentConfig.TypeMeta.APIVersion),
scheduler.WithKubeConfig(cc.KubeConfig),
scheduler.WithProfiles(cc.ComponentConfig.Profiles...),
scheduler.WithPercentageOfNodesToScore(cc.ComponentConfig.PercentageOfNodesToScore),
scheduler.WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),
scheduler.WithPodMaxBackoffSeconds(cc.ComponentConfig.PodMaxBackoffSeconds),
scheduler.WithPodInitialBackoffSeconds(cc.ComponentConfig.PodInitialBackoffSeconds),
scheduler.WithPodMaxInUnschedulablePodsDuration(cc.PodMaxInUnschedulablePodsDuration),
scheduler.WithExtenders(cc.ComponentConfig.Extenders...),
scheduler.WithParallelism(cc.ComponentConfig.Parallelism),
scheduler.WithBuildFrameworkCapturer(func(profile kubeschedulerconfig.KubeSchedulerProfile) {
// Profiles are processed during Framework instantiation to set default plugins and configurations. Capturing them for logging
completedProfiles = append(completedProfiles, profile)
}),
)
...
}
scheduler.New 中，会根据配置生成profile，而 profile.NewMap 会完成这一步func New(client clientset.Interface,
...
clusterEventMap := make(map[framework.ClusterEvent]sets.String)
profiles, err := profile.NewMap(options.profiles, registry, recorderFactory, stopCh,
frameworkruntime.WithComponentConfigVersion(options.componentConfigVersion),
frameworkruntime.WithClientSet(client),
frameworkruntime.WithKubeConfig(options.kubeConfig),
frameworkruntime.WithInformerFactory(informerFactory),
frameworkruntime.WithSnapshotSharedLister(snapshot),
frameworkruntime.WithPodNominator(nominator),
frameworkruntime.WithCaptureProfile(frameworkruntime.CaptureProfile(options.frameworkCapturer)),
frameworkruntime.WithClusterEventMap(clusterEventMap),
frameworkruntime.WithParallelism(int(options.parallelism)),
frameworkruntime.WithExtenders(extenders),
)
...
}
newProfile 返回的则是一个创建好的 frameworkfunc newProfile(cfg config.KubeSchedulerProfile, r frameworkruntime.Registry, recorderFact RecorderFactory,
stopCh <-chan struct{}, opts ...frameworkruntime.Option) (framework.Framework, error) {
recorder := recorderFact(cfg.SchedulerName)
opts = append(opts, frameworkruntime.WithEventRecorder(recorder))
return frameworkruntime.NewFramework(r, &cfg, stopCh, opts...)
}
pluginsNeeded，这里会根据配置中开启的插件而返回一个插件集，这个就是最终在每个扩展点中药执行的插件。func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) sets.String {
pgSet := sets.String{}
if plugins == nil {
return pgSet
}
find := func(pgs *config.PluginSet) {
for _, pg := range pgs.Enabled {
pgSet.Insert(pg.Name)
}
}
// 获取到所有的扩展点，找到为Enabled的插件加入到pgSet
for _, e := range f.getExtensionPoints(plugins) {
find(e.plugins)
}
// Parse MultiPoint separately since they are not returned by f.getExtensionPoints()
find(&plugins.MultiPoint)
return pgSet
}
NodePorts，NodeResourcesFit，podtopologyspreadNodePorts 插件用于检查Pod请求的端口，在节点上是否为空闲端口。FilterPlugin 和 PreFilterPluginframework 中 PreFilter 扩展点被调用。func (pl *NodePorts) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s := getContainerPorts(pod) // 或得Pod得端口
// 写入状态
cycleState.Write(preFilterStateKey, preFilterState(s))
return nil, nil
}
framework 中 Filter 扩展点被调用。// Filter invoked at the filter extension point.
func (pl *NodePorts) Filter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
wantPorts, err := getPreFilterState(cycleState)
if err != nil {
return framework.AsStatus(err)
}
fits := fitsPorts(wantPorts, nodeInfo)
if !fits {
return framework.NewStatus(framework.Unschedulable, ErrReason)
}
return nil
}
func fitsPorts(wantPorts []*v1.ContainerPort, nodeInfo *framework.NodeInfo) bool {
// 对比existingPorts 和 wantPorts是否冲突，冲突则调度失败
existingPorts := nodeInfo.UsedPorts
for _, cp := range wantPorts {
if existingPorts.CheckConflict(cp.HostIP, string(cp.Protocol), cp.HostPort) {
return false
}
}
return true
}
register 中注册得func New(_ runtime.Object, _ framework.Handle) (framework.Plugin, error) {
return &NodePorts{}, nil
}
func (f *frameworkImpl) RunFilterPlugins(
ctx context.Context,
state *framework.CycleState,
pod *v1.Pod,
nodeInfo *framework.NodeInfo,
) framework.PluginToStatus {
statuses := make(framework.PluginToStatus)
for _, pl := range f.filterPlugins {
pluginStatus := f.runFilterPlugin(ctx, pl, state, pod, nodeInfo)
if !pluginStatus.IsSuccess() {
if !pluginStatus.IsUnschedulable()
errStatus := framework.AsStatus(fmt.Errorf("running %q filter plugin: %w", pl.Name(), pluginStatus.AsError())).WithFailedPlugin(pl.Name())
return map[string]*framework.Status{pl.Name(): errStatus}
}
pluginStatus.SetFailedPlugin(pl.Name())
statuses[pl.Name()] = pluginStatus
}
}
return statuses
}
Code、reasons、（可选）err 和 failedPlugin （失败的那个插件名）组成。当 code 不是 Success 时，应说明原因。而且，当 code 为 Success 时，其他所有字段都应为空。nil 状态也被视为成功。type Status struct {
code Code
reasons []string
err error
// failedPlugin is an optional field that records the plugin name a Pod failed by.
// It's set by the framework when code is Error, Unschedulable or UnschedulableAndUnresolvable.
failedPlugin string
}
computePodResourceRequest ，返回值（ framework.Resource）覆盖了每一个维度中资源的最大宽度。因为将按照 init-containers , containers 得顺序运行，会通过迭代方式收集每个维度中的最大值。计算时会对常规容器的资源向量求和，因为containers 运行会同时运行多个容器。计算示例为：
Pod:
InitContainers
IC1:
CPU: 2
Memory: 1G
IC2:
CPU: 2
Memory: 3G
Containers
C1:
CPU: 2
Memory: 1G
C2:
CPU: 1
Memory: 1G
在维度1中（InitContainers）所需资源最大值时，CPU=2, Memory=3G；而维度2（Containers）所需资源最大值为：CPU=2, Memory=1G；那么最终结果为 CPU=3, Memory=3G，因为在维度1，最大资源时Memory=3G；而维度2最大资源是CPU=1+2, Memory=1+1，取每个维度中最大资源最大宽度即为 CPU=3, Memory=3G。
下面则看下代码得实现
func computePodResourceRequest(pod *v1.Pod) *preFilterState {
result := &preFilterState{}
for _, container := range pod.Spec.Containers {
result.Add(container.Resources.Requests)
}
// 取最大得资源
for _, container := range pod.Spec.InitContainers {
result.SetMaxResource(container.Resources.Requests)
}
// 如果Overhead正在使用，需要将其计算到总资源中
if pod.Spec.Overhead != nil {
result.Add(pod.Spec.Overhead)
}
return result
}
// SetMaxResource 是比较ResourceList并为每个资源取最大值。
func (r *Resource) SetMaxResource(rl v1.ResourceList) {
if r == nil {
return
}
for rName, rQuantity := range rl {
switch rName {
case v1.ResourceMemory:
r.Memory = max(r.Memory, rQuantity.Value())
case v1.ResourceCPU:
r.MilliCPU = max(r.MilliCPU, rQuantity.MilliValue())
case v1.ResourceEphemeralStorage:
if utilfeature.DefaultFeatureGate.Enabled(features.LocalStorageCapacityIsolation) {
r.EphemeralStorage = max(r.EphemeralStorage, rQuantity.Value())
}
default:
if schedutil.IsScalarResourceName(rName) {
r.SetScalar(rName, max(r.ScalarResources[rName], rQuantity.Value()))
}
}
}
}
leastAllocate
LeastAllocated 是 NodeResourcesFit 的打分策略 ，LeastAllocated 打分的标准是更偏向于请求资源较少的Node。将会先计算出Node上调度的pod请求的内存、CPU与其他资源的百分比，然后并根据请求的比例与容量的平均值的最小值进行优先级排序。
计算公式是这样的：\(\frac{\frac{cpu((capacity-requested) \times MaxNodeScore \times cpuWeight)}{capacity} + \frac{memory((capacity-requested) \times MaxNodeScore \times memoryWeight}{capacity}) + ...}{weightSum}\)
下面来看下实现
func leastResourceScorer(resToWeightMap resourceToWeightMap) func(resourceToValueMap, resourceToValueMap) int64 {
return func(requested, allocable resourceToValueMap) int64 {
var nodeScore, weightSum int64
for resource := range requested {
weight := resToWeightMap[resource]
// 计算出的资源分数乘weight
resourceScore := leastRequestedScore(requested[resource], allocable[resource])
nodeScore += resourceScore * weight
weightSum += weight
}
if weightSum == 0 {
return 0
}
// 最终除weightSum
return nodeScore / weightSum
}
}
leastRequestedScore 计算标准为未使用容量的计算范围为 0~MaxNodeScore，0 为最低优先级，MaxNodeScore 为最高优先级。未使用的资源越多，得分越高。
func leastRequestedScore(requested, capacity int64) int64 {
if capacity == 0 {
return 0
}
if requested > capacity {
return 0
}
// 容量 - 请求的 x 预期值（100）/ 容量
return ((capacity - requested) * int64(framework.MaxNodeScore)) / capacity
}
Topology pkg\scheduler\framework\plugins\podtopologyspread\plugin.go
var systemDefaultConstraints = []v1.TopologySpreadConstraint{
{
TopologyKey: v1.LabelHostname,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 3,
},
{
TopologyKey: v1.LabelTopologyZone,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 5,
},
}
可以通过在配置文件中留空，来禁用默认配置
defaultConstraints: []defaultingType: List
apiVersion: kubescheduler.config.k8s.io/v1beta3
kind: KubeSchedulerConfiguration
profiles:
- schedulerName: default-scheduler
pluginConfig:
- name: PodTopologySpread
args:
defaultConstraints: []
defaultingType: List
通过源码学习Topology
podtopologyspread 实现了4种扩展点方法，包含 filter 和 score
PreFilter
可以看到 PreFilter 的核心为 calPreFilterState
func (pl *PodTopologySpread) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s, err := pl.calPreFilterState(ctx, pod)
if err != nil {
return nil, framework.AsStatus(err)
}
cycleState.Write(preFilterStateKey, s)
return nil, nil
}
calPreFilterState 主要功能是用在计算如何在拓扑域中分布Pod，首先看段代码时，需要掌握下属几个概念
- update 函数实际上时用于计算 preFilter 中 最后的计算结果会保存在
CycleState 中
cycleState.Write(preFilterStateKey, s)
Filter 主要是从 PreFilter 处理的过程中拿到状态 preFilterState，然后看下每个拓扑约束中的 MaxSkew 是否合法，具体的计算公式为：\(matchNum + selfMatchNum - minMatchNum\)
matchNum：Prefilter 中计算出的对应的拓扑分布数量，可以在Framework 中会运行 ScoreExtension ，即 NormalizeScore
// Run NormalizeScore method for each ScorePlugin in parallel.
f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
pl := f.scorePlugins[index]
nodeScoreList := pluginToNodeScores[pl.Name()]
if pl.ScoreExtensions() == nil {
return
}
status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList)
if !status.IsSuccess() {
err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
errCh.SendErrorWithCancel(err, cancel)
return
}
})
if err := errCh.ReceiveError(); err != nil {
return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
}
NormalizeScore 会为所有的node根据之前计算出的权重进行打分
func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status {
s, err := getPreScoreState(cycleState)
if err != nil {
return framework.AsStatus(err)
}
if s == nil {
return nil
}
// 计算 和
var minScore int64 = math.MaxInt64
var maxScore int64
for i, score := range scores {
// it's mandatory to check if is present in m.IgnoredNodes
if s.IgnoredNodes.Has(score.Name) {
scores[i].Score = invalidScore
continue
}
if score.Score < minScore {
minScore = score.Score
}
if score.Score > maxScore {
maxScore = score.Score
}
}
for i := range scores {
if scores[i].Score == invalidScore {
scores[i].Score = 0
continue
}
if maxScore == 0 {
scores[i].Score = framework.MaxNodeScore
continue
}
s := scores[i].Score
scores[i].Score = framework.MaxNodeScore * (maxScore + minScore - s) / maxScore
}
return nil
}
到此，对于pod拓扑插件功能大概可以明了了，
- Filter 部分（
PreFilter，Filter）完成拓扑对(Topology Pair)划分
- Score部分（
PreScore, Score , NormalizeScore ）主要是对拓扑对（可以理解为拓扑结构划分）来选择一个最适合的pod的节点（即分数最优的节点）
而在 scoring_test.go 给了很多用例，可以更深入的了解这部分算法
Reference
[1] scheduling code hierarchy
[2] scheduler algorithm
[3] in tree VS out of tree volume plugins
[4] scheduler_framework_plugins
[5] scheduling config
[6] topology spread constraints
var systemDefaultConstraints = []v1.TopologySpreadConstraint{
{
TopologyKey: v1.LabelHostname,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 3,
},
{
TopologyKey: v1.LabelTopologyZone,
WhenUnsatisfiable: v1.ScheduleAnyway,
MaxSkew: 5,
},
}
defaultConstraints: []defaultingType: ListapiVersion: kubescheduler.config.k8s.io/v1beta3
kind: KubeSchedulerConfiguration
profiles:
- schedulerName: default-scheduler
pluginConfig:
- name: PodTopologySpread
args:
defaultConstraints: []
defaultingType: List
podtopologyspread 实现了4种扩展点方法，包含 filter 和 scorePreFilter 的核心为 calPreFilterStatefunc (pl *PodTopologySpread) PreFilter(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod) (*framework.PreFilterResult, *framework.Status) {
s, err := pl.calPreFilterState(ctx, pod)
if err != nil {
return nil, framework.AsStatus(err)
}
cycleState.Write(preFilterStateKey, s)
return nil, nil
}
CycleState 中
cycleState.Write(preFilterStateKey, s)
Filter 主要是从 PreFilter 处理的过程中拿到状态 preFilterState，然后看下每个拓扑约束中的 MaxSkew 是否合法，具体的计算公式为：\(matchNum + selfMatchNum - minMatchNum\)
matchNum：Prefilter 中计算出的对应的拓扑分布数量，可以在Framework 中会运行ScoreExtension，即NormalizeScore// Run NormalizeScore method for each ScorePlugin in parallel. f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) { pl := f.scorePlugins[index] nodeScoreList := pluginToNodeScores[pl.Name()] if pl.ScoreExtensions() == nil { return } status := f.runScoreExtension(ctx, pl, state, pod, nodeScoreList) if !status.IsSuccess() { err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError()) errCh.SendErrorWithCancel(err, cancel) return } }) if err := errCh.ReceiveError(); err != nil { return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err)) }NormalizeScore 会为所有的node根据之前计算出的权重进行打分
func (pl *PodTopologySpread) NormalizeScore(ctx context.Context, cycleState *framework.CycleState, pod *v1.Pod, scores framework.NodeScoreList) *framework.Status { s, err := getPreScoreState(cycleState) if err != nil { return framework.AsStatus(err) } if s == nil { return nil } // 计算和 var minScore int64 = math.MaxInt64 var maxScore int64 for i, score := range scores { // it's mandatory to check if is present in m.IgnoredNodes if s.IgnoredNodes.Has(score.Name) { scores[i].Score = invalidScore continue } if score.Score < minScore { minScore = score.Score } if score.Score > maxScore { maxScore = score.Score } } for i := range scores { if scores[i].Score == invalidScore { scores[i].Score = 0 continue } if maxScore == 0 { scores[i].Score = framework.MaxNodeScore continue } s := scores[i].Score scores[i].Score = framework.MaxNodeScore * (maxScore + minScore - s) / maxScore } return nil } 到此，对于pod拓扑插件功能大概可以明了了，
- Filter 部分（
PreFilter，Filter）完成拓扑对(Topology Pair)划分 - Score部分（
PreScore,Score,NormalizeScore）主要是对拓扑对（可以理解为拓扑结构划分）来选择一个最适合的pod的节点（即分数最优的节点）
而在 scoring_test.go 给了很多用例，可以更深入的了解这部分算法
Reference
[1] scheduling code hierarchy
[2] scheduler algorithm
[3] in tree VS out of tree volume plugins
[4] scheduler_framework_plugins
[5] scheduling config
[6] topology spread constraints
- Filter 部分（