github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-29 14:37:00 +00:00
Code Issues Projects Releases Wiki Activity

Remove GetBinder member and replace it with a method.

Browse Source 
Signed-off-by: Aldo Culquicondor <acondor@google.com>
This commit is contained in:
Aldo Culquicondor 2020-01-16 16:55:34 -05:00
parent 7a39e1396a
commit eb265bc7db
5 changed files with 205 additions and 170 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
pkg/scheduler/BUILD
View File

@ -90,7 +90,6 @@ go_test(
"//staging/src/k8s.io/apimachinery/pkg/runtime:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
"//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
@ -102,6 +101,7 @@ go_test(
"//staging/src/k8s.io/client-go/testing:go_default_library",
"//staging/src/k8s.io/client-go/tools/cache:go_default_library",
"//staging/src/k8s.io/client-go/tools/events:go_default_library",
"//vendor/github.com/google/go-cmp/cmp:go_default_library",
],
)

27
pkg/scheduler/factory.go
View File

@ -157,7 +157,6 @@ func (c *Configurator) create(extenders []algorithm.SchedulerExtender) (*Schedul
return &Scheduler{
SchedulerCache: c.schedulerCache,
Algorithm: algo,
GetBinder: getBinderFunc(c.client, extenders),
Framework: framework,
NextPod: internalqueue.MakeNextPodFunc(podQueue),
Error: MakeDefaultErrorFunc(c.client, podQueue, c.schedulerCache),
@ -378,19 +377,6 @@ func getPredicateConfigs(keys sets.String, lr *frameworkplugins.LegacyRegistry,
return &plugins, pluginConfig, nil
}
// getBinderFunc returns a func which returns an extender that supports bind or a default binder based on the given pod.
func getBinderFunc(client clientset.Interface, extenders []algorithm.SchedulerExtender) func(pod *v1.Pod) Binder {
defaultBinder := &binder{client}
return func(pod *v1.Pod) Binder {
for _, extender := range extenders {
if extender.IsBinder() && extender.IsInterested(pod) {
return extender
}
}
return defaultBinder
}
}
type podInformer struct {
informer cache.SharedIndexInformer
}
@ -482,19 +468,6 @@ func MakeDefaultErrorFunc(client clientset.Interface, podQueue internalqueue.Sch
}
}
type binder struct {
Client clientset.Interface
}
// Implement Binder interface
var _ Binder = &binder{}
// Bind just does a POST binding RPC.
func (b *binder) Bind(binding *v1.Binding) error {
klog.V(3).Infof("Attempting to bind %v to %v", binding.Name, binding.Target.Name)
return b.Client.CoreV1().Pods(binding.Namespace).Bind(binding)
}
// GetPodDisruptionBudgetLister returns pdb lister from the given informer factory. Returns nil if PodDisruptionBudget feature is disabled.
func GetPodDisruptionBudgetLister(informerFactory informers.SharedInformerFactory) policylisters.PodDisruptionBudgetLister {
if utilfeature.DefaultFeatureGate.Enabled(kubefeatures.PodDisruptionBudget) {

106
pkg/scheduler/factory_test.go
View File

@ -20,7 +20,6 @@ import (
"context"
"encoding/json"
"errors"
"fmt"
"reflect"
"testing"
"time"
@ -38,7 +37,6 @@ import (
"k8s.io/client-go/tools/cache"
apitesting "k8s.io/kubernetes/pkg/api/testing"
kubefeatures "k8s.io/kubernetes/pkg/features"
"k8s.io/kubernetes/pkg/scheduler/algorithm"
schedulerapi "k8s.io/kubernetes/pkg/scheduler/apis/config"
"k8s.io/kubernetes/pkg/scheduler/apis/config/scheme"
extenderv1 "k8s.io/kubernetes/pkg/scheduler/apis/extender/v1"
@ -373,49 +371,32 @@ func testClientGetPodRequest(client *fake.Clientset, t *testing.T, podNs string,
}
}
func TestBind(t *testing.T) {
table := []struct {
name string
binding *v1.Binding
}{
{
name: "binding can bind and validate request",
binding: &v1.Binding{
ObjectMeta: metav1.ObjectMeta{
Namespace: metav1.NamespaceDefault,
Name: "foo",
},
Target: v1.ObjectReference{
Name: "foohost.kubernetes.mydomain.com",
},
},
// TODO(#87157): Move to DefaultBinding Plugin tests when it is introduced.
func TestDefaultBinding(t *testing.T) {
binding := &v1.Binding{
ObjectMeta: metav1.ObjectMeta{
Namespace: metav1.NamespaceDefault,
Name: "foo",
},
Target: v1.ObjectReference{
Name: "foohost.kubernetes.mydomain.com",
},
}
for _, test := range table {
t.Run(test.name, func(t *testing.T) {
testBind(test.binding, t)
})
}
}
func testBind(binding *v1.Binding, t *testing.T) {
testPod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: binding.GetName(), Namespace: metav1.NamespaceDefault},
Spec: apitesting.V1DeepEqualSafePodSpec(),
}
client := fake.NewSimpleClientset(&v1.PodList{Items: []v1.Pod{*testPod}})
b := binder{client}
if err := b.Bind(binding); err != nil {
sched := Scheduler{client: client}
if err := sched.defaultBinding(binding); err != nil {
t.Errorf("Unexpected error: %v", err)
return
}
pod := client.CoreV1().Pods(metav1.NamespaceDefault).(*fakeV1.FakePods)
actualBinding, err := pod.GetBinding(binding.GetName())
pods := client.CoreV1().Pods(metav1.NamespaceDefault).(*fakeV1.FakePods)
actualBinding, err := pods.GetBinding(binding.GetName())
if err != nil {
t.Fatalf("Unexpected error: %v", err)
return
@ -461,6 +442,7 @@ type fakeExtender struct {
isBinder bool
interestedPodName string
ignorable bool
gotBind bool
}
func (f *fakeExtender) Name() string {
@ -496,6 +478,7 @@ func (f *fakeExtender) Prioritize(
func (f *fakeExtender) Bind(binding *v1.Binding) error {
if f.isBinder {
f.gotBind = true
return nil
}
return errors.New("not a binder")
@ -509,65 +492,6 @@ func (f *fakeExtender) IsInterested(pod *v1.Pod) bool {
return pod != nil && pod.Name == f.interestedPodName
}
func TestGetBinderFunc(t *testing.T) {
table := []struct {
podName string
extenders []algorithm.SchedulerExtender
expectedBinderType string
name string
}{
{
name: "the extender is not a binder",
podName: "pod0",
extenders: []algorithm.SchedulerExtender{
&fakeExtender{isBinder: false, interestedPodName: "pod0"},
},
expectedBinderType: "*scheduler.binder",
},
{
name: "one of the extenders is a binder and interested in pod",
podName: "pod0",
extenders: []algorithm.SchedulerExtender{
&fakeExtender{isBinder: false, interestedPodName: "pod0"},
&fakeExtender{isBinder: true, interestedPodName: "pod0"},
},
expectedBinderType: "*scheduler.fakeExtender",
},
{
name: "one of the extenders is a binder, but not interested in pod",
podName: "pod1",
extenders: []algorithm.SchedulerExtender{
&fakeExtender{isBinder: false, interestedPodName: "pod1"},
&fakeExtender{isBinder: true, interestedPodName: "pod0"},
},
expectedBinderType: "*scheduler.binder",
},
}
for _, test := range table {
t.Run(test.name, func(t *testing.T) {
testGetBinderFunc(test.expectedBinderType, test.podName, test.extenders, t)
})
}
}
func testGetBinderFunc(expectedBinderType, podName string, extenders []algorithm.SchedulerExtender, t *testing.T) {
pod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: podName,
},
}
f := &Configurator{}
binderFunc := getBinderFunc(f.client, extenders)
binder := binderFunc(pod)
binderType := fmt.Sprintf("%s", reflect.TypeOf(binder))
if binderType != expectedBinderType {
t.Errorf("Expected binder %q but got %q", expectedBinderType, binderType)
}
}
type TestPlugin struct {
name string
}

87
pkg/scheduler/scheduler.go
View File

@ -82,7 +82,6 @@ type Scheduler struct {
SchedulerCache internalcache.Cache
Algorithm core.ScheduleAlgorithm
GetBinder func(pod *v1.Pod) Binder
// PodConditionUpdater is used only in case of scheduling errors. If we succeed
// with scheduling, PodScheduled condition will be updated in apiserver in /bind
// handler so that binding and setting PodCondition it is atomic.
@ -119,6 +118,9 @@ type Scheduler struct {
SchedulingQueue internalqueue.SchedulingQueue
scheduledPodsHasSynced func() bool
// TODO(#87157): Remove this when the DefaultBinding Plugin is introduced.
client clientset.Interface
}
// Cache returns the cache in scheduler for test to check the data in scheduler.
@ -333,6 +335,7 @@ func New(client clientset.Interface,
sched.podConditionUpdater = &podConditionUpdaterImpl{client}
sched.podPreemptor = &podPreemptorImpl{client}
sched.scheduledPodsHasSynced = podInformer.Informer().HasSynced
sched.client = client
AddAllEventHandlers(sched, options.schedulerName, informerFactory, podInformer)
return sched, nil
@ -505,28 +508,61 @@ func (sched *Scheduler) assume(assumed *v1.Pod, host string) error {
return nil
}
// bind binds a pod to a given node defined in a binding object. We expect this to run asynchronously, so we
// handle binding metrics internally.
func (sched *Scheduler) bind(ctx context.Context, assumed *v1.Pod, targetNode string, state *framework.CycleState) error {
bindingStart := time.Now()
bindStatus := sched.Framework.RunBindPlugins(ctx, state, assumed, targetNode)
var err error
if !bindStatus.IsSuccess() {
if bindStatus.Code() == framework.Skip {
// All bind plugins chose to skip binding of this pod, call original binding function.
// If binding succeeds then PodScheduled condition will be updated in apiserver so that
// it's atomic with setting host.
err = sched.GetBinder(assumed).Bind(&v1.Binding{
ObjectMeta: metav1.ObjectMeta{Namespace: assumed.Namespace, Name: assumed.Name, UID: assumed.UID},
Target: v1.ObjectReference{
Kind: "Node",
Name: targetNode,
},
})
} else {
err = fmt.Errorf("Bind failure, code: %d: %v", bindStatus.Code(), bindStatus.Message())
}
// bind binds a pod to a given node defined in a binding object.
// The precedence for binding is: (1) extenders, (2) plugins and (3) default binding.
// We expect this to run asynchronously, so we handle binding metrics internally.
func (sched *Scheduler) bind(ctx context.Context, assumed *v1.Pod, targetNode string, state *framework.CycleState) (err error) {
start := time.Now()
defer func() {
sched.finishBinding(assumed, targetNode, start, err)
}()
binding := &v1.Binding{
ObjectMeta: metav1.ObjectMeta{
Namespace: assumed.Namespace,
Name: assumed.Name,
UID: assumed.UID,
},
Target: v1.ObjectReference{
Kind: "Node",
Name: targetNode,
},
}
bound, err := sched.extendersBinding(assumed, binding)
if bound {
return err
}
bindStatus := sched.Framework.RunBindPlugins(ctx, state, assumed, targetNode)
if bindStatus.IsSuccess() {
return nil
}
if bindStatus.Code() != framework.Skip {
return fmt.Errorf("bind failure, code: %d: %v", bindStatus.Code(), bindStatus.Message())
}
// All bind plugins chose to skip binding of this pod, call original binding
// function. If binding succeeds then PodScheduled condition will be updated
// in apiserver so that it's atomic with setting host.
return sched.defaultBinding(binding)
}
// TODO(#87159): Move this to a Plugin.
func (sched *Scheduler) extendersBinding(assumed *v1.Pod, binding *v1.Binding) (bool, error) {
for _, extender := range sched.Algorithm.Extenders() {
if !extender.IsBinder() || !extender.IsInterested(assumed) {
continue
}
return true, extender.Bind(binding)
}
return false, nil
}
// TODO(#87157): Move this to a Plugin.
func (sched *Scheduler) defaultBinding(binding *v1.Binding) error {
klog.V(3).Infof("Attempting to bind %v/%v to %v", binding.Namespace, binding.Name, binding.Target.Name)
return sched.client.CoreV1().Pods(binding.Namespace).Bind(binding)
}
func (sched *Scheduler) finishBinding(assumed *v1.Pod, targetNode string, start time.Time, err error) {
if finErr := sched.SchedulerCache.FinishBinding(assumed); finErr != nil {
klog.Errorf("scheduler cache FinishBinding failed: %v", finErr)
}
@ -535,13 +571,12 @@ func (sched *Scheduler) bind(ctx context.Context, assumed *v1.Pod, targetNode st
if err := sched.SchedulerCache.ForgetPod(assumed); err != nil {
klog.Errorf("scheduler cache ForgetPod failed: %v", err)
}
return err
return
}
metrics.BindingLatency.Observe(metrics.SinceInSeconds(bindingStart))
metrics.DeprecatedSchedulingDuration.WithLabelValues(metrics.Binding).Observe(metrics.SinceInSeconds(bindingStart))
metrics.BindingLatency.Observe(metrics.SinceInSeconds(start))
metrics.DeprecatedSchedulingDuration.WithLabelValues(metrics.Binding).Observe(metrics.SinceInSeconds(start))
sched.Recorder.Eventf(assumed, nil, v1.EventTypeNormal, "Scheduled", "Binding", "Successfully assigned %v/%v to %v", assumed.Namespace, assumed.Name, targetNode)
return nil
}
// scheduleOne does the entire scheduling workflow for a single pod. It is serialized on the scheduling algorithm's host fitting.

153
pkg/scheduler/scheduler_test.go
View File

@ -27,6 +27,7 @@ import (
"testing"
"time"
"github.com/google/go-cmp/cmp"
v1 "k8s.io/api/core/v1"
"k8s.io/api/events/v1beta1"
"k8s.io/apimachinery/pkg/api/resource"
@ -34,12 +35,12 @@ import (
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/diff"
"k8s.io/apimachinery/pkg/util/sets"
"k8s.io/apimachinery/pkg/util/wait"
"k8s.io/client-go/informers"
clientsetfake "k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/kubernetes/scheme"
clienttesting "k8s.io/client-go/testing"
clientcache "k8s.io/client-go/tools/cache"
"k8s.io/client-go/tools/events"
volumescheduling "k8s.io/kubernetes/pkg/controller/volume/scheduling"
@ -66,12 +67,6 @@ var (
emptyFramework, _ = framework.NewFramework(emptyPluginRegistry, nil, nil)
)
type fakeBinder struct {
b func(binding *v1.Binding) error
}
func (fb fakeBinder) Bind(binding *v1.Binding) error { return fb.b(binding) }
type fakePodConditionUpdater struct{}
func (fc fakePodConditionUpdater) update(pod *v1.Pod, podCondition *v1.PodCondition) error {
@ -258,6 +253,7 @@ func TestScheduler(t *testing.T) {
expectForgetPod: podWithID("foo", testNode.Name),
eventReason: "FailedScheduling",
}, {
name: "deleting pod",
sendPod: deletingPod("foo"),
algo: mockScheduler{core.ScheduleResult{}, nil},
eventReason: "FailedScheduling",
@ -292,16 +288,18 @@ func TestScheduler(t *testing.T) {
return pod.UID == gotAssumedPod.UID
},
}
client := clientsetfake.NewSimpleClientset(item.sendPod)
client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) {
if action.GetSubresource() != "binding" {
return false, nil, nil
}
gotBinding = action.(clienttesting.CreateAction).GetObject().(*v1.Binding)
return true, gotBinding, item.injectBindError
})
s := &Scheduler{
SchedulerCache: sCache,
Algorithm: item.algo,
GetBinder: func(pod *v1.Pod) Binder {
return fakeBinder{func(b *v1.Binding) error {
gotBinding = b
return item.injectBindError
}}
},
SchedulerCache: sCache,
Algorithm: item.algo,
podConditionUpdater: fakePodConditionUpdater{},
Error: func(p *framework.PodInfo, err error) {
gotPod = p.Pod
@ -313,12 +311,13 @@ func TestScheduler(t *testing.T) {
Framework: emptyFramework,
Recorder: eventBroadcaster.NewRecorder(scheme.Scheme, "scheduler"),
VolumeBinder: volumebinder.NewFakeVolumeBinder(&volumescheduling.FakeVolumeBinderConfig{AllBound: true}),
client: client,
}
called := make(chan struct{})
stopFunc := eventBroadcaster.StartEventWatcher(func(obj runtime.Object) {
e, _ := obj.(*v1beta1.Event)
if e, a := item.eventReason, e.Reason; e != a {
t.Errorf("expected %v, got %v", e, a)
if e.Reason != item.eventReason {
t.Errorf("got event %v, want %v", e.Reason, item.eventReason)
}
close(called)
})
@ -336,8 +335,8 @@ func TestScheduler(t *testing.T) {
if e, a := item.expectError, gotError; !reflect.DeepEqual(e, a) {
t.Errorf("error: wanted %v, got %v", e, a)
}
if e, a := item.expectBind, gotBinding; !reflect.DeepEqual(e, a) {
t.Errorf("error: %s", diff.ObjectDiff(e, a))
if diff := cmp.Diff(item.expectBind, gotBinding); diff != "" {
t.Errorf("got binding diff (-want, +got): %s", diff)
}
stopFunc()
})
@ -626,15 +625,19 @@ func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache internalcache.C
bindingChan := make(chan *v1.Binding, 1)
errChan := make(chan error, 1)
client := clientsetfake.NewSimpleClientset()
client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) {
var b *v1.Binding
if action.GetSubresource() == "binding" {
b := action.(clienttesting.CreateAction).GetObject().(*v1.Binding)
bindingChan <- b
}
return true, b, nil
})
sched := &Scheduler{
SchedulerCache: scache,
Algorithm: algo,
GetBinder: func(pod *v1.Pod) Binder {
return fakeBinder{func(b *v1.Binding) error {
bindingChan <- b
return nil
}}
},
NextPod: func() *framework.PodInfo {
return &framework.PodInfo{Pod: clientcache.Pop(queuedPodStore).(*v1.Pod)}
},
@ -646,6 +649,7 @@ func setupTestScheduler(queuedPodStore *clientcache.FIFO, scache internalcache.C
podPreemptor: fakePodPreemptor{},
Framework: fwk,
VolumeBinder: volumebinder.NewFakeVolumeBinder(&volumescheduling.FakeVolumeBinderConfig{AllBound: true}),
client: client,
}
if recorder != nil {
@ -933,3 +937,102 @@ priorities:
}
}
}
func TestSchedulerBinding(t *testing.T) {
table := []struct {
podName string
extenders []algorithm.SchedulerExtender
wantBinderID int
name string
}{
{
name: "the extender is not a binder",
podName: "pod0",
extenders: []algorithm.SchedulerExtender{
&fakeExtender{isBinder: false, interestedPodName: "pod0"},
},
wantBinderID: -1, // default binding.
},
{
name: "one of the extenders is a binder and interested in pod",
podName: "pod0",
extenders: []algorithm.SchedulerExtender{
&fakeExtender{isBinder: false, interestedPodName: "pod0"},
&fakeExtender{isBinder: true, interestedPodName: "pod0"},
},
wantBinderID: 1,
},
{
name: "one of the extenders is a binder, but not interested in pod",
podName: "pod1",
extenders: []algorithm.SchedulerExtender{
&fakeExtender{isBinder: false, interestedPodName: "pod1"},
&fakeExtender{isBinder: true, interestedPodName: "pod0"},
},
wantBinderID: -1, // default binding.
},
}
for _, test := range table {
t.Run(test.name, func(t *testing.T) {
pod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: test.podName,
},
}
defaultBound := false
client := clientsetfake.NewSimpleClientset(pod)
client.PrependReactor("create", "pods", func(action clienttesting.Action) (bool, runtime.Object, error) {
if action.GetSubresource() == "binding" {
defaultBound = true
}
return false, nil, nil
})
fwk, err := framework.NewFramework(framework.Registry{}, nil, nil)
if err != nil {
t.Fatal(err)
}
stop := make(chan struct{})
defer close(stop)
scache := internalcache.New(100*time.Millisecond, stop)
algo := core.NewGenericScheduler(
scache,
nil,
nil,
fwk,
test.extenders,
nil,
nil,
nil,
false,
0,
false,
)
sched := Scheduler{
Algorithm: algo,
Framework: fwk,
Recorder: &events.FakeRecorder{},
SchedulerCache: scache,
client: client,
}
err = sched.bind(context.Background(), pod, "node", nil)
if err != nil {
t.Error(err)
}
// Checking default binding.
if wantBound := test.wantBinderID == -1; defaultBound != wantBound {
t.Errorf("got bound with default binding: %v, want %v", defaultBound, wantBound)
}
// Checking extenders binding.
for i, ext := range test.extenders {
wantBound := i == test.wantBinderID
if gotBound := ext.(*fakeExtender).gotBind; gotBound != wantBound {
t.Errorf("got bound with extender #%d: %v, want %v", i, gotBound, wantBound)
}
}
})
}
}