Repository: clusternet/clusternet
Pull request #322 (2022-05-07 06:18:14)

add scheduler framework
Signed-off-by: Garrybest <garrybest@foxmail.com>

<!--  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:


codecov-commenter: # [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/322?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) Report
> Merging [#322](https://codecov.io/gh/clusternet/clusternet/pull/322?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (b6b36df) into [main](https://codecov.io/gh/clusternet/clusternet/commit/4ab555b215aa504dc2aab6ccf801a3071dcab921?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (4ab555b) will **increase** coverage by `1.11%`.
> The diff coverage is `n/a`.

```diff
@@            Coverage Diff             @@
##             main     #322      +/-   ##
==========================================
+ Coverage   11.09%   12.21%   +1.11%     
==========================================
  Files          47       47              
  Lines        4991     4815     -176     
==========================================
+ Hits          554      588      +34     
+ Misses       4399     4184     -215     
- Partials       38       43       +5     
```


| [Impacted Files](https://codecov.io/gh/clusternet/clusternet/pull/322?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) | Coverage Δ | |
|---|---|---|
| [...ithub.com/clusternet/clusternet/pkg/utils/slice.go](https://codecov.io/gh/clusternet/clusternet/pull/322/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy91dGlscy9zbGljZS5nbw==) | `58.33% <0.00%> (-41.67%)` | :arrow_down: |
| [...rnet/clusternet/pkg/scheduler/algorithm/generic.go](https://codecov.io/gh/clusternet/clusternet/pull/322/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvYWxnb3JpdGhtL2dlbmVyaWMuZ28=) | | |
| [...sternet/pkg/scheduler/algorithm/dividing/static.go](https://codecov.io/gh/clusternet/clusternet/pull/322/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvYWxnb3JpdGhtL2RpdmlkaW5nL3N0YXRpYy5nbw==) | | |
| [...eduler/framework/plugins/defaultassigner/static.go](https://codecov.io/gh/clusternet/clusternet/pull/322/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvZnJhbWV3b3JrL3BsdWdpbnMvZGVmYXVsdGFzc2lnbmVyL3N0YXRpYy5nbw==) | `50.00% <0.00%> (ø)` | |
| [...ler/framework/plugins/defaultassigner/algorithm.go](https://codecov.io/gh/clusternet/clusternet/pull/322/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvZnJhbWV3b3JrL3BsdWdpbnMvZGVmYXVsdGFzc2lnbmVyL2FsZ29yaXRobS5nbw==) | `87.50% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/clusternet/clusternet/pull/322?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/322?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Last update [4ab555b...b6b36df](https://codecov.io/gh/clusternet/clusternet/pull/322?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet).


Patch:
diff --git a/pkg/scheduler/algorithm/generic.go b/pkg/scheduler/algorithm/generic.go
index bfbad3a..82d496a 100644
--- a/pkg/scheduler/algorithm/generic.go
+++ b/pkg/scheduler/algorithm/generic.go
@@ -21,6 +21,7 @@ package algorithm
 import (
 	"context"
 	"fmt"
+	"sort"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -31,13 +32,13 @@ import (
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
 	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
-	"github.com/clusternet/clusternet/pkg/scheduler/algorithm/dividing"
 	schedulerapis "github.com/clusternet/clusternet/pkg/scheduler/apis"
 	schedulercache "github.com/clusternet/clusternet/pkg/scheduler/cache"
 	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
 	"github.com/clusternet/clusternet/pkg/scheduler/framework/runtime"
 	"github.com/clusternet/clusternet/pkg/scheduler/metrics"
 	"github.com/clusternet/clusternet/pkg/scheduler/parallelize"
+	"github.com/clusternet/clusternet/pkg/utils"
 )
 
 const (
@@ -73,6 +74,7 @@ func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework
 		return result, ErrNoClustersAvailable
 	}
 
+	// Step 1: Filter clusters.
 	feasibleClusters, diagnosis, err := g.findClustersThatFitSubscription(ctx, fwk, sub)
 	if err != nil {
 		return result, err
@@ -87,12 +89,19 @@ func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework
 		}
 	}
 
-	priorityList, err := prioritizeClusters(ctx, fwk, sub, feasibleClusters)
+	// Step 2: Estimate max available replicas if necessary.
+	availableList, err := estimateReplicas(ctx, fwk, sub, finv, feasibleClusters)
 	if err != nil {
 		return result, err
 	}
 
-	clusters, err := g.selectClusters(priorityList, sub, finv)
+	// Step 3: Prioritize clusters.
+	priorityList, err := prioritizeClusters(ctx, fwk, sub, feasibleClusters, availableList)
+	if err != nil {
+		return result, err
+	}
+
+	clusters, err := g.selectClusters(ctx, priorityList, fwk, sub, finv)
 	trace.Step("Prioritizing done")
 
 	return ScheduleResult{
@@ -102,9 +111,51 @@ func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework
 	}, err
 }
 
+func estimateReplicas(ctx context.Context, fwk framework.Framework, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster) (framework.ClusterScoreList, error) {
+	availableList := make(framework.ClusterScoreList, len(clusters))
+	for i := range clusters {
+		availableList[i] = framework.ClusterScore{
+			NamespacedName: klog.KObj(clusters[i]).String(),
+		}
+	}
+
+	// If not dividing, finv does not exist.
+	if sub.Spec.SchedulingStrategy != appsapi.DividingSchedulingStrategyType {
+		return availableList, nil
+	}
+
+	// Initialize.
+	for i := range availableList {
+		availableList[i].MaxAvailableReplicas = make(framework.FeedReplicas, len(finv.Spec.Feeds))
+	}
+
+	if sub.Spec.DividingScheduling == nil || sub.Spec.DividingScheduling.Type == appsapi.StaticReplicaDividingType || !fwk.HasEstimatePlugins() {
+		return availableList, nil
+	}
+
+	// Run PreEstimate plugins.
+	preEstimateStatus := fwk.RunPreEstimatePlugins(ctx, sub, finv, clusters)
+	if !preEstimateStatus.IsSuccess() {
+		return nil, preEstimateStatus.AsError()
+	}
+
+	// Run the Estimate plugins.
+	availableList, estimateStatus := fwk.RunEstimatePlugins(ctx, sub, finv, clusters, availableList)
+	if !estimateStatus.IsSuccess() {
+		return nil, estimateStatus.AsError()
+	}
+
+	if klog.V(10).Enabled() {
+		for i := range availableList {
+			klog.V(10).InfoS("Estimated cluster's max available replicas for subscription", "subscription", klog.KObj(sub), "cluster", availableList[i].NamespacedName, "max available replicas", availableList[i].MaxAvailableReplicas)
+		}
+	}
+	return availableList, nil
+}
+
 // selectClusters takes a prioritized list of clusters and then picks a fraction of clusters
 // in a reservoir sampling manner from the clusters that had the highest score.
-func (g *genericScheduler) selectClusters(clusterScoreList framework.ClusterScoreList, sub *appsapi.Subscription, finv *appsapi.FeedInventory) (framework.TargetClusters, error) {
+func (g *genericScheduler) selectClusters(ctx context.Context, clusterScoreList framework.ClusterScoreList, fwk framework.Framework, sub *appsapi.Subscription, finv *appsapi.FeedInventory) (framework.TargetClusters, error) {
 	if len(clusterScoreList) == 0 {
 		return framework.TargetClusters{}, fmt.Errorf("empty clusterScoreList")
 	}
@@ -115,13 +166,35 @@ func (g *genericScheduler) selectClusters(clusterScoreList framework.ClusterScor
 		selected.BindingClusters = append(selected.BindingClusters, clusterScore.NamespacedName)
 	}
 
-	if sub.Spec.SchedulingStrategy == appsapi.DividingSchedulingStrategyType {
-		if err := g.divideReplicas(&selected, sub, finv); err != nil {
-			return framework.TargetClusters{}, fmt.Errorf("failed to divide replicas: %v", err)
+	if sub.Spec.SchedulingStrategy != appsapi.DividingSchedulingStrategyType {
+		return selected, nil
+	}
+
+	selected.Replicas = make(map[string][]int32)
+	// transfer to available replicas
+	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])
 		}
 	}
 
-	return selected, nil
+	// Run PreAssign plugins.
+	preAssignStatus := fwk.RunPreAssignPlugins(ctx, sub, finv, selected)
+	if !preAssignStatus.IsSuccess() {
+		return framework.TargetClusters{}, preAssignStatus.AsError()
+	}
+
+	// Run the Assign plugins.
+	selected, assignStatus := fwk.RunAssignPlugins(ctx, sub, finv, selected)
+	if assignStatus.IsSuccess() {
+		return selected, nil
+	}
+	if assignStatus.Code() == framework.Error {
+		return framework.TargetClusters{}, assignStatus.AsError()
+	}
+
+	return framework.TargetClusters{}, fmt.Errorf("assign status: %s, %v", assignStatus.Code().String(), assignStatus.Message())
 }
 
 // numFeasibleClustersToFind returns the number of feasible clusters that once found, the scheduler stops
@@ -263,40 +336,17 @@ func (g *genericScheduler) findClustersThatPassFilters(ctx context.Context, fwk
 	return feasibleClusters, nil
 }
 
-func (g *genericScheduler) divideReplicas(selected *framework.TargetClusters, sub *appsapi.Subscription, finv *appsapi.FeedInventory) error {
-	switch sub.Spec.DividingScheduling.Type {
-	case appsapi.StaticReplicaDividingType:
-		return g.staticDivideReplicas(selected, sub, finv)
-	}
-	return nil
-}
-
-func (g *genericScheduler) staticDivideReplicas(selected *framework.TargetClusters, sub *appsapi.Subscription, finv *appsapi.FeedInventory) (err error) {
-	clusters := make([]*clusterapi.ManagedCluster, len(selected.BindingClusters))
-	for i, name := range selected.BindingClusters {
-		if clusters[i], err = g.cache.Get(name); err != nil {
-			return err
-		}
-	}
-
-	return dividing.StaticDivideReplicas(selected, sub, clusters, finv)
-}
-
 // prioritizeClusters prioritizes the clusters by running the score plugins,
 // which return a score for each cluster from the call to RunScorePlugins().
 // The scores from each plugin are added together to make the score for that cluster, then
 // any extenders are run as well.
 // All scores are finally combined (added) to get the total weighted scores of all clusters
-func prioritizeClusters(ctx context.Context, fwk framework.Framework, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster) (framework.ClusterScoreList, error) {
+func prioritizeClusters(ctx context.Context, fwk framework.Framework, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster, result framework.ClusterScoreList) (framework.ClusterScoreList, error) {
 	// If no priority configs are provided, then all clusters will have a score of one.
 	// This is required to generate the priority list in the required format
 	if !fwk.HasScorePlugins() {
-		result := make(framework.ClusterScoreList, 0, len(clusters))
-		for i := range clusters {
-			result = append(result, framework.ClusterScore{
-				NamespacedName: klog.KObj(clusters[i]).String(),
-				Score:          1,
-			})
+		for i := range result {
+			result[i].Score = 1
 		}
 		return result, nil
 	}
@@ -316,24 +366,21 @@ func prioritizeClusters(ctx context.Context, fwk framework.Framework, sub *appsa
 	if klog.V(10).Enabled() {
 		for plugin, clusterScoreList := range scoresMap {
 			for _, clusterScore := range clusterScoreList {
-				klog.InfoS("Plugin scored cluster for subscription", "subscription", klog.KObj(sub), "plugin", plugin, "cluster", clusterScore.NamespacedName, "score", clusterScore.Score)
+				klog.V(10).InfoS("Plugin scored cluster for subscription", "subscription", klog.KObj(sub), "plugin", plugin, "cluster", clusterScore.NamespacedName, "score", clusterScore.Score)
 			}
 		}
 	}
 
-	// Summarize all scores.
-	result := make(framework.ClusterScoreList, 0, len(clusters))
-
 	for i := range clusters {
-		result = append(result, framework.ClusterScore{NamespacedName: klog.KObj(clusters[i]).String(), Score: 0})
 		for j := range scoresMap {
 			result[i].Score += scoresMap[j][i].Score
 		}
 	}
+	sort.Sort(result)
 
 	if klog.V(10).Enabled() {
 		for i := range result {
-			klog.InfoS("Calculated cluster's final score for subscription", "subscription", klog.KObj(sub), "cluster", result[i].NamespacedName, "score", result[i].Score)
+			klog.V(10).InfoS("Calculated cluster's final score for subscription", "subscription", klog.KObj(sub), "cluster", result[i].NamespacedName, "score", result[i].Score)
 		}
 	}
 	return result, nil
diff --git a/pkg/scheduler/apis/types.go b/pkg/scheduler/apis/types.go
index 9ce6d54..5c31356 100644
--- a/pkg/scheduler/apis/types.go
+++ b/pkg/scheduler/apis/types.go
@@ -39,12 +39,24 @@ type Plugins struct {
 	// PostFilter is a list of plugins that are invoked after filtering phase, no matter whether filtering succeeds or not.
 	PostFilter PluginSet
 
+	// PreEstimate is a list of plugins that are invoked before estimating.
+	PreEstimate PluginSet
+
+	// Estimate is a list of plugins that should be invoked when estimate max available replicas for clusters that have passed the filtering phase.
+	Estimate PluginSet
+
 	// PreScore is a list of plugins that are invoked before scoring.
 	PreScore PluginSet
 
 	// Score is a list of plugins that should be invoked when ranking clusters that have passed the filtering phase.
 	Score PluginSet
 
+	// PreAssign is a list of plugins that are invoked before assigning.
+	PreAssign PluginSet
+
+	// Assign is a list of plugins that should be invoked when assigning replicas.
+	Assign PluginSet
+
 	// Reserve is a list of plugins invoked when reserving/unreserving resources
 	// after a cluster is assigned to run the Subscription.
 	Reserve PluginSet
@@ -111,8 +123,12 @@ func (p *Plugins) Names() []string {
 		p.Filter,
 		p.PostFilter,
 		p.Reserve,
+		p.PreEstimate,
+		p.Estimate,
 		p.PreScore,
 		p.Score,
+		p.PreAssign,
+		p.Assign,
 		p.PreBind,
 		p.Bind,
 		p.PostBind,
diff --git a/pkg/scheduler/default_plugins.go b/pkg/scheduler/default_plugins.go
index 25a9fda..0c33525 100644
--- a/pkg/scheduler/default_plugins.go
+++ b/pkg/scheduler/default_plugins.go
@@ -30,13 +30,21 @@ func getDefaultPlugins() *schedulerapis.Plugins {
 				{Name: names.TaintToleration},
 			},
 		},
-		PostFilter: schedulerapis.PluginSet{},
-		PreScore:   schedulerapis.PluginSet{},
+		PostFilter:  schedulerapis.PluginSet{},
+		PreEstimate: schedulerapis.PluginSet{},
+		Estimate:    schedulerapis.PluginSet{},
+		PreScore:    schedulerapis.PluginSet{},
 		Score: schedulerapis.PluginSet{
 			Enabled: []schedulerapis.Plugin{
 				{Name: names.TaintToleration, Weight: 3},
 			},
 		},
+		PreAssign: schedulerapis.PluginSet{},
+		Assign: schedulerapis.PluginSet{
+			Enabled: []schedulerapis.Plugin{
+				{Name: names.StaticAssigner},
+			},
+		},
 		Reserve: schedulerapis.PluginSet{},
 		Permit:  schedulerapis.PluginSet{},
 		PreBind: schedulerapis.PluginSet{},
diff --git a/pkg/scheduler/framework/interfaces/interface.go b/pkg/scheduler/framework/interfaces/interface.go
index 1247449..29bfbc1 100644
--- a/pkg/scheduler/framework/interfaces/interface.go
+++ b/pkg/scheduler/framework/interfaces/interface.go
@@ -33,18 +33,45 @@ import (
 	clusternet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
 	informers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions"
 	schedulerapis "github.com/clusternet/clusternet/pkg/scheduler/apis"
+	"github.com/clusternet/clusternet/pkg/scheduler/cache"
 	"github.com/clusternet/clusternet/pkg/scheduler/parallelize"
 )
 
 // ClusterScoreList declares a list of clusters and their scores.
 type ClusterScoreList []ClusterScore
 
+func (cs ClusterScoreList) Len() int {
+	return len(cs)
+}
+
+func (cs ClusterScoreList) Less(i, j int) bool {
+	return cs[i].Score > cs[j].Score
+}
+
+func (cs ClusterScoreList) Swap(i, j int) {
+	cs[i], cs[j] = cs[j], cs[i]
+}
+
 // ClusterScore is a struct with cluster id and score.
 type ClusterScore struct {
-	NamespacedName string // the namespaced name key of a ManagedCluster
-	Score          int64
+	NamespacedName       string // the namespaced name key of a ManagedCluster
+	Score                int64
+	MaxAvailableReplicas FeedReplicas
 }
 
+func (cs ClusterScoreList) ClusterNames() []string {
+	clusters := make([]string, 0, len(cs))
+	for _, score := range cs {
+		clusters = append(clusters, score.NamespacedName)
+	}
+	return clusters
+}
+
+type FeedReplicas []int32
+
+// PluginToClusterReplicas declares a map from plugin name to its ClusterScoreList.
+type PluginToClusterReplicas map[string]ClusterScoreList
+
 // PluginToClusterScores declares a map from plugin name to its ClusterScoreList.
 type PluginToClusterScores map[string]ClusterScoreList
 
@@ -170,6 +197,29 @@ type PostFilterPlugin interface {
 	PostFilter(ctx context.Context, sub *appsapi.Subscription, filteredClusterStatusMap ClusterToStatusMap) (*PostFilterResult, *Status)
 }
 
+// PreEstimatePlugin is an interface for "PreEstimate" plugin. PreEstimate is an
+// informational extension point. Plugins will be called with a list of clusters
+// that passed the filtering phase.
+type PreEstimatePlugin interface {
+	Plugin
+
+	// PreEstimate is called by the scheduling framework after a list of managed clusters
+	// passed the filtering phase. All pre-estimate plugins must return success or
+	// the subscription will be rejected.
+	PreEstimate(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster) *Status
+}
+
+// EstimatePlugin is an interface that must be implemented by "Estimate" plugins to estimate
+// max available replicas for clusters that passed the filtering phase.
+type EstimatePlugin interface {
+	Plugin
+
+	// Estimate is called on each filtered cluster. It must return success and available
+	// replicas indicating the feasible domain for sake of division. All estimate plugins
+	// must return success or the subscription will be rejected.
+	Estimate(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, cluster *clusterapi.ManagedCluster) (FeedReplicas, *Status)
+}
+
 // PreScorePlugin is an interface for "PreScore" plugin. PreScore is an
 // informational extension point. Plugins will be called with a list of clusters
 // that passed the filtering phase. A plugin may use this data to update internal
@@ -205,6 +255,29 @@ type ScorePlugin interface {
 	ScoreExtensions() ScoreExtensions
 }
 
+// PreAssignPlugin is an interface for "PreAssign" plugin. PreAssign is an
+// informational extension point. These are meant to prepare the state of
+// the assign.
+type PreAssignPlugin interface {
+	Plugin
+
+	// PreAssign is called by the scheduling framework after a list of managed clusters
+	// passed the filtering phase. All pre-estimate plugins must return success or
+	// the subscription will be rejected.
+	PreAssign(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas TargetClusters) *Status
+}
+
+// AssignPlugin is an interface that must be implemented by "Assign" plugins to assign replicas
+// for each selected cluster.
+type AssignPlugin interface {
+	Plugin
+
+	// Assign is called on each selected cluster. It will assign replicas for each
+	// selected cluster. The return result is a map whose key is feed key and
+	// value is respective assigned replicas for each selected cluster.
+	Assign(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas TargetClusters) (TargetClusters, *Status)
+}
+
 // ReservePlugin is an interface for plugins with Reserve and Unreserve
 // methods. These are meant to update the state of the plugin. This concept
 // used to be called 'assume' in the original scheduler. These plugins should
@@ -293,6 +366,31 @@ type Framework interface {
 	// cluster state to make the subscription potentially schedulable in a future scheduling cycle.
 	RunPostFilterPlugins(ctx context.Context, sub *appsapi.Subscription, filteredClusterStatusMap ClusterToStatusMap) (*PostFilterResult, *Status)
 
+	// RunPreEstimatePlugins runs the set of configured PreEstimate plugins. It returns
+	// *Status and its code is set to non-success if any of the plugins returns
+	// anything but Success. If a non-success status is returned, then the scheduling
+	// cycle is aborted.
+	RunPreEstimatePlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster) *Status
+
+	// RunEstimatePlugins runs the set of configured Estimate plugins. It returns a map that
+	// stores for each Estimate plugin name the corresponding ClusterScoreList(s).
+	// It also returns *Status, which is set to non-success if any of the plugins returns
+	// a non-success status.
+	RunEstimatePlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster, availableList ClusterScoreList) (ClusterScoreList, *Status)
+
+	// RunPreAssignPlugins runs the set of configured PreAssign plugins. It returns
+	// *Status and its code is set to non-success if any of the plugins returns
+	// anything but Success. If a non-success status is returned, then the scheduling
+	// cycle is aborted.
+	RunPreAssignPlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, selected TargetClusters) *Status
+
+	// RunAssignPlugins runs the set of configured Assign plugins. An Assign plugin may choose
+	// whether to handle the given subscription. If an Assign plugin chooses to skip the
+	// assigning, it should return code=5("skip") status. Otherwise, it should return "Error"
+	// or "Success". If none of the plugins handled assigning, RunAssignPlugins returns
+	// code=5("skip") status.
+	RunAssignPlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, selected TargetClusters) (TargetClusters, *Status)
+
 	// RunReservePluginsReserve runs the Reserve method of the set of
 	// configured Reserve plugins. If any of these calls returns an error, it
 	// does not continue running the remaining ones and returns the error. In
@@ -340,6 +438,9 @@ type Framework interface {
 	// HasScorePlugins returns true if at least one Score plugin is defined.
 	HasScorePlugins() bool
 
+	// HasEstimatePlugins returns true if at least one Estimate plugin is defined.
+	HasEstimatePlugins() bool
+
 	// ListPlugins returns a map of extension point name to list of configured Plugins.
 	ListPlugins() *schedulerapis.Plugins
 
@@ -354,6 +455,9 @@ type Handle interface {
 	// PluginsRunner abstracts operations to run some plugins.
 	PluginsRunner
 
+	// ClusterCache returns a cluster cache.
+	ClusterCache() cache.Cache
+
 	// IterateOverWaitingSubscriptions acquires a read lock and iterates over the WaitingSubscriptions map.
 	IterateOverWaitingSubscriptions(callback func(WaitingSubscription))
 
diff --git a/pkg/scheduler/algorithm/dividing/static.go b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
similarity index 99%
rename from pkg/scheduler/algorithm/dividing/static.go
rename to pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
index b504546..0f915ea 100644
--- a/pkg/scheduler/algorithm/dividing/static.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
@@ -1,4 +1,4 @@
-package dividing
+package defaultassigner
 
 import (
 	"sort"
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/static.go b/pkg/scheduler/framework/plugins/defaultassigner/static.go
new file mode 100644
index 0000000..3773472
--- /dev/null
+++ b/pkg/scheduler/framework/plugins/defaultassigner/static.go
@@ -0,0 +1,70 @@
+/*
+Copyright 2022 The Clusternet Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package defaultassigner
+
+import (
+	"context"
+
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/klog/v2"
+
+	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"
+)
+
+// StaticAssigner assigns replicas to clusters.
+type StaticAssigner struct {
+	handle framework.Handle
+}
+
+var _ framework.AssignPlugin = &StaticAssigner{}
+
+// New creates a DefaultAssigner.
+func New(_ runtime.Object, handle framework.Handle) (framework.Plugin, error) {
+	return &StaticAssigner{handle: handle}, nil
+}
+
+// Name returns the name of the plugin.
+func (pl *StaticAssigner) Name() string {
+	return names.StaticAssigner
+}
+
+// Assign assigns subscriptions to clusters using the clusternet client.
+func (pl *StaticAssigner) Assign(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) (framework.TargetClusters, *framework.Status) {
+	klog.V(3).InfoS("Attempting to assign replicas to clusters",
+		"subscription", klog.KObj(sub), "clusters", availableReplicas.BindingClusters)
+	if sub.Spec.DividingScheduling == nil || sub.Spec.DividingScheduling.Type != appsapi.StaticReplicaDividingType {
+		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)
+	}
+
+	if err = StaticDivideReplicas(&result, sub, clusters, finv); err != nil {
+		return framework.TargetClusters{}, framework.AsStatus(err)
+	}
+
+	return result, nil
+}
diff --git a/pkg/scheduler/framework/plugins/names/names.go b/pkg/scheduler/framework/plugins/names/names.go
index d89ca79..42a285b 100644
--- a/pkg/scheduler/framework/plugins/names/names.go
+++ b/pkg/scheduler/framework/plugins/names/names.go
@@ -17,6 +17,8 @@ limitations under the License.
 package names
 
 const (
+	StaticAssigner = "StaticAssigner"
+
 	DefaultBinder = "DefaultBinder"
 
 	TaintToleration = "TaintToleration"
diff --git a/pkg/scheduler/framework/plugins/registry.go b/pkg/scheduler/framework/plugins/registry.go
index d361f8e..8489524 100644
--- a/pkg/scheduler/framework/plugins/registry.go
+++ b/pkg/scheduler/framework/plugins/registry.go
@@ -17,6 +17,7 @@ limitations under the License.
 package plugins
 
 import (
+	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins/defaultassigner"
 	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins/defaultbinder"
 	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins/names"
 	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins/tainttoleration"
@@ -28,5 +29,6 @@ func NewInTreeRegistry() runtime.Registry {
 	return runtime.Registry{
 		names.DefaultBinder:   defaultbinder.New,
 		names.TaintToleration: tainttoleration.New,
+		names.StaticAssigner:  defaultassigner.New,
 	}
 }
diff --git a/pkg/scheduler/framework/runtime/framework.go b/pkg/scheduler/framework/runtime/framework.go
index 7dc077c..6053dee 100644
--- a/pkg/scheduler/framework/runtime/framework.go
+++ b/pkg/scheduler/framework/runtime/framework.go
@@ -35,9 +35,11 @@ import (
 	clusternet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
 	informers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions"
 	schedulerapis "github.com/clusternet/clusternet/pkg/scheduler/apis"
+	"github.com/clusternet/clusternet/pkg/scheduler/cache"
 	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
 	"github.com/clusternet/clusternet/pkg/scheduler/metrics"
 	"github.com/clusternet/clusternet/pkg/scheduler/parallelize"
+	"github.com/clusternet/clusternet/pkg/utils"
 )
 
 const (
@@ -47,8 +49,12 @@ const (
 	maxTimeout              = 15 * time.Minute
 	preFilter               = "PreFilter"
 	postFilter              = "PostFilter"
+	preEstimate             = "PreEstimate"
+	estimate                = "Estimate"
 	preScore                = "PreScore"
 	score                   = "Score"
+	preAssign               = "PreAssign"
+	assign                  = "Assign"
 	scoreExtensionNormalize = "ScoreExtensionNormalize"
 	preBind                 = "PreBind"
 	bind                    = "Bind"
@@ -66,8 +72,12 @@ type frameworkImpl struct {
 	preFilterPlugins     []framework.PreFilterPlugin
 	filterPlugins        []framework.FilterPlugin
 	postFilterPlugins    []framework.PostFilterPlugin
+	preEstimatePlugins   []framework.PreEstimatePlugin
+	estimatePlugins      []framework.EstimatePlugin
 	preScorePlugins      []framework.PreScorePlugin
 	scorePlugins         []framework.ScorePlugin
+	preAssignPlugins     []framework.PreAssignPlugin
+	assignPlugins        []framework.AssignPlugin
 	reservePlugins       []framework.ReservePlugin
 	preBindPlugins       []framework.PreBindPlugin
 	bindPlugins          []framework.BindPlugin
@@ -78,6 +88,7 @@ type frameworkImpl struct {
 	kubeConfig      *restclient.Config
 	eventRecorder   record.EventRecorder
 	informerFactory informers.SharedInformerFactory
+	cache           cache.Cache
 
 	metricsRecorder *metricsRecorder
 	profileName     string
@@ -106,8 +117,12 @@ func (f *frameworkImpl) getExtensionPoints(plugins *schedulerapis.Plugins) []ext
 		{&plugins.Filter, &f.filterPlugins},
 		{&plugins.PostFilter, &f.postFilterPlugins},
 		{&plugins.Reserve, &f.reservePlugins},
+		{&plugins.PreEstimate, &f.preEstimatePlugins},
+		{&plugins.Estimate, &f.estimatePlugins},
 		{&plugins.PreScore, &f.preScorePlugins},
 		{&plugins.Score, &f.scorePlugins},
+		{&plugins.PreAssign, &f.preAssignPlugins},
+		{&plugins.Assign, &f.assignPlugins},
 		{&plugins.PreBind, &f.preBindPlugins},
 		{&plugins.Bind, &f.bindPlugins},
 		{&plugins.PostBind, &f.postBindPlugins},
@@ -118,6 +133,7 @@ func (f *frameworkImpl) getExtensionPoints(plugins *schedulerapis.Plugins) []ext
 type frameworkOptions struct {
 	clientSet       clusternet.Interface
 	kubeConfig      *restclient.Config
+	cache           cache.Cache
 	eventRecorder   record.EventRecorder
 	informerFactory informers.SharedInformerFactory
 	runAllFilters   bool
@@ -156,6 +172,13 @@ func WithInformerFactory(informerFactory informers.SharedInformerFactory) Option
 	}
 }
 
+// WithCache sets scheduler cache for the scheduling frameworkImpl.
+func WithCache(cache cache.Cache) Option {
+	return func(o *frameworkOptions) {
+		o.cache = cache
+	}
+}
+
 // WithRunAllFilters sets the runAllFilters flag, which means RunFilterPlugins accumulates
 // all failure Statuses.
 func WithRunAllFilters(runAllFilters bool) Option {
@@ -198,6 +221,7 @@ func NewFramework(r Registry, plugins *schedulerapis.Plugins, opts ...Option) (f
 		runAllFilters:        options.runAllFilters,
 		parallelizer:         options.parallelizer,
 		profileName:          "default",
+		cache:                options.cache,
 	}
 
 	if r == nil {
@@ -357,6 +381,68 @@ func (f *frameworkImpl) runPostFilterPlugin(ctx context.Context, pl framework.Po
 	return r, s
 }
 
+func (f *frameworkImpl) RunPreEstimatePlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster) (status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(preEstimate, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	for _, pl := range f.preEstimatePlugins {
+		status = f.runPreEstimatePlugin(ctx, pl, sub, finv, clusters)
+		if !status.IsSuccess() {
+			return framework.AsStatus(fmt.Errorf("running PreEstimate plugin %q: %w", pl.Name(), status.AsError()))
+		}
+	}
+
+	return nil
+}
+
+func (f *frameworkImpl) runPreEstimatePlugin(ctx context.Context, pl framework.PreEstimatePlugin, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster) *framework.Status {
+	startTime := time.Now()
+	status := pl.PreEstimate(ctx, sub, finv, clusters)
+	f.metricsRecorder.observePluginDurationAsync(preEstimate, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return status
+}
+
+func (f *frameworkImpl) RunEstimatePlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, clusters []*clusterapi.ManagedCluster, availableList framework.ClusterScoreList) (res framework.ClusterScoreList, status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(estimate, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	ctx, cancel := context.WithCancel(ctx)
+	errCh := parallelize.NewErrorChannel()
+
+	// Run Estimate method for each cluster in parallel.
+	f.Parallelizer().Until(ctx, len(clusters), func(index int) {
+		for i, pl := range f.estimatePlugins {
+			replicas, status := f.runEstimatePlugin(ctx, pl, sub, finv, clusters[index])
+			if !status.IsSuccess() {
+				err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
+				errCh.SendErrorWithCancel(err, cancel)
+				return
+			}
+			if i == 0 {
+				// First plugin, just set the replicas.
+				availableList[index].MaxAvailableReplicas = replicas
+			} else {
+				// Get the minimum replicas from all estimators.
+				availableList[index].MaxAvailableReplicas = mergeFeedReplicas(availableList[index].MaxAvailableReplicas, replicas)
+			}
+		}
+	})
+	if err := errCh.ReceiveError(); err != nil {
+		return nil, framework.AsStatus(fmt.Errorf("running Estimate plugins: %w", err))
+	}
+
+	return availableList, nil
+}
+
+func (f *frameworkImpl) runEstimatePlugin(ctx context.Context, pl framework.EstimatePlugin, sub *appsapi.Subscription, finv *appsapi.FeedInventory, cluster *clusterapi.ManagedCluster) (framework.FeedReplicas, *framework.Status) {
+	startTime := time.Now()
+	replicas, status := pl.Estimate(ctx, sub, finv, cluster)
+	f.metricsRecorder.observePluginDurationAsync(estimate, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return replicas, status
+}
+
 // RunPreScorePlugins runs the set of configured pre-score plugins. If any
 // of these plugins returns any status other than "Success", the given subscription is rejected.
 func (f *frameworkImpl) RunPreScorePlugins(ctx context.Context, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster) (status *framework.Status) {
@@ -473,6 +559,59 @@ func (f *frameworkImpl) runScoreExtension(ctx context.Context, pl framework.Scor
 	return status
 }
 
+func (f *frameworkImpl) RunPreAssignPlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) (status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(preAssign, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	for _, pl := range f.preAssignPlugins {
+		status = f.runPreAssignPlugin(ctx, pl, sub, finv, availableReplicas)
+		if !status.IsSuccess() {
+			err := status.AsError()
+			klog.ErrorS(err, "Failed running PreAssign plugin", "plugin", pl.Name(), "sub", klog.KObj(sub))
+			return framework.AsStatus(fmt.Errorf("running PreAssign plugin %q: %w", pl.Name(), err))
+		}
+	}
+	return nil
+}
+
+func (f *frameworkImpl) runPreAssignPlugin(ctx context.Context, pl framework.PreAssignPlugin, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) *framework.Status {
+	startTime := time.Now()
+	status := pl.PreAssign(ctx, sub, finv, availableReplicas)
+	f.metricsRecorder.observePluginDurationAsync(preAssign, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return status
+}
+
+func (f *frameworkImpl) RunAssignPlugins(ctx context.Context, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) (result framework.TargetClusters, status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(assign, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	if len(f.assignPlugins) == 0 {
+		return framework.TargetClusters{}, framework.NewStatus(framework.Skip, "empty assign plugins")
+	}
+	for _, ap := range f.assignPlugins {
+		result, status = f.runAssignPlugin(ctx, ap, sub, finv, availableReplicas)
+		if status != nil && status.Code() == framework.Skip {
+			continue
+		}
+		if !status.IsSuccess() {
+			err := status.AsError()
+			klog.ErrorS(err, "Failed running Assign plugin", "plugin", ap.Name(), "sub", klog.KObj(sub))
+			return framework.TargetClusters{}, framework.AsStatus(fmt.Errorf("running Assign plugin %q: %w", ap.Name(), err))
+		}
+		return result, status
+	}
+	return result, status
+}
+
+func (f *frameworkImpl) runAssignPlugin(ctx context.Context, ap framework.AssignPlugin, sub *appsapi.Subscription, finv *appsapi.FeedInventory, availableReplicas framework.TargetClusters) (framework.TargetClusters, *framework.Status) {
+	startTime := time.Now()
+	result, status := ap.Assign(ctx, sub, finv, availableReplicas)
+	f.metricsRecorder.observePluginDurationAsync(assign, ap.Name(), status, metrics.SinceInSeconds(startTime))
+	return result, status
+}
+
 // RunPreBindPlugins runs the set of configured prebind plugins. It returns a
 // failure (bool) if any of the plugins returns an error. It also returns an
 // error containing the rejection message or the error occurred in the plugin.
@@ -711,6 +850,11 @@ func (f *frameworkImpl) HasScorePlugins() bool {
 	return len(f.scorePlugins) > 0
 }
 
+// HasEstimatePlugins returns true if at least one estimate plugin is defined.
+func (f *frameworkImpl) HasEstimatePlugins() bool {
+	return len(f.estimatePlugins) > 0
+}
+
 // ListPlugins returns a map of extension point name to plugin names configured at each extension
 // point. Returns nil if no plugins where configured.
 func (f *frameworkImpl) ListPlugins() *schedulerapis.Plugins {
@@ -765,3 +909,22 @@ func (f *frameworkImpl) ProfileName() string {
 func (f *frameworkImpl) Parallelizer() parallelize.Parallelizer {
 	return f.parallelizer
 }
+
+// ClusterCache returns a cluster cache.
+func (f *frameworkImpl) ClusterCache() cache.Cache {
+	return f.cache
+}
+
+func mergeFeedReplicas(a, b framework.FeedReplicas) framework.FeedReplicas {
+	if len(a) < len(b) {
+		a, b = b, a
+	}
+	res := make(framework.FeedReplicas, len(a))
+	for i := 0; i < len(b); i++ {
+		res[i] = utils.MinInt32(a[i], b[i])
+	}
+	for i := len(b); i < len(a); i++ {
+		res[i] = a[i]
+	}
+	return res
+}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index aa30cd1..c1039ed 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -141,6 +141,7 @@ func NewScheduler(schedulerOptions *options.SchedulerOptions) (*Scheduler, error
 	framework, err := frameworkruntime.NewFramework(sched.registry, getDefaultPlugins(),
 		frameworkruntime.WithEventRecorder(recorder),
 		frameworkruntime.WithInformerFactory(clusternetInformerFactory),
+		frameworkruntime.WithCache(schedulerCache),
 		frameworkruntime.WithClientSet(clusternetClient),
 		frameworkruntime.WithKubeConfig(clientConfig),
 		frameworkruntime.WithParallelism(parallelize.DefaultParallelism),
diff --git a/pkg/utils/slice.go b/pkg/utils/slice.go
index eb18efe..addcdb9 100644
--- a/pkg/utils/slice.go
+++ b/pkg/utils/slice.go
@@ -68,3 +68,31 @@ func RemoveString(slice []string, s string) []string {
 	}
 	return newSlice
 }
+
+func MaxInt(a, b int) int {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func MinInt(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}
+
+func MaxInt32(a, b int32) int32 {
+	if a > b {
+		return a
+	}
+	return b
+}
+
+func MinInt32(a, b int32) int32 {
+	if a < b {
+		return a
+	}
+	return b
+}


Test patch:
diff --git a/pkg/scheduler/algorithm/dividing/static_test.go b/pkg/scheduler/algorithm/dividing/static_test.go
deleted file mode 100644
index 464546a..0000000
--- a/pkg/scheduler/algorithm/dividing/static_test.go
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
-Copyright 2022 The Clusternet Authors.
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package dividing
-
-import (
-	"reflect"
-	"testing"
-)
-
-func TestSortByWeight(t *testing.T) {
-	tests := []struct {
-		name    string
-		weights []int64
-		want    []int
-	}{
-		{
-			weights: []int64{3, 1, 2},
-			want:    []int{0, 2, 1},
-		},
-		{
-			weights: []int64{1, 2, 3, 4},
-			want:    []int{3, 2, 1, 0},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := sortByWeight(tt.weights); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("sortByWeight() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
-
-func TestStaticDivideReplicas(t *testing.T) {
-	tests := []struct {
-		name            string
-		desiredReplicas int32
-		wl              *weightList
-		want            []int32
-	}{
-		{
-			name:            "exact division",
-			desiredReplicas: 5,
-			wl: &weightList{
-				weightSum: 5,
-				weights:   []int64{2, 1, 0, 2},
-			},
-			want: []int32{2, 1, 0, 2},
-		},
-		{
-			name:            "exact division (2 times)",
-			desiredReplicas: 10,
-			wl: &weightList{
-				weightSum: 5,
-				weights:   []int64{2, 1, 2},
-			},
-			want: []int32{4, 2, 4},
-		},
-		{
-			name:            "differential division - 1",
-			desiredReplicas: 6,
-			wl: &weightList{
-				weightSum: 5,
-				weights:   []int64{2, 1, 0, 2},
-			},
-			want: []int32{3, 1, 0, 2},
-		},
-		{
-			name:            "differential division -2 ",
-			desiredReplicas: 5,
-			wl: &weightList{
-				weightSum: 8,
-				weights:   []int64{5, 1, 0, 2},
-			},
-			want: []int32{4, 0, 0, 1},
-		},
-	}
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if got := staticDivideReplicas(tt.desiredReplicas, tt.wl); !reflect.DeepEqual(got, tt.want) {
-				t.Errorf("staticDivideReplicas() = %v, want %v", got, tt.want)
-			}
-		})
-	}
-}
diff --git a/pkg/scheduler/algorithm/generic_test.go b/pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go
similarity index 75%
rename from pkg/scheduler/algorithm/generic_test.go
rename to pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go
index 50b31b6..c628091 100644
--- a/pkg/scheduler/algorithm/generic_test.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/algorithm_test.go
@@ -1,4 +1,20 @@
-package algorithm
+/*
+Copyright 2022 The Clusternet Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package defaultassigner
 
 import (
 	"context"
@@ -16,9 +32,87 @@ import (
 	"github.com/clusternet/clusternet/pkg/known"
 	schedulercache "github.com/clusternet/clusternet/pkg/scheduler/cache"
 	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+	"github.com/clusternet/clusternet/pkg/scheduler/framework/runtime"
 	"github.com/clusternet/clusternet/pkg/utils"
 )
 
+func TestSortByWeight(t *testing.T) {
+	tests := []struct {
+		name    string
+		weights []int64
+		want    []int
+	}{
+		{
+			weights: []int64{3, 1, 2},
+			want:    []int{0, 2, 1},
+		},
+		{
+			weights: []int64{1, 2, 3, 4},
+			want:    []int{3, 2, 1, 0},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := sortByWeight(tt.weights); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("sortByWeight() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestStaticDivideReplicas(t *testing.T) {
+	tests := []struct {
+		name            string
+		desiredReplicas int32
+		wl              *weightList
+		want            []int32
+	}{
+		{
+			name:            "exact division",
+			desiredReplicas: 5,
+			wl: &weightList{
+				weightSum: 5,
+				weights:   []int64{2, 1, 0, 2},
+			},
+			want: []int32{2, 1, 0, 2},
+		},
+		{
+			name:            "exact division (2 times)",
+			desiredReplicas: 10,
+			wl: &weightList{
+				weightSum: 5,
+				weights:   []int64{2, 1, 2},
+			},
+			want: []int32{4, 2, 4},
+		},
+		{
+			name:            "differential division - 1",
+			desiredReplicas: 6,
+			wl: &weightList{
+				weightSum: 5,
+				weights:   []int64{2, 1, 0, 2},
+			},
+			want: []int32{3, 1, 0, 2},
+		},
+		{
+			name:            "differential division -2 ",
+			desiredReplicas: 5,
+			wl: &weightList{
+				weightSum: 8,
+				weights:   []int64{5, 1, 0, 2},
+			},
+			want: []int32{4, 0, 0, 1},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := staticDivideReplicas(tt.desiredReplicas, tt.wl); !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("staticDivideReplicas() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
 func newClusterWithLabels(name, namespace string, labels map[string]string) *clusterapi.ManagedCluster {
 	return &clusterapi.ManagedCluster{
 		ObjectMeta: metav1.ObjectMeta{
@@ -303,22 +397,24 @@ func TestDivideReplicas(t *testing.T) {
 			clusternetInformerFactory := informers.NewSharedInformerFactory(clusternetClient, known.DefaultResync)
 			clusterInformer := clusternetInformerFactory.Clusters().V1beta1().ManagedClusters()
 
-			g := &genericScheduler{
-				cache: schedulercache.New(clusterInformer.Lister()),
+			fwk, _ := runtime.NewFramework(nil, nil, runtime.WithCache(schedulercache.New(clusterInformer.Lister())))
+			assigner, err := New(nil, fwk)
+			if err != nil {
+				t.Fatalf("Creating plugin error: %v", err)
 			}
 
 			clusternetInformerFactory.Start(ctx.Done())
 			if !cache.WaitForCacheSync(ctx.Done(), clusterInformer.Informer().HasSynced) {
-				t.Fatalf("divideReplicas() error = %v, wantErr %v", fmt.Errorf("failed to wait for cache sync"), tt.wantErr)
+				t.Fatalf("Assign() error = %v, wantErr %v", fmt.Errorf("failed to wait for cache sync"), tt.wantErr)
 			}
 
-			if err := g.divideReplicas(tt.args.selected, tt.args.sub, tt.args.finv); (err != nil) != tt.wantErr {
-				t.Errorf("divideReplicas() error = %v, wantErr %v", err, tt.wantErr)
+			result, status := assigner.(framework.AssignPlugin).Assign(ctx, tt.args.sub, tt.args.finv, *tt.args.selected)
+			if (status != nil) != tt.wantErr {
+				t.Errorf("Assign() error = %v, wantErr %v", err, tt.wantErr)
 			}
-			if !reflect.DeepEqual(tt.args.selected.Replicas, tt.wantReplicas) {
-				t.Errorf("divideReplicas() gotResponse = %v, want %v", tt.args.selected.Replicas, tt.wantReplicas)
+			if !reflect.DeepEqual(result.Replicas, tt.wantReplicas) {
+				t.Errorf("Assign() gotResponse = %v, want %v", result.Replicas, tt.wantReplicas)
 			}
-
 		})
 	}
 }