feat: make kube-api-qps configurable

fix

Author: andyzhangx

pkg/blob/azure.go

@@ -51,7 +51,7 @@ func IsAzureStackCloud(cloud *azure.Cloud) bool {
 }
 
 // getCloudProvider get Azure Cloud Provider
-func getCloudProvider(kubeconfig, nodeID, secretName, secretNamespace, userAgent string, allowEmptyCloudConfig bool) (*azure.Cloud, error) {
+func getCloudProvider(kubeconfig, nodeID, secretName, secretNamespace, userAgent string, allowEmptyCloudConfig bool, kubeAPIQPS float64, kubeAPIBurst int) (*azure.Cloud, error) {
 	var (
 		config     *azure.Config
 		kubeClient *clientset.Clientset
@@ -70,10 +70,10 @@ func getCloudProvider(kubeconfig, nodeID, secretName, secretNamespace, userAgent
 	kubeCfg, err := getKubeConfig(kubeconfig)
 	if err == nil && kubeCfg != nil {
 		// set QPS and QPS Burst as higher values
-		kubeCfg.QPS = 25
-		kubeCfg.Burst = 50
-		kubeClient, err = clientset.NewForConfig(kubeCfg)
-		if err != nil {
+		klog.V(2).Infof("set QPS(%f) and QPS Burst(%d) for driver kubeClient", float32(kubeAPIQPS), kubeAPIBurst)
+		kubeCfg.QPS = float32(kubeAPIQPS)
+		kubeCfg.Burst = kubeAPIBurst
+		if kubeClient, err = clientset.NewForConfig(kubeCfg); err != nil {
 			klog.Warningf("NewForConfig failed with error: %v", err)
 		}
 	} else {

pkg/blob/azure_test.go

@@ -169,7 +169,7 @@ users:
 			}
 			os.Setenv(DefaultAzureCredentialFileEnv, fakeCredFile)
 		}
-		cloud, err := getCloudProvider(test.kubeconfig, test.nodeID, "", "", test.userAgent, test.allowEmptyCloudConfig)
+		cloud, err := getCloudProvider(test.kubeconfig, test.nodeID, "", "", test.userAgent, test.allowEmptyCloudConfig, 25.0, 50)
 		if !reflect.DeepEqual(err, test.expectedErr) && test.expectedErr != nil && !strings.Contains(err.Error(), test.expectedErr.Error()) {
 			t.Errorf("desc: %s,\n input: %q, GetCloudProvider err: %v, expectedErr: %v", test.desc, test.kubeconfig, err, test.expectedErr)
 		}

pkg/blob/blob.go

@@ -145,6 +145,8 @@ type DriverOptions struct {
 	EnableGetVolumeStats                   bool
 	AppendTimeStampInCacheDir              bool
 	MountPermissions                       uint64
+	KubeAPIQPS                             float64
+	KubeAPIBurst                           int
 }
 
 // Driver implements all interfaces of CSI drivers
@@ -166,6 +168,8 @@ type Driver struct {
 	appendTimeStampInCacheDir              bool
 	blobfuseProxyConnTimout                int
 	mountPermissions                       uint64
+	kubeAPIQPS                             float64
+	kubeAPIBurst                           int
 	mounter                                *mount.SafeFormatAndMount
 	volLockMap                             *util.LockMap
 	// A map storing all volumes with ongoing operations so that additional operations
@@ -200,6 +204,8 @@ func NewDriver(options *DriverOptions) *Driver {
 		allowEmptyCloudConfig:                  options.AllowEmptyCloudConfig,
 		enableGetVolumeStats:                   options.EnableGetVolumeStats,
 		mountPermissions:                       options.MountPermissions,
+		kubeAPIQPS:                             options.KubeAPIQPS,
+		kubeAPIBurst:                           options.KubeAPIBurst,
 	}
 	d.Name = options.DriverName
 	d.Version = driverVersion
@@ -226,7 +232,7 @@ func (d *Driver) Run(endpoint, kubeconfig string, testBool bool) {
 
 	userAgent := GetUserAgent(d.Name, d.customUserAgent, d.userAgentSuffix)
 	klog.V(2).Infof("driver userAgent: %s", userAgent)
-	d.cloud, err = getCloudProvider(kubeconfig, d.NodeID, d.cloudConfigSecretName, d.cloudConfigSecretNamespace, userAgent, d.allowEmptyCloudConfig)
+	d.cloud, err = getCloudProvider(kubeconfig, d.NodeID, d.cloudConfigSecretName, d.cloudConfigSecretNamespace, userAgent, d.allowEmptyCloudConfig, d.kubeAPIQPS, d.kubeAPIBurst)
 	if err != nil {
 		klog.Fatalf("failed to get Azure Cloud Provider, error: %v", err)
 	}

pkg/blobplugin/main.go

@@ -55,6 +55,8 @@ var (
 	appendTimeStampInCacheDir              = flag.Bool("append-timestamp-cache-dir", false, "append timestamp into cache directory on agent node")
 	mountPermissions                       = flag.Uint64("mount-permissions", 0777, "mounted folder permissions")
 	allowInlineVolumeKeyAccessWithIdentity = flag.Bool("allow-inline-volume-key-access-with-idenitity", false, "allow accessing storage account key using cluster identity for inline volume")
+	kubeAPIQPS                             = flag.Float64("kube-api-qps", 25.0, "QPS to use while communicating with the kubernetes apiserver.")
+	kubeAPIBurst                           = flag.Int("kube-api-burst", 50, "Burst to use while communicating with the kubernetes apiserver.")
 )
 
 func main() {
@@ -91,6 +93,8 @@ func handle() {
 		AppendTimeStampInCacheDir:              *appendTimeStampInCacheDir,
 		MountPermissions:                       *mountPermissions,
 		AllowInlineVolumeKeyAccessWithIdentity: *allowInlineVolumeKeyAccessWithIdentity,
+		KubeAPIQPS:                             *kubeAPIQPS,
+		KubeAPIBurst:                           *kubeAPIBurst,
 	}
 	driver := blob.NewDriver(&driverOptions)
 	if driver == nil {