chore: support to configurate the qps and burst for client config

cmd/dataprotection/main.go

@@ -24,6 +24,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"runtime"
 	"strings"
 
 	// Import all Kubernetes client auth plugins (e.g. Azure, GCP, OIDC, etc.)
@@ -33,7 +34,7 @@ import (
 	snapshotv1 "github.com/kubernetes-csi/external-snapshotter/client/v6/apis/volumesnapshot/v1"
 	"github.com/spf13/pflag"
 	corev1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/runtime"
+	k8sruntime "k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	discoverycli "k8s.io/client-go/discovery"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
@@ -71,7 +72,7 @@ const (
 )
 
 var (
-	scheme   = runtime.NewScheme()
+	scheme   = k8sruntime.NewScheme()
 	setupLog = ctrl.Log.WithName("setup")
 )
 
@@ -99,6 +100,7 @@ func init() {
 	viper.SetDefault(constant.CfgKeyCtrlrMgrNS, "default")
 	viper.SetDefault(constant.KubernetesClusterDomainEnv, constant.DefaultDNSDomain)
 	viper.SetDefault(dptypes.CfgKeyGCFrequencySeconds, dptypes.DefaultGCFrequencySeconds)
+	viper.SetDefault(dptypes.CfgDataProtectionReconcileWorkers, runtime.NumCPU())
 }
 
 func main() {
@@ -165,7 +167,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+	mgr, err := ctrl.NewManager(intctrlutil.GeKubeRestConfig(), ctrl.Options{
 		Scheme:                 scheme,
 		MetricsBindAddress:     metricsAddr,
 		Port:                   9443,

cmd/manager/main.go

@@ -248,7 +248,7 @@ func main() {
 		os.Exit(1)
 	}
 
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+	mgr, err := ctrl.NewManager(intctrlutil.GeKubeRestConfig(), ctrl.Options{
 		Scheme:                 scheme,
 		MetricsBindAddress:     metricsAddr,
 		Port:                   9443,

controllers/dataprotection/backup_controller.go

@@ -126,7 +126,7 @@ func (r *BackupReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	b := ctrl.NewControllerManagedBy(mgr).
 		For(&dpv1alpha1.Backup{}).
 		WithOptions(controller.Options{
-			MaxConcurrentReconciles: viper.GetInt(maxConcurDataProtectionReconKey),
+			MaxConcurrentReconciles: viper.GetInt(dptypes.CfgDataProtectionReconcileWorkers),
 		}).
 		Owns(&batchv1.Job{}).
 		Watches(&batchv1.Job{}, handler.EnqueueRequestsFromMapFunc(r.parseBackupJob))

controllers/dataprotection/types.go

@@ -20,22 +20,16 @@ along with this program.  If not, see <http://www.gnu.org/licenses/>.
 package dataprotection
 
 import (
-	"runtime"
 	"time"
 
 	corev1 "k8s.io/api/core/v1"
-
-	viper "github.com/apecloud/kubeblocks/pkg/viperx"
 )
 
 const (
 	trueVal = "true"
 )
 
 const (
-	// settings keys
-	maxConcurDataProtectionReconKey = "MAXCONCURRENTRECONCILES_DATAPROTECTION"
-
 	// label keys
 	dataProtectionBackupRepoKey          = "dataprotection.kubeblocks.io/backup-repo-name"
 	dataProtectionWaitRepoPreparationKey = "dataprotection.kubeblocks.io/wait-repo-preparation"
@@ -100,7 +94,3 @@ const (
 )
 
 var reconcileInterval = time.Second
-
-func init() {
-	viper.SetDefault(maxConcurDataProtectionReconKey, runtime.NumCPU()*2)
-}

controllers/workloads/replicatedstatemachine_controller.go

@@ -30,6 +30,7 @@ import (
 	"k8s.io/client-go/tools/record"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 
 	workloads "github.com/apecloud/kubeblocks/apis/workloads/v1alpha1"
@@ -154,6 +155,9 @@ func (r *ReplicatedStateMachineReconciler) SetupWithManager(mgr ctrl.Manager) er
 
 		return ctrl.NewControllerManagedBy(mgr).
 			For(&workloads.ReplicatedStateMachine{}).
+			WithOptions(controller.Options{
+				MaxConcurrentReconciles: viper.GetInt(constant.CfgKBReconcileWorkers),
+			}).
 			Watches(&appsv1.StatefulSet{}, stsHandler).
 			Watches(&batchv1.Job{}, jobHandler).
 			Watches(&corev1.Pod{}, podHandler).
@@ -165,6 +169,9 @@ func (r *ReplicatedStateMachineReconciler) SetupWithManager(mgr ctrl.Manager) er
 	podHandler := handler.NewBuilder(ctx).AddFinder(stsOwnerFinder).AddFinder(rsmOwnerFinder).Build()
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&workloads.ReplicatedStateMachine{}).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: viper.GetInt(constant.CfgKBReconcileWorkers),
+		}).
 		Owns(&appsv1.StatefulSet{}).
 		Owns(&batchv1.Job{}).
 		Watches(&corev1.Pod{}, podHandler).

deploy/helm/templates/dataprotection.yaml

@@ -88,6 +88,18 @@ spec:
             - name: ENABLE_WEBHOOKS
               value: "true"
             {{- end }}
+            {{- if .Values.dataProtection.reconcileWorkers }}
+            - name: DATAPROTECTION_RECONCILE_WORKERS
+              value: {{ .Values.dataProtection.reconcileWorkers | quote }}
+            {{- end }}
+            {{- if .Values.client.qps }}
+            - name: CLIENT_QPS
+              value: {{ .Values.client.qps | quote }}
+            {{- end }}
+            {{- if .Values.client.burst }}
+            - name: CLIENT_BURST
+              value: {{ .Values.client.burst | quote }}
+            {{- end }}
             - name: DP_ENCRYPTION_KEY
               valueFrom:
                 secretKeyRef:

deploy/helm/templates/deployment.yaml

@@ -73,6 +73,14 @@ spec:
             - name: KUBEBLOCKS_RECONCILE_WORKERS
               value: {{ .Values.reconcileWorkers | quote }}
             {{- end }}
+            {{- if .Values.client.qps }}
+            - name: CLIENT_QPS
+              value: {{ .Values.client.qps | quote }}
+            {{- end }}
+            {{- if .Values.client.burst }}
+            - name: CLIENT_BURST
+              value: {{ .Values.client.burst | quote }}
+            {{- end }}
             {{- with .Values.nodeSelector }}
             - name: CM_NODE_SELECTOR
               value: {{ toJson . | quote }}

deploy/helm/values.yaml

@@ -31,6 +31,13 @@ replicaCount: 1
 ##
 reconcileWorkers: ""
 
+## k8s client configuration.
+client:
+  # default is 20
+  qps: ""
+  # default is 30
+  burst: ""
+
 ## @param nameOverride
 ##
 nameOverride: ""
@@ -311,7 +318,8 @@ dataProtection:
   # if 'get/list' role of the backup CR are compromised.
   encryptionKey: ""
   gcFrequencySeconds: 3600
-
+  ## MaxConcurrentReconciles for backup controller.
+  reconcileWorkers: ""
   image:
     # if the value of dataProtection.image.registry is not specified using `--set`, it will be set to the value of 'image.registry' by default
     registry: ""

pkg/constant/const.go

@@ -51,6 +51,8 @@ const (
 	CfgKeyDPEncryptionKey = "DP_ENCRYPTION_KEY"
 
 	CfgKBReconcileWorkers = "KUBEBLOCKS_RECONCILE_WORKERS"
+	CfgClientQPS          = "CLIENT_QPS"
+	CfgClientBurst        = "CLIENT_BURST"
 )
 
 const (

pkg/controllerutil/util.go

@@ -26,6 +26,8 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/rest"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
 
@@ -119,3 +121,16 @@ func SetOwnerReference(owner, object metav1.Object) error {
 func SetControllerReference(owner, object metav1.Object) error {
 	return controllerutil.SetControllerReference(owner, object, innerScheme)
 }
+
+func GeKubeRestConfig() *rest.Config {
+	cfg := ctrl.GetConfigOrDie()
+	clientQPS := viper.GetInt(constant.CfgClientQPS)
+	if clientQPS != 0 {
+		cfg.QPS = float32(clientQPS)
+	}
+	clientBurst := viper.GetInt(constant.CfgClientBurst)
+	if clientBurst != 0 {
+		cfg.Burst = clientBurst
+	}
+	return cfg
+}

pkg/dataprotection/types/constant.go

@@ -50,6 +50,11 @@ const (
 	ReconfigureRefAnnotationKey = "dataprotection.kubeblocks.io/reconfigure-ref"
 )
 
+const (
+	// CfgDataProtectionReconcileWorkers the max reconcile workers for MaxConcurrentReconciles
+	CfgDataProtectionReconcileWorkers = "DATAPROTECTION_RECONCILE_WORKERS"
+)
+
 // label keys
 const (
 	// ClusterUIDLabelKey specifies the cluster UID label key.

pkg/viperx/viperx.go

@@ -51,6 +51,10 @@ func GetInt32(key string) int32 {
 	return rCall(key, viper.GetInt32)
 }
 
+func GetFloat64(key string) float64 {
+	return rCall(key, viper.GetFloat64)
+}
+
 func GetString(key string) string {
 	return rCall(key, viper.GetString)
 }