add: enable-pull-scheduler-estimator-event flag

Signed-off-by: prodan <pengshihaoren@gmail.com>

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
+	// EnablePullSchedulerEstimatorEvent represents whether the pull mode enabled the accurate scheduler estimator event.
+	EnablePullSchedulerEstimatorEvent 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.EnablePullSchedulerEstimatorEvent, "enable-pull-scheduler-estimator-event", true, "Enable scheduler estimator event handler in pull mode.")
 	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

@@ -118,6 +118,7 @@ func run(opts *options.Options, stopChan <-chan struct{}, registryOptions ...Opt
 		scheduler.WithSchedulerEstimatorPort(opts.SchedulerEstimatorPort),
 		scheduler.WithSchedulerEstimatorTimeout(opts.SchedulerEstimatorTimeout),
 		scheduler.WithEnableEmptyWorkloadPropagation(opts.EnableEmptyWorkloadPropagation),
+		scheduler.WithEnablePullSchedulerEstimatorEvent(opts.EnablePullSchedulerEstimatorEvent),
 	)
 	if err != nil {
 		return fmt.Errorf("couldn't create scheduler: %w", err)

pkg/scheduler/event_handler.go

@@ -190,7 +190,7 @@ func (s *Scheduler) addCluster(obj interface{}) {
 	}
 	klog.V(3).Infof("Add event for cluster %s", cluster.Name)
 
-	if s.enableSchedulerEstimator {
+	if s.enableSchedulerEstimatorEvent(cluster.Spec.SyncMode) {
 		s.schedulerEstimatorWorker.Add(cluster.Name)
 	}
 }
@@ -245,7 +245,7 @@ func (s *Scheduler) updateCluster(_, newObj interface{}) {
 	}
 	klog.V(3).Infof("Update event for cluster %s", newCluster.Name)
 
-	if s.enableSchedulerEstimator {
+	if s.enableSchedulerEstimatorEvent(newCluster.Spec.SyncMode) {
 		s.schedulerEstimatorWorker.Add(newCluster.Name)
 	}
 
@@ -283,7 +283,15 @@ func (s *Scheduler) deleteCluster(obj interface{}) {
 	s.enqueueAffectedBinding(cluster.Name)
 	s.enqueueAffectedClusterBinding(cluster.Name)
 
-	if s.enableSchedulerEstimator {
+	if s.enableSchedulerEstimatorEvent(cluster.Spec.SyncMode) {
 		s.schedulerEstimatorWorker.Add(cluster.Name)
 	}
 }
+
+func (s *Scheduler) enableSchedulerEstimatorEvent(clusterSyncMode clusterv1alpha1.ClusterSyncMode) bool {
+	if s.enableSchedulerEstimator && clusterSyncMode == clusterv1alpha1.Pull {
+		return s.enablePullSchedulerEstimatorEvent
+	}
+
+	return s.enableSchedulerEstimator
+}

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
+	enablePullSchedulerEstimatorEvent 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
+	// enablePullSchedulerEstimatorEvent represents whether the pull mode enabled the accurate scheduler estimator event.
+	enablePullSchedulerEstimatorEvent 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 {
 	}
 }
 
+// WithEnablePullSchedulerEstimatorEvent sets the enablePullSchedulerEstimatorEvent for scheduler
+func WithEnablePullSchedulerEstimatorEvent(enablePullSchedulerEstimatorEvent bool) Option {
+	return func(o *schedulerOptions) {
+		o.enablePullSchedulerEstimatorEvent = enablePullSchedulerEstimatorEvent
+	}
+}
+
 // 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.enablePullSchedulerEstimatorEvent = options.enablePullSchedulerEstimatorEvent
 		sched.schedulerEstimatorPort = options.schedulerEstimatorPort
 		sched.schedulerEstimatorCache = estimatorclient.NewSchedulerEstimatorCache()
 		schedulerEstimatorWorkerOptions := util.Options{