github/kubernetes
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #75848 from bsalamat/framework_update

Browse Source 
Implement/update interfaces and skeleton for the scheduling framework
This commit is contained in:
Kubernetes Prow Robot
2019-04-30 00:03:36 -07:00
committed by GitHub
parent cd4ec14772 83828bcb2d
commit cf22c56c7f
37 changed files with 906 additions and 727 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
test/integration/scheduler/BUILD
View File

@@ -11,8 +11,8 @@ go_test(
size = "large",
srcs = [
"extender_test.go",
"framework_test.go",
"main_test.go",
"plugin_test.go",
"predicates_test.go",
"preemption_test.go",
"priorities_test.go",
@@ -35,8 +35,8 @@ go_test(
"//pkg/scheduler/api:go_default_library",
"//pkg/scheduler/apis/config:go_default_library",
"//pkg/scheduler/factory:go_default_library",
"//pkg/scheduler/framework/v1alpha1:go_default_library",
"//pkg/scheduler/nodeinfo:go_default_library",
"//pkg/scheduler/plugins/v1alpha1:go_default_library",
"//pkg/volume:go_default_library",
"//pkg/volume/testing:go_default_library",
"//plugin/pkg/admission/podtolerationrestriction:go_default_library",
@@ -47,6 +47,7 @@ go_test(
"//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/intstr:go_default_library",
@@ -95,7 +96,7 @@ go_library(
"//pkg/scheduler/algorithmprovider:go_default_library",
"//pkg/scheduler/api:go_default_library",
"//pkg/scheduler/factory:go_default_library",
"//pkg/scheduler/plugins/v1alpha1:go_default_library",
"//pkg/scheduler/framework/v1alpha1:go_default_library",
"//pkg/util/taints:go_default_library",
"//staging/src/k8s.io/api/core/v1:go_default_library",
"//staging/src/k8s.io/api/policy/v1beta1:go_default_library",

218
test/integration/scheduler/framework_test.go Normal file
View File

@@ -0,0 +1,218 @@
/*
Copyright 2018 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 scheduler
import (
"fmt"
"k8s.io/apimachinery/pkg/runtime"
"testing"
"time"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/wait"
framework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
)
// TesterPlugin is common ancestor for a test plugin that allows injection of
// failures and some other test functionalities.
type TesterPlugin struct {
numReserveCalled int
numPrebindCalled int
failReserve bool
failPrebind bool
rejectPrebind bool
}
type ReservePlugin struct {
TesterPlugin
}
type PrebindPlugin struct {
TesterPlugin
}
const (
reservePluginName = "reserve-plugin"
prebindPluginName = "prebind-plugin"
)
var _ = framework.ReservePlugin(&ReservePlugin{})
var _ = framework.PrebindPlugin(&PrebindPlugin{})
// Name returns name of the plugin.
func (rp *ReservePlugin) Name() string {
return reservePluginName
}
// Reserve is a test function that returns an error or nil, depending on the
// value of "failReserve".
func (rp *ReservePlugin) Reserve(pc *framework.PluginContext, pod *v1.Pod, nodeName string) *framework.Status {
rp.numReserveCalled++
if rp.failReserve {
return framework.NewStatus(framework.Error, fmt.Sprintf("injecting failure for pod %v", pod.Name))
}
return nil
}
var resPlugin = &ReservePlugin{}
var pbdPlugin = &PrebindPlugin{}
// NewReservePlugin is the factory for reserve plugin.
func NewReservePlugin(_ *runtime.Unknown, _ framework.FrameworkHandle) (framework.Plugin, error) {
return resPlugin, nil
}
// Name returns name of the plugin.
func (pp *PrebindPlugin) Name() string {
return prebindPluginName
}
// Prebind is a test function that returns (true, nil) or errors for testing.
func (pp *PrebindPlugin) Prebind(pc *framework.PluginContext, pod *v1.Pod, nodeName string) *framework.Status {
pp.numPrebindCalled++
if pp.failPrebind {
return framework.NewStatus(framework.Error, fmt.Sprintf("injecting failure for pod %v", pod.Name))
}
if pp.rejectPrebind {
return framework.NewStatus(framework.Unschedulable, fmt.Sprintf("reject pod %v", pod.Name))
}
return nil
}
// NewPrebindPlugin is the factory for prebind plugin.
func NewPrebindPlugin(_ *runtime.Unknown, _ framework.FrameworkHandle) (framework.Plugin, error) {
return pbdPlugin, nil
}
// TestReservePlugin tests invocation of reserve plugins.
func TestReservePlugin(t *testing.T) {
// Create a plugin registry for testing. Register only a reserve plugin.
registry := framework.Registry{reservePluginName: NewReservePlugin}
// Create the master and the scheduler with the test plugin set.
context := initTestSchedulerWithOptions(t,
initTestMaster(t, "reserve-plugin", nil),
false, nil, registry, false, time.Second)
defer cleanupTest(t, context)
cs := context.clientSet
// Add a few nodes.
_, err := createNodes(cs, "test-node", nil, 2)
if err != nil {
t.Fatalf("Cannot create nodes: %v", err)
}
for _, fail := range []bool{false, true} {
resPlugin.failReserve = fail
// Create a best effort pod.
pod, err := createPausePod(cs,
initPausePod(cs, &pausePodConfig{Name: "test-pod", Namespace: context.ns.Name}))
if err != nil {
t.Errorf("Error while creating a test pod: %v", err)
}
if fail {
if err = wait.Poll(10*time.Millisecond, 30*time.Second, podSchedulingError(cs, pod.Namespace, pod.Name)); err != nil {
t.Errorf("Didn't expected the pod to be scheduled. error: %v", err)
}
} else {
if err = waitForPodToSchedule(cs, pod); err != nil {
t.Errorf("Expected the pod to be scheduled. error: %v", err)
}
}
if resPlugin.numReserveCalled == 0 {
t.Errorf("Expected the reserve plugin to be called.")
}
cleanupPods(cs, t, []*v1.Pod{pod})
}
}
// TestPrebindPlugin tests invocation of prebind plugins.
func TestPrebindPlugin(t *testing.T) {
// Create a plugin registry for testing. Register only a reserve plugin.
registry := framework.Registry{prebindPluginName: NewPrebindPlugin}
// Create the master and the scheduler with the test plugin set.
context := initTestSchedulerWithOptions(t,
initTestMaster(t, "prebind-plugin", nil),
false, nil, registry, false, time.Second)
defer cleanupTest(t, context)
cs := context.clientSet
// Add a few nodes.
_, err := createNodes(cs, "test-node", nil, 2)
if err != nil {
t.Fatalf("Cannot create nodes: %v", err)
}
tests := []struct {
fail bool
reject bool
}{
{
fail: false,
reject: false,
},
{
fail: true,
reject: false,
},
{
fail: false,
reject: true,
},
{
fail: true,
reject: true,
},
}
for i, test := range tests {
pbdPlugin.failPrebind = test.fail
pbdPlugin.rejectPrebind = test.reject
// Create a best effort pod.
pod, err := createPausePod(cs,
initPausePod(cs, &pausePodConfig{Name: "test-pod", Namespace: context.ns.Name}))
if err != nil {
t.Errorf("Error while creating a test pod: %v", err)
}
if test.fail {
if err = wait.Poll(10*time.Millisecond, 30*time.Second, podSchedulingError(cs, pod.Namespace, pod.Name)); err != nil {
t.Errorf("test #%v: Expected a scheduling error, but didn't get it. error: %v", i, err)
}
} else {
if test.reject {
if err = waitForPodUnschedulable(cs, pod); err != nil {
t.Errorf("test #%v: Didn't expected the pod to be scheduled. error: %v", i, err)
}
} else {
if err = waitForPodToSchedule(cs, pod); err != nil {
t.Errorf("test #%v: Expected the pod to be scheduled. error: %v", i, err)
}
}
}
if pbdPlugin.numPrebindCalled == 0 {
t.Errorf("Expected the prebind plugin to be called.")
}
cleanupPods(cs, t, []*v1.Pod{pod})
}
}

269
test/integration/scheduler/plugin_test.go
View File

@@ -1,269 +0,0 @@
/*
Copyright 2018 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 scheduler
import (
"fmt"
"testing"
"time"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/util/wait"
plugins "k8s.io/kubernetes/pkg/scheduler/plugins/v1alpha1"
)
// StatefulMultipointExample is an example plugin that is executed at multiple extension points.
// This plugin is stateful. It receives arguments at initialization (NewMultipointPlugin)
// and changes its state when it is executed.
type TesterPlugin struct {
numReserveCalled int
numPrebindCalled int
failReserve bool
failPrebind bool
rejectPrebind bool
}
var _ = plugins.ReservePlugin(&TesterPlugin{})
var _ = plugins.PrebindPlugin(&TesterPlugin{})
// Name returns name of the plugin.
func (tp *TesterPlugin) Name() string {
return "tester-plugin"
}
// Reserve is a test function that returns an error or nil, depending on the
// value of "failReserve".
func (tp *TesterPlugin) Reserve(ps plugins.PluginSet, pod *v1.Pod, nodeName string) error {
tp.numReserveCalled++
if tp.failReserve {
return fmt.Errorf("injecting failure for pod %v", pod.Name)
}
return nil
}
// Prebind is a test function that returns (true, nil) or errors for testing.
func (tp *TesterPlugin) Prebind(ps plugins.PluginSet, pod *v1.Pod, nodeName string) (bool, error) {
var err error
tp.numPrebindCalled++
if tp.failPrebind {
err = fmt.Errorf("injecting failure for pod %v", pod.Name)
}
if tp.rejectPrebind {
return false, err
}
return true, err
}
// TestPluginSet is a plugin set used for testing purposes.
type TestPluginSet struct {
data *plugins.PluginData
reservePlugins []plugins.ReservePlugin
prebindPlugins []plugins.PrebindPlugin
}
var _ = plugins.PluginSet(&TestPluginSet{})
// ReservePlugins returns a slice of default reserve plugins.
func (r *TestPluginSet) ReservePlugins() []plugins.ReservePlugin {
return r.reservePlugins
}
// PrebindPlugins returns a slice of default prebind plugins.
func (r *TestPluginSet) PrebindPlugins() []plugins.PrebindPlugin {
return r.prebindPlugins
}
// Data returns a pointer to PluginData.
func (r *TestPluginSet) Data() *plugins.PluginData {
return r.data
}
// TestReservePlugin tests invocation of reserve plugins.
func TestReservePlugin(t *testing.T) {
// Create a plugin set for testing. Register only a reserve plugin.
testerPlugin := &TesterPlugin{}
testPluginSet := &TestPluginSet{
data: &plugins.PluginData{
Ctx: plugins.NewPluginContext(),
},
reservePlugins: []plugins.ReservePlugin{testerPlugin},
}
// Create the master and the scheduler with the test plugin set.
context := initTestSchedulerWithOptions(t,
initTestMaster(t, "reserve-plugin", nil),
false, nil, testPluginSet, false, time.Second)
defer cleanupTest(t, context)
cs := context.clientSet
// Add a few nodes.
_, err := createNodes(cs, "test-node", nil, 2)
if err != nil {
t.Fatalf("Cannot create nodes: %v", err)
}
for _, fail := range []bool{false, true} {
testerPlugin.failReserve = fail
// Create a best effort pod.
pod, err := createPausePod(cs,
initPausePod(cs, &pausePodConfig{Name: "test-pod", Namespace: context.ns.Name}))
if err != nil {
t.Errorf("Error while creating a test pod: %v", err)
}
if fail {
if err = wait.Poll(10*time.Millisecond, 30*time.Second, podSchedulingError(cs, pod.Namespace, pod.Name)); err != nil {
t.Errorf("Didn't expected the pod to be scheduled. error: %v", err)
}
} else {
if err = waitForPodToSchedule(cs, pod); err != nil {
t.Errorf("Expected the pod to be scheduled. error: %v", err)
}
}
if testerPlugin.numReserveCalled == 0 {
t.Errorf("Expected the reserve plugin to be called.")
}
cleanupPods(cs, t, []*v1.Pod{pod})
}
}
// TestPrebindPlugin tests invocation of prebind plugins.
func TestPrebindPlugin(t *testing.T) {
// Create a plugin set for testing. Register only a prebind plugin.
testerPlugin := &TesterPlugin{}
testPluginSet := &TestPluginSet{
data: &plugins.PluginData{
Ctx: plugins.NewPluginContext(),
},
prebindPlugins: []plugins.PrebindPlugin{testerPlugin},
}
// Create the master and the scheduler with the test plugin set.
context := initTestSchedulerWithOptions(t,
initTestMaster(t, "prebind-plugin", nil),
false, nil, testPluginSet, false, time.Second)
defer cleanupTest(t, context)
cs := context.clientSet
// Add a few nodes.
_, err := createNodes(cs, "test-node", nil, 2)
if err != nil {
t.Fatalf("Cannot create nodes: %v", err)
}
tests := []struct {
fail bool
reject bool
}{
{
fail: false,
reject: false,
},
{
fail: true,
reject: false,
},
{
fail: false,
reject: true,
},
{
fail: true,
reject: true,
},
}
for i, test := range tests {
testerPlugin.failPrebind = test.fail
testerPlugin.rejectPrebind = test.reject
// Create a best effort pod.
pod, err := createPausePod(cs,
initPausePod(cs, &pausePodConfig{Name: "test-pod", Namespace: context.ns.Name}))
if err != nil {
t.Errorf("Error while creating a test pod: %v", err)
}
if test.fail {
if err = wait.Poll(10*time.Millisecond, 30*time.Second, podSchedulingError(cs, pod.Namespace, pod.Name)); err != nil {
t.Errorf("test #%v: Expected a scheduling error, but didn't get it. error: %v", i, err)
}
} else {
if test.reject {
if err = waitForPodUnschedulable(cs, pod); err != nil {
t.Errorf("test #%v: Didn't expected the pod to be scheduled. error: %v", i, err)
}
} else {
if err = waitForPodToSchedule(cs, pod); err != nil {
t.Errorf("test #%v: Expected the pod to be scheduled. error: %v", i, err)
}
}
}
if testerPlugin.numPrebindCalled == 0 {
t.Errorf("Expected the prebind plugin to be called.")
}
cleanupPods(cs, t, []*v1.Pod{pod})
}
}
// TestContextCleanup tests that data inserted in the pluginContext is removed
// after a scheduling cycle is over.
func TestContextCleanup(t *testing.T) {
// Create a plugin set for testing.
testerPlugin := &TesterPlugin{}
testPluginSet := &TestPluginSet{
data: &plugins.PluginData{
Ctx: plugins.NewPluginContext(),
},
reservePlugins: []plugins.ReservePlugin{testerPlugin},
prebindPlugins: []plugins.PrebindPlugin{testerPlugin},
}
// Create the master and the scheduler with the test plugin set.
context := initTestSchedulerWithOptions(t,
initTestMaster(t, "plugin-context-cleanup", nil),
false, nil, testPluginSet, false, time.Second)
defer cleanupTest(t, context)
cs := context.clientSet
// Add a few nodes.
_, err := createNodes(cs, "test-node", nil, 2)
if err != nil {
t.Fatalf("Cannot create nodes: %v", err)
}
// Insert something in the plugin context.
testPluginSet.Data().Ctx.Write("test", "foo")
// Create and schedule a best effort pod.
pod, err := runPausePod(cs,
initPausePod(cs, &pausePodConfig{Name: "test-pod", Namespace: context.ns.Name}))
if err != nil {
t.Errorf("Error while creating or scheduling a test pod: %v", err)
}
// Make sure the data inserted in the plugin context is removed.
_, err = testPluginSet.Data().Ctx.Read("test")
if err == nil || err.Error() != plugins.NotFound {
t.Errorf("Expected the plugin context to be cleaned up after a scheduling cycle. error: %v", err)
}
cleanupPods(cs, t, []*v1.Pod{pod})
}

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

@@ -43,6 +43,7 @@ import (
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
kubeschedulerconfig "k8s.io/kubernetes/pkg/scheduler/apis/config"
"k8s.io/kubernetes/pkg/scheduler/factory"
schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
schedulernodeinfo "k8s.io/kubernetes/pkg/scheduler/nodeinfo"
"k8s.io/kubernetes/test/integration/framework"
)
@@ -263,6 +264,7 @@ priorities: []
},
},
nil,
schedulerframework.NewRegistry(),
scheduler.WithName(v1.DefaultSchedulerName),
scheduler.WithHardPodAffinitySymmetricWeight(v1.DefaultHardPodAffinitySymmetricWeight),
scheduler.WithBindTimeoutSeconds(defaultBindTimeout),
@@ -331,6 +333,7 @@ func TestSchedulerCreationFromNonExistentConfigMap(t *testing.T) {
},
},
nil,
schedulerframework.NewRegistry(),
scheduler.WithName(v1.DefaultSchedulerName),
scheduler.WithHardPodAffinitySymmetricWeight(v1.DefaultHardPodAffinitySymmetricWeight),
scheduler.WithBindTimeoutSeconds(defaultBindTimeout))
@@ -595,7 +598,7 @@ func TestMultiScheduler(t *testing.T) {
stopCh := make(chan struct{})
defer close(stopCh)
schedulerConfigFactory2 := createConfiguratorWithPodInformer(fooScheduler, clientSet2, podInformer2, informerFactory2, stopCh)
schedulerConfigFactory2 := createConfiguratorWithPodInformer(fooScheduler, clientSet2, podInformer2, informerFactory2, schedulerframework.NewRegistry(), stopCh)
schedulerConfig2, err := schedulerConfigFactory2.Create()
if err != nil {
t.Errorf("Couldn't create scheduler config: %v", err)

18
test/integration/scheduler/util.go
View File

@@ -49,7 +49,7 @@ import (
_ "k8s.io/kubernetes/pkg/scheduler/algorithmprovider"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
"k8s.io/kubernetes/pkg/scheduler/factory"
plugins "k8s.io/kubernetes/pkg/scheduler/plugins/v1alpha1"
schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
taintutils "k8s.io/kubernetes/pkg/util/taints"
"k8s.io/kubernetes/test/integration/framework"
imageutils "k8s.io/kubernetes/test/utils/image"
@@ -73,6 +73,7 @@ func createConfiguratorWithPodInformer(
clientSet clientset.Interface,
podInformer coreinformers.PodInformer,
informerFactory informers.SharedInformerFactory,
pluginRegistry schedulerframework.Registry,
stopCh <-chan struct{},
) factory.Configurator {
return factory.NewConfigFactory(&factory.ConfigFactoryArgs{
@@ -88,6 +89,7 @@ func createConfiguratorWithPodInformer(
ServiceInformer: informerFactory.Core().V1().Services(),
PdbInformer: informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
StorageClassInformer: informerFactory.Storage().V1().StorageClasses(),
Registry: pluginRegistry,
HardPodAffinitySymmetricWeight: v1.DefaultHardPodAffinitySymmetricWeight,
DisablePreemption: false,
PercentageOfNodesToScore: schedulerapi.DefaultPercentageOfNodesToScore,
@@ -146,7 +148,7 @@ func initTestScheduler(
) *testContext {
// Pod preemption is enabled by default scheduler configuration, but preemption only happens when PodPriority
// feature gate is enabled at the same time.
return initTestSchedulerWithOptions(t, context, setPodInformer, policy, nil, false, time.Second)
return initTestSchedulerWithOptions(t, context, setPodInformer, policy, schedulerframework.NewRegistry(), false, time.Second)
}
// initTestSchedulerWithOptions initializes a test environment and creates a scheduler with default
@@ -156,7 +158,7 @@ func initTestSchedulerWithOptions(
context *testContext,
setPodInformer bool,
policy *schedulerapi.Policy,
pluginSet plugins.PluginSet,
pluginRegistry schedulerframework.Registry,
disablePreemption bool,
resyncPeriod time.Duration,
) *testContext {
@@ -173,7 +175,7 @@ func initTestSchedulerWithOptions(
}
context.schedulerConfigFactory = createConfiguratorWithPodInformer(
v1.DefaultSchedulerName, context.clientSet, podInformer, context.informerFactory, context.stopCh)
v1.DefaultSchedulerName, context.clientSet, podInformer, context.informerFactory, pluginRegistry, context.stopCh)
var err error
@@ -208,11 +210,6 @@ func initTestSchedulerWithOptions(
controller.WaitForCacheSync("scheduler", context.schedulerConfig.StopEverything, podInformer.Informer().HasSynced)
}
// Set pluginSet if provided. DefaultPluginSet is used if this is not specified.
if pluginSet != nil {
context.schedulerConfig.PluginSet = pluginSet
}
eventBroadcaster := record.NewBroadcaster()
context.schedulerConfig.Recorder = eventBroadcaster.NewRecorder(
legacyscheme.Scheme,
@@ -259,7 +256,8 @@ func initTest(t *testing.T, nsPrefix string) *testContext {
// configuration but with pod preemption disabled.
func initTestDisablePreemption(t *testing.T, nsPrefix string) *testContext {
return initTestSchedulerWithOptions(
t, initTestMaster(t, nsPrefix, nil), true, nil, nil, true, time.Second)
t, initTestMaster(t, nsPrefix, nil), true, nil,
schedulerframework.NewRegistry(), true, time.Second)
}
// cleanupTest deletes the scheduler and the test namespace. It should be called