Merge pull request #81911 from ahmad-diaa/use-scheduler-in-createConfiguratorArgsWithPodInformer

Use scheduler.New() Instead of factory.NewConfigFactory() in Integration Test scheduler_test.go

pkg/scheduler/scheduler.go

@@ -181,7 +181,7 @@ func New(client clientset.Interface,
 	registry framework.Registry,
 	plugins *kubeschedulerconfig.Plugins,
 	pluginConfig []kubeschedulerconfig.PluginConfig,
-	opts ...func(o *schedulerOptions)) (*Scheduler, error) {
+	opts ...Option) (*Scheduler, error) {
 
 	options := defaultSchedulerOptions
 	for _, opt := range opts {

test/integration/scheduler/BUILD

@@ -93,6 +93,7 @@ go_library(
         "//pkg/scheduler:go_default_library",
         "//pkg/scheduler/algorithmprovider:go_default_library",
         "//pkg/scheduler/api:go_default_library",
+        "//pkg/scheduler/api/latest:go_default_library",
         "//pkg/scheduler/apis/config:go_default_library",
         "//pkg/scheduler/factory:go_default_library",
         "//pkg/scheduler/framework/plugins:go_default_library",
@@ -104,6 +105,7 @@ go_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/labels: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/util/uuid:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",

test/integration/scheduler/framework_test.go

@@ -449,7 +449,7 @@ func TestPreFilterPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "prefilter-plugin", nil),
-		false, nil, registry, plugins, emptyPluginConfig, false, time.Second)
+		false, nil, registry, plugins, emptyPluginConfig, time.Second)
 
 	defer cleanupTest(t, context)
 
@@ -627,7 +627,7 @@ func TestReservePlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "reserve-plugin", nil),
-		false, nil, registry, plugins, emptyPluginConfig, false, time.Second)
+		false, nil, registry, plugins, emptyPluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -691,7 +691,7 @@ func TestPrebindPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "prebind-plugin", nil),
-		false, nil, registry, plugins, preBindPluginConfig, false, time.Second)
+		false, nil, registry, plugins, preBindPluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -792,7 +792,7 @@ func TestUnreservePlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "unreserve-plugin", nil),
-		false, nil, registry, plugins, pluginConfig, false, time.Second)
+		false, nil, registry, plugins, pluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -907,7 +907,7 @@ func TestBindPlugin(t *testing.T) {
 
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t, testContext,
-		false, nil, registry, plugins, pluginConfig, false, time.Second)
+		false, nil, registry, plugins, pluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1079,7 +1079,7 @@ func TestPostBindPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "postbind-plugin", nil),
-		false, nil, registry, plugins, pluginConfig, false, time.Second)
+		false, nil, registry, plugins, pluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1160,7 +1160,7 @@ func TestPermitPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "permit-plugin", nil),
-		false, nil, registry, plugins, pluginConfig, false, time.Second)
+		false, nil, registry, plugins, pluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1272,7 +1272,7 @@ func TestCoSchedulingWithPermitPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "permit-plugin", nil),
-		false, nil, registry, plugins, pluginConfig, false, time.Second)
+		false, nil, registry, plugins, pluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1362,7 +1362,7 @@ func TestFilterPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "filter-plugin", nil),
-		false, nil, registry, plugin, emptyPluginConfig, false, time.Second)
+		false, nil, registry, plugin, emptyPluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1422,7 +1422,7 @@ func TestPostFilterPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "post-filter-plugin", nil),
-		false, nil, registry, pluginsConfig, emptyPluginConfig, false, time.Second)
+		false, nil, registry, pluginsConfig, emptyPluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1487,7 +1487,7 @@ func TestPreemptWithPermitPlugin(t *testing.T) {
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "preempt-with-permit-plugin", nil),
-		false, nil, registry, plugins, pluginConfig, false, time.Second)
+		false, nil, registry, plugins, pluginConfig, time.Second)
 	defer cleanupTest(t, context)
 
 	cs := context.clientSet
@@ -1558,7 +1558,7 @@ func initTestContextForScorePlugin(t *testing.T, plugins *schedulerconfig.Plugin
 	// Create the master and the scheduler with the test plugin set.
 	context := initTestSchedulerWithOptions(t,
 		initTestMaster(t, "score-plugin", nil),
-		false, nil, registry, plugins, emptyPluginConfig, false, time.Second)
+		false, nil, registry, plugins, emptyPluginConfig, time.Second)
 
 	cs := context.clientSet
 	_, err := createNodes(cs, "test-node", nil, 10)

test/integration/scheduler/scheduler_test.go

@@ -352,7 +352,7 @@ func TestUnschedulableNodes(t *testing.T) {
 	context := initTest(t, "unschedulable-nodes")
 	defer cleanupTest(t, context)
 
-	nodeLister := context.schedulerConfigArgs.NodeInformer.Lister()
+	nodeLister := context.informerFactory.Core().V1().Nodes().Lister()
 	// NOTE: This test cannot run in parallel, because it is creating and deleting
 	// non-namespaced objects (Nodes).
 	defer context.clientSet.CoreV1().Nodes().DeleteCollection(nil, metav1.ListOptions{})
@@ -596,38 +596,7 @@ func TestMultiScheduler(t *testing.T) {
 	}
 
 	// 5. create and start a scheduler with name "foo-scheduler"
-	clientSet2 := clientset.NewForConfigOrDie(&restclient.Config{Host: context.httpServer.URL, ContentConfig: restclient.ContentConfig{GroupVersion: &schema.GroupVersion{Group: "", Version: "v1"}}})
+	context = initTestSchedulerWithOptions(t, context, true, nil, schedulerplugins.NewDefaultRegistry(), nil, []kubeschedulerconfig.PluginConfig{}, time.Second, scheduler.WithName(fooScheduler))
-	informerFactory2 := informers.NewSharedInformerFactory(context.clientSet, 0)
-	podInformer2 := factory.NewPodInformer(context.clientSet, 0)
-
-	stopCh := make(chan struct{})
-	defer close(stopCh)
-
-	schedulerConfigFactory2 := factory.NewConfigFactory(createConfiguratorArgsWithPodInformer(fooScheduler, clientSet2, podInformer2, informerFactory2, schedulerplugins.NewDefaultRegistry(),
-		nil, []kubeschedulerconfig.PluginConfig{}, stopCh))
-	schedulerConfig2, err := schedulerConfigFactory2.Create()
-	if err != nil {
-		t.Errorf("Couldn't create scheduler config: %v", err)
-	}
-	eventBroadcaster2 := events.NewBroadcaster(&events.EventSinkImpl{Interface: clientSet2.EventsV1beta1().Events("")})
-	schedulerConfig2.Recorder = eventBroadcaster2.NewRecorder(legacyscheme.Scheme, "k8s.io/"+fooScheduler)
-	eventBroadcaster2.StartRecordingToSink(stopCh)
-
-	sched2 := scheduler.NewFromConfig(schedulerConfig2)
-	scheduler.AddAllEventHandlers(sched2,
-		fooScheduler,
-		context.informerFactory.Core().V1().Nodes(),
-		podInformer2,
-		context.informerFactory.Core().V1().PersistentVolumes(),
-		context.informerFactory.Core().V1().PersistentVolumeClaims(),
-		context.informerFactory.Core().V1().Services(),
-		context.informerFactory.Storage().V1().StorageClasses(),
-		context.informerFactory.Storage().V1beta1().CSINodes(),
-	)
-
-	go podInformer2.Informer().Run(stopCh)
-	informerFactory2.Start(stopCh)
-	sched2.Run()
 
 	//	6. **check point-2**:
 	//		- testPodWithAnnotationFitsFoo should be scheduled

test/integration/scheduler/util.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/apimachinery/pkg/util/wait"
@@ -48,6 +49,7 @@ import (
 	podutil "k8s.io/kubernetes/pkg/api/v1/pod"
 	"k8s.io/kubernetes/pkg/controller/disruption"
 	"k8s.io/kubernetes/pkg/scheduler"
+	latestschedulerapi "k8s.io/kubernetes/pkg/scheduler/api/latest"
 	schedulerconfig "k8s.io/kubernetes/pkg/scheduler/apis/config"
 
 	// Register defaults in pkg/scheduler/algorithmprovider.
@@ -62,49 +64,32 @@ import (
 )
 
 type testContext struct {
-	closeFn             framework.CloseFunc
+	closeFn         framework.CloseFunc
-	httpServer          *httptest.Server
+	httpServer      *httptest.Server
-	ns                  *v1.Namespace
+	ns              *v1.Namespace
-	clientSet           *clientset.Clientset
+	clientSet       *clientset.Clientset
-	informerFactory     informers.SharedInformerFactory
+	informerFactory informers.SharedInformerFactory
-	schedulerConfigArgs *factory.ConfigFactoryArgs
+	scheduler       *scheduler.Scheduler
-	schedulerConfig     *factory.Config
+	stopCh          chan struct{}
-	scheduler           *scheduler.Scheduler
-	stopCh              chan struct{}
 }
 
-// createConfiguratorWithPodInformer creates a configurator for scheduler.
+func createAlgorithmSourceFromPolicy(policy *schedulerapi.Policy, clientSet clientset.Interface) schedulerconfig.SchedulerAlgorithmSource {
-func createConfiguratorArgsWithPodInformer(
+	policyString := runtime.EncodeOrDie(latestschedulerapi.Codec, policy)
-	schedulerName string,
+	configPolicyName := "scheduler-custom-policy-config"
-	clientSet clientset.Interface,
+	policyConfigMap := v1.ConfigMap{
-	podInformer coreinformers.PodInformer,
+		ObjectMeta: metav1.ObjectMeta{Namespace: metav1.NamespaceSystem, Name: configPolicyName},
-	informerFactory informers.SharedInformerFactory,
+		Data:       map[string]string{schedulerconfig.SchedulerPolicyConfigMapKey: policyString},
-	pluginRegistry schedulerframework.Registry,
+	}
-	plugins *schedulerconfig.Plugins,
+	policyConfigMap.APIVersion = "v1"
-	pluginConfig []schedulerconfig.PluginConfig,
+	clientSet.CoreV1().ConfigMaps(metav1.NamespaceSystem).Create(&policyConfigMap)
-	stopCh <-chan struct{},
+
-) *factory.ConfigFactoryArgs {
+	return schedulerconfig.SchedulerAlgorithmSource{
-	return &factory.ConfigFactoryArgs{
+		Policy: &schedulerconfig.SchedulerPolicySource{
-		Client:                         clientSet,
+			ConfigMap: &schedulerconfig.SchedulerPolicyConfigMapSource{
-		NodeInformer:                   informerFactory.Core().V1().Nodes(),
+				Namespace: policyConfigMap.Namespace,
-		PodInformer:                    podInformer,
+				Name:      policyConfigMap.Name,
-		PvInformer:                     informerFactory.Core().V1().PersistentVolumes(),
+			},
-		PvcInformer:                    informerFactory.Core().V1().PersistentVolumeClaims(),
+		},
-		ReplicationControllerInformer:  informerFactory.Core().V1().ReplicationControllers(),
-		ReplicaSetInformer:             informerFactory.Apps().V1().ReplicaSets(),
-		StatefulSetInformer:            informerFactory.Apps().V1().StatefulSets(),
-		ServiceInformer:                informerFactory.Core().V1().Services(),
-		PdbInformer:                    informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
-		StorageClassInformer:           informerFactory.Storage().V1().StorageClasses(),
-		CSINodeInformer:                informerFactory.Storage().V1beta1().CSINodes(),
-		Registry:                       pluginRegistry,
-		Plugins:                        plugins,
-		PluginConfig:                   pluginConfig,
-		HardPodAffinitySymmetricWeight: v1.DefaultHardPodAffinitySymmetricWeight,
-		DisablePreemption:              false,
-		PercentageOfNodesToScore:       schedulerapi.DefaultPercentageOfNodesToScore,
-		BindTimeoutSeconds:             600,
-		StopCh:                         stopCh,
 	}
 }
 
@@ -158,7 +143,7 @@ func initTestScheduler(
 ) *testContext {
 	// Pod preemption is enabled by default scheduler configuration.
 	return initTestSchedulerWithOptions(t, context, setPodInformer, policy, schedulerplugins.NewDefaultRegistry(),
-		nil, []schedulerconfig.PluginConfig{}, false, time.Second)
+		nil, []schedulerconfig.PluginConfig{}, time.Second)
 }
 
 // initTestSchedulerWithOptions initializes a test environment and creates a scheduler with default
@@ -171,8 +156,8 @@ func initTestSchedulerWithOptions(
 	pluginRegistry schedulerframework.Registry,
 	plugins *schedulerconfig.Plugins,
 	pluginConfig []schedulerconfig.PluginConfig,
-	disablePreemption bool,
 	resyncPeriod time.Duration,
+	opts ...scheduler.Option,
 ) *testContext {
 	// 1. Create scheduler
 	context.informerFactory = informers.NewSharedInformerFactory(context.clientSet, resyncPeriod)
@@ -185,58 +170,61 @@ func initTestSchedulerWithOptions(
 	} else {
 		podInformer = context.informerFactory.Core().V1().Pods()
 	}
-
-	context.schedulerConfigArgs = createConfiguratorArgsWithPodInformer(
-		v1.DefaultSchedulerName, context.clientSet, podInformer, context.informerFactory, pluginRegistry, plugins,
-		pluginConfig, context.stopCh)
-	configFactory := factory.NewConfigFactory(context.schedulerConfigArgs)
-
 	var err error
-
-	if policy != nil {
-		context.schedulerConfig, err = configFactory.CreateFromConfig(*policy)
-	} else {
-		context.schedulerConfig, err = configFactory.Create()
-	}
-
-	if err != nil {
-		t.Fatalf("Couldn't create scheduler config: %v", err)
-	}
-
-	// set DisablePreemption option
-	context.schedulerConfig.DisablePreemption = disablePreemption
 	eventBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{
 		Interface: context.clientSet.EventsV1beta1().Events(""),
 	})
-	context.schedulerConfig.Recorder = eventBroadcaster.NewRecorder(
+	recorder := eventBroadcaster.NewRecorder(
 		legacyscheme.Scheme,
 		v1.DefaultSchedulerName,
 	)
-
+	var algorithmSrc schedulerconfig.SchedulerAlgorithmSource
-	context.scheduler = scheduler.NewFromConfig(context.schedulerConfig)
+	if policy != nil {
-
+		algorithmSrc = createAlgorithmSourceFromPolicy(policy, context.clientSet)
-	scheduler.AddAllEventHandlers(context.scheduler,
+	} else {
-		v1.DefaultSchedulerName,
+		provider := schedulerconfig.SchedulerDefaultProviderName
+		algorithmSrc = schedulerconfig.SchedulerAlgorithmSource{
+			Provider: &provider,
+		}
+	}
+	opts = append([]scheduler.Option{scheduler.WithBindTimeoutSeconds(600)}, opts...)
+	context.scheduler, err = scheduler.New(
+		context.clientSet,
 		context.informerFactory.Core().V1().Nodes(),
 		podInformer,
 		context.informerFactory.Core().V1().PersistentVolumes(),
 		context.informerFactory.Core().V1().PersistentVolumeClaims(),
+		context.informerFactory.Core().V1().ReplicationControllers(),
+		context.informerFactory.Apps().V1().ReplicaSets(),
+		context.informerFactory.Apps().V1().StatefulSets(),
 		context.informerFactory.Core().V1().Services(),
+		context.informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
 		context.informerFactory.Storage().V1().StorageClasses(),
 		context.informerFactory.Storage().V1beta1().CSINodes(),
+		recorder,
+		algorithmSrc,
+		context.stopCh,
+		pluginRegistry,
+		plugins,
+		pluginConfig,
+		opts...,
 	)
 
+	if err != nil {
+		t.Fatalf("Couldn't create scheduler: %v", err)
+	}
+
 	// set setPodInformer if provided.
 	if setPodInformer {
-		go podInformer.Informer().Run(context.schedulerConfig.StopEverything)
+		go podInformer.Informer().Run(context.scheduler.StopEverything)
-		cache.WaitForNamedCacheSync("scheduler", context.schedulerConfig.StopEverything, podInformer.Informer().HasSynced)
+		cache.WaitForNamedCacheSync("scheduler", context.scheduler.StopEverything, podInformer.Informer().HasSynced)
 	}
 
 	stopCh := make(chan struct{})
 	eventBroadcaster.StartRecordingToSink(stopCh)
 
-	context.informerFactory.Start(context.schedulerConfig.StopEverything)
+	context.informerFactory.Start(context.scheduler.StopEverything)
-	context.informerFactory.WaitForCacheSync(context.schedulerConfig.StopEverything)
+	context.informerFactory.WaitForCacheSync(context.scheduler.StopEverything)
 
 	context.scheduler.Run()
 	return context
@@ -268,9 +256,9 @@ func initDisruptionController(t *testing.T, context *testContext) *disruption.Di
 		mapper,
 		scaleClient)
 
-	informers.Start(context.schedulerConfig.StopEverything)
+	informers.Start(context.scheduler.StopEverything)
-	informers.WaitForCacheSync(context.schedulerConfig.StopEverything)
+	informers.WaitForCacheSync(context.scheduler.StopEverything)
-	go dc.Run(context.schedulerConfig.StopEverything)
+	go dc.Run(context.scheduler.StopEverything)
 	return dc
 }
 
@@ -286,7 +274,7 @@ func initTestDisablePreemption(t *testing.T, nsPrefix string) *testContext {
 	return initTestSchedulerWithOptions(
 		t, initTestMaster(t, nsPrefix, nil), true, nil,
 		schedulerplugins.NewDefaultRegistry(), nil, []schedulerconfig.PluginConfig{},
-		true, time.Second)
+		time.Second, scheduler.WithPreemptionDisabled(true))
 }
 
 // cleanupTest deletes the scheduler and the test namespace. It should be called