Repository: clusternet/clusternet
Pull request #204 (2021-12-13 09:46:51)

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



Patch:
diff --git a/cmd/clusternet-scheduler/app/app.go b/cmd/clusternet-scheduler/app/app.go
new file mode 100644
index 0000000..ea40b53
--- /dev/null
+++ b/cmd/clusternet-scheduler/app/app.go
@@ -0,0 +1,82 @@
+/*
+Copyright 2021 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 app
+
+import (
+	"context"
+
+	"github.com/spf13/cobra"
+	"github.com/spf13/pflag"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/klog/v2"
+
+	_ "github.com/clusternet/clusternet/pkg/features"
+	"github.com/clusternet/clusternet/pkg/scheduler"
+	"github.com/clusternet/clusternet/pkg/scheduler/options"
+	"github.com/clusternet/clusternet/pkg/version"
+)
+
+var (
+	// the command name
+	cmdName = "clusternet-scheduler"
+)
+
+// NewSchedulerCommand creates a *cobra.Command object with default parameters
+func NewSchedulerCommand(ctx context.Context) *cobra.Command {
+	opts, err := options.NewSchedulerOptions()
+	if err != nil {
+		klog.Fatalf("unable to initialize command options: %v", err)
+	}
+
+	cmd := &cobra.Command{
+		Use:  cmdName,
+		Long: `cluster-wise scheduler`,
+		Run: func(cmd *cobra.Command, args []string) {
+			if err = version.PrintAndExitIfRequested(cmdName); err != nil {
+				klog.Exit(err)
+			}
+
+			if err = opts.Complete(); err != nil {
+				klog.Exit(err)
+			}
+			if err = opts.Validate(args); err != nil {
+				klog.Exit(err)
+			}
+
+			cmd.Flags().VisitAll(func(flag *pflag.Flag) {
+				klog.V(1).Infof("FLAG: --%s=%q", flag.Name, flag.Value)
+			})
+
+			// TODO: start metrics server
+
+			sched, err := scheduler.NewScheduler(opts)
+			if err != nil {
+				klog.Exit(err)
+			}
+			if err = sched.Run(ctx); err != nil {
+				klog.Exit(err)
+			}
+		},
+	}
+
+	flags := cmd.Flags()
+	version.AddVersionFlag(flags)
+	opts.AddFlags(flags)
+	utilfeature.DefaultMutableFeatureGate.AddFlag(flags)
+
+	return cmd
+}
diff --git a/cmd/clusternet-scheduler/scheduler.go b/cmd/clusternet-scheduler/scheduler.go
new file mode 100644
index 0000000..3ad96ee
--- /dev/null
+++ b/cmd/clusternet-scheduler/scheduler.go
@@ -0,0 +1,48 @@
+/*
+Copyright 2021 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 main
+
+import (
+	goflag "flag"
+	"fmt"
+	"math/rand"
+	"os"
+	"time"
+
+	"github.com/spf13/pflag"
+	"k8s.io/component-base/logs"
+
+	"github.com/clusternet/clusternet/cmd/clusternet-scheduler/app"
+	"github.com/clusternet/clusternet/pkg/utils"
+)
+
+func main() {
+	rand.Seed(time.Now().UTC().UnixNano())
+
+	logs.InitLogs()
+	defer logs.FlushLogs()
+
+	ctx := utils.GracefulStopWithContext()
+	command := app.NewSchedulerCommand(ctx)
+	pflag.CommandLine.SetNormalizeFunc(utils.WordSepNormalizeFunc)
+	pflag.CommandLine.AddGoFlagSet(goflag.CommandLine)
+
+	if err := command.Execute(); err != nil {
+		fmt.Fprintf(os.Stderr, "%v\n", err)
+		os.Exit(1)
+	}
+}
diff --git a/go.mod b/go.mod
index ab85893..4847e9d 100644
--- a/go.mod
+++ b/go.mod
@@ -7,6 +7,7 @@ require (
 	github.com/emicklei/go-restful v2.9.5+incompatible
 	github.com/evanphx/json-patch v4.9.0+incompatible
 	github.com/go-openapi/spec v0.19.5
+	github.com/google/go-cmp v0.5.5
 	github.com/gorilla/websocket v1.4.2
 	github.com/opencontainers/runc v1.0.3 // indirect
 	github.com/rancher/remotedialer v0.2.6-0.20210318171128-d1ebd5202be4
diff --git a/manifests/crds/apps.clusternet.io_subscriptions.yaml b/manifests/crds/apps.clusternet.io_subscriptions.yaml
index d39052a..5429ea5 100644
--- a/manifests/crds/apps.clusternet.io_subscriptions.yaml
+++ b/manifests/crds/apps.clusternet.io_subscriptions.yaml
@@ -68,6 +68,13 @@ spec:
                 default: default
                 description: If specified, the Subscription will be handled by specified scheduler. If not specified, the Subscription will be handled by default scheduler.
                 type: string
+              schedulingStrategy:
+                default: Replication
+                description: If specified, the Subscription will be handled with specified SchedulingStrategy. Otherwise, with generic SchedulingStrategy.
+                enum:
+                - Replication
+                - Balancing
+                type: string
               subscribers:
                 description: Subscribers subscribes
                 items:
@@ -114,6 +121,11 @@ spec:
           status:
             description: SubscriptionStatus defines the observed state of Subscription
             properties:
+              bindingNamespaces:
+                description: Namespaces of targeted clusters that Subscription binds to.
+                items:
+                  type: string
+                type: array
               completedReleases:
                 description: Total number of completed releases targeted by this deployment.
                 format: int32
diff --git a/pkg/agent/deployer/generic/generic.go b/pkg/agent/deployer/generic/generic.go
index c4660c7..e871a9e 100644
--- a/pkg/agent/deployer/generic/generic.go
+++ b/pkg/agent/deployer/generic/generic.go
@@ -86,7 +86,6 @@ func NewDeployer(syncMode clusterapi.ClusterSyncMode, appPusherEnabled bool,
 	descController, err := description.NewController(clusternetClient,
 		clusternetInformerFactory.Apps().V1alpha1().Descriptions(),
 		clusternetInformerFactory.Apps().V1alpha1().HelmReleases(),
-		clusternetInformerFactory.Apps().V1alpha1().HelmCharts(),
 		deployer.recorder,
 		deployer.handleDescription)
 	if err != nil {
diff --git a/pkg/apis/apps/v1alpha1/subscription.go b/pkg/apis/apps/v1alpha1/subscription.go
index 6848011..bfce74c 100644
--- a/pkg/apis/apps/v1alpha1/subscription.go
+++ b/pkg/apis/apps/v1alpha1/subscription.go
@@ -46,6 +46,15 @@ type SubscriptionSpec struct {
 	// +kubebuilder:default=default
 	SchedulerName string `json:"schedulerName,omitempty"`
 
+	// If specified, the Subscription will be handled with specified SchedulingStrategy.
+	// Otherwise, with generic SchedulingStrategy.
+	//
+	// +optional
+	// +kubebuilder:validation:Type=string
+	// +kubebuilder:validation:Enum=Replication;Balancing
+	// +kubebuilder:default=Replication
+	SchedulingStrategy SchedulingStrategyType `json:"schedulingStrategy,omitempty"`
+
 	// Subscribers subscribes
 	//
 	// +required
@@ -61,6 +70,11 @@ type SubscriptionSpec struct {
 
 // SubscriptionStatus defines the observed state of Subscription
 type SubscriptionStatus struct {
+	// Namespaces of targeted clusters that Subscription binds to.
+	//
+	// +optional
+	BindingNamespaces []string `json:"bindingNamespaces,omitempty"`
+
 	// Total number of Helm releases desired by this Subscription.
 	//
 	// +optional
@@ -111,6 +125,17 @@ type Feed struct {
 	Name string `json:"name"`
 }
 
+type SchedulingStrategyType string
+
+const (
+	// ReplicaSchedulingStrategy places and maintains a copy of this Subscription on each matched clusters.
+	ReplicaSchedulingStrategy SchedulingStrategyType = "Replication"
+
+	// BalancingSchedulingStrategy balances and divides the replicas of a Subscription to several matching clusters.
+	// TODO
+	BalancingSchedulingStrategy SchedulingStrategyType = "Balancing"
+)
+
 // +kubebuilder:object:root=true
 // +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
 
diff --git a/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go b/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go
index 4b17b19..06daf13 100644
--- a/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go
+++ b/pkg/apis/apps/v1alpha1/zz_generated.deepcopy.go
@@ -739,7 +739,7 @@ func (in *Subscription) DeepCopyInto(out *Subscription) {
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	in.Spec.DeepCopyInto(&out.Spec)
-	out.Status = in.Status
+	in.Status.DeepCopyInto(&out.Status)
 	return
 }
 
@@ -825,6 +825,11 @@ func (in *SubscriptionSpec) DeepCopy() *SubscriptionSpec {
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *SubscriptionStatus) DeepCopyInto(out *SubscriptionStatus) {
 	*out = *in
+	if in.BindingNamespaces != nil {
+		in, out := &in.BindingNamespaces, &out.BindingNamespaces
+		*out = make([]string, len(*in))
+		copy(*out, *in)
+	}
 	return
 }
 
diff --git a/pkg/controllers/apps/description/description.go b/pkg/controllers/apps/description/description.go
index 3a8e9e3..4a6dd75 100644
--- a/pkg/controllers/apps/description/description.go
+++ b/pkg/controllers/apps/description/description.go
@@ -25,8 +25,6 @@ import (
 	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"
-	"k8s.io/apimachinery/pkg/selection"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/tools/cache"
@@ -69,9 +67,7 @@ type Controller struct {
 }
 
 func NewController(clusternetClient clusternetClientSet.Interface,
-	descInformer appInformers.DescriptionInformer,
-	hrInformer appInformers.HelmReleaseInformer,
-	helmChartInformer appInformers.HelmChartInformer,
+	descInformer appInformers.DescriptionInformer, hrInformer appInformers.HelmReleaseInformer,
 	recorder record.EventRecorder, syncHandlerFunc SyncHandlerFunc) (*Controller, error) {
 	if syncHandlerFunc == nil {
 		return nil, fmt.Errorf("syncHandlerFunc must be set")
@@ -97,10 +93,6 @@ func NewController(clusternetClient clusternetClientSet.Interface,
 	hrInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
 		DeleteFunc: c.deleteHelmRelease,
 	})
-
-	helmChartInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
-		UpdateFunc: c.updateHelmChart,
-	})
 	return c, nil
 }
 
@@ -200,43 +192,6 @@ func (c *Controller) deleteHelmRelease(obj interface{}) {
 	c.enqueue(desc)
 }
 
-func (c *Controller) updateHelmChart(old, cur interface{}) {
-	oldChart := old.(*appsapi.HelmChart)
-	newChart := cur.(*appsapi.HelmChart)
-
-	// Decide whether discovery has reported a spec change.
-	if reflect.DeepEqual(oldChart.Spec, newChart.Spec) {
-		return
-	}
-
-	baseUIDs := []string{}
-	for k, v := range newChart.GetLabels() {
-		if v == "Base" {
-			baseUIDs = append(baseUIDs, k)
-		}
-	}
-	if len(baseUIDs) == 0 {
-		return
-	}
-
-	klog.V(5).Infof("HelmChart %s/%s has bases: %s", newChart.Namespace, newChart.Name, baseUIDs)
-	requirement, err := labels.NewRequirement(known.ConfigUIDLabel, selection.In, baseUIDs)
-	if err != nil {
-		utilruntime.HandleError(fmt.Errorf("failed to create label requirement: %v", err))
-		return
-	}
-	selector := labels.NewSelector().Add(*requirement)
-	descs, err := c.descLister.Descriptions("").List(selector)
-	if err != nil {
-		utilruntime.HandleError(fmt.Errorf("failed to list descriptions: %v", err))
-		return
-	}
-	for _, desc := range descs {
-		klog.V(4).Infof("updating Description %q", klog.KObj(desc))
-		c.workqueue.Add(klog.KObj(desc).String())
-	}
-}
-
 // resolveControllerRef returns the controller referenced by a ControllerRef,
 // or nil if the ControllerRef could not be resolved to a matching controller
 // of the correct Kind.
diff --git a/pkg/hub/deployer/generic/generic.go b/pkg/hub/deployer/generic/generic.go
index 3d3dfbb..0b0e40c 100644
--- a/pkg/hub/deployer/generic/generic.go
+++ b/pkg/hub/deployer/generic/generic.go
@@ -78,7 +78,6 @@ func NewDeployer(apiserverURL string, clusternetClient *clusternetclientset.Clie
 	descController, err := description.NewController(clusternetClient,
 		clusternetInformerFactory.Apps().V1alpha1().Descriptions(),
 		clusternetInformerFactory.Apps().V1alpha1().HelmReleases(),
-		clusternetInformerFactory.Apps().V1alpha1().HelmCharts(),
 		deployer.recorder,
 		deployer.handleDescription)
 	if err != nil {
diff --git a/pkg/hub/deployer/helm/helm.go b/pkg/hub/deployer/helm/helm.go
index 450dfed..1eb21dd 100644
--- a/pkg/hub/deployer/helm/helm.go
+++ b/pkg/hub/deployer/helm/helm.go
@@ -122,7 +122,6 @@ func NewDeployer(apiserverURL string, clusternetClient *clusternetclientset.Clie
 	descController, err := description.NewController(clusternetClient,
 		clusternetInformerFactory.Apps().V1alpha1().Descriptions(),
 		clusternetInformerFactory.Apps().V1alpha1().HelmReleases(),
-		clusternetInformerFactory.Apps().V1alpha1().HelmCharts(),
 		deployer.recorder,
 		deployer.handleDescription)
 	if err != nil {
diff --git a/pkg/known/constant.go b/pkg/known/constant.go
index ad115e1..b9e842b 100644
--- a/pkg/known/constant.go
+++ b/pkg/known/constant.go
@@ -56,3 +56,9 @@ const (
 const (
 	Category = "clusternet.shadow"
 )
+
+const (
+	// NoteLengthLimit denotes the maximum note length.
+	// copied from k8s.io/kubernetes/pkg/apis/core/validation/events.go
+	NoteLengthLimit = 1024
+)
diff --git a/pkg/scheduler/algorithm/generic.go b/pkg/scheduler/algorithm/generic.go
new file mode 100644
index 0000000..4d06dd9
--- /dev/null
+++ b/pkg/scheduler/algorithm/generic.go
@@ -0,0 +1,311 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/generic_scheduler.go and modified
+
+package algorithm
+
+import (
+	"context"
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
+	utiltrace "k8s.io/utils/trace"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	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"
+)
+
+const (
+	// minFeasibleClustersToFind is the minimum number of clusters that would be scored
+	// in each scheduling cycle. This is a semi-arbitrary value to ensure that a
+	// certain minimum of clusters are checked for feasibility. This in turn helps
+	// ensure a minimum level of spreading.
+	minFeasibleClustersToFind = 100
+	// minFeasibleClustersPercentageToFind is the minimum percentage of clusters that
+	// would be scored in each scheduling cycle. This is a semi-arbitrary value
+	// to ensure that a certain minimum of clusters are checked for feasibility.
+	// This in turn helps ensure a minimum level of spreading.
+	minFeasibleClustersPercentageToFind = 5
+)
+
+// ErrNoClustersAvailable is used to describe the error that no clusters available to schedule subscriptions.
+var ErrNoClustersAvailable = fmt.Errorf("no clusters available to schedule subscriptions")
+
+type genericScheduler struct {
+	cache                       schedulercache.Cache
+	percentageOfClustersToScore int32
+	nextStartClusterIndex       int
+}
+
+// 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) {
+	trace := utiltrace.New("Scheduling", utiltrace.Field{Key: "namespace", Value: sub.Namespace}, utiltrace.Field{Key: "name", Value: sub.Name})
+	defer trace.LogIfLong(100 * time.Millisecond)
+
+	if g.cache.NumClusters() == 0 {
+		return result, ErrNoClustersAvailable
+	}
+
+	feasibleClusters, diagnosis, err := g.findClustersThatFitSubscription(ctx, fwk, sub)
+	if err != nil {
+		return result, err
+	}
+	trace.Step("Computing predicates done")
+
+	if len(feasibleClusters) == 0 {
+		return result, &framework.FitError{
+			Subscription:   sub,
+			NumAllClusters: g.cache.NumClusters(),
+			Diagnosis:      diagnosis,
+		}
+	}
+
+	priorityList, err := prioritizeClusters(ctx, fwk, sub, feasibleClusters)
+	if err != nil {
+		return result, err
+	}
+
+	clusters, err := g.selectClusters(priorityList, sub)
+	trace.Step("Prioritizing done")
+
+	return ScheduleResult{
+		SuggestedClusters: clusters,
+		EvaluatedClusters: len(feasibleClusters) + len(diagnosis.ClusterToStatusMap),
+		FeasibleClusters:  len(feasibleClusters),
+	}, err
+}
+
+// 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) {
+	if len(clusterScoreList) == 0 {
+		return nil, fmt.Errorf("empty clusterScoreList")
+	}
+
+	var selected []string
+	for _, clusterScore := range clusterScoreList {
+		// TODO: sampling with scores
+		selected = append(selected, clusterScore.NamespacedName)
+	}
+	return selected, nil
+}
+
+// numFeasibleClustersToFind returns the number of feasible clusters that once found, the scheduler stops
+// its search for more feasible clusters.
+func (g *genericScheduler) numFeasibleClustersToFind(numAllClusters int32, schedulingStrategy appsapi.SchedulingStrategyType) (numClusters int32) {
+	if numAllClusters < minFeasibleClustersToFind || g.percentageOfClustersToScore >= 100 || schedulingStrategy == appsapi.ReplicaSchedulingStrategy {
+		return numAllClusters
+	}
+
+	adaptivePercentage := g.percentageOfClustersToScore
+	if adaptivePercentage <= 0 {
+		basePercentageOfClustersToScore := int32(50)
+		adaptivePercentage = basePercentageOfClustersToScore - numAllClusters/125
+		if adaptivePercentage < minFeasibleClustersPercentageToFind {
+			adaptivePercentage = minFeasibleClustersPercentageToFind
+		}
+	}
+
+	numClusters = numAllClusters * adaptivePercentage / 100
+	if numClusters < minFeasibleClustersToFind {
+		return minFeasibleClustersToFind
+	}
+
+	return numClusters
+}
+
+// Filters the clusters to find the ones that fit the subscription based on the framework filter plugins.
+func (g *genericScheduler) findClustersThatFitSubscription(ctx context.Context, fwk framework.Framework, sub *appsapi.Subscription) ([]*clusterapi.ManagedCluster, framework.Diagnosis, error) {
+	diagnosis := framework.Diagnosis{
+		ClusterToStatusMap:   make(framework.ClusterToStatusMap),
+		UnschedulablePlugins: sets.NewString(),
+	}
+
+	// Run "prefilter" plugins.
+	s := fwk.RunPreFilterPlugins(ctx, sub)
+	allClusters, err := g.cache.List()
+	if err != nil {
+		return nil, diagnosis, err
+	}
+	if !s.IsSuccess() {
+		if !s.IsUnschedulable() {
+			return nil, diagnosis, s.AsError()
+		}
+		// All clusters will have the same status. Some non trivial refactoring is
+		// needed to avoid this copy.
+		for _, n := range allClusters {
+			diagnosis.ClusterToStatusMap[klog.KObj(n).String()] = s
+		}
+		// Status satisfying IsUnschedulable() gets injected into diagnosis.UnschedulablePlugins.
+		diagnosis.UnschedulablePlugins.Insert(s.FailedPlugin())
+		return nil, diagnosis, nil
+	}
+
+	feasibleClusters, err := g.findClustersThatPassFilters(ctx, fwk, sub, diagnosis, allClusters)
+	if err != nil {
+		return nil, diagnosis, err
+	}
+	return feasibleClusters, diagnosis, nil
+}
+
+// findClustersThatPassFilters finds the clusters that fit the filter plugins.
+func (g *genericScheduler) findClustersThatPassFilters(ctx context.Context, fwk framework.Framework,
+	sub *appsapi.Subscription, diagnosis framework.Diagnosis,
+	clusters []*clusterapi.ManagedCluster) ([]*clusterapi.ManagedCluster, error) {
+	numClustersToFind := g.numFeasibleClustersToFind(int32(len(clusters)), sub.Spec.SchedulingStrategy)
+
+	// Create feasible list with enough space to avoid growing it
+	// and allow assigning.
+	feasibleClusters := make([]*clusterapi.ManagedCluster, numClustersToFind)
+
+	if !fwk.HasFilterPlugins() {
+		length := len(clusters)
+		for i := range feasibleClusters {
+			feasibleClusters[i] = clusters[(g.nextStartClusterIndex+i)%length]
+		}
+		g.nextStartClusterIndex = (g.nextStartClusterIndex + len(feasibleClusters)) % length
+		return feasibleClusters, nil
+	}
+
+	errCh := parallelize.NewErrorChannel()
+	var statusesLock sync.Mutex
+	var feasibleClustersLen int32
+	ctx, cancel := context.WithCancel(ctx)
+	checkCluster := func(i int) {
+		// We check the clusters starting from where we left off in the previous scheduling cycle,
+		// this is to make sure all clusters have the same chance of being examined across subscriptions.
+		cluster := clusters[(g.nextStartClusterIndex+i)%len(clusters)]
+
+		status := fwk.RunFilterPlugins(ctx, sub, cluster).Merge()
+		if status.Code() == framework.Error {
+			errCh.SendErrorWithCancel(status.AsError(), cancel)
+			return
+		}
+		if status.IsSuccess() {
+			length := atomic.AddInt32(&feasibleClustersLen, 1)
+			if length > numClustersToFind {
+				cancel()
+				atomic.AddInt32(&feasibleClustersLen, -1)
+			} else {
+				feasibleClusters[length-1] = cluster
+			}
+		} else {
+			statusesLock.Lock()
+			diagnosis.ClusterToStatusMap[klog.KObj(cluster).String()] = status
+			diagnosis.UnschedulablePlugins.Insert(status.FailedPlugin())
+			statusesLock.Unlock()
+		}
+	}
+
+	beginCheckCluster := time.Now()
+	statusCode := framework.Success
+	defer func() {
+		// We record Filter extension point latency here instead of in framework.go because framework.RunFilterPlugins
+		// function is called for each cluster, whereas we want to have an overall latency for all clusters per scheduling cycle.
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(runtime.Filter, statusCode.String(), fwk.ProfileName()).Observe(metrics.SinceInSeconds(beginCheckCluster))
+	}()
+
+	// Stops searching for more clusters once the configured number of feasible clusters
+	// are found.
+	fwk.Parallelizer().Until(ctx, len(clusters), checkCluster)
+	processedClusters := int(feasibleClustersLen) + len(diagnosis.ClusterToStatusMap)
+	g.nextStartClusterIndex = (g.nextStartClusterIndex + processedClusters) % len(clusters)
+
+	feasibleClusters = feasibleClusters[:feasibleClustersLen]
+	if err := errCh.ReceiveError(); err != nil {
+		statusCode = framework.Error
+		return nil, err
+	}
+	return feasibleClusters, nil
+}
+
+// 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) {
+	// 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,
+			})
+		}
+		return result, nil
+	}
+
+	// Run PreScore plugins.
+	preScoreStatus := fwk.RunPreScorePlugins(ctx, sub, clusters)
+	if !preScoreStatus.IsSuccess() {
+		return nil, preScoreStatus.AsError()
+	}
+
+	// Run the Score plugins.
+	scoresMap, scoreStatus := fwk.RunScorePlugins(ctx, sub, clusters)
+	if !scoreStatus.IsSuccess() {
+		return nil, scoreStatus.AsError()
+	}
+
+	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)
+			}
+		}
+	}
+
+	// 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
+		}
+	}
+
+	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)
+		}
+	}
+	return result, nil
+}
+
+// NewGenericScheduler creates a genericScheduler object.
+func NewGenericScheduler(cache schedulercache.Cache) ScheduleAlgorithm {
+	return &genericScheduler{
+		cache:                       cache,
+		percentageOfClustersToScore: schedulerapis.DefaultPercentageOfClustersToScore,
+	}
+}
diff --git a/pkg/scheduler/algorithm/interface.go b/pkg/scheduler/algorithm/interface.go
new file mode 100644
index 0000000..077aa2e
--- /dev/null
+++ b/pkg/scheduler/algorithm/interface.go
@@ -0,0 +1,41 @@
+/*
+Copyright 2021 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 algorithm
+
+import (
+	"context"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+)
+
+// 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)
+}
+
+// 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
+	// Number of clusters scheduler evaluated on one subscription scheduled
+	EvaluatedClusters int
+	// Number of feasible clusters on one subscription scheduled
+	FeasibleClusters int
+}
diff --git a/pkg/scheduler/apis/types.go b/pkg/scheduler/apis/types.go
new file mode 100644
index 0000000..9ce6d54
--- /dev/null
+++ b/pkg/scheduler/apis/types.go
@@ -0,0 +1,128 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/apis/config/types.go and modified
+
+package apis
+
+import (
+	"math"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+)
+
+// Plugins include multiple extension points. When specified, the list of plugins for
+// a particular extension point are the only ones enabled. If an extension point is
+// omitted from the config, then the default set of plugins is used for that extension point.
+// Enabled plugins are called in the order specified here, after default plugins. If they need to
+// be invoked before default plugins, default plugins must be disabled and re-enabled here in desired order.
+type Plugins struct {
+	// PreFilter is a list of plugins that should be invoked at "PreFilter" extension point of the scheduling framework.
+	PreFilter PluginSet
+
+	// Filter is a list of plugins that should be invoked when filtering out clusters that cannot run the Subscription.
+	Filter PluginSet
+
+	// PostFilter is a list of plugins that are invoked after filtering phase, no matter whether filtering succeeds or not.
+	PostFilter 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
+
+	// Reserve is a list of plugins invoked when reserving/unreserving resources
+	// after a cluster is assigned to run the Subscription.
+	Reserve PluginSet
+
+	// Permit is a list of plugins that control binding of a Subscription. These plugins can prevent or delay binding of a Subscription.
+	Permit PluginSet
+
+	// PreBind is a list of plugins that should be invoked before a Subscription is bound.
+	PreBind PluginSet
+
+	// Bind is a list of plugins that should be invoked at "Bind" extension point of the scheduling framework.
+	// The scheduler call these plugins in order. Scheduler skips the rest of these plugins as soon as one returns success.
+	Bind PluginSet
+
+	// PostBind is a list of plugins that should be invoked after a Subscription is successfully bound.
+	PostBind PluginSet
+}
+
+// PluginSet specifies enabled and disabled plugins for an extension point.
+// If an array is empty, missing, or nil, default plugins at that extension point will be used.
+type PluginSet struct {
+	// Enabled specifies plugins that should be enabled in addition to default plugins.
+	// These are called after default plugins and in the same order specified here.
+	Enabled []Plugin
+	// Disabled specifies default plugins that should be disabled.
+	// When all default plugins need to be disabled, an array containing only one "*" should be provided.
+	Disabled []Plugin
+}
+
+// Plugin specifies a plugin name and its weight when applicable. Weight is used only for Score plugins.
+type Plugin struct {
+	// Name defines the name of plugin
+	Name string
+	// Weight defines the weight of plugin, only used for Score plugins.
+	Weight int32
+}
+
+/*
+ * NOTE: The following variables and methods are intentionally left out of the staging mirror.
+ */
+const (
+	// DefaultPercentageOfClustersToScore defines the percentage of clusters of all clusters
+	// that once found feasible, the scheduler stops looking for more clusters.
+	// A value of 0 means adaptive, meaning the scheduler figures out a proper default.
+	DefaultPercentageOfClustersToScore = 0
+
+	// MaxCustomPriorityScore is the max score UtilizationShapePoint expects.
+	MaxCustomPriorityScore int64 = 10
+
+	// MaxTotalScore is the maximum total score.
+	MaxTotalScore int64 = math.MaxInt64
+
+	// MaxWeight defines the max weight value allowed for custom PriorityPolicy
+	MaxWeight = MaxTotalScore / MaxCustomPriorityScore
+)
+
+// Names returns the list of enabled plugin names.
+func (p *Plugins) Names() []string {
+	if p == nil {
+		return nil
+	}
+	extensions := []PluginSet{
+		p.PreFilter,
+		p.Filter,
+		p.PostFilter,
+		p.Reserve,
+		p.PreScore,
+		p.Score,
+		p.PreBind,
+		p.Bind,
+		p.PostBind,
+		p.Permit,
+	}
+	n := sets.NewString()
+	for _, e := range extensions {
+		for _, pg := range e.Enabled {
+			n.Insert(pg.Name)
+		}
+	}
+	return n.List()
+}
diff --git a/pkg/scheduler/cache/cache.go b/pkg/scheduler/cache/cache.go
new file mode 100644
index 0000000..c3ec491
--- /dev/null
+++ b/pkg/scheduler/cache/cache.go
@@ -0,0 +1,77 @@
+/*
+Copyright 2021 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 cache
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/labels"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/klog/v2"
+
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	clusterlisters "github.com/clusternet/clusternet/pkg/generated/listers/clusters/v1beta1"
+)
+
+type Cache interface {
+	// NumClusters returns the number of clusters in the cache.
+	NumClusters() int
+
+	// List returns the list of ManagedCluster(s).
+	List() ([]*clusterapi.ManagedCluster, error)
+
+	// Get returns the ManagedCluster of the given managed cluster.
+	Get(namespacedName string) (*clusterapi.ManagedCluster, error)
+}
+
+type schedulerCache struct {
+	clusterListers clusterlisters.ManagedClusterLister
+}
+
+// NumClusters returns the number of clusters in the cache.
+func (s *schedulerCache) NumClusters() int {
+	clusters, err := s.List()
+	if err != nil {
+		klog.Errorf("failed to list clusters: %v", err)
+		return 0
+	}
+	return len(clusters)
+}
+
+// List returns the list of clusters in the cache.
+func (s *schedulerCache) List() ([]*clusterapi.ManagedCluster, error) {
+	return s.clusterListers.List(labels.Everything())
+}
+
+// Get returns the ManagedCluster of the given cluster.
+func (s *schedulerCache) Get(namespacedName string) (*clusterapi.ManagedCluster, error) {
+	// Convert the namespace/name string into a distinct namespace and name
+	ns, name, err := cache.SplitMetaNamespaceKey(namespacedName)
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", namespacedName))
+		return nil, err
+	}
+
+	return s.clusterListers.ManagedClusters(ns).Get(name)
+}
+
+func New(clusterListers clusterlisters.ManagedClusterLister) Cache {
+	return &schedulerCache{
+		clusterListers: clusterListers,
+	}
+}
diff --git a/pkg/scheduler/framework/interfaces/interface.go b/pkg/scheduler/framework/interfaces/interface.go
new file mode 100644
index 0000000..0dfef98
--- /dev/null
+++ b/pkg/scheduler/framework/interfaces/interface.go
@@ -0,0 +1,403 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file defines the scheduling framework plugin interfaces.
+// The interfaces are copied from k8s.io/kubernetes/pkg/scheduler/framework/interface.go and modified
+
+package interfaces
+
+import (
+	"context"
+	"math"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/informers"
+	clientset "k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	schedulerapis "github.com/clusternet/clusternet/pkg/scheduler/apis"
+	"github.com/clusternet/clusternet/pkg/scheduler/parallelize"
+)
+
+// ClusterScoreList declares a list of clusters and their scores.
+type ClusterScoreList []ClusterScore
+
+// ClusterScore is a struct with cluster id and score.
+type ClusterScore struct {
+	NamespacedName string // the namespaced name key of a ManagedCluster
+	Score          int64
+}
+
+// PluginToClusterScores declares a map from plugin name to its ClusterScoreList.
+type PluginToClusterScores map[string]ClusterScoreList
+
+// ClusterToStatusMap declares map from cluster namespaced key to its status.
+type ClusterToStatusMap map[string]*Status
+
+// statusPrecedence defines a map from status to its precedence, larger value means higher precedent.
+var statusPrecedence = map[Code]int{
+	Error:                        3,
+	UnschedulableAndUnresolvable: 2,
+	Unschedulable:                1,
+	// Any other statuses we know today, `Skip` or `Wait`, will take precedence over `Success`.
+	Success: -1,
+}
+
+const (
+	// MaxClusterScore is the maximum score a Score plugin is expected to return.
+	MaxClusterScore int64 = 100
+
+	// MinClusterScore is the minimum score a Score plugin is expected to return.
+	MinClusterScore int64 = 0
+
+	// MaxTotalScore is the maximum total score.
+	MaxTotalScore int64 = math.MaxInt64
+)
+
+// PluginToStatus maps plugin name to status. Currently, used to identify which Filter plugin
+// returned which status.
+type PluginToStatus map[string]*Status
+
+// Merge merges the statuses in the map into one. The resulting status code have the following
+// precedence: Error, UnschedulableAndUnresolvable, Unschedulable.
+func (p PluginToStatus) Merge() *Status {
+	if len(p) == 0 {
+		return nil
+	}
+
+	finalStatus := NewStatus(Success)
+	for _, s := range p {
+		if s.Code() == Error {
+			finalStatus.err = s.AsError()
+		}
+		if statusPrecedence[s.Code()] > statusPrecedence[finalStatus.code] {
+			finalStatus.code = s.Code()
+			// Same as code, we keep the most relevant failedPlugin in the returned Status.
+			finalStatus.failedPlugin = s.FailedPlugin()
+		}
+
+		for _, r := range s.reasons {
+			finalStatus.AppendReason(r)
+		}
+	}
+
+	return finalStatus
+}
+
+// WaitingSubscription represents a subscription currently waiting in the permit phase.
+type WaitingSubscription interface {
+	// GetSubscription returns a reference to the waiting subscription.
+	GetSubscription() *appsapi.Subscription
+
+	// GetPendingPlugins returns a list of pending Permit plugin's name.
+	GetPendingPlugins() []string
+
+	// Allow declares the waiting subscription is allowed to be scheduled by the plugin named as "pluginName".
+	// If this is the last remaining plugin to allow, then a success signal is delivered
+	// to unblock the subscription.
+	Allow(pluginName string)
+
+	// Reject declares the waiting subscription Unschedulable.
+	Reject(pluginName, msg string)
+}
+
+// Plugin is the parent type for all the scheduling framework plugins.
+type Plugin interface {
+	Name() string
+}
+
+// PreFilterPlugin is an interface that must be implemented by "PreFilter" plugins.
+// These plugins are called at the beginning of the scheduling cycle.
+type PreFilterPlugin interface {
+	Plugin
+
+	// PreFilter is called at the beginning of the scheduling cycle. All PreFilter
+	// plugins must return success or the subscription will be rejected.
+	PreFilter(ctx context.Context, sub *appsapi.Subscription) *Status
+}
+
+// FilterPlugin is an interface for Filter plugins. These plugins are called at the
+// filter extension point for filtering out hosts that cannot run a subscription.
+// This concept used to be called 'predicate' in the original scheduler.
+// These plugins should return Success, Unschedulable or Error in Status.code.
+// However, the scheduler accepts other valid codes as well.
+// Anything other than "Success" will lead to exclusion of the given host from
+// running the subscription.
+type FilterPlugin interface {
+	Plugin
+
+	// Filter is called by the scheduling framework.
+	// All FilterPlugins should return "Success" to declare that
+	// the given managed cluster fits the subscription. If Filter doesn't return "Success",
+	// it will return Unschedulable, UnschedulableAndUnresolvable or Error.
+	// For the cluster being evaluated, Filter plugins should look at the passed
+	// cluster's information (e.g., subscriptions considered to be running on the cluster).
+	Filter(ctx context.Context, sub *appsapi.Subscription, cluster *clusterapi.ManagedCluster) *Status
+}
+
+// PostFilterPlugin is an interface for "PostFilter" plugins. These plugins are called
+// after a subscription cannot be scheduled.
+type PostFilterPlugin interface {
+	Plugin
+
+	// PostFilter is called by the scheduling framework.
+	// A PostFilter plugin should return one of the following statuses:
+	// - Unschedulable: the plugin gets executed successfully but the subscription cannot be made schedulable.
+	// - Success: the plugin gets executed successfully and the subscription can be made schedulable.
+	// - Error: the plugin aborts due to some internal error.
+	//
+	// Informational plugins should be configured ahead of other ones, and always return Unschedulable status.
+	// Optionally, a non-nil PostFilterResult may be returned along with a Success status. For example,
+	// a preemption plugin may choose to return nominatedClusterName, so that framework can reuse that to update the
+	// preemptor subscription's .spec.status.nominatedClusterName field.
+	PostFilter(ctx context.Context, sub *appsapi.Subscription, filteredClusterStatusMap ClusterToStatusMap) (*PostFilterResult, *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
+// state or to generate logs/metrics.
+type PreScorePlugin interface {
+	Plugin
+
+	// PreScore is called by the scheduling framework after a list of managed clusters
+	// passed the filtering phase. All prescore plugins must return success or
+	// the subscription will be rejected
+	PreScore(ctx context.Context, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster) *Status
+}
+
+// ScoreExtensions is an interface for Score extended functionality.
+type ScoreExtensions interface {
+	// NormalizeScore is called for all cluster scores produced by the same plugin's "Score"
+	// method. A successful run of NormalizeScore will update the scores list and return
+	// a success status.
+	NormalizeScore(ctx context.Context, sub *appsapi.Subscription, scores ClusterScoreList) *Status
+}
+
+// ScorePlugin is an interface that must be implemented by "Score" plugins to rank
+// clusters that passed the filtering phase.
+type ScorePlugin interface {
+	Plugin
+
+	// Score is called on each filtered cluster. It must return success and an integer
+	// indicating the rank of the cluster. All scoring plugins must return success or
+	// the subscription will be rejected.
+	Score(ctx context.Context, sub *appsapi.Subscription, namespacedCluster string) (int64, *Status)
+
+	// ScoreExtensions returns a ScoreExtensions interface if it implements one, or nil if does not.
+	ScoreExtensions() ScoreExtensions
+}
+
+// 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
+// return only Success or Error in Status.code. However, the scheduler accepts
+// other valid codes as well. Anything other than Success will lead to
+// rejection of the subscription.
+type ReservePlugin interface {
+	Plugin
+
+	// 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
+	// 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)
+}
+
+// PreBindPlugin is an interface that must be implemented by "PreBind" plugins.
+// These plugins are called before a subscription being scheduled.
+type PreBindPlugin interface {
+	Plugin
+
+	// 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, clusterNamespaces []string) *Status
+}
+
+// PostBindPlugin is an interface that must be implemented by "PostBind" plugins.
+// These plugins are called after a subscription is successfully bound to a group of ManagedCluster.
+type PostBindPlugin interface {
+	Plugin
+
+	// PostBind is called after a subscription is successfully bound. These plugins are
+	// 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)
+}
+
+// PermitPlugin is an interface that must be implemented by "Permit" plugins.
+// These plugins are called before a subscription is bound to a group of ManagedCluster.
+type PermitPlugin interface {
+	Plugin
+
+	// Permit is called before binding a subscription (and before prebind plugins). Permit
+	// plugins are used to prevent or delay the binding of a Subscription. A permit plugin
+	// must return success or wait with timeout duration, or the subscription will be rejected.
+	// 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)
+}
+
+// BindPlugin is an interface that must be implemented by "Bind" plugins. Bind
+// plugins are used to bind a subscription to a group of ManagedCluster.
+type BindPlugin interface {
+	Plugin
+
+	// Bind plugins will not be called until all pre-bind plugins have completed. Each
+	// bind plugin is called in the configured order. A bind plugin may choose whether
+	// or not to handle the given Subscription. If a bind plugin chooses to handle a Subscription, the
+	// 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
+}
+
+// Framework manages the set of plugins in use by the scheduling framework.
+// Configured plugins are called at specified points in a scheduling context.
+type Framework interface {
+	Handle
+
+	// RunPreFilterPlugins runs the set of configured PreFilter 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.
+	RunPreFilterPlugins(ctx context.Context, sub *appsapi.Subscription) *Status
+
+	// RunPostFilterPlugins runs the set of configured PostFilter plugins.
+	// PostFilter plugins can either be informational, in which case should be configured
+	// to execute first and return Unschedulable status, or ones that try to change the
+	// cluster state to make the subscription potentially schedulable in a future scheduling cycle.
+	RunPostFilterPlugins(ctx context.Context, sub *appsapi.Subscription, filteredClusterStatusMap ClusterToStatusMap) (*PostFilterResult, *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
+	// such case, subscription will not be scheduled.
+	RunReservePluginsReserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string) *Status
+
+	// RunReservePluginsUnreserve runs the Unreserve method of the set of
+	// configured Reserve plugins.
+	RunReservePluginsUnreserve(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string)
+
+	// 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
+	// running the remaining plugins and returns an error. Otherwise, if any of the
+	// 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
+
+	// 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
+
+	// RunPreBindPlugins runs the set of configured PreBind plugins. It returns
+	// *Status and its code is set to non-success if any of the plugins returns
+	// 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
+
+	// RunPostBindPlugins runs the set of configured PostBind plugins.
+	RunPostBindPlugins(ctx context.Context, sub *appsapi.Subscription, namespacedClusters []string)
+
+	// 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
+
+	// HasFilterPlugins returns true if at least one Filter plugin is defined.
+	HasFilterPlugins() bool
+
+	// HasPostFilterPlugins returns true if at least one PostFilter plugin is defined.
+	HasPostFilterPlugins() bool
+
+	// HasScorePlugins returns true if at least one Score plugin is defined.
+	HasScorePlugins() bool
+
+	// ListPlugins returns a map of extension point name to list of configured Plugins.
+	ListPlugins() *schedulerapis.Plugins
+
+	// ProfileName returns the profile name associated to this framework.
+	ProfileName() string
+}
+
+// Handle provides data and some tools that plugins can use. It is
+// passed to the plugin factories at the time of plugin initialization. Plugins
+// must store and use this handle to call framework functions.
+type Handle interface {
+	// PluginsRunner abstracts operations to run some plugins.
+	PluginsRunner
+
+	// IterateOverWaitingSubscriptions acquires a read lock and iterates over the WaitingSubscriptions map.
+	IterateOverWaitingSubscriptions(callback func(WaitingSubscription))
+
+	// GetWaitingSubscription returns a waiting subscription given its UID.
+	GetWaitingSubscription(uid types.UID) WaitingSubscription
+
+	// RejectWaitingSubscription rejects a waiting subscription given its UID.
+	RejectWaitingSubscription(uid types.UID)
+
+	// ClientSet returns a kubernetes clientSet.
+	ClientSet() clientset.Interface
+
+	// KubeConfig returns the raw kube schedulerapis.
+	KubeConfig() *restclient.Config
+
+	// EventRecorder returns an event recorder.
+	EventRecorder() record.EventRecorder
+
+	SharedInformerFactory() informers.SharedInformerFactory
+
+	// Parallelizer returns a parallelizer holding parallelism for scheduler.
+	Parallelizer() parallelize.Parallelizer
+}
+
+// PostFilterResult wraps needed info for scheduler framework to act upon PostFilter phase.
+type PostFilterResult struct {
+	NominatedNamespacedClusters []string
+}
+
+// PluginsRunner abstracts operations to run some plugins.
+// This is used by preemption PostFilter plugins when evaluating the feasibility of
+// scheduling the subscription on clusters when certain running subscriptions get evicted.
+type PluginsRunner interface {
+	// RunPreScorePlugins runs the set of configured PreScore plugins. If any
+	// of these plugins returns any status other than "Success", the given subscription is rejected.
+	RunPreScorePlugins(context.Context, *appsapi.Subscription, []*clusterapi.ManagedCluster) *Status
+
+	// RunScorePlugins runs the set of configured Score plugins. It returns a map that
+	// stores for each Score 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.
+	RunScorePlugins(context.Context, *appsapi.Subscription, []*clusterapi.ManagedCluster) (PluginToClusterScores, *Status)
+
+	// RunFilterPlugins runs the set of configured Filter plugins for subscription on
+	// the given cluster.
+	RunFilterPlugins(context.Context, *appsapi.Subscription, *clusterapi.ManagedCluster) PluginToStatus
+}
diff --git a/pkg/scheduler/framework/interfaces/interface_copied.go b/pkg/scheduler/framework/interfaces/interface_copied.go
new file mode 100644
index 0000000..eb871cf
--- /dev/null
+++ b/pkg/scheduler/framework/interfaces/interface_copied.go
@@ -0,0 +1,176 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/framework/interfaces.go and modified
+
+package interfaces
+
+import (
+	"errors"
+	"strings"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+)
+
+// Code is the Status code/type which is returned from plugins.
+type Code int
+
+// These are predefined codes used in a Status.
+const (
+	// Success means that plugin ran correctly and found subscription schedulable.
+	// NOTE: A nil status is also considered as "Success".
+	Success Code = iota
+	// Error is used for internal plugin errors, unexpected input, etc.
+	Error
+	// Unschedulable is used when a plugin finds a subscription unschedulable. The scheduler might attempt to
+	// preempt other subscriptions to get this subscription scheduled. Use UnschedulableAndUnresolvable to make the
+	// scheduler skip preemption.
+	// The accompanying status message should explain why the subscription is unschedulable.
+	Unschedulable
+	// UnschedulableAndUnresolvable is used when a plugin finds a subscription unschedulable and
+	// preemption would not change anything. Plugins should return Unschedulable if it is possible
+	// that the subscription can get scheduled with preemption.
+	// The accompanying status message should explain why the subscription is unschedulable.
+	UnschedulableAndUnresolvable
+	// Wait is used when a Permit plugin finds a subscription scheduling should wait.
+	Wait
+	// Skip is used when a Bind plugin chooses to skip binding.
+	Skip
+)
+
+// This list should be exactly the same as the codes iota defined above in the same order.
+var codes = []string{"Success", "Error", "Unschedulable", "UnschedulableAndUnresolvable", "Wait", "Skip"}
+
+func (c Code) String() string {
+	return codes[c]
+}
+
+// Status indicates the result of running a plugin. It consists of a code, a
+// message, (optionally) an error and an plugin name it fails by. When the status
+// code is not `Success`, the reasons should explain why.
+// NOTE: A nil Status is also considered as Success.
+// Copied from k8s.io/kubernetes/pkg/scheduler/framework/interface.go and modified
+type Status struct {
+	code    Code
+	reasons []string
+	err     error
+	// failedPlugin is an optional field that records the plugin name a Subscription failed by.
+	// It's set by the framework when code is Error, Unschedulable or UnschedulableAndUnresolvable.
+	failedPlugin string
+}
+
+// Code returns code of the Status.
+func (s *Status) Code() Code {
+	if s == nil {
+		return Success
+	}
+	return s.code
+}
+
+// Message returns a concatenated message on reasons of the Status.
+func (s *Status) Message() string {
+	if s == nil {
+		return ""
+	}
+	return strings.Join(s.reasons, ", ")
+}
+
+// SetFailedPlugin sets the given plugin name to s.failedPlugin.
+func (s *Status) SetFailedPlugin(plugin string) {
+	s.failedPlugin = plugin
+}
+
+// WithFailedPlugin sets the given plugin name to s.failedPlugin,
+// and returns the given status object.
+func (s *Status) WithFailedPlugin(plugin string) *Status {
+	s.SetFailedPlugin(plugin)
+	return s
+}
+
+// FailedPlugin returns the failed plugin name.
+func (s *Status) FailedPlugin() string {
+	return s.failedPlugin
+}
+
+// Reasons returns reasons of the Status.
+func (s *Status) Reasons() []string {
+	return s.reasons
+}
+
+// AppendReason appends given reason to the Status.
+func (s *Status) AppendReason(reason string) {
+	s.reasons = append(s.reasons, reason)
+}
+
+// IsSuccess returns true if and only if "Status" is nil or Code is "Success".
+func (s *Status) IsSuccess() bool {
+	return s.Code() == Success
+}
+
+// IsUnschedulable returns true if "Status" is Unschedulable (Unschedulable or UnschedulableAndUnresolvable).
+func (s *Status) IsUnschedulable() bool {
+	code := s.Code()
+	return code == Unschedulable || code == UnschedulableAndUnresolvable
+}
+
+// AsError returns nil if the status is a success; otherwise returns an "error" object
+// with a concatenated message on reasons of the Status.
+func (s *Status) AsError() error {
+	if s.IsSuccess() {
+		return nil
+	}
+	if s.err != nil {
+		return s.err
+	}
+	return errors.New(s.Message())
+}
+
+// Equal checks equality of two statuses. This is useful for testing with
+// cmp.Equal.
+func (s *Status) Equal(x *Status) bool {
+	if s == nil || x == nil {
+		return s.IsSuccess() && x.IsSuccess()
+	}
+	if s.code != x.code {
+		return false
+	}
+	if s.code == Error {
+		return cmp.Equal(s.err, x.err, cmpopts.EquateErrors())
+	}
+	return cmp.Equal(s.reasons, x.reasons)
+}
+
+// NewStatus makes a Status out of the given arguments and returns its pointer.
+func NewStatus(code Code, reasons ...string) *Status {
+	s := &Status{
+		code:    code,
+		reasons: reasons,
+	}
+	if code == Error {
+		s.err = errors.New(s.Message())
+	}
+	return s
+}
+
+// AsStatus wraps an error in a Status.
+func AsStatus(err error) *Status {
+	return &Status{
+		code:    Error,
+		reasons: []string{err.Error()},
+		err:     err,
+	}
+}
diff --git a/pkg/scheduler/framework/interfaces/types.go b/pkg/scheduler/framework/interfaces/types.go
new file mode 100644
index 0000000..cf083f9
--- /dev/null
+++ b/pkg/scheduler/framework/interfaces/types.go
@@ -0,0 +1,66 @@
+/*
+Copyright 2021 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 interfaces
+
+import (
+	"fmt"
+	"sort"
+	"strings"
+
+	"k8s.io/apimachinery/pkg/util/sets"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+)
+
+// Diagnosis records the details to diagnose a scheduling failure.
+type Diagnosis struct {
+	ClusterToStatusMap   ClusterToStatusMap
+	UnschedulablePlugins sets.String
+}
+
+// FitError describes a fit error of a subscription.
+type FitError struct {
+	Subscription   *appsapi.Subscription
+	NumAllClusters int
+	Diagnosis      Diagnosis
+}
+
+const (
+	// NoClusterAvailableMsg is used to format message when no clusters available.
+	NoClusterAvailableMsg = "0/%v clusters are available"
+)
+
+// Error returns detailed information of why the subscription failed to fit on each cluster
+func (f *FitError) Error() string {
+	reasons := make(map[string]int)
+	for _, status := range f.Diagnosis.ClusterToStatusMap {
+		for _, reason := range status.Reasons() {
+			reasons[reason]++
+		}
+	}
+
+	sortReasonsHistogram := func() []string {
+		var reasonStrings []string
+		for k, v := range reasons {
+			reasonStrings = append(reasonStrings, fmt.Sprintf("%v %v", v, k))
+		}
+		sort.Strings(reasonStrings)
+		return reasonStrings
+	}
+	reasonMsg := fmt.Sprintf(NoClusterAvailableMsg+": %v.", f.NumAllClusters, strings.Join(sortReasonsHistogram(), ", "))
+	return reasonMsg
+}
diff --git a/pkg/scheduler/framework/plugins/registry.go b/pkg/scheduler/framework/plugins/registry.go
new file mode 100644
index 0000000..00debf9
--- /dev/null
+++ b/pkg/scheduler/framework/plugins/registry.go
@@ -0,0 +1,27 @@
+/*
+Copyright 2021 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 plugins
+
+import (
+	"github.com/clusternet/clusternet/pkg/scheduler/framework/runtime"
+)
+
+// NewInTreeRegistry builds the registry with all the in-tree plugins.
+func NewInTreeRegistry() runtime.Registry {
+	// TODO: register plugins
+	return runtime.Registry{}
+}
diff --git a/pkg/scheduler/framework/runtime/framework.go b/pkg/scheduler/framework/runtime/framework.go
new file mode 100644
index 0000000..19cb5fb
--- /dev/null
+++ b/pkg/scheduler/framework/runtime/framework.go
@@ -0,0 +1,722 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/framework/runtime/framework.go and modified
+
+package runtime
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+	"k8s.io/client-go/informers"
+	clientset "k8s.io/client-go/kubernetes"
+	restclient "k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	schedulerapis "github.com/clusternet/clusternet/pkg/scheduler/apis"
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+	"github.com/clusternet/clusternet/pkg/scheduler/metrics"
+	"github.com/clusternet/clusternet/pkg/scheduler/parallelize"
+)
+
+const (
+	// Filter is the name of the filter extension point.
+	Filter = "Filter"
+	// Specifies the maximum timeout a permit plugin can return.
+	maxTimeout              = 15 * time.Minute
+	preFilter               = "PreFilter"
+	postFilter              = "PostFilter"
+	preScore                = "PreScore"
+	score                   = "Score"
+	scoreExtensionNormalize = "ScoreExtensionNormalize"
+	preBind                 = "PreBind"
+	bind                    = "Bind"
+	postBind                = "PostBind"
+	reserve                 = "Reserve"
+	unreserve               = "Unreserve"
+	permit                  = "Permit"
+)
+
+// frameworkImpl is the component responsible for initializing and running scheduler
+// plugins.
+type frameworkImpl struct {
+	registry             Registry
+	waitingSubscriptions *waitingSubscriptionsMap
+	scorePluginWeight    map[string]int
+	preFilterPlugins     []framework.PreFilterPlugin
+	filterPlugins        []framework.FilterPlugin
+	postFilterPlugins    []framework.PostFilterPlugin
+	preScorePlugins      []framework.PreScorePlugin
+	scorePlugins         []framework.ScorePlugin
+	reservePlugins       []framework.ReservePlugin
+	preBindPlugins       []framework.PreBindPlugin
+	bindPlugins          []framework.BindPlugin
+	postBindPlugins      []framework.PostBindPlugin
+	permitPlugins        []framework.PermitPlugin
+
+	clientSet       clientset.Interface
+	kubeConfig      *restclient.Config
+	eventRecorder   record.EventRecorder
+	informerFactory informers.SharedInformerFactory
+
+	metricsRecorder *metricsRecorder
+	profileName     string
+
+	parallelizer parallelize.Parallelizer
+
+	// Indicates that RunFilterPlugins should accumulate all failed statuses and not return
+	// after the first failure.
+	runAllFilters bool
+}
+
+// extensionPoint encapsulates desired and applied set of plugins at a specific extension
+// point. This is used to simplify iterating over all extension points supported by the
+// frameworkImpl.
+type extensionPoint struct {
+	// the set of plugins to be configured at this extension point.
+	plugins *schedulerapis.PluginSet
+	// a pointer to the slice storing plugins implementations that will run at this
+	// extension point.
+	slicePtr interface{}
+}
+
+func (f *frameworkImpl) getExtensionPoints(plugins *schedulerapis.Plugins) []extensionPoint {
+	return []extensionPoint{
+		{&plugins.PreFilter, &f.preFilterPlugins},
+		{&plugins.Filter, &f.filterPlugins},
+		{&plugins.PostFilter, &f.postFilterPlugins},
+		{&plugins.Reserve, &f.reservePlugins},
+		{&plugins.PreScore, &f.preScorePlugins},
+		{&plugins.Score, &f.scorePlugins},
+		{&plugins.PreBind, &f.preBindPlugins},
+		{&plugins.Bind, &f.bindPlugins},
+		{&plugins.PostBind, &f.postBindPlugins},
+		{&plugins.Permit, &f.permitPlugins},
+	}
+}
+
+type frameworkOptions struct {
+	clientSet       clientset.Interface
+	kubeConfig      *restclient.Config
+	eventRecorder   record.EventRecorder
+	informerFactory informers.SharedInformerFactory
+	runAllFilters   bool
+	parallelizer    parallelize.Parallelizer
+	metricsRecorder *metricsRecorder
+}
+
+// Option for the frameworkImpl.
+type Option func(*frameworkOptions)
+
+// WithClientSet sets clientSet for the scheduling frameworkImpl.
+func WithClientSet(clientSet clientset.Interface) Option {
+	return func(o *frameworkOptions) {
+		o.clientSet = clientSet
+	}
+}
+
+// WithKubeConfig sets kubeConfig for the scheduling frameworkImpl.
+func WithKubeConfig(kubeConfig *restclient.Config) Option {
+	return func(o *frameworkOptions) {
+		o.kubeConfig = kubeConfig
+	}
+}
+
+// WithEventRecorder sets clientSet for the scheduling frameworkImpl.
+func WithEventRecorder(recorder record.EventRecorder) Option {
+	return func(o *frameworkOptions) {
+		o.eventRecorder = recorder
+	}
+}
+
+// WithInformerFactory sets informer factory for the scheduling frameworkImpl.
+func WithInformerFactory(informerFactory informers.SharedInformerFactory) Option {
+	return func(o *frameworkOptions) {
+		o.informerFactory = informerFactory
+	}
+}
+
+// WithRunAllFilters sets the runAllFilters flag, which means RunFilterPlugins accumulates
+// all failure Statuses.
+func WithRunAllFilters(runAllFilters bool) Option {
+	return func(o *frameworkOptions) {
+		o.runAllFilters = runAllFilters
+	}
+}
+
+// WithParallelism sets parallelism for the scheduling frameworkImpl.
+func WithParallelism(parallelism int) Option {
+	return func(o *frameworkOptions) {
+		o.parallelizer = parallelize.NewParallelizer(parallelism)
+	}
+}
+
+func defaultFrameworkOptions() frameworkOptions {
+	return frameworkOptions{
+		metricsRecorder: newMetricsRecorder(1000, time.Second),
+		parallelizer:    parallelize.NewParallelizer(parallelize.DefaultParallelism),
+	}
+}
+
+var _ framework.Framework = &frameworkImpl{}
+
+// NewFramework initializes plugins given the configuration and the registry.
+func NewFramework(r Registry, opts ...Option) (framework.Framework, error) {
+	options := defaultFrameworkOptions()
+	for _, opt := range opts {
+		opt(&options)
+	}
+
+	f := &frameworkImpl{
+		registry:             r,
+		scorePluginWeight:    make(map[string]int),
+		waitingSubscriptions: newWaitingSubscriptionsMap(),
+		clientSet:            options.clientSet,
+		kubeConfig:           options.kubeConfig,
+		eventRecorder:        options.eventRecorder,
+		informerFactory:      options.informerFactory,
+		metricsRecorder:      options.metricsRecorder,
+		runAllFilters:        options.runAllFilters,
+		parallelizer:         options.parallelizer,
+		profileName:          "default",
+	}
+
+	// TODO: check pluginsNeeded
+
+	// Verifying the score weights again since Plugin.Name() could return a different
+	// value from the one used in the configuration.
+	for _, scorePlugin := range f.scorePlugins {
+		if f.scorePluginWeight[scorePlugin.Name()] == 0 {
+			return nil, fmt.Errorf("score plugin %q is not configured with weight", scorePlugin.Name())
+		}
+	}
+
+	if len(f.bindPlugins) == 0 {
+		return nil, fmt.Errorf("at least one bind plugin is needed")
+	}
+
+	return f, nil
+}
+
+// RunPreFilterPlugins runs the set of configured PreFilter 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.
+func (f *frameworkImpl) RunPreFilterPlugins(ctx context.Context, sub *appsapi.Subscription) (status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(preFilter, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	for _, pl := range f.preFilterPlugins {
+		status = f.runPreFilterPlugin(ctx, pl, sub)
+		if !status.IsSuccess() {
+			status.SetFailedPlugin(pl.Name())
+			if status.IsUnschedulable() {
+				return status
+			}
+			return framework.AsStatus(fmt.Errorf("running PreFilter plugin %q: %w", pl.Name(), status.AsError())).WithFailedPlugin(pl.Name())
+		}
+	}
+
+	return nil
+}
+
+func (f *frameworkImpl) runPreFilterPlugin(ctx context.Context, pl framework.PreFilterPlugin, sub *appsapi.Subscription) *framework.Status {
+	startTime := time.Now()
+	status := pl.PreFilter(ctx, sub)
+	f.metricsRecorder.observePluginDurationAsync(preFilter, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return status
+}
+
+// RunFilterPlugins runs the set of configured Filter plugins for subscription on
+// the given cluster. If any of these plugins doesn't return "Success", the
+// given cluster is not suitable for running subscription.
+// Meanwhile, the failure message and status are set for the given cluster.
+func (f *frameworkImpl) RunFilterPlugins(ctx context.Context, sub *appsapi.Subscription, cluster *clusterapi.ManagedCluster) framework.PluginToStatus {
+	statuses := make(framework.PluginToStatus)
+	for _, pl := range f.filterPlugins {
+		pluginStatus := f.runFilterPlugin(ctx, pl, sub, cluster)
+		if !pluginStatus.IsSuccess() {
+			if !pluginStatus.IsUnschedulable() {
+				// Filter plugins are not supposed to return any status other than
+				// Success or Unschedulable.
+				errStatus := framework.AsStatus(fmt.Errorf("running %q filter plugin: %w", pl.Name(), pluginStatus.AsError())).WithFailedPlugin(pl.Name())
+				return map[string]*framework.Status{pl.Name(): errStatus}
+			}
+			pluginStatus.SetFailedPlugin(pl.Name())
+			statuses[pl.Name()] = pluginStatus
+			if !f.runAllFilters {
+				// Exit early if we don't need to run all filters.
+				return statuses
+			}
+		}
+	}
+
+	return statuses
+}
+
+func (f *frameworkImpl) runFilterPlugin(ctx context.Context, pl framework.FilterPlugin, sub *appsapi.Subscription, cluster *clusterapi.ManagedCluster) *framework.Status {
+	startTime := time.Now()
+	status := pl.Filter(ctx, sub, cluster)
+	f.metricsRecorder.observePluginDurationAsync(Filter, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return status
+}
+
+// RunPostFilterPlugins runs the set of configured PostFilter plugins until the first
+// Success or Error is met, otherwise continues to execute all plugins.
+func (f *frameworkImpl) RunPostFilterPlugins(ctx context.Context, sub *appsapi.Subscription, filteredClusterStatusMap framework.ClusterToStatusMap) (_ *framework.PostFilterResult, status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(postFilter, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+
+	statuses := make(framework.PluginToStatus)
+	for _, pl := range f.postFilterPlugins {
+		r, s := f.runPostFilterPlugin(ctx, pl, sub, filteredClusterStatusMap)
+		if s.IsSuccess() {
+			return r, s
+		} else if !s.IsUnschedulable() {
+			// Any status other than Success or Unschedulable is Error.
+			return nil, framework.AsStatus(s.AsError())
+		}
+		statuses[pl.Name()] = s
+	}
+
+	return nil, statuses.Merge()
+}
+
+func (f *frameworkImpl) runPostFilterPlugin(ctx context.Context, pl framework.PostFilterPlugin, sub *appsapi.Subscription, filteredClusterStatusMap framework.ClusterToStatusMap) (*framework.PostFilterResult, *framework.Status) {
+	startTime := time.Now()
+	r, s := pl.PostFilter(ctx, sub, filteredClusterStatusMap)
+	f.metricsRecorder.observePluginDurationAsync(postFilter, pl.Name(), s, metrics.SinceInSeconds(startTime))
+	return r, s
+}
+
+// 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) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(preScore, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	for _, pl := range f.preScorePlugins {
+		status = f.runPreScorePlugin(ctx, pl, sub, clusters)
+		if !status.IsSuccess() {
+			return framework.AsStatus(fmt.Errorf("running PreScore plugin %q: %w", pl.Name(), status.AsError()))
+		}
+	}
+
+	return nil
+}
+
+func (f *frameworkImpl) runPreScorePlugin(ctx context.Context, pl framework.PreScorePlugin, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster) *framework.Status {
+	startTime := time.Now()
+	status := pl.PreScore(ctx, sub, clusters)
+	f.metricsRecorder.observePluginDurationAsync(preScore, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return status
+}
+
+// RunScorePlugins runs the set of configured scoring plugins. It returns a list that
+// stores for each scoring 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.
+func (f *frameworkImpl) RunScorePlugins(ctx context.Context, sub *appsapi.Subscription, clusters []*clusterapi.ManagedCluster) (ps framework.PluginToClusterScores, status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(score, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	pluginToClusterScores := make(framework.PluginToClusterScores, len(f.scorePlugins))
+	for _, pl := range f.scorePlugins {
+		pluginToClusterScores[pl.Name()] = make(framework.ClusterScoreList, len(clusters))
+	}
+	ctx, cancel := context.WithCancel(ctx)
+	errCh := parallelize.NewErrorChannel()
+
+	// Run Score method for each cluster in parallel.
+	f.Parallelizer().Until(ctx, len(clusters), func(index int) {
+		for _, pl := range f.scorePlugins {
+			clusterNamespacedName := klog.KObj(clusters[index]).String()
+			s, status := f.runScorePlugin(ctx, pl, sub, clusterNamespacedName)
+			if !status.IsSuccess() {
+				err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
+				errCh.SendErrorWithCancel(err, cancel)
+				return
+			}
+			pluginToClusterScores[pl.Name()][index] = framework.ClusterScore{
+				NamespacedName: clusterNamespacedName,
+				Score:          s,
+			}
+		}
+	})
+	if err := errCh.ReceiveError(); err != nil {
+		return nil, framework.AsStatus(fmt.Errorf("running Score plugins: %w", err))
+	}
+
+	// Run NormalizeScore method for each ScorePlugin in parallel.
+	f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
+		pl := f.scorePlugins[index]
+		ClusterScoreList := pluginToClusterScores[pl.Name()]
+		if pl.ScoreExtensions() == nil {
+			return
+		}
+		status := f.runScoreExtension(ctx, pl, sub, ClusterScoreList)
+		if !status.IsSuccess() {
+			err := fmt.Errorf("plugin %q failed with: %w", pl.Name(), status.AsError())
+			errCh.SendErrorWithCancel(err, cancel)
+			return
+		}
+	})
+	if err := errCh.ReceiveError(); err != nil {
+		return nil, framework.AsStatus(fmt.Errorf("running Normalize on Score plugins: %w", err))
+	}
+
+	// Apply score defaultWeights for each ScorePlugin in parallel.
+	f.Parallelizer().Until(ctx, len(f.scorePlugins), func(index int) {
+		pl := f.scorePlugins[index]
+		// Score plugins' weight has been checked when they are initialized.
+		weight := f.scorePluginWeight[pl.Name()]
+		ClusterScoreList := pluginToClusterScores[pl.Name()]
+
+		for i, clusterScore := range ClusterScoreList {
+			// return error if score plugin returns invalid score.
+			if clusterScore.Score > framework.MaxClusterScore || clusterScore.Score < framework.MinClusterScore {
+				err := fmt.Errorf("plugin %q returns an invalid score %v, it should in the range of [%v, %v] after normalizing", pl.Name(), clusterScore.Score, framework.MinClusterScore, framework.MaxClusterScore)
+				errCh.SendErrorWithCancel(err, cancel)
+				return
+			}
+			ClusterScoreList[i].Score = clusterScore.Score * int64(weight)
+		}
+	})
+	if err := errCh.ReceiveError(); err != nil {
+		return nil, framework.AsStatus(fmt.Errorf("applying score defaultWeights on Score plugins: %w", err))
+	}
+
+	return pluginToClusterScores, nil
+}
+
+func (f *frameworkImpl) runScorePlugin(ctx context.Context, pl framework.ScorePlugin, sub *appsapi.Subscription, clusterNamespace string) (int64, *framework.Status) {
+	startTime := time.Now()
+	s, status := pl.Score(ctx, sub, clusterNamespace)
+	f.metricsRecorder.observePluginDurationAsync(score, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return s, status
+}
+
+func (f *frameworkImpl) runScoreExtension(ctx context.Context, pl framework.ScorePlugin, sub *appsapi.Subscription, ClusterScoreList framework.ClusterScoreList) *framework.Status {
+	startTime := time.Now()
+	status := pl.ScoreExtensions().NormalizeScore(ctx, sub, ClusterScoreList)
+	f.metricsRecorder.observePluginDurationAsync(scoreExtensionNormalize, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return 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.
+func (f *frameworkImpl) RunPreBindPlugins(ctx context.Context, sub *appsapi.Subscription, clusterNamespaces []string) (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, clusterNamespaces)
+		if !status.IsSuccess() {
+			err := status.AsError()
+			klog.ErrorS(err, "Failed running PreBind plugin", "plugin", pl.Name(), "sub", klog.KObj(sub))
+			return framework.AsStatus(fmt.Errorf("running PreBind plugin %q: %w", pl.Name(), err))
+		}
+	}
+	return nil
+}
+
+func (f *frameworkImpl) runPreBindPlugin(ctx context.Context, pl framework.PreBindPlugin, sub *appsapi.Subscription, clusterNamespaces []string) *framework.Status {
+	startTime := time.Now()
+	status := pl.PreBind(ctx, sub, clusterNamespaces)
+	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, clusterNamespaces []string) (status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(bind, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	if len(f.bindPlugins) == 0 {
+		return framework.NewStatus(framework.Skip, "")
+	}
+	for _, bp := range f.bindPlugins {
+		status = f.runBindPlugin(ctx, bp, sub, clusterNamespaces)
+		if status != nil && status.Code() == framework.Skip {
+			continue
+		}
+		if !status.IsSuccess() {
+			err := status.AsError()
+			klog.ErrorS(err, "Failed running Bind plugin", "plugin", bp.Name(), "sub", klog.KObj(sub))
+			return framework.AsStatus(fmt.Errorf("running Bind plugin %q: %w", bp.Name(), err))
+		}
+		return status
+	}
+	return status
+}
+
+func (f *frameworkImpl) runBindPlugin(ctx context.Context, bp framework.BindPlugin, sub *appsapi.Subscription, clusterNamespaces []string) *framework.Status {
+	startTime := time.Now()
+	status := bp.Bind(ctx, sub, clusterNamespaces)
+	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, clusterNamespaces []string) {
+	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, clusterNamespaces)
+	}
+}
+
+func (f *frameworkImpl) runPostBindPlugin(ctx context.Context, pl framework.PostBindPlugin, sub *appsapi.Subscription, clusterNamespaces []string) {
+	startTime := time.Now()
+	pl.PostBind(ctx, sub, clusterNamespaces)
+	f.metricsRecorder.observePluginDurationAsync(postBind, pl.Name(), nil, metrics.SinceInSeconds(startTime))
+}
+
+// RunReservePluginsReserve runs the Reserve method in the set of configured
+// reserve plugins. If any of these plugins returns an error, it does not
+// 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, clusterNamespaces []string) (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, clusterNamespaces)
+		if !status.IsSuccess() {
+			err := status.AsError()
+			klog.ErrorS(err, "Failed running Reserve plugin", "plugin", pl.Name(), "subscription", klog.KObj(sub))
+			return framework.AsStatus(fmt.Errorf("running Reserve plugin %q: %w", pl.Name(), err))
+		}
+	}
+	return nil
+}
+
+func (f *frameworkImpl) runReservePluginReserve(ctx context.Context, pl framework.ReservePlugin, sub *appsapi.Subscription, clusterNamespaces []string) *framework.Status {
+	startTime := time.Now()
+	status := pl.Reserve(ctx, sub, clusterNamespaces)
+	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, clusterNamespaces []string) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(unreserve, framework.Success.String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	// 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, clusterNamespaces)
+	}
+}
+
+func (f *frameworkImpl) runReservePluginUnreserve(ctx context.Context, pl framework.ReservePlugin, sub *appsapi.Subscription, clusterNamespaces []string) {
+	startTime := time.Now()
+	pl.Unreserve(ctx, sub, clusterNamespaces)
+	f.metricsRecorder.observePluginDurationAsync(unreserve, pl.Name(), nil, metrics.SinceInSeconds(startTime))
+}
+
+// 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
+// running the remaining plugins and returns an error. Otherwise, if any of the
+// 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, clusterNamespaces []string) (status *framework.Status) {
+	startTime := time.Now()
+	defer func() {
+		metrics.FrameworkExtensionPointDuration.WithLabelValues(permit, status.Code().String(), f.profileName).Observe(metrics.SinceInSeconds(startTime))
+	}()
+	pluginsWaitTime := make(map[string]time.Duration)
+	statusCode := framework.Success
+	for _, pl := range f.permitPlugins {
+		status, timeout := f.runPermitPlugin(ctx, pl, sub, clusterNamespaces)
+		if !status.IsSuccess() {
+			if status.IsUnschedulable() {
+				msg := fmt.Sprintf("rejected subscription %q by permit plugin %q: %v", sub.Name, pl.Name(), status.Message())
+				klog.V(4).Infof(msg)
+				status.SetFailedPlugin(pl.Name())
+				return status
+			}
+			if status.Code() == framework.Wait {
+				// Not allowed to be greater than maxTimeout.
+				if timeout > maxTimeout {
+					timeout = maxTimeout
+				}
+				pluginsWaitTime[pl.Name()] = timeout
+				statusCode = framework.Wait
+			} else {
+				err := status.AsError()
+				klog.ErrorS(err, "Failed running Permit plugin", "plugin", pl.Name(), "subscription", klog.KObj(sub))
+				return framework.AsStatus(fmt.Errorf("running Permit plugin %q: %w", pl.Name(), err)).WithFailedPlugin(pl.Name())
+			}
+		}
+	}
+	if statusCode == framework.Wait {
+		waitingSubscription := newWaitingSubscription(sub, pluginsWaitTime)
+		f.waitingSubscriptions.add(waitingSubscription)
+		msg := fmt.Sprintf("one or more plugins asked to wait and no plugin rejected subscription %q", sub.Name)
+		klog.V(4).Infof(msg)
+		return framework.NewStatus(framework.Wait, msg)
+	}
+	return nil
+}
+
+func (f *frameworkImpl) runPermitPlugin(ctx context.Context, pl framework.PermitPlugin, sub *appsapi.Subscription, clusterNamespaces []string) (*framework.Status, time.Duration) {
+	startTime := time.Now()
+	status, timeout := pl.Permit(ctx, sub, clusterNamespaces)
+	f.metricsRecorder.observePluginDurationAsync(permit, pl.Name(), status, metrics.SinceInSeconds(startTime))
+	return status, timeout
+}
+
+// WaitOnPermit will block, if the subscription is a waiting subscription, until the waiting subscription is rejected or allowed.
+func (f *frameworkImpl) WaitOnPermit(ctx context.Context, sub *appsapi.Subscription) *framework.Status {
+	waitingSubscription := f.waitingSubscriptions.get(sub.UID)
+	if waitingSubscription == nil {
+		return nil
+	}
+	defer f.waitingSubscriptions.remove(sub.UID)
+	klog.V(4).Infof("subscription %q waiting on permit", sub.Name)
+
+	startTime := time.Now()
+	s := <-waitingSubscription.s
+	metrics.PermitWaitDuration.WithLabelValues(s.Code().String()).Observe(metrics.SinceInSeconds(startTime))
+
+	if !s.IsSuccess() {
+		if s.IsUnschedulable() {
+			msg := fmt.Sprintf("subscription %q rejected while waiting on permit: %v", sub.Name, s.Message())
+			klog.V(4).Infof(msg)
+			s.SetFailedPlugin(s.FailedPlugin())
+			return s
+		}
+		err := s.AsError()
+		klog.ErrorS(err, "Failed waiting on permit for subscription", "subscription", klog.KObj(sub))
+		return framework.AsStatus(fmt.Errorf("waiting on permit for subscription: %w", err)).WithFailedPlugin(s.FailedPlugin())
+	}
+	return nil
+}
+
+// IterateOverWaitingSubscriptions acquires a read lock and iterates over the WaitingSubscriptions map.
+func (f *frameworkImpl) IterateOverWaitingSubscriptions(callback func(framework.WaitingSubscription)) {
+	f.waitingSubscriptions.iterate(callback)
+}
+
+// GetWaitingSubscription returns a reference to a WaitingSubscription given its UID.
+func (f *frameworkImpl) GetWaitingSubscription(uid types.UID) framework.WaitingSubscription {
+	if wp := f.waitingSubscriptions.get(uid); wp != nil {
+		return wp
+	}
+	return nil // Returning nil instead of *waitingSubscription(nil).
+}
+
+// RejectWaitingSubscription rejects a WaitingSubscription given its UID.
+func (f *frameworkImpl) RejectWaitingSubscription(uid types.UID) {
+	waitingSubscription := f.waitingSubscriptions.get(uid)
+	if waitingSubscription != nil {
+		waitingSubscription.Reject("", "removed")
+	}
+}
+
+// HasFilterPlugins returns true if at least one filter plugin is defined.
+func (f *frameworkImpl) HasFilterPlugins() bool {
+	return len(f.filterPlugins) > 0
+}
+
+// HasPostFilterPlugins returns true if at least one postFilter plugin is defined.
+func (f *frameworkImpl) HasPostFilterPlugins() bool {
+	return len(f.postFilterPlugins) > 0
+}
+
+// HasScorePlugins returns true if at least one score plugin is defined.
+func (f *frameworkImpl) HasScorePlugins() bool {
+	return len(f.scorePlugins) > 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 {
+	m := schedulerapis.Plugins{}
+
+	for _, e := range f.getExtensionPoints(&m) {
+		plugins := reflect.ValueOf(e.slicePtr).Elem()
+		extName := plugins.Type().Elem().Name()
+		var cfgs []schedulerapis.Plugin
+		for i := 0; i < plugins.Len(); i++ {
+			name := plugins.Index(i).Interface().(framework.Plugin).Name()
+			p := schedulerapis.Plugin{Name: name}
+			if extName == "ScorePlugin" {
+				// Weights apply only to score plugins.
+				p.Weight = int32(f.scorePluginWeight[name])
+			}
+			cfgs = append(cfgs, p)
+		}
+		if len(cfgs) > 0 {
+			e.plugins.Enabled = cfgs
+		}
+	}
+	return &m
+}
+
+// ClientSet returns a kubernetes clientset.
+func (f *frameworkImpl) ClientSet() clientset.Interface {
+	return f.clientSet
+}
+
+// KubeConfig returns a kubernetes schedulerapis.
+func (f *frameworkImpl) KubeConfig() *restclient.Config {
+	return f.kubeConfig
+}
+
+// EventRecorder returns an event recorder.
+func (f *frameworkImpl) EventRecorder() record.EventRecorder {
+	return f.eventRecorder
+}
+
+// SharedInformerFactory returns a shared informer factory.
+func (f *frameworkImpl) SharedInformerFactory() informers.SharedInformerFactory {
+	return f.informerFactory
+}
+
+// ProfileName returns the profile name associated to this framework.
+func (f *frameworkImpl) ProfileName() string {
+	return f.profileName
+}
+
+// Parallelizer returns a parallelizer holding parallelism for scheduler.
+func (f *frameworkImpl) Parallelizer() parallelize.Parallelizer {
+	return f.parallelizer
+}
diff --git a/pkg/scheduler/framework/runtime/metrics_recorder.go b/pkg/scheduler/framework/runtime/metrics_recorder.go
new file mode 100644
index 0000000..402d347
--- /dev/null
+++ b/pkg/scheduler/framework/runtime/metrics_recorder.go
@@ -0,0 +1,105 @@
+/*
+Copyright 2019 The Kubernetes 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/framework/runtime/metrics_recorder.go
+
+package runtime
+
+import (
+	"time"
+
+	k8smetrics "k8s.io/component-base/metrics"
+
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+	"github.com/clusternet/clusternet/pkg/scheduler/metrics"
+)
+
+// frameworkMetric is the data structure passed in the buffer channel between the main framework thread
+// and the metricsRecorder goroutine.
+type frameworkMetric struct {
+	metric      *k8smetrics.HistogramVec
+	labelValues []string
+	value       float64
+}
+
+// metricRecorder records framework metrics in a separate goroutine to avoid overhead in the critical path.
+type metricsRecorder struct {
+	// bufferCh is a channel that serves as a metrics buffer before the metricsRecorder goroutine reports it.
+	bufferCh chan *frameworkMetric
+	// if bufferSize is reached, incoming metrics will be discarded.
+	bufferSize int
+	// how often the recorder runs to flush the metrics.
+	interval time.Duration
+
+	// stopCh is used to stop the goroutine which periodically flushes metrics. It's currently only
+	// used in tests.
+	stopCh chan struct{}
+	// isStoppedCh indicates whether the goroutine is stopped. It's used in tests only to make sure
+	// the metric flushing goroutine is stopped so that tests can collect metrics for verification.
+	isStoppedCh chan struct{}
+}
+
+func newMetricsRecorder(bufferSize int, interval time.Duration) *metricsRecorder {
+	recorder := &metricsRecorder{
+		bufferCh:    make(chan *frameworkMetric, bufferSize),
+		bufferSize:  bufferSize,
+		interval:    interval,
+		stopCh:      make(chan struct{}),
+		isStoppedCh: make(chan struct{}),
+	}
+	go recorder.run()
+	return recorder
+}
+
+// observePluginDurationAsync observes the plugin_execution_duration_seconds metric.
+// The metric will be flushed to Prometheus asynchronously.
+func (r *metricsRecorder) observePluginDurationAsync(extensionPoint, pluginName string, status *framework.Status, value float64) {
+	newMetric := &frameworkMetric{
+		metric:      metrics.PluginExecutionDuration,
+		labelValues: []string{pluginName, extensionPoint, status.Code().String()},
+		value:       value,
+	}
+	select {
+	case r.bufferCh <- newMetric:
+	default:
+	}
+}
+
+// run flushes buffered metrics into Prometheus every second.
+func (r *metricsRecorder) run() {
+	for {
+		select {
+		case <-r.stopCh:
+			close(r.isStoppedCh)
+			return
+		default:
+		}
+		r.flushMetrics()
+		time.Sleep(r.interval)
+	}
+}
+
+// flushMetrics tries to clean up the bufferCh by reading at most bufferSize metrics.
+func (r *metricsRecorder) flushMetrics() {
+	for i := 0; i < r.bufferSize; i++ {
+		select {
+		case m := <-r.bufferCh:
+			m.metric.WithLabelValues(m.labelValues...).Observe(m.value)
+		default:
+			return
+		}
+	}
+}
diff --git a/pkg/scheduler/framework/runtime/registry.go b/pkg/scheduler/framework/runtime/registry.go
new file mode 100644
index 0000000..6faaa7a
--- /dev/null
+++ b/pkg/scheduler/framework/runtime/registry.go
@@ -0,0 +1,65 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/framework/runtime/registry.go and modified
+
+package runtime
+
+import (
+	"fmt"
+
+	"k8s.io/apimachinery/pkg/runtime"
+
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+)
+
+// PluginFactory is a function that builds a plugin.
+type PluginFactory = func(configuration runtime.Object, f framework.Handle) (framework.Plugin, error)
+
+// Registry is a collection of all available plugins. The framework uses a
+// registry to enable and initialize configured plugins.
+// All plugins must be in the registry before initializing the framework.
+type Registry map[string]PluginFactory
+
+// Register adds a new plugin to the registry. If a plugin with the same name
+// exists, it returns an error.
+func (r Registry) Register(name string, factory PluginFactory) error {
+	if _, ok := r[name]; ok {
+		return fmt.Errorf("a plugin named %v already exists", name)
+	}
+	r[name] = factory
+	return nil
+}
+
+// Unregister removes an existing plugin from the registry. If no plugin with
+// the provided name exists, it returns an error.
+func (r Registry) Unregister(name string) error {
+	if _, ok := r[name]; !ok {
+		return fmt.Errorf("no plugin named %v exists", name)
+	}
+	delete(r, name)
+	return nil
+}
+
+// Merge merges the provided registry to the current one.
+func (r Registry) Merge(in Registry) error {
+	for name, factory := range in {
+		if err := r.Register(name, factory); err != nil {
+			return err
+		}
+	}
+	return nil
+}
diff --git a/pkg/scheduler/framework/runtime/waiting_susbcriptions_map.go b/pkg/scheduler/framework/runtime/waiting_susbcriptions_map.go
new file mode 100644
index 0000000..4a1e453
--- /dev/null
+++ b/pkg/scheduler/framework/runtime/waiting_susbcriptions_map.go
@@ -0,0 +1,168 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/framework/runtime/waiting_pods_map.go and modified.
+
+package runtime
+
+import (
+	"fmt"
+	"sync"
+	"time"
+
+	"k8s.io/apimachinery/pkg/types"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+)
+
+// waitingSubscriptionsMap a thread-safe map used to maintain subscriptions waiting in the permit phase.
+type waitingSubscriptionsMap struct {
+	subs map[types.UID]*waitingSubscription
+	mu   sync.RWMutex
+}
+
+// newWaitingSubscriptionsMap returns a new waitingSubscriptionsMap.
+func newWaitingSubscriptionsMap() *waitingSubscriptionsMap {
+	return &waitingSubscriptionsMap{
+		subs: make(map[types.UID]*waitingSubscription),
+	}
+}
+
+// add a new WaitingSubscription to the map.
+func (m *waitingSubscriptionsMap) add(wp *waitingSubscription) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.subs[wp.GetSubscription().UID] = wp
+}
+
+// remove a WaitingSubscription from the map.
+func (m *waitingSubscriptionsMap) remove(uid types.UID) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	delete(m.subs, uid)
+}
+
+// get a WaitingSubscription from the map.
+func (m *waitingSubscriptionsMap) get(uid types.UID) *waitingSubscription {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	return m.subs[uid]
+}
+
+// iterate acquires a read lock and iterates over the WaitingSubscriptions map.
+func (m *waitingSubscriptionsMap) iterate(callback func(subscription framework.WaitingSubscription)) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+	for _, v := range m.subs {
+		callback(v)
+	}
+}
+
+// waitingSubscription represents a subscription waiting in the permit phase.
+type waitingSubscription struct {
+	sub            *appsapi.Subscription
+	pendingPlugins map[string]*time.Timer
+	s              chan *framework.Status
+	mu             sync.RWMutex
+}
+
+var _ framework.WaitingSubscription = &waitingSubscription{}
+
+// newWaitingSubscription returns a new waitingSubscription instance.
+func newWaitingSubscription(sub *appsapi.Subscription, pluginsMaxWaitTime map[string]time.Duration) *waitingSubscription {
+	wp := &waitingSubscription{
+		sub: sub,
+		// Allow() and Reject() calls are non-blocking. This property is guaranteed
+		// by using non-blocking send to this channel. This channel has a buffer of size 1
+		// to ensure that non-blocking send will not be ignored - possible situation when
+		// receiving from this channel happens after non-blocking send.
+		s: make(chan *framework.Status, 1),
+	}
+
+	wp.pendingPlugins = make(map[string]*time.Timer, len(pluginsMaxWaitTime))
+	// The time.AfterFunc calls wp.Reject which iterates through pendingPlugins map. Acquire the
+	// lock here so that time.AfterFunc can only execute after newWaitingSubscription finishes.
+	wp.mu.Lock()
+	defer wp.mu.Unlock()
+	for k, v := range pluginsMaxWaitTime {
+		plugin, waitTime := k, v
+		wp.pendingPlugins[plugin] = time.AfterFunc(waitTime, func() {
+			msg := fmt.Sprintf("rejected due to timeout after waiting %v at plugin %v",
+				waitTime, plugin)
+			wp.Reject(plugin, msg)
+		})
+	}
+
+	return wp
+}
+
+// GetSubscription returns a reference to the waiting subscription.
+func (w *waitingSubscription) GetSubscription() *appsapi.Subscription {
+	return w.sub
+}
+
+// GetPendingPlugins returns a list of pending permit plugin's name.
+func (w *waitingSubscription) GetPendingPlugins() []string {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+	plugins := make([]string, 0, len(w.pendingPlugins))
+	for p := range w.pendingPlugins {
+		plugins = append(plugins, p)
+	}
+
+	return plugins
+}
+
+// Allow declares the waiting subscription is allowed to be scheduled by plugin pluginName.
+// If this is the last remaining plugin to allow, then a success signal is delivered
+// to unblock the subscription.
+func (w *waitingSubscription) Allow(pluginName string) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if timer, exist := w.pendingPlugins[pluginName]; exist {
+		timer.Stop()
+		delete(w.pendingPlugins, pluginName)
+	}
+
+	// Only signal success status after all plugins have allowed
+	if len(w.pendingPlugins) != 0 {
+		return
+	}
+
+	// The select clause works as a non-blocking send.
+	// If there is no receiver, it's a no-op (default case).
+	select {
+	case w.s <- framework.NewStatus(framework.Success, ""):
+	default:
+	}
+}
+
+// Reject declares the waiting subscription unschedulable.
+func (w *waitingSubscription) Reject(pluginName, msg string) {
+	w.mu.RLock()
+	defer w.mu.RUnlock()
+	for _, timer := range w.pendingPlugins {
+		timer.Stop()
+	}
+
+	// The select clause works as a non-blocking send.
+	// If there is no receiver, it's a no-op (default case).
+	select {
+	case w.s <- framework.NewStatus(framework.Unschedulable, msg).WithFailedPlugin(pluginName):
+	default:
+	}
+}
diff --git a/pkg/scheduler/metrics/metrics.go b/pkg/scheduler/metrics/metrics.go
new file mode 100644
index 0000000..4413a70
--- /dev/null
+++ b/pkg/scheduler/metrics/metrics.go
@@ -0,0 +1,148 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/metrics/metrics.go and modified
+
+package metrics
+
+import (
+	"sync"
+	"time"
+
+	"k8s.io/component-base/metrics"
+	"k8s.io/component-base/metrics/legacyregistry"
+)
+
+const (
+	// SchedulerSubsystem - subsystem name used by scheduler
+	SchedulerSubsystem = "clusternet_scheduler"
+
+	// Binding - binding operation label value
+	Binding = "binding"
+)
+
+// All the histogram based metrics have 1ms as size for the smallest bucket.
+var (
+	scheduleAttempts = metrics.NewCounterVec(
+		&metrics.CounterOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "schedule_attempts_total",
+			Help:           "Number of attempts to schedule subscriptions, by the result. 'unschedulable' means a subscription could not be scheduled, while 'error' means an internal scheduler problem.",
+			StabilityLevel: metrics.ALPHA,
+		}, []string{"result", "profile"})
+
+	e2eSchedulingLatency = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "e2e_scheduling_duration_seconds",
+			Help:           "E2e scheduling latency in seconds (scheduling algorithm + binding)",
+			Buckets:        metrics.ExponentialBuckets(0.001, 2, 15),
+			StabilityLevel: metrics.ALPHA,
+		}, []string{"result", "profile"})
+
+	schedulingLatency = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "scheduling_attempt_duration_seconds",
+			Help:           "Scheduling attempt latency in seconds (scheduling algorithm + binding)",
+			Buckets:        metrics.ExponentialBuckets(0.001, 2, 15),
+			StabilityLevel: metrics.STABLE,
+		}, []string{"result", "profile"})
+
+	SchedulingAlgorithmLatency = metrics.NewHistogram(
+		&metrics.HistogramOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "scheduling_algorithm_duration_seconds",
+			Help:           "Scheduling algorithm latency in seconds",
+			Buckets:        metrics.ExponentialBuckets(0.001, 2, 15),
+			StabilityLevel: metrics.ALPHA,
+		},
+	)
+
+	SchedulerGoroutines = metrics.NewGaugeVec(
+		&metrics.GaugeOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "scheduler_goroutines",
+			Help:           "Number of running goroutines split by the work they do such as binding.",
+			StabilityLevel: metrics.ALPHA,
+		}, []string{"work"})
+
+	FrameworkExtensionPointDuration = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem: SchedulerSubsystem,
+			Name:      "framework_extension_point_duration_seconds",
+			Help:      "Latency for running all plugins of a specific extension point.",
+			// Start with 0.1ms with the last bucket being [~200ms, Inf)
+			Buckets:        metrics.ExponentialBuckets(0.0001, 2, 12),
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"extension_point", "status", "profile"})
+
+	PluginExecutionDuration = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem: SchedulerSubsystem,
+			Name:      "plugin_execution_duration_seconds",
+			Help:      "Duration for running a plugin at a specific extension point.",
+			// Start with 0.01ms with the last bucket being [~22ms, Inf). We use a small factor (1.5)
+			// so that we have better granularity since plugin latency is very sensitive.
+			Buckets:        metrics.ExponentialBuckets(0.00001, 1.5, 20),
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"plugin", "extension_point", "status"})
+
+	PermitWaitDuration = metrics.NewHistogramVec(
+		&metrics.HistogramOpts{
+			Subsystem:      SchedulerSubsystem,
+			Name:           "permit_wait_duration_seconds",
+			Help:           "Duration of waiting on permit.",
+			Buckets:        metrics.ExponentialBuckets(0.001, 2, 15),
+			StabilityLevel: metrics.ALPHA,
+		},
+		[]string{"result"})
+
+	metricsList = []metrics.Registerable{
+		scheduleAttempts,
+		e2eSchedulingLatency,
+		SchedulingAlgorithmLatency,
+		FrameworkExtensionPointDuration,
+		PluginExecutionDuration,
+		SchedulerGoroutines,
+		PermitWaitDuration,
+	}
+)
+
+var registerMetrics sync.Once
+
+// Register all metrics.
+func Register() {
+	// Register the metrics.
+	registerMetrics.Do(func() {
+		RegisterMetrics(metricsList...)
+	})
+}
+
+// RegisterMetrics registers a list of metrics.
+// This function is exported because it is intended to be used by out-of-tree plugins to register their custom metrics.
+func RegisterMetrics(extraMetrics ...metrics.Registerable) {
+	for _, metric := range extraMetrics {
+		legacyregistry.MustRegister(metric)
+	}
+}
+
+// SinceInSeconds gets the time since the specified start in seconds.
+func SinceInSeconds(start time.Time) float64 {
+	return time.Since(start).Seconds()
+}
diff --git a/pkg/scheduler/metrics/profile_metrics.go b/pkg/scheduler/metrics/profile_metrics.go
new file mode 100644
index 0000000..1259e45
--- /dev/null
+++ b/pkg/scheduler/metrics/profile_metrics.go
@@ -0,0 +1,50 @@
+/*
+Copyright 2021 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/metrics/profile_metrics.go and modified
+// This file contains helpers for metrics that are associated to a profile.
+
+package metrics
+
+var (
+	scheduledResult     = "scheduled"
+	unschedulableResult = "unschedulable"
+	errorResult         = "error"
+)
+
+// SubscriptionScheduled can record a successful scheduling attempt and the duration
+// since `start`.
+func SubscriptionScheduled(profile string, duration float64) {
+	observeScheduleAttemptAndLatency(scheduledResult, profile, duration)
+}
+
+// SubscriptionUnschedulable can record a scheduling attempt for an unschedulable subscription
+// and the duration since `start`.
+func SubscriptionUnschedulable(profile string, duration float64) {
+	observeScheduleAttemptAndLatency(unschedulableResult, profile, duration)
+}
+
+// SubscriptionScheduleError can record a scheduling attempt that had an error and the
+// duration since `start`.
+func SubscriptionScheduleError(profile string, duration float64) {
+	observeScheduleAttemptAndLatency(errorResult, profile, duration)
+}
+
+func observeScheduleAttemptAndLatency(result, profile string, duration float64) {
+	e2eSchedulingLatency.WithLabelValues(result, profile).Observe(duration)
+	schedulingLatency.WithLabelValues(result, profile).Observe(duration)
+	scheduleAttempts.WithLabelValues(result, profile).Inc()
+}
diff --git a/pkg/scheduler/options/options.go b/pkg/scheduler/options/options.go
new file mode 100644
index 0000000..5845fc5
--- /dev/null
+++ b/pkg/scheduler/options/options.go
@@ -0,0 +1,94 @@
+/*
+Copyright 2021 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 options
+
+import (
+	"github.com/spf13/pflag"
+	"k8s.io/apimachinery/pkg/runtime"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	componentbaseconfig "k8s.io/component-base/config"
+	componentbaseoptions "k8s.io/component-base/config/options"
+	componentbaseconfigv1alpha1 "k8s.io/component-base/config/v1alpha1"
+	"k8s.io/klog/v2"
+
+	"github.com/clusternet/clusternet/pkg/known"
+)
+
+// SchedulerOptions has all the params needed to run a Scheduler
+type SchedulerOptions struct {
+	// LeaderElection defines the configuration of leader election client.
+	LeaderElection componentbaseconfig.LeaderElectionConfiguration
+
+	// ClientConnection specifies the kubeconfig file and client connection
+	// settings for the proxy server to use when communicating with the apiserver.
+	ClientConnection componentbaseconfig.ClientConnectionConfiguration
+}
+
+// NewSchedulerOptions returns a new SchedulerOptions
+func NewSchedulerOptions() (*SchedulerOptions, error) {
+	versionedClientConnection := componentbaseconfigv1alpha1.ClientConnectionConfiguration{}
+	versionedLeaderElection := componentbaseconfigv1alpha1.LeaderElectionConfiguration{
+		ResourceLock:      "lease", // Use lease-based leader election to reduce cost.
+		ResourceName:      "clusternet-scheduler",
+		ResourceNamespace: known.ClusternetSystemNamespace,
+	}
+	// Use the default ClientConnectionConfiguration and LeaderElectionConfiguration options
+	componentbaseconfigv1alpha1.RecommendedDefaultClientConnectionConfiguration(&versionedClientConnection)
+	componentbaseconfigv1alpha1.RecommendedDefaultLeaderElectionConfiguration(&versionedLeaderElection)
+
+	o := &SchedulerOptions{
+		ClientConnection: componentbaseconfig.ClientConnectionConfiguration{},
+		LeaderElection:   componentbaseconfig.LeaderElectionConfiguration{},
+	}
+
+	schedScheme := runtime.NewScheme()
+	utilruntime.Must(componentbaseconfigv1alpha1.AddToScheme(schedScheme))
+	if err := schedScheme.Convert(&versionedClientConnection, &o.ClientConnection, nil); err != nil {
+		return nil, err
+	}
+	if err := schedScheme.Convert(&versionedLeaderElection, &o.LeaderElection, nil); err != nil {
+		return nil, err
+	}
+
+	return o, nil
+}
+
+// Validate validates SchedulerOptions
+func (o *SchedulerOptions) Validate(args []string) error {
+	errors := []error{}
+	return utilerrors.NewAggregate(errors)
+}
+
+// Complete fills in fields required to have valid data
+func (o *SchedulerOptions) Complete() error {
+	// TODO
+
+	return nil
+}
+
+// AddFlags adds flags for SchedulerOptions.
+func (o *SchedulerOptions) AddFlags(fs *pflag.FlagSet) {
+	fs.StringVar(&o.ClientConnection.Kubeconfig, "kubeconfig", o.ClientConnection.Kubeconfig, "Path to a kubeconfig file pointing at the 'core' kubernetes server. Only required if out-of-cluster.")
+	fs.Float32Var(&o.ClientConnection.QPS, "kube-api-qps", o.ClientConnection.QPS, "QPS to use while talking with the 'core' kubernetes apiserver.")
+	fs.Int32Var(&o.ClientConnection.Burst, "kube-api-burst", o.ClientConnection.Burst, "Burst to use while talking with 'core' kubernetes apiserver.")
+
+	componentbaseoptions.BindLeaderElectionFlags(&o.LeaderElection, fs)
+	if err := fs.MarkHidden("leader-elect-resource-lock"); err != nil {
+		klog.Errorf("failed to set a flag to hidden: %v", err)
+	}
+}
diff --git a/pkg/scheduler/parallelize/error_channel.go b/pkg/scheduler/parallelize/error_channel.go
new file mode 100644
index 0000000..13a11d8
--- /dev/null
+++ b/pkg/scheduler/parallelize/error_channel.go
@@ -0,0 +1,63 @@
+/*
+Copyright 2019 The Kubernetes 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/internal/parallelize/error_channel.go
+
+package parallelize
+
+import (
+	"context"
+)
+
+// ErrorChannel supports non-blocking send and receive operation to capture error.
+// A maximum of one error is kept in the channel and the rest of the errors sent
+// are ignored, unless the existing error is received and the channel becomes empty
+// again.
+type ErrorChannel struct {
+	errCh chan error
+}
+
+// SendError sends an error without blocking the sender.
+func (e *ErrorChannel) SendError(err error) {
+	select {
+	case e.errCh <- err:
+	default:
+	}
+}
+
+// SendErrorWithCancel sends an error without blocking the sender and calls
+// cancel function.
+func (e *ErrorChannel) SendErrorWithCancel(err error, cancel context.CancelFunc) {
+	e.SendError(err)
+	cancel()
+}
+
+// ReceiveError receives an error from channel without blocking on the receiver.
+func (e *ErrorChannel) ReceiveError() error {
+	select {
+	case err := <-e.errCh:
+		return err
+	default:
+		return nil
+	}
+}
+
+// NewErrorChannel returns a new ErrorChannel.
+func NewErrorChannel() *ErrorChannel {
+	return &ErrorChannel{
+		errCh: make(chan error, 1),
+	}
+}
diff --git a/pkg/scheduler/parallelize/parallelism.go b/pkg/scheduler/parallelize/parallelism.go
new file mode 100644
index 0000000..c66aaf6
--- /dev/null
+++ b/pkg/scheduler/parallelize/parallelism.go
@@ -0,0 +1,58 @@
+/*
+Copyright 2020 The Kubernetes 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/internal/parallelize/parallelism.go
+
+package parallelize
+
+import (
+	"context"
+	"math"
+
+	"k8s.io/client-go/util/workqueue"
+)
+
+// DefaultParallelism is the default parallelism used in scheduler.
+const DefaultParallelism int = 16
+
+// Parallelizer holds the parallelism for scheduler.
+type Parallelizer struct {
+	parallelism int
+}
+
+// NewParallelizer returns an object holding the parallelism.
+func NewParallelizer(p int) Parallelizer {
+	return Parallelizer{parallelism: p}
+}
+
+// chunkSizeFor returns a chunk size for the given number of items to use for
+// parallel work. The size aims to produce good CPU utilization.
+// returns max(1, min(sqrt(n), n/Parallelism))
+func chunkSizeFor(n, parallelism int) int {
+	s := int(math.Sqrt(float64(n)))
+
+	if r := n/parallelism + 1; s > r {
+		s = r
+	} else if s < 1 {
+		s = 1
+	}
+	return s
+}
+
+// Until is a wrapper around workqueue.ParallelizeUntil to use in scheduling algorithms.
+func (p Parallelizer) Until(ctx context.Context, pieces int, doWorkPiece workqueue.DoWorkPieceFunc) {
+	workqueue.ParallelizeUntil(ctx, p.parallelism, pieces, doWorkPiece, workqueue.WithChunkSize(chunkSizeFor(pieces, p.parallelism)))
+}
diff --git a/pkg/scheduler/scheduler.go b/pkg/scheduler/scheduler.go
new file mode 100644
index 0000000..3434435
--- /dev/null
+++ b/pkg/scheduler/scheduler.go
@@ -0,0 +1,402 @@
+/*
+Copyright 2021 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 scheduler
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+	"k8s.io/controller-manager/pkg/clientbuilder"
+	"k8s.io/klog/v2"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	clusternet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
+	informers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions"
+	applisters "github.com/clusternet/clusternet/pkg/generated/listers/apps/v1alpha1"
+	"github.com/clusternet/clusternet/pkg/known"
+	"github.com/clusternet/clusternet/pkg/scheduler/algorithm"
+	schedulercache "github.com/clusternet/clusternet/pkg/scheduler/cache"
+	framework "github.com/clusternet/clusternet/pkg/scheduler/framework/interfaces"
+	"github.com/clusternet/clusternet/pkg/scheduler/framework/plugins"
+	frameworkruntime "github.com/clusternet/clusternet/pkg/scheduler/framework/runtime"
+	"github.com/clusternet/clusternet/pkg/scheduler/metrics"
+	"github.com/clusternet/clusternet/pkg/scheduler/options"
+	"github.com/clusternet/clusternet/pkg/utils"
+)
+
+// These are reasons for a subscription's transition to a condition.
+const (
+	// ReasonUnschedulable reason in SubscriptionScheduled SubscriptionCondition means that the scheduler
+	// can't schedule the subscription right now, for example due to insufficient resources in the clusters.
+	ReasonUnschedulable = "Unschedulable"
+
+	// SchedulerError is the reason recorded for events when an error occurs during scheduling a subscription.
+	SchedulerError = "SchedulerError"
+)
+
+// Scheduler defines configuration for clusternet scheduler
+type Scheduler struct {
+	schedulerOptions *options.SchedulerOptions
+
+	kubeClient                *kubernetes.Clientset
+	clusternetClient          *clusternet.Clientset
+	ClusternetInformerFactory informers.SharedInformerFactory
+
+	subsLister applisters.SubscriptionLister
+	subsSynced cache.InformerSynced
+
+	// default in-tree registry
+	registry frameworkruntime.Registry
+
+	scheduleAlgorithm algorithm.ScheduleAlgorithm
+
+	// SchedulingQueue holds subscriptions to be scheduled
+	SchedulingQueue workqueue.RateLimitingInterface
+
+	framework framework.Framework
+}
+
+// NewScheduler returns a new Scheduler.
+func NewScheduler(schedulerOptions *options.SchedulerOptions) (*Scheduler, error) {
+	clientConfig, err := utils.LoadsKubeConfig(schedulerOptions.ClientConnection.Kubeconfig, 1)
+	if err != nil {
+		return nil, err
+	}
+	clientConfig.QPS = schedulerOptions.ClientConnection.QPS
+	clientConfig.Burst = int(schedulerOptions.ClientConnection.Burst)
+
+	// creating the clientset
+	rootClientBuilder := clientbuilder.SimpleControllerClientBuilder{
+		ClientConfig: clientConfig,
+	}
+	kubeClient := kubernetes.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("kubeclient-clusternet-scheduler"))
+	clusternetClient := clusternet.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-client-scheduler"))
+	clusternetInformerFactory := informers.NewSharedInformerFactory(clusternetClient, known.DefaultResync)
+
+	// create event recorder
+	broadcaster := record.NewBroadcaster()
+	broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
+		Interface: kubeClient.CoreV1().Events(""),
+	})
+	utilruntime.Must(appsapi.AddToScheme(scheme.Scheme))
+	utilruntime.Must(clusterapi.AddToScheme(scheme.Scheme))
+	recorder := broadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "clusternet-scheduler"})
+
+	schedulerCache := schedulercache.New(clusternetInformerFactory.Clusters().V1beta1().ManagedClusters().Lister())
+
+	sched := &Scheduler{
+		schedulerOptions:          schedulerOptions,
+		kubeClient:                kubeClient,
+		clusternetClient:          clusternetClient,
+		ClusternetInformerFactory: clusternetInformerFactory,
+		subsLister:                clusternetInformerFactory.Apps().V1alpha1().Subscriptions().Lister(),
+		subsSynced:                clusternetInformerFactory.Apps().V1alpha1().Subscriptions().Informer().HasSynced,
+		registry:                  plugins.NewInTreeRegistry(),
+		scheduleAlgorithm:         algorithm.NewGenericScheduler(schedulerCache),
+		SchedulingQueue:           workqueue.NewRateLimitingQueue(workqueue.DefaultItemBasedRateLimiter()),
+	}
+
+	framework, err := frameworkruntime.NewFramework(sched.registry,
+		frameworkruntime.WithEventRecorder(recorder),
+	)
+	if err != nil {
+		return nil, err
+	}
+	sched.framework = framework
+
+	// register all metrics
+	metrics.Register()
+
+	addAllEventHandlers(sched)
+	return sched, nil
+}
+
+// Run begins watching and scheduling. It starts scheduling and blocked until the context is done.
+func (sched *Scheduler) Run(ctx context.Context) error {
+	defer sched.SchedulingQueue.ShutDown()
+
+	// Start all informers.
+	sched.ClusternetInformerFactory.Start(ctx.Done())
+
+	// Wait for all caches to sync before scheduling.
+	sched.ClusternetInformerFactory.WaitForCacheSync(ctx.Done())
+
+	// if leader election is disabled, so runCommand inline until done.
+	if !sched.schedulerOptions.LeaderElection.LeaderElect {
+		wait.UntilWithContext(ctx, sched.scheduleOne, 0)
+		klog.Warning("finished without leader elect")
+		return nil
+	}
+
+	// leader election is enabled, runCommand via LeaderElector until done and exit.
+	curIdentity, err := utils.GenerateIdentity()
+	if err != nil {
+		return err
+	}
+	le, err := leaderelection.NewLeaderElector(*utils.NewLeaderElectionConfigWithDefaultValue(
+		curIdentity,
+		sched.schedulerOptions.LeaderElection.ResourceName,
+		sched.schedulerOptions.LeaderElection.ResourceNamespace,
+		sched.schedulerOptions.LeaderElection.LeaseDuration.Duration,
+		sched.schedulerOptions.LeaderElection.RenewDeadline.Duration,
+		sched.schedulerOptions.LeaderElection.RetryPeriod.Duration,
+		sched.kubeClient,
+		leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				wait.UntilWithContext(ctx, sched.scheduleOne, 0)
+			},
+			OnStoppedLeading: func() {
+				klog.Error("leader election got lost")
+			},
+			OnNewLeader: func(identity string) {
+				// we're notified when new leader elected
+				if identity == curIdentity {
+					// I just got the lock
+					return
+				}
+				klog.Infof("new leader elected: %s", identity)
+			},
+		},
+	))
+	if err != nil {
+		return err
+	}
+	le.Run(ctx)
+	return nil
+}
+
+// scheduleOne does the entire scheduling workflow for a single subscription.
+// It is serialized on the scheduling algorithm's cluster fitting.
+func (sched *Scheduler) scheduleOne(ctx context.Context) {
+	key, shutdown := sched.SchedulingQueue.Get()
+	if shutdown {
+		klog.Error("failed to get next unscheduled subscription from closed queue")
+		return
+	}
+	defer sched.SchedulingQueue.Done(key)
+
+	// TODO: scheduling
+	// Convert the namespace/name string into a distinct namespace and name
+	ns, name, err := cache.SplitMetaNamespaceKey(key.(string))
+	if err != nil {
+		utilruntime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return
+	}
+
+	sub, err := sched.subsLister.Subscriptions(ns).Get(name)
+	if err != nil {
+		utilruntime.HandleError(err)
+		return
+	}
+	klog.V(3).InfoS("Attempting to schedule subscription", "subscription", klog.KObj(sub))
+
+	// 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)
+	if err != nil {
+		sched.recordSchedulingFailure(sub, err, ReasonUnschedulable)
+		return
+	}
+	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() {
+		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.recordSchedulingFailure(sub, sts.AsError(), SchedulerError)
+		return
+	}
+
+	// Run "permit" plugins.
+	runPermitStatus := sched.framework.RunPermitPlugins(schedulingCycleCtx, sub, namespacedClusters)
+	if runPermitStatus.Code() != framework.Wait && !runPermitStatus.IsSuccess() {
+		var reason string
+		if runPermitStatus.IsUnschedulable() {
+			metrics.SubscriptionUnschedulable(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
+			reason = ReasonUnschedulable
+		} else {
+			metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
+			reason = SchedulerError
+		}
+		// One of the plugins returned status different from success or wait.
+		sched.framework.RunReservePluginsUnreserve(schedulingCycleCtx, sub, namespacedClusters)
+		sched.recordSchedulingFailure(sub, runPermitStatus.AsError(), reason)
+		return
+	}
+
+	// bind the subscription to multiple clusters asynchronously (we can do this b/c of the assumption step above).
+	go func() {
+		bindingCycleCtx, cancel := context.WithCancel(ctx)
+		defer cancel()
+		metrics.SchedulerGoroutines.WithLabelValues(metrics.Binding).Inc()
+		defer metrics.SchedulerGoroutines.WithLabelValues(metrics.Binding).Dec()
+
+		waitOnPermitStatus := sched.framework.WaitOnPermit(bindingCycleCtx, sub)
+		if !waitOnPermitStatus.IsSuccess() {
+			var reason string
+			if waitOnPermitStatus.IsUnschedulable() {
+				metrics.SubscriptionUnschedulable(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
+				reason = ReasonUnschedulable
+			} else {
+				metrics.SubscriptionScheduleError(sched.framework.ProfileName(), metrics.SinceInSeconds(start))
+				reason = SchedulerError
+			}
+			// trigger un-reserve plugins to clean up state associated with the reserved subscription
+			sched.framework.RunReservePluginsUnreserve(bindingCycleCtx, sub, namespacedClusters)
+			sched.recordSchedulingFailure(sub, waitOnPermitStatus.AsError(), reason)
+			return
+		}
+
+		// Run "prebind" plugins.
+		preBindStatus := sched.framework.RunPreBindPlugins(bindingCycleCtx, sub, namespacedClusters)
+		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.recordSchedulingFailure(sub, preBindStatus.AsError(), SchedulerError)
+			return
+		}
+
+		err := sched.bind(bindingCycleCtx, sub, namespacedClusters)
+		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.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)
+		}
+	}()
+}
+
+// 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, targetClusterNamespaces []string) (err error) {
+	defer func() {
+		// finish binding
+		if err != nil {
+			klog.V(1).InfoS("Failed to bind sub", "sub", klog.KObj(sub))
+			return
+		}
+		sched.framework.EventRecorder().Eventf(
+			sub,
+			corev1.EventTypeNormal,
+			"Scheduled",
+			"Binding",
+			"Successfully assigned %s to %s", klog.KObj(sub), strings.Join(targetClusterNamespaces, ","),
+		)
+	}()
+
+	bindStatus := sched.framework.RunBindPlugins(ctx, sub, targetClusterNamespaces)
+	if bindStatus.IsSuccess() {
+		return nil
+	}
+	if bindStatus.Code() == framework.Error {
+		return bindStatus.AsError()
+	}
+	return fmt.Errorf("bind status: %s, %v", bindStatus.Code().String(), bindStatus.Message())
+}
+
+// recordSchedulingFailure records an event for the subscription that indicates the
+// subscription has failed to schedule. Also, update the subscription condition.
+func (sched *Scheduler) recordSchedulingFailure(sub *appsapi.Subscription, err error, _ string) {
+	klog.V(2).InfoS("Unable to schedule subscription; waiting", "subscription", klog.KObj(sub), "err", err)
+
+	msg := truncateMessage(err.Error())
+	sched.framework.EventRecorder().Event(sub, corev1.EventTypeWarning, "FailedScheduling", msg)
+
+	// TODO: update subscription condition
+
+	// re-added to the queue for re-processing
+	sched.SchedulingQueue.AddRateLimited(klog.KObj(sub).String())
+}
+
+// addAllEventHandlers is a helper function used in tests and in Scheduler
+// to add event handlers for various informers.
+func addAllEventHandlers(sched *Scheduler) {
+	sched.ClusternetInformerFactory.Apps().V1alpha1().Subscriptions().Informer().AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: func(obj interface{}) bool {
+			switch t := obj.(type) {
+			case *appsapi.Subscription:
+				sub := obj.(*appsapi.Subscription)
+				if sub.DeletionTimestamp != nil {
+					return false
+				}
+
+				// TODO: filter scheduler name
+				return true
+			case cache.DeletedFinalStateUnknown:
+				if _, ok := t.Obj.(*appsapi.Subscription); ok {
+					return true
+				}
+				utilruntime.HandleError(fmt.Errorf("unable to convert object %T to *Subscription in %T", obj, sched))
+				return false
+			default:
+				utilruntime.HandleError(fmt.Errorf("unable to handle object in %T: %T", sched, obj))
+				return false
+			}
+		},
+		Handler: cache.ResourceEventHandlerFuncs{
+			AddFunc: func(obj interface{}) {
+				// TODO
+				sub := obj.(*appsapi.Subscription)
+				sched.SchedulingQueue.AddRateLimited(klog.KObj(sub).String())
+			},
+			UpdateFunc: func(oldObj, newObj interface{}) {
+				// TODO
+				newSub := newObj.(*appsapi.Subscription)
+				sched.SchedulingQueue.AddRateLimited(klog.KObj(newSub).String())
+			},
+		},
+	})
+}
+
+// truncateMessage truncates a message if it hits the NoteLengthLimit.
+// copied from k8s.io/kubernetes/pkg/scheduler/scheduler.go
+func truncateMessage(message string) string {
+	max := known.NoteLengthLimit
+	if len(message) <= max {
+		return message
+	}
+	suffix := " ..."
+	return message[:max-len(suffix)] + suffix
+}
diff --git a/pkg/utils/leaderelection.go b/pkg/utils/leaderelection.go
new file mode 100644
index 0000000..7e9be77
--- /dev/null
+++ b/pkg/utils/leaderelection.go
@@ -0,0 +1,67 @@
+/*
+Copyright 2021 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 utils
+
+import (
+	"os"
+	"time"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/uuid"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/leaderelection/resourcelock"
+)
+
+func GenerateIdentity() (string, error) {
+	hostname, err := os.Hostname()
+	if err != nil {
+		return "", err
+	}
+
+	// add a uniquifier so that two processes on the same host don't accidentally both become active
+	identity := hostname + "_" + string(uuid.NewUUID())
+	return identity, nil
+}
+
+func NewLeaderElectionConfigWithDefaultValue(identity, leaseName, leaseNamespace string,
+	leaseDuration time.Duration, renewDeadline time.Duration, retryPeriod time.Duration,
+	clientset kubernetes.Interface, callbacks leaderelection.LeaderCallbacks) *leaderelection.LeaderElectionConfig {
+	return &leaderelection.LeaderElectionConfig{
+		Lock: &resourcelock.LeaseLock{
+			LeaseMeta: metav1.ObjectMeta{
+				Name:      leaseName,
+				Namespace: leaseNamespace,
+			},
+			Client: clientset.CoordinationV1(),
+			LockConfig: resourcelock.ResourceLockConfig{
+				Identity: identity,
+			},
+		},
+		// IMPORTANT: you MUST ensure that any code you have that
+		// is protected by the lease must terminate **before**
+		// you call cancel. Otherwise, you could have a background
+		// loop still running and another process could
+		// get elected before your background loop finished, violating
+		// the stated goal of the lease.
+		ReleaseOnCancel: true,
+		LeaseDuration:   leaseDuration,
+		RenewDeadline:   renewDeadline,
+		RetryPeriod:     retryPeriod,
+		Callbacks:       callbacks,
+	}
+}


Test patch:
diff --git a/pkg/scheduler/parallelize/error_channel_test.go b/pkg/scheduler/parallelize/error_channel_test.go
new file mode 100644
index 0000000..fb04868
--- /dev/null
+++ b/pkg/scheduler/parallelize/error_channel_test.go
@@ -0,0 +1,49 @@
+/*
+Copyright 2019 The Kubernetes 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/internal/parallelize/error_channel_test.go
+
+package parallelize
+
+import (
+	"context"
+	"errors"
+	"testing"
+)
+
+func TestErrorChannel(t *testing.T) {
+	errCh := NewErrorChannel()
+
+	if actualErr := errCh.ReceiveError(); actualErr != nil {
+		t.Errorf("expect nil from err channel, but got %v", actualErr)
+	}
+
+	err := errors.New("unknown error")
+	errCh.SendError(err)
+	if actualErr := errCh.ReceiveError(); actualErr != err {
+		t.Errorf("expect %v from err channel, but got %v", err, actualErr)
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	errCh.SendErrorWithCancel(err, cancel)
+	if actualErr := errCh.ReceiveError(); actualErr != err {
+		t.Errorf("expect %v from err channel, but got %v", err, actualErr)
+	}
+
+	if ctxErr := ctx.Err(); ctxErr != context.Canceled {
+		t.Errorf("expect context canceled, but got %v", ctxErr)
+	}
+}
diff --git a/pkg/scheduler/parallelize/parallelism_test.go b/pkg/scheduler/parallelize/parallelism_test.go
new file mode 100644
index 0000000..be7600c
--- /dev/null
+++ b/pkg/scheduler/parallelize/parallelism_test.go
@@ -0,0 +1,56 @@
+/*
+Copyright 2020 The Kubernetes 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.
+*/
+
+// This file was copied from k8s.io/kubernetes/pkg/scheduler/internal/parallelize/parallelism_test.go
+
+package parallelize
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestChunkSize(t *testing.T) {
+	tests := []struct {
+		input      int
+		wantOutput int
+	}{
+		{
+			input:      32,
+			wantOutput: 3,
+		},
+		{
+			input:      16,
+			wantOutput: 2,
+		},
+		{
+			input:      1,
+			wantOutput: 1,
+		},
+		{
+			input:      0,
+			wantOutput: 1,
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(fmt.Sprintf("%d", test.input), func(t *testing.T) {
+			if chunkSizeFor(test.input, DefaultParallelism) != test.wantOutput {
+				t.Errorf("Expected: %d, got: %d", test.wantOutput, chunkSizeFor(test.input, DefaultParallelism))
+			}
+		})
+	}
+}