feat: add hub election leader controller

charts/yurt-manager/crds/apps.openyurt.io_nodepools.yaml

@@ -373,6 +373,12 @@ spec:
                     LeaderNodeLabelSelector is used only when LeaderElectionStrategy is mark. leader Yurhub will be
                     elected from nodes that filtered by this label selector.
                   type: object
+                leaderReplicas:
+                  description: |-
+                    LeaderReplicas is used for specifying the number of leader replicas in the nodepool.
+                    If the field is not specified, the default value is 1.
+                  format: int32
+                  type: integer
                 poolScopeMetadata:
                   description: |-
                     PoolScopeMetadata is used for specifying resources which will be shared in the nodepool.

charts/yurt-manager/templates/yurt-manager-auto-generated.yaml

@@ -50,6 +50,12 @@ metadata:
 ---
 apiVersion: v1
 kind: ServiceAccount
+metadata:
+  name: yurt-manager-hubleader-controller
+  namespace: {{ .Release.Namespace }}
+---
+apiVersion: v1
+kind: ServiceAccount
 metadata:
   name: yurt-manager-load-balancer-set-controller
   namespace: {{ .Release.Namespace }}
@@ -471,6 +477,21 @@ rules:
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRole
+metadata:
+  name: yurt-manager-hubleader-controller
+rules:
+- apiGroups:
+  - apps.openyurt.io
+  resources:
+  - nodepool
+  - nodepool/status
+  verbs:
+  - get
+  - patch
+  - update
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
 metadata:
   name: yurt-manager-load-balancer-set-controller
 rules:
@@ -1035,6 +1056,19 @@ subjects:
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: ClusterRoleBinding
+metadata:
+  name: yurt-manager-hubleader-controller-binding
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: yurt-manager-hubleader-controller
+subjects:
+- kind: ServiceAccount
+  name: yurt-manager-hubleader-controller
+  namespace: {{ .Release.Namespace }}
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
 metadata:
   name: yurt-manager-load-balancer-set-controller-binding
 roleRef:

cmd/yurt-manager/app/options/hubleadercontroller.go

@@ -0,0 +1,69 @@
+/*
+Copyright 2025 The OpenYurt 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"
+
+	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/hubleader/config"
+)
+
+type HubLeaderControllerOptions struct {
+	*config.HubLeaderControllerConfiguration
+}
+
+func NewHubLeaderControllerOptions() *HubLeaderControllerOptions {
+	return &HubLeaderControllerOptions{
+		&config.HubLeaderControllerConfiguration{
+			ConcurrentHubLeaderWorkers: 3,
+		},
+	}
+}
+
+// AddFlags adds flags related to hubleader for yurt-manager to the specified FlagSet.
+func (h *HubLeaderControllerOptions) AddFlags(fs *pflag.FlagSet) {
+	if h == nil {
+		return
+	}
+
+	fs.Int32Var(
+		&h.ConcurrentHubLeaderWorkers,
+		"concurrent-hubleader-workers",
+		h.ConcurrentHubLeaderWorkers,
+		"The number of nodepool objects that are allowed to reconcile concurrently.",
+	)
+}
+
+// ApplyTo fills up hubleader config with options.
+func (h *HubLeaderControllerOptions) ApplyTo(cfg *config.HubLeaderControllerConfiguration) error {
+	if h == nil {
+		return nil
+	}
+
+	cfg.ConcurrentHubLeaderWorkers = h.ConcurrentHubLeaderWorkers
+
+	return nil
+}
+
+// Validate checks validation of HubLeaderControllerOptions.
+func (h *HubLeaderControllerOptions) Validate() []error {
+	if h == nil {
+		return nil
+	}
+	errs := []error{}
+	return errs
+}

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

@@ -46,6 +46,7 @@
 	GatewayDNSController          *GatewayDNSControllerOptions
 	GatewayInternalSvcController  *GatewayInternalSvcControllerOptions
 	GatewayPublicSvcController    *GatewayPublicSvcControllerOptions
+	HubLeaderController           *HubLeaderControllerOptions
 }
 
 // NewYurtManagerOptions creates a new YurtManagerOptions with a default config.
@@ -73,6 +74,7 @@
 		GatewayDNSController:          NewGatewayDNSControllerOptions(),
 		GatewayInternalSvcController:  NewGatewayInternalSvcControllerOptions(),
 		GatewayPublicSvcController:    NewGatewayPublicSvcControllerOptions(),
+		HubLeaderController:           NewHubLeaderControllerOptions(),
 	}
 
 	return &s, nil
@@ -101,6 +103,7 @@
 	y.GatewayDNSController.AddFlags(fss.FlagSet("gatewaydns controller"))
 	y.GatewayInternalSvcController.AddFlags(fss.FlagSet("gatewayinternalsvc controller"))
 	y.GatewayPublicSvcController.AddFlags(fss.FlagSet("gatewaypublicsvc controller"))
+	y.HubLeaderController.AddFlags(fss.FlagSet("hubleader controller"))
 	return fss
 }
 
@@ -128,6 +131,7 @@
 	errs = append(errs, y.GatewayDNSController.Validate()...)
 	errs = append(errs, y.GatewayInternalSvcController.Validate()...)
 	errs = append(errs, y.GatewayPublicSvcController.Validate()...)
+	errs = append(errs, y.HubLeaderController.Validate()...)
 	return utilerrors.NewAggregate(errs)
 }
 
@@ -196,11 +200,17 @@
 	if err := y.GatewayPublicSvcController.ApplyTo(&c.ComponentConfig.GatewayPublicSvcController); err != nil {
 		return err
 	}
+	if err := y.HubLeaderController.ApplyTo(&c.ComponentConfig.HubLeaderController); err != nil {
+		return err
+	}
 	return nil
 }
 
 // Config return a yurt-manager config objective
-func (y *YurtManagerOptions) Config(allControllers []string, controllerAliases map[string]string) (*config.Config, error) {
+func (y *YurtManagerOptions) Config(
+	allControllers []string,
+	controllerAliases map[string]string,
+) (*config.Config, error) {
 	if err := y.Validate(allControllers, controllerAliases); err != nil {
 		return nil, err
 	}

cmd/yurt-manager/names/controller_names.go

@@ -37,6 +37,7 @@ const (
 	NodeLifeCycleController                = "node-life-cycle-controller"
 	NodeBucketController                   = "node-bucket-controller"
 	LoadBalancerSetController              = "load-balancer-set-controller"
+	HubLeaderController                    = "hubleader-controller"
 )
 
 func YurtManagerControllerAliases() map[string]string {

pkg/apis/apps/v1alpha1/nodepool_conversion.go

@@ -46,6 +46,7 @@
 	// Set interconnectivity to false which will not use leader election strategy or reuse list/watch events
 	dst.Spec.InterConnectivity = false
 	dst.Spec.LeaderElectionStrategy = string(v1beta2.ElectionStrategyRandom)
+	dst.Spec.LeaderReplicas = 1
 
 	klog.V(4).Infof("convert from v1alpha1 to v1beta1 for nodepool %s", dst.Name)
 

pkg/apis/apps/v1beta1/nodepool_conversion.go

@@ -46,6 +46,7 @@
 	// Set interconnectivity to false which will not use leader election strategy or reuse list/watch events
 	dst.Spec.InterConnectivity = false
 	dst.Spec.LeaderElectionStrategy = string(v1beta2.ElectionStrategyRandom)
+	dst.Spec.LeaderReplicas = 1
 
 	klog.V(4).Infof("convert from v1beta to v1beta2 for nodepool %s", dst.Name)
 

pkg/apis/apps/v1beta2/default.go

@@ -23,4 +23,7 @@
 		obj.Annotations = make(map[string]string)
 	}
 
+	if obj.Spec.LeaderReplicas <= 0 {
+		obj.Spec.LeaderReplicas = 1
+	}
 }

pkg/apis/apps/v1beta2/nodepool_types.go

@@ -84,6 +84,11 @@ type NodePoolSpec struct {
 	// PoolScopeMetadata is used for specifying resources which will be shared in the nodepool.
 	// And it is supported to modify dynamically. and the default value is v1.Service and discovery.Endpointslice.
 	PoolScopeMetadata []metav1.GroupVersionKind `json:"poolScopeMetadata,omitempty"`
+
+	// LeaderReplicas is used for specifying the number of leader replicas in the nodepool.
+	// If the field is not specified, the default value is 1.
+	// + optional
+	LeaderReplicas int32 `json:"leaderReplicas,omitempty"`
 }
 
 // NodePoolStatus defines the observed state of NodePool

pkg/yurtmanager/controller/apis/config/types.go

@@ -23,6 +23,7 @@ import (
 
 	csrapproverconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/csrapprover/config"
 	daemonpodupdaterconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/daemonpodupdater/config"
+	hubleaderconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/hubleader/config"
 	loadbalancersetconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/loadbalancerset/loadbalancerset/config"
 	nodebucketconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/nodebucket/config"
 	nodepoolconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/nodepool/config"
@@ -105,6 +106,9 @@ type YurtManagerConfiguration struct {
 
 	// GatewayPublicSvcController holds configuration for GatewayPublicSvcController related features.
 	GatewayPublicSvcController gatewaypublicsvcconfig.GatewayPublicSvcControllerConfiguration
+
+	// HubLeaderController holds configuration for HubLeaderController related features.
+	HubLeaderController hubleaderconfig.HubLeaderControllerConfiguration
 }
 
 type GenericConfiguration struct {

pkg/yurtmanager/controller/hubleader/config/types.go

@@ -0,0 +1,22 @@
+/*
+Copyright 2023 The OpenYurt 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 config
+
+// HubLeaderControllerConfiguration contains elements describing HubLeaderController.
+type HubLeaderControllerConfiguration struct {
+	ConcurrentHubLeaderWorkers int32
+}