Repository: clusternet/clusternet
Pull request #292 (2022-04-06 08:12:58)

add static replica dividing
<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?

kind/feature
kind/api
kind/change

#### 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/292?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) Report
> Merging [#292](https://codecov.io/gh/clusternet/clusternet/pull/292?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (19ab889) into [main](https://codecov.io/gh/clusternet/clusternet/commit/0f5de61e9c8149d62a4cd6eec7319f16b1a2978b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (0f5de61) will **decrease** coverage by `0.34%`.
> The diff coverage is `n/a`.

```diff
@@            Coverage Diff             @@
##             main     #292      +/-   ##
==========================================
- Coverage   10.53%   10.18%   -0.35%     
==========================================
  Files          39       40       +1     
  Lines        4283     4507     +224     
==========================================
+ Hits          451      459       +8     
- Misses       3800     4015     +215     
- Partials       32       33       +1     
```


| [Impacted Files](https://codecov.io/gh/clusternet/clusternet/pull/292?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) | Coverage Δ | |
|---|---|---|
| [...github.com/clusternet/clusternet/pkg/utils/feed.go](https://codecov.io/gh/clusternet/clusternet/pull/292/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy91dGlscy9mZWVkLmdv) | `31.48% <0.00%> (-1.86%)` | :arrow_down: |
| [.../framework/plugins/defaultbinder/default\_binder.go](https://codecov.io/gh/clusternet/clusternet/pull/292/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvZnJhbWV3b3JrL3BsdWdpbnMvZGVmYXVsdGJpbmRlci9kZWZhdWx0X2JpbmRlci5nbw==) | `81.81% <0.00%> (ø)` | |
| [...rnet/clusternet/pkg/scheduler/algorithm/generic.go](https://codecov.io/gh/clusternet/clusternet/pull/292/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet#diff-c3JjL2dpdGh1Yi5jb20vY2x1c3Rlcm5ldC9jbHVzdGVybmV0L3BrZy9zY2hlZHVsZXIvYWxnb3JpdGhtL2dlbmVyaWMuZ28=) | `3.66% <0.00%> (ø)` | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/clusternet/clusternet/pull/292?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/292?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Last update [0f5de61...19ab889](https://codecov.io/gh/clusternet/clusternet/pull/292?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/manifests/crds/apps.clusternet.io_subscriptions.yaml b/manifests/crds/apps.clusternet.io_subscriptions.yaml
index 93f258e..31627ba 100644
--- a/manifests/crds/apps.clusternet.io_subscriptions.yaml
+++ b/manifests/crds/apps.clusternet.io_subscriptions.yaml
@@ -192,7 +192,7 @@ spec:
                       description: Static weight of subscriber when dividing replicas.
                         Present only for static divided scheduling.
                       format: int32
-                      minimum: 1
+                      minimum: 0
                       type: integer
                   required:
                   - clusterAffinity
@@ -218,12 +218,13 @@ spec:
                 description: Total number of Helm releases desired by this Subscription.
                 type: integer
               replicas:
-                description: Desired replicas of targeted clusters for each feed.
-                items:
-                  additionalProperties:
+                additionalProperties:
+                  items:
+                    format: int32
                     type: integer
-                  type: object
-                type: array
+                  type: array
+                description: Desired replicas of targeted clusters for each feed.
+                type: object
               specHash:
                 description: SpecHash calculates the hash value of current SubscriptionSpec.
                 format: int64
diff --git a/manifests/crds/clusters.clusternet.io_managedclusters.yaml b/manifests/crds/clusters.clusternet.io_managedclusters.yaml
index 01d630e..f4ad494 100644
--- a/manifests/crds/clusters.clusternet.io_managedclusters.yaml
+++ b/manifests/crds/clusters.clusternet.io_managedclusters.yaml
@@ -37,8 +37,8 @@ spec:
     - jsonPath: .status.k8sVersion
       name: KUBERNETES
       type: string
-    - jsonPath: .status.conditions[?(@.type=='Ready')].status
-      name: STATUS
+    - jsonPath: .status.readyz
+      name: READYZ
       type: string
     - jsonPath: .metadata.creationTimestamp
       name: AGE
diff --git a/pkg/apis/apps/v1alpha1/subscription.go b/pkg/apis/apps/v1alpha1/subscription.go
index a24b9e9..5152b7c 100644
--- a/pkg/apis/apps/v1alpha1/subscription.go
+++ b/pkg/apis/apps/v1alpha1/subscription.go
@@ -89,7 +89,7 @@ type SubscriptionStatus struct {
 	// Desired replicas of targeted clusters for each feed.
 	//
 	// +optional
-	Replicas []map[string]int `json:"replicas,omitempty"`
+	Replicas map[string][]int32 `json:"replicas,omitempty"`
 
 	// SpecHash calculates the hash value of current SubscriptionSpec.
 	//
@@ -119,8 +119,8 @@ type Subscriber struct {
 	// Present only for static divided scheduling.
 	//
 	// +optional
-	// +kubebuilder:validation:Minimum=1
-	Weight *int32 `json:"weight,omitempty"`
+	// +kubebuilder:validation:Minimum=0
+	Weight int32 `json:"weight,omitempty"`
 }
 
 // Feed defines the resource to be selected.
diff --git a/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go
index 6e8daae..62575f9 100644
--- a/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go
@@ -890,11 +890,6 @@ func (in *Subscriber) DeepCopyInto(out *Subscriber) {
 		*out = new(v1.LabelSelector)
 		(*in).DeepCopyInto(*out)
 	}
-	if in.Weight != nil {
-		in, out := &in.Weight, &out.Weight
-		*out = new(int32)
-		**out = **in
-	}
 	return
 }
 
@@ -1019,15 +1014,17 @@ func (in *SubscriptionStatus) DeepCopyInto(out *SubscriptionStatus) {
 	}
 	if in.Replicas != nil {
 		in, out := &in.Replicas, &out.Replicas
-		*out = make([]map[string]int, len(*in))
-		for i := range *in {
-			if (*in)[i] != nil {
-				in, out := &(*in)[i], &(*out)[i]
-				*out = make(map[string]int, len(*in))
-				for key, val := range *in {
-					(*out)[key] = val
-				}
+		*out = make(map[string][]int32, len(*in))
+		for key, val := range *in {
+			var outVal []int32
+			if val == nil {
+				(*out)[key] = nil
+			} else {
+				in, out := &val, &outVal
+				*out = make([]int32, len(*in))
+				copy(*out, *in)
 			}
+			(*out)[key] = outVal
 		}
 	}
 	return
diff --git a/pkg/apis/clusters/v1beta1/types.go b/pkg/apis/clusters/v1beta1/types.go
index 7ccf59d..6bd83d4 100644
--- a/pkg/apis/clusters/v1beta1/types.go
+++ b/pkg/apis/clusters/v1beta1/types.go
@@ -282,7 +282,7 @@ type ManagedClusterStatus struct {
 // +kubebuilder:printcolumn:name="CLUSTER TYPE",type=string,JSONPath=`.spec.clusterType`,description="The type of the cluster",priority=100
 // +kubebuilder:printcolumn:name="SYNC MODE",type=string,JSONPath=`.spec.syncMode`,description="The cluster sync mode"
 // +kubebuilder:printcolumn:name="KUBERNETES",type=string,JSONPath=".status.k8sVersion"
-// +kubebuilder:printcolumn:name="STATUS",type=string,JSONPath=".status.conditions[?(@.type=='Ready')].status"
+// +kubebuilder:printcolumn:name="READYZ",type=string,JSONPath=".status.readyz"
 // +kubebuilder:printcolumn:name="AGE",type="date",JSONPath=".metadata.creationTimestamp"
 
 // ManagedCluster is the Schema for the managedclusters API
diff --git a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
index a7e6ac1..c195157 100644
--- a/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
+++ b/pkg/controllers/clusters/clusterstatus/clusterstatus_controller.go
@@ -169,7 +169,7 @@ func (c *Controller) getHealthStatus(ctx context.Context, path string) bool {
 }
 
 func (c *Controller) getCondition(status clusterapi.ManagedClusterStatus) metav1.Condition {
-	if (status.Livez && status.Readyz) || status.Healthz {
+	if status.Livez && status.Readyz {
 		return metav1.Condition{
 			Type:               clusterapi.ClusterReady,
 			Status:             metav1.ConditionTrue,
diff --git a/pkg/hub/deployer/deployer.go b/pkg/hub/deployer/deployer.go
index fe55b9b..dc8c240 100644
--- a/pkg/hub/deployer/deployer.go
+++ b/pkg/hub/deployer/deployer.go
@@ -583,9 +583,18 @@ func (deployer *Deployer) populateDescriptions(base *appsapi.Base) error {
 			Finalizers: []string{
 				known.AppFinalizer,
 			},
+			OwnerReferences: []metav1.OwnerReference{
+				{
+					APIVersion:         baseKind.Version,
+					Kind:               baseKind.Kind,
+					Name:               base.Name,
+					UID:                base.UID,
+					Controller:         utilpointer.BoolPtr(true),
+					BlockOwnerDeletion: utilpointer.BoolPtr(true),
+				},
+			},
 		},
 	}
-	descTemplate.SetOwnerReferences([]metav1.OwnerReference{*metav1.NewControllerRef(base, baseKind)})
 
 	var allErrs []error
 	if len(allChartRefs) > 0 {
diff --git a/pkg/hub/deployer/helm/helm.go b/pkg/hub/deployer/helm/helm.go
index b6273c9..db6b32e 100644
--- a/pkg/hub/deployer/helm/helm.go
+++ b/pkg/hub/deployer/helm/helm.go
@@ -291,6 +291,16 @@ func (deployer *Deployer) populateHelmRelease(desc *appsapi.Description) error {
 				Finalizers: []string{
 					known.AppFinalizer,
 				},
+				OwnerReferences: []metav1.OwnerReference{
+					{
+						APIVersion:         descriptionKind.Version,
+						Kind:               descriptionKind.Kind,
+						Name:               desc.Name,
+						UID:                desc.UID,
+						Controller:         utilpointer.BoolPtr(true),
+						BlockOwnerDeletion: utilpointer.BoolPtr(true),
+					},
+				},
 			},
 			Spec: appsapi.HelmReleaseSpec{
 				ReleaseName:     utilpointer.String(chart.Name), // default to be the HelmChart name
@@ -299,7 +309,6 @@ func (deployer *Deployer) populateHelmRelease(desc *appsapi.Description) error {
 				Overrides:       desc.Spec.Raw[idx],
 			},
 		}
-		hr.SetOwnerReferences([]metav1.OwnerReference{*metav1.NewControllerRef(desc, descriptionKind)})
 		hrsToBeDeleted.Delete(klog.KObj(hr).String())
 
 		err = deployer.syncHelmRelease(desc, hr)
diff --git a/pkg/scheduler/algorithm/dividing/static.go b/pkg/scheduler/algorithm/dividing/static.go
new file mode 100644
index 0000000..fe85116
--- /dev/null
+++ b/pkg/scheduler/algorithm/dividing/static.go
@@ -0,0 +1,114 @@
+package dividing
+
+import (
+	"sort"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+
+	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/utils"
+)
+
+type weightList struct {
+	weightSum int64
+	weights   []int64
+}
+
+// StaticDivideReplicas will fill the target replicas of all feeds based on a weight list and feed finv.
+func StaticDivideReplicas(selected *framework.TargetClusters, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster, finv *appsapi.FeedInventory) error {
+	wl := getWeightList(sub.Spec.Subscribers, clusters)
+	if wl.weightSum == 0 {
+		return nil
+	}
+	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
+			}
+		}
+		if order != nil && order.DesiredReplicas != nil {
+			replicas := staticDivideReplicas(*order.DesiredReplicas, &wl)
+			selected.Replicas[utils.GetFeedKey(&order.Feed)] = replicas
+		} else {
+			selected.Replicas[utils.GetFeedKey(&order.Feed)] = []int32{}
+		}
+	}
+	return nil
+}
+
+// staticDivideReplicas divides replicas by the weight list.
+func staticDivideReplicas(desiredReplicas int32, wl *weightList) []int32 {
+	res := make([]int32, len(wl.weights))
+	remain := wl.weightSum
+	for i, weight := range wl.weights {
+		replica := weight * int64(desiredReplicas) / wl.weightSum
+		res[i] = int32(replica)
+		remain -= replica
+	}
+	if remain == 0 {
+		return res
+	}
+
+	remainInt := int(remain)
+	clusterIndices := sortByWeight(wl.weights)
+	if remainInt <= len(clusterIndices) {
+		for i := 0; i < remainInt; i++ {
+			res[clusterIndices[i]]++
+		}
+	} else {
+		avg, bounder := remainInt/len(clusterIndices), remainInt%len(clusterIndices)
+		for i := range clusterIndices {
+			if i < bounder {
+				res[clusterIndices[i]] += int32(avg) + 1
+			} else {
+				res[clusterIndices[i]] += int32(avg)
+			}
+		}
+	}
+	return res
+}
+
+func sortByWeight(weights []int64) []int {
+	clusterIndices := make([]int, len(weights))
+	for i := range weights {
+		clusterIndices[i] = i
+	}
+	sort.Slice(clusterIndices, func(i, j int) bool {
+		return weights[clusterIndices[i]] > weights[clusterIndices[j]]
+	})
+	return clusterIndices
+}
+
+// getWeightList analyzes weight from subscribers and form a weightList based on the sequence of clusters.
+func getWeightList(subscribers []appsapi.Subscriber, clusters []*clusterapi.ManagedCluster) weightList {
+	res := weightList{
+		weights: make([]int64, 0, len(clusters)),
+	}
+	for _, cluster := range clusters {
+		var clusterWeightSum int64
+		for _, subscriber := range subscribers {
+			if subscriber.Weight <= 0 {
+				continue
+			}
+			selector, err := metav1.LabelSelectorAsSelector(subscriber.ClusterAffinity)
+			if err != nil {
+				continue
+			}
+			if !selector.Matches(labels.Set(cluster.Labels)) {
+				continue
+			}
+			clusterWeightSum += int64(subscriber.Weight)
+			res.weightSum += int64(subscriber.Weight)
+		}
+		res.weights = append(res.weights, clusterWeightSum)
+	}
+	return res
+}
diff --git a/pkg/scheduler/algorithm/generic.go b/pkg/scheduler/algorithm/generic.go
index a3c0dd0..bfbad3a 100644
--- a/pkg/scheduler/algorithm/generic.go
+++ b/pkg/scheduler/algorithm/generic.go
@@ -31,6 +31,7 @@ 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"
@@ -64,7 +65,7 @@ type genericScheduler struct {
 // Schedule tries to schedule the given subscription to multiple clusters.
 // If it succeeds, it will return the namespaced names of ManagedClusters.
 // If it fails, it will return a FitError error with reasons.
-func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework, sub *appsapi.Subscription) (result ScheduleResult, err error) {
+func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework, sub *appsapi.Subscription, finv *appsapi.FeedInventory) (result ScheduleResult, err error) {
 	trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: sub.Namespace}, utiltrace.Field{Key: "name", Value: sub.Name})
 	defer trace.LogIfLong(100 * time.Millisecond)
 
@@ -91,7 +92,7 @@ func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework
 		return result, err
 	}
 
-	clusters, err := g.selectClusters(priorityList, sub)
+	clusters, err := g.selectClusters(priorityList, sub, finv)
 	trace.Step("Prioritizing done")
 
 	return ScheduleResult{
@@ -103,16 +104,23 @@ func (g *genericScheduler) Schedule(ctx context.Context, fwk framework.Framework
 
 // 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, _ *appsapi.Subscription) ([]string, error) {
+func (g *genericScheduler) selectClusters(clusterScoreList framework.ClusterScoreList, sub *appsapi.Subscription, finv *appsapi.FeedInventory) (framework.TargetClusters, error) {
 	if len(clusterScoreList) == 0 {
-		return nil, fmt.Errorf("empty clusterScoreList")
+		return framework.TargetClusters{}, fmt.Errorf("empty clusterScoreList")
 	}
 
-	var selected []string
+	var selected framework.TargetClusters
 	for _, clusterScore := range clusterScoreList {
 		// TODO: sampling with scores
-		selected = append(selected, clusterScore.NamespacedName)
+		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)
+		}
+	}
+
 	return selected, nil
 }
 
@@ -255,6 +263,25 @@ 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
diff --git a/pkg/scheduler/algorithm/interface.go b/pkg/scheduler/algorithm/interface.go
index 077aa2e..2ecbc22 100644
--- a/pkg/scheduler/algorithm/interface.go
+++ b/pkg/scheduler/algorithm/interface.go
@@ -26,14 +26,14 @@ import (
 // ScheduleAlgorithm is an interface implemented by things that know how to schedule resources to target
 // managed clusters.
 type ScheduleAlgorithm interface {
-	Schedule(context.Context, framework.Framework, *appsapi.Subscription) (scheduleResult ScheduleResult, err error)
+	Schedule(context.Context, framework.Framework, *appsapi.Subscription, *appsapi.FeedInventory) (scheduleResult ScheduleResult, err error)
 }
 
 // ScheduleResult represents the result of one subscription scheduled. It will contain
 // the final selected clusters, along with the selected intermediate information.
 type ScheduleResult struct {
 	// the scheduler suggest clusters (namespaced name)
-	SuggestedClusters []string
+	SuggestedClusters framework.TargetClusters
 	// Number of clusters scheduler evaluated on one subscription scheduled
 	EvaluatedClusters int
 	// Number of feasible clusters on one subscription scheduled
diff --git a/pkg/scheduler/framework/interfaces/interface.go b/pkg/scheduler/framework/interfaces/interface.go
index 12281cd..1247449 100644
--- a/pkg/scheduler/framework/interfaces/interface.go
+++ b/pkg/scheduler/framework/interfaces/interface.go
@@ -217,13 +217,13 @@ type ReservePlugin interface {
 	// Reserve is called by the scheduling framework when the scheduler cache is
 	// updated. If this method returns a failed Status, the scheduler will call
 	// the Unreserve method for all enabled ReservePlugins.
-	Reserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	Reserve(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 	// Unreserve is called by the scheduling framework when a reserved subscription was
 	// rejected, an error occurred during reservation of subsequent plugins, or
 	// in a later phase. The Unreserve method implementation must be idempotent
 	// and may be called by the scheduler even if the corresponding Reserve
 	// method for the same plugin was not called.
-	Unreserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string)
+	Unreserve(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters)
 }
 
 // PreBindPlugin is an interface that must be implemented by "PreBind" plugins.
@@ -233,7 +233,7 @@ type PreBindPlugin interface {
 
 	// PreBind is called before binding a subscription. All prebind plugins must return
 	// success or the subscription will be rejected and won't be sent for binding.
-	PreBind(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	PreBind(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 }
 
 // PostBindPlugin is an interface that must be implemented by "PostBind" plugins.
@@ -245,7 +245,7 @@ type PostBindPlugin interface {
 	// informational. A common application of this extension point is for cleaning
 	// up. If a plugin needs to clean-up its state after a subscription is scheduled and
 	// bound, PostBind is the extension point that it should register.
-	PostBind(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string)
+	PostBind(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters)
 }
 
 // PermitPlugin is an interface that must be implemented by "Permit" plugins.
@@ -259,7 +259,7 @@ type PermitPlugin interface {
 	// The subscription will also be rejected if the wait timeout or the subscription is rejected while
 	// waiting. Note that if the plugin returns "wait", the framework will wait only
 	// after running the remaining plugins given that no other plugin rejects the subscription.
-	Permit(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) (*Status, time.Duration)
+	Permit(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) (*Status, time.Duration)
 }
 
 // BindPlugin is an interface that must be implemented by "Bind" plugins. Bind
@@ -273,7 +273,7 @@ type BindPlugin interface {
 	// remaining bind plugins are skipped. When a bind plugin does not handle a Subscription,
 	// it must return Skip in its Status code. If a bind plugin returns an Error, the
 	// subscription is rejected and will not be bound.
-	Bind(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	Bind(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 }
 
 // Framework manages the set of plugins in use by the scheduling framework.
@@ -297,11 +297,11 @@ type Framework interface {
 	// configured Reserve plugins. If any of these calls returns an error, it
 	// does not continue running the remaining ones and returns the error. In
 	// such case, subscription will not be scheduled.
-	RunReservePluginsReserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	RunReservePluginsReserve(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 
 	// RunReservePluginsUnreserve runs the Unreserve method of the set of
 	// configured Reserve plugins.
-	RunReservePluginsUnreserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string)
+	RunReservePluginsUnreserve(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters)
 
 	// RunPermitPlugins runs the set of configured Permit plugins. If any of these
 	// plugins returns a status other than "Success" or "Wait", it does not continue
@@ -309,7 +309,7 @@ type Framework interface {
 	// plugins returns "Wait", then this function will create and add waiting subscription
 	// to a map of currently waiting subscriptions and return status with "Wait" code.
 	// Subscription will remain waiting subscription for the minimum duration returned by the Permit plugins.
-	RunPermitPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	RunPermitPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 
 	// WaitOnPermit will block, if the subscription is a waiting subscription, until the waiting subscription is rejected or allowed.
 	WaitOnPermit(ctx context.Context, sub *appsapi.Subscription) *Status
@@ -319,17 +319,17 @@ type Framework interface {
 	// anything but Success. If the Status code is Unschedulable, it is
 	// considered as a scheduling check failure, otherwise, it is considered as an
 	// internal error. In either case the subscription is not going to be bound.
-	RunPreBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	RunPreBindPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 
 	// RunPostBindPlugins runs the set of configured PostBind plugins.
-	RunPostBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string)
+	RunPostBindPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters)
 
 	// RunBindPlugins runs the set of configured Bind plugins. A Bind plugin may choose
 	// whether or not to handle the given subscription. If a Bind plugin chooses to skip the
 	// binding, it should return code=5("skip") status. Otherwise, it should return "Error"
 	// or "Success". If none of the plugins handled binding, RunBindPlugins returns
 	// code=5("skip") status.
-	RunBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+	RunBindPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters TargetClusters) *Status
 
 	// HasFilterPlugins returns true if at least one Filter plugin is defined.
 	HasFilterPlugins() bool
@@ -380,7 +380,7 @@ type Handle interface {
 
 // PostFilterResult wraps needed info for scheduler framework to act upon PostFilter phase.
 type PostFilterResult struct {
-	NominatedNamespacedClusters []string
+	NominatedtargetClusters TargetClusters
 }
 
 // PluginsRunner abstracts operations to run some plugins.
diff --git a/pkg/scheduler/framework/interfaces/types.go b/pkg/scheduler/framework/interfaces/types.go
index cf083f9..d4bca97 100644
--- a/pkg/scheduler/framework/interfaces/types.go
+++ b/pkg/scheduler/framework/interfaces/types.go
@@ -39,6 +39,32 @@ type FitError struct {
 	Diagnosis      Diagnosis
 }
 
+// TargetClusters represents the scheduling result.
+type TargetClusters struct {
+	// Namespaced names of targeted clusters that Subscription binds to.
+	BindingClusters []string
+
+	// Desired replicas of targeted clusters for each feed.
+	Replicas map[string][]int32
+}
+
+func (t TargetClusters) Len() int {
+	return len(t.BindingClusters)
+}
+
+func (t TargetClusters) Less(i, j int) bool {
+	return t.BindingClusters[i] < t.BindingClusters[j]
+}
+
+func (t TargetClusters) Swap(i, j int) {
+	t.BindingClusters[i], t.BindingClusters[j] = t.BindingClusters[j], t.BindingClusters[i]
+	for _, replicas := range t.Replicas {
+		if len(replicas) == len(t.BindingClusters) {
+			replicas[i], replicas[j] = replicas[j], replicas[i]
+		}
+	}
+}
+
 const (
 	// NoClusterAvailableMsg is used to format message when no clusters available.
 	NoClusterAvailableMsg = "0/%v clusters are available"
diff --git a/pkg/scheduler/framework/plugins/defaultbinder/default_binder.go b/pkg/scheduler/framework/plugins/defaultbinder/default_binder.go
index 3332fff..021801f 100644
--- a/pkg/scheduler/framework/plugins/defaultbinder/default_binder.go
+++ b/pkg/scheduler/framework/plugins/defaultbinder/default_binder.go
@@ -48,19 +48,18 @@ func (pl *DefaultBinder) Name() string {
 }
 
 // Bind binds subscriptions to clusters using the clusternet client.
-func (pl *DefaultBinder) Bind(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *framework.Status {
+func (pl *DefaultBinder) Bind(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) *framework.Status {
 	klog.V(3).InfoS("Attempting to bind subscription to clusters",
-		"subscription", klog.KObj(sub), "clusters", namespacedClusters)
+		"subscription", klog.KObj(sub), "clusters", targetClusters.BindingClusters)
 
 	// use an ordered list
-	sort.SliceStable(namespacedClusters, func(i, j int) bool {
-		return namespacedClusters[i] < namespacedClusters[j]
-	})
+	sort.Stable(targetClusters)
 
 	subCopy := sub.DeepCopy()
-	subCopy.Status.BindingClusters = namespacedClusters
+	subCopy.Status.BindingClusters = targetClusters.BindingClusters
+	subCopy.Status.Replicas = targetClusters.Replicas
 	subCopy.Status.SpecHash = utils.HashSubscriptionSpec(&subCopy.Spec)
-	subCopy.Status.DesiredReleases = len(namespacedClusters)
+	subCopy.Status.DesiredReleases = len(targetClusters.BindingClusters)
 
 	_, err := pl.handle.ClientSet().AppsV1alpha1().Subscriptions(sub.Namespace).UpdateStatus(ctx, subCopy, metav1.UpdateOptions{})
 	if err != nil {
diff --git a/pkg/scheduler/framework/runtime/framework.go b/pkg/scheduler/framework/runtime/framework.go
index f686afa..7dc077c 100644
--- a/pkg/scheduler/framework/runtime/framework.go
+++ b/pkg/scheduler/framework/runtime/framework.go
@@ -476,13 +476,13 @@ func (f *frameworkImpl) runScoreExtension(ctx context.Context, pl framework.Scor
 // 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.
-func (f *frameworkImpl) RunPreBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) (status *framework.Status) {
+func (f *frameworkImpl) RunPreBindPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) (status *framework.Status) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(preBind, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
 	}()
 	for _, pl := range f.preBindPlugins {
-		status = f.runPreBindPlugin(ctx, pl, sub, namespacedClusters)
+		status = f.runPreBindPlugin(ctx, pl, sub, targetClusters)
 		if !status.IsSuccess() {
 			err := status.AsError()
 			klog.ErrorS(err, "Failed running PreBind plugin", "plugin", pl.Name(), "sub", klog.KObj(sub))
@@ -492,15 +492,15 @@ func (f *frameworkImpl) RunPreBindPlugins(ctx context.Context, sub *appsapi.Subs
 	return nil
 }
 
-func (f *frameworkImpl) runPreBindPlugin(ctx context.Context, pl framework.PreBindPlugin, sub *appsapi.Subscription, namespacedClusters []string) *framework.Status {
+func (f *frameworkImpl) runPreBindPlugin(ctx context.Context, pl framework.PreBindPlugin, sub *appsapi.Subscription, targetClusters framework.TargetClusters) *framework.Status {
 	startTime := time.Now()
-	status := pl.PreBind(ctx, sub, namespacedClusters)
+	status := pl.PreBind(ctx, sub, targetClusters)
 	f.metricsRecorder.observePluginDurationAsync(preBind, pl.Name(), status, metrics.SinceInSeconds(startTime))
 	return status
 }
 
 // RunBindPlugins runs the set of configured bind plugins until one returns a non `Skip` status.
-func (f *frameworkImpl) RunBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) (status *framework.Status) {
+func (f *frameworkImpl) RunBindPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) (status *framework.Status) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(bind, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
@@ -509,7 +509,7 @@ func (f *frameworkImpl) RunBindPlugins(ctx context.Context, sub *appsapi.Subscri
 		return framework.NewStatus(framework.Skip, "")
 	}
 	for _, bp := range f.bindPlugins {
-		status = f.runBindPlugin(ctx, bp, sub, namespacedClusters)
+		status = f.runBindPlugin(ctx, bp, sub, targetClusters)
 		if status != nil && status.Code() == framework.Skip {
 			continue
 		}
@@ -523,27 +523,27 @@ func (f *frameworkImpl) RunBindPlugins(ctx context.Context, sub *appsapi.Subscri
 	return status
 }
 
-func (f *frameworkImpl) runBindPlugin(ctx context.Context, bp framework.BindPlugin, sub *appsapi.Subscription, namespacedClusters []string) *framework.Status {
+func (f *frameworkImpl) runBindPlugin(ctx context.Context, bp framework.BindPlugin, sub *appsapi.Subscription, targetClusters framework.TargetClusters) *framework.Status {
 	startTime := time.Now()
-	status := bp.Bind(ctx, sub, namespacedClusters)
+	status := bp.Bind(ctx, sub, targetClusters)
 	f.metricsRecorder.observePluginDurationAsync(bind, bp.Name(), status, metrics.SinceInSeconds(startTime))
 	return status
 }
 
 // RunPostBindPlugins runs the set of configured postbind plugins.
-func (f *frameworkImpl) RunPostBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) {
+func (f *frameworkImpl) RunPostBindPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(postBind, framework.Success.String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
 	}()
 	for _, pl := range f.postBindPlugins {
-		f.runPostBindPlugin(ctx, pl, sub, namespacedClusters)
+		f.runPostBindPlugin(ctx, pl, sub, targetClusters)
 	}
 }
 
-func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.PostBindPlugin, sub *appsapi.Subscription, namespacedClusters []string) {
+func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.PostBindPlugin, sub *appsapi.Subscription, targetClusters framework.TargetClusters) {
 	startTime := time.Now()
-	pl.PostBind(ctx, sub, namespacedClusters)
+	pl.PostBind(ctx, sub, targetClusters)
 	f.metricsRecorder.observePluginDurationAsync(postBind, pl.Name(), nil, metrics.SinceInSeconds(startTime))
 }
 
@@ -552,13 +552,13 @@ func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.Post
 // continue running the remaining ones and returns the error. In such a case,
 // the subscription will not be scheduled and the caller will be expected to call
 // RunReservePluginsUnreserve.
-func (f *frameworkImpl) RunReservePluginsReserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) (status *framework.Status) {
+func (f *frameworkImpl) RunReservePluginsReserve(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) (status *framework.Status) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(reserve, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
 	}()
 	for _, pl := range f.reservePlugins {
-		status = f.runReservePluginReserve(ctx, pl, sub, namespacedClusters)
+		status = f.runReservePluginReserve(ctx, pl, sub, targetClusters)
 		if !status.IsSuccess() {
 			err := status.AsError()
 			klog.ErrorS(err, "Failed running Reserve plugin", "plugin", pl.Name(), "subscription", klog.KObj(sub))
@@ -568,16 +568,16 @@ func (f *frameworkImpl) RunReservePluginsReserve(ctx context.Context, sub *appsa
 	return nil
 }
 
-func (f *frameworkImpl) runReservePluginReserve(ctx context.Context, pl framework.ReservePlugin, sub *appsapi.Subscription, namespacedClusters []string) *framework.Status {
+func (f *frameworkImpl) runReservePluginReserve(ctx context.Context, pl framework.ReservePlugin, sub *appsapi.Subscription, targetClusters framework.TargetClusters) *framework.Status {
 	startTime := time.Now()
-	status := pl.Reserve(ctx, sub, namespacedClusters)
+	status := pl.Reserve(ctx, sub, targetClusters)
 	f.metricsRecorder.observePluginDurationAsync(reserve, pl.Name(), status, metrics.SinceInSeconds(startTime))
 	return status
 }
 
 // RunReservePluginsUnreserve runs the Unreserve method in the set of
 // configured reserve plugins.
-func (f *frameworkImpl) RunReservePluginsUnreserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) {
+func (f *frameworkImpl) RunReservePluginsUnreserve(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(unreserve, framework.Success.String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
@@ -585,13 +585,13 @@ func (f *frameworkImpl) RunReservePluginsUnreserve(ctx context.Context, sub *app
 	// Execute the Unreserve operation of each reserve plugin in the
 	// *reverse* order in which the Reserve operation was executed.
 	for i := len(f.reservePlugins) - 1; i >= 0; i-- {
-		f.runReservePluginUnreserve(ctx, f.reservePlugins[i], sub, namespacedClusters)
+		f.runReservePluginUnreserve(ctx, f.reservePlugins[i], sub, targetClusters)
 	}
 }
 
-func (f *frameworkImpl) runReservePluginUnreserve(ctx context.Context, pl framework.ReservePlugin, sub *appsapi.Subscription, namespacedClusters []string) {
+func (f *frameworkImpl) runReservePluginUnreserve(ctx context.Context, pl framework.ReservePlugin, sub *appsapi.Subscription, targetClusters framework.TargetClusters) {
 	startTime := time.Now()
-	pl.Unreserve(ctx, sub, namespacedClusters)
+	pl.Unreserve(ctx, sub, targetClusters)
 	f.metricsRecorder.observePluginDurationAsync(unreserve, pl.Name(), nil, metrics.SinceInSeconds(startTime))
 }
 
@@ -601,7 +601,7 @@ func (f *frameworkImpl) runReservePluginUnreserve(ctx context.Context, pl framew
 // plugins returns "Wait", then this function will create and add waiting subscription
 // to a map of currently waiting subs and return status with "Wait" code.
 // Subscription will remain waiting subscription for the minimum duration returned by the permit plugins.
-func (f *frameworkImpl) RunPermitPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) (status *framework.Status) {
+func (f *frameworkImpl) RunPermitPlugins(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) (status *framework.Status) {
 	startTime := time.Now()
 	defer func() {
 		metrics.FrameworkExtensionPointDuration.WithLabelValues(permit, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
@@ -609,7 +609,7 @@ func (f *frameworkImpl) RunPermitPlugins(ctx context.Context, sub *appsapi.Subsc
 	pluginsWaitTime := make(map[string]time.Duration)
 	statusCode := framework.Success
 	for _, pl := range f.permitPlugins {
-		status, timeout := f.runPermitPlugin(ctx, pl, sub, namespacedClusters)
+		status, timeout := f.runPermitPlugin(ctx, pl, sub, targetClusters)
 		if !status.IsSuccess() {
 			if status.IsUnschedulable() {
 				msg := fmt.Sprintf("rejected subscription %q by permit plugin %q: %v", sub.Name, pl.Name(), status.Message())
@@ -641,9 +641,9 @@ func (f *frameworkImpl) RunPermitPlugins(ctx context.Context, sub *appsapi.Subsc
 	return nil
 }
 
-func (f *frameworkImpl) runPermitPlugin(ctx context.Context, pl framework.PermitPlugin, sub *appsapi.Subscription, namespacedClusters []string) (*framework.Status, time.Duration) {
+func (f *frameworkImpl) runPermitPlugin(ctx context.Context, pl framework.PermitPlugin, sub *appsapi.Subscription, targetClusters framework.TargetClusters) (*framework.Status, time.Duration) {
 	startTime := time.Now()
-	status, timeout := pl.Permit(ctx, sub, namespacedClusters)
+	status, timeout := pl.Permit(ctx, sub, targetClusters)
 	f.metricsRecorder.observePluginDurationAsync(permit, pl.Name(), status, metrics.SinceInSeconds(startTime))
 	return status, timeout
 }
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
index a969031..387d90b 100644
--- a/pkg/scheduler/scheduler.go
+++ b/pkg/scheduler/scheduler.go
@@ -25,6 +25,7 @@ import (
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -77,6 +78,9 @@ type Scheduler struct {
 	subsLister applisters.SubscriptionLister
 	subsSynced cache.InformerSynced
 
+	inventoryLister applisters.FeedInventoryLister
+	inventorySynced cache.InformerSynced
+
 	// default in-tree registry
 	registry frameworkruntime.Registry
 
@@ -126,6 +130,8 @@ func NewScheduler(schedulerOptions *options.SchedulerOptions) (*Scheduler, error
 		ClusternetInformerFactory: clusternetInformerFactory,
 		subsLister:                clusternetInformerFactory.Apps().V1alpha1().Subscriptions().Lister(),
 		subsSynced:                clusternetInformerFactory.Apps().V1alpha1().Subscriptions().Informer().HasSynced,
+		inventoryLister:           clusternetInformerFactory.Apps().V1alpha1().FeedInventories().Lister(),
+		inventorySynced:           clusternetInformerFactory.Apps().V1alpha1().FeedInventories().Informer().HasSynced,
 		registry:                  plugins.NewInTreeRegistry(),
 		scheduleAlgorithm:         algorithm.NewGenericScheduler(schedulerCache),
 		SchedulingQueue:           workqueue.NewRateLimitingQueue(workqueue.DefaultItemBasedRateLimiter()),
@@ -231,13 +237,26 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 	}
 	klog.V(3).InfoS("Attempting to schedule subscription", "subscription", klog.KObj(sub))
 
+	var finv *appsapi.FeedInventory
+	if sub.Spec.SchedulingStrategy == appsapi.DividingSchedulingStrategyType {
+		finv, err = sched.inventoryLister.FeedInventories(ns).Get(name)
+		// TODO(Garrybest): Ignore some circumstances when subscription does not need to be scheduled so far.
+		// E.g., when feed in subscription has been modified but finv has not been updated.
+		if err != nil {
+			if !errors.IsNotFound(err) {
+				utilruntime.HandleError(err)
+			}
+			return
+		}
+	}
+
 	// Synchronously attempt to find a fit for the subscription.
 	start := time.Now()
 
 	schedulingCycleCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	scheduleResult, err := sched.scheduleAlgorithm.Schedule(schedulingCycleCtx, sched.framework, sub)
+	scheduleResult, err := sched.scheduleAlgorithm.Schedule(schedulingCycleCtx, sched.framework, sub, finv)
 	if err != nil {
 		sched.recordSchedulingFailure(sub, err, ReasonUnschedulable)
 		return
@@ -245,20 +264,17 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 	metrics.SchedulingAlgorithmLatency.Observe(metrics.SinceInSeconds(start))
 
 	// Run the Reserve method of reserve plugins.
-	var namespacedClusters []string
-	for _, cluster := range scheduleResult.SuggestedClusters {
-		namespacedClusters = append(namespacedClusters, cluster)
-	}
-	if sts := sched.framework.RunReservePluginsReserve(schedulingCycleCtx, sub, namespacedClusters); !sts.IsSuccess() {
+	targetClusters := scheduleResult.SuggestedClusters
+	if sts := sched.framework.RunReservePluginsReserve(schedulingCycleCtx, sub, targetClusters); !sts.IsSuccess() {
 		metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 		// trigger un-reserve to clean up state associated with the reserved subscription
-		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, sub, namespacedClusters)
+		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, sub, targetClusters)
 		sched.recordSchedulingFailure(sub, sts.AsError(), SchedulerError)
 		return
 	}
 
 	// Run "permit" plugins.
-	runPermitStatus := sched.framework.RunPermitPlugins(schedulingCycleCtx, sub, namespacedClusters)
+	runPermitStatus := sched.framework.RunPermitPlugins(schedulingCycleCtx, sub, targetClusters)
 	if runPermitStatus.Code() != framework.Wait && !runPermitStatus.IsSuccess() {
 		var reason string
 		if runPermitStatus.IsUnschedulable() {
@@ -269,7 +285,7 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 			reason = SchedulerError
 		}
 		// One of the plugins returned status different from success or wait.
-		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, sub, namespacedClusters)
+		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, sub, targetClusters)
 		sched.recordSchedulingFailure(sub, runPermitStatus.AsError(), reason)
 		return
 	}
@@ -292,39 +308,39 @@ func (sched *Scheduler) scheduleOne(ctx context.Context) {
 				reason = SchedulerError
 			}
 			// trigger un-reserve plugins to clean up state associated with the reserved subscription
-			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, namespacedClusters)
+			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, targetClusters)
 			sched.recordSchedulingFailure(sub, waitOnPermitStatus.AsError(), reason)
 			return
 		}
 
 		// Run "prebind" plugins.
-		preBindStatus := sched.framework.RunPreBindPlugins(bindingCycleCtx, sub, namespacedClusters)
+		preBindStatus := sched.framework.RunPreBindPlugins(bindingCycleCtx, sub, targetClusters)
 		if !preBindStatus.IsSuccess() {
 			metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 			// trigger un-reserve plugins to clean up state associated with the reserved subscription
-			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, namespacedClusters)
+			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, targetClusters)
 			sched.recordSchedulingFailure(sub, preBindStatus.AsError(), SchedulerError)
 			return
 		}
 
-		err := sched.bind(bindingCycleCtx, sub, namespacedClusters)
+		err := sched.bind(bindingCycleCtx, sub, targetClusters)
 		if err != nil {
 			metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 			// trigger un-reserve plugins to clean up state associated with the reserved subscription
-			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, namespacedClusters)
+			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, targetClusters)
 			sched.recordSchedulingFailure(sub, fmt.Errorf("binding rejected: %w", err), SchedulerError)
 		} else {
 			metrics.SubscriptionScheduled(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
 
 			// Run "postbind" plugins.
-			sched.framework.RunPostBindPlugins(bindingCycleCtx, sub, namespacedClusters)
+			sched.framework.RunPostBindPlugins(bindingCycleCtx, sub, targetClusters)
 		}
 	}()
 }
 
 // bind a subscription to given clusters.
 // We expect this to run asynchronously, so we handle binding metrics internally.
-func (sched *Scheduler) bind(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) (err error) {
+func (sched *Scheduler) bind(ctx context.Context, sub *appsapi.Subscription, targetClusters framework.TargetClusters) (err error) {
 	defer func() {
 		// finish binding
 		if err != nil {
@@ -336,11 +352,11 @@ func (sched *Scheduler) bind(ctx context.Context, sub *appsapi.Subscription, nam
 			corev1.EventTypeNormal,
 			"Scheduled",
 			"Successfully bound %s to %s",
-			klog.KObj(sub), strings.Join(namespacedClusters, ","),
+			klog.KObj(sub), strings.Join(targetClusters.BindingClusters, ","),
 		)
 	}()
 
-	bindStatus := sched.framework.RunBindPlugins(ctx, sub, namespacedClusters)
+	bindStatus := sched.framework.RunBindPlugins(ctx, sub, targetClusters)
 	if bindStatus.IsSuccess() {
 		return nil
 	}
@@ -418,6 +434,22 @@ func (sched *Scheduler) addAllEventHandlers() {
 		},
 	})
 
+	sched.ClusternetInformerFactory.Apps().V1alpha1().FeedInventories().Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc: func(obj interface{}) {
+			finv := obj.(*appsapi.FeedInventory)
+			sched.SchedulingQueue.AddRateLimited(klog.KObj(finv).String())
+		},
+		UpdateFunc: func(oldObj, newObj interface{}) {
+			oldInventory := oldObj.(*appsapi.FeedInventory)
+			newInventory := newObj.(*appsapi.FeedInventory)
+			if newInventory.ResourceVersion == oldInventory.ResourceVersion {
+				// Periodic resync will send update events for all known Inventory.
+				return
+			}
+			sched.SchedulingQueue.AddRateLimited(klog.KObj(newInventory).String())
+		},
+	})
+
 	enqueueSubscriptionForClusterFunc := func(mcls *clusterapi.ManagedCluster) {
 		sched.lock.RLock()
 		defer sched.lock.RUnlock()
diff --git a/pkg/utils/feed.go b/pkg/utils/feed.go
index fec278c..43e9fd6 100644
--- a/pkg/utils/feed.go
+++ b/pkg/utils/feed.go
@@ -174,3 +174,11 @@ func HashSubscriptionSpec(subscriptionSpec *appsapi.SubscriptionSpec) uint64 {
 	printer.Fprintf(hasher, "%#v", specJSON)
 	return uint64(hasher.Sum32())
 }
+
+func GetFeedKey(feed *appsapi.Feed) string {
+	if len(feed.Namespace) != 0 {
+		return fmt.Sprintf("%s/%s/%s/%s", feed.APIVersion, feed.Kind, feed.Namespace, feed.Name)
+	} else {
+		return fmt.Sprintf("%s/%s/%s", feed.APIVersion, feed.Kind, feed.Name)
+	}
+}


Test patch:
diff --git a/pkg/scheduler/algorithm/generic_test.go b/pkg/scheduler/algorithm/generic_test.go
new file mode 100644
index 0000000..07a8da0
--- /dev/null
+++ b/pkg/scheduler/algorithm/generic_test.go
@@ -0,0 +1,324 @@
+package algorithm
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/tools/cache"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	"github.com/clusternet/clusternet/pkg/generated/clientset/versioned/fake"
+	informers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions"
+	"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/utils"
+)
+
+func newClusterWithLabels(name, namespace string, labels map[string]string) *clusterapi.ManagedCluster {
+	return &clusterapi.ManagedCluster{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+			Labels:    labels,
+		},
+	}
+}
+
+func newDeploymentFeed(name string) *appsapi.Feed {
+	return &appsapi.Feed{
+		Kind:       "Deployment",
+		APIVersion: "apps/v1",
+		Namespace:  "default",
+		Name:       name,
+	}
+}
+
+func Test_genericScheduler_divideReplicas(t *testing.T) {
+	desiredReplicas := int32(12)
+	type args struct {
+		selected *framework.TargetClusters
+		sub      *appsapi.Subscription
+		finv     *appsapi.FeedInventory
+	}
+	tests := []struct {
+		name         string
+		args         args
+		wantReplicas map[string][]int32
+		wantErr      bool
+	}{
+		{
+			name: "1 cluster",
+			args: args{
+				selected: &framework.TargetClusters{
+					BindingClusters: []string{"cluster-ns-01/member1"},
+				},
+				sub: &appsapi.Subscription{
+					Spec: appsapi.SubscriptionSpec{
+						SchedulingStrategy: appsapi.DividingSchedulingStrategyType,
+						DividingScheduling: &appsapi.DividingSchedulingStrategy{Type: appsapi.StaticReplicaDividingType},
+						Subscribers: []appsapi.Subscriber{
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"a": "1",
+									},
+								},
+								Weight: 1,
+							},
+						},
+						Feeds: []appsapi.Feed{*newDeploymentFeed("example-1")},
+					},
+				},
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{Feeds: []appsapi.FeedOrder{
+						{
+							Feed:            *newDeploymentFeed("example-1"),
+							DesiredReplicas: &desiredReplicas,
+						},
+					}},
+				},
+			},
+			wantReplicas: map[string][]int32{
+				utils.GetFeedKey(newDeploymentFeed("example-1")): {desiredReplicas},
+			},
+			wantErr: false,
+		},
+		{
+			name: "2 clusters",
+			args: args{
+				selected: &framework.TargetClusters{
+					BindingClusters: []string{"cluster-ns-01/member1", "cluster-ns-01/member2"},
+				},
+				sub: &appsapi.Subscription{
+					Spec: appsapi.SubscriptionSpec{
+						SchedulingStrategy: appsapi.DividingSchedulingStrategyType,
+						DividingScheduling: &appsapi.DividingSchedulingStrategy{Type: appsapi.StaticReplicaDividingType},
+						Subscribers: []appsapi.Subscriber{
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"a": "1",
+									},
+								},
+								Weight: 1,
+							},
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"b": "2",
+									},
+								},
+								Weight: 2,
+							},
+						},
+						Feeds: []appsapi.Feed{*newDeploymentFeed("example-1")},
+					},
+				},
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{Feeds: []appsapi.FeedOrder{
+						{
+							Feed:            *newDeploymentFeed("example-1"),
+							DesiredReplicas: &desiredReplicas,
+						},
+					}},
+				},
+			},
+			wantReplicas: map[string][]int32{
+				utils.GetFeedKey(newDeploymentFeed("example-1")): {4, 8},
+			},
+			wantErr: false,
+		},
+		{
+			name: "3 clusters",
+			args: args{
+				selected: &framework.TargetClusters{
+					BindingClusters: []string{"cluster-ns-01/member1", "cluster-ns-01/member2", "cluster-ns-02/member3"},
+				},
+				sub: &appsapi.Subscription{
+					Spec: appsapi.SubscriptionSpec{
+						SchedulingStrategy: appsapi.DividingSchedulingStrategyType,
+						DividingScheduling: &appsapi.DividingSchedulingStrategy{Type: appsapi.StaticReplicaDividingType},
+						Subscribers: []appsapi.Subscriber{
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"a": "1",
+									},
+								},
+								Weight: 1,
+							},
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"b": "2",
+									},
+								},
+								Weight: 2,
+							},
+						},
+						Feeds: []appsapi.Feed{*newDeploymentFeed("example-1")},
+					},
+				},
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{Feeds: []appsapi.FeedOrder{
+						{
+							Feed:            *newDeploymentFeed("example-1"),
+							DesiredReplicas: &desiredReplicas,
+						},
+					}},
+				},
+			},
+			wantReplicas: map[string][]int32{
+				utils.GetFeedKey(newDeploymentFeed("example-1")): {3, 6, 3},
+			},
+			wantErr: false,
+		},
+		{
+			name: "3 clusters with duplicated weight",
+			args: args{
+				selected: &framework.TargetClusters{
+					BindingClusters: []string{"cluster-ns-01/member1", "cluster-ns-01/member2", "cluster-ns-02/member3"},
+				},
+				sub: &appsapi.Subscription{
+					Spec: appsapi.SubscriptionSpec{
+						SchedulingStrategy: appsapi.DividingSchedulingStrategyType,
+						DividingScheduling: &appsapi.DividingSchedulingStrategy{Type: appsapi.StaticReplicaDividingType},
+						Subscribers: []appsapi.Subscriber{
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"a": "1",
+									},
+								},
+								Weight: 1,
+							},
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"b": "2",
+									},
+								},
+								Weight: 0,
+							},
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"b": "1",
+									},
+								},
+								Weight: 2,
+							},
+						},
+						Feeds: []appsapi.Feed{*newDeploymentFeed("example-1")},
+					},
+				},
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{Feeds: []appsapi.FeedOrder{
+						{
+							Feed:            *newDeploymentFeed("example-1"),
+							DesiredReplicas: &desiredReplicas,
+						},
+					}},
+				},
+			},
+			wantReplicas: map[string][]int32{
+				utils.GetFeedKey(newDeploymentFeed("example-1")): {3, 0, 9},
+			},
+			wantErr: false,
+		},
+		{
+			name: "2 feeds, 3 clusters with duplicated weight",
+			args: args{
+				selected: &framework.TargetClusters{
+					BindingClusters: []string{"cluster-ns-01/member1", "cluster-ns-01/member2", "cluster-ns-02/member3"},
+				},
+				sub: &appsapi.Subscription{
+					Spec: appsapi.SubscriptionSpec{
+						SchedulingStrategy: appsapi.DividingSchedulingStrategyType,
+						DividingScheduling: &appsapi.DividingSchedulingStrategy{Type: appsapi.StaticReplicaDividingType},
+						Subscribers: []appsapi.Subscriber{
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"a": "1",
+									},
+								},
+								Weight: 1,
+							},
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"b": "2",
+									},
+								},
+								Weight: 0,
+							},
+							{
+								ClusterAffinity: &metav1.LabelSelector{
+									MatchLabels: map[string]string{
+										"b": "1",
+									},
+								},
+								Weight: 2,
+							},
+						},
+						Feeds: []appsapi.Feed{*newDeploymentFeed("example-1"), *newDeploymentFeed("example-2")},
+					},
+				},
+				finv: &appsapi.FeedInventory{
+					Spec: appsapi.FeedInventorySpec{Feeds: []appsapi.FeedOrder{
+						{
+							Feed:            *newDeploymentFeed("example-1"),
+							DesiredReplicas: &desiredReplicas,
+						},
+						{
+							Feed:            *newDeploymentFeed("example-2"),
+							DesiredReplicas: &desiredReplicas,
+						},
+					}},
+				},
+			},
+			wantReplicas: map[string][]int32{
+				utils.GetFeedKey(newDeploymentFeed("example-1")): {3, 0, 9},
+				utils.GetFeedKey(newDeploymentFeed("example-2")): {3, 0, 9},
+			},
+			wantErr: false,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+
+			clusternetClient := fake.NewSimpleClientset(
+				newClusterWithLabels("member1", "cluster-ns-01", map[string]string{"a": "1"}),
+				newClusterWithLabels("member2", "cluster-ns-01", map[string]string{"b": "2"}),
+				newClusterWithLabels("member3", "cluster-ns-02", map[string]string{"a": "1", "b": "1"}),
+			)
+			clusternetInformerFactory := informers.NewSharedInformerFactory(clusternetClient, known.DefaultResync)
+			clusterInformer := clusternetInformerFactory.Clusters().V1beta1().ManagedClusters()
+
+			g := &genericScheduler{
+				cache: schedulercache.New(clusterInformer.Lister()),
+			}
+
+			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)
+			}
+
+			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)
+			}
+			if !reflect.DeepEqual(tt.args.selected.Replicas, tt.wantReplicas) {
+				t.Errorf("divideReplicas() gotResponse = %v, want %v", tt.args.selected.Replicas, tt.wantReplicas)
+			}
+
+		})
+	}
+}
diff --git a/pkg/scheduler/framework/plugins/defaultbinder/default_binder_test.go b/pkg/scheduler/framework/plugins/defaultbinder/default_binder_test.go
index 453c474..eff3951 100644
--- a/pkg/scheduler/framework/plugins/defaultbinder/default_binder_test.go
+++ b/pkg/scheduler/framework/plugins/defaultbinder/default_binder_test.go
@@ -28,6 +28,7 @@ import (
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
 	"github.com/clusternet/clusternet/pkg/generated/clientset/versioned/fake"
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
 	frameworkruntime "github.com/clusternet/clusternet/pkg/scheduler/framework/runtime"
 )
 
@@ -35,9 +36,11 @@ func TestDefaultBinder(t *testing.T) {
 	testSubscription := &appsapi.Subscription{
 		ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "ns"},
 	}
-	testClusters := []string{
-		"cluster-ns-03/xyz",
-		"cluster-ns-01/def",
+	testClusters := framework.TargetClusters{
+		BindingClusters: []string{
+			"cluster-ns-03/xyz",
+			"cluster-ns-01/def",
+		},
 	}
 	tests := []struct {
 		name           string