Merge pull request #781 from andyzhangx/increase-blob-api-qps

fix: increase api qps for blob kubeclient

Author: andyzhangx

pkg/blob/azure.go

@@ -30,7 +30,7 @@ import (
 
 	"github.com/Azure/go-autorest/autorest"
 
-	"k8s.io/client-go/kubernetes"
+	clientset "k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/klog/v2"
@@ -52,6 +52,12 @@ func IsAzureStackCloud(cloud *azure.Cloud) bool {
 
 // getCloudProvider get Azure Cloud Provider
 func getCloudProvider(kubeconfig, nodeID, secretName, secretNamespace, userAgent string, allowEmptyCloudConfig bool) (*azure.Cloud, error) {
+	var (
+		config     *azure.Config
+		kubeClient *clientset.Clientset
+		fromSecret bool
+	)
+
 	az := &azure.Cloud{
 		InitSecretConfig: azure.InitSecretConfig{
 			SecretName:      secretName,
@@ -61,19 +67,22 @@ func getCloudProvider(kubeconfig, nodeID, secretName, secretNamespace, userAgent
 	}
 	az.Environment.StorageEndpointSuffix = storage.DefaultBaseURL
 
-	kubeClient, err := getKubeClient(kubeconfig)
-	if err != nil {
+	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.Warningf("NewForConfig failed with error: %v", err)
+		}
+	} else {
 		klog.Warningf("get kubeconfig(%s) failed with error: %v", kubeconfig, err)
 		if !os.IsNotExist(err) && !errors.Is(err, rest.ErrNotInCluster) {
 			return az, fmt.Errorf("failed to get KubeClient: %w", err)
 		}
 	}
 
-	var (
-		config     *azure.Config
-		fromSecret bool
-	)
-
 	if kubeClient != nil {
 		klog.V(2).Infof("reading cloud config from secret %s/%s", az.SecretNamespace, az.SecretName)
 		az.KubeClient = kubeClient
@@ -251,11 +260,7 @@ func (d *Driver) updateSubnetServiceEndpoints(ctx context.Context, vnetResourceG
 	return nil
 }
 
-func getKubeClient(kubeconfig string) (*kubernetes.Clientset, error) {
-	var (
-		config *rest.Config
-		err    error
-	)
+func getKubeConfig(kubeconfig string) (config *rest.Config, err error) {
 	if kubeconfig != "" {
 		if config, err = clientcmd.BuildConfigFromFlags("", kubeconfig); err != nil {
 			return nil, err
@@ -265,6 +270,5 @@ func getKubeClient(kubeconfig string) (*kubernetes.Clientset, error) {
 			return nil, err
 		}
 	}
-
-	return kubernetes.NewForConfig(config)
+	return config, err
 }

pkg/blob/azure_test.go

@@ -375,3 +375,86 @@ func TestUpdateSubnetServiceEndpoints(t *testing.T) {
 		t.Run(tc.name, tc.testFunc)
 	}
 }
+
+func TestGetKubeConfig(t *testing.T) {
+	emptyKubeConfig := "empty-Kube-Config"
+	validKubeConfig := "valid-Kube-Config"
+	fakeContent := `
+apiVersion: v1
+clusters:
+- cluster:
+    server: https://localhost:8080
+  name: foo-cluster
+contexts:
+- context:
+    cluster: foo-cluster
+    user: foo-user
+    namespace: bar
+  name: foo-context
+current-context: foo-context
+kind: Config
+users:
+- name: foo-user
+  user:
+    exec:
+      apiVersion: client.authentication.k8s.io/v1beta1
+      args:
+      - arg-1
+      - arg-2
+      command: foo-command
+`
+	err := createTestFile(emptyKubeConfig)
+	if err != nil {
+		t.Error(err)
+	}
+	defer func() {
+		if err := os.Remove(emptyKubeConfig); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	err = createTestFile(validKubeConfig)
+	if err != nil {
+		t.Error(err)
+	}
+	defer func() {
+		if err := os.Remove(validKubeConfig); err != nil {
+			t.Error(err)
+		}
+	}()
+
+	if err := ioutil.WriteFile(validKubeConfig, []byte(fakeContent), 0666); err != nil {
+		t.Error(err)
+	}
+
+	tests := []struct {
+		desc                     string
+		kubeconfig               string
+		expectError              bool
+		envVariableHasConfig     bool
+		envVariableConfigIsValid bool
+	}{
+		{
+			desc:                     "[success] valid kube config passed",
+			kubeconfig:               validKubeConfig,
+			expectError:              false,
+			envVariableHasConfig:     false,
+			envVariableConfigIsValid: false,
+		},
+		{
+			desc:                     "[failure] invalid kube config passed",
+			kubeconfig:               emptyKubeConfig,
+			expectError:              true,
+			envVariableHasConfig:     false,
+			envVariableConfigIsValid: false,
+		},
+	}
+
+	for _, test := range tests {
+		_, err := getKubeConfig(test.kubeconfig)
+		receiveError := (err != nil)
+		if test.expectError != receiveError {
+			t.Errorf("desc: %s,\n input: %q, GetCloudProvider err: %v, expectErr: %v", test.desc, test.kubeconfig, err, test.expectError)
+		}
+	}
+}