Merge pull request #2064 from prodanlabs/fix-scheduler

karmada-scheduler add disable-scheduler-estimator-in-pull-mode flag

cmd/scheduler/app/options/options.go

@@ -41,6 +41,8 @@ type Options struct {
 
 	// EnableSchedulerEstimator represents whether the accurate scheduler estimator should be enabled.
 	EnableSchedulerEstimator bool
+	// DisableSchedulerEstimatorInPullMode represents whether to disable the scheduler estimator in pull mode.
+	DisableSchedulerEstimatorInPullMode bool
 	// SchedulerEstimatorTimeout specifies the timeout period of calling the accurate scheduler estimator service.
 	SchedulerEstimatorTimeout metav1.Duration
 	// SchedulerEstimatorPort is the port that the accurate scheduler estimator server serves at.
@@ -80,6 +82,7 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.Float32Var(&o.KubeAPIQPS, "kube-api-qps", 40.0, "QPS to use while talking with karmada-apiserver. Doesn't cover events and node heartbeat apis which rate limiting is controlled by a different set of flags.")
 	fs.IntVar(&o.KubeAPIBurst, "kube-api-burst", 60, "Burst to use while talking with karmada-apiserver. Doesn't cover events and node heartbeat apis which rate limiting is controlled by a different set of flags.")
 	fs.BoolVar(&o.EnableSchedulerEstimator, "enable-scheduler-estimator", false, "Enable calling cluster scheduler estimator for adjusting replicas.")
+	fs.BoolVar(&o.DisableSchedulerEstimatorInPullMode, "disable-scheduler-estimator-in-pull-mode", false, "Disable the scheduler estimator for clusters in pull mode, which takes effect only when enable-scheduler-estimator is true.")
 	fs.DurationVar(&o.SchedulerEstimatorTimeout.Duration, "scheduler-estimator-timeout", 3*time.Second, "Specifies the timeout period of calling the scheduler estimator service.")
 	fs.IntVar(&o.SchedulerEstimatorPort, "scheduler-estimator-port", defaultEstimatorPort, "The secure port on which to connect the accurate scheduler estimator.")
 	fs.BoolVar(&o.EnableEmptyWorkloadPropagation, "enable-empty-workload-propagation", false, "Enable workload with replicas 0 to be propagated to member clusters.")

cmd/scheduler/app/scheduler.go

@@ -115,6 +115,7 @@ func run(opts *options.Options, stopChan <-chan struct{}, registryOptions ...Opt
 	sched, err := scheduler.NewScheduler(dynamicClientSet, karmadaClient, kubeClientSet,
 		scheduler.WithOutOfTreeRegistry(outOfTreeRegistry),
 		scheduler.WithEnableSchedulerEstimator(opts.EnableSchedulerEstimator),
+		scheduler.WithDisableSchedulerEstimatorInPullMode(opts.DisableSchedulerEstimatorInPullMode),
 		scheduler.WithSchedulerEstimatorPort(opts.SchedulerEstimatorPort),
 		scheduler.WithSchedulerEstimatorTimeout(opts.SchedulerEstimatorTimeout),
 		scheduler.WithEnableEmptyWorkloadPropagation(opts.EnableEmptyWorkloadPropagation),

pkg/scheduler/scheduler.go

@@ -80,17 +80,20 @@ type Scheduler struct {
 
 	eventRecorder record.EventRecorder
 
-	enableSchedulerEstimator bool
-	schedulerEstimatorCache  *estimatorclient.SchedulerEstimatorCache
-	schedulerEstimatorPort   int
-	schedulerEstimatorWorker util.AsyncWorker
+	enableSchedulerEstimator            bool
+	disableSchedulerEstimatorInPullMode bool
+	schedulerEstimatorCache             *estimatorclient.SchedulerEstimatorCache
+	schedulerEstimatorPort              int
+	schedulerEstimatorWorker            util.AsyncWorker
 
 	enableEmptyWorkloadPropagation bool
 }
 
 type schedulerOptions struct {
 	// enableSchedulerEstimator represents whether the accurate scheduler estimator should be enabled.
 	enableSchedulerEstimator bool
+	// disableSchedulerEstimatorInPullMode represents whether to disable the scheduler estimator in pull mode.
+	disableSchedulerEstimatorInPullMode bool
 	// schedulerEstimatorTimeout specifies the timeout period of calling the accurate scheduler estimator service.
 	schedulerEstimatorTimeout metav1.Duration
 	// schedulerEstimatorPort is the port that the accurate scheduler estimator server serves at.
@@ -111,6 +114,13 @@ func WithEnableSchedulerEstimator(enableSchedulerEstimator bool) Option {
 	}
 }
 
+// WithDisableSchedulerEstimatorInPullMode sets the disableSchedulerEstimatorInPullMode for scheduler
+func WithDisableSchedulerEstimatorInPullMode(disableSchedulerEstimatorInPullMode bool) Option {
+	return func(o *schedulerOptions) {
+		o.disableSchedulerEstimatorInPullMode = disableSchedulerEstimatorInPullMode
+	}
+}
+
 // WithSchedulerEstimatorTimeout sets the schedulerEstimatorTimeout for scheduler
 func WithSchedulerEstimatorTimeout(schedulerEstimatorTimeout metav1.Duration) Option {
 	return func(o *schedulerOptions) {
@@ -183,6 +193,7 @@ func NewScheduler(dynamicClient dynamic.Interface, karmadaClient karmadaclientse
 
 	if options.enableSchedulerEstimator {
 		sched.enableSchedulerEstimator = options.enableSchedulerEstimator
+		sched.disableSchedulerEstimatorInPullMode = options.disableSchedulerEstimatorInPullMode
 		sched.schedulerEstimatorPort = options.schedulerEstimatorPort
 		sched.schedulerEstimatorCache = estimatorclient.NewSchedulerEstimatorCache()
 		schedulerEstimatorWorkerOptions := util.Options{
@@ -575,14 +586,18 @@ func (s *Scheduler) reconcileEstimatorConnection(key util.QueueKey) error {
 		return fmt.Errorf("failed to reconcile estimator connection as invalid key: %v", key)
 	}
 
-	_, err := s.clusterLister.Get(name)
+	cluster, err := s.clusterLister.Get(name)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			s.schedulerEstimatorCache.DeleteCluster(name)
 			return nil
 		}
 		return err
 	}
+	if cluster.Spec.SyncMode == clusterv1alpha1.Pull && s.disableSchedulerEstimatorInPullMode {
+		return nil
+	}
+
 	return estimatorclient.EstablishConnection(name, s.schedulerEstimatorCache, s.schedulerEstimatorPort)
 }
 
@@ -593,6 +608,9 @@ func (s *Scheduler) establishEstimatorConnections() {
 		return
 	}
 	for i := range clusterList.Items {
+		if clusterList.Items[i].Spec.SyncMode == clusterv1alpha1.Pull && s.disableSchedulerEstimatorInPullMode {
+			continue
+		}
 		if err = estimatorclient.EstablishConnection(clusterList.Items[i].Name, s.schedulerEstimatorCache, s.schedulerEstimatorPort); err != nil {
 			klog.Error(err)
 		}