mirror of
https://github.com/k3s-io/kubernetes.git
synced 2025-12-07 18:06:21 +00:00
Revert "enhancement(scheduler): share waitingPods among profiles"
This reverts commit 227c1915db.
This commit is contained in:
kerthcet
@@ -21,16 +21,13 @@ import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/google/go-cmp/cmp"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
eventsv1 "k8s.io/api/events/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/util/sets"
|
||||
utilfeature "k8s.io/apiserver/pkg/util/feature"
|
||||
"k8s.io/client-go/informers"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
@@ -440,14 +437,12 @@ func initScheduler(ctx context.Context, cache internalcache.Cache, queue interna
|
||||
tf.RegisterBindPlugin(defaultbinder.Name, defaultbinder.New),
|
||||
}
|
||||
eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: client.EventsV1()})
|
||||
waitingPods := frameworkruntime.NewWaitingPodsMap()
|
||||
fwk, err := tf.NewFramework(ctx,
|
||||
registerPluginFuncs,
|
||||
testSchedulerName,
|
||||
frameworkruntime.WithClientSet(client),
|
||||
frameworkruntime.WithInformerFactory(informerFactory),
|
||||
frameworkruntime.WithEventRecorder(eventBroadcaster.NewRecorder(scheme.Scheme, testSchedulerName)),
|
||||
frameworkruntime.WithWaitingPods(waitingPods),
|
||||
)
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
@@ -596,7 +591,6 @@ const (
|
||||
queueSort = "no-op-queue-sort-plugin"
|
||||
fakeBind = "bind-plugin"
|
||||
emptyEventExtensions = "emptyEventExtensions"
|
||||
fakePermit = "fakePermit"
|
||||
)
|
||||
|
||||
func Test_buildQueueingHintMap(t *testing.T) {
|
||||
@@ -911,160 +905,6 @@ func newFramework(ctx context.Context, r frameworkruntime.Registry, profile sche
|
||||
)
|
||||
}
|
||||
|
||||
func TestFrameworkHandler_IterateOverWaitingPods(t *testing.T) {
|
||||
const (
|
||||
testSchedulerProfile1 = "test-scheduler-profile-1"
|
||||
testSchedulerProfile2 = "test-scheduler-profile-2"
|
||||
testSchedulerProfile3 = "test-scheduler-profile-3"
|
||||
)
|
||||
|
||||
nodes := []runtime.Object{
|
||||
st.MakeNode().Name("node1").UID("node1").Obj(),
|
||||
st.MakeNode().Name("node2").UID("node2").Obj(),
|
||||
st.MakeNode().Name("node3").UID("node3").Obj(),
|
||||
}
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
profiles []schedulerapi.KubeSchedulerProfile
|
||||
waitSchedulingPods []*v1.Pod
|
||||
expectPodNamesInWaitingPods []string
|
||||
}{
|
||||
{
|
||||
name: "pods with same profile are waiting on permit stage",
|
||||
profiles: []schedulerapi.KubeSchedulerProfile{
|
||||
{
|
||||
SchedulerName: testSchedulerProfile1,
|
||||
Plugins: &schedulerapi.Plugins{
|
||||
QueueSort: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "PrioritySort"}}},
|
||||
Permit: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: fakePermit}}},
|
||||
Bind: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "DefaultBinder"}}},
|
||||
},
|
||||
},
|
||||
},
|
||||
waitSchedulingPods: []*v1.Pod{
|
||||
st.MakePod().Name("pod1").UID("pod1").SchedulerName(testSchedulerProfile1).Obj(),
|
||||
st.MakePod().Name("pod2").UID("pod2").SchedulerName(testSchedulerProfile1).Obj(),
|
||||
st.MakePod().Name("pod3").UID("pod3").SchedulerName(testSchedulerProfile1).Obj(),
|
||||
},
|
||||
expectPodNamesInWaitingPods: []string{"pod1", "pod2", "pod3"},
|
||||
},
|
||||
{
|
||||
name: "pods with different profiles are waiting on permit stage",
|
||||
profiles: []schedulerapi.KubeSchedulerProfile{
|
||||
{
|
||||
SchedulerName: testSchedulerProfile1,
|
||||
Plugins: &schedulerapi.Plugins{
|
||||
QueueSort: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "PrioritySort"}}},
|
||||
Permit: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: fakePermit}}},
|
||||
Bind: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "DefaultBinder"}}},
|
||||
},
|
||||
},
|
||||
{
|
||||
SchedulerName: testSchedulerProfile2,
|
||||
Plugins: &schedulerapi.Plugins{
|
||||
QueueSort: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "PrioritySort"}}},
|
||||
Permit: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: fakePermit}}},
|
||||
Bind: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "DefaultBinder"}}},
|
||||
},
|
||||
},
|
||||
{
|
||||
SchedulerName: testSchedulerProfile3,
|
||||
Plugins: &schedulerapi.Plugins{
|
||||
QueueSort: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "PrioritySort"}}},
|
||||
Permit: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: fakePermit}}},
|
||||
Bind: schedulerapi.PluginSet{Enabled: []schedulerapi.Plugin{{Name: "DefaultBinder"}}},
|
||||
},
|
||||
},
|
||||
},
|
||||
waitSchedulingPods: []*v1.Pod{
|
||||
st.MakePod().Name("pod1").UID("pod1").SchedulerName(testSchedulerProfile1).Obj(),
|
||||
st.MakePod().Name("pod2").UID("pod2").SchedulerName(testSchedulerProfile1).Obj(),
|
||||
st.MakePod().Name("pod3").UID("pod3").SchedulerName(testSchedulerProfile2).Obj(),
|
||||
st.MakePod().Name("pod4").UID("pod4").SchedulerName(testSchedulerProfile3).Obj(),
|
||||
},
|
||||
expectPodNamesInWaitingPods: []string{"pod1", "pod2", "pod3", "pod4"},
|
||||
},
|
||||
}
|
||||
|
||||
for _, tc := range cases {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
// Set up scheduler for the 3 nodes.
|
||||
objs := append([]runtime.Object{&v1.Namespace{ObjectMeta: metav1.ObjectMeta{Name: ""}}}, nodes...)
|
||||
fakeClient := fake.NewSimpleClientset(objs...)
|
||||
informerFactory := informers.NewSharedInformerFactory(fakeClient, 0)
|
||||
eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{Interface: fakeClient.EventsV1()})
|
||||
eventRecorder := eventBroadcaster.NewRecorder(scheme.Scheme, fakePermit)
|
||||
|
||||
outOfTreeRegistry := frameworkruntime.Registry{
|
||||
fakePermit: newFakePermitPlugin(eventRecorder),
|
||||
}
|
||||
|
||||
_, ctx := ktesting.NewTestContext(t)
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
scheduler, err := New(
|
||||
ctx,
|
||||
fakeClient,
|
||||
informerFactory,
|
||||
nil,
|
||||
profile.NewRecorderFactory(eventBroadcaster),
|
||||
WithProfiles(tc.profiles...),
|
||||
WithFrameworkOutOfTreeRegistry(outOfTreeRegistry),
|
||||
)
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("Failed to create scheduler: %v", err)
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
waitSchedulingPodNumber := len(tc.waitSchedulingPods)
|
||||
wg.Add(waitSchedulingPodNumber)
|
||||
stopFn, err := eventBroadcaster.StartEventWatcher(func(obj runtime.Object) {
|
||||
e, ok := obj.(*eventsv1.Event)
|
||||
if !ok || (e.Reason != podWaitingReason) {
|
||||
return
|
||||
}
|
||||
wg.Done()
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer stopFn()
|
||||
|
||||
// Run scheduler.
|
||||
informerFactory.Start(ctx.Done())
|
||||
informerFactory.WaitForCacheSync(ctx.Done())
|
||||
go scheduler.Run(ctx)
|
||||
|
||||
// Send pods to be scheduled.
|
||||
for _, p := range tc.waitSchedulingPods {
|
||||
_, err = fakeClient.CoreV1().Pods("").Create(ctx, p, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// Wait all pods in waitSchedulingPods to be scheduled.
|
||||
wg.Wait()
|
||||
|
||||
// Ensure that all waitingPods in scheduler can be obtained from any profiles.
|
||||
for _, fwk := range scheduler.Profiles {
|
||||
actualPodNamesInWaitingPods := sets.NewString()
|
||||
fwk.IterateOverWaitingPods(func(pod framework.WaitingPod) {
|
||||
actualPodNamesInWaitingPods.Insert(pod.GetPod().Name)
|
||||
})
|
||||
// Validate the name of pods in waitingPods matches expectations.
|
||||
if actualPodNamesInWaitingPods.Len() != len(tc.expectPodNamesInWaitingPods) ||
|
||||
!actualPodNamesInWaitingPods.HasAll(tc.expectPodNamesInWaitingPods...) {
|
||||
t.Fatalf("Unexpected waitingPods in scheduler profile %s, expect: %#v, got: %#v", fwk.ProfileName(), tc.expectPodNamesInWaitingPods, actualPodNamesInWaitingPods.List())
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
var _ framework.QueueSortPlugin = &fakeQueueSortPlugin{}
|
||||
|
||||
// fakeQueueSortPlugin is a no-op implementation for QueueSort extension point.
|
||||
@@ -1164,36 +1004,3 @@ func (*emptyEventsToRegisterPlugin) Filter(_ context.Context, _ *framework.Cycle
|
||||
}
|
||||
|
||||
func (*emptyEventsToRegisterPlugin) EventsToRegister() []framework.ClusterEventWithHint { return nil }
|
||||
|
||||
// fakePermitPlugin only implements PermitPlugin interface.
|
||||
type fakePermitPlugin struct {
|
||||
eventRecorder events.EventRecorder
|
||||
}
|
||||
|
||||
func newFakePermitPlugin(eventRecorder events.EventRecorder) frameworkruntime.PluginFactory {
|
||||
return func(ctx context.Context, configuration runtime.Object, f framework.Handle) (framework.Plugin, error) {
|
||||
pl := &fakePermitPlugin{
|
||||
eventRecorder: eventRecorder,
|
||||
}
|
||||
return pl, nil
|
||||
}
|
||||
}
|
||||
|
||||
func (f fakePermitPlugin) Name() string {
|
||||
return fakePermit
|
||||
}
|
||||
|
||||
const (
|
||||
podWaitingReason = "podWaiting"
|
||||
)
|
||||
|
||||
func (f fakePermitPlugin) Permit(ctx context.Context, state *framework.CycleState, p *v1.Pod, nodeName string) (*framework.Status, time.Duration) {
|
||||
defer func() {
|
||||
// Send event with podWaiting reason to broadcast this pod is already waiting in the permit stage.
|
||||
f.eventRecorder.Eventf(p, nil, v1.EventTypeWarning, podWaitingReason, "", "")
|
||||
}()
|
||||
|
||||
return framework.NewStatus(framework.Wait), 100 * time.Second
|
||||
}
|
||||
|
||||
var _ framework.PermitPlugin = &fakePermitPlugin{}
|
||||
|
||||
Reference in New Issue
Block a user