Repository: clusternet/clusternet
Pull request #638 (2023-03-23 16:36:41)

introduce new component clusternet-controller-manager
<!--  Thanks for sending a pull request!

-->

#### What type of PR is this?

#### What this PR does / why we need it:

#### Which issue(s) this PR fixes:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
-->
xref #590 

#### Special notes for your reviewer:


zxbyoyoyo: 


> I think `clusternet-hub` is fine. There is no need to rename to `clusternet-apiserver`. Also we've been using `clusternet-hub` for a long time, including the container image name.

before the refactoring ,clusternet-hub contains apiserver、controller capability, so it is ok to use `clusternet-hub`. but now apiserver, controller has been split out separately, apiserver called clusternet-hubb is not quite appropriate.

I think we need to redefine the concept, such as hub、agent represents the role of the cluster in clusternet, hub cluster h will contain apiserver, controller, scheduler components.

replacement cost we can do in some tasks to do
dixudx: > before the refactoring, clusternet-hub contains apiserver, controller capability, so it is ok to use `clusternet-hub`. but now apiserver, controller has been split out separately, apiserver called clusternet-hub is not quite appropriate.
> 
> I think we need to redefine the concept, such as hub, agent represents the role of the cluster in clusternet, hub cluster h will contain apiserver, controller, scheduler components.
> 
> replacement cost we can do in some tasks to do

Even though the controller parts were moved out from `clusternet-hub`, `clusternet-hub` is still not a purely apiserver. It also holds some proxy capabilities, peer connections, etc. We can regard `clusternet-hub` as an apiserver, but it goes beyond that.
dixudx: Regarding to the renaming of `clusternet-hub` to `clusternet-apiserver`, I'd prefer to the renaming part into a new PR where we can discuss it there.
codecov-commenter: ## [Codecov](https://codecov.io/gh/clusternet/clusternet/pull/638?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) Report
> Merging [#638](https://codecov.io/gh/clusternet/clusternet/pull/638?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (9f4bb91) into [main](https://codecov.io/gh/clusternet/clusternet/commit/e2ffa5fa54a7e81045fbef497d0dfd68a4bcc7f6?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet) (e2ffa5f) will **decrease** coverage by `0.38%`.
> The diff coverage is `n/a`.

:mega: This organization is not using Codecov’s [GitHub App Integration](https://github.com/apps/codecov). We recommend you install it so Codecov can continue to function properly for your repositories. [Learn more](https://about.codecov.io/blog/codecov-is-updating-its-github-integration/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)

```diff
@@            Coverage Diff             @@
##             main     #638      +/-   ##
==========================================
- Coverage   13.60%   13.23%   -0.38%     
==========================================
  Files          66       71       +5     
  Lines        7492     7800     +308     
==========================================
+ Hits         1019     1032      +13     
- Misses       6391     6685     +294     
- Partials       82       83       +1     
```


[see 17 files with indirect coverage changes](https://codecov.io/gh/clusternet/clusternet/pull/638/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=clusternet)


Patch:
diff --git a/.gitignore b/.gitignore
index d4908ff..0283b6d 100644
--- a/.gitignore
+++ b/.gitignore
@@ -37,3 +37,6 @@ dist
 
 # go vendor
 vendor
+
+# Test
+coverage.out
\ No newline at end of file
diff --git a/cmd/clusternet-controller-manager/app/app.go b/cmd/clusternet-controller-manager/app/app.go
new file mode 100644
index 0000000..a82bb5e
--- /dev/null
+++ b/cmd/clusternet-controller-manager/app/app.go
@@ -0,0 +1,109 @@
+/*
+Copyright 2023 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"
+	"fmt"
+	"os"
+
+	"github.com/clusternet/clusternet/cmd/clusternet-controller-manager/app/options"
+	"github.com/spf13/cobra"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	cliflag "k8s.io/component-base/cli/flag"
+	"k8s.io/component-base/cli/globalflag"
+	"k8s.io/component-base/logs"
+	logsapi "k8s.io/component-base/logs/api/v1"
+	"k8s.io/component-base/term"
+	"k8s.io/klog/v2"
+
+	_ "github.com/clusternet/clusternet/pkg/features"
+	"github.com/clusternet/clusternet/pkg/version"
+)
+
+func init() {
+	utilruntime.Must(logsapi.AddFeatureGates(utilfeature.DefaultMutableFeatureGate))
+}
+
+var (
+	// the command name
+	cmdName = "clusternet-controller-manager"
+)
+
+// NewControllerManagerCommand creates a *cobra.Command object with default parameters
+func NewControllerManagerCommand(stopCh context.Context) *cobra.Command {
+	opts, err := options.NewControllerManagerOptions()
+	if err != nil {
+		klog.Fatalf("unable to initialize command options: %v", err)
+	}
+
+	cmd := &cobra.Command{
+		Use:  cmdName,
+		Long: `Running in parent cluster,responsible for running all controllers`,
+		Args: func(cmd *cobra.Command, args []string) error {
+			for _, arg := range args {
+				if len(arg) > 0 {
+					return fmt.Errorf("%q does not take any arguments, got %q", cmd.CommandPath(), args)
+				}
+			}
+			return nil
+		},
+		Run: func(cmd *cobra.Command, args []string) {
+			if err = version.PrintAndExitIfRequested(cmdName); err != nil {
+				klog.Exit(err)
+			}
+
+			// Activate logging as soon as possible, after that
+			// show flags with the final logging configuration.
+			if err := logsapi.ValidateAndApply(opts.Logs, utilfeature.DefaultFeatureGate); err != nil {
+				fmt.Fprintf(os.Stderr, "%v\n", err)
+				os.Exit(1)
+			}
+			cliflag.PrintFlags(cmd.Flags())
+
+			if err = opts.Complete(); err != nil {
+				klog.Exit(err)
+			}
+			if err = opts.Validate(); err != nil {
+				klog.Exit(err)
+			}
+
+			cm, err2 := NewControllerManager(opts)
+			if err2 != nil {
+				klog.Exit(err2)
+			}
+			if err = cm.Run(stopCh); err != nil {
+				klog.Exit(err)
+			}
+
+			// TODO: add logic
+		},
+	}
+
+	nfs := opts.Flags(NewControllerInitializers().KnownControllerNames())
+	version.AddVersionFlag(nfs.FlagSet("global"))
+	globalflag.AddGlobalFlags(nfs.FlagSet("global"), cmd.Name(), logs.SkipLoggingConfigurationFlags())
+	fs := cmd.Flags()
+	for _, f := range nfs.FlagSets {
+		fs.AddFlagSet(f)
+	}
+	cols, _, _ := term.TerminalSize(cmd.OutOrStdout())
+	cliflag.SetUsageAndHelpFunc(cmd, *nfs, cols)
+
+	return cmd
+}
diff --git a/cmd/clusternet-controller-manager/app/context/context.go b/cmd/clusternet-controller-manager/app/context/context.go
new file mode 100644
index 0000000..1dc714c
--- /dev/null
+++ b/cmd/clusternet-controller-manager/app/context/context.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2023 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 context
+
+import (
+	"context"
+
+	"github.com/clusternet/clusternet/cmd/clusternet-controller-manager/app/options"
+	clusternet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
+	informers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions"
+	"k8s.io/apimachinery/pkg/util/sets"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/controller-manager/pkg/clientbuilder"
+	"k8s.io/klog/v2"
+	mcsclientset "sigs.k8s.io/mcs-api/pkg/client/clientset/versioned"
+	mcsinformers "sigs.k8s.io/mcs-api/pkg/client/informers/externalversions"
+)
+
+// ControllerContext defines the context object for controller.
+type ControllerContext struct {
+	Opts                      *options.ControllerManagerOptions
+	KubeConfig                *rest.Config
+	KubeClient                *kubernetes.Clientset
+	ElectionClient            *kubernetes.Clientset
+	ClusternetClient          *clusternet.Clientset
+	McsClient                 *mcsclientset.Clientset
+	ClusternetInformerFactory informers.SharedInformerFactory
+	KubeInformerFactory       kubeinformers.SharedInformerFactory
+	McsInformerFactory        mcsinformers.SharedInformerFactory
+	ClientBuilder             clientbuilder.ControllerClientBuilder
+	EventRecorder             record.EventRecorder
+
+	// InformersStarted is closed after all of the controllers have been initialized and are running.  After this point it is safe,
+	// for an individual controller to start the shared informers. Before it is closed, they should not.
+	InformersStarted chan struct{}
+}
+
+// StartControllers starts a set of controllers with a specified ControllerContext
+func (c *ControllerContext) StartControllers(ctx context.Context, initializers Initializers) error {
+	for controllerName, initFn := range initializers {
+		if !c.IsControllerEnabled(controllerName) {
+			klog.Warningf("%q is disabled", controllerName)
+			continue
+		}
+		klog.V(1).Infof("Starting %q", controllerName)
+		started, err := initFn(c, ctx)
+		if err != nil {
+			klog.Errorf("Error starting %q", controllerName)
+			return err
+		}
+		if !started {
+			klog.Warningf("Skipping %q", controllerName)
+			continue
+		}
+		klog.Infof("Started %q", controllerName)
+	}
+	return nil
+}
+
+func (c *ControllerContext) StartShardInformerFactories(ctx context.Context) {
+	c.KubeInformerFactory.Start(ctx.Done())
+	c.ClusternetInformerFactory.Start(ctx.Done())
+	c.McsInformerFactory.Start(ctx.Done())
+}
+
+// IsControllerEnabled check if a specified controller enabled or not.
+func (c *ControllerContext) IsControllerEnabled(name string) bool {
+	hasStar := false
+	for _, ctrl := range c.Opts.Controllers {
+		if ctrl == name {
+			return true
+		}
+		if ctrl == "-"+name {
+			return false
+		}
+		if ctrl == "*" {
+			hasStar = true
+		}
+	}
+	// if we get here, there was no explicit choice
+	if !hasStar {
+		// nothing on by default
+		return false
+	}
+	return true
+}
+
+// InitFunc is used to launch a particular controller.
+// Any error returned will cause the controller process to `Fatal`
+// The bool indicates whether the controller was enabled.
+type InitFunc func(controllerContext *ControllerContext, ctx context.Context) (enabled bool, err error)
+
+// Initializers is a public map of named controller groups
+type Initializers map[string]InitFunc
+
+// KnownControllerNames returns all known controller names
+func (i Initializers) KnownControllerNames() []string {
+	return sets.StringKeySet(i).List()
+}
diff --git a/cmd/clusternet-controller-manager/app/controllermanager.go b/cmd/clusternet-controller-manager/app/controllermanager.go
new file mode 100644
index 0000000..4543cd0
--- /dev/null
+++ b/cmd/clusternet-controller-manager/app/controllermanager.go
@@ -0,0 +1,269 @@
+/*
+Copyright 2023 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"
+
+	"k8s.io/apimachinery/pkg/version"
+
+	controllercontext "github.com/clusternet/clusternet/cmd/clusternet-controller-manager/app/context"
+	"github.com/clusternet/clusternet/cmd/clusternet-controller-manager/app/options"
+	"github.com/clusternet/clusternet/pkg/controllers/clusters/clusterlifecycle"
+	"github.com/clusternet/clusternet/pkg/controllers/clusters/clusterlifecycle/discovery"
+	"github.com/clusternet/clusternet/pkg/controllers/mcs"
+	"github.com/clusternet/clusternet/pkg/features"
+	clusternet "github.com/clusternet/clusternet/pkg/generated/clientset/versioned"
+	"github.com/clusternet/clusternet/pkg/hub/approver"
+	"github.com/clusternet/clusternet/pkg/hub/deployer"
+	"github.com/clusternet/clusternet/pkg/known"
+	"github.com/clusternet/clusternet/pkg/utils"
+	corev1 "k8s.io/api/core/v1"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	"k8s.io/client-go/dynamic"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	v1core "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/leaderelection"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/controller-manager/pkg/clientbuilder"
+	"k8s.io/klog/v2"
+	mcsclientset "sigs.k8s.io/mcs-api/pkg/client/clientset/versioned"
+	mcsinformers "sigs.k8s.io/mcs-api/pkg/client/informers/externalversions"
+
+	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
+	clusterapi "github.com/clusternet/clusternet/pkg/apis/clusters/v1beta1"
+	informers "github.com/clusternet/clusternet/pkg/generated/informers/externalversions"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/client-go/kubernetes/scheme"
+)
+
+// ControllerManager defines configuration for controllers
+type ControllerManager struct {
+	options           *options.ControllerManagerOptions
+	controllerContext *controllercontext.ControllerContext
+}
+
+// NewControllerManager returns a new ControllerManager.
+func NewControllerManager(opts *options.ControllerManagerOptions) (*ControllerManager, error) {
+
+	config, err := utils.LoadsKubeConfig(&opts.ClientConnection)
+	if err != nil {
+		return nil, err
+	}
+
+	// creating the clientset
+	rootClientBuilder := clientbuilder.SimpleControllerClientBuilder{
+		ClientConfig: config,
+	}
+	kubeClient := kubernetes.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-controller-manager-kube-client"))
+	clusternetClient := clusternet.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-controller-manager-client"))
+	electionClient := kubernetes.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-controller-manager-election-client"))
+	mcsClient := mcsclientset.NewForConfigOrDie(rootClientBuilder.ConfigOrDie("clusternet-controller-manager-mcs-client"))
+
+	broadcaster := record.NewBroadcaster()
+	if kubeClient != nil {
+		klog.Infof("sending events to api server")
+		broadcaster.StartRecordingToSink(&v1core.EventSinkImpl{
+			Interface: kubeClient.CoreV1().Events(""),
+		})
+	} else {
+		klog.Warningf("no api server defined - no events will be sent to API server.")
+	}
+	utilruntime.Must(appsapi.AddToScheme(scheme.Scheme))
+	utilruntime.Must(clusterapi.AddToScheme(scheme.Scheme))
+	recorder := broadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: "clusternet-controller-manager"})
+
+	cm := &ControllerManager{
+		options: opts,
+		controllerContext: &controllercontext.ControllerContext{
+			Opts:                      opts,
+			KubeConfig:                config,
+			KubeClient:                kubeClient,
+			ElectionClient:            electionClient,
+			ClusternetClient:          clusternetClient,
+			McsClient:                 mcsClient,
+			ClusternetInformerFactory: informers.NewSharedInformerFactory(clusternetClient, known.DefaultResync),
+			KubeInformerFactory:       kubeinformers.NewSharedInformerFactory(kubeClient, known.DefaultResync),
+			McsInformerFactory:        mcsinformers.NewSharedInformerFactory(mcsClient, known.DefaultResync),
+			ClientBuilder:             rootClientBuilder,
+			EventRecorder:             recorder,
+			InformersStarted:          make(chan struct{}),
+		},
+	}
+	return cm, nil
+}
+
+// Run starts a new ControllerManager use given ControllerManagerOptions
+func (cm *ControllerManager) Run(stopCh context.Context) error {
+	klog.Info("starting clusternet-controller-manager ...")
+	err := cm.options.Config()
+	if err != nil {
+		return err
+	}
+	curIdentity, err := utils.GenerateIdentity()
+	if err != nil {
+		return err
+	}
+
+	// create lease for clusternet controllers
+	controllerLease, err := leaderelection.NewLeaderElector(*utils.NewLeaderElectionConfigWithDefaultValue(
+		curIdentity,
+		cm.options.LeaderElection.ResourceName,
+		cm.options.LeaderElection.ResourceNamespace,
+		cm.options.LeaderElection.LeaseDuration.Duration,
+		cm.options.LeaderElection.RenewDeadline.Duration,
+		cm.options.LeaderElection.RetryPeriod.Duration,
+		cm.controllerContext.ElectionClient,
+		leaderelection.LeaderCallbacks{
+			OnStartedLeading: func(ctx context.Context) {
+				klog.Info("leader elected: starting main controllers.")
+				cm.run(ctx)
+			},
+			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
+	}
+
+	// No leader election, run directly
+	if !cm.options.LeaderElection.LeaderElect {
+		cm.run(stopCh)
+		<-stopCh.Done()
+		return nil
+	}
+	// leader election run
+	controllerLease.Run(stopCh)
+	<-stopCh.Done()
+	return nil
+}
+
+func (cm *ControllerManager) run(ctx context.Context) {
+	if err := cm.controllerContext.StartControllers(ctx, NewControllerInitializers()); err != nil {
+		klog.Fatalf("error starting controllers: %v", err)
+	}
+	cm.controllerContext.StartShardInformerFactories(ctx)
+	close(cm.controllerContext.InformersStarted)
+	<-ctx.Done()
+}
+
+func NewControllerInitializers() controllercontext.Initializers {
+	var initializers = make(controllercontext.Initializers)
+	initializers["crrapprover"] = startCRRApproveController
+	initializers["serviceimport"] = startServiceImportController
+	initializers["clusterlifecycle"] = startClusterLifecycleController
+	if utilfeature.DefaultFeatureGate.Enabled(features.Deployer) {
+		initializers["deployer"] = startDeployerControllers
+	}
+	initializers["clusterdiscovery"] = startClusterDiscoveryController
+	return initializers
+}
+
+func startCRRApproveController(controllerCtx *controllercontext.ControllerContext, ctx context.Context) (bool, error) {
+	socketConnectionEnabled := utilfeature.DefaultFeatureGate.Enabled(features.SocketConnection)
+	approve, err := approver.NewCRRApprover(
+		controllerCtx.KubeClient,
+		controllerCtx.ClusternetClient,
+		controllerCtx.ClusternetInformerFactory,
+		controllerCtx.KubeInformerFactory,
+		socketConnectionEnabled,
+	)
+	if err != nil {
+		return false, err
+	}
+	go approve.Run(controllerCtx.Opts.Threadiness, ctx.Done())
+	return true, nil
+}
+
+func startDeployerControllers(controllerCtx *controllercontext.ControllerContext, ctx context.Context) (bool, error) {
+	deployer, err := deployer.NewDeployer(
+		controllerCtx.KubeConfig.Host,
+		controllerCtx.Opts.LeaderElection.ResourceNamespace,
+		controllerCtx.Opts.ReservedNamespace,
+		controllerCtx.KubeClient,
+		controllerCtx.ClusternetClient,
+		controllerCtx.ClusternetInformerFactory,
+		controllerCtx.KubeInformerFactory,
+		controllerCtx.EventRecorder,
+		controllerCtx.Opts.AnonymousAuthSupported,
+	)
+	if err != nil {
+		return false, err
+	}
+	go deployer.Run(controllerCtx.Opts.Threadiness, ctx.Done())
+	return true, nil
+}
+
+func startClusterLifecycleController(controllerCtx *controllercontext.ControllerContext, ctx context.Context) (bool, error) {
+	clusterLifecycle := clusterlifecycle.NewController(
+		controllerCtx.ClusternetClient,
+		controllerCtx.ClusternetInformerFactory.Clusters().V1beta1().ManagedClusters(),
+		controllerCtx.EventRecorder,
+	)
+	go clusterLifecycle.Run(controllerCtx.Opts.Threadiness, ctx.Done())
+	return true, nil
+}
+
+func startServiceImportController(controllerCtx *controllercontext.ControllerContext, ctx context.Context) (bool, error) {
+	var ver *version.Info
+	ver, err := controllerCtx.McsClient.Discovery().ServerVersion()
+	if err != nil {
+		return false, err
+	}
+	serviceImportEnabled, err := utils.MultiClusterServiceEnabled(ver.String())
+	if err != nil {
+		return false, err
+	}
+	if serviceImportEnabled {
+		go mcs.NewServiceImportController(
+			controllerCtx.KubeClient,
+			controllerCtx.KubeInformerFactory.Discovery().V1().EndpointSlices(),
+			controllerCtx.McsClient,
+			controllerCtx.McsInformerFactory,
+		).Run(ctx)
+	}
+	return true, nil
+}
+
+func startClusterDiscoveryController(controllerCtx *controllercontext.ControllerContext, ctx context.Context) (bool, error) {
+	if len(controllerCtx.Opts.ClusterAPIKubeconfig) == 0 {
+		klog.Warning("will not discovery clusters created by cluster-api providers due to empty kubeconfig path")
+		return false, nil
+	}
+	clusterAPICfg, err := utils.LoadsKubeConfigFromFile(controllerCtx.Opts.ClusterAPIKubeconfig)
+	if err != nil {
+		return false, err
+	}
+	clusterDiscovery := discovery.NewController(
+		dynamic.NewForConfigOrDie(clusterAPICfg),
+		kubernetes.NewForConfigOrDie(clusterAPICfg),
+		known.DefaultResync,
+	)
+	go clusterDiscovery.Run(ctx)
+	return true, nil
+}
diff --git a/cmd/clusternet-controller-manager/app/options/options.go b/cmd/clusternet-controller-manager/app/options/options.go
new file mode 100644
index 0000000..d5cebed
--- /dev/null
+++ b/cmd/clusternet-controller-manager/app/options/options.go
@@ -0,0 +1,150 @@
+/*
+Copyright 2023 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 (
+	"fmt"
+	"net"
+	"strings"
+
+	"github.com/clusternet/clusternet/pkg/known"
+	"github.com/clusternet/clusternet/pkg/utils"
+	utilerrors "k8s.io/apimachinery/pkg/util/errors"
+	apiserveroptions "k8s.io/apiserver/pkg/server/options"
+	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	cliflag "k8s.io/component-base/cli/flag"
+	"k8s.io/component-base/metrics"
+	controllermanageroptions "k8s.io/controller-manager/options"
+)
+
+const (
+	// DefaultClusternetControllerManagerPort is the default port for the clusternet-controller-manager running in parent cluster.
+	// May be overridden by a flag at startup.
+	DefaultClusternetControllerManagerPort = 10660
+)
+
+// ControllerManagerOptions holds the command-line options for command
+type ControllerManagerOptions struct {
+
+	// Whether the anonymous access is allowed by the kube-apiserver,
+	// i.e. flag "--anonymous-auth=true" is set to kube-apiserver.
+	// If enabled, then the deployers in Clusternet will use anonymous when proxying requests to child clusters.
+	// If not, serviceaccount "clusternet-hub-proxy" will be used instead.
+	AnonymousAuthSupported bool
+	// default namespace to create Manifest in
+	// default to be "clusternet-reserved"
+	ReservedNamespace    string
+	ClusterAPIKubeconfig string
+
+	//Threadiness of controller workers, default to be 10
+	Threadiness int
+
+	// Controllers is the list of controllers to enable or disable
+	// '*' means "all enabled by default controllers"
+	// 'foo' means "enable 'foo'"
+	// '-foo' means "disable 'foo'"
+	// first item for a particular name wins
+	Controllers []string
+
+	*utils.ControllerOptions
+
+	SecureServing *apiserveroptions.SecureServingOptionsWithLoopback
+
+	// DebuggingOptions holds the Debugging options.
+	DebuggingOptions *controllermanageroptions.DebuggingOptions
+
+	Metrics *metrics.Options
+}
+
+// NewControllerManagerOptions creates a new *options with sane defaults
+func NewControllerManagerOptions() (*ControllerManagerOptions, error) {
+	controllerOptions, err := utils.NewControllerOptions("clusternet-hub", known.ClusternetSystemNamespace)
+	if err != nil {
+		return nil, err
+	}
+
+	opts := &ControllerManagerOptions{
+		SecureServing:          apiserveroptions.NewSecureServingOptions().WithLoopback(),
+		DebuggingOptions:       controllermanageroptions.RecommendedDebuggingOptions(),
+		ControllerOptions:      controllerOptions,
+		Metrics:                metrics.NewOptions(),
+		AnonymousAuthSupported: true,
+		ReservedNamespace:      known.ClusternetReservedNamespace,
+		Threadiness:            known.DefaultThreadiness,
+	}
+	// Set the PairName but leave certificate directory blank to generate in-memory by default
+	opts.SecureServing.ServerCert.CertDirectory = ""
+	opts.SecureServing.ServerCert.PairName = "clusternet-controller-manager"
+	opts.SecureServing.BindPort = DefaultClusternetControllerManagerPort
+	//opts.initFlags()
+	return opts, nil
+}
+
+func (o *ControllerManagerOptions) Config() error {
+	if o.SecureServing != nil {
+		if err := o.SecureServing.MaybeDefaultWithSelfSignedCerts("localhost", nil, []net.IP{net.ParseIP("127.0.0.1")}); err != nil {
+			return fmt.Errorf("error creating self-signed certificates: %v", err)
+		}
+	}
+
+	o.Metrics.Apply()
+	return nil
+}
+
+// Complete completes all the required options.
+func (o *ControllerManagerOptions) Complete() error {
+	allErrs := []error{}
+
+	// complete leader election and client connection options
+	if err := o.ControllerOptions.Complete(); err != nil {
+		allErrs = append(allErrs, err)
+	}
+	return utilerrors.NewAggregate(allErrs)
+}
+
+// Validate validates all the required options.
+func (o *ControllerManagerOptions) Validate() error {
+	allErrs := []error{}
+
+	// validate leader election and client connection options
+	if err := o.ControllerOptions.Validate(); err != nil {
+		allErrs = append(allErrs, err)
+	}
+
+	return utilerrors.NewAggregate(allErrs)
+}
+
+// Flags initializes flags by section name.
+func (o *ControllerManagerOptions) Flags(allControllers []string) *cliflag.NamedFlagSets {
+	fss := &cliflag.NamedFlagSets{}
+	o.SecureServing.AddFlags(fss.FlagSet("secure serving"))
+	o.DebuggingOptions.AddFlags(fss.FlagSet("profiling"))
+	o.Metrics.AddFlags(fss.FlagSet("metrics"))
+	// flags for leader election and client connection
+	o.ControllerOptions.AddFlagSets(fss)
+	utilfeature.DefaultMutableFeatureGate.AddFlag(fss.FlagSet("feature gate"))
+	miscfs := fss.FlagSet("misc")
+	miscfs.BoolVar(&o.AnonymousAuthSupported, "anonymous-auth-supported", o.AnonymousAuthSupported, "Whether the anonymous access is allowed by the 'core' kubernetes server")
+	miscfs.StringVar(&o.ReservedNamespace, "reserved-namespace", o.ReservedNamespace, "The default namespace to create Manifest in")
+	miscfs.IntVar(&o.Threadiness, "threadiness", o.Threadiness, "The number of threads to use for controller workers")
+	miscfs.StringVar(&o.ClusterAPIKubeconfig, "cluster-api-kubeconfig", o.ClusterAPIKubeconfig, "Path to a kubeconfig file pointing at the management cluster for cluster-api.")
+	miscfs.StringSliceVar(&o.Controllers, "controllers", o.Controllers, fmt.Sprintf(""+
+		"A list of controllers to enable. '*' enables all on-by-default controllers, 'foo' enables the controller "+
+		"named 'foo', '-foo' disables the controller named 'foo'.\nAll controllers: %s",
+		strings.Join(allControllers, ", ")))
+	return fss
+}
diff --git a/cmd/clusternet-controller-manager/controller-manager.go b/cmd/clusternet-controller-manager/controller-manager.go
new file mode 100644
index 0000000..ac52587
--- /dev/null
+++ b/cmd/clusternet-controller-manager/controller-manager.go
@@ -0,0 +1,35 @@
+/*
+Copyright 2023 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 (
+	"math/rand"
+	"os"
+	"time"
+
+	"github.com/clusternet/clusternet/cmd/clusternet-controller-manager/app"
+	"github.com/clusternet/clusternet/pkg/utils"
+	"k8s.io/component-base/cli"
+	_ "k8s.io/component-base/logs/json/register" // for JSON log format registration
+)
+
+func main() {
+	rand.Seed(time.Now().UTC().UnixNano())
+	command := app.NewControllerManagerCommand(utils.GracefulStopWithContext())
+	code := cli.Run(command)
+	os.Exit(code)
+}
diff --git a/pkg/scheduler/framework/interfaces/types.go b/pkg/scheduler/framework/interfaces/types.go
index 12dbd5d..2a10871 100644
--- a/pkg/scheduler/framework/interfaces/types.go
+++ b/pkg/scheduler/framework/interfaces/types.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"k8s.io/apimachinery/pkg/util/sets"
+	"k8s.io/klog/v2"
 
 	appsapi "github.com/clusternet/clusternet/pkg/apis/apps/v1alpha1"
 )
@@ -97,11 +98,11 @@ func (t *TargetClusters) DeepCopy() *TargetClusters {
 	return obj
 }
 
-func (t *TargetClusters) Merge(b *TargetClusters) {
-	// TODO: use matrix addition to merge
-	// 1. Align t and b using the union of clusters and feeds , no value use 0 fill
-	// 2. Addition t and b which filled
-	// 3. Remove all 0 rows and columns
+// If b.Replicas have feed more then one , use this method, it works good
+// https://github.com/clusternet/clusternet/blob/v0.14.0/pkg/scheduler/framework/interfaces/types.go#L100
+// MergeOneFeed use for merge two TargetClusters when b.Replicas only one feed
+func (t *TargetClusters) MergeOneFeed(b *TargetClusters) {
+	klog.Info("merge one feed : ", b)
 	if b == nil || len(b.BindingClusters) == 0 {
 		return
 	}
@@ -126,25 +127,36 @@ func (t *TargetClusters) Merge(b *TargetClusters) {
 		m[cluster] = i
 	}
 
-	for bi, bCluster := range b.BindingClusters {
-		ti, exist := m[bCluster]
-		if !exist {
-			// this cluster is a new one, we should append
-			t.BindingClusters = append(t.BindingClusters, bCluster)
-			for feed, tReplicas := range t.Replicas {
-				if len(tReplicas) != 0 && len(tReplicas) < len(t.BindingClusters) {
-					t.Replicas[feed] = append(t.Replicas[feed], 0)
-				}
+	// only one feed in b.Replicas, get feed and replicas by loop
+	for feed, replicas := range b.Replicas {
+		if len(replicas) == 0 {
+			if _, isok := t.Replicas[feed]; !isok {
+				t.Replicas[feed] = make([]int32, 0)
 			}
-			ti = len(t.BindingClusters) - 1
-			m[bCluster] = ti
+			return
 		}
-		for feed, bReplicas := range b.Replicas {
-			if len(bReplicas) != 0 {
-				if len(t.Replicas[feed]) == 0 {
-					t.Replicas[feed] = make([]int32, len(b.Replicas[feed]))
+		for bi, cluster := range b.BindingClusters {
+			// same cluster , use b binding cluster index get replica, and assign to t replicas
+			if ti, exist := m[cluster]; exist {
+				if len(t.Replicas[feed]) != len(t.BindingClusters) {
+					t.Replicas[feed] = make([]int32, len(t.BindingClusters))
+					t.Replicas[feed][ti] = replicas[bi]
+					continue
+				}
+				t.Replicas[feed][ti] += replicas[bi]
+			} else {
+				// new cluster, add cluster to t binding cluster
+				t.BindingClusters = append(t.BindingClusters, cluster)
+				// add 0 to exist feed but new cluster, if feed is nil, skip it
+				for f := range t.Replicas {
+					if f != feed {
+						if len(t.Replicas[f]) != 0 {
+							t.Replicas[f] = append(t.Replicas[f], 0)
+						}
+					} else {
+						t.Replicas[f] = append(t.Replicas[f], b.Replicas[feed][bi])
+					}
 				}
-				t.Replicas[feed][ti] += bReplicas[bi]
 			}
 		}
 	}
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
index 84a7980..091e055 100644
--- a/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/algorithm.go
@@ -47,7 +47,7 @@ func DynamicDivideReplicas(sub *appsapi.Subscription, deviations []appsapi.FeedO
 		if err != nil {
 			return *result, err
 		}
-		result.Merge(&r)
+		result.MergeOneFeed(&r)
 	}
 	return *result, nil
 }


Test patch:
diff --git a/pkg/scheduler/framework/interfaces/types_test.go b/pkg/scheduler/framework/interfaces/types_test.go
index 30b9feb..6f67fdf 100644
--- a/pkg/scheduler/framework/interfaces/types_test.go
+++ b/pkg/scheduler/framework/interfaces/types_test.go
@@ -38,7 +38,12 @@ func TestTargetClustersWithFirstScheduling_Merge(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t1 *testing.T) {
 			tcp := source.DeepCopy()
-			tcp.Merge(tt.b)
+			for k, v := range tt.b.Replicas {
+				tcp.MergeOneFeed(&TargetClusters{
+					BindingClusters: tt.b.BindingClusters,
+					Replicas:        map[string][]int32{k: v},
+				})
+			}
 			if !reflect.DeepEqual(tcp, tt.result) {
 				t.Errorf("Merge() %s gotResponse = %v, want %v", tt.name, tcp, tt.result)
 			}
@@ -159,10 +164,125 @@ func TestTargetClusters_Merge(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t1 *testing.T) {
 			tcp := template.DeepCopy()
-			tcp.Merge(tt.b)
+			for k, v := range tt.b.Replicas {
+				tcp.MergeOneFeed(&TargetClusters{
+					BindingClusters: tt.b.BindingClusters,
+					Replicas:        map[string][]int32{k: v},
+				})
+			}
 			if !reflect.DeepEqual(tcp, tt.result) {
 				t.Errorf("Merge() %s gotResponse = %v, want %v", tt.name, tcp, tt.result)
 			}
 		})
 	}
 }
+
+func TestTargetCLusters_MergeOneFeed(t *testing.T) {
+	template := &TargetClusters{
+		BindingClusters: []string{"c1", "c2", "c3"},
+		Replicas: map[string][]int32{
+			"f1": {1, 2, 3},
+			"f2": {},
+			"f3": {1, 0, 3},
+		},
+	}
+	tests := []struct {
+		name   string
+		b      *TargetClusters
+		result *TargetClusters
+	}{
+		{
+			name: "merge one feed",
+			b: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c2", "c3",
+				},
+				Replicas: map[string][]int32{
+					"f1": {4, 4, 5},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{"c1", "c2", "c3"},
+				Replicas: map[string][]int32{
+					"f1": {5, 6, 8},
+					"f2": {},
+					"f3": {1, 0, 3},
+				},
+			},
+		},
+		{
+			name: "merge not exist feed",
+			b: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c2", "c3",
+				},
+				Replicas: map[string][]int32{
+					"f4": {1, 1, 1},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c2", "c3",
+				},
+				Replicas: map[string][]int32{
+					"f1": {1, 2, 3},
+					"f2": {},
+					"f3": {1, 0, 3},
+					"f4": {1, 1, 1},
+				},
+			},
+		},
+		{
+			name: "merge not exist cluster",
+			b: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c4",
+				},
+				Replicas: map[string][]int32{
+					"f1": {1, 1},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c2", "c3", "c4",
+				},
+				Replicas: map[string][]int32{
+					"f1": {2, 2, 3, 1},
+					"f2": {},
+					"f3": {1, 0, 3, 0},
+				},
+			},
+		},
+		{
+			name: "merge not exist cluster and feed",
+			b: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c4",
+				},
+				Replicas: map[string][]int32{
+					"f2": {1, 4},
+				},
+			},
+			result: &TargetClusters{
+				BindingClusters: []string{
+					"c1", "c2", "c3", "c4",
+				},
+				Replicas: map[string][]int32{
+					"f1": {1, 2, 3, 0},
+					"f2": {1, 0, 0, 4},
+					"f3": {1, 0, 3, 0},
+				},
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t1 *testing.T) {
+			tcopy := template.DeepCopy()
+			tcopy.MergeOneFeed(tt.b)
+			if !reflect.DeepEqual(tcopy, tt.result) {
+				t.Errorf("MergeOneFeed() %s \ngot = %v\nwant= %v", tt.name, tcopy, tt.result)
+			}
+		})
+	}
+}
diff --git a/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go b/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go
index 165624e..06c617e 100644
--- a/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go
+++ b/pkg/scheduler/framework/plugins/defaultassigner/dynamic_test.go
@@ -280,11 +280,11 @@ func TestDynamicAssigner_Assign(t *testing.T) {
 				},
 			},
 			want: framework.TargetClusters{
-				BindingClusters: []string{"c1", "c2", "c3", "c4"},
+				BindingClusters: []string{"c1", "c2", "c3"},
 				Replicas: map[string][]int32{
-					utils.GetFeedKey(newDeploymentFeed("f1")): {1, 3, 0, 0}, // already scheduled: 4
+					utils.GetFeedKey(newDeploymentFeed("f1")): {1, 3, 0}, // already scheduled: 4
 					utils.GetFeedKey(newServiceFeed("f2")):    {},
-					utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1, 0}, // already scheduled: 3
+					utils.GetFeedKey(newDeploymentFeed("f3")): {1, 1, 1}, // already scheduled: 3
 					utils.GetFeedKey(newServiceFeed("f4")):    {},
 				},
 			},