github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-21 19:01:49 +00:00
Code Issues Projects Releases Wiki Activity

scheduler: merge Reserve and Unreserve plugins

Browse Source 
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.
This commit is contained in:
Adhityaa Chandrasekar 2020-06-15 21:52:54 +00:00
parent 8adcd7978e
commit ec83143342
23 changed files with 243 additions and 288 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cmd/kube-scheduler/app/server_test.go
View File

@ -200,7 +200,6 @@ profiles:
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
}
@ -234,7 +233,6 @@ profiles:
"QueueSortPlugin": {{Name: "PrioritySort"}},
"ScorePlugin": {{Name: "InterPodAffinity", Weight: 1}, {Name: "TaintToleration", Weight: 1}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
},
@ -251,7 +249,6 @@ profiles:
"BindPlugin": {{Name: "DefaultBinder"}},
"QueueSortPlugin": {{Name: "PrioritySort"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
},
@ -333,7 +330,6 @@ profiles:
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
},

5
pkg/scheduler/algorithmprovider/registry.go
View File

@ -142,11 +142,6 @@ func getDefaultConfig() *schedulerapi.Plugins {
{Name: volumebinding.Name},
},
},
Unreserve: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},
},
},
PreBind: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},

15
pkg/scheduler/algorithmprovider/registry_test.go
View File

@ -112,11 +112,6 @@ func TestClusterAutoscalerProvider(t *testing.T) {
{Name: volumebinding.Name},
},
},
Unreserve: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},
},
},
PreBind: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},
@ -209,11 +204,6 @@ func TestApplyFeatureGates(t *testing.T) {
{Name: volumebinding.Name},
},
},
Unreserve: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},
},
},
PreBind: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},
@ -292,11 +282,6 @@ func TestApplyFeatureGates(t *testing.T) {
{Name: volumebinding.Name},
},
},
Unreserve: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},
},
},
PreBind: &schedulerapi.PluginSet{
Enabled: []schedulerapi.Plugin{
{Name: volumebinding.Name},

15
pkg/scheduler/apis/config/testing/compatibility_test.go
View File

@ -704,7 +704,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
wantExtenders: []config.Extender{{
@ -811,7 +810,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
wantExtenders: []config.Extender{{
@ -931,7 +929,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
wantExtenders: []config.Extender{{
@ -1053,7 +1050,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
wantExtenders: []config.Extender{{
@ -1175,7 +1171,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
wantExtenders: []config.Extender{{
@ -1301,7 +1296,6 @@ func TestCompatibility_v1_Scheduler(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
},
wantExtenders: []config.Extender{{
@ -1430,7 +1424,6 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
},
"BindPlugin": {{Name: "DefaultBinder"}},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
}
@ -1500,7 +1493,6 @@ func TestAlgorithmProviderCompatibility(t *testing.T) {
{Name: "DefaultPodTopologySpread", Weight: 1},
},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
"BindPlugin": {{Name: "DefaultBinder"}},
},
@ -1591,7 +1583,6 @@ func TestPluginsConfigurationCompatibility(t *testing.T) {
{Name: "DefaultPodTopologySpread", Weight: 1},
},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
"BindPlugin": {{Name: "DefaultBinder"}},
}
@ -1812,11 +1803,6 @@ func TestPluginsConfigurationCompatibility(t *testing.T) {
{Name: "VolumeBinding"},
},
},
Unreserve: &config.PluginSet{
Disabled: []config.Plugin{
{Name: "VolumeBinding"},
},
},
},
wantPlugins: map[string][]config.Plugin{
"QueueSortPlugin": {
@ -1941,7 +1927,6 @@ func TestPluginsConfigurationCompatibility(t *testing.T) {
{Name: "NodeResourcesBalancedAllocation", Weight: 24},
},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
"BindPlugin": {{Name: "DefaultBinder"}},
},

8
pkg/scheduler/apis/config/types.go
View File

@ -192,7 +192,8 @@ type Plugins struct {
// Score is a list of plugins that should be invoked when ranking nodes that have passed the filtering phase.
Score *PluginSet
// Reserve is a list of plugins invoked when reserving a node to run the pod.
// Reserve is a list of plugins invoked when reserving/unreserving resources
// after a node is assigned to run the pod.
Reserve *PluginSet
// Permit is a list of plugins that control binding of a Pod. These plugins can prevent or delay binding of a Pod.
@ -207,9 +208,6 @@ type Plugins struct {
// PostBind is a list of plugins that should be invoked after a pod is successfully bound.
PostBind *PluginSet
// Unreserve is a list of plugins invoked when a pod that was previously reserved is rejected in a later phase.
Unreserve *PluginSet
}
// PluginSet specifies enabled and disabled plugins for an extension point.
@ -288,7 +286,6 @@ func (p *Plugins) Append(src *Plugins) {
p.PreBind = appendPluginSet(p.PreBind, src.PreBind)
p.Bind = appendPluginSet(p.Bind, src.Bind)
p.PostBind = appendPluginSet(p.PostBind, src.PostBind)
p.Unreserve = appendPluginSet(p.Unreserve, src.Unreserve)
}
// Apply merges the plugin configuration from custom plugins, handling disabled sets.
@ -308,7 +305,6 @@ func (p *Plugins) Apply(customPlugins *Plugins) {
p.PreBind = mergePluginSets(p.PreBind, customPlugins.PreBind)
p.Bind = mergePluginSets(p.Bind, customPlugins.Bind)
p.PostBind = mergePluginSets(p.PostBind, customPlugins.PostBind)
p.Unreserve = mergePluginSets(p.Unreserve, customPlugins.Unreserve)
}
func mergePluginSets(defaultPluginSet, customPluginSet *PluginSet) *PluginSet {

6
pkg/scheduler/apis/config/types_test.go
View File

@ -64,7 +64,6 @@ func TestPluginsAppend(t *testing.T) {
PreBind: &PluginSet{},
Bind: &PluginSet{},
PostBind: &PluginSet{},
Unreserve: &PluginSet{},
},
},
{
@ -126,7 +125,6 @@ func TestPluginsApply(t *testing.T) {
PreBind: &PluginSet{Enabled: []Plugin{}},
Bind: &PluginSet{Enabled: []Plugin{}},
PostBind: &PluginSet{Enabled: []Plugin{}},
Unreserve: &PluginSet{Enabled: []Plugin{}},
},
},
{
@ -168,7 +166,6 @@ func TestPluginsApply(t *testing.T) {
PreBind: &PluginSet{Enabled: []Plugin{}},
Bind: &PluginSet{Enabled: []Plugin{}},
PostBind: &PluginSet{Enabled: []Plugin{}},
Unreserve: &PluginSet{Enabled: []Plugin{}},
},
},
{
@ -211,7 +208,6 @@ func TestPluginsApply(t *testing.T) {
PreBind: &PluginSet{Enabled: []Plugin{}},
Bind: &PluginSet{Enabled: []Plugin{}},
PostBind: &PluginSet{Enabled: []Plugin{}},
Unreserve: &PluginSet{Enabled: []Plugin{}},
},
},
{
@ -252,7 +248,6 @@ func TestPluginsApply(t *testing.T) {
PreBind: &PluginSet{Enabled: []Plugin{}},
Bind: &PluginSet{Enabled: []Plugin{}},
PostBind: &PluginSet{Enabled: []Plugin{}},
Unreserve: &PluginSet{Enabled: []Plugin{}},
},
},
{
@ -282,7 +277,6 @@ func TestPluginsApply(t *testing.T) {
PreBind: nil,
Bind: nil,
PostBind: nil,
Unreserve: nil,
},
},
}

18
pkg/scheduler/apis/config/v1beta1/zz_generated.conversion.go generated
View File

@ -740,15 +740,6 @@ func autoConvert_v1beta1_Plugins_To_config_Plugins(in *v1beta1.Plugins, out *con
} else {
out.PostBind = nil
}
if in.Unreserve != nil {
in, out := &in.Unreserve, &out.Unreserve
*out = new(config.PluginSet)
if err := Convert_v1beta1_PluginSet_To_config_PluginSet(*in, *out, s); err != nil {
return err
}
} else {
out.Unreserve = nil
}
return nil
}
@ -857,15 +848,6 @@ func autoConvert_config_Plugins_To_v1beta1_Plugins(in *config.Plugins, out *v1be
} else {
out.PostBind = nil
}
if in.Unreserve != nil {
in, out := &in.Unreserve, &out.Unreserve
*out = new(v1beta1.PluginSet)
if err := Convert_config_PluginSet_To_v1beta1_PluginSet(*in, *out, s); err != nil {
return err
}
} else {
out.Unreserve = nil
}
return nil
}

5
pkg/scheduler/apis/config/zz_generated.deepcopy.go generated
View File

@ -486,11 +486,6 @@ func (in *Plugins) DeepCopyInto(out *Plugins) {
*out = new(PluginSet)
(*in).DeepCopyInto(*out)
}
if in.Unreserve != nil {
in, out := &in.Unreserve, &out.Unreserve
*out = new(PluginSet)
(*in).DeepCopyInto(*out)
}
return
}

17
pkg/scheduler/framework/plugins/examples/multipoint/multipoint.go
View File

@ -50,7 +50,7 @@ func (s *stateData) Clone() framework.StateData {
return copy
}
// Reserve is the functions invoked by the framework at "reserve" extension point.
// Reserve is the function invoked by the framework at "reserve" extension point.
func (mc CommunicatingPlugin) Reserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
if pod == nil {
return framework.NewStatus(framework.Error, "pod cannot be nil")
@ -63,7 +63,20 @@ func (mc CommunicatingPlugin) Reserve(ctx context.Context, state *framework.Cycl
return nil
}
// PreBind is the functions invoked by the framework at "prebind" extension point.
// Unreserve is the function invoked by the framework when any error happens
// during "reserve" extension point or later.
func (mc CommunicatingPlugin) Unreserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
if pod.Name == "my-test-pod" {
state.Lock()
// The pod is at the end of its lifecycle -- let's clean up the allocated
// resources. In this case, our clean up is simply deleting the key written
// in the Reserve operation.
state.Delete(framework.StateKey(pod.Name))
state.Unlock()
}
}
// PreBind is the function invoked by the framework at "prebind" extension point.
func (mc CommunicatingPlugin) PreBind(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
if pod == nil {
return framework.NewStatus(framework.Error, "pod cannot be nil")

31
pkg/scheduler/framework/plugins/examples/stateful/stateful.go
View File

@ -31,8 +31,7 @@ import (
// This plugin is stateful. It receives arguments at initialization (NewMultipointPlugin)
// and changes its state when it is executed.
type MultipointExample struct {
mpState map[int]string
numRuns int
executionPoints []string
mu sync.RWMutex
}
@ -47,19 +46,35 @@ func (mp *MultipointExample) Name() string {
return Name
}
// Reserve is the functions invoked by the framework at "reserve" extension point.
// Reserve is the function invoked by the framework at "reserve" extension
// point. In this trivial example, the Reserve method allocates an array of
// strings.
func (mp *MultipointExample) Reserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
// Reserve is not called concurrently, and so we don't need to lock.
mp.numRuns++
mp.executionPoints = append(mp.executionPoints, "reserve")
return nil
}
// PreBind is the functions invoked by the framework at "prebind" extension point.
// Unreserve is the function invoked by the framework when any error happens
// during "reserve" extension point or later. In this example, the Unreserve
// method loses its reference to the string slice, allowing it to be garbage
// collected, and thereby "unallocating" the reserved resources.
func (mp *MultipointExample) Unreserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
// Unlike Reserve, the Unreserve method may be called concurrently since
// there is no guarantee that there will only one unreserve operation at any
// given point in time (for example, during the binding cycle).
mp.mu.Lock()
defer mp.mu.Unlock()
mp.executionPoints = nil
}
// PreBind is the function invoked by the framework at "prebind" extension
// point.
func (mp *MultipointExample) PreBind(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
// PreBind could be called concurrently for different pods.
mp.mu.Lock()
defer mp.mu.Unlock()
mp.numRuns++
mp.executionPoints = append(mp.executionPoints, "pre-bind")
if pod == nil {
return framework.NewStatus(framework.Error, "pod must not be nil")
}
@ -72,8 +87,6 @@ func New(config *runtime.Unknown, _ framework.FrameworkHandle) (framework.Plugin
klog.Error("MultipointExample configuration cannot be empty")
return nil, fmt.Errorf("MultipointExample configuration cannot be empty")
}
mp := MultipointExample{
mpState: make(map[int]string),
}
mp := MultipointExample{}
return &mp, nil
}

1
pkg/scheduler/framework/plugins/legacy_registry.go
View File

@ -273,7 +273,6 @@ func NewLegacyRegistry() *LegacyRegistry {
plugins.Filter = appendToPluginSet(plugins.Filter, volumebinding.Name, nil)
plugins.Reserve = appendToPluginSet(plugins.Reserve, volumebinding.Name, nil)
plugins.PreBind = appendToPluginSet(plugins.PreBind, volumebinding.Name, nil)
plugins.Unreserve = appendToPluginSet(plugins.Unreserve, volumebinding.Name, nil)
return
})
registry.registerPredicateConfigProducer(NoDiskConflictPred,

1
pkg/scheduler/framework/plugins/legacy_registry_test.go
View File

@ -115,7 +115,6 @@ func TestRegisterConfigProducers(t *testing.T) {
PreBind: &config.PluginSet{},
Bind: &config.PluginSet{},
PostBind: &config.PluginSet{},
Unreserve: &config.PluginSet{},
}
if diff := cmp.Diff(wantPlugins, gotPlugins); diff != "" {

1
pkg/scheduler/framework/plugins/volumebinding/volume_binding.go
View File

@ -66,7 +66,6 @@ var _ framework.PreFilterPlugin = &VolumeBinding{}
var _ framework.FilterPlugin = &VolumeBinding{}
var _ framework.ReservePlugin = &VolumeBinding{}
var _ framework.PreBindPlugin = &VolumeBinding{}
var _ framework.UnreservePlugin = &VolumeBinding{}
// Name is the name of the plugin used in Registry and configurations.
const Name = "VolumeBinding"

30
pkg/scheduler/framework/runtime/framework.go
View File

@ -77,7 +77,6 @@ type frameworkImpl struct {
preBindPlugins []framework.PreBindPlugin
bindPlugins []framework.BindPlugin
postBindPlugins []framework.PostBindPlugin
unreservePlugins []framework.UnreservePlugin
permitPlugins []framework.PermitPlugin
clientSet clientset.Interface
@ -116,7 +115,6 @@ func (f *frameworkImpl) getExtensionPoints(plugins *config.Plugins) []extensionP
{plugins.PreBind, &f.preBindPlugins},
{plugins.Bind, &f.bindPlugins},
{plugins.PostBind, &f.postBindPlugins},
{plugins.Unreserve, &f.unreservePlugins},
{plugins.Permit, &f.permitPlugins},
{plugins.QueueSort, &f.queueSortPlugins},
}
@ -787,18 +785,19 @@ func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.Post
f.metricsRecorder.observePluginDurationAsync(postBind, pl.Name(), nil, metrics.SinceInSeconds(startTime))
}
// RunReservePlugins runs 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 case, pod will not be scheduled.
func (f *frameworkImpl) RunReservePlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) (status *framework.Status) {
// 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.runReservePlugin(ctx, pl, state, pod, nodeName)
status = f.runReservePluginReserve(ctx, pl, state, pod, nodeName)
if !status.IsSuccess() {
msg := fmt.Sprintf("error while running %q reserve plugin for pod %q: %v", pl.Name(), pod.Name, status.Message())
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)
}
@ -806,7 +805,7 @@ func (f *frameworkImpl) RunReservePlugins(ctx context.Context, state *framework.
return nil
}
func (f *frameworkImpl) runReservePlugin(ctx context.Context, pl framework.ReservePlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
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)
}
@ -816,18 +815,21 @@ func (f *frameworkImpl) runReservePlugin(ctx context.Context, pl framework.Reser
return status
}
// RunUnreservePlugins runs the set of configured unreserve plugins.
func (f *frameworkImpl) RunUnreservePlugins(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
// 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))
}()
for _, pl := range f.unreservePlugins {
f.runUnreservePlugin(ctx, pl, state, pod, nodeName)
// 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) runUnreservePlugin(ctx context.Context, pl framework.UnreservePlugin, state *framework.CycleState, pod *v1.Pod, nodeName string) {
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

29
pkg/scheduler/framework/runtime/framework_test.go
View File

@ -184,6 +184,9 @@ func (pl *TestPlugin) Reserve(ctx context.Context, state *v1alpha1.CycleState, p
return v1alpha1.NewStatus(v1alpha1.Code(pl.inj.ReserveStatus), "injected status")
}
func (pl *TestPlugin) Unreserve(ctx context.Context, state *v1alpha1.CycleState, p *v1.Pod, nodeName string) {
}
func (pl *TestPlugin) PreBind(ctx context.Context, state *v1alpha1.CycleState, p *v1.Pod, nodeName string) *v1alpha1.Status {
return v1alpha1.NewStatus(v1alpha1.Code(pl.inj.PreBindStatus), "injected status")
}
@ -191,9 +194,6 @@ func (pl *TestPlugin) PreBind(ctx context.Context, state *v1alpha1.CycleState, p
func (pl *TestPlugin) PostBind(ctx context.Context, state *v1alpha1.CycleState, p *v1.Pod, nodeName string) {
}
func (pl *TestPlugin) Unreserve(ctx context.Context, state *v1alpha1.CycleState, p *v1.Pod, nodeName string) {
}
func (pl *TestPlugin) Permit(ctx context.Context, state *v1alpha1.CycleState, p *v1.Pod, nodeName string) (*v1alpha1.Status, time.Duration) {
return v1alpha1.NewStatus(v1alpha1.Code(pl.inj.PermitStatus), "injected status"), time.Duration(0)
}
@ -1322,7 +1322,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.Unschedulable)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin" reserve plugin for pod "": injected status`),
},
{
name: "ErrorReservePlugin",
@ -1332,7 +1332,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.Error)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin" reserve plugin for pod "": injected status`),
},
{
name: "UnschedulableReservePlugin",
@ -1342,7 +1342,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.UnschedulableAndUnresolvable)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin" reserve plugin for pod "": injected status`),
},
{
name: "SuccessSuccessReservePlugins",
@ -1370,7 +1370,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.Error)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin" reserve plugin for pod "": injected status`),
},
{
name: "SuccessErrorReservePlugins",
@ -1384,7 +1384,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.Error)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin 1" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin 1" reserve plugin for pod "": injected status`),
},
{
name: "ErrorSuccessReservePlugin",
@ -1398,7 +1398,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.Success)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin" reserve plugin for pod "": injected status`),
},
{
name: "UnschedulableAndSuccessReservePlugin",
@ -1412,7 +1412,7 @@ func TestReservePlugins(t *testing.T) {
inj: injectedResult{ReserveStatus: int(v1alpha1.Success)},
},
},
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running "TestPlugin" reserve plugin for pod "": injected status`),
wantStatus: v1alpha1.NewStatus(v1alpha1.Error, `error while running Reserve in "TestPlugin" reserve plugin for pod "": injected status`),
},
}
@ -1440,7 +1440,7 @@ func TestReservePlugins(t *testing.T) {
t.Fatalf("fail to create framework: %s", err)
}
status := f.RunReservePlugins(context.TODO(), nil, pod, "")
status := f.RunReservePluginsReserve(context.TODO(), nil, pod, "")
if !reflect.DeepEqual(status, tt.wantStatus) {
t.Errorf("wrong status code. got %v, want %v", status, tt.wantStatus)
@ -1602,13 +1602,13 @@ func TestRecordingMetrics(t *testing.T) {
},
{
name: "Reserve - Success",
action: func(f v1alpha1.Framework) { f.RunReservePlugins(context.Background(), state, pod, "") },
action: func(f v1alpha1.Framework) { f.RunReservePluginsReserve(context.Background(), state, pod, "") },
wantExtensionPoint: "Reserve",
wantStatus: v1alpha1.Success,
},
{
name: "Unreserve - Success",
action: func(f v1alpha1.Framework) { f.RunUnreservePlugins(context.Background(), state, pod, "") },
action: func(f v1alpha1.Framework) { f.RunReservePluginsUnreserve(context.Background(), state, pod, "") },
wantExtensionPoint: "Unreserve",
wantStatus: v1alpha1.Success,
},
@ -1660,7 +1660,7 @@ func TestRecordingMetrics(t *testing.T) {
},
{
name: "Reserve - Error",
action: func(f v1alpha1.Framework) { f.RunReservePlugins(context.Background(), state, pod, "") },
action: func(f v1alpha1.Framework) { f.RunReservePluginsReserve(context.Background(), state, pod, "") },
inject: injectedResult{ReserveStatus: int(v1alpha1.Error)},
wantExtensionPoint: "Reserve",
wantStatus: v1alpha1.Error,
@ -1718,7 +1718,6 @@ func TestRecordingMetrics(t *testing.T) {
PreBind: pluginSet,
Bind: pluginSet,
PostBind: pluginSet,
Unreserve: pluginSet,
}
recorder := newMetricsRecorder(100, time.Nanosecond)
f, err := newFrameworkWithQueueSortAndBind(r, plugins, emptyArgs, withMetricsRecorder(recorder), WithProfileName(testProfileName))

41
pkg/scheduler/framework/v1alpha1/interface.go
View File

@ -324,17 +324,21 @@ type ScorePlugin interface {
ScoreExtensions() ScoreExtensions
}
// ReservePlugin is an interface for Reserve plugins. These plugins are called
// at the reservation point. These are meant to update the state of the plugin.
// This concept used to be called 'assume' in the original scheduler.
// These plugins should return only Success or Error in Status.code. However,
// the scheduler accepts other valid codes as well. Anything other than Success
// will lead to rejection of the pod.
// ReservePlugin is an interface for plugins with Reserve and Unreserve
// methods. These are meant to update the state of the plugin. This concept
// used to be called 'assume' in the original scheduler. These plugins should
// return only Success or Error in Status.code. However, the scheduler accepts
// other valid codes as well. Anything other than Success will lead to
// rejection of the pod.
type ReservePlugin interface {
Plugin
// Reserve is called by the scheduling framework when the scheduler cache is
// updated.
Reserve(ctx context.Context, state *CycleState, p *v1.Pod, nodeName string) *Status
// Unreserve is called by the scheduling framework when a reserved pod was
// rejected, an error occurred during reservation of subsequent plugins, or
// in a later phase. The Unreserve method implementation must be idempotent.
Unreserve(ctx context.Context, state *CycleState, p *v1.Pod, nodeName string)
}
// PreBindPlugin is an interface that must be implemented by "prebind" plugins.
@ -357,17 +361,6 @@ type PostBindPlugin interface {
PostBind(ctx context.Context, state *CycleState, p *v1.Pod, nodeName string)
}
// UnreservePlugin is an interface for Unreserve plugins. This is an informational
// extension point. If a pod was reserved and then rejected in a later phase, then
// un-reserve plugins will be notified. Un-reserve plugins should clean up state
// associated with the reserved Pod.
type UnreservePlugin interface {
Plugin
// Unreserve is called by the scheduling framework when a reserved pod was
// rejected in a later phase.
Unreserve(ctx context.Context, state *CycleState, p *v1.Pod, nodeName string)
}
// PermitPlugin is an interface that must be implemented by "permit" plugins.
// These plugins are called before a pod is bound to a node.
type PermitPlugin interface {
@ -452,13 +445,15 @@ type Framework interface {
// RunPostBindPlugins runs the set of configured postbind plugins.
RunPostBindPlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// RunReservePlugins runs 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 case, pod will not be scheduled.
RunReservePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// RunReservePluginsReserve runs the Reserve method of the set of
// configured reserve plugins. If any of these calls returns an error, it
// does not continue running the remaining ones and returns the error. In
// such case, pod will not be scheduled.
RunReservePluginsReserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string) *Status
// RunUnreservePlugins runs the set of configured unreserve plugins.
RunUnreservePlugins(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// RunReservePluginsUnreserve runs the Unreserve method of the set of
// configured reserve plugins.
RunReservePluginsUnreserve(ctx context.Context, state *CycleState, pod *v1.Pod, nodeName string)
// 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

14
pkg/scheduler/scheduler.go
View File

@ -516,8 +516,8 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
assumedPodInfo := podInfo.DeepCopy()
assumedPod := assumedPodInfo.Pod
// Run "reserve" plugins.
if sts := prof.RunReservePlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
// Run the Reserve method of reserve plugins.
if sts := prof.RunReservePluginsReserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost); !sts.IsSuccess() {
sched.recordSchedulingFailure(prof, assumedPodInfo, sts.AsError(), SchedulerError, "")
metrics.PodScheduleError(prof.Name, metrics.SinceInSeconds(start))
return
@ -534,7 +534,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
sched.recordSchedulingFailure(prof, assumedPodInfo, err, SchedulerError, "")
metrics.PodScheduleError(prof.Name, metrics.SinceInSeconds(start))
// trigger un-reserve plugins to clean up state associated with the reserved Pod
prof.RunUnreservePlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
prof.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
return
}
@ -553,7 +553,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
klog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
}
// One of the plugins returned status different than success or wait.
prof.RunUnreservePlugins(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
prof.RunReservePluginsUnreserve(schedulingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
sched.recordSchedulingFailure(prof, assumedPodInfo, runPermitStatus.AsError(), reason, "")
return
}
@ -579,7 +579,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
klog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
}
// trigger un-reserve plugins to clean up state associated with the reserved Pod
prof.RunUnreservePlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
prof.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
sched.recordSchedulingFailure(prof, assumedPodInfo, waitOnPermitStatus.AsError(), reason, "")
return
}
@ -594,7 +594,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
klog.Errorf("scheduler cache ForgetPod failed: %v", forgetErr)
}
// trigger un-reserve plugins to clean up state associated with the reserved Pod
prof.RunUnreservePlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
prof.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
sched.recordSchedulingFailure(prof, assumedPodInfo, preBindStatus.AsError(), reason, "")
return
}
@ -603,7 +603,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
if err != nil {
metrics.PodScheduleError(prof.Name, metrics.SinceInSeconds(start))
// trigger un-reserve plugins to clean up state associated with the reserved Pod
prof.RunUnreservePlugins(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
prof.RunReservePluginsUnreserve(bindingCycleCtx, state, assumedPod, scheduleResult.SuggestedHost)
sched.recordSchedulingFailure(prof, assumedPodInfo, fmt.Errorf("Binding rejected: %v", err), SchedulerError, "")
} else {
// Calculating nodeResourceString can be heavy. Avoid it if klog verbosity is below 2.

4
pkg/scheduler/scheduler_test.go
View File

@ -826,7 +826,7 @@ func setupTestSchedulerWithVolumeBinding(volumeBinder scheduling.SchedulerVolume
st.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New),
st.RegisterPluginAsExtensions(volumebinding.Name, func(plArgs runtime.Object, handle framework.FrameworkHandle) (framework.Plugin, error) {
return &volumebinding.VolumeBinding{Binder: volumeBinder}, nil
}, "PreFilter", "Filter", "Reserve", "Unreserve", "PreBind"),
}, "PreFilter", "Filter", "Reserve", "PreBind"),
}
s, bindingChan, errChan := setupTestScheduler(queuedPodStore, scache, informerFactory, broadcaster, fns...)
informerFactory.Start(stop)
@ -919,7 +919,7 @@ func TestSchedulerWithVolumeBinding(t *testing.T) {
},
expectAssumeCalled: true,
eventReason: "FailedScheduling",
expectError: fmt.Errorf("error while running %q reserve plugin for pod %q: %v", volumebinding.Name, "foo", assumeErr),
expectError: fmt.Errorf("error while running Reserve in %q reserve plugin for pod %q: %v", volumebinding.Name, "foo", assumeErr),
},
{
name: "bind error",

2
pkg/scheduler/testing/framework_helpers.go
View File

@ -100,8 +100,6 @@ func getPluginSetByExtension(plugins *schedulerapi.Plugins, extension string) *s
return initializeIfNeeded(&plugins.Bind)
case "Reserve":
return initializeIfNeeded(&plugins.Reserve)
case "Unreserve":
return initializeIfNeeded(&plugins.Unreserve)
case "Permit":
return initializeIfNeeded(&plugins.Permit)
case "PreBind":

6
staging/src/k8s.io/kube-scheduler/config/v1beta1/types.go
View File

@ -175,7 +175,8 @@ type Plugins struct {
// Score is a list of plugins that should be invoked when ranking nodes that have passed the filtering phase.
Score *PluginSet `json:"score,omitempty"`
// Reserve is a list of plugins invoked when reserving a node to run the pod.
// Reserve is a list of plugins invoked when reserving/unreserving resources
// after a node is assigned to run the pod.
Reserve *PluginSet `json:"reserve,omitempty"`
// Permit is a list of plugins that control binding of a Pod. These plugins can prevent or delay binding of a Pod.
@ -190,9 +191,6 @@ type Plugins struct {
// PostBind is a list of plugins that should be invoked after a pod is successfully bound.
PostBind *PluginSet `json:"postBind,omitempty"`
// Unreserve is a list of plugins invoked when a pod that was previously reserved is rejected in a later phase.
Unreserve *PluginSet `json:"unreserve,omitempty"`
}
// PluginSet specifies enabled and disabled plugins for an extension point.

5
staging/src/k8s.io/kube-scheduler/config/v1beta1/zz_generated.deepcopy.go generated
View File

@ -449,11 +449,6 @@ func (in *Plugins) DeepCopyInto(out *Plugins) {
*out = new(PluginSet)
(*in).DeepCopyInto(*out)
}
if in.Unreserve != nil {
in, out := &in.Unreserve, &out.Unreserve
*out = new(PluginSet)
(*in).DeepCopyInto(*out)
}
return
}

157
test/integration/scheduler/framework_test.go
View File

@ -67,8 +67,11 @@ type PostFilterPlugin struct {
}
type ReservePlugin struct {
name string
numReserveCalled int
failReserve bool
numUnreserveCalled int
pluginInvokeEventChan chan pluginInvokeEvent
}
type PreScorePlugin struct {
@ -96,12 +99,6 @@ type PostBindPlugin struct {
pluginInvokeEventChan chan pluginInvokeEvent
}
type UnreservePlugin struct {
name string
numUnreserveCalled int
pluginInvokeEventChan chan pluginInvokeEvent
}
type PermitPlugin struct {
name string
numPermitCalled int
@ -126,7 +123,6 @@ const (
preScorePluginName = "prescore-plugin"
reservePluginName = "reserve-plugin"
preBindPluginName = "prebind-plugin"
unreservePluginName = "unreserve-plugin"
postBindPluginName = "postbind-plugin"
permitPluginName = "permit-plugin"
)
@ -142,7 +138,6 @@ var _ framework.PreScorePlugin = &PreScorePlugin{}
var _ framework.PreBindPlugin = &PreBindPlugin{}
var _ framework.BindPlugin = &BindPlugin{}
var _ framework.PostBindPlugin = &PostBindPlugin{}
var _ framework.UnreservePlugin = &UnreservePlugin{}
var _ framework.PermitPlugin = &PermitPlugin{}
// newPlugin returns a plugin factory with specified Plugin.
@ -247,11 +242,11 @@ func (fp *FilterPlugin) Filter(ctx context.Context, state *framework.CycleState,
// Name returns name of the plugin.
func (rp *ReservePlugin) Name() string {
return reservePluginName
return rp.name
}
// Reserve is a test function that returns an error or nil, depending on the
// value of "failReserve".
// Reserve is a test function that increments an intenral counter and returns
// an error or nil, depending on the value of "failReserve".
func (rp *ReservePlugin) Reserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) *framework.Status {
rp.numReserveCalled++
if rp.failReserve {
@ -260,9 +255,20 @@ func (rp *ReservePlugin) Reserve(ctx context.Context, state *framework.CycleStat
return nil
}
// reset used to reset reserve plugin.
// Unreserve is a test function that increments an internal counter and emits
// an event to a channel. While Unreserve implementations should normally be
// idempotent, we relax that requirement here for testing purposes.
func (rp *ReservePlugin) Unreserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
rp.numUnreserveCalled++
if rp.pluginInvokeEventChan != nil {
rp.pluginInvokeEventChan <- pluginInvokeEvent{pluginName: rp.Name(), val: rp.numUnreserveCalled}
}
}
// reset used to reset internal counters.
func (rp *ReservePlugin) reset() {
rp.numReserveCalled = 0
rp.numUnreserveCalled = 0
}
// Name returns name of the plugin.
@ -411,25 +417,6 @@ func (pp *PostFilterPlugin) PostFilter(ctx context.Context, state *framework.Cyc
return nil, framework.NewStatus(framework.Success, fmt.Sprintf("make room for pod %v to be schedulable", pod.Name))
}
// Name returns name of the plugin.
func (up *UnreservePlugin) Name() string {
return up.name
}
// Unreserve is a test function that returns an error or nil, depending on the
// value of "failUnreserve".
func (up *UnreservePlugin) Unreserve(ctx context.Context, state *framework.CycleState, pod *v1.Pod, nodeName string) {
up.numUnreserveCalled++
if up.pluginInvokeEventChan != nil {
up.pluginInvokeEventChan <- pluginInvokeEvent{pluginName: up.Name(), val: up.numUnreserveCalled}
}
}
// reset used to reset numUnreserveCalled.
func (up *UnreservePlugin) reset() {
up.numUnreserveCalled = 0
}
// Name returns name of the plugin.
func (pp *PermitPlugin) Name() string {
return pp.name
@ -810,7 +797,7 @@ func TestNormalizeScorePlugin(t *testing.T) {
}
// TestReservePlugin tests invocation of reserve plugins.
func TestReservePlugin(t *testing.T) {
func TestReservePluginReserve(t *testing.T) {
// Create a plugin registry for testing. Register only a reserve plugin.
reservePlugin := &ReservePlugin{}
registry := frameworkruntime.Registry{reservePluginName: newPlugin(reservePlugin)}
@ -830,7 +817,7 @@ func TestReservePlugin(t *testing.T) {
}
// Create the master and the scheduler with the test plugin set.
testCtx := initTestSchedulerForFrameworkTest(t, testutils.InitTestMaster(t, "reserve-plugin", nil), 2,
testCtx := initTestSchedulerForFrameworkTest(t, testutils.InitTestMaster(t, "reserve-plugin-reserve", nil), 2,
scheduler.WithProfiles(prof),
scheduler.WithFrameworkOutOfTreeRegistry(registry))
defer testutils.CleanupTest(t, testCtx)
@ -962,26 +949,40 @@ func TestPrebindPlugin(t *testing.T) {
}
}
// TestUnreservePlugin tests invocation of un-reserve plugin
func TestUnreservePlugin(t *testing.T) {
// TODO: register more plugin which would trigger un-reserve plugin
preBindPlugin := &PreBindPlugin{}
unreservePlugin := &UnreservePlugin{name: unreservePluginName}
registry := frameworkruntime.Registry{
unreservePluginName: newPlugin(unreservePlugin),
preBindPluginName: newPlugin(preBindPlugin),
// TestUnreserveReservePlugin tests invocation of the Unreserve operation in
// reserve plugins through failures in execution points such as pre-bind. Also
// tests that the order of invocation of Unreserve operation is executed in the
// reverse order of invocation of the Reserve operation.
func TestReservePluginUnreserve(t *testing.T) {
numReservePlugins := 3
pluginInvokeEventChan := make(chan pluginInvokeEvent, numReservePlugins)
preBindPlugin := &PreBindPlugin{
failPreBind: true,
}
var reservePlugins []*ReservePlugin
for i := 0; i < numReservePlugins; i++ {
reservePlugins = append(reservePlugins, &ReservePlugin{
name: fmt.Sprintf("%s-%d", reservePluginName, i),
pluginInvokeEventChan: pluginInvokeEventChan,
})
}
// Setup initial unreserve and prebind plugin for testing.
registry := frameworkruntime.Registry{
// TODO(#92229): test more failure points that would trigger Unreserve in
// reserve plugins than just one pre-bind plugin.
preBindPluginName: newPlugin(preBindPlugin),
}
for _, pl := range reservePlugins {
registry[pl.Name()] = newPlugin(pl)
}
// Setup initial reserve and prebind plugin for testing.
prof := schedulerconfig.KubeSchedulerProfile{
SchedulerName: v1.DefaultSchedulerName,
Plugins: &schedulerconfig.Plugins{
Unreserve: &schedulerconfig.PluginSet{
Enabled: []schedulerconfig.Plugin{
{
Name: unreservePluginName,
},
},
Reserve: &schedulerconfig.PluginSet{
// filled by looping over reservePlugins
},
PreBind: &schedulerconfig.PluginSet{
Enabled: []schedulerconfig.Plugin{
@ -992,9 +993,14 @@ func TestUnreservePlugin(t *testing.T) {
},
},
}
for _, pl := range reservePlugins {
prof.Plugins.Reserve.Enabled = append(prof.Plugins.Reserve.Enabled, schedulerconfig.Plugin{
Name: pl.Name(),
})
}
// Create the master and the scheduler with the test plugin set.
testCtx := initTestSchedulerForFrameworkTest(t, testutils.InitTestMaster(t, "unreserve-plugin", nil), 2,
testCtx := initTestSchedulerForFrameworkTest(t, testutils.InitTestMaster(t, "reserve-plugin-unreserve", nil), 2,
scheduler.WithProfiles(prof),
scheduler.WithFrameworkOutOfTreeRegistry(registry))
defer testutils.CleanupTest(t, testCtx)
@ -1004,11 +1010,11 @@ func TestUnreservePlugin(t *testing.T) {
preBindFail bool
}{
{
name: "fail preBind unreserve plugin",
name: "fail preBind",
preBindFail: true,
},
{
name: "do not fail preBind unreserve plugin",
name: "pass preBind",
preBindFail: false,
},
}
@ -1025,22 +1031,34 @@ func TestUnreservePlugin(t *testing.T) {
if test.preBindFail {
if err = wait.Poll(10*time.Millisecond, 30*time.Second, podSchedulingError(testCtx.ClientSet, pod.Namespace, pod.Name)); err != nil {
t.Errorf("Expected a scheduling error, but didn't get it. error: %v", err)
t.Errorf("Expected a scheduling error, but didn't get it: %v", err)
}
for i := numReservePlugins - 1; i >= 0; i-- {
select {
case event := <-pluginInvokeEventChan:
expectedPluginName := reservePlugins[i].Name()
if expectedPluginName != event.pluginName {
t.Errorf("event.pluginName = %s, want %s", event.pluginName, expectedPluginName)
}
case <-time.After(time.Second * 30):
t.Errorf("pluginInvokeEventChan receive timed out")
}
if unreservePlugin.numUnreserveCalled == 0 || unreservePlugin.numUnreserveCalled != preBindPlugin.numPreBindCalled {
t.Errorf("Expected the unreserve plugin to be called %d times, was called %d times.", preBindPlugin.numPreBindCalled, unreservePlugin.numUnreserveCalled)
}
} else {
if err = testutils.WaitForPodToSchedule(testCtx.ClientSet, pod); err != nil {
t.Errorf("Expected the pod to be scheduled. error: %v", err)
t.Errorf("Expected the pod to be scheduled, got an error: %v", err)
}
for i, pl := range reservePlugins {
if pl.numUnreserveCalled != 0 {
t.Errorf("reservePlugins[%d].numUnreserveCalled = %d, want 0", i, pl.numUnreserveCalled)
}
if unreservePlugin.numUnreserveCalled > 0 {
t.Errorf("Didn't expect the unreserve plugin to be called, was called %d times.", unreservePlugin.numUnreserveCalled)
}
}
unreservePlugin.reset()
preBindPlugin.reset()
for _, pl := range reservePlugins {
pl.reset()
}
testutils.CleanupPods(testCtx.ClientSet, t, []*v1.Pod{pod})
})
}
@ -1056,12 +1074,13 @@ func TestBindPlugin(t *testing.T) {
testContext := testutils.InitTestMaster(t, "bind-plugin", nil)
bindPlugin1 := &BindPlugin{PluginName: "bind-plugin-1", client: testContext.ClientSet}
bindPlugin2 := &BindPlugin{PluginName: "bind-plugin-2", client: testContext.ClientSet}
unreservePlugin := &UnreservePlugin{name: "mock-unreserve-plugin"}
reservePlugin := &ReservePlugin{name: "mock-reserve-plugin"}
postBindPlugin := &PostBindPlugin{name: "mock-post-bind-plugin"}
// Create a plugin registry for testing. Register an unreserve, a bind plugin and a postBind plugin.
// Create a plugin registry for testing. Register reserve, bind, and
// postBind plugins.
registry := frameworkruntime.Registry{
unreservePlugin.Name(): func(_ runtime.Object, _ framework.FrameworkHandle) (framework.Plugin, error) {
return unreservePlugin, nil
reservePlugin.Name(): func(_ runtime.Object, _ framework.FrameworkHandle) (framework.Plugin, error) {
return reservePlugin, nil
},
bindPlugin1.Name(): func(_ runtime.Object, _ framework.FrameworkHandle) (framework.Plugin, error) {
return bindPlugin1, nil
@ -1078,8 +1097,8 @@ func TestBindPlugin(t *testing.T) {
prof := schedulerconfig.KubeSchedulerProfile{
SchedulerName: v1.DefaultSchedulerName,
Plugins: &schedulerconfig.Plugins{
Unreserve: &schedulerconfig.PluginSet{
Enabled: []schedulerconfig.Plugin{{Name: unreservePlugin.Name()}},
Reserve: &schedulerconfig.PluginSet{
Enabled: []schedulerconfig.Plugin{{Name: reservePlugin.Name()}},
},
Bind: &schedulerconfig.PluginSet{
// Put DefaultBinder last.
@ -1137,7 +1156,7 @@ func TestBindPlugin(t *testing.T) {
{
name: "bind plugin fails to bind the pod",
bindPluginStatuses: []*framework.Status{framework.NewStatus(framework.Error, "failed to bind"), framework.NewStatus(framework.Success, "")},
expectInvokeEvents: []pluginInvokeEvent{{pluginName: bindPlugin1.Name(), val: 1}, {pluginName: unreservePlugin.Name(), val: 1}, {pluginName: bindPlugin1.Name(), val: 2}, {pluginName: unreservePlugin.Name(), val: 2}},
expectInvokeEvents: []pluginInvokeEvent{{pluginName: bindPlugin1.Name(), val: 1}, {pluginName: reservePlugin.Name(), val: 1}, {pluginName: bindPlugin1.Name(), val: 2}, {pluginName: reservePlugin.Name(), val: 2}},
},
}
@ -1151,7 +1170,7 @@ func TestBindPlugin(t *testing.T) {
bindPlugin1.pluginInvokeEventChan = pluginInvokeEventChan
bindPlugin2.pluginInvokeEventChan = pluginInvokeEventChan
unreservePlugin.pluginInvokeEventChan = pluginInvokeEventChan
reservePlugin.pluginInvokeEventChan = pluginInvokeEventChan
postBindPlugin.pluginInvokeEventChan = pluginInvokeEventChan
// Create a best effort pod.
@ -1194,8 +1213,8 @@ func TestBindPlugin(t *testing.T) {
}); err != nil {
t.Errorf("Expected the postbind plugin to be called once, was called %d times.", postBindPlugin.numPostBindCalled)
}
if unreservePlugin.numUnreserveCalled != 0 {
t.Errorf("Expected the unreserve plugin not to be called, was called %d times.", unreservePlugin.numUnreserveCalled)
if reservePlugin.numUnreserveCalled != 0 {
t.Errorf("Expected unreserve to not be called, was called %d times.", reservePlugin.numUnreserveCalled)
}
} else {
// bind plugin fails to bind the pod
@ -1223,7 +1242,7 @@ func TestBindPlugin(t *testing.T) {
postBindPlugin.reset()
bindPlugin1.reset()
bindPlugin2.reset()
unreservePlugin.reset()
reservePlugin.reset()
testutils.CleanupPods(testCtx.ClientSet, t, []*v1.Pod{pod})
})
}

2
test/integration/scheduler/scheduler_test.go
View File

@ -145,7 +145,6 @@ func TestSchedulerCreationFromConfigMap(t *testing.T) {
{Name: "TaintToleration", Weight: 1},
},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
"BindPlugin": {{Name: "DefaultBinder"}},
},
@ -239,7 +238,6 @@ kind: Policy
{Name: "TaintToleration", Weight: 1},
},
"ReservePlugin": {{Name: "VolumeBinding"}},
"UnreservePlugin": {{Name: "VolumeBinding"}},
"PreBindPlugin": {{Name: "VolumeBinding"}},
"BindPlugin": {{Name: "DefaultBinder"}},
},