github
/
k3s
mirror of https://github.com/k3s-io/k3s
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Removed old scheduler constructor.

pull/6/head
Klaus Ma 2017-05-08 15:10:25 +08:00
parent 1780a527f6
commit 63d194b626
6 changed files with 200 additions and 76 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
plugin/pkg/scheduler/BUILD
View File

@ -35,7 +35,10 @@ go_test(
go_library(
name = "go_default_library",
srcs = ["scheduler.go"],
srcs = [
"scheduler.go",
"testutil.go",
],
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",

16
plugin/pkg/scheduler/scheduler.go
View File

@ -20,10 +20,12 @@ import (
"time"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/record"
"k8s.io/kubernetes/pkg/api/v1"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
corelisters "k8s.io/kubernetes/pkg/client/listers/core/v1"
"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm"
schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
@ -33,8 +35,6 @@ import (
"k8s.io/kubernetes/plugin/pkg/scheduler/util"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/client-go/tools/cache"
)
// Binder knows how to write a binding.
@ -121,16 +121,6 @@ type Config struct {
StopEverything chan struct{}
}
// New returns a new scheduler.
// TODO replace this with NewFromConfigurator.
func New(c *Config) *Scheduler {
s := &Scheduler{
config: c,
}
metrics.Register()
return s
}
// NewFromConfigurator returns a new scheduler that is created entirely by the Configurator. Assumes Create() is implemented.
// Supports intermediate Config mutation for now if you provide modifier functions which will run after Config is created.
func NewFromConfigurator(c Configurator, modifiers ...func(c *Config)) (*Scheduler, error) {

125
plugin/pkg/scheduler/scheduler_test.go
View File

@ -154,31 +154,34 @@ func TestScheduler(t *testing.T) {
var gotPod *v1.Pod
var gotAssumedPod *v1.Pod
var gotBinding *v1.Binding
c := &Config{
SchedulerCache: &schedulertesting.FakeCache{
AssumeFunc: func(pod *v1.Pod) {
gotAssumedPod = pod
configurator := &FakeConfigurator{
Config: &Config{
SchedulerCache: &schedulertesting.FakeCache{
AssumeFunc: func(pod *v1.Pod) {
gotAssumedPod = pod
},
},
NodeLister: schedulertesting.FakeNodeLister(
[]*v1.Node{&testNode},
),
Algorithm: item.algo,
Binder: fakeBinder{func(b *v1.Binding) error {
gotBinding = b
return item.injectBindError
}},
PodConditionUpdater: fakePodConditionUpdater{},
Error: func(p *v1.Pod, err error) {
gotPod = p
gotError = err
},
NextPod: func() *v1.Pod {
return item.sendPod
},
Recorder: eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "scheduler"}),
},
NodeLister: schedulertesting.FakeNodeLister(
[]*v1.Node{&testNode},
),
Algorithm: item.algo,
Binder: fakeBinder{func(b *v1.Binding) error {
gotBinding = b
return item.injectBindError
}},
PodConditionUpdater: fakePodConditionUpdater{},
Error: func(p *v1.Pod, err error) {
gotPod = p
gotError = err
},
NextPod: func() *v1.Pod {
return item.sendPod
},
Recorder: eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: "scheduler"}),
}
s := New(c)
s, _ := NewFromConfigurator(configurator, nil...)
called := make(chan struct{})
events := eventBroadcaster.StartEventWatcher(func(e *clientv1.Event) {
if e, a := item.eventReason, e.Reason; e != a {
@ -495,24 +498,29 @@ func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache schedulercache.
[]algorithm.SchedulerExtender{})
bindingChan := make(chan *v1.Binding, 1)
errChan := make(chan error, 1)
cfg := &Config{
SchedulerCache: scache,
NodeLister: nodeLister,
Algorithm: algo,
Binder: fakeBinder{func(b *v1.Binding) error {
bindingChan <- b
return nil
}},
NextPod: func() *v1.Pod {
return clientcache.Pop(queuedPodStore).(*v1.Pod)
configurator := &FakeConfigurator{
Config: &Config{
SchedulerCache: scache,
NodeLister: nodeLister,
Algorithm: algo,
Binder: fakeBinder{func(b *v1.Binding) error {
bindingChan <- b
return nil
}},
NextPod: func() *v1.Pod {
return clientcache.Pop(queuedPodStore).(*v1.Pod)
},
Error: func(p *v1.Pod, err error) {
errChan <- err
},
Recorder: &record.FakeRecorder{},
PodConditionUpdater: fakePodConditionUpdater{},
},
Error: func(p *v1.Pod, err error) {
errChan <- err
},
Recorder: &record.FakeRecorder{},
PodConditionUpdater: fakePodConditionUpdater{},
}
return New(cfg), bindingChan, errChan
sched, _ := NewFromConfigurator(configurator, nil...)
return sched, bindingChan, errChan
}
func setupTestSchedulerLongBindingWithRetry(queuedPodStore *clientcache.FIFO, scache schedulercache.Cache, nodeLister schedulertesting.FakeNodeLister, predicateMap map[string]algorithm.FitPredicate, stop chan struct{}, bindingTime time.Duration) (*Scheduler, chan *v1.Binding) {
@ -525,24 +533,29 @@ func setupTestSchedulerLongBindingWithRetry(queuedPodStore *clientcache.FIFO, sc
algorithm.EmptyMetadataProducer,
[]algorithm.SchedulerExtender{})
bindingChan := make(chan *v1.Binding, 2)
cfg := &Config{
SchedulerCache: scache,
NodeLister: nodeLister,
Algorithm: algo,
Binder: fakeBinder{func(b *v1.Binding) error {
time.Sleep(bindingTime)
bindingChan <- b
return nil
}},
NextPod: func() *v1.Pod {
return clientcache.Pop(queuedPodStore).(*v1.Pod)
configurator := &FakeConfigurator{
Config: &Config{
SchedulerCache: scache,
NodeLister: nodeLister,
Algorithm: algo,
Binder: fakeBinder{func(b *v1.Binding) error {
time.Sleep(bindingTime)
bindingChan <- b
return nil
}},
NextPod: func() *v1.Pod {
return clientcache.Pop(queuedPodStore).(*v1.Pod)
},
Error: func(p *v1.Pod, err error) {
queuedPodStore.AddIfNotPresent(p)
},
Recorder: &record.FakeRecorder{},
PodConditionUpdater: fakePodConditionUpdater{},
StopEverything: stop,
},
Error: func(p *v1.Pod, err error) {
queuedPodStore.AddIfNotPresent(p)
},
Recorder: &record.FakeRecorder{},
PodConditionUpdater: fakePodConditionUpdater{},
StopEverything: stop,
}
return New(cfg), bindingChan
sched, _ := NewFromConfigurator(configurator, nil...)
return sched, bindingChan
}

115
plugin/pkg/scheduler/testutil.go Normal file
View File

@ -0,0 +1,115 @@
/*
Copyright 2017 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/util/sets"
"k8s.io/client-go/tools/cache"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
corelisters "k8s.io/kubernetes/pkg/client/listers/core/v1"
"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm"
schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
"k8s.io/kubernetes/plugin/pkg/scheduler/util"
)
// FakeConfigurator is an implementation for test.
type FakeConfigurator struct {
Config *Config
}
// GetPriorityFunctionConfigs is not implemented yet.
func (fc *FakeConfigurator) GetPriorityFunctionConfigs(priorityKeys sets.String) ([]algorithm.PriorityConfig, error) {
return nil, fmt.Errorf("not implemented")
}
// GetPriorityMetadataProducer is not implemented yet.
func (fc *FakeConfigurator) GetPriorityMetadataProducer() (algorithm.MetadataProducer, error) {
return nil, fmt.Errorf("not implemented")
}
// GetPredicateMetadataProducer is not implemented yet.
func (fc *FakeConfigurator) GetPredicateMetadataProducer() (algorithm.MetadataProducer, error) {
return nil, fmt.Errorf("not implemented")
}
// GetPredicates is not implemented yet.
func (fc *FakeConfigurator) GetPredicates(predicateKeys sets.String) (map[string]algorithm.FitPredicate, error) {
return nil, fmt.Errorf("not implemented")
}
// GetHardPodAffinitySymmetricWeight is not implemented yet.
func (fc *FakeConfigurator) GetHardPodAffinitySymmetricWeight() int {
panic("not implemented")
}
// GetSchedulerName is not implemented yet.
func (fc *FakeConfigurator) GetSchedulerName() string {
panic("not implemented")
}
// MakeDefaultErrorFunc is not implemented yet.
func (fc *FakeConfigurator) MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue *cache.FIFO) func(pod *v1.Pod, err error) {
return nil
}
// ResponsibleForPod is not implemented yet.
func (fc *FakeConfigurator) ResponsibleForPod(pod *v1.Pod) bool {
panic("not implemented")
}
// GetNodeLister is not implemented yet.
func (fc *FakeConfigurator) GetNodeLister() corelisters.NodeLister {
return nil
}
// GetClient is not implemented yet.
func (fc *FakeConfigurator) GetClient() clientset.Interface {
return nil
}
// GetScheduledPodLister is not implemented yet.
func (fc *FakeConfigurator) GetScheduledPodLister() corelisters.PodLister {
return nil
}
// Run is not implemented yet.
func (fc *FakeConfigurator) Run() {
panic("not implemented")
}
// Create returns FakeConfigurator.Config
func (fc *FakeConfigurator) Create() (*Config, error) {
return fc.Config, nil
}
// CreateFromProvider returns FakeConfigurator.Config
func (fc *FakeConfigurator) CreateFromProvider(providerName string) (*Config, error) {
return fc.Config, nil
}
// CreateFromConfig returns FakeConfigurator.Config
func (fc *FakeConfigurator) CreateFromConfig(policy schedulerapi.Policy) (*Config, error) {
return fc.Config, nil
}
// CreateFromKeys returns FakeConfigurator.Config
func (fc *FakeConfigurator) CreateFromKeys(predicateKeys, priorityKeys sets.String, extenders []algorithm.SchedulerExtender) (*Config, error) {
return fc.Config, nil
}

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

@ -341,7 +341,7 @@ func TestSchedulerExtender(t *testing.T) {
eventBroadcaster := record.NewBroadcaster()
schedulerConfig.Recorder = eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: v1.DefaultSchedulerName})
eventBroadcaster.StartRecordingToSink(&v1core.EventSinkImpl{Interface: v1core.New(clientSet.Core().RESTClient()).Events("")})
scheduler := scheduler.New(schedulerConfig)
scheduler, _ := scheduler.NewFromConfigurator(&scheduler.FakeConfigurator{Config: schedulerConfig}, nil...)
informerFactory.Start(schedulerConfig.StopEverything)
scheduler.Run()

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

@ -261,8 +261,8 @@ func TestUnschedulableNodes(t *testing.T) {
schedulerConfig.Recorder = eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: v1.DefaultSchedulerName})
eventBroadcaster.StartRecordingToSink(&clientv1core.EventSinkImpl{Interface: clientv1core.New(clientSet.Core().RESTClient()).Events("")})
informerFactory.Start(schedulerConfig.StopEverything)
scheduler.New(schedulerConfig).Run()
sched, _ := scheduler.NewFromConfigurator(&scheduler.FakeConfigurator{Config: schedulerConfig}, nil...)
sched.Run()
defer close(schedulerConfig.StopEverything)
DoTestUnschedulableNodes(t, clientSet, ns, schedulerConfigFactory.GetNodeLister())
@ -543,7 +543,8 @@ func TestMultiScheduler(t *testing.T) {
schedulerConfig.Recorder = eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: v1.DefaultSchedulerName})
eventBroadcaster.StartRecordingToSink(&clientv1core.EventSinkImpl{Interface: clientv1core.New(clientSet.Core().RESTClient()).Events("")})
informerFactory.Start(schedulerConfig.StopEverything)
scheduler.New(schedulerConfig).Run()
sched, _ := scheduler.NewFromConfigurator(&scheduler.FakeConfigurator{Config: schedulerConfig}, nil...)
sched.Run()
// default-scheduler will be stopped later
// 2. create a node
@ -627,8 +628,9 @@ func TestMultiScheduler(t *testing.T) {
schedulerConfig2.Recorder = eventBroadcaster2.NewRecorder(api.Scheme, clientv1.EventSource{Component: "foo-scheduler"})
eventBroadcaster.StartRecordingToSink(&clientv1core.EventSinkImpl{Interface: clientv1core.New(clientSet2.Core().RESTClient()).Events("")})
informerFactory2.Start(schedulerConfig2.StopEverything)
scheduler.New(schedulerConfig2).Run()
sched2, _ := scheduler.NewFromConfigurator(&scheduler.FakeConfigurator{Config: schedulerConfig2}, nil...)
sched2.Run()
defer close(schedulerConfig2.StopEverything)
// 6. **check point-2**:
@ -735,7 +737,8 @@ func TestAllocatable(t *testing.T) {
schedulerConfig.Recorder = eventBroadcaster.NewRecorder(api.Scheme, clientv1.EventSource{Component: v1.DefaultSchedulerName})
eventBroadcaster.StartRecordingToSink(&clientv1core.EventSinkImpl{Interface: clientv1core.New(clientSet.Core().RESTClient()).Events("")})
informerFactory.Start(schedulerConfig.StopEverything)
scheduler.New(schedulerConfig).Run()
sched, _ := scheduler.NewFromConfigurator(&scheduler.FakeConfigurator{Config: schedulerConfig}, nil...)
sched.Run()
// default-scheduler will be stopped later
defer close(schedulerConfig.StopEverything)