ec83143342
Previously, separate interfaces were defined for Reserve and Unreserve plugins. However, in nearly all cases, a plugin that allocates a resource using Reserve will likely want to register itself for Unreserve as well in order to free the allocated resource at the end of a failed scheduling/binding cycle. Having separate plugins for Reserve and Unreserve also adds unnecessary config toil. To that end, this patch aims to merge the two plugins into a single interface called a ReservePlugin that requires implementing both the Reserve and Unreserve methods.
1035 lines
38 KiB
Go
1035 lines
38 KiB
Go
/*
|
|
Copyright 2019 The Kubernetes Authors.
|
|
|
|
Licensed under the Apache License, Version 2.0 (the "License");
|
|
you may not use this file except in compliance with the License.
|
|
You may obtain a copy of the License at
|
|
|
|
http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
Unless required by applicable law or agreed to in writing, software
|
|
distributed under the License is distributed on an "AS IS" BASIS,
|
|
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
See the License for the specific language governing permissions and
|
|
limitations under the License.
|
|
*/
|
|
|
|
package runtime
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"reflect"
|
|
"time"
|
|
|
|
v1 "k8s.io/api/core/v1"
|
|
"k8s.io/apimachinery/pkg/runtime"
|
|
"k8s.io/apimachinery/pkg/types"
|
|
"k8s.io/apimachinery/pkg/util/sets"
|
|
"k8s.io/client-go/informers"
|
|
clientset "k8s.io/client-go/kubernetes"
|
|
"k8s.io/client-go/tools/events"
|
|
"k8s.io/klog/v2"
|
|
"k8s.io/kube-scheduler/config/v1beta1"
|
|
"k8s.io/kubernetes/pkg/scheduler/apis/config"
|
|
"k8s.io/kubernetes/pkg/scheduler/apis/config/scheme"
|
|
framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
|
|
"k8s.io/kubernetes/pkg/scheduler/internal/parallelize"
|
|
"k8s.io/kubernetes/pkg/scheduler/metrics"
|
|
)
|
|
|
|
const (
|
|
// Filter is the name of the filter extension point.
|
|
Filter = "Filter"
|
|
// Specifies the maximum timeout a permit plugin can return.
|
|
maxTimeout time.Duration = 15 * time.Minute
|
|
preFilter = "PreFilter"
|
|
preFilterExtensionAddPod = "PreFilterExtensionAddPod"
|
|
preFilterExtensionRemovePod = "PreFilterExtensionRemovePod"
|
|
postFilter = "PostFilter"
|
|
preScore = "PreScore"
|
|
score = "Score"
|
|
scoreExtensionNormalize = "ScoreExtensionNormalize"
|
|
preBind = "PreBind"
|
|
bind = "Bind"
|
|
postBind = "PostBind"
|
|
reserve = "Reserve"
|
|
unreserve = "Unreserve"
|
|
permit = "Permit"
|
|
)
|
|
|
|
var configDecoder = scheme.Codecs.UniversalDecoder()
|
|
|
|
// frameworkImpl is the component responsible for initializing and running scheduler
|
|
// plugins.
|
|
type frameworkImpl struct {
|
|
registry Registry
|
|
snapshotSharedLister framework.SharedLister
|
|
waitingPods *waitingPodsMap
|
|
pluginNameToWeightMap 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
|
|
eventRecorder events.EventRecorder
|
|
informerFactory informers.SharedInformerFactory
|
|
|
|
metricsRecorder *metricsRecorder
|
|
profileName string
|
|
|
|
preemptHandle framework.PreemptHandle
|
|
|
|
// Indicates that RunFilterPlugins should accumulate all failed statuses and not return
|
|
// after the first failure.
|
|
runAllFilters bool
|
|
}
|
|
|
|
// extensionPoint encapsulates desired and applied set of plugins at a specific extension
|
|
// point. This is used to simplify iterating over all extension points supported by the
|
|
// frameworkImpl.
|
|
type extensionPoint struct {
|
|
// the set of plugins to be configured at this extension point.
|
|
plugins *config.PluginSet
|
|
// a pointer to the slice storing plugins implementations that will run at this
|
|
// extension point.
|
|
slicePtr interface{}
|
|
}
|
|
|
|
func (f *frameworkImpl) getExtensionPoints(plugins *config.Plugins) []extensionPoint {
|
|
return []extensionPoint{
|
|
{plugins.PreFilter, &f.preFilterPlugins},
|
|
{plugins.Filter, &f.filterPlugins},
|
|
{plugins.PostFilter, &f.postFilterPlugins},
|
|
{plugins.Reserve, &f.reservePlugins},
|
|
{plugins.PreScore, &f.preScorePlugins},
|
|
{plugins.Score, &f.scorePlugins},
|
|
{plugins.PreBind, &f.preBindPlugins},
|
|
{plugins.Bind, &f.bindPlugins},
|
|
{plugins.PostBind, &f.postBindPlugins},
|
|
{plugins.Permit, &f.permitPlugins},
|
|
{plugins.QueueSort, &f.queueSortPlugins},
|
|
}
|
|
}
|
|
|
|
type frameworkOptions struct {
|
|
clientSet clientset.Interface
|
|
eventRecorder events.EventRecorder
|
|
informerFactory informers.SharedInformerFactory
|
|
snapshotSharedLister framework.SharedLister
|
|
metricsRecorder *metricsRecorder
|
|
profileName string
|
|
podNominator framework.PodNominator
|
|
extenders []framework.Extender
|
|
runAllFilters bool
|
|
}
|
|
|
|
// Option for the frameworkImpl.
|
|
type Option func(*frameworkOptions)
|
|
|
|
// WithClientSet sets clientSet for the scheduling frameworkImpl.
|
|
func WithClientSet(clientSet clientset.Interface) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.clientSet = clientSet
|
|
}
|
|
}
|
|
|
|
// WithEventRecorder sets clientSet for the scheduling frameworkImpl.
|
|
func WithEventRecorder(recorder events.EventRecorder) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.eventRecorder = recorder
|
|
}
|
|
}
|
|
|
|
// WithInformerFactory sets informer factory for the scheduling frameworkImpl.
|
|
func WithInformerFactory(informerFactory informers.SharedInformerFactory) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.informerFactory = informerFactory
|
|
}
|
|
}
|
|
|
|
// WithSnapshotSharedLister sets the SharedLister of the snapshot.
|
|
func WithSnapshotSharedLister(snapshotSharedLister framework.SharedLister) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.snapshotSharedLister = snapshotSharedLister
|
|
}
|
|
}
|
|
|
|
// WithRunAllFilters sets the runAllFilters flag, which means RunFilterPlugins accumulates
|
|
// all failure Statuses.
|
|
func WithRunAllFilters(runAllFilters bool) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.runAllFilters = runAllFilters
|
|
}
|
|
}
|
|
|
|
// WithProfileName sets the profile name.
|
|
func WithProfileName(name string) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.profileName = name
|
|
}
|
|
}
|
|
|
|
// withMetricsRecorder is only used in tests.
|
|
func withMetricsRecorder(recorder *metricsRecorder) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.metricsRecorder = recorder
|
|
}
|
|
}
|
|
|
|
// WithPodNominator sets podNominator for the scheduling frameworkImpl.
|
|
func WithPodNominator(nominator framework.PodNominator) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.podNominator = nominator
|
|
}
|
|
}
|
|
|
|
// WithExtenders sets extenders for the scheduling frameworkImpl.
|
|
func WithExtenders(extenders []framework.Extender) Option {
|
|
return func(o *frameworkOptions) {
|
|
o.extenders = extenders
|
|
}
|
|
}
|
|
|
|
var defaultFrameworkOptions = frameworkOptions{
|
|
metricsRecorder: newMetricsRecorder(1000, time.Second),
|
|
}
|
|
|
|
// TODO(#91029): move this to frameworkImpl runtime package.
|
|
var _ framework.PreemptHandle = &preemptHandle{}
|
|
|
|
type preemptHandle struct {
|
|
extenders []framework.Extender
|
|
framework.PodNominator
|
|
framework.PluginsRunner
|
|
}
|
|
|
|
// Extenders returns the registered extenders.
|
|
func (ph *preemptHandle) Extenders() []framework.Extender {
|
|
return ph.extenders
|
|
}
|
|
|
|
var _ framework.Framework = &frameworkImpl{}
|
|
|
|
// NewFramework initializes plugins given the configuration and the registry.
|
|
func NewFramework(r Registry, plugins *config.Plugins, args []config.PluginConfig, opts ...Option) (framework.Framework, error) {
|
|
options := defaultFrameworkOptions
|
|
for _, opt := range opts {
|
|
opt(&options)
|
|
}
|
|
|
|
f := &frameworkImpl{
|
|
registry: r,
|
|
snapshotSharedLister: options.snapshotSharedLister,
|
|
pluginNameToWeightMap: make(map[string]int),
|
|
waitingPods: newWaitingPodsMap(),
|
|
clientSet: options.clientSet,
|
|
eventRecorder: options.eventRecorder,
|
|
informerFactory: options.informerFactory,
|
|
metricsRecorder: options.metricsRecorder,
|
|
profileName: options.profileName,
|
|
runAllFilters: options.runAllFilters,
|
|
}
|
|
f.preemptHandle = &preemptHandle{
|
|
extenders: options.extenders,
|
|
PodNominator: options.podNominator,
|
|
PluginsRunner: f,
|
|
}
|
|
if plugins == nil {
|
|
return f, nil
|
|
}
|
|
|
|
// get needed plugins from config
|
|
pg := f.pluginsNeeded(plugins)
|
|
|
|
pluginConfig := make(map[string]runtime.Object, len(args))
|
|
for i := range args {
|
|
name := args[i].Name
|
|
if _, ok := pluginConfig[name]; ok {
|
|
return nil, fmt.Errorf("repeated config for plugin %s", name)
|
|
}
|
|
pluginConfig[name] = args[i].Args
|
|
}
|
|
|
|
pluginsMap := make(map[string]framework.Plugin)
|
|
var totalPriority int64
|
|
for name, factory := range r {
|
|
// initialize only needed plugins.
|
|
if _, ok := pg[name]; !ok {
|
|
continue
|
|
}
|
|
|
|
args, err := getPluginArgsOrDefault(pluginConfig, name)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("getting args for Plugin %q: %w", name, err)
|
|
}
|
|
p, err := factory(args, f)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("error initializing plugin %q: %v", name, err)
|
|
}
|
|
pluginsMap[name] = p
|
|
|
|
// a weight of zero is not permitted, plugins can be disabled explicitly
|
|
// when configured.
|
|
f.pluginNameToWeightMap[name] = int(pg[name].Weight)
|
|
if f.pluginNameToWeightMap[name] == 0 {
|
|
f.pluginNameToWeightMap[name] = 1
|
|
}
|
|
// Checks totalPriority against MaxTotalScore to avoid overflow
|
|
if int64(f.pluginNameToWeightMap[name])*framework.MaxNodeScore > framework.MaxTotalScore-totalPriority {
|
|
return nil, fmt.Errorf("total score of Score plugins could overflow")
|
|
}
|
|
totalPriority += int64(f.pluginNameToWeightMap[name]) * framework.MaxNodeScore
|
|
}
|
|
|
|
for _, e := range f.getExtensionPoints(plugins) {
|
|
if err := updatePluginList(e.slicePtr, e.plugins, pluginsMap); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
// Verifying the score weights again since Plugin.Name() could return a different
|
|
// value from the one used in the configuration.
|
|
for _, scorePlugin := range f.scorePlugins {
|
|
if f.pluginNameToWeightMap[scorePlugin.Name()] == 0 {
|
|
return nil, fmt.Errorf("score plugin %q is not configured with weight", scorePlugin.Name())
|
|
}
|
|
}
|
|
|
|
if len(f.queueSortPlugins) == 0 {
|
|
return nil, fmt.Errorf("no queue sort plugin is enabled")
|
|
}
|
|
if len(f.queueSortPlugins) > 1 {
|
|
return nil, fmt.Errorf("only one queue sort plugin can be enabled")
|
|
}
|
|
if len(f.bindPlugins) == 0 {
|
|
return nil, fmt.Errorf("at least one bind plugin is needed")
|
|
}
|
|
|
|
return f, nil
|
|
}
|
|
|
|
// getPluginArgsOrDefault returns a configuration provided by the user or builds
|
|
// a default from the scheme. Returns `nil, nil` if the plugin does not have a
|
|
// defined arg types, such as in-tree plugins that don't require configuration
|
|
// or out-of-tree plugins.
|
|
func getPluginArgsOrDefault(pluginConfig map[string]runtime.Object, name string) (runtime.Object, error) {
|
|
res, ok := pluginConfig[name]
|
|
if ok {
|
|
return res, nil
|
|
}
|
|
// Use defaults from latest config API version.
|
|
gvk := v1beta1.SchemeGroupVersion.WithKind(name + "Args")
|
|
obj, _, err := configDecoder.Decode(nil, &gvk, nil)
|
|
if runtime.IsNotRegisteredError(err) {
|
|
// This plugin is out-of-tree or doesn't require configuration.
|
|
return nil, nil
|
|
}
|
|
return obj, err
|
|
}
|
|
|
|
func updatePluginList(pluginList interface{}, pluginSet *config.PluginSet, pluginsMap map[string]framework.Plugin) error {
|
|
if pluginSet == nil {
|
|
return nil
|
|
}
|
|
|
|
plugins := reflect.ValueOf(pluginList).Elem()
|
|
pluginType := plugins.Type().Elem()
|
|
set := sets.NewString()
|
|
for _, ep := range pluginSet.Enabled {
|
|
pg, ok := pluginsMap[ep.Name]
|
|
if !ok {
|
|
return fmt.Errorf("%s %q does not exist", pluginType.Name(), ep.Name)
|
|
}
|
|
|
|
if !reflect.TypeOf(pg).Implements(pluginType) {
|
|
return fmt.Errorf("plugin %q does not extend %s plugin", ep.Name, pluginType.Name())
|
|
}
|
|
|
|
if set.Has(ep.Name) {
|
|
return fmt.Errorf("plugin %q already registered as %q", ep.Name, pluginType.Name())
|
|
}
|
|
|
|
set.Insert(ep.Name)
|
|
|
|
newPlugins := reflect.Append(plugins, reflect.ValueOf(pg))
|
|
plugins.Set(newPlugins)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// QueueSortFunc returns the function to sort pods in scheduling queue
|
|
func (f *frameworkImpl) QueueSortFunc() framework.LessFunc {
|
|
if f == nil {
|
|
// If frameworkImpl is nil, simply keep their order unchanged.
|
|
// NOTE: this is primarily for tests.
|
|
return func(_, _ *framework.QueuedPodInfo) bool { return false }
|
|
}
|
|
|
|
if len(f.queueSortPlugins) == 0 {
|
|
panic("No QueueSort plugin is registered in the frameworkImpl.")
|
|
}
|
|
|
|
// Only one QueueSort plugin can be enabled.
|
|
return f.queueSortPlugins[0].Less
|
|
}
|
|
|
|
// RunPreFilterPlugins runs the set of configured PreFilter plugins. It returns
|
|
// *Status and its code is set to non-success if any of the plugins returns
|
|
// anything but Success. If a non-success status is returned, then the scheduling
|
|
// cycle is aborted.
|
|
func (f *frameworkImpl) RunPreFilterPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod) (status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(preFilter, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
for _, pl := range f.preFilterPlugins {
|
|
status = f.runPreFilterPlugin(ctx, pl, state, pod)
|
|
if !status.IsSuccess() {
|
|
if status.IsUnschedulable() {
|
|
msg := fmt.Sprintf("rejected by %q at prefilter: %v", pl.Name(), status.Message())
|
|
klog.V(4).Infof(msg)
|
|
return framework.NewStatus(status.Code(), msg)
|
|
}
|
|
msg := fmt.Sprintf("error while running %q prefilter plugin for pod %q: %v", pl.Name(), pod.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runPreFilterPlugin(ctx context.Context, pl framework.PreFilterPlugin, state *framework.CycleState, pod *v1.Pod) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.PreFilter(ctx, state, pod)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.PreFilter(ctx, state, pod)
|
|
f.metricsRecorder.observePluginDurationAsync(preFilter, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunPreFilterExtensionAddPod calls the AddPod interface for the set of configured
|
|
// PreFilter plugins. It returns directly if any of the plugins return any
|
|
// status other than Success.
|
|
func (f *frameworkImpl) RunPreFilterExtensionAddPod(
|
|
ctx context.Context,
|
|
state *framework.CycleState,
|
|
podToSchedule *v1.Pod,
|
|
podToAdd *v1.Pod,
|
|
nodeInfo *framework.NodeInfo,
|
|
) (status *framework.Status) {
|
|
for _, pl := range f.preFilterPlugins {
|
|
if pl.PreFilterExtensions() == nil {
|
|
continue
|
|
}
|
|
status = f.runPreFilterExtensionAddPod(ctx, pl, state, podToSchedule, podToAdd, nodeInfo)
|
|
if !status.IsSuccess() {
|
|
msg := fmt.Sprintf("error while running AddPod for plugin %q while scheduling pod %q: %v",
|
|
pl.Name(), podToSchedule.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runPreFilterExtensionAddPod(ctx context.Context, pl framework.PreFilterPlugin, state *framework.CycleState, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.PreFilterExtensions().AddPod(ctx, state, podToSchedule, podToAdd, nodeInfo)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.PreFilterExtensions().AddPod(ctx, state, podToSchedule, podToAdd, nodeInfo)
|
|
f.metricsRecorder.observePluginDurationAsync(preFilterExtensionAddPod, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunPreFilterExtensionRemovePod calls the RemovePod interface for the set of configured
|
|
// PreFilter plugins. It returns directly if any of the plugins return any
|
|
// status other than Success.
|
|
func (f *frameworkImpl) RunPreFilterExtensionRemovePod(
|
|
ctx context.Context,
|
|
state *framework.CycleState,
|
|
podToSchedule *v1.Pod,
|
|
podToRemove *v1.Pod,
|
|
nodeInfo *framework.NodeInfo,
|
|
) (status *framework.Status) {
|
|
for _, pl := range f.preFilterPlugins {
|
|
if pl.PreFilterExtensions() == nil {
|
|
continue
|
|
}
|
|
status = f.runPreFilterExtensionRemovePod(ctx, pl, state, podToSchedule, podToRemove, nodeInfo)
|
|
if !status.IsSuccess() {
|
|
msg := fmt.Sprintf("error while running RemovePod for plugin %q while scheduling pod %q: %v",
|
|
pl.Name(), podToSchedule.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runPreFilterExtensionRemovePod(ctx context.Context, pl framework.PreFilterPlugin, state *framework.CycleState, podToSchedule *v1.Pod, podToAdd *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.PreFilterExtensions().RemovePod(ctx, state, podToSchedule, podToAdd, nodeInfo)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.PreFilterExtensions().RemovePod(ctx, state, podToSchedule, podToAdd, nodeInfo)
|
|
f.metricsRecorder.observePluginDurationAsync(preFilterExtensionRemovePod, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunFilterPlugins runs the set of configured Filter plugins for pod on
|
|
// the given node. If any of these plugins doesn't return "Success", the
|
|
// given node is not suitable for running pod.
|
|
// Meanwhile, the failure message and status are set for the given node.
|
|
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() {
|
|
// Filter plugins are not supposed to return any status other than
|
|
// Success or Unschedulable.
|
|
errStatus := framework.NewStatus(framework.Error, fmt.Sprintf("running %q filter plugin for pod %q: %v", pl.Name(), pod.Name, pluginStatus.Message()))
|
|
return map[string]*framework.Status{pl.Name(): errStatus}
|
|
}
|
|
statuses[pl.Name()] = pluginStatus
|
|
if !f.runAllFilters {
|
|
// Exit early if we don't need to run all filters.
|
|
return statuses
|
|
}
|
|
}
|
|
}
|
|
|
|
return statuses
|
|
}
|
|
|
|
func (f *frameworkImpl) runFilterPlugin(ctx context.Context, pl framework.FilterPlugin, state *framework.CycleState, pod *v1.Pod, nodeInfo *framework.NodeInfo) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.Filter(ctx, state, pod, nodeInfo)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.Filter(ctx, state, pod, nodeInfo)
|
|
f.metricsRecorder.observePluginDurationAsync(Filter, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunPostFilterPlugins runs the set of configured PostFilter plugins until the first
|
|
// Success or Error is met, otherwise continues to execute all plugins.
|
|
func (f *frameworkImpl) RunPostFilterPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, filteredNodeStatusMap framework.NodeToStatusMap) (_ *framework.PostFilterResult, status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(postFilter, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
|
|
statuses := make(framework.PluginToStatus)
|
|
for _, pl := range f.postFilterPlugins {
|
|
r, s := f.runPostFilterPlugin(ctx, pl, state, pod, filteredNodeStatusMap)
|
|
if s.IsSuccess() {
|
|
return r, s
|
|
} else if !s.IsUnschedulable() {
|
|
// Any status other than Success or Unschedulable is Error.
|
|
return nil, framework.NewStatus(framework.Error, s.Message())
|
|
}
|
|
statuses[pl.Name()] = s
|
|
}
|
|
|
|
return nil, statuses.Merge()
|
|
}
|
|
|
|
func (f *frameworkImpl) runPostFilterPlugin(ctx context.Context, pl framework.PostFilterPlugin, state *framework.CycleState, pod *v1.Pod, filteredNodeStatusMap framework.NodeToStatusMap) (*framework.PostFilterResult, *framework.Status) {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.PostFilter(ctx, state, pod, filteredNodeStatusMap)
|
|
}
|
|
startTime := time.Now()
|
|
r, s := pl.PostFilter(ctx, state, pod, filteredNodeStatusMap)
|
|
f.metricsRecorder.observePluginDurationAsync(postFilter, pl.Name(), s, metrics.SinceInSeconds(startTime))
|
|
return r, s
|
|
}
|
|
|
|
// RunPreScorePlugins runs the set of configured pre-score plugins. If any
|
|
// of these plugins returns any status other than "Success", the given pod is rejected.
|
|
func (f *frameworkImpl) RunPreScorePlugins(
|
|
ctx context.Context,
|
|
state *framework.CycleState,
|
|
pod *v1.Pod,
|
|
nodes []*v1.Node,
|
|
) (status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(preScore, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
for _, pl := range f.preScorePlugins {
|
|
status = f.runPreScorePlugin(ctx, pl, state, pod, nodes)
|
|
if !status.IsSuccess() {
|
|
msg := fmt.Sprintf("error while running %q prescore plugin for pod %q: %v", pl.Name(), pod.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runPreScorePlugin(ctx context.Context, pl framework.PreScorePlugin, state *framework.CycleState, pod *v1.Pod, nodes []*v1.Node) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.PreScore(ctx, state, pod, nodes)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.PreScore(ctx, state, pod, nodes)
|
|
f.metricsRecorder.observePluginDurationAsync(preScore, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunScorePlugins runs the set of configured scoring plugins. It returns a list that
|
|
// stores for each scoring plugin name the corresponding NodeScoreList(s).
|
|
// It also returns *Status, which is set to non-success if any of the plugins returns
|
|
// a non-success status.
|
|
func (f *frameworkImpl) RunScorePlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodes []*v1.Node) (ps framework.PluginToNodeScores, status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(score, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
pluginToNodeScores := make(framework.PluginToNodeScores, len(f.scorePlugins))
|
|
for _, pl := range f.scorePlugins {
|
|
pluginToNodeScores[pl.Name()] = make(framework.NodeScoreList, len(nodes))
|
|
}
|
|
ctx, cancel := context.WithCancel(ctx)
|
|
errCh := parallelize.NewErrorChannel()
|
|
|
|
// Run Score method for each node in parallel.
|
|
parallelize.Until(ctx, len(nodes), func(index int) {
|
|
for _, pl := range f.scorePlugins {
|
|
nodeName := nodes[index].Name
|
|
s, status := f.runScorePlugin(ctx, pl, state, pod, nodeName)
|
|
if !status.IsSuccess() {
|
|
errCh.SendErrorWithCancel(fmt.Errorf(status.Message()), cancel)
|
|
return
|
|
}
|
|
pluginToNodeScores[pl.Name()][index] = framework.NodeScore{
|
|
Name: nodeName,
|
|
Score: int64(s),
|
|
}
|
|
}
|
|
})
|
|
if err := errCh.ReceiveError(); err != nil {
|
|
msg := fmt.Sprintf("error while running score plugin for pod %q: %v", pod.Name, err)
|
|
klog.Error(msg)
|
|
return nil, framework.NewStatus(framework.Error, msg)
|
|
}
|
|
|
|
// Run NormalizeScore method for each ScorePlugin in parallel.
|
|
parallelize.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("normalize score plugin %q failed with error %v", pl.Name(), status.Message())
|
|
errCh.SendErrorWithCancel(err, cancel)
|
|
return
|
|
}
|
|
})
|
|
if err := errCh.ReceiveError(); err != nil {
|
|
msg := fmt.Sprintf("error while running normalize score plugin for pod %q: %v", pod.Name, err)
|
|
klog.Error(msg)
|
|
return nil, framework.NewStatus(framework.Error, msg)
|
|
}
|
|
|
|
// Apply score defaultWeights for each ScorePlugin in parallel.
|
|
parallelize.Until(ctx, len(f.scorePlugins), func(index int) {
|
|
pl := f.scorePlugins[index]
|
|
// Score plugins' weight has been checked when they are initialized.
|
|
weight := f.pluginNameToWeightMap[pl.Name()]
|
|
nodeScoreList := pluginToNodeScores[pl.Name()]
|
|
|
|
for i, nodeScore := range nodeScoreList {
|
|
// return error if score plugin returns invalid score.
|
|
if nodeScore.Score > int64(framework.MaxNodeScore) || nodeScore.Score < int64(framework.MinNodeScore) {
|
|
err := fmt.Errorf("score plugin %q returns an invalid score %v, it should in the range of [%v, %v] after normalizing", pl.Name(), nodeScore.Score, framework.MinNodeScore, framework.MaxNodeScore)
|
|
errCh.SendErrorWithCancel(err, cancel)
|
|
return
|
|
}
|
|
nodeScoreList[i].Score = nodeScore.Score * int64(weight)
|
|
}
|
|
})
|
|
if err := errCh.ReceiveError(); err != nil {
|
|
msg := fmt.Sprintf("error while applying score defaultWeights for pod %q: %v", pod.Name, err)
|
|
klog.Error(msg)
|
|
return nil, framework.NewStatus(framework.Error, msg)
|
|
}
|
|
|
|
return pluginToNodeScores, nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runScorePlugin(ctx context.Context, pl framework.ScorePlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) (int64, *framework.Status) {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.Score(ctx, state, pod, nodeName)
|
|
}
|
|
startTime := time.Now()
|
|
s, status := pl.Score(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(score, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return s, status
|
|
}
|
|
|
|
func (f *frameworkImpl) runScoreExtension(ctx context.Context, pl framework.ScorePlugin, state *framework.CycleState, pod *v1.Pod, nodeScoreList framework.NodeScoreList) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.ScoreExtensions().NormalizeScore(ctx, state, pod, nodeScoreList)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.ScoreExtensions().NormalizeScore(ctx, state, pod, nodeScoreList)
|
|
f.metricsRecorder.observePluginDurationAsync(scoreExtensionNormalize, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunPreBindPlugins runs the set of configured prebind plugins. It returns a
|
|
// failure (bool) if any of the plugins returns an error. It also returns an
|
|
// error containing the rejection message or the error occurred in the plugin.
|
|
func (f *frameworkImpl) RunPreBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(preBind, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
for _, pl := range f.preBindPlugins {
|
|
status = f.runPreBindPlugin(ctx, pl, state, pod, nodeName)
|
|
if !status.IsSuccess() {
|
|
msg := fmt.Sprintf("error while running %q prebind plugin for pod %q: %v", pl.Name(), pod.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runPreBindPlugin(ctx context.Context, pl framework.PreBindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.PreBind(ctx, state, pod, nodeName)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.PreBind(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(preBind, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunBindPlugins runs the set of configured bind plugins until one returns a non `Skip` status.
|
|
func (f *frameworkImpl) RunBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(bind, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
if len(f.bindPlugins) == 0 {
|
|
return framework.NewStatus(framework.Skip, "")
|
|
}
|
|
for _, bp := range f.bindPlugins {
|
|
status = f.runBindPlugin(ctx, bp, state, pod, nodeName)
|
|
if status != nil && status.Code() == framework.Skip {
|
|
continue
|
|
}
|
|
if !status.IsSuccess() {
|
|
msg := fmt.Sprintf("plugin %q failed to bind pod \"%v/%v\": %v", bp.Name(), pod.Namespace, pod.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
return status
|
|
}
|
|
return status
|
|
}
|
|
|
|
func (f *frameworkImpl) runBindPlugin(ctx context.Context, bp framework.BindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return bp.Bind(ctx, state, pod, nodeName)
|
|
}
|
|
startTime := time.Now()
|
|
status := bp.Bind(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(bind, bp.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunPostBindPlugins runs the set of configured postbind plugins.
|
|
func (f *frameworkImpl) RunPostBindPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(postBind, framework.Success.String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
for _, pl := range f.postBindPlugins {
|
|
f.runPostBindPlugin(ctx, pl, state, pod, nodeName)
|
|
}
|
|
}
|
|
|
|
func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.PostBindPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
pl.PostBind(ctx, state, pod, nodeName)
|
|
return
|
|
}
|
|
startTime := time.Now()
|
|
pl.PostBind(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(postBind, pl.Name(), nil, metrics.SinceInSeconds(startTime))
|
|
}
|
|
|
|
// RunReservePluginsReserve runs the Reserve method in the set of configured
|
|
// reserve plugins. If any of these plugins returns an error, it does not
|
|
// continue running the remaining ones and returns the error. In such a case,
|
|
// the pod will not be scheduled.
|
|
func (f *frameworkImpl) RunReservePluginsReserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(reserve, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
for _, pl := range f.reservePlugins {
|
|
status = f.runReservePluginReserve(ctx, pl, state, pod, nodeName)
|
|
if !status.IsSuccess() {
|
|
msg := fmt.Sprintf("error while running Reserve in %q reserve plugin for pod %q: %v", pl.Name(), pod.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runReservePluginReserve(ctx context.Context, pl framework.ReservePlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.Reserve(ctx, state, pod, nodeName)
|
|
}
|
|
startTime := time.Now()
|
|
status := pl.Reserve(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(reserve, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status
|
|
}
|
|
|
|
// RunReservePluginsUnreserve runs the Unreserve method in the set of
|
|
// configured reserve plugins.
|
|
func (f *frameworkImpl) RunReservePluginsUnreserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(unreserve, framework.Success.String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
// Execute the Unreserve operation of each reserve plugin in the
|
|
// *reverse* order in which the Reserve operation was executed.
|
|
for i := len(f.reservePlugins) - 1; i >= 0; i-- {
|
|
f.runReservePluginUnreserve(ctx, f.reservePlugins[i], state, pod, nodeName)
|
|
}
|
|
}
|
|
|
|
func (f *frameworkImpl) runReservePluginUnreserve(ctx context.Context, pl framework.ReservePlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
pl.Unreserve(ctx, state, pod, nodeName)
|
|
return
|
|
}
|
|
startTime := time.Now()
|
|
pl.Unreserve(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(unreserve, pl.Name(), nil, metrics.SinceInSeconds(startTime))
|
|
}
|
|
|
|
// RunPermitPlugins runs the set of configured permit plugins. If any of these
|
|
// plugins returns a status other than "Success" or "Wait", it does not continue
|
|
// running the remaining plugins and returns an error. Otherwise, if any of the
|
|
// plugins returns "Wait", then this function will create and add waiting pod
|
|
// to a map of currently waiting pods and return status with "Wait" code.
|
|
// Pod will remain waiting pod for the minimum duration returned by the permit plugins.
|
|
func (f *frameworkImpl) RunPermitPlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
|
|
startTime := time.Now()
|
|
defer func() {
|
|
metrics.FrameworkExtensionPointDuration.WithLabelValues(permit, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
|
|
}()
|
|
pluginsWaitTime := make(map[string]time.Duration)
|
|
statusCode := framework.Success
|
|
for _, pl := range f.permitPlugins {
|
|
status, timeout := f.runPermitPlugin(ctx, pl, state, pod, nodeName)
|
|
if !status.IsSuccess() {
|
|
if status.IsUnschedulable() {
|
|
msg := fmt.Sprintf("rejected pod %q by permit plugin %q: %v", pod.Name, pl.Name(), status.Message())
|
|
klog.V(4).Infof(msg)
|
|
return framework.NewStatus(status.Code(), msg)
|
|
}
|
|
if status.Code() == framework.Wait {
|
|
// Not allowed to be greater than maxTimeout.
|
|
if timeout > maxTimeout {
|
|
timeout = maxTimeout
|
|
}
|
|
pluginsWaitTime[pl.Name()] = timeout
|
|
statusCode = framework.Wait
|
|
} else {
|
|
msg := fmt.Sprintf("error while running %q permit plugin for pod %q: %v", pl.Name(), pod.Name, status.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
}
|
|
}
|
|
if statusCode == framework.Wait {
|
|
waitingPod := newWaitingPod(pod, pluginsWaitTime)
|
|
f.waitingPods.add(waitingPod)
|
|
msg := fmt.Sprintf("one or more plugins asked to wait and no plugin rejected pod %q", pod.Name)
|
|
klog.V(4).Infof(msg)
|
|
return framework.NewStatus(framework.Wait, msg)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (f *frameworkImpl) runPermitPlugin(ctx context.Context, pl framework.PermitPlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) (*framework.Status, time.Duration) {
|
|
if !state.ShouldRecordPluginMetrics() {
|
|
return pl.Permit(ctx, state, pod, nodeName)
|
|
}
|
|
startTime := time.Now()
|
|
status, timeout := pl.Permit(ctx, state, pod, nodeName)
|
|
f.metricsRecorder.observePluginDurationAsync(permit, pl.Name(), status, metrics.SinceInSeconds(startTime))
|
|
return status, timeout
|
|
}
|
|
|
|
// WaitOnPermit will block, if the pod is a waiting pod, until the waiting pod is rejected or allowed.
|
|
func (f *frameworkImpl) WaitOnPermit(ctx context.Context, pod *v1.Pod) (status *framework.Status) {
|
|
waitingPod := f.waitingPods.get(pod.UID)
|
|
if waitingPod == nil {
|
|
return nil
|
|
}
|
|
defer f.waitingPods.remove(pod.UID)
|
|
klog.V(4).Infof("pod %q waiting on permit", pod.Name)
|
|
|
|
startTime := time.Now()
|
|
s := <-waitingPod.s
|
|
metrics.PermitWaitDuration.WithLabelValues(s.Code().String()).Observe(metrics.SinceInSeconds(startTime))
|
|
|
|
if !s.IsSuccess() {
|
|
if s.IsUnschedulable() {
|
|
msg := fmt.Sprintf("pod %q rejected while waiting on permit: %v", pod.Name, s.Message())
|
|
klog.V(4).Infof(msg)
|
|
return framework.NewStatus(s.Code(), msg)
|
|
}
|
|
msg := fmt.Sprintf("error received while waiting on permit for pod %q: %v", pod.Name, s.Message())
|
|
klog.Error(msg)
|
|
return framework.NewStatus(framework.Error, msg)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// SnapshotSharedLister returns the scheduler's SharedLister of the latest NodeInfo
|
|
// snapshot. The snapshot is taken at the beginning of a scheduling cycle and remains
|
|
// unchanged until a pod finishes "Reserve". There is no guarantee that the information
|
|
// remains unchanged after "Reserve".
|
|
func (f *frameworkImpl) SnapshotSharedLister() framework.SharedLister {
|
|
return f.snapshotSharedLister
|
|
}
|
|
|
|
// IterateOverWaitingPods acquires a read lock and iterates over the WaitingPods map.
|
|
func (f *frameworkImpl) IterateOverWaitingPods(callback func(framework.WaitingPod)) {
|
|
f.waitingPods.iterate(callback)
|
|
}
|
|
|
|
// GetWaitingPod returns a reference to a WaitingPod given its UID.
|
|
func (f *frameworkImpl) GetWaitingPod(uid types.UID) framework.WaitingPod {
|
|
if wp := f.waitingPods.get(uid); wp != nil {
|
|
return wp
|
|
}
|
|
return nil // Returning nil instead of *waitingPod(nil).
|
|
}
|
|
|
|
// RejectWaitingPod rejects a WaitingPod given its UID.
|
|
func (f *frameworkImpl) RejectWaitingPod(uid types.UID) {
|
|
waitingPod := f.waitingPods.get(uid)
|
|
if waitingPod != nil {
|
|
waitingPod.Reject("removed")
|
|
}
|
|
}
|
|
|
|
// HasFilterPlugins returns true if at least one filter plugin is defined.
|
|
func (f *frameworkImpl) HasFilterPlugins() bool {
|
|
return len(f.filterPlugins) > 0
|
|
}
|
|
|
|
// HasPostFilterPlugins returns true if at least one postFilter plugin is defined.
|
|
func (f *frameworkImpl) HasPostFilterPlugins() bool {
|
|
return len(f.postFilterPlugins) > 0
|
|
}
|
|
|
|
// HasScorePlugins returns true if at least one score plugin is defined.
|
|
func (f *frameworkImpl) HasScorePlugins() bool {
|
|
return len(f.scorePlugins) > 0
|
|
}
|
|
|
|
// ListPlugins returns a map of extension point name to plugin names configured at each extension
|
|
// point. Returns nil if no plugins where configured.
|
|
func (f *frameworkImpl) ListPlugins() map[string][]config.Plugin {
|
|
m := make(map[string][]config.Plugin)
|
|
|
|
for _, e := range f.getExtensionPoints(&config.Plugins{}) {
|
|
plugins := reflect.ValueOf(e.slicePtr).Elem()
|
|
extName := plugins.Type().Elem().Name()
|
|
var cfgs []config.Plugin
|
|
for i := 0; i < plugins.Len(); i++ {
|
|
name := plugins.Index(i).Interface().(framework.Plugin).Name()
|
|
p := config.Plugin{Name: name}
|
|
if extName == "ScorePlugin" {
|
|
// Weights apply only to score plugins.
|
|
p.Weight = int32(f.pluginNameToWeightMap[name])
|
|
}
|
|
cfgs = append(cfgs, p)
|
|
}
|
|
if len(cfgs) > 0 {
|
|
m[extName] = cfgs
|
|
}
|
|
}
|
|
if len(m) > 0 {
|
|
return m
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ClientSet returns a kubernetes clientset.
|
|
func (f *frameworkImpl) ClientSet() clientset.Interface {
|
|
return f.clientSet
|
|
}
|
|
|
|
// EventRecorder returns an event recorder.
|
|
func (f *frameworkImpl) EventRecorder() events.EventRecorder {
|
|
return f.eventRecorder
|
|
}
|
|
|
|
// SharedInformerFactory returns a shared informer factory.
|
|
func (f *frameworkImpl) SharedInformerFactory() informers.SharedInformerFactory {
|
|
return f.informerFactory
|
|
}
|
|
|
|
func (f *frameworkImpl) pluginsNeeded(plugins *config.Plugins) map[string]config.Plugin {
|
|
pgMap := make(map[string]config.Plugin)
|
|
|
|
if plugins == nil {
|
|
return pgMap
|
|
}
|
|
|
|
find := func(pgs *config.PluginSet) {
|
|
if pgs == nil {
|
|
return
|
|
}
|
|
for _, pg := range pgs.Enabled {
|
|
pgMap[pg.Name] = pg
|
|
}
|
|
}
|
|
for _, e := range f.getExtensionPoints(plugins) {
|
|
find(e.plugins)
|
|
}
|
|
return pgMap
|
|
}
|
|
|
|
// PreemptHandle returns the internal preemptHandle object.
|
|
func (f *frameworkImpl) PreemptHandle() framework.PreemptHandle {
|
|
return f.preemptHandle
|
|
}
|