Repository: clusternet/clusternet
Pull request #395 (2022-06-25 14:52:09)

add scale functionality to dynamic assigner
<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?

#### What this PR does / why we need it:

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
Fixes #

#### Special notes for your reviewer:



Patch:
diff --git a/deploy/agent/clusternet_agent_deployment.yaml b/deploy/agent/clusternet_agent_deployment.yaml
index 63224e1..f8a5cc1 100644
--- a/deploy/agent/clusternet_agent_deployment.yaml
+++ b/deploy/agent/clusternet_agent_deployment.yaml
@@ -58,12 +58,6 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: metadata.namespace
-            # Set the CLUSTER_CIDR and SERVICE_CIDR when deploy agent with k8s control-plane not running as static pod
-            # for example，k3s and rke  
-            # - name: CLUSTER_CIDR
-            #   value: 10.233.12.0/18
-            # - name: SERVICE_CIDR
-            #   value: 10.235.12.0/18
           command:
             - /usr/local/bin/clusternet-agent
             - --cluster-reg-token=$(REG_TOKEN)
diff --git a/deploy/scheduler/clusternet_scheduler_rbac.yaml b/deploy/scheduler/clusternet_scheduler_rbac.yaml
index 9628677..80ac140 100644
--- a/deploy/scheduler/clusternet_scheduler_rbac.yaml
+++ b/deploy/scheduler/clusternet_scheduler_rbac.yaml
@@ -35,17 +35,3 @@ subjects:
   - kind: ServiceAccount
     name: clusternet-scheduler
     namespace: clusternet-system
-
----
-apiVersion: rbac.authorization.k8s.io/v1
-kind: ClusterRoleBinding
-metadata:
-  name: clusternet:scheduler
-roleRef:
-  apiGroup: rbac.authorization.k8s.io
-  kind: ClusterRole
-  name: clusternet:system:socketsproxy
-subjects:
-  - kind: ServiceAccount
-    name: clusternet-scheduler
-    namespace: clusternet-system
diff --git a/manifests/crds/clusters.clusternet.io_managedclusters.yaml b/manifests/crds/clusters.clusternet.io_managedclusters.yaml
index 085eda3..d7f2925 100644
--- a/manifests/crds/clusters.clusternet.io_managedclusters.yaml
+++ b/manifests/crds/clusters.clusternet.io_managedclusters.yaml
@@ -282,10 +282,6 @@ spec:
               predictorAddress:
                 description: PredictorAddress shows the predictor address
                 type: string
-              predictorDirectAccess:
-                description: PredictorDirectAccess indicates whether the predictor
-                  can be accessed directly by clusternet-scheduler
-                type: boolean
               predictorEnabled:
                 description: PredictorEnabled indicates whether predictor is enabled.
                 type: boolean
diff --git a/pkg/agent/agent.go b/pkg/agent/agent.go
index 2c319c9..d31eada 100644
--- a/pkg/agent/agent.go
+++ b/pkg/agent/agent.go
@@ -35,7 +35,6 @@ import (
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/leaderelection"
-	"k8s.io/controller-manager/pkg/clientbuilder"
 	"k8s.io/klog/v2"
 	utilpointer "k8s.io/utils/pointer"
 
@@ -73,9 +72,6 @@ type Agent struct {
 	// clientset for child cluster
 	childKubeClientSet kubernetes.Interface
 
-	// clientset for child cluster election
-	childElectionClientSet kubernetes.Interface
-
 	// kube informer factory for child cluster
 	kubeInformerFactory kubeinformers.SharedInformerFactory
 
@@ -90,7 +86,7 @@ type Agent struct {
 
 	deployer *deployer.Deployer
 
-	predictor *predictor.Server
+	predictor *predictor.PredictorServer
 }
 
 // NewAgent returns a new Agent.
@@ -108,16 +104,8 @@ func NewAgent(registrationOpts *ClusterRegistrationOptions, controllerOpts *util
 	if err != nil {
 		return nil, err
 	}
-
 	// create clientset for child cluster
-	clientBuilder := clientbuilder.SimpleControllerClientBuilder{
-		ClientConfig: childKubeConfig,
-	}
-	childKubeClientSet := kubernetes.NewForConfigOrDie(clientBuilder.ConfigOrDie("clusternet-agent-kube-client"))
-	var electionClientSet *kubernetes.Clientset
-	if controllerOpts.LeaderElection.LeaderElect {
-		electionClientSet = kubernetes.NewForConfigOrDie(clientBuilder.ConfigOrDie("clusternet-agent-election-client"))
-	}
+	childKubeClientSet := kubernetes.NewForConfigOrDie(childKubeConfig)
 
 	// create metrics client for child cluster
 	metricClient := metricsv.NewForConfigOrDie(childKubeConfig)
@@ -125,24 +113,17 @@ func NewAgent(registrationOpts *ClusterRegistrationOptions, controllerOpts *util
 	// creates the informer factory
 	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(childKubeClientSet, known.DefaultResync)
 
-	var p *predictor.Server
-	if registrationOpts.serveInternalPredictor {
-		// predictorAddr is in the form "host:port"
-		predictorAddr := strings.TrimLeft(registrationOpts.PredictorAddress, "http://")
-		predictorAddr = strings.TrimLeft(predictorAddr, "https://")
-		p, err = predictor.NewServer(childKubeConfig, childKubeClientSet, kubeInformerFactory, predictorAddr)
-		if err != nil {
-			return nil, err
-		}
+	predictor, err := predictor.NewPredictorServer(childKubeConfig, childKubeClientSet, kubeInformerFactory)
+	if err != nil {
+		return nil, err
 	}
 
 	agent := &Agent{
-		Identity:               identity,
-		childKubeClientSet:     childKubeClientSet,
-		childElectionClientSet: electionClientSet,
-		kubeInformerFactory:    kubeInformerFactory,
-		registrationOptions:    registrationOpts,
-		controllerOptions:      controllerOpts,
+		Identity:            identity,
+		childKubeClientSet:  childKubeClientSet,
+		kubeInformerFactory: kubeInformerFactory,
+		registrationOptions: registrationOpts,
+		controllerOptions:   controllerOpts,
 		statusManager: NewStatusManager(
 			childKubeConfig.Host,
 			registrationOpts,
@@ -154,7 +135,7 @@ func NewAgent(registrationOpts *ClusterRegistrationOptions, controllerOpts *util
 			registrationOpts.ClusterSyncMode,
 			childKubeConfig.Host,
 			controllerOpts.LeaderElection.ResourceNamespace),
-		predictor: p,
+		predictor: predictor,
 	}
 	return agent, nil
 }
@@ -183,7 +164,7 @@ func (agent *Agent) Run(ctx context.Context) error {
 		agent.controllerOptions.LeaderElection.LeaseDuration.Duration,
 		agent.controllerOptions.LeaderElection.RenewDeadline.Duration,
 		agent.controllerOptions.LeaderElection.RetryPeriod.Duration,
-		agent.childElectionClientSet,
+		agent.childKubeClientSet,
 		leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(ctx context.Context) {
 				agent.run(ctx)
@@ -237,11 +218,10 @@ func (agent *Agent) run(ctx context.Context) {
 		}
 	}, time.Duration(0))
 
-	if agent.predictor != nil {
-		go wait.UntilWithContext(ctx, func(ctx context.Context) {
-			agent.predictor.Run(ctx)
-		}, time.Duration(0))
-	}
+	go wait.UntilWithContext(ctx, func(ctx context.Context) {
+		agent.predictor.Run(ctx)
+	}, time.Duration(0))
+
 	<-ctx.Done()
 }
 
diff --git a/pkg/agent/constant.go b/pkg/agent/constant.go
index 29e1fb1..1623826 100644
--- a/pkg/agent/constant.go
+++ b/pkg/agent/constant.go
@@ -52,15 +52,6 @@ const (
 
 	// UseMetricsServer flag specifies whether to collect metrics from metrics server running in the cluster
 	UseMetricsServer = "use-metrics-server"
-
-	// PredictorAddress flag specifies the address of external predictor
-	PredictorAddress = "predictor-addr"
-
-	// PredictorDirectAccess flag indicates whether the predictor can be accessed directly by clusternet-scheduler
-	PredictorDirectAccess = "predictor-direct-access"
-
-	// PredictorPort flag specifies the port on which to serve built-in predictor
-	PredictorPort = "predictor-port"
 )
 
 // default values
diff --git a/pkg/agent/option.go b/pkg/agent/option.go
index 4e2f825..0ea1500 100644
--- a/pkg/agent/option.go
+++ b/pkg/agent/option.go
@@ -58,12 +58,6 @@ type ClusterRegistrationOptions struct {
 
 	// PredictorAddress specifies the address of predictor
 	PredictorAddress string
-	// PredictorDirectAccess indicates whether the predictor can be accessed directly by clusternet-scheduler
-	PredictorDirectAccess bool
-	// PredictorPort specifies the port on which to serve built-in predictor
-	PredictorPort int
-	// serveInternalPredictor indicates whether to serve built-in predictor. It is not a flag.
-	serveInternalPredictor bool
 
 	// UseMetricsServer specifies whether to collect metrics from metrics server
 	UseMetricsServer bool
@@ -79,8 +73,6 @@ func NewClusterRegistrationOptions() *ClusterRegistrationOptions {
 		ClusterSyncMode:               string(clusterapi.Pull),
 		ClusterStatusReportFrequency:  metav1.Duration{Duration: DefaultClusterStatusReportFrequency},
 		ClusterStatusCollectFrequency: metav1.Duration{Duration: DefaultClusterStatusCollectFrequency},
-		PredictorPort:                 8080,
-		PredictorDirectAccess:         false,
 	}
 }
 
@@ -109,13 +101,9 @@ func (opts *ClusterRegistrationOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.DurationVar(&opts.ClusterStatusCollectFrequency.Duration, ClusterStatusCollectFrequency, opts.ClusterStatusCollectFrequency.Duration,
 		"Specifies how often the agent collects current child cluster status")
 	fs.BoolVar(&opts.TunnelLogging, "enable-tunnel-logging", opts.TunnelLogging, "Enable tunnel logging")
-	fs.BoolVar(&opts.UseMetricsServer, UseMetricsServer, opts.UseMetricsServer, "Use metrics server")
-	fs.StringVar(&opts.PredictorAddress, PredictorAddress, opts.PredictorAddress,
-		"Set address of external predictor, such as https://abc.com:8080. If not set, built-in predictor will be used when feature gate 'Predictor' is enabled.")
-	fs.BoolVar(&opts.PredictorDirectAccess, PredictorDirectAccess, opts.PredictorDirectAccess,
-		"Whether the predictor be accessed directly by clusternet-scheduler")
-	fs.IntVar(&opts.PredictorPort, PredictorPort, opts.PredictorPort,
-		"Set port on which to serve built-in predictor server. It is only used when feature gate 'Predictor' is enabled and '--predictor-addr' is not set.")
+	fs.BoolVar(&opts.UseMetricsServer, "use-metrics-server", opts.UseMetricsServer, "Use metrics server")
+	fs.StringVar(&opts.PredictorAddress, "external-predictor-addr", opts.PredictorAddress,
+		"Set address of external predictor. If not set, built-in predictor will be used when feature gate 'Predictor' is enabled.")
 }
 
 // Complete completes all the required options.
@@ -125,11 +113,6 @@ func (opts *ClusterRegistrationOptions) Complete() []error {
 	opts.ClusterName = strings.TrimSpace(opts.ClusterName)
 	opts.ClusterNamePrefix = strings.TrimSpace(opts.ClusterNamePrefix)
 
-	if utilfeature.DefaultFeatureGate.Enabled(features.Predictor) && opts.PredictorAddress == "" {
-		opts.PredictorAddress = fmt.Sprintf("http://localhost:%d", opts.PredictorPort)
-		opts.serveInternalPredictor = true
-	}
-
 	return allErrs
 }
 
diff --git a/pkg/agent/status_manager.go b/pkg/agent/status_manager.go
index 9c71b43..01f6059 100644
--- a/pkg/agent/status_manager.go
+++ b/pkg/agent/status_manager.go
@@ -59,12 +59,11 @@ func NewStatusManager(
 		statusReportFrequency: regOpts.ClusterStatusReportFrequency,
 		clusterStatusController: clusterstatus.NewController(
 			apiserverURL,
+			regOpts.PredictorAddress,
+			regOpts.UseMetricsServer,
 			kubeClient,
 			metricClient,
 			kubeInformerFactory,
-			regOpts.PredictorAddress,
-			regOpts.PredictorDirectAccess,
-			regOpts.UseMetricsServer,
 			regOpts.ClusterStatusCollectFrequency,
 			regOpts.ClusterStatusReportFrequency,
 		),
diff --git a/pkg/apis/clusters/v1beta1/types.go b/pkg/apis/clusters/v1beta1/types.go
index 54bd59c..5c82db5 100644
--- a/pkg/apis/clusters/v1beta1/types.go
+++ b/pkg/apis/clusters/v1beta1/types.go
@@ -289,9 +289,6 @@ type ManagedClusterStatus struct {
 	// PredictorAddress shows the predictor address
 	// +optional
 	PredictorAddress string `json:"predictorAddress,omitempty"`
-
-	// PredictorDirectAccess indicates whether the predictor can be accessed directly by clusternet-scheduler
-	PredictorDirectAccess bool `json:"predictorDirectAccess,omitempty"`
 }
 
 // +genclient
diff --git a/pkg/apis/scheduler/interface.go b/pkg/apis/scheduler/interface.go
index ddaab2f..7407101 100644
--- a/pkg/apis/scheduler/interface.go
+++ b/pkg/apis/scheduler/interface.go
@@ -24,24 +24,6 @@ import (
 	"github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
 )
 
-const (
-	DefaultAcceptableReplicasKey = "default"
-)
-
-// PredictorReplicas indicates a map of label to replicas with this constraint. Here the label constraint
-// could be topology constraints, such as
-// {
-//    "topology.kubernetes.io/zone=zone1,topology.kubernetes.io/region=region1": 3,
-//    "topology.kubernetes.io/zone=zone2,topology.kubernetes.io/region=region1": 5,
-// }.
-type PredictorReplicas map[string]int32
-
-// PredictorResults specify
-type PredictorResults struct {
-	// Replicas records the max acceptable replicas in this cluster
-	Replicas PredictorReplicas `json:"replicas,omitemtpy"`
-}
-
 // PredictorProvider is an interface that provides replicas predictions.
 type PredictorProvider interface {
 	// MaxAcceptableReplicas indicates the maximum acceptable replicas that the cluster could admit.
@@ -51,7 +33,7 @@ type PredictorProvider interface {
 	//    "topology.kubernetes.io/zone=zone1,topology.kubernetes.io/region=region1": 3,
 	//    "topology.kubernetes.io/zone=zone2,topology.kubernetes.io/region=region1": 5,
 	// }.
-	MaxAcceptableReplicas(ctx context.Context, requirements v1alpha1.ReplicaRequirements) (PredictorResults, error)
+	MaxAcceptableReplicas(ctx context.Context, requirements v1alpha1.ReplicaRequirements) (map[string]int32, error)
 
 	// UnschedulableReplicas returns current unschedulable replicas.
 	UnschedulableReplicas(ctx context.Context, gvk metav1.GroupVersionKind, namespacedName string,
@@ -59,14 +41,3 @@ type PredictorProvider interface {
 
 	// TODO: resource reservations
 }
-
-const (
-	// RootPathReplicas specifies the root path for predictor replica server
-	RootPathReplicas = "/replicas"
-
-	// SubPathPredict specifies the sub path for predicting replicas
-	SubPathPredict = "/predict"
-
-	// SubPathUnscheduled specifies the sub path for listing unscheduled replicas
-	SubPathUnscheduled = "/unscheduled"
-)
diff --git a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
index 9f7ef35..49a6916 100644
--- a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
+++ b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
@@ -52,44 +52,40 @@ type Controller struct {
 	apiserverURL       string
 	appPusherEnabled   bool
 	useSocket          bool
+	predictorEnable    bool
 	useMetricsServer   bool
+	predictorAddress   string
 	nodeLister         corev1lister.NodeLister
 	nodeSynced         cache.InformerSynced
 	podLister          corev1lister.PodLister
 	podSynced          cache.InformerSynced
-
-	predictorEnable       bool
-	predictorAddress      string
-	predictorDirectAccess bool
 }
 
 func NewController(
-	apiserverURL string,
+	apiserverURL, predictorAddress string,
+	useMetricsServer bool,
 	kubeClient kubernetes.Interface,
 	metricClient *metricsv.Clientset,
 	kubeInformerFactory informers.SharedInformerFactory,
-	predictorAddress string,
-	predictorDirectAccess, useMetricsServer bool,
 	collectingPeriod metav1.Duration,
 	heartbeatFrequency metav1.Duration,
 ) *Controller {
 	return &Controller{
-		kubeClient:            kubeClient,
-		metricClientset:       metricClient,
-		lock:                  &sync.Mutex{},
-		collectingPeriod:      collectingPeriod,
-		heartbeatFrequency:    heartbeatFrequency,
-		apiserverURL:          apiserverURL,
-		appPusherEnabled:      utilfeature.DefaultFeatureGate.Enabled(features.AppPusher),
-		useSocket:             utilfeature.DefaultFeatureGate.Enabled(features.SocketConnection),
-		predictorEnable:       utilfeature.DefaultFeatureGate.Enabled(features.Predictor),
-		useMetricsServer:      useMetricsServer,
-		predictorAddress:      predictorAddress,
-		predictorDirectAccess: predictorDirectAccess,
-		nodeLister:            kubeInformerFactory.Core().V1().Nodes().Lister(),
-		nodeSynced:            kubeInformerFactory.Core().V1().Nodes().Informer().HasSynced,
-		podLister:             kubeInformerFactory.Core().V1().Pods().Lister(),
-		podSynced:             kubeInformerFactory.Core().V1().Pods().Informer().HasSynced,
+		kubeClient:         kubeClient,
+		metricClientset:    metricClient,
+		lock:               &sync.Mutex{},
+		collectingPeriod:   collectingPeriod,
+		heartbeatFrequency: heartbeatFrequency,
+		apiserverURL:       apiserverURL,
+		appPusherEnabled:   utilfeature.DefaultFeatureGate.Enabled(features.AppPusher),
+		useSocket:          utilfeature.DefaultFeatureGate.Enabled(features.SocketConnection),
+		predictorEnable:    utilfeature.DefaultFeatureGate.Enabled(features.Predictor),
+		useMetricsServer:   useMetricsServer,
+		predictorAddress:   predictorAddress,
+		nodeLister:         kubeInformerFactory.Core().V1().Nodes().Lister(),
+		nodeSynced:         kubeInformerFactory.Core().V1().Nodes().Informer().HasSynced,
+		podLister:          kubeInformerFactory.Core().V1().Pods().Lister(),
+		podSynced:          kubeInformerFactory.Core().V1().Pods().Informer().HasSynced,
 	}
 }
 
@@ -157,7 +153,6 @@ func (c *Controller) collectingClusterStatus(ctx context.Context) {
 	status.Conditions = []metav1.Condition{c.getCondition(status)}
 	status.PredictorEnabled = c.predictorEnable
 	status.PredictorAddress = c.predictorAddress
-	status.PredictorDirectAccess = c.predictorDirectAccess
 	c.setClusterStatus(status)
 }
 
diff --git a/pkg/controllers/clusters/clusterstatus/network.go b/pkg/controllers/clusters/clusterstatus/network.go
index a54bbae..8b07e58 100644
--- a/pkg/controllers/clusters/clusterstatus/network.go
+++ b/pkg/controllers/clusters/clusterstatus/network.go
@@ -18,7 +18,6 @@ package clusterstatus
 
 import (
 	"errors"
-	"os"
 	"strings"
 
 	corev1 "k8s.io/api/core/v1"
@@ -35,11 +34,6 @@ func findServiceIPRange(podLister corev1Lister.PodLister) (string, error) {
 	if clusterIPRange != "" {
 		return clusterIPRange, nil
 	}
-	// Try to find the pod IP range from the env.
-	clusterIPRange = os.Getenv("SERVICE_CIDR")
-	if clusterIPRange != "" {
-		return clusterIPRange, nil
-	}
 	return "", errors.New("can't get ServiceIPRange")
 }
 
@@ -63,11 +57,7 @@ func findPodIPRange(nodeLister corev1Lister.NodeLister, podLister corev1Lister.P
 	if podIPRange != "" {
 		return podIPRange, nil
 	}
-	// Try to find the pod IP range from the env.
-	podIPRange = os.Getenv("CLUSTER_CIDR")
-	if podIPRange != "" {
-		return podIPRange, nil
-	}
+
 	return "", errors.New("can't get PodIPRange")
 }
 
diff --git a/pkg/features/features.go b/pkg/features/features.go
index a95b37b..e6c2eb3 100644
--- a/pkg/features/features.go
+++ b/pkg/features/features.go
@@ -81,7 +81,7 @@ const (
 	FeedInventory featuregate.Feature = "FeedInventory"
 
 	// Predictor will predictor child cluster resource before scheduling.
-	// This feature gate needs a running predictor, either build-in or external.
+	// This feature gate need a running predictor , build-in agent or external predictor
 	//
 	// owner: @yinsenyan
 	// alpha: v0.10.0
diff --git a/pkg/hub/hub.go b/pkg/hub/hub.go
index 2dd9d3d..7c2717b 100644
--- a/pkg/hub/hub.go
+++ b/pkg/hub/hub.go
@@ -83,7 +83,6 @@ type Hub struct {
 	aggregatorInformerFactory aggregatorinformers.SharedInformerFactory
 
 	kubeClient       *kubernetes.Clientset
-	electionClient   *kubernetes.Clientset
 	clusternetClient *clusternet.Clientset
 	clientBuilder    clientbuilder.ControllerClientBuilder
 
@@ -114,11 +113,6 @@ func NewHub(opts *options.HubServerOptions) (*Hub, error) {
 	kubeClient := kubernetes.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-hub-kube-client"))
 	clusternetClient := clusternet.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-hub-client"))
 
-	var electionClient *kubernetes.Clientset
-	if opts.LeaderElection.LeaderElect {
-		electionClient = kubernetes.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-hub-election-client"))
-	}
-
 	//deployer.broadcaster.StartStructuredLogging(5)
 	broadcaster := record.NewBroadcaster()
 	if kubeClient != nil {
@@ -162,7 +156,6 @@ func NewHub(opts *options.HubServerOptions) (*Hub, error) {
 		crrApprover:               approver,
 		options:                   opts,
 		kubeClient:                kubeClient,
-		electionClient:            electionClient,
 		clusternetClient:          clusternetClient,
 		clientBuilder:             rootClientBuilder,
 		clusternetInformerFactory: clusternetInformerFactory,
@@ -211,7 +204,7 @@ func (hub *Hub) Run(ctx context.Context) error {
 		hub.options.LeaderElection.LeaseDuration.Duration,
 		hub.options.LeaderElection.RenewDeadline.Duration,
 		hub.options.LeaderElection.RetryPeriod.Duration,
-		hub.electionClient,
+		hub.kubeClient,
 		leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(ctx context.Context) {
 				hub.runControllers(ctx)
@@ -244,7 +237,7 @@ func (hub *Hub) Run(ctx context.Context) error {
 			Token:    hub.options.PeerToken,
 		},
 		hub.options.LeaderElection,
-		hub.electionClient,
+		hub.kubeClient,
 	)
 	if err != nil {
 		return err
@@ -347,7 +340,7 @@ func (hub *Hub) Run(ctx context.Context) error {
 			}()
 
 			go leasegc.NewLeaseGC(
-				hub.electionClient,
+				hub.kubeClient,
 				hub.options.LeaderElection.LeaseDuration.Duration,
 				hub.options.LeaderElection.ResourceNamespace,
 				func(lease *coordinationapi.Lease) bool {
@@ -367,7 +360,7 @@ func (hub *Hub) Run(ctx context.Context) error {
 				}
 			}()
 
-			go refreshingPeers(hub.electionClient, hub.peerID, hub.options.LeaderElection.ResourceNamespace,
+			go refreshingPeers(hub.kubeClient, hub.peerID, hub.options.LeaderElection.ResourceNamespace,
 				server.PeerDialer, postStartHookContext.StopCh)
 
 			select {
diff --git a/pkg/predictor/apis/types.go b/pkg/predictor/apis/types.go
index 39ed80a..7812a46 100644
--- a/pkg/predictor/apis/types.go
+++ b/pkg/predictor/apis/types.go
@@ -17,6 +17,8 @@ limitations under the License.
 package apis
 
 import (
+	"math"
+
 	"k8s.io/apimachinery/pkg/util/sets"
 )
 
@@ -73,6 +75,25 @@ type Plugin struct {
 	Weight int32
 }
 
+/*
+ * NOTE: The following variables and methods are intentionally left out of the staging mirror.
+ */
+const (
+	// DefaultPercentageOfNodesToScore defines the percentage of nodes of all nodes
+	// that once found feasible, the predictor stops looking for more nodes.
+	// A value of 0 means adaptive, meaning the predictor figures out a proper default.
+	DefaultPercentageOfNodesToScore = 0
+
+	// MaxCustomPriorityScore is the max score UtilizationShapePoint expects.
+	MaxCustomPriorityScore int64 = 10
+
+	// MaxTotalScore is the maximum total score.
+	MaxTotalScore int64 = math.MaxInt64
+
+	// MaxWeight defines the max weight value allowed for custom PriorityPolicy
+	MaxWeight = MaxTotalScore / MaxCustomPriorityScore
+)
+
 // Names returns the list of enabled plugin names.
 func (p *Plugins) Names() []string {
 	if p == nil {
diff --git a/pkg/predictor/framework/interfaces/interface.go b/pkg/predictor/framework/interfaces/interface.go
index 8708c60..8c1823e 100644
--- a/pkg/predictor/framework/interfaces/interface.go
+++ b/pkg/predictor/framework/interfaces/interface.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/client-go/tools/record"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	schedulerapi "github.com/clusternet/clusternet/pkg/apis/scheduler"
 	predictorapis "github.com/clusternet/clusternet/pkg/predictor/apis"
 	"github.com/clusternet/clusternet/pkg/scheduler/parallelize"
 )
@@ -227,7 +226,7 @@ type AggregatePlugin interface {
 	// Aggregate is called on all selected nodes. It will calculate the sum of acceptable
 	// replicas for selected nodes. The return result is a map whose key indicates the
 	// specific calculation strategy and value is max acceptable replicas.
-	Aggregate(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores NodeScoreList) (schedulerapi.PredictorReplicas, *Status)
+	Aggregate(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores NodeScoreList) (AcceptableReplicas, *Status)
 }
 
 // Framework manages the set of plugins in use by the predictor framework.
@@ -268,7 +267,7 @@ type Framework interface {
 	// whose key indicates the specific calculation strategy and value is max acceptable replicas
 	// It also returns *Status, which is set to non-success if any of the plugins returns
 	// a non-success status.
-	RunAggregatePlugins(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores NodeScoreList) (schedulerapi.PredictorReplicas, *Status)
+	RunAggregatePlugins(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores NodeScoreList) (AcceptableReplicas, *Status)
 
 	// HasFilterPlugins returns true if at least one Filter plugin is defined.
 	HasFilterPlugins() bool
diff --git a/pkg/predictor/framework/interfaces/types.go b/pkg/predictor/framework/interfaces/types.go
index 7b1f4da..6acaf8d 100644
--- a/pkg/predictor/framework/interfaces/types.go
+++ b/pkg/predictor/framework/interfaces/types.go
@@ -20,6 +20,11 @@ import (
 	v1 "k8s.io/api/core/v1"
 )
 
+const DefaultAcceptableReplicasKey = "default"
+
+// AcceptableReplicas records the max acceptable replicas in this cluster
+type AcceptableReplicas map[string]int32
+
 // NodeInfo is node level aggregated information.
 type NodeInfo struct {
 	// Overall node information.
diff --git a/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate.go b/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate.go
index 89c3d06..e0b6592 100644
--- a/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate.go
+++ b/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate.go
@@ -23,7 +23,6 @@ import (
 	"k8s.io/klog/v2"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	schedulerapi "github.com/clusternet/clusternet/pkg/apis/scheduler"
 	framework "github.com/clusternet/clusternet/pkg/predictor/framework/interfaces"
 	"github.com/clusternet/clusternet/pkg/predictor/framework/plugins/names"
 )
@@ -46,15 +45,15 @@ func (pl *DefaultAggregator) Name() string {
 }
 
 // Aggregate aggregate acceptable replicas for all filtered node
-func (pl *DefaultAggregator) Aggregate(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores framework.NodeScoreList) (schedulerapi.PredictorReplicas, *framework.Status) {
+func (pl *DefaultAggregator) Aggregate(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores framework.NodeScoreList) (framework.AcceptableReplicas, *framework.Status) {
 	klog.V(5).InfoS("Attempting to Aggregate replicas")
-	result := make(schedulerapi.PredictorReplicas)
+	result := make(framework.AcceptableReplicas)
 	maxAcceptableReplicas := int32(0)
 	for i := range scores {
 		maxAcceptableReplicas += scores[i].MaxAvailableReplicas
 	}
 	// Todo : support topology-aware replicas
-	result[schedulerapi.DefaultAcceptableReplicasKey] = maxAcceptableReplicas
+	result[framework.DefaultAcceptableReplicasKey] = maxAcceptableReplicas
 
 	return result, nil
 }
diff --git a/pkg/predictor/framework/runtime/framework.go b/pkg/predictor/framework/runtime/framework.go
index 656f894..79bfe08 100644
--- a/pkg/predictor/framework/runtime/framework.go
+++ b/pkg/predictor/framework/runtime/framework.go
@@ -30,7 +30,6 @@ import (
 	"k8s.io/client-go/tools/record"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	schedulerapi "github.com/clusternet/clusternet/pkg/apis/scheduler"
 	predictorapis "github.com/clusternet/clusternet/pkg/predictor/apis"
 	framework "github.com/clusternet/clusternet/pkg/predictor/framework/interfaces"
 	"github.com/clusternet/clusternet/pkg/predictor/metrics"
@@ -543,7 +542,7 @@ func (f *frameworkImpl) runPreAggregatePlugin(ctx context.Context, pl framework.
 	return status
 }
 
-func (f *frameworkImpl) RunAggregatePlugins(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores framework.NodeScoreList) (result schedulerapi.PredictorReplicas, status *framework.Status) {
+func (f *frameworkImpl) RunAggregatePlugins(ctx context.Context, requirements *appsapi.ReplicaRequirements, scores framework.NodeScoreList) (result framework.AcceptableReplicas, status *framework.Status) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(aggregate, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
@@ -551,7 +550,7 @@ func (f *frameworkImpl) RunAggregatePlugins(ctx context.Context, requirements *a
 	if len(f.aggregatePlugins) == 0 {
 		return nil, framework.NewStatus(framework.Skip, "empty aggregate plugins")
 	}
-	result = make(schedulerapi.PredictorReplicas)
+	result = make(framework.AcceptableReplicas)
 	for _, ap := range f.aggregatePlugins {
 		aggreateResult, pluginStatus := f.runAggregatePlugin(ctx, ap, requirements, scores)
 		if pluginStatus != nil && pluginStatus.Code() == framework.Skip {
@@ -570,7 +569,7 @@ func (f *frameworkImpl) RunAggregatePlugins(ctx context.Context, requirements *a
 	return result, nil
 }
 
-func (f *frameworkImpl) runAggregatePlugin(ctx context.Context, ap framework.AggregatePlugin, requirements *appsapi.ReplicaRequirements, scores framework.NodeScoreList) (result schedulerapi.PredictorReplicas, status *framework.Status) {
+func (f *frameworkImpl) runAggregatePlugin(ctx context.Context, ap framework.AggregatePlugin, requirements *appsapi.ReplicaRequirements, scores framework.NodeScoreList) (result framework.AcceptableReplicas, status *framework.Status) {
 	startTime := time.Now()
 	result, status = ap.Aggregate(ctx, requirements, scores)
 	f.metricsRecorder.observePluginDurationAsync(aggregate, ap.Name(), status, metrics.SinceInSeconds(startTime))
diff --git a/pkg/predictor/predictor.go b/pkg/predictor/predictor.go
index 058c349..d270ecb 100644
--- a/pkg/predictor/predictor.go
+++ b/pkg/predictor/predictor.go
@@ -26,7 +26,6 @@ import (
 	"k8s.io/klog/v2"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	schedulerapi "github.com/clusternet/clusternet/pkg/apis/scheduler"
 	framework "github.com/clusternet/clusternet/pkg/predictor/framework/interfaces"
 	frameworkruntime "github.com/clusternet/clusternet/pkg/predictor/framework/runtime"
 	"github.com/clusternet/clusternet/pkg/predictor/metrics"
@@ -224,7 +223,7 @@ func aggregateReplicas(
 	fwk framework.Framework,
 	requirements *appsapi.ReplicaRequirements,
 	result framework.NodeScoreList,
-) (schedulerapi.PredictorReplicas, error) {
+) (framework.AcceptableReplicas, error) {
 	if !fwk.HasAggregatePlugins() {
 		return nil, fmt.Errorf("no calculate plugin is registry")
 	}
diff --git a/pkg/predictor/server.go b/pkg/predictor/server.go
index 98eaf48..fe3ed43 100644
--- a/pkg/predictor/server.go
+++ b/pkg/predictor/server.go
@@ -18,17 +18,14 @@ package predictor
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
-	"io/ioutil"
-	"net/http"
 	"sync/atomic"
 
-	"github.com/emicklei/go-restful"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/informers"
 	kubeinformers "k8s.io/client-go/informers"
 	informerv1 "k8s.io/client-go/informers/core/v1"
 	"k8s.io/client-go/kubernetes"
@@ -57,10 +54,10 @@ const (
 // and returns the pods that assigned to the node.
 type GetPodsAssignedToNodeFunc func(string) ([]*corev1.Pod, error)
 
-type Server struct {
+type PredictorServer struct {
 	kubeClient kubernetes.Interface
 
-	informerFactory kubeinformers.SharedInformerFactory
+	informerFactory informers.SharedInformerFactory
 	nodeInformer    informerv1.NodeInformer
 	podInformer     informerv1.PodInformer
 	nodeLister      listerv1.NodeLister
@@ -71,17 +68,13 @@ type Server struct {
 	// default in-tree registry
 	registry  frameworkruntime.Registry
 	framework framework.Framework
-
-	address     string
-	restfulCont *restful.Container
 }
 
-func NewServer(
+func NewPredictorServer(
 	clientConfig *restclient.Config,
 	kubeClient kubernetes.Interface,
 	informerFactory kubeinformers.SharedInformerFactory,
-	addr string,
-) (*Server, error) {
+) (*PredictorServer, error) {
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
 		Interface: kubeClient.CoreV1().Events(""),
@@ -108,8 +101,7 @@ func NewServer(
 		return nil, err
 	}
 
-	return &Server{
-		address:         addr,
+	ps := &PredictorServer{
 		kubeClient:      kubeClient,
 		informerFactory: informerFactory,
 		nodeInformer:    informerFactory.Core().V1().Nodes(),
@@ -119,45 +111,24 @@ func NewServer(
 		registry:        registry,
 		framework:       framework,
 		getPodFunc:      getPodFunc,
-		restfulCont:     restful.NewContainer(),
-	}, nil
+	}
+	return ps, nil
 }
 
-var _ schedulerapi.PredictorProvider = &Server{}
+var _ schedulerapi.PredictorProvider = &PredictorServer{}
 
-func (s *Server) Run(ctx context.Context) {
+func (ps *PredictorServer) Run(ctx context.Context) {
 	klog.Infof("Starting clusternet predictor")
 	defer klog.Infof("Shutting down clusternet predictor")
 
 	// Wait for all caches to sync before scheduling.
-	if !cache.WaitForCacheSync(ctx.Done(), s.podInformer.Informer().HasSynced, s.nodeInformer.Informer().HasSynced) {
+	if !cache.WaitForCacheSync(ctx.Done(), ps.podInformer.Informer().HasSynced, ps.nodeInformer.Informer().HasSynced) {
 		klog.Errorf("failed to wait for caches to sync")
 		return
 	}
 
-	s.installDefaultHandlers(ctx)
-
-	srv := http.Server{
-		Addr:              s.address,
-		Handler:           s.restfulCont,
-		TLSConfig:         nil,
-		ReadTimeout:       0,
-		ReadHeaderTimeout: 0,
-		WriteTimeout:      0,
-		IdleTimeout:       0,
-		MaxHeaderBytes:    0,
-		TLSNextProto:      nil,
-		ConnState:         nil,
-		ErrorLog:          nil,
-		BaseContext:       nil,
-		ConnContext:       nil,
-	}
-
-	defer srv.Close()
-	if err := srv.ListenAndServe(); err != nil {
-		// Error starting or closing listener:
-		klog.ErrorS(err, "Failed to listen and serve predictor")
-	}
+	// TODO: add communication interface to scheduler
+	<-ctx.Done()
 }
 
 // BuildGetPodsAssignedToNodeFunc establishes an indexer to map the pods and their assigned nodes.
@@ -200,59 +171,20 @@ func BuildGetPodsAssignedToNodeFunc(podInformer informerv1.PodInformer) (GetPods
 	return getPodsAssignedToNode, nil
 }
 
-// installDefaultHandlers registers the default set of supported HTTP request
-// patterns with the restful Container.
-func (s *Server) installDefaultHandlers(ctx context.Context) {
-	ws := new(restful.WebService)
-	ws.Path(schedulerapi.RootPathReplicas)
-
-	ws.Route(ws.POST(schedulerapi.SubPathPredict).To(func(request *restful.Request, response *restful.Response) {
-		defer request.Request.Body.Close()
-		data, err := ioutil.ReadAll(request.Request.Body)
-		if err != nil {
-			http.Error(response, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		var require appsapi.ReplicaRequirements
-		err = json.Unmarshal(data, &require)
-		if err != nil {
-			http.Error(response, err.Error(), http.StatusBadRequest)
-			return
-		}
+func (ps *PredictorServer) MaxAcceptableReplicas(ctx context.Context, requirements appsapi.ReplicaRequirements) (map[string]int32, error) {
 
-		replicas, err := s.MaxAcceptableReplicas(ctx, require)
-		if err != nil {
-			klog.Warningf("failed to predict max acceptable replicas for %q, error: %v", data, err)
-			http.Error(response, err.Error(), http.StatusInternalServerError)
-			return
-		}
-
-		err = response.WriteAsJson(replicas)
-		if err != nil {
-			http.Error(response, err.Error(), http.StatusInternalServerError)
-			return
-		}
-	}).Operation("predictReplicas"))
-
-	ws.Route(ws.GET(schedulerapi.SubPathUnscheduled).To(func(request *restful.Request, response *restful.Response) {
-		// TODO: unschedulable replicas http handler function logic
-		http.Error(response, "Not Implemented", http.StatusInternalServerError)
-		return
-	}).Operation("getUnscheduledReplicas"))
-
-	s.restfulCont.Add(ws)
-}
+	defaultAcceptableReplicas := map[string]int32{
+		framework.DefaultAcceptableReplicasKey: 0,
+	}
 
-func (s *Server) MaxAcceptableReplicas(ctx context.Context, requirements appsapi.ReplicaRequirements) (schedulerapi.PredictorResults, error) {
-	nodes, err := s.nodeLister.List(labels.SelectorFromSet(requirements.NodeSelector))
+	nodes, err := ps.nodeLister.List(labels.SelectorFromSet(requirements.NodeSelector))
 	if err != nil {
-		return schedulerapi.PredictorResults{}, fmt.Errorf("failed to get nodes that match node selector, err: %v", err)
+		return defaultAcceptableReplicas, fmt.Errorf("failed to get nodes that match node selector, err: %v", err)
 	}
 	nodeInfoList := make([]*framework.NodeInfo, len(nodes))
 	var nodesLen int32
 	getNodeInfo := func(i int) {
-		pods, err := s.getPodFunc(nodes[i].Name)
+		pods, err := ps.getPodFunc(nodes[i].Name)
 		if err != nil {
 			klog.V(6).InfoS("failed to get pods in nodes", "nodes", nodes[i].Name)
 		} else {
@@ -261,39 +193,39 @@ func (s *Server) MaxAcceptableReplicas(ctx context.Context, requirements appsapi
 			nodeInfoList[length-1] = nodeInfo
 		}
 	}
-	s.framework.Parallelizer().Until(ctx, len(nodes), getNodeInfo)
+	ps.framework.Parallelizer().Until(ctx, len(nodes), getNodeInfo)
 
 	// Step 1: Filter Nodes.
-	feasibleNodes, err := findNodesThatFitRequirements(ctx, s.framework, &requirements, nodeInfoList)
+	feasibleNodes, err := findNodesThatFitRequirements(ctx, ps.framework, &requirements, nodeInfoList)
 	if err != nil {
-		return schedulerapi.PredictorResults{}, err
+		return defaultAcceptableReplicas, err
 	}
 
 	if len(feasibleNodes) == 0 {
-		return schedulerapi.PredictorResults{}, fmt.Errorf("no feasible nodes found")
+		return defaultAcceptableReplicas, fmt.Errorf("no feasible nodes found")
 	}
 
 	// Step 2: cal max available replicas for each feasibleNodes.
-	nodeScoreList, err := computeReplicas(ctx, s.framework, &requirements, feasibleNodes)
+	nodeScoreList, err := computeReplicas(ctx, ps.framework, &requirements, feasibleNodes)
 	if err != nil {
-		return schedulerapi.PredictorResults{}, err
+		return defaultAcceptableReplicas, err
 	}
 
 	// Step 3: Prioritize clusters.
-	priorityList, err := prioritizeNodes(ctx, s.framework, &requirements, feasibleNodes, nodeScoreList)
+	priorityList, err := prioritizeNodes(ctx, ps.framework, &requirements, feasibleNodes, nodeScoreList)
 	if err != nil {
-		return schedulerapi.PredictorResults{}, err
+		return defaultAcceptableReplicas, err
 	}
 
 	// step4 aggregate the max available replicas
-	result, err := aggregateReplicas(ctx, s.framework, &requirements, priorityList)
+	result, err := aggregateReplicas(ctx, ps.framework, &requirements, priorityList)
 	if err != nil {
-		return schedulerapi.PredictorResults{}, err
+		return defaultAcceptableReplicas, err
 	}
-	return schedulerapi.PredictorResults{Replicas: result}, nil
+	return result, nil
 }
 
-func (s *Server) UnschedulableReplicas(ctx context.Context, gvk metav1.GroupVersionKind, namespacedName string,
+func (ps *PredictorServer) UnschedulableReplicas(ctx context.Context, gvk metav1.GroupVersionKind, namespacedName string,
 	labelSelector map[string]string) (int32, error) {
 	// TODO: add real logic
 	return 0, nil
diff --git a/pkg/scheduler/algorithm/generic.go b/pkg/scheduler/algorithm/generic.go
index d46c07a..f69c040 100644
--- a/pkg/scheduler/algorithm/generic.go
+++ b/pkg/scheduler/algorithm/generic.go
@@ -172,16 +172,10 @@ func (g *genericScheduler) selectClusters(ctx context.Context, state *framework.
 
 	selected.Replicas = make(map[string][]int32)
 	// transfer to available replicas
-	for index, feedOrder := range finv.Spec.Feeds {
-		if feedOrder.DesiredReplicas == nil {
-			continue
-		}
-		for _, clusterScore := range clusterScoreList {
-			feedKey := utils.GetFeedKey(feedOrder.Feed)
-			if clusterScore.MaxAvailableReplicas[index] == nil {
-				return framework.TargetClusters{}, fmt.Errorf("unable to get replicas for feed %q in cluster %q", feedKey, clusterScore.NamespacedName)
-			}
-			selected.Replicas[feedKey] = append(selected.Replicas[feedKey], *clusterScore.MaxAvailableReplicas[index])
+	for _, clusterScore := range clusterScoreList {
+		for i := range finv.Spec.Feeds {
+			feed := &finv.Spec.Feeds[i].Feed
+			selected.Replicas[utils.GetFeedKey(*feed)] = append(selected.Replicas[utils.GetFeedKey(*feed)], clusterScore.MaxAvailableReplicas[i])
 		}
 	}
 
diff --git a/pkg/scheduler/framework/interfaces/interface.go b/pkg/scheduler/framework/interfaces/interface.go
index d44561b..cf0191f 100644
--- a/pkg/scheduler/framework/interfaces/interface.go
+++ b/pkg/scheduler/framework/interfaces/interface.go
@@ -67,7 +67,7 @@ func (cs ClusterScoreList) ClusterNames() []string {
 	return clusters
 }
 
-type FeedReplicas []*int32
+type FeedReplicas []int32
 
 // PluginToClusterReplicas declares a map from plugin name to its ClusterScoreList.
 type PluginToClusterReplicas map[string]ClusterScoreList
diff --git a/pkg/scheduler/framework/interfaces/types.go b/pkg/scheduler/framework/interfaces/types.go
index a51252a..5f13647 100644
--- a/pkg/scheduler/framework/interfaces/types.go
+++ b/pkg/scheduler/framework/interfaces/types.go
@@ -48,6 +48,10 @@ type TargetClusters struct {
 	Replicas map[string][]int32
 }
 
+func NewTargetClusters(bindingClusters []string, replicas map[string][]int32) *TargetClusters {
+	return &TargetClusters{BindingClusters: bindingClusters, Replicas: replicas}
+}
+
 func (t TargetClusters) Len() int {
 	return len(t.BindingClusters)
 }
@@ -93,6 +97,52 @@ func (t *TargetClusters) DeepCopy() *TargetClusters {
 	return obj
 }
 
+func (t *TargetClusters) Merge(b *TargetClusters) {
+	if b == nil || len(b.BindingClusters) == 0 {
+		return
+	}
+	if t.Replicas == nil {
+		t.Replicas = make(map[string][]int32)
+	}
+	// If the feed replicas does not exist in the former target clusters, we initialize them as an empty one.
+	for feed, replicas := range b.Replicas {
+		if _, exist := t.Replicas[feed]; exist {
+			continue
+		}
+		if len(replicas) == 0 {
+			t.Replicas[feed] = make([]int32, 0)
+		} else {
+			t.Replicas[feed] = make([]int32, len(t.BindingClusters))
+		}
+	}
+
+	// transfer from cluster to index
+	m := make(map[string]int)
+	for i, cluster := range t.BindingClusters {
+		m[cluster] = i
+	}
+
+	for bi, bCluster := range b.BindingClusters {
+		ti, exist := m[bCluster]
+		if !exist {
+			// this cluster is a new one, we should append
+			t.BindingClusters = append(t.BindingClusters, bCluster)
+			for feed, tReplicas := range t.Replicas {
+				if len(tReplicas) != 0 {
+					t.Replicas[feed] = append(t.Replicas[feed], 0)
+				}
+			}
+			ti = len(t.BindingClusters) - 1
+			m[bCluster] = ti
+		}
+		for feed, bReplicas := range b.Replicas {
+			if len(bReplicas) != 0 {
+				t.Replicas[feed][ti] += bReplicas[bi]
+			}
+		}
+	}
+}
+
 const (
 	// NoClusterAvailableMsg is used to format message when no clusters available.
 	NoClusterAvailableMsg = "0/%v clusters are available"
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
index ebdc4f3..4e541b0 100644
--- a/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
@@ -17,48 +17,94 @@ type weightList struct {
 	weights   []int64
 }
 
-// DynamicDivideReplicas will fill the target replicas of all feeds based on predictor result and feed finv.
-func DynamicDivideReplicas(selected *framework.TargetClusters, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster, finv *appsapi.FeedInventory) error {
-	if selected.Replicas == nil {
-		selected.Replicas = make(map[string][]int32)
-	}
-	for _, feed := range sub.Spec.Feeds {
-		var order *appsapi.FeedOrder
-		for i := range finv.Spec.Feeds {
-			if feed == finv.Spec.Feeds[i].Feed {
-				order = &finv.Spec.Feeds[i]
-				break
+// DynamicDivideReplicas will fill the target replicas of all feeds based on predictor result and deviation.
+// First time scheduling is considered as a special kind of scaling. When the desired replicas in deviation
+// are negative, it means we should try to scale down, otherwise we try to scale up deviation replicas.
+func DynamicDivideReplicas(sub *appsapi.Subscription, deviations []appsapi.FeedOrder, availableReplicas framework.TargetClusters) (framework.TargetClusters, error) {
+	var err error
+
+	result := framework.NewTargetClusters(sub.Status.BindingClusters, sub.Status.Replicas).DeepCopy()
+	strategy := sub.Spec.DividingScheduling.DynamicDividing.Strategy
+
+	for i := range deviations {
+		d := deviations[i].DesiredReplicas
+		_, scheduled := sub.Status.Replicas[utils.GetFeedKey(deviations[i].Feed)]
+		var r framework.TargetClusters
+		switch {
+		// First scheduling is considered as a special kind of scaling up.
+		case !scheduled || (d != nil && *d > 0):
+			switch strategy {
+			case appsapi.SpreadDividingStrategy:
+				r = spreadScaleUp(&deviations[i], availableReplicas)
+			}
+		case d != nil && *d < 0:
+			switch strategy {
+			case appsapi.SpreadDividingStrategy:
+				r = spreadScaleDown(sub, deviations[i])
 			}
 		}
-		if order != nil && order.DesiredReplicas != nil {
-			replicas := dymamicDivideReplicas(*order.DesiredReplicas, selected.Replicas[order.Name])
-			selected.Replicas[utils.GetFeedKey(order.Feed)] = replicas
-		} else {
-			selected.Replicas[utils.GetFeedKey(order.Feed)] = []int32{}
+		if err != nil {
+			return *result, err
 		}
+		result.Merge(&r)
 	}
-	return nil
+	return *result, nil
+}
+
+func spreadScaleUp(d *appsapi.FeedOrder, availableReplicas framework.TargetClusters) framework.TargetClusters {
+	result := framework.TargetClusters{
+		BindingClusters: availableReplicas.BindingClusters,
+		Replicas:        make(map[string][]int32),
+	}
+	feedKey := utils.GetFeedKey(d.Feed)
+	if d.DesiredReplicas != nil {
+		replicas := dynamicDivideReplicas(*d.DesiredReplicas, availableReplicas.Replicas[feedKey])
+		result.Replicas[feedKey] = replicas
+	} else {
+		result.Replicas[feedKey] = []int32{}
+	}
+	return result
+}
+
+func spreadScaleDown(sub *appsapi.Subscription, d appsapi.FeedOrder) framework.TargetClusters {
+	result := framework.TargetClusters{
+		BindingClusters: sub.Status.BindingClusters,
+		Replicas:        make(map[string][]int32),
+	}
+	feedKey := utils.GetFeedKey(d.Feed)
+	if d.DesiredReplicas != nil {
+		replicas := dynamicDivideReplicas(*d.DesiredReplicas, sub.Status.Replicas[feedKey])
+		result.Replicas[feedKey] = replicas
+	} else {
+		result.Replicas[feedKey] = []int32{}
+	}
+	return result
 }
 
 // dynamicDivideReplicas divides replicas by the MaxAvailableReplicas
-func dymamicDivideReplicas(desiredReplicas int32, maxAvailableReplicas []int32) []int32 {
+func dynamicDivideReplicas(desiredReplicas int32, maxAvailableReplicas []int32) []int32 {
 	res := make([]int32, len(maxAvailableReplicas))
-	sumAvailableReplicas := utils.SumArrayInt32(maxAvailableReplicas)
+	weightSum := utils.SumArrayInt32(maxAvailableReplicas)
 
-	if desiredReplicas > sumAvailableReplicas {
+	if weightSum == 0 || desiredReplicas > weightSum {
 		return maxAvailableReplicas
 	}
 
-	for k := range res {
-		res[k] = (maxAvailableReplicas[k] / sumAvailableReplicas) * desiredReplicas
+	remain := desiredReplicas
+	for i, weight := range maxAvailableReplicas {
+		replica := weight * desiredReplicas / weightSum
+		res[i] = replica
+		remain -= replica
 	}
 
-	// Even out the rest replicas among all candidate clusters.
-	rest := desiredReplicas - utils.SumArrayInt32(res)
-	if rest > 0 {
-		for i := 0; i < int(rest) && i < len(res); i++ {
+	if remain > 0 {
+		for i := 0; i < int(remain) && i < len(res); i++ {
 			res[i]++
 		}
+	} else if remain < 0 {
+		for i := 0; i < int(-remain) && i < len(res); i++ {
+			res[i]--
+		}
 	}
 
 	return res
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/dynamic.go b/pkg/scheduler/framework/plugins/defaultassigner/dynamic.go
index 9c813c0..479ad48 100644
--- a/pkg/scheduler/framework/plugins/defaultassigner/dynamic.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/dynamic.go
@@ -18,14 +18,22 @@ package defaultassigner
 
 import (
 	"context"
+	"fmt"
 
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/pointer"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
 	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
 	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins/names"
+	"github.com/clusternet/clusternet/pkg/utils"
+)
+
+const (
+	// preAssignStateKey is the key in CycleState to DynamicAssigner pre-computed data.
+	// Using the name of the plugin will likely help us avoid collisions with other plugins.
+	preAssignStateKey = "PreAssign" + names.DynamicAssigner
 )
 
 // DynamicAssigner assigns replicas to clusters.
@@ -34,8 +42,33 @@ type DynamicAssigner struct {
 }
 
 var _ framework.AssignPlugin = &DynamicAssigner{}
+var _ framework.PreAssignPlugin = &DynamicAssigner{}
+
+// preAssignState computed at PreAssign and used at Assign.
+type preAssignState struct {
+	deviations []appsapi.FeedOrder
+}
+
+// Clone the prefilter state.
+func (s *preAssignState) Clone() framework.StateData {
+	return s
+}
+
+func getPreAssignState(cycleState *framework.CycleState) (*preAssignState, error) {
+	c, err := cycleState.Read(preAssignStateKey)
+	if err != nil {
+		// preAssignState doesn't exist, likely PreAssign wasn't invoked.
+		return nil, fmt.Errorf("error reading %q from cycleState: %w", preAssignStateKey, err)
+	}
+
+	s, ok := c.(*preAssignState)
+	if !ok {
+		return nil, fmt.Errorf("%+v convert to DynamicAssigner.preAssignState error", c)
+	}
+	return s, nil
+}
 
-// NewDynamicAssigner creates a default dynamic assigner.
+// NewDynamicAssigner creates a DefaultAssigner.
 func NewDynamicAssigner(_ runtime.Object, handle framework.Handle) (framework.Plugin, error) {
 	return &DynamicAssigner{handle: handle}, nil
 }
@@ -45,6 +78,32 @@ func (pl *DynamicAssigner) Name() string {
 	return names.DynamicAssigner
 }
 
+func (pl *DynamicAssigner) PreAssign(ctx context.Context, state *framework.CycleState, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) *framework.Status {
+	if sub.Spec.DividingScheduling == nil || sub.Spec.DividingScheduling.Type != appsapi.DynamicReplicaDividingType {
+		return nil
+	}
+	if sub.Spec.DividingScheduling.DynamicDividing == nil {
+		return framework.AsStatus(fmt.Errorf("must specify field DynamicDividing when dividing type is dynamic"))
+	}
+
+	// Scheduler only cares about the deviation replicas in case rescheduling running replicas.
+	// When scaling down, the deviation replicas would be negative.
+	s := new(preAssignState)
+	deviations := make([]appsapi.FeedOrder, len(finv.Spec.Feeds))
+	for i := range finv.Spec.Feeds {
+		deviations[i] = *finv.Spec.Feeds[i].DeepCopy()
+		if deviations[i].DesiredReplicas == nil {
+			continue
+		}
+		currentReplicas := utils.SumArrayInt32(sub.Status.Replicas[utils.GetFeedKey(deviations[i].Feed)])
+		deviations[i].DesiredReplicas = pointer.Int32(*deviations[i].DesiredReplicas - currentReplicas)
+	}
+	s.deviations = deviations
+
+	state.Write(preAssignStateKey, s)
+	return nil
+}
+
 // Assign assigns subscriptions to clusters using the clusternet client.
 func (pl *DynamicAssigner) Assign(ctx context.Context, state *framework.CycleState, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) (framework.TargetClusters, *framework.Status) {
 	klog.V(3).InfoS("Attempting to assign replicas to clusters",
@@ -52,17 +111,13 @@ func (pl *DynamicAssigner) Assign(ctx context.Context, state *framework.CycleSta
 	if sub.Spec.DividingScheduling == nil || sub.Spec.DividingScheduling.Type != appsapi.DynamicReplicaDividingType {
 		return framework.TargetClusters{}, framework.NewStatus(framework.Skip, "")
 	}
-	var err error
-	var result framework.TargetClusters
-	clusters := make([]*clusterapi.ManagedCluster, len(availableReplicas.BindingClusters))
-	for i, name := range availableReplicas.BindingClusters {
-		if clusters[i], err = pl.handle.ClusterCache().Get(name); err != nil {
-			return framework.TargetClusters{}, framework.AsStatus(err)
-		}
-		result.BindingClusters = append(result.BindingClusters, name)
+	s, err := getPreAssignState(state)
+	if err != nil {
+		return framework.TargetClusters{}, framework.AsStatus(err)
 	}
 
-	if err = DynamicDivideReplicas(&result, sub, clusters, finv); err != nil {
+	result, err := DynamicDivideReplicas(sub, s.deviations, availableReplicas)
+	if err != nil {
 		return framework.TargetClusters{}, framework.AsStatus(err)
 	}
 
diff --git a/pkg/scheduler/framework/plugins/predictor/predictor.go b/pkg/scheduler/framework/plugins/predictor/predictor.go
index b976994..5127285 100644
--- a/pkg/scheduler/framework/plugins/predictor/predictor.go
+++ b/pkg/scheduler/framework/plugins/predictor/predictor.go
@@ -17,41 +17,23 @@ limitations under the License.
 package predictor
 
 import (
-	"bytes"
 	"context"
-	"crypto/tls"
-	"encoding/json"
-	"fmt"
-	"io/ioutil"
-	"net/http"
-	"strings"
-	"time"
 
 	"k8s.io/apimachinery/pkg/runtime"
-	restclient "k8s.io/client-go/rest"
-	"k8s.io/klog/v2"
-	utilpointer "k8s.io/utils/pointer"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
-	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
-	proxiesapi "github.com/clusternet/clusternet/pkg/apis/proxies/v1alpha1"
-	schedulerapi "github.com/clusternet/clusternet/pkg/apis/scheduler"
-	"github.com/clusternet/clusternet/pkg/features"
+	clsapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	"github.com/clusternet/clusternet/pkg/apis/scheduler"
 	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
 	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins/names"
 )
 
-var _ framework.PredictPlugin = &Predictor{}
-
 // Predictor is a plugin than checks if a subscription need resources
 type Predictor struct {
 	handle framework.Handle
 }
 
-// New initializes a new plugin and returns it.
-func New(_ runtime.Object, h framework.Handle) (framework.Plugin, error) {
-	return &Predictor{handle: h}, nil
-}
+var _ framework.PredictPlugin = &Predictor{}
 
 // Name returns name of the plugin. It is used in logs, etc.
 func (pl *Predictor) Name() string {
@@ -59,80 +41,30 @@ func (pl *Predictor) Name() string {
 }
 
 // Predict invoked by scheduler predictor plugin
-func (pl *Predictor) Predict(_ context.Context, state *framework.CycleState, _ *appsapi.Subscription, finv *appsapi.FeedInventory,
-	mcls *clusterapi.ManagedCluster) (feedReplicas framework.FeedReplicas, s *framework.Status) {
-	if !mcls.Status.PredictorEnabled {
-		return nil, framework.NewStatus(framework.Skip, "predictor is disabled")
-	}
-
-	predictorAddress := mcls.Status.PredictorAddress
-	httpClient := &http.Client{
-		Transport: &http.Transport{
-			TLSClientConfig: &tls.Config{
-				InsecureSkipVerify: true,
-			},
-		},
-	}
-	var err error
-	if !mcls.Status.PredictorDirectAccess {
-		klog.V(5).Infof("the predictor of cluster %s can not be accessed directly", mcls.Spec.ClusterID)
-		if !mcls.Status.UseSocket {
-			return nil, framework.AsStatus(fmt.Errorf("cluster %s does not enable feature gate %s",
-				mcls.Spec.ClusterID, features.SocketConnection))
-		}
-
-		httpClient, err = restclient.HTTPClientFor(pl.handle.KubeConfig())
-		if err != nil {
-			return nil, framework.AsStatus(err)
-		}
+func (pl *Predictor) Predict(ctx context.Context, state *framework.CycleState, sub *appsapi.Subscription, finv *appsapi.FeedInventory, cluster *clsapi.ManagedCluster) (feedReplicas framework.FeedReplicas, s *framework.Status) {
+	var pp scheduler.PredictorProvider
 
-		predictorAddress = strings.Replace(predictorAddress, "http://", "http/", 1)
-		predictorAddress = strings.Replace(predictorAddress, "https://", "https/", 1)
-		predictorAddress = strings.Join([]string{
-			strings.TrimRight(pl.handle.KubeConfig().Host, "/"),
-			fmt.Sprintf("apis/%s/sockets/%s/proxy/%s", proxiesapi.SchemeGroupVersion.String(), mcls.Spec.ClusterID, predictorAddress),
-		}, "/")
+	if cluster.Status.PredictorEnabled {
+		// TODO: use pkg/predictor create a new predictor provider
+	} else {
+		return nil, framework.NewStatus(framework.Skip, "predictor is disabled")
 	}
-	httpClient.Timeout = time.Second * 32
-
-	for _, feedOrder := range finv.Spec.Feeds {
-		if feedOrder.DesiredReplicas == nil {
-			feedReplicas = append(feedReplicas, nil)
-			continue
-		}
 
-		replica, err2 := predictMaxAcceptableReplicas(httpClient, predictorAddress, feedOrder.ReplicaRequirements)
-		if err2 != nil {
-			return nil, framework.AsStatus(err2)
+	for _, feed := range finv.Spec.Feeds {
+		if feed.ReplicaRequirements.Resources.Size() == 0 {
+			feedReplicas = append(feedReplicas, 0)
+		} else {
+			replica, err := pp.MaxAcceptableReplicas(ctx, feed.ReplicaRequirements)
+			if err != nil {
+				return nil, framework.AsStatus(err)
+			}
+			feedReplicas = append(feedReplicas, replica[cluster.Name])
 		}
-		// Todo : support topology-aware replicas
-		feedReplicas = append(feedReplicas, utilpointer.Int32(replica[schedulerapi.DefaultAcceptableReplicasKey]))
 	}
 	return
 }
 
-func predictMaxAcceptableReplicas(httpClient *http.Client, address string, require appsapi.ReplicaRequirements) (map[string]int32, error) {
-	payload, err := json.Marshal(require)
-	if err != nil {
-		return nil, err
-	}
-
-	resp, err := httpClient.Post(address, "application/json", bytes.NewBuffer(payload))
-	if err != nil {
-		return nil, err
-	}
-
-	defer resp.Body.Close()
-	data, err := ioutil.ReadAll(resp.Body)
-	if err != nil {
-		return nil, err
-	}
-
-	predictorResults := &schedulerapi.PredictorResults{}
-	err = json.Unmarshal(data, predictorResults)
-	if err != nil {
-		return nil, err
-	}
-
-	return predictorResults.Replicas, nil
+// New initializes a new plugin and returns it.
+func New(_ runtime.Object, h framework.Handle) (framework.Plugin, error) {
+	return &Predictor{handle: h}, nil
 }
diff --git a/pkg/scheduler/framework/runtime/framework.go b/pkg/scheduler/framework/runtime/framework.go
index 0fa1398..b5d3d51 100644
--- a/pkg/scheduler/framework/runtime/framework.go
+++ b/pkg/scheduler/framework/runtime/framework.go
@@ -29,7 +29,6 @@ import (
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/record"
 	"k8s.io/klog/v2"
-	utilpointer "k8s.io/utils/pointer"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
 	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
@@ -415,8 +414,8 @@ func (f *frameworkImpl) RunPredictPlugins(ctx context.Context, state *framework.
 	// Run Predict method for each cluster in parallel.
 	f.Parallelizer().Until(ctx, len(clusters), func(index int) {
 		for i, pl := range f.predictPlugins {
-			replicas, status2 := f.runPredictPlugin(ctx, pl, state, sub, finv, clusters[index])
-			if !status2.IsSuccess() {
+			replicas, status := f.runPredictPlugin(ctx, pl, state, sub, finv, clusters[index])
+			if !status.IsSuccess() {
 				err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
 				errCh.SendErrorWithCancel(err, cancel)
 				return
@@ -922,12 +921,7 @@ func mergeFeedReplicas(a, b framework.FeedReplicas) framework.FeedReplicas {
 	}
 	res := make(framework.FeedReplicas, len(a))
 	for i := 0; i < len(b); i++ {
-		if a[i] != nil && b[i] != nil {
-			res[i] = utilpointer.Int32(utils.MinInt32(*a[i], *b[i]))
-			continue
-		}
-
-		res[i] = nil
+		res[i] = utils.MinInt32(a[i], b[i])
 	}
 	for i := len(b); i < len(a); i++ {
 		res[i] = a[i]
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index b2d1adb..01b89a0 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -72,7 +72,6 @@ type Scheduler struct {
 	schedulerOptions *options.SchedulerOptions
 
 	kubeClient                *kubernetes.Clientset
-	electionClient            *kubernetes.Clientset
 	clusternetClient          *clusternet.Clientset
 	ClusternetInformerFactory informers.SharedInformerFactory
 
@@ -113,11 +112,6 @@ func NewScheduler(schedulerOptions *options.SchedulerOptions) (*Scheduler, error
 	clusternetClient := clusternet.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-client-scheduler"))
 	clusternetInformerFactory := informers.NewSharedInformerFactory(clusternetClient, known.DefaultResync)
 
-	var electionClient *kubernetes.Clientset
-	if schedulerOptions.LeaderElection.LeaderElect {
-		electionClient = kubernetes.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-scheduler-election-client"))
-	}
-
 	// create event recorder
 	broadcaster := record.NewBroadcaster()
 	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
@@ -132,7 +126,6 @@ func NewScheduler(schedulerOptions *options.SchedulerOptions) (*Scheduler, error
 	sched := &Scheduler{
 		schedulerOptions:          schedulerOptions,
 		kubeClient:                kubeClient,
-		electionClient:            electionClient,
 		clusternetClient:          clusternetClient,
 		ClusternetInformerFactory: clusternetInformerFactory,
 		subsLister:                clusternetInformerFactory.Apps().V1alpha1().Subscriptions().Lister(),
@@ -195,7 +188,7 @@ func (sched *Scheduler) Run(ctx context.Context) error {
 		sched.schedulerOptions.LeaderElection.LeaseDuration.Duration,
 		sched.schedulerOptions.LeaderElection.RenewDeadline.Duration,
 		sched.schedulerOptions.LeaderElection.RetryPeriod.Duration,
-		sched.electionClient,
+		sched.kubeClient,
 		leaderelection.LeaderCallbacks{
 			OnStartedLeading: func(ctx context.Context) {
 				wait.UntilWithContext(ctx, sched.scheduleOne, 0)


Test patch:
diff --git a/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate_test.go b/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate_test.go
index c7a6b32..802fd62 100644
--- a/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate_test.go
+++ b/pkg/predictor/framework/plugins/defaultaggregate/default_aggregate_test.go
@@ -21,7 +21,6 @@ import (
 	"fmt"
 	"testing"
 
-	schedulerapi "github.com/clusternet/clusternet/pkg/apis/scheduler"
 	framework "github.com/clusternet/clusternet/pkg/predictor/framework/interfaces"
 )
 
@@ -53,9 +52,9 @@ func TestDefaultAggregator(t *testing.T) {
 			if got := status.AsError(); got != nil {
 				t.Errorf("status does not match %q, want nil", got)
 			}
-			if maxAvailableReplicas[schedulerapi.DefaultAcceptableReplicasKey] != test.want {
+			if maxAvailableReplicas[framework.DefaultAcceptableReplicasKey] != test.want {
 				t.Errorf("got different maxAvailableReplicas got %q, want %q",
-					maxAvailableReplicas[schedulerapi.DefaultAcceptableReplicasKey], test.want)
+					maxAvailableReplicas[framework.DefaultAcceptableReplicasKey], test.want)
 			}
 		})
 	}
diff --git a/pkg/scheduler/framework/interfaces/types_test.go b/pkg/scheduler/framework/interfaces/types_test.go
new file mode 100644
index 0000000..d65f41f
--- /dev/null
+++ b/pkg/scheduler/framework/interfaces/types_test.go
@@ -0,0 +1,110 @@
+package interfaces
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestTargetClusters_Merge(t *testing.T) {
+	template := &TargetClusters{
+		BindingClusters: []string{"c1", "c2", "c3"},
+		Replicas: map[string][]int32{
+			"f1": {1, 2, 3},
+			"f2": {},
+			"f3": {1, 0, 1},
+		},
+	}
+	tests := []struct {
+		name   string
+		b      *TargetClusters
+		result *TargetClusters
+	}{
+		{
+			name: "f1 and f3 in c2 and c3",
+			b: &TargetClusters{
+				BindingClusters: []string{"c2", "c3"},
+				Replicas: map[string][]int32{
+					"f1": {1, 2},
+					"f3": {1, 0},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					"f1": {1, 3, 5},
+					"f2": {},
+					"f3": {1, 1, 1},
+				},
+			},
+		},
+		{
+			name: "f1 and f3 in c2 and c4",
+			b: &TargetClusters{
+				BindingClusters: []string{"c2", "c4"},
+				Replicas: map[string][]int32{
+					"f1": {1, 2},
+					"f3": {1, 2},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3", "c4"},
+				Replicas: map[string][]int32{
+					"f1": {1, 3, 3, 2},
+					"f2": {},
+					"f3": {1, 1, 1, 2},
+				},
+			},
+		},
+		{
+			name: "f3, f4 and f5 in c2 and c4",
+			b: &TargetClusters{
+				BindingClusters: []string{"c2", "c4"},
+				Replicas: map[string][]int32{
+					"f3": {1, 2},
+					"f4": {1, 2},
+					"f5": {2, 3},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3", "c4"},
+				Replicas: map[string][]int32{
+					"f1": {1, 2, 3, 0},
+					"f2": {},
+					"f3": {1, 1, 1, 2},
+					"f4": {0, 1, 0, 2},
+					"f5": {0, 2, 0, 3},
+				},
+			},
+		},
+		{
+			name: "f3, f4 and f5 in c0 and c4",
+			b: &TargetClusters{
+				BindingClusters: []string{"c0", "c4"},
+				Replicas: map[string][]int32{
+					"f3": {1, 2},
+					"f4": {1, 2},
+					"f5": {},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3", "c0", "c4"},
+				Replicas: map[string][]int32{
+					"f1": {1, 2, 3, 0, 0},
+					"f2": {},
+					"f3": {1, 0, 1, 1, 2},
+					"f4": {0, 0, 0, 1, 2},
+					"f5": {},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t1 *testing.T) {
+			tcp := template.DeepCopy()
+			tcp.Merge(tt.b)
+			if !reflect.DeepEqual(tcp, tt.result) {
+				t.Errorf("Merge() gotResponse = %v, want %v", tcp, tt.result)
+			}
+		})
+	}
+}
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go b/pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go
index 5153f89..b8cd2a6 100644
--- a/pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go
@@ -132,6 +132,15 @@ func newDeploymentFeed(name string) appsapi.Feed {
 	}
 }
 
+func newServiceFeed(name string) appsapi.Feed {
+	return appsapi.Feed{
+		Kind:       "Service",
+		APIVersion: "v1",
+		Namespace:  "default",
+		Name:       name,
+	}
+}
+
 func TestDivideReplicas(t *testing.T) {
 	desiredReplicas := int32(12)
 	type args struct {
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go b/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go
new file mode 100644
index 0000000..165624e
--- /dev/null
+++ b/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go
@@ -0,0 +1,312 @@
+package defaultassigner
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"k8s.io/utils/pointer"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+	"github.com/clusternet/clusternet/pkg/utils"
+)
+
+var (
+	subTemplate = &appsapi.Subscription{
+		Spec: appsapi.SubscriptionSpec{
+			SchedulingStrategy: appsapi.DividingSchedulingStrategyType,
+			DividingScheduling: &appsapi.DividingScheduling{
+				Type: appsapi.DynamicReplicaDividingType,
+				DynamicDividing: &appsapi.DynamicDividing{
+					Strategy: appsapi.SpreadDividingStrategy,
+				},
+			},
+			Feeds: []appsapi.Feed{
+				newDeploymentFeed("f1"),
+				newServiceFeed("f2"),
+				newDeploymentFeed("f3"),
+			},
+		},
+		Status: appsapi.SubscriptionStatus{
+			BindingClusters: []string{"c1", "c2", "c3"},
+			Replicas: map[string][]int32{
+				utils.GetFeedKey(newDeploymentFeed("f1")): {1, 3, 0}, // already scheduled: 4
+				utils.GetFeedKey(newServiceFeed("f2")):    {},
+				utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1}, // already scheduled: 3
+			},
+		},
+	}
+)
+
+func TestDynamicAssigner_Assign(t *testing.T) {
+	type args struct {
+		sub               *appsapi.Subscription
+		finv              *appsapi.FeedInventory
+		availableReplicas framework.TargetClusters
+	}
+	tests := []struct {
+		name string
+		args args
+		want framework.TargetClusters
+	}{
+		{
+			name: "scale up f1",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(8)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(3)},
+						},
+					},
+				},
+				availableReplicas: framework.TargetClusters{
+					BindingClusters: []string{"c1", "c2", "c3"},
+					Replicas: map[string][]int32{
+						utils.GetFeedKey(newDeploymentFeed("f1")): {10, 10, 10},
+						utils.GetFeedKey(newServiceFeed("f2")):    {},
+						utils.GetFeedKey(newDeploymentFeed("f3")): {10, 10, 10},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {3, 4, 1},
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1},
+				},
+			},
+		},
+		{
+			name: "scale up f1 and f3",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(8)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(9)},
+						},
+					},
+				},
+				availableReplicas: framework.TargetClusters{
+					BindingClusters: []string{"c1", "c2", "c3"},
+					Replicas: map[string][]int32{
+						utils.GetFeedKey(newDeploymentFeed("f1")): {10, 10, 10},
+						utils.GetFeedKey(newServiceFeed("f2")):    {},
+						utils.GetFeedKey(newDeploymentFeed("f3")): {10, 20, 30},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {3, 4, 1},
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {2, 3, 4},
+				},
+			},
+		},
+		{
+			name: "scale down f1",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(2)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(3)},
+						},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {0, 2, 0},
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1},
+				},
+			},
+		},
+		{
+			name: "scale down f1 and f3",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(2)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(1)},
+						},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {0, 2, 0},
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {0, 0, 1},
+				},
+			},
+		},
+		{
+			name: "scale up f1 and f3 with new candidate clusters",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(8)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(9)},
+						},
+					},
+				},
+				availableReplicas: framework.TargetClusters{
+					BindingClusters: []string{"c1", "c2", "c4"},
+					Replicas: map[string][]int32{
+						utils.GetFeedKey(newDeploymentFeed("f1")): {10, 10, 10},
+						utils.GetFeedKey(newServiceFeed("f2")):    {},
+						utils.GetFeedKey(newDeploymentFeed("f3")): {10, 20, 30},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3", "c4"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {3, 4, 0, 1},
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {2, 3, 1, 3},
+				},
+			},
+		},
+		{
+			name: "scale up f1 with new candidate clusters and scale down f3",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(8)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(1)},
+						},
+					},
+				},
+				availableReplicas: framework.TargetClusters{
+					BindingClusters: []string{"c1", "c2", "c4"},
+					Replicas: map[string][]int32{
+						utils.GetFeedKey(newDeploymentFeed("f1")): {10, 10, 10},
+						utils.GetFeedKey(newServiceFeed("f2")):    {},
+						utils.GetFeedKey(newDeploymentFeed("f3")): {10, 20, 30},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3", "c4"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {3, 4, 0, 1},
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {0, 0, 1, 0},
+				},
+			},
+		},
+		{
+			name: "do nothing",
+			args: args{
+				sub: subTemplate.DeepCopy(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(4)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(3)},
+						},
+					},
+				},
+				availableReplicas: framework.TargetClusters{
+					BindingClusters: []string{"c1", "c2", "c4"},
+					Replicas: map[string][]int32{
+						utils.GetFeedKey(newDeploymentFeed("f1")): {10, 10, 10},
+						utils.GetFeedKey(newServiceFeed("f2")):    {},
+						utils.GetFeedKey(newDeploymentFeed("f3")): {10, 20, 30},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {1, 3, 0}, // already scheduled: 4
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1}, // already scheduled: 3
+				},
+			},
+		},
+		{
+			name: "add new secret",
+			args: args{
+				sub: func() *appsapi.Subscription {
+					sub := subTemplate.DeepCopy()
+					sub.Spec.Feeds = append(sub.Spec.Feeds, newServiceFeed("f4"))
+					return sub
+				}(),
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{
+						Feeds: []appsapi.FeedOrder{
+							{Feed: newDeploymentFeed("f1"), DesiredReplicas: pointer.Int32(4)},
+							{Feed: newServiceFeed("f2")},
+							{Feed: newDeploymentFeed("f3"), DesiredReplicas: pointer.Int32(3)},
+							{Feed: newServiceFeed("f4")},
+						},
+					},
+				},
+				availableReplicas: framework.TargetClusters{
+					BindingClusters: []string{"c1", "c2", "c4"},
+					Replicas: map[string][]int32{
+						utils.GetFeedKey(newDeploymentFeed("f1")): {10, 10, 10},
+						utils.GetFeedKey(newServiceFeed("f2")):    {},
+						utils.GetFeedKey(newDeploymentFeed("f3")): {10, 20, 30},
+						utils.GetFeedKey(newServiceFeed("f4")):    {},
+					},
+				},
+			},
+			want: framework.TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3", "c4"},
+				Replicas: map[string][]int32{
+					utils.GetFeedKey(newDeploymentFeed("f1")): {1, 3, 0, 0}, // already scheduled: 4
+					utils.GetFeedKey(newServiceFeed("f2")):    {},
+					utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1, 0}, // already scheduled: 3
+					utils.GetFeedKey(newServiceFeed("f4")):    {},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			pl := new(DynamicAssigner)
+			state := framework.NewCycleState()
+
+			status := pl.PreAssign(context.TODO(), state, tt.args.sub, tt.args.finv, tt.args.availableReplicas)
+			if status != nil {
+				t.Errorf("PreAssign() got status = %v, want nil", status)
+			}
+
+			got, status := pl.Assign(context.TODO(), state, tt.args.sub, tt.args.finv, tt.args.availableReplicas)
+			if status != nil {
+				t.Errorf("Assign() got status = %v, want nil", status)
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("Assign() got = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}