Improve configuration of concurrent endpoints workers (issue #1910)

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

@@ -0,0 +1,63 @@
+/*
+Copyright 2024 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/servicetopology/endpoints/config"
+)
+
+type EndPointsControllerOptions struct {
+	*config.ServiceTopologyEndPointsControllerConfiguration
+}
+
+func NewEndPointsControllerOptions() *EndPointsControllerOptions {
+	return &EndPointsControllerOptions{
+		&config.ServiceTopologyEndPointsControllerConfiguration{
+			ConcurrentEndPointsWorkers: 3,
+		},
+	}
+}
+
+// AddFlags adds flags related to servicetopology endpoints for yurt-manager to the specified FlagSet.
+func (n *EndPointsControllerOptions) AddFlags(fs *pflag.FlagSet) {
+	if n == nil {
+		return
+	}
+
+	fs.Int32Var(&n.ConcurrentEndPointsWorkers, "servicetopology-endpoints-workers", n.ConcurrentEndPointsWorkers, "Max concurrent workers for Servicetopology-endpoints controller.")
+}
+
+// ApplyTo fils up servicetopolgy endpoints config with options.
+func (o *EndPointsControllerOptions) ApplyTo(cfg *config.ServiceTopologyEndPointsControllerConfiguration) error {
+	if o == nil {
+		return nil
+	}
+
+	cfg.ConcurrentEndPointsWorkers = o.ConcurrentEndPointsWorkers
+	return nil
+}
+
+// Validate checks validation of EndPointsControllerOptions.
+func (o *EndPointsControllerOptions) Validate() []error {
+	if o == nil {
+		return nil
+	}
+	errs := []error{}
+	return errs
+}

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

@@ -39,6 +39,7 @@ type YurtManagerOptions struct {
 	YurtAppOverriderController *YurtAppOverriderControllerOptions
 	NodeLifeCycleController    *NodeLifecycleControllerOptions
 	NodeBucketController       *NodeBucketControllerOptions
+	EndPointsController        *EndPointsControllerOptions
 	LoadBalancerSetController  *LoadBalancerSetControllerOptions
 }
 
@@ -60,6 +61,7 @@ func NewYurtManagerOptions() (*YurtManagerOptions, error) {
 		YurtAppOverriderController: NewYurtAppOverriderControllerOptions(),
 		NodeLifeCycleController:    NewNodeLifecycleControllerOptions(),
 		NodeBucketController:       NewNodeBucketControllerOptions(),
+		EndPointsController:        NewEndPointsControllerOptions(),
 		LoadBalancerSetController:  NewLoadBalancerSetControllerOptions(),
 	}
 
@@ -81,6 +83,7 @@ func (y *YurtManagerOptions) Flags(allControllers, disabledByDefaultControllers
 	y.YurtAppOverriderController.AddFlags(fss.FlagSet("yurtappoverrider controller"))
 	y.NodeLifeCycleController.AddFlags(fss.FlagSet("nodelifecycle controller"))
 	y.NodeBucketController.AddFlags(fss.FlagSet("nodebucket controller"))
+	y.EndPointsController.AddFlags(fss.FlagSet("endpoints controller"))
 	y.LoadBalancerSetController.AddFlags(fss.FlagSet("loadbalancerset controller"))
 	return fss
 }
@@ -101,6 +104,7 @@ func (y *YurtManagerOptions) Validate(allControllers []string, controllerAliases
 	errs = append(errs, y.YurtAppOverriderController.Validate()...)
 	errs = append(errs, y.NodeLifeCycleController.Validate()...)
 	errs = append(errs, y.NodeBucketController.Validate()...)
+	errs = append(errs, y.EndPointsController.Validate()...)
 	errs = append(errs, y.LoadBalancerSetController.Validate()...)
 	return utilerrors.NewAggregate(errs)
 }
@@ -146,6 +150,9 @@ func (y *YurtManagerOptions) ApplyTo(c *config.Config, controllerAliases map[str
 	if err := y.NodeBucketController.ApplyTo(&c.ComponentConfig.NodeBucketController); err != nil {
 		return err
 	}
+	if err := y.EndPointsController.ApplyTo(&c.ComponentConfig.ServiceTopologyEndpointsController); err != nil {
+		return err
+	}
 	if err := y.LoadBalancerSetController.ApplyTo(&c.ComponentConfig.LoadBalancerSetController); err != nil {
 		return err
 	}

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

@@ -28,6 +28,7 @@
 	nodepoolconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/nodepool/config"
 	platformadminconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/platformadmin/config"
 	gatewaypickupconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/raven/gatewaypickup/config"
+	endpointsconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/servicetopology/endpoints/config"
 	yurtappdaemonconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/yurtappdaemon/config"
 	yurtappoverriderconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/yurtappoverrider/config"
 	yurtappsetconfig "github.com/openyurtio/openyurt/pkg/yurtmanager/controller/yurtappset/config"
@@ -79,8 +80,15 @@
 	//  NodeBucketController holds configuration for NodeBucketController related features.
 	NodeBucketController nodebucketconfig.NodeBucketControllerConfiguration
 
+	//  PoolServiceController holds configuration for PoolServiceController related features.
+	PoolServiceController poolserviceconfig.PoolServiceControllerConfiguration
+
+	// EndPointsController holds configuration for EndPointsController related features.
+	ServiceTopologyEndpointsController endpointsconfig.ServiceTopologyEndPointsControllerConfiguration
+
 	//  LoadBalancerSetController holds configuration for LoadBalancerSetController related features.
 	LoadBalancerSetController loadbalancersetconfig.LoadBalancerSetControllerConfiguration
+
 }
 
 type GenericConfiguration struct {

pkg/yurtmanager/controller/servicetopology/endpoints/config/types.go

@@ -0,0 +1,19 @@
+/*
+Copyright 2024 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
+
+// EndPointsControllerConfiguration contains elements describing EndPOintsController
+type ServiceTopologyEndPointsControllerConfiguration struct {
+	ConcurrentEndPointsWorkers int32
+}

pkg/yurtmanager/controller/servicetopology/endpoints/endpoints_controller.go

@@ -18,7 +18,6 @@ package endpoints
 
 import (
 	"context"
-	"flag"
 	"fmt"
 
 	corev1 "k8s.io/api/core/v1"
@@ -35,13 +34,8 @@ import (
 	"github.com/openyurtio/openyurt/pkg/yurtmanager/controller/servicetopology/adapter"
 )
 
-func init() {
-	flag.IntVar(&concurrentReconciles, "servicetopology-endpoints-workers", concurrentReconciles, "Max concurrent workers for Servicetopology-endpoints controller.")
-}
-
 var (
-	concurrentReconciles = 3
-	controllerKind       = corev1.SchemeGroupVersion.WithKind("Endpoints")
+	controllerKind = corev1.SchemeGroupVersion.WithKind("Endpoints")
 )
 
 func Format(format string, args ...interface{}) string {
@@ -53,7 +47,7 @@ func Format(format string, args ...interface{}) string {
 // and Start it when the Manager is Started.
 func Add(ctx context.Context, c *appconfig.CompletedConfig, mgr manager.Manager) error {
 	klog.Infof("servicetopology-endpoints-controller add controller %s", controllerKind.String())
-	return add(mgr, newReconciler(c, mgr))
+	return add(mgr, c, newReconciler(c, mgr))
 }
 
 var _ reconcile.Reconciler = &ReconcileServicetopologyEndpoints{}
@@ -76,9 +70,9 @@ func (r *ReconcileServicetopologyEndpoints) InjectClient(c client.Client) error
 }
 
 // add adds a new Controller to mgr with r as the reconcile.Reconciler
-func add(mgr manager.Manager, r reconcile.Reconciler) error {
+func add(mgr manager.Manager, cfg *appconfig.CompletedConfig, r reconcile.Reconciler) error {
 	// Create a new controller
-	c, err := controller.New(names.ServiceTopologyEndpointsController, mgr, controller.Options{Reconciler: r, MaxConcurrentReconciles: concurrentReconciles})
+	c, err := controller.New(names.ServiceTopologyEndpointsController, mgr, controller.Options{Reconciler: r, MaxConcurrentReconciles: int(cfg.ComponentConfig.ServiceTopologyEndpointsController.ConcurrentEndPointsWorkers)})
 	if err != nil {
 		return err
 	}