Merge pull request #609 from jxustc/main

add sharding feature for ClusterSynchroManager

Author: Iceber

cmd/binding-apiserver/app/binding_apiserver.go

@@ -57,7 +57,7 @@ func NewClusterPediaServerCommand(ctx context.Context) *cobra.Command {
 				return err
 			}
 
-			synchromanager := synchromanager.NewManager(crdclient, config.StorageFactory, clustersynchro.ClusterSyncConfig{})
+			synchromanager := synchromanager.NewManager(crdclient, config.StorageFactory, clustersynchro.ClusterSyncConfig{}, "")
 			go synchromanager.Run(1, ctx.Done())
 
 			server, err := completedConfig.New()

cmd/clustersynchro-manager/app/config/config.go

@@ -18,6 +18,7 @@ type Config struct {
 	EventRecorder record.EventRecorder
 
 	WorkerNumber            int
+	ShardingName            string
 	MetricsServerConfig     metrics.Config
 	KubeMetricsServerConfig *kubestatemetrics.ServerConfig
 	StorageFactory          storage.StorageFactory

cmd/clustersynchro-manager/app/options/options.go

@@ -47,6 +47,7 @@ type Options struct {
 
 	WorkerNumber            int // WorkerNumber is the number of worker goroutines
 	PageSizeForResourceSync int64
+	ShardingName            string
 }
 
 func NewClusterSynchroManagerOptions() (*Options, error) {
@@ -90,6 +91,7 @@ func (o *Options) Flags() cliflag.NamedFlagSets {
 	genericfs.Float32Var(&o.ClientConnection.QPS, "kube-api-qps", o.ClientConnection.QPS, "QPS to use while talking with kubernetes apiserver.")
 	genericfs.Int32Var(&o.ClientConnection.Burst, "kube-api-burst", o.ClientConnection.Burst, "Burst to use while talking with kubernetes apiserver.")
 	genericfs.IntVar(&o.WorkerNumber, "worker-number", o.WorkerNumber, "The number of worker goroutines.")
+	genericfs.StringVar(&o.ShardingName, "sharding-name", o.ShardingName, "The sharding name of manager.")
 
 	syncfs := fss.FlagSet("resource sync")
 	syncfs.Int64Var(&o.PageSizeForResourceSync, "page-size", o.PageSizeForResourceSync, "The requested chunk size of initial and resync watch lists for resource sync")
@@ -160,13 +162,18 @@ func (o *Options) Config() (*config.Config, error) {
 	}
 	kubeStateMetricsServerConfig := o.KubeStateMetrics.ServerConfig(metricsConfig)
 
+	if o.ShardingName != "" {
+		o.LeaderElection.ResourceName = fmt.Sprintf("%s-%s", o.LeaderElection.ResourceName, o.ShardingName)
+	}
+
 	return &config.Config{
 		CRDClient:     crdclient,
 		Kubeconfig:    kubeconfig,
 		EventRecorder: eventRecorder,
 
 		StorageFactory: storagefactory,
 		WorkerNumber:   o.WorkerNumber,
+		ShardingName:   o.ShardingName,
 
 		MetricsServerConfig:     metricsConfig,
 		KubeMetricsServerConfig: kubeStateMetricsServerConfig,

cmd/clustersynchro-manager/app/synchro.go

@@ -85,7 +85,7 @@ func NewClusterSynchroManagerCommand(ctx context.Context) *cobra.Command {
 }
 
 func Run(ctx context.Context, c *config.Config) error {
-	synchromanager := synchromanager.NewManager(c.CRDClient, c.StorageFactory, c.ClusterSyncConfig)
+	synchromanager := synchromanager.NewManager(c.CRDClient, c.StorageFactory, c.ClusterSyncConfig, c.ShardingName)
 
 	go func() {
 		metrics.RunServer(c.MetricsServerConfig)

kustomize/crds/cluster.clusterpedia.io_pediaclusters.yaml

@@ -36,6 +36,10 @@ spec:
       name: ClusterHealthy
       priority: 10
       type: string
+    - jsonPath: .status.shardingName
+      name: ShardingName
+      priority: 10
+      type: string
     name: v1alpha2
     schema:
       openAPIV3Schema:
@@ -68,6 +72,8 @@ spec:
               kubeconfig:
                 format: byte
                 type: string
+              shardingName:
+                type: string
               syncAllCustomResources:
                 type: boolean
               syncResources:
@@ -169,6 +175,8 @@ spec:
                   - type
                   type: object
                 type: array
+              shardingName:
+                type: string
               syncResources:
                 items:
                   properties:

pkg/generated/openapi/zz_generated.openapi.go

@@ -48,6 +48,7 @@ type Manager struct {
 	clusterpediaclient crdclientset.Interface
 	informerFactory    externalversions.SharedInformerFactory
 
+	shardingName               string
 	queue                      workqueue.RateLimitingInterface
 	storage                    storage.StorageFactory
 	clusterlister              clusterlister.PediaClusterLister
@@ -62,14 +63,15 @@ type Manager struct {
 
 var _ kubestatemetrics.ClusterMetricsWriterListGetter = &Manager{}
 
-func NewManager(client crdclientset.Interface, storage storage.StorageFactory, syncConfig clustersynchro.ClusterSyncConfig) *Manager {
+func NewManager(client crdclientset.Interface, storage storage.StorageFactory, syncConfig clustersynchro.ClusterSyncConfig, shardingName string) *Manager {
 	factory := externalversions.NewSharedInformerFactory(client, 0)
 	clusterinformer := factory.Cluster().V1alpha2().PediaClusters()
 	clusterSyncResourcesInformer := factory.Cluster().V1alpha2().ClusterSyncResources()
 
 	manager := &Manager{
 		informerFactory:    factory,
 		clusterpediaclient: client,
+		shardingName:       shardingName,
 
 		storage:                    storage,
 		clusterlister:              clusterinformer.Lister(),
@@ -166,7 +168,9 @@ func (manager *Manager) addCluster(obj interface{}) {
 func (manager *Manager) updateCluster(older, newer interface{}) {
 	oldObj := older.(*clusterv1alpha2.PediaCluster)
 	newObj := newer.(*clusterv1alpha2.PediaCluster)
-	if newObj.DeletionTimestamp.IsZero() && equality.Semantic.DeepEqual(oldObj.Spec, newObj.Spec) {
+	if newObj.DeletionTimestamp.IsZero() &&
+		equality.Semantic.DeepEqual(oldObj.Spec, newObj.Spec) &&
+		oldObj.Status.ShardingName == newObj.Status.ShardingName {
 		return
 	}
 
@@ -183,6 +187,17 @@ func (manager *Manager) enqueue(obj interface{}) {
 		return
 	}
 
+	if cluster, ok := obj.(*clusterv1alpha2.PediaCluster); ok {
+		currentSharding := cluster.Status.ShardingName
+		if cluster.Spec.ShardingName != manager.shardingName {
+			if currentSharding == nil || *currentSharding != manager.shardingName {
+				return
+			}
+		} else if currentSharding != nil && *currentSharding != manager.shardingName {
+			return
+		}
+	}
+
 	manager.queue.Add(key)
 }
 
@@ -256,6 +271,17 @@ func (manager *Manager) processNextCluster() (continued bool) {
 
 // if err returned is not nil, cluster will be requeued
 func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster) controller.Result {
+	if cluster.Status.ShardingName == nil && cluster.Spec.ShardingName != manager.shardingName {
+		return controller.NoRequeueResult
+	}
+
+	if cluster.Status.ShardingName != nil && *cluster.Status.ShardingName != manager.shardingName {
+		return controller.NoRequeueResult
+	}
+	// After the above filtering， The cluster will be in the following state：
+	// 1. spec.sharding == manager.shardingName and status.sharding == nil
+	// 2. spec.sharding == manager.shardingName and status != nil and status.sharding == manager.shardingName
+	// 3. spec.sharding != manager.shardingName and status != nil and status.sharding == manager.shardingName
 	if !cluster.DeletionTimestamp.IsZero() {
 		klog.InfoS("remove cluster", "cluster", cluster.Name)
 		if err := manager.removeCluster(cluster.Name); err != nil {
@@ -286,6 +312,20 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 		}
 	}
 
+	if cluster.Spec.ShardingName != manager.shardingName {
+		// status.sharding == manager.shardingName
+		manager.stopClusterSynchro(cluster.Name)
+
+		if err := manager.UpdateClusterShardingStatus(context.TODO(), cluster.Name, nil); err != nil {
+			klog.ErrorS(err, "Failed to remove cluster shardingName status", "cluster", cluster.Name)
+			return controller.RequeueResult(defaultRetryNum)
+		}
+
+		return controller.NoRequeueResult
+	}
+
+	cluster.Status.ShardingName = &manager.shardingName
+
 	manager.synchrolock.RLock()
 	synchro := manager.synchros[cluster.Name]
 	manager.synchrolock.RUnlock()
@@ -387,6 +427,11 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 			return controller.NoRequeueResult
 		}
 
+		if err := manager.UpdateClusterShardingStatus(context.TODO(), cluster.Name, &manager.shardingName); err != nil {
+			klog.ErrorS(err, "Failed to update cluster shardingName status", "cluster", cluster.Name)
+			return controller.RequeueResult(defaultRetryNum)
+		}
+
 		manager.synchroWaitGroup.StartWithChannel(manager.stopCh, synchro.Run)
 
 		manager.synchrolock.Lock()
@@ -398,6 +443,17 @@ func (manager *Manager) reconcileCluster(cluster *clusterv1alpha2.PediaCluster)
 	return controller.NoRequeueResult
 }
 
+func (manager *Manager) stopClusterSynchro(name string) {
+	manager.synchrolock.Lock()
+	synchro := manager.synchros[name]
+	delete(manager.synchros, name)
+	manager.synchrolock.Unlock()
+
+	if synchro != nil {
+		synchro.Shutdown(true)
+	}
+}
+
 func (manager *Manager) removeCluster(name string) error {
 	manager.synchrolock.Lock()
 	synchro := manager.synchros[name]
@@ -532,6 +588,12 @@ func (manager *Manager) updateClusterStatus(ctx context.Context, name string, up
 	})
 }
 
+func (manager *Manager) UpdateClusterShardingStatus(ctx context.Context, name string, shardingName *string) error {
+	return manager.updateClusterStatus(ctx, name, func(clusterStatus *clusterv1alpha2.ClusterStatus) {
+		clusterStatus.ShardingName = shardingName
+	})
+}
+
 func buildClusterConfig(cluster *clusterv1alpha2.PediaCluster) (*rest.Config, error) {
 	if len(cluster.Spec.Kubeconfig) != 0 {
 		clientconfig, err := clientcmd.NewClientConfigFromBytes(cluster.Spec.Kubeconfig)

pkg/synchromanager/clustersynchro_manager.go

@@ -55,6 +55,7 @@ const (
 // +kubebuilder:printcolumn:name="Validated",type=string,JSONPath=".status.conditions[?(@.type == 'Validated')].reason",priority=10
 // +kubebuilder:printcolumn:name="SynchroRunning",type=string,JSONPath=".status.conditions[?(@.type == 'SynchroRunning')].reason",priority=10
 // +kubebuilder:printcolumn:name="ClusterHealthy",type=string,JSONPath=".status.conditions[?(@.type == 'ClusterHealthy')].reason",priority=10
+// +kubebuilder:printcolumn:name="ShardingName",type=string,JSONPath=".status.shardingName",priority=10
 type PediaCluster struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
@@ -93,6 +94,9 @@ type ClusterSpec struct {
 
 	// +optional
 	SyncResourcesRefName string `json:"syncResourcesRefName,omitempty"`
+
+	// +optional
+	ShardingName string `json:"shardingName,omitempty"`
 }
 
 type ClusterGroupResources struct {
@@ -119,6 +123,9 @@ type ClusterStatus struct {
 
 	// +optional
 	SyncResources []ClusterGroupResourcesStatus `json:"syncResources,omitempty"`
+
+	// +optional
+	ShardingName *string `json:"shardingName,omitempty"`
 }
 
 type ClusterGroupResourcesStatus struct {