Commit a98d14d2 authored by jayunit100's avatar jayunit100

[scheduler] interface for configuration factory, configurator.

parent c14fa94a
......@@ -19,6 +19,8 @@ go_library(
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",
"//pkg/client/cache:go_default_library",
"//pkg/client/clientset_generated/clientset:go_default_library",
"//pkg/client/record:go_default_library",
"//pkg/client/restclient:go_default_library",
"//pkg/util:go_default_library",
......@@ -29,11 +31,13 @@ go_library(
"//plugin/pkg/scheduler/api:go_default_library",
"//plugin/pkg/scheduler/metrics:go_default_library",
"//plugin/pkg/scheduler/schedulercache:go_default_library",
"//plugin/pkg/scheduler/util:go_default_library",
"//vendor:github.com/golang/glog",
"//vendor:github.com/golang/groupcache/lru",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
"//vendor:k8s.io/apimachinery/pkg/util/errors",
"//vendor:k8s.io/apimachinery/pkg/util/net",
"//vendor:k8s.io/apimachinery/pkg/util/sets",
"//vendor:k8s.io/apimachinery/pkg/util/wait",
],
)
......@@ -87,6 +91,7 @@ filegroup(
"//plugin/pkg/scheduler/metrics:all-srcs",
"//plugin/pkg/scheduler/schedulercache:all-srcs",
"//plugin/pkg/scheduler/testing:all-srcs",
"//plugin/pkg/scheduler/util:all-srcs",
],
tags = ["automanaged"],
)
......@@ -29,6 +29,7 @@ go_library(
"//plugin/pkg/scheduler/api:go_default_library",
"//plugin/pkg/scheduler/api/validation:go_default_library",
"//plugin/pkg/scheduler/schedulercache:go_default_library",
"//plugin/pkg/scheduler/util:go_default_library",
"//vendor:github.com/golang/glog",
"//vendor:k8s.io/apimachinery/pkg/api/errors",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
......@@ -61,9 +62,9 @@ go_test(
"//plugin/pkg/scheduler/api:go_default_library",
"//plugin/pkg/scheduler/api/latest:go_default_library",
"//plugin/pkg/scheduler/schedulercache:go_default_library",
"//plugin/pkg/scheduler/util:go_default_library",
"//vendor:k8s.io/apimachinery/pkg/apis/meta/v1",
"//vendor:k8s.io/apimachinery/pkg/runtime",
"//vendor:k8s.io/apimachinery/pkg/types",
],
)
......
......@@ -25,7 +25,6 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/kubernetes/pkg/api"
"k8s.io/kubernetes/pkg/api/testapi"
apitesting "k8s.io/kubernetes/pkg/api/testing"
......@@ -38,6 +37,7 @@ import (
schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
latestschedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api/latest"
"k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache"
"k8s.io/kubernetes/plugin/pkg/scheduler/util"
)
func TestCreate(t *testing.T) {
......@@ -152,13 +152,8 @@ func TestDefaultErrorFunc(t *testing.T) {
defer server.Close()
factory := NewConfigFactory(clientset.NewForConfigOrDie(&restclient.Config{Host: server.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &api.Registry.GroupOrDie(v1.GroupName).GroupVersion}}), v1.DefaultSchedulerName, v1.DefaultHardPodAffinitySymmetricWeight, v1.DefaultFailureDomains)
queue := cache.NewFIFO(cache.MetaNamespaceKeyFunc)
podBackoff := podBackoff{
perPodBackoff: map[types.NamespacedName]*backoffEntry{},
clock: &fakeClock{},
defaultDuration: 1 * time.Millisecond,
maxDuration: 1 * time.Second,
}
errFunc := factory.makeDefaultErrorFunc(&podBackoff, queue)
podBackoff := util.CreatePodBackoff(1*time.Millisecond, 1*time.Second)
errFunc := factory.MakeDefaultErrorFunc(podBackoff, queue)
errFunc(testPod, nil)
for {
......@@ -202,14 +197,6 @@ func TestNodeEnumerator(t *testing.T) {
}
}
type fakeClock struct {
t time.Time
}
func (f *fakeClock) Now() time.Time {
return f.t
}
func TestBind(t *testing.T) {
table := []struct {
binding *v1.Binding
......@@ -245,66 +232,6 @@ func TestBind(t *testing.T) {
}
}
func TestBackoff(t *testing.T) {
clock := fakeClock{}
backoff := podBackoff{
perPodBackoff: map[types.NamespacedName]*backoffEntry{},
clock: &clock,
defaultDuration: 1 * time.Second,
maxDuration: 60 * time.Second,
}
tests := []struct {
podID types.NamespacedName
expectedDuration time.Duration
advanceClock time.Duration
}{
{
podID: types.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 1 * time.Second,
},
{
podID: types.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 2 * time.Second,
},
{
podID: types.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 4 * time.Second,
},
{
podID: types.NamespacedName{Namespace: "default", Name: "bar"},
expectedDuration: 1 * time.Second,
advanceClock: 120 * time.Second,
},
// 'foo' should have been gc'd here.
{
podID: types.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 1 * time.Second,
},
}
for _, test := range tests {
duration := backoff.getEntry(test.podID).getBackoff(backoff.maxDuration)
if duration != test.expectedDuration {
t.Errorf("expected: %s, got %s for %s", test.expectedDuration.String(), duration.String(), test.podID)
}
clock.t = clock.t.Add(test.advanceClock)
backoff.gc()
}
fooID := types.NamespacedName{Namespace: "default", Name: "foo"}
backoff.perPodBackoff[fooID].backoff = 60 * time.Second
duration := backoff.getEntry(fooID).getBackoff(backoff.maxDuration)
if duration != 60*time.Second {
t.Errorf("expected: 60, got %s", duration.String())
}
// Verify that we split on namespaces correctly, same name, different namespace
fooID.Namespace = "other"
duration = backoff.getEntry(fooID).getBackoff(backoff.maxDuration)
if duration != 1*time.Second {
t.Errorf("expected: 1, got %s", duration.String())
}
}
// TestResponsibleForPod tests if a pod with an annotation that should cause it to
// be picked up by the default scheduler, is in fact picked by the default scheduler
// Two schedulers are made in the test: one is default scheduler and other scheduler
......@@ -363,8 +290,8 @@ func TestResponsibleForPod(t *testing.T) {
}
for _, test := range tests {
podOfDefault := factoryDefaultScheduler.responsibleForPod(test.pod)
podOfFoo := factoryFooScheduler.responsibleForPod(test.pod)
podOfDefault := factoryDefaultScheduler.ResponsibleForPod(test.pod)
podOfFoo := factoryFooScheduler.ResponsibleForPod(test.pod)
results := []bool{podOfDefault, podOfFoo}
expected := []bool{test.pickedByDefault, test.pickedByFoo}
if !reflect.DeepEqual(results, expected) {
......
......@@ -24,10 +24,16 @@ import (
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/pkg/client/record"
"k8s.io/kubernetes/plugin/pkg/scheduler/algorithm"
schedulerapi "k8s.io/kubernetes/plugin/pkg/scheduler/api"
"k8s.io/kubernetes/plugin/pkg/scheduler/metrics"
"k8s.io/kubernetes/plugin/pkg/scheduler/schedulercache"
"k8s.io/kubernetes/plugin/pkg/scheduler/util"
clientset "k8s.io/kubernetes/pkg/client/clientset_generated/clientset"
"github.com/golang/glog"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/kubernetes/pkg/client/cache"
)
// Binder knows how to write a binding.
......@@ -45,6 +51,33 @@ type Scheduler struct {
config *Config
}
// These are the functions which need to be provided in order to build a Scheduler configuration.
// An implementation of this can be seen in factory.go.
type Configurator interface {
GetPriorityFunctionConfigs(priorityKeys sets.String) ([]algorithm.PriorityConfig, error)
GetPriorityMetadataProducer() (algorithm.MetadataProducer, error)
GetPredicateMetadataProducer() (algorithm.MetadataProducer, error)
GetPredicates(predicateKeys sets.String) (map[string]algorithm.FitPredicate, error)
GetHardPodAffinitySymmetricWeight() int
GetFailureDomains() []string
GetSchedulerName() string
MakeDefaultErrorFunc(backoff *util.PodBackoff, podQueue *cache.FIFO) func(pod *v1.Pod, err error)
// Probably doesn't need to be public. But exposed for now in case.
ResponsibleForPod(pod *v1.Pod) bool
// Needs to be exposed for things like integration tests where we want to make fake nodes.
GetNodeStore() cache.Store
GetClient() clientset.Interface
GetScheduledPodListerIndexer() cache.Indexer
Run()
Create() (*Config, error)
CreateFromProvider(providerName string) (*Config, error)
CreateFromConfig(policy schedulerapi.Policy) (*Config, error)
CreateFromKeys(predicateKeys, priorityKeys sets.String, extenders []algorithm.SchedulerExtender) (*Config, error)
}
type Config struct {
// It is expected that changes made via SchedulerCache will be observed
// by NodeLister and Algorithm.
......
package(default_visibility = ["//visibility:public"])
licenses(["notice"])
load(
"@io_bazel_rules_go//go:def.bzl",
"go_library",
"go_test",
)
go_test(
name = "go_default_test",
srcs = ["backoff_utils_test.go"],
library = ":go_default_library",
tags = ["automanaged"],
deps = ["//vendor:k8s.io/apimachinery/pkg/types"],
)
go_library(
name = "go_default_library",
srcs = ["backoff_utils.go"],
tags = ["automanaged"],
deps = [
"//vendor:github.com/golang/glog",
"//vendor:k8s.io/apimachinery/pkg/types",
],
)
filegroup(
name = "package-srcs",
srcs = glob(["**"]),
tags = ["automanaged"],
visibility = ["//visibility:private"],
)
filegroup(
name = "all-srcs",
srcs = [":package-srcs"],
tags = ["automanaged"],
)
/*
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 util
import (
"fmt"
"github.com/golang/glog"
ktypes "k8s.io/apimachinery/pkg/types"
"sync"
"sync/atomic"
"time"
)
type clock interface {
Now() time.Time
}
type realClock struct{}
func (realClock) Now() time.Time {
return time.Now()
}
// backoffEntry is single threaded. in particular, it only allows a single action to be waiting on backoff at a time.
// It is expected that all users will only use the public TryWait(...) method
// It is also not safe to copy this object.
type backoffEntry struct {
backoff time.Duration
lastUpdate time.Time
reqInFlight int32
}
// tryLock attempts to acquire a lock via atomic compare and swap.
// returns true if the lock was acquired, false otherwise
func (b *backoffEntry) tryLock() bool {
return atomic.CompareAndSwapInt32(&b.reqInFlight, 0, 1)
}
// unlock returns the lock. panics if the lock isn't held
func (b *backoffEntry) unlock() {
if !atomic.CompareAndSwapInt32(&b.reqInFlight, 1, 0) {
panic(fmt.Sprintf("unexpected state on unlocking: %+v", b))
}
}
// TryWait tries to acquire the backoff lock, maxDuration is the maximum allowed period to wait for.
func (b *backoffEntry) TryWait(maxDuration time.Duration) bool {
if !b.tryLock() {
return false
}
defer b.unlock()
b.wait(maxDuration)
return true
}
func (entry *backoffEntry) getBackoff(maxDuration time.Duration) time.Duration {
duration := entry.backoff
newDuration := time.Duration(duration) * 2
if newDuration > maxDuration {
newDuration = maxDuration
}
entry.backoff = newDuration
glog.V(4).Infof("Backing off %s for pod %+v", duration.String(), entry)
return duration
}
func (entry *backoffEntry) wait(maxDuration time.Duration) {
time.Sleep(entry.getBackoff(maxDuration))
}
type PodBackoff struct {
perPodBackoff map[ktypes.NamespacedName]*backoffEntry
lock sync.Mutex
clock clock
defaultDuration time.Duration
maxDuration time.Duration
}
func (p *PodBackoff) MaxDuration() time.Duration {
return p.maxDuration
}
func CreateDefaultPodBackoff() *PodBackoff {
return CreatePodBackoff(1*time.Second, 60*time.Second)
}
func CreatePodBackoff(defaultDuration, maxDuration time.Duration) *PodBackoff {
return CreatePodBackoffWithClock(defaultDuration, maxDuration, realClock{})
}
func CreatePodBackoffWithClock(defaultDuration, maxDuration time.Duration, clock clock) *PodBackoff {
return &PodBackoff{
perPodBackoff: map[ktypes.NamespacedName]*backoffEntry{},
clock: clock,
defaultDuration: defaultDuration,
maxDuration: maxDuration,
}
}
func (p *PodBackoff) GetEntry(podID ktypes.NamespacedName) *backoffEntry {
p.lock.Lock()
defer p.lock.Unlock()
entry, ok := p.perPodBackoff[podID]
if !ok {
entry = &backoffEntry{backoff: p.defaultDuration}
p.perPodBackoff[podID] = entry
}
entry.lastUpdate = p.clock.Now()
return entry
}
func (p *PodBackoff) Gc() {
p.lock.Lock()
defer p.lock.Unlock()
now := p.clock.Now()
for podID, entry := range p.perPodBackoff {
if now.Sub(entry.lastUpdate) > p.maxDuration {
delete(p.perPodBackoff, podID)
}
}
}
/*
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 util
import (
ktypes "k8s.io/apimachinery/pkg/types"
"testing"
"time"
)
type fakeClock struct {
t time.Time
}
func (f *fakeClock) Now() time.Time {
return f.t
}
func TestBackoff(t *testing.T) {
clock := fakeClock{}
backoff := CreatePodBackoffWithClock(1*time.Second, 60*time.Second, &clock)
tests := []struct {
podID ktypes.NamespacedName
expectedDuration time.Duration
advanceClock time.Duration
}{
{
podID: ktypes.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 1 * time.Second,
},
{
podID: ktypes.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 2 * time.Second,
},
{
podID: ktypes.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 4 * time.Second,
},
{
podID: ktypes.NamespacedName{Namespace: "default", Name: "bar"},
expectedDuration: 1 * time.Second,
advanceClock: 120 * time.Second,
},
// 'foo' should have been gc'd here.
{
podID: ktypes.NamespacedName{Namespace: "default", Name: "foo"},
expectedDuration: 1 * time.Second,
},
}
for _, test := range tests {
duration := backoff.GetEntry(test.podID).getBackoff(backoff.maxDuration)
if duration != test.expectedDuration {
t.Errorf("expected: %s, got %s for %s", test.expectedDuration.String(), duration.String(), test.podID)
}
clock.t = clock.t.Add(test.advanceClock)
backoff.Gc()
}
fooID := ktypes.NamespacedName{Namespace: "default", Name: "foo"}
backoff.perPodBackoff[fooID].backoff = 60 * time.Second
duration := backoff.GetEntry(fooID).getBackoff(backoff.maxDuration)
if duration != 60*time.Second {
t.Errorf("expected: 60, got %s", duration.String())
}
// Verify that we split on namespaces correctly, same name, different namespace
fooID.Namespace = "other"
duration = backoff.GetEntry(fooID).getBackoff(backoff.maxDuration)
if duration != 1*time.Second {
t.Errorf("expected: 1, got %s", duration.String())
}
}
......@@ -71,7 +71,7 @@ func TestUnschedulableNodes(t *testing.T) {
defer close(schedulerConfig.StopEverything)
DoTestUnschedulableNodes(t, clientSet, ns, schedulerConfigFactory.NodeLister.Store)
DoTestUnschedulableNodes(t, clientSet, ns, schedulerConfigFactory.GetNodeStore())
}
func podScheduled(c clientset.Interface, podNamespace, podName string) wait.ConditionFunc {
......
......@@ -37,7 +37,7 @@ go_test(
tags = ["automanaged"],
deps = [
"//pkg/api/v1:go_default_library",
"//plugin/pkg/scheduler/factory:go_default_library",
"//plugin/pkg/scheduler:go_default_library",
"//test/integration/framework:go_default_library",
"//test/utils:go_default_library",
"//vendor:github.com/golang/glog",
......
......@@ -56,7 +56,7 @@ func BenchmarkScheduling1000Nodes1000Pods(b *testing.B) {
func benchmarkScheduling(numNodes, numScheduledPods int, b *testing.B) {
schedulerConfigFactory, finalFunc := mustSetupScheduler()
defer finalFunc()
c := schedulerConfigFactory.Client
c := schedulerConfigFactory.GetClient()
nodePreparer := framework.NewIntegrationTestNodePreparer(
c,
......@@ -74,7 +74,7 @@ func benchmarkScheduling(numNodes, numScheduledPods int, b *testing.B) {
podCreator.CreatePods()
for {
scheduled := schedulerConfigFactory.ScheduledPodLister.Indexer.List()
scheduled := schedulerConfigFactory.GetScheduledPodListerIndexer().List()
if len(scheduled) >= numScheduledPods {
break
}
......@@ -89,7 +89,7 @@ func benchmarkScheduling(numNodes, numScheduledPods int, b *testing.B) {
for {
// This can potentially affect performance of scheduler, since List() is done under mutex.
// TODO: Setup watch on apiserver and wait until all pods scheduled.
scheduled := schedulerConfigFactory.ScheduledPodLister.Indexer.List()
scheduled := schedulerConfigFactory.GetScheduledPodListerIndexer().List()
if len(scheduled) >= numScheduledPods+b.N {
break
}
......
......@@ -24,12 +24,12 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/kubernetes/pkg/api/v1"
"k8s.io/kubernetes/plugin/pkg/scheduler/factory"
"k8s.io/kubernetes/test/integration/framework"
testutils "k8s.io/kubernetes/test/utils"
"github.com/golang/glog"
"github.com/renstrom/dedent"
"k8s.io/kubernetes/plugin/pkg/scheduler"
)
const (
......@@ -74,7 +74,7 @@ func TestSchedule100Node3KNodeAffinityPods(t *testing.T) {
})
}
config.nodePreparer = framework.NewIntegrationTestNodePreparer(
config.schedulerConfigFactory.Client,
config.schedulerSupportFunctions.GetClient(),
nodeStrategies,
"scheduler-perf-",
)
......@@ -106,7 +106,7 @@ func TestSchedule100Node3KNodeAffinityPods(t *testing.T) {
}),
)
}
config.podCreator = testutils.NewTestPodCreator(config.schedulerConfigFactory.Client, podCreatorConfig)
config.podCreator = testutils.NewTestPodCreator(config.schedulerSupportFunctions.GetClient(), podCreatorConfig)
if min := schedulePods(config); min < threshold30K {
t.Errorf("Too small pod scheduling throughput for 30k pods. Expected %v got %v", threshold30K, min)
......@@ -144,19 +144,19 @@ func TestSchedule1000Node30KPods(t *testing.T) {
// }
type testConfig struct {
numPods int
numNodes int
nodePreparer testutils.TestNodePreparer
podCreator *testutils.TestPodCreator
schedulerConfigFactory *factory.ConfigFactory
destroyFunc func()
numPods int
numNodes int
nodePreparer testutils.TestNodePreparer
podCreator *testutils.TestPodCreator
schedulerSupportFunctions scheduler.Configurator
destroyFunc func()
}
func baseConfig() *testConfig {
schedulerConfigFactory, destroyFunc := mustSetupScheduler()
return &testConfig{
schedulerConfigFactory: schedulerConfigFactory,
destroyFunc: destroyFunc,
schedulerSupportFunctions: schedulerConfigFactory,
destroyFunc: destroyFunc,
}
}
......@@ -164,14 +164,14 @@ func defaultSchedulerBenchmarkConfig(numNodes, numPods int) *testConfig {
baseConfig := baseConfig()
nodePreparer := framework.NewIntegrationTestNodePreparer(
baseConfig.schedulerConfigFactory.Client,
baseConfig.schedulerSupportFunctions.GetClient(),
[]testutils.CountToStrategy{{Count: numNodes, Strategy: &testutils.TrivialNodePrepareStrategy{}}},
"scheduler-perf-",
)
config := testutils.NewTestPodCreatorConfig()
config.AddStrategy("sched-test", numPods, testutils.NewSimpleWithControllerCreatePodStrategy("rc1"))
podCreator := testutils.NewTestPodCreator(baseConfig.schedulerConfigFactory.Client, config)
podCreator := testutils.NewTestPodCreator(baseConfig.schedulerSupportFunctions.GetClient(), config)
baseConfig.nodePreparer = nodePreparer
baseConfig.podCreator = podCreator
......@@ -203,7 +203,7 @@ func schedulePods(config *testConfig) int32 {
// Bake in time for the first pod scheduling event.
for {
time.Sleep(50 * time.Millisecond)
scheduled := config.schedulerConfigFactory.ScheduledPodLister.Indexer.List()
scheduled := config.schedulerSupportFunctions.GetScheduledPodListerIndexer().List()
// 30,000 pods -> wait till @ least 300 are scheduled to start measuring.
// TODO Find out why sometimes there may be scheduling blips in the beggining.
if len(scheduled) > config.numPods/100 {
......@@ -218,7 +218,7 @@ func schedulePods(config *testConfig) int32 {
// This can potentially affect performance of scheduler, since List() is done under mutex.
// Listing 10000 pods is an expensive operation, so running it frequently may impact scheduler.
// TODO: Setup watch on apiserver and wait until all pods scheduled.
scheduled := config.schedulerConfigFactory.ScheduledPodLister.Indexer.List()
scheduled := config.schedulerSupportFunctions.GetScheduledPodListerIndexer().List()
// We will be completed when all pods are done being scheduled.
// return the worst-case-scenario interval that was seen during this time.
......
......@@ -40,7 +40,7 @@ import (
// remove resources after finished.
// Notes on rate limiter:
// - client rate limit is set to 5000.
func mustSetupScheduler() (schedulerConfigFactory *factory.ConfigFactory, destroyFunc func()) {
func mustSetupScheduler() (schedulerConfigFactory scheduler.Configurator, destroyFunc func()) {
h := &framework.MasterHolder{Initialized: make(chan struct{})}
s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment