Merge pull request #80697 from draveness/feature/remove-scheduler-new-config-factory

feat: use scheduler.New instead in createSchedulerConfigurator

test/integration/scheduler_perf/BUILD

@@ -14,8 +14,10 @@ go_library(
     ],
     importpath = "k8s.io/kubernetes/test/integration/scheduler_perf",
     deps = [
-        "//pkg/scheduler/factory:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/labels:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
+        "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/client-go/rest:go_default_library",
         "//test/integration/util:go_default_library",
@@ -40,9 +42,9 @@ go_test(
         "//staging/src/k8s.io/api/storage/v1beta1:go_default_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/apiserver/pkg/util/feature:go_default_library",
-        "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
         "//staging/src/k8s.io/component-base/featuregate/testing:go_default_library",
         "//staging/src/k8s.io/csi-translation-lib/plugins:go_default_library",
         "//test/integration/framework:go_default_library",

test/integration/scheduler_perf/scheduler_bench_test.go

@@ -21,7 +21,7 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	storagev1beta1 "k8s.io/api/storage/v1beta1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,6 +31,7 @@ import (
 	csilibplugins "k8s.io/csi-translation-lib/plugins"
 	"k8s.io/klog"
 	"k8s.io/kubernetes/pkg/features"
+	"k8s.io/kubernetes/pkg/scheduler/factory"
 	"k8s.io/kubernetes/pkg/volume/util"
 	"k8s.io/kubernetes/test/integration/framework"
 	testutils "k8s.io/kubernetes/test/utils"
@@ -358,12 +359,12 @@ func benchmarkScheduling(numNodes, numExistingPods, minPods int,
 	if b.N < minPods {
 		b.N = minPods
 	}
-	schedulerConfigArgs, finalFunc := mustSetupScheduler()
+	finalFunc, clientset := mustSetupScheduler()
 	defer finalFunc()
-	c := schedulerConfigArgs.Client
 
+	podInformer := factory.NewPodInformer(clientset, 0)
 	nodePreparer := framework.NewIntegrationTestNodePreparer(
-		c,
+		clientset,
 		[]testutils.CountToStrategy{{Count: numNodes, Strategy: nodeStrategy}},
 		"scheduler-perf-",
 	)
@@ -374,12 +375,11 @@ func benchmarkScheduling(numNodes, numExistingPods, minPods int,
 
 	config := testutils.NewTestPodCreatorConfig()
 	config.AddStrategy("sched-test", numExistingPods, setupPodStrategy)
-	podCreator := testutils.NewTestPodCreator(c, config)
+	podCreator := testutils.NewTestPodCreator(clientset, config)
 	podCreator.CreatePods()
 
-	podLister := schedulerConfigArgs.PodInformer.Lister()
 	for {
-		scheduled, err := getScheduledPods(podLister)
+		scheduled, err := getScheduledPods(podInformer)
 		if err != nil {
 			klog.Fatalf("%v", err)
 		}
@@ -392,12 +392,11 @@ func benchmarkScheduling(numNodes, numExistingPods, minPods int,
 	b.ResetTimer()
 	config = testutils.NewTestPodCreatorConfig()
 	config.AddStrategy("sched-test", b.N, testPodStrategy)
-	podCreator = testutils.NewTestPodCreator(c, config)
+	podCreator = testutils.NewTestPodCreator(clientset, config)
 	podCreator.CreatePods()
 	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, err := getScheduledPods(podLister)
+		scheduled, err := getScheduledPods(podInformer)
 		if err != nil {
 			klog.Fatalf("%v", err)
 		}

test/integration/scheduler_perf/scheduler_test.go

@@ -23,14 +23,15 @@ import (
 	"testing"
 	"time"
 
-	"k8s.io/api/core/v1"
+	v1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/labels"
-	listers "k8s.io/client-go/listers/core/v1"
-	"k8s.io/klog"
+	coreinformers "k8s.io/client-go/informers/core/v1"
+	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/kubernetes/pkg/scheduler/factory"
 	testutils "k8s.io/kubernetes/test/utils"
+
+	"k8s.io/klog"
 )
 
 const (
@@ -107,18 +108,19 @@ type testConfig struct {
 	numNodes            int
 	mutatedNodeTemplate *v1.Node
 	mutatedPodTemplate  *v1.Pod
-	schedulerSupport    *factory.ConfigFactoryArgs
+	clientset           clientset.Interface
+	podInformer         coreinformers.PodInformer
 	destroyFunc         func()
 }
 
 // getBaseConfig returns baseConfig after initializing number of nodes and pods.
 func getBaseConfig(nodes int, pods int) *testConfig {
-	schedulerConfigArgs, destroyFunc := mustSetupScheduler()
+	destroyFunc, clientset := mustSetupScheduler()
 	return &testConfig{
-		schedulerSupport: schedulerConfigArgs,
-		destroyFunc:      destroyFunc,
-		numNodes:         nodes,
-		numPods:          pods,
+		clientset:   clientset,
+		destroyFunc: destroyFunc,
+		numNodes:    nodes,
+		numPods:     pods,
 	}
 }
 
@@ -134,11 +136,12 @@ func schedulePods(config *testConfig) int32 {
 	// We are interested in low scheduling rates (i.e. qps=2),
 	minQPS := int32(math.MaxInt32)
 	start := time.Now()
-	podLister := config.schedulerSupport.PodInformer.Lister()
+
+	podInformer := factory.NewPodInformer(config.clientset, 0)
 	// Bake in time for the first pod scheduling event.
 	for {
 		time.Sleep(50 * time.Millisecond)
-		scheduled, err := getScheduledPods(podLister)
+		scheduled, err := getScheduledPods(podInformer)
 		if err != nil {
 			klog.Fatalf("%v", err)
 		}
@@ -153,14 +156,11 @@ func schedulePods(config *testConfig) int32 {
 
 	// Now that scheduling has started, lets start taking the pulse on how many pods are happening per second.
 	for {
-		// 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, err := getScheduledPods(podLister)
+		scheduled, err := getScheduledPods(podInformer)
 		if err != nil {
 			klog.Fatalf("%v", err)
 		}
-
 		// We will be completed when all pods are done being scheduled.
 		// return the worst-case-scenario interval that was seen during this time.
 		// Note this should never be low due to cold-start, so allow bake in sched time if necessary.
@@ -186,20 +186,6 @@ func schedulePods(config *testConfig) int32 {
 	}
 }
 
-func getScheduledPods(lister listers.PodLister) ([]*v1.Pod, error) {
-	all, err := lister.List(labels.Everything())
-	if err != nil {
-		return nil, err
-	}
-	scheduled := make([]*v1.Pod, 0, len(all))
-	for _, pod := range all {
-		if len(pod.Spec.NodeName) > 0 {
-			scheduled = append(scheduled, pod)
-		}
-	}
-	return scheduled, nil
-}
-
 // mutateNodeTemplate returns the modified node needed for creation of nodes.
 func (na nodeAffinity) mutateNodeTemplate(node *v1.Node) {
 	labels := make(map[string]string)
@@ -237,17 +223,18 @@ func (na nodeAffinity) mutatePodTemplate(pod *v1.Pod) {
 // generateNodes generates nodes to be used for scheduling.
 func (inputConfig *schedulerPerfConfig) generateNodes(config *testConfig) {
 	for i := 0; i < inputConfig.NodeCount; i++ {
-		config.schedulerSupport.Client.CoreV1().Nodes().Create(config.mutatedNodeTemplate)
+		config.clientset.CoreV1().Nodes().Create(config.mutatedNodeTemplate)
+
 	}
 	for i := 0; i < config.numNodes-inputConfig.NodeCount; i++ {
-		config.schedulerSupport.Client.CoreV1().Nodes().Create(baseNodeTemplate)
+		config.clientset.CoreV1().Nodes().Create(baseNodeTemplate)
 	}
 }
 
 // generatePods generates pods to be used for scheduling.
 func (inputConfig *schedulerPerfConfig) generatePods(config *testConfig) {
-	testutils.CreatePod(config.schedulerSupport.Client, "sample", inputConfig.PodCount, config.mutatedPodTemplate)
-	testutils.CreatePod(config.schedulerSupport.Client, "sample", config.numPods-inputConfig.PodCount, basePodTemplate)
+	testutils.CreatePod(config.clientset, "sample", inputConfig.PodCount, config.mutatedPodTemplate)
+	testutils.CreatePod(config.clientset, "sample", config.numPods-inputConfig.PodCount, basePodTemplate)
 }
 
 // generatePodAndNodeTopology is the wrapper function for modifying both pods and node objects.

test/integration/scheduler_perf/util.go

@@ -17,21 +17,23 @@ limitations under the License.
 package benchmark
 
 import (
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime/schema"
+	coreinformers "k8s.io/client-go/informers/core/v1"
 	clientset "k8s.io/client-go/kubernetes"
 	restclient "k8s.io/client-go/rest"
-	"k8s.io/kubernetes/pkg/scheduler/factory"
 	"k8s.io/kubernetes/test/integration/util"
 )
 
 // mustSetupScheduler starts the following components:
 // - k8s api server (a.k.a. master)
 // - scheduler
-// It returns scheduler config factory and destroyFunc which should be used to
+// It returns clientset and destroyFunc which should be used to
 // remove resources after finished.
 // Notes on rate limiter:
 //   - client rate limit is set to 5000.
-func mustSetupScheduler() (*factory.ConfigFactoryArgs, util.ShutdownFunc) {
+func mustSetupScheduler() (util.ShutdownFunc, clientset.Interface) {
 	apiURL, apiShutdown := util.StartApiserver()
 	clientSet := clientset.NewForConfigOrDie(&restclient.Config{
 		Host:          apiURL,
@@ -39,11 +41,27 @@ func mustSetupScheduler() (*factory.ConfigFactoryArgs, util.ShutdownFunc) {
 		QPS:           5000.0,
 		Burst:         5000,
 	})
-	schedulerConfigArgs, schedulerShutdown := util.StartScheduler(clientSet)
+	_, schedulerShutdown := util.StartScheduler(clientSet)
 
 	shutdownFunc := func() {
 		schedulerShutdown()
 		apiShutdown()
 	}
-	return schedulerConfigArgs, shutdownFunc
+
+	return shutdownFunc, clientSet
+}
+
+func getScheduledPods(podInformer coreinformers.PodInformer) ([]*v1.Pod, error) {
+	pods, err := podInformer.Lister().List(labels.Everything())
+	if err != nil {
+		return nil, err
+	}
+	scheduled := make([]*v1.Pod, 0, len(pods))
+	for i := range pods {
+		pod := pods[i]
+		if len(pod.Spec.NodeName) > 0 {
+			scheduled = append(scheduled, pod)
+		}
+	}
+	return scheduled, nil
 }

test/integration/util/BUILD

@@ -16,8 +16,8 @@ go_library(
         "//pkg/api/legacyscheme:go_default_library",
         "//pkg/scheduler:go_default_library",
         "//pkg/scheduler/algorithmprovider/defaults:go_default_library",
-        "//pkg/scheduler/api:go_default_library",
-        "//pkg/scheduler/factory:go_default_library",
+        "//pkg/scheduler/apis/config:go_default_library",
+        "//pkg/scheduler/framework/v1alpha1:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",

test/integration/util/util.go

@@ -30,8 +30,8 @@ import (
 
 	// import DefaultProvider
 	_ "k8s.io/kubernetes/pkg/scheduler/algorithmprovider/defaults"
-	schedulerapi "k8s.io/kubernetes/pkg/scheduler/api"
-	"k8s.io/kubernetes/pkg/scheduler/factory"
+	schedulerconfig "k8s.io/kubernetes/pkg/scheduler/apis/config"
+	schedulerframework "k8s.io/kubernetes/pkg/scheduler/framework/v1alpha1"
 	"k8s.io/kubernetes/test/integration/framework"
 )
 
@@ -57,9 +57,8 @@ func StartApiserver() (string, ShutdownFunc) {
 }
 
 // StartScheduler configures and starts a scheduler given a handle to the clientSet interface
-// and event broadcaster. It returns a handle to the configurator args for the running scheduler
-// and the shutdown function to stop it.
-func StartScheduler(clientSet clientset.Interface) (*factory.ConfigFactoryArgs, ShutdownFunc) {
+// and event broadcaster. It returns the running scheduler and the shutdown function to stop it.
+func StartScheduler(clientSet clientset.Interface) (*scheduler.Scheduler, ShutdownFunc) {
 	informerFactory := informers.NewSharedInformerFactory(clientSet, 0)
 	stopCh := make(chan struct{})
 	evtBroadcaster := events.NewBroadcaster(&events.EventSinkImpl{
@@ -67,16 +66,15 @@ func StartScheduler(clientSet clientset.Interface) (*factory.ConfigFactoryArgs,
 
 	evtBroadcaster.StartRecordingToSink(stopCh)
 
-	configuratorArgs := createSchedulerConfiguratorArgs(clientSet, informerFactory, stopCh)
-	configurator := factory.NewConfigFactory(configuratorArgs)
+	recorder := evtBroadcaster.NewRecorder(
+		legacyscheme.Scheme,
+		v1.DefaultSchedulerName,
+	)
 
-	config, err := configurator.CreateFromConfig(schedulerapi.Policy{})
+	sched, err := createScheduler(clientSet, informerFactory, recorder, stopCh)
 	if err != nil {
 		klog.Fatalf("Error creating scheduler: %v", err)
 	}
-	config.Recorder = evtBroadcaster.NewRecorder(legacyscheme.Scheme, "scheduler")
-
-	sched := scheduler.NewFromConfig(config)
 	scheduler.AddAllEventHandlers(sched,
 		v1.DefaultSchedulerName,
 		informerFactory.Core().V1().Nodes(),
@@ -96,32 +94,38 @@ func StartScheduler(clientSet clientset.Interface) (*factory.ConfigFactoryArgs,
 		close(stopCh)
 		klog.Infof("destroyed scheduler")
 	}
-	return configuratorArgs, shutdownFunc
+	return sched, shutdownFunc
 }
 
-// createSchedulerConfigurator create a configurator for scheduler with given informer factory.
-func createSchedulerConfiguratorArgs(
+// createScheduler create a scheduler with given informer factory and default name.
+func createScheduler(
 	clientSet clientset.Interface,
 	informerFactory informers.SharedInformerFactory,
+	recorder events.EventRecorder,
 	stopCh <-chan struct{},
-) *factory.ConfigFactoryArgs {
+) (*scheduler.Scheduler, error) {
+	defaultProviderName := schedulerconfig.SchedulerDefaultProviderName
 
-	return &factory.ConfigFactoryArgs{
-		Client:                         clientSet,
-		NodeInformer:                   informerFactory.Core().V1().Nodes(),
-		PodInformer:                    informerFactory.Core().V1().Pods(),
-		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(),
-		HardPodAffinitySymmetricWeight: v1.DefaultHardPodAffinitySymmetricWeight,
-		DisablePreemption:              false,
-		PercentageOfNodesToScore:       schedulerapi.DefaultPercentageOfNodesToScore,
-		StopCh:                         stopCh,
-	}
+	return scheduler.New(
+		clientSet,
+		informerFactory.Core().V1().Nodes(),
+		informerFactory.Core().V1().Pods(),
+		informerFactory.Core().V1().PersistentVolumes(),
+		informerFactory.Core().V1().PersistentVolumeClaims(),
+		informerFactory.Core().V1().ReplicationControllers(),
+		informerFactory.Apps().V1().ReplicaSets(),
+		informerFactory.Apps().V1().StatefulSets(),
+		informerFactory.Core().V1().Services(),
+		informerFactory.Policy().V1beta1().PodDisruptionBudgets(),
+		informerFactory.Storage().V1().StorageClasses(),
+		informerFactory.Storage().V1beta1().CSINodes(),
+		recorder,
+		schedulerconfig.SchedulerAlgorithmSource{
+			Provider: &defaultProviderName,
+		},
+		stopCh,
+		schedulerframework.NewRegistry(),
+		nil,
+		[]schedulerconfig.PluginConfig{},
+	)
 }