backoff when err in reconcile mount pod (#720)

Signed-off-by: zwwhdls <[email protected]>

pkg/config/config.go

@@ -41,7 +41,7 @@ var (
 	MountLabels        = ""
 	HostIp             = ""
 	KubeletPort        = ""
-	ReconcileTimeout   = 1 * time.Minute
+	ReconcileTimeout   = 5 * time.Minute
 	ReconcilerInterval = 5
 
 	CSIPod = corev1.Pod{}

pkg/controller/reconciler.go

@@ -18,10 +18,12 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"strconv"
 	"time"
 
 	"golang.org/x/sync/errgroup"
+	"k8s.io/client-go/util/flowcontrol"
 	"k8s.io/klog"
 	k8sexec "k8s.io/utils/exec"
 	"k8s.io/utils/mount"
@@ -30,6 +32,11 @@ import (
 	"github.com/juicedata/juicefs-csi-driver/pkg/k8sclient"
 )
 
+const (
+	retryPeriod    = 5 * time.Second
+	maxRetryPeriod = 300 * time.Second
+)
+
 type PodReconciler struct {
 	mount.SafeFormatAndMount
 	*k8sclient.K8sClient
@@ -64,6 +71,7 @@ func StartReconciler() error {
 }
 
 func doReconcile(ks *k8sclient.K8sClient, kc *k8sclient.KubeletClient) {
+	backOff := flowcontrol.NewBackOff(retryPeriod, maxRetryPeriod)
 	for {
 		ctx := context.TODO()
 		timeoutCtx, cancel := context.WithTimeout(context.Background(), config.ReconcileTimeout)
@@ -89,6 +97,8 @@ func doReconcile(ks *k8sclient.K8sClient, kc *k8sclient.KubeletClient) {
 			if value, ok := pod.Labels[config.PodTypeKey]; !ok || value != config.PodTypeValue {
 				continue
 			}
+			backOffID := fmt.Sprintf("mountpod/%s", pod.Name)
+
 			g.Go(func() error {
 				mounter := mount.SafeFormatAndMount{
 					Interface: mount.New(""),
@@ -104,14 +114,20 @@ func doReconcile(ks *k8sclient.K8sClient, kc *k8sclient.KubeletClient) {
 					klog.Infof("goroutine of pod %s cancel", pod.Name)
 					return nil
 				default:
-					err := podDriver.Run(ctx, pod)
-					if err != nil {
-						klog.Errorf("Driver check pod %s error: %v", pod.Name, err)
+					if !backOff.IsInBackOffSinceUpdate(backOffID, backOff.Clock.Now()) {
+						err = podDriver.Run(ctx, pod)
+						if err != nil {
+							klog.Errorf("Driver check pod %s error, will retry: %v", pod.Name, err)
+							backOff.Next(backOffID, time.Now())
+							return err
+						}
+						backOff.Reset(backOffID)
 					}
-					return err
 				}
+				return nil
 			})
 		}
+		backOff.GC()
 		g.Wait()
 	finish:
 		cancel()

pkg/k8sclient/client.go

@@ -22,6 +22,7 @@ import (
 	"io"
 	"net/url"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
@@ -84,6 +85,22 @@ func NewClient() (*K8sClient, error) {
 		return nil, status.Error(codes.NotFound, "Can't get kube InClusterConfig")
 	}
 	config.Timeout = timeout
+
+	if os.Getenv("KUBE_QPS") != "" {
+		kubeQpsInt, err := strconv.Atoi(os.Getenv("KUBE_QPS"))
+		if err != nil {
+			return nil, err
+		}
+		config.QPS = float32(kubeQpsInt)
+	}
+	if os.Getenv("KUBE_BURST") != "" {
+		kubeBurstInt, err := strconv.Atoi(os.Getenv("KUBE_BURST"))
+		if err != nil {
+			return nil, err
+		}
+		config.Burst = kubeBurstInt
+	}
+
 	client, err := kubernetes.NewForConfig(config)
 	if err != nil {
 		return nil, err