Fix rpc retry policy not taking effect (#17673)

Signed-off-by: SimFG <bang.fu@zilliz.com>

configs/milvus.yaml

@@ -256,6 +256,10 @@ grpc:
     dialTimeout:      5000
     keepAliveTime:    10000
     keepAliveTimeout: 20000
+    maxMaxAttempts: 5
+    initialBackOff: 1.0
+    maxBackoff: 60.0
+    backoffMultiplier: 2.0
 
 # Configure the proxy tls enable.
 tls:

go.mod

@@ -56,6 +56,7 @@ require (
 	golang.org/x/exp v0.0.0-20211216164055-b2b84827b756
 	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 	google.golang.org/grpc v1.44.0
+	google.golang.org/grpc/examples v0.0.0-20220617181431-3e7b97febc7f // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0
 	stathat.com/c/consistent v1.0.0
 )
@@ -66,5 +67,4 @@ replace (
 	github.com/dgrijalva/jwt-go => github.com/golang-jwt/jwt v3.2.2+incompatible // Fix security alert for jwt-go 3.2.0
 	github.com/go-kit/kit => github.com/go-kit/kit v0.1.0
 	github.com/streamnative/pulsarctl => github.com/xiaofan-luan/pulsarctl v0.0.2
-	google.golang.org/grpc => google.golang.org/grpc v1.38.0
 )

internal/distributed/datacoord/client/client.go

@@ -60,6 +60,11 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.data.DataCoord",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 		sess: sess,
 	}

internal/distributed/datanode/client/client.go

@@ -53,6 +53,11 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.data.DataNode",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.DataNodeRole)

internal/distributed/indexcoord/client/client.go

@@ -60,6 +60,11 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.index.IndexCoord",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 		sess: sess,
 	}

internal/distributed/indexnode/client/client.go

@@ -53,6 +53,11 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.index.IndexNode",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.IndexNodeRole)

internal/distributed/proxy/client/client.go

@@ -53,6 +53,11 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.proxy.Proxy",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.ProxyRole)

internal/distributed/querycoord/client/client.go

@@ -59,6 +59,11 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.query.QueryCoord",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 		sess: sess,
 	}

internal/distributed/querynode/client/client.go

@@ -54,6 +54,11 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.query.QueryNode",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.QueryNodeRole)

internal/distributed/rootcoord/client/client.go

@@ -67,6 +67,11 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 			DialTimeout:            ClientParams.DialTimeout,
 			KeepAliveTime:          ClientParams.KeepAliveTime,
 			KeepAliveTimeout:       ClientParams.KeepAliveTimeout,
+			RetryServiceNameConfig: "milvus.proto.rootcoord.RootCoord",
+			MaxAttempts:            ClientParams.MaxAttempts,
+			InitialBackoff:         ClientParams.InitialBackoff,
+			MaxBackoff:             ClientParams.MaxBackoff,
+			BackoffMultiplier:      ClientParams.BackoffMultiplier,
 		},
 		sess: sess,
 	}

internal/util/grpcclient/client.go

@@ -22,6 +22,8 @@ import (
 	"sync"
 	"time"
 
+	"google.golang.org/grpc/backoff"
+
 	"github.com/milvus-io/milvus/internal/util"
 
 	"github.com/milvus-io/milvus/internal/util/crypto"
@@ -32,7 +34,6 @@ import (
 	"github.com/milvus-io/milvus/internal/util/trace"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/backoff"
 	"google.golang.org/grpc/keepalive"
 )
 
@@ -59,10 +60,16 @@ type ClientBase struct {
 	role                   string
 	ClientMaxSendSize      int
 	ClientMaxRecvSize      int
+	RetryServiceNameConfig string
 
 	DialTimeout      time.Duration
 	KeepAliveTime    time.Duration
 	KeepAliveTimeout time.Duration
+
+	MaxAttempts       int
+	InitialBackoff    float32
+	MaxBackoff        float32
+	BackoffMultiplier float32
 }
 
 // SetRole sets role of client
@@ -138,18 +145,17 @@ func (c *ClientBase) connect(ctx context.Context) error {
 	dialContext, cancel := context.WithTimeout(ctx, c.DialTimeout)
 
 	// refer to https://github.com/grpc/grpc-proto/blob/master/grpc/service_config/service_config.proto
-	retryPolicy := `{
+	retryPolicy := fmt.Sprintf(`{
 		"methodConfig": [{
-		  "name": [{}],
+		  "name": [{"service": "%s"}],
-		  "waitForReady": false,
 		  "retryPolicy": {
-			  "MaxAttempts": 4,
+			  "MaxAttempts": %d,
-			  "InitialBackoff": ".1s",
+			  "InitialBackoff": "%fs",
-			  "MaxBackoff": ".4s",
+			  "MaxBackoff": "%fs",
-			  "BackoffMultiplier": 1.6,
+			  "BackoffMultiplier": %f,
 			  "RetryableStatusCodes": [ "UNAVAILABLE" ]
 		  }
-		}]}`
+		}]}`, c.RetryServiceNameConfig, c.MaxAttempts, c.InitialBackoff, c.MaxBackoff, c.BackoffMultiplier)
 
 	conn, err := grpc.DialContext(
 		dialContext,

internal/util/grpcclient/client_test.go

@@ -19,10 +19,21 @@ package grpcclient
 import (
 	"context"
 	"errors"
+	"fmt"
+	"log"
+	"net"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
+	"google.golang.org/grpc/examples/helloworld/helloworld"
+	"google.golang.org/grpc/keepalive"
+
+	"github.com/milvus-io/milvus/internal/util/typeutil"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/reflection"
+
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -245,3 +256,82 @@ func TestClientBase_Recall(t *testing.T) {
 	})
 
 }
+
+type server struct {
+	helloworld.UnimplementedGreeterServer
+	reqCounter   uint
+	SuccessCount uint
+}
+
+func (s *server) SayHello(ctx context.Context, in *helloworld.HelloRequest) (*helloworld.HelloReply, error) {
+	log.Printf("Received: %s", in.Name)
+	s.reqCounter++
+	if s.reqCounter%s.SuccessCount == 0 {
+		log.Printf("success %d", s.reqCounter)
+		return &helloworld.HelloReply{Message: strings.ToUpper(in.Name)}, nil
+	}
+	return nil, status.Errorf(codes.Unavailable, "server: fail it")
+}
+
+func TestClientBase_RetryPolicy(t *testing.T) {
+	// server
+	port := ":50051"
+	address := "localhost:50051"
+	lis, err := net.Listen("tcp", port)
+	if err != nil {
+		log.Fatalf("failed to listen: %v", err)
+	}
+	var kaep = keepalive.EnforcementPolicy{
+		MinTime:             5 * time.Second,
+		PermitWithoutStream: true,
+	}
+	var kasp = keepalive.ServerParameters{
+		Time:    60 * time.Second,
+		Timeout: 60 * time.Second,
+	}
+
+	maxAttempts := 5
+	s := grpc.NewServer(
+		grpc.KeepaliveEnforcementPolicy(kaep),
+		grpc.KeepaliveParams(kasp),
+	)
+	helloworld.RegisterGreeterServer(s, &server{SuccessCount: uint(maxAttempts)})
+	reflection.Register(s)
+	go func() {
+		if err := s.Serve(lis); err != nil {
+			log.Fatalf("failed to serve: %v", err)
+		}
+	}()
+	defer s.Stop()
+
+	clientBase := ClientBase{
+		ClientMaxRecvSize:      1 * 1024 * 1024,
+		ClientMaxSendSize:      1 * 1024 * 1024,
+		DialTimeout:            60 * time.Second,
+		KeepAliveTime:          60 * time.Second,
+		KeepAliveTimeout:       60 * time.Second,
+		RetryServiceNameConfig: "helloworld.Greeter",
+		MaxAttempts:            maxAttempts,
+		InitialBackoff:         10.0,
+		MaxBackoff:             60.0,
+		BackoffMultiplier:      2.0,
+	}
+	clientBase.SetRole(typeutil.DataCoordRole)
+	clientBase.SetGetAddrFunc(func() (string, error) {
+		return address, nil
+	})
+	clientBase.SetNewGrpcClientFunc(func(cc *grpc.ClientConn) interface{} {
+		return helloworld.NewGreeterClient(cc)
+	})
+	defer clientBase.Close()
+
+	ctx := context.Background()
+	name := fmt.Sprintf("hello world %d", time.Now().Second())
+	res, err := clientBase.Call(ctx, func(client interface{}) (interface{}, error) {
+		c := client.(helloworld.GreeterClient)
+		fmt.Println("client base...")
+		return c.SayHello(ctx, &helloworld.HelloRequest{Name: name})
+	})
+	assert.Nil(t, err)
+	assert.Equal(t, res.(*helloworld.HelloReply).Message, strings.ToUpper(name))
+}

internal/util/paramtable/grpc_param.go

@@ -12,6 +12,7 @@
 package paramtable
 
 import (
+	"fmt"
 	"math"
 	"strconv"
 	"sync"
@@ -43,6 +44,12 @@ const (
 	DefaultKeepAliveTime    = 10000 * time.Millisecond
 	DefaultKeepAliveTimeout = 20000 * time.Millisecond
 
+	// Grpc retry policy
+	DefaultMaxAttempts               = 5
+	DefaultInitialBackoff    float32 = 1.0
+	DefaultMaxBackoff        float32 = 60.0
+	DefaultBackoffMultiplier float32 = 2.0
+
 	ProxyInternalPort = 19529
 	ProxyExternalPort = 19530
 )
@@ -185,6 +192,11 @@ type GrpcClientConfig struct {
 	DialTimeout      time.Duration
 	KeepAliveTime    time.Duration
 	KeepAliveTimeout time.Duration
+
+	MaxAttempts       int
+	InitialBackoff    float32
+	MaxBackoff        float32
+	BackoffMultiplier float32
 }
 
 // InitOnce initialize grpc client config once
@@ -202,116 +214,248 @@ func (p *GrpcClientConfig) init(domain string) {
 	p.initDialTimeout()
 	p.initKeepAliveTimeout()
 	p.initKeepAliveTime()
+	p.initMaxAttempts()
+	p.initInitialBackoff()
+	p.initMaxBackoff()
+	p.initBackoffMultiplier()
+}
+
+func (p *GrpcClientConfig) ParseConfig(funcDesc string, key string, backKey string, parseValue func(string) (interface{}, error), applyValue func(interface{}, error)) {
+	var err error
+
+	valueStr, err := p.Load(key)
+	if err != nil && backKey != "" {
+		valueStr, err = p.Load(backKey)
+	}
+	if err != nil {
+		log.Warn(fmt.Sprintf("Failed to load %s, set to default", key), zap.String("role", p.Domain), zap.Error(err))
+		applyValue(nil, err)
+	} else {
+		value, err := parseValue(valueStr)
+		if err != nil {
+			log.Warn(fmt.Sprintf("Failed to parse %s, set to default", key),
+				zap.String("role", p.Domain), zap.String(key, valueStr), zap.Error(err))
+			applyValue(nil, err)
+		} else {
+			applyValue(value, nil)
+		}
+	}
+
+	log.Debug(funcDesc, zap.String("role", p.Domain), zap.Int(key, p.ClientMaxSendSize))
 }
 
 func (p *GrpcClientConfig) initClientMaxSendSize() {
-	var err error
+	funcDesc := "Init client max send size"
-
+	key := "grpc.clientMaxSendSize"
-	valueStr, err := p.Load("grpc.clientMaxSendSize")
+	p.ParseConfig(funcDesc, key, fmt.Sprintf("%s.%s", p.Domain, key),
-	if err != nil {
+		func(s string) (interface{}, error) {
-		valueStr, err = p.Load(p.Domain + ".grpc.clientMaxSendSize")
+			return strconv.Atoi(s)
-	}
+		},
+		func(i interface{}, err error) {
 			if err != nil {
 				p.ClientMaxSendSize = DefaultClientMaxSendSize
-	} else {
+				return
-		value, err := strconv.Atoi(valueStr)
+			}
-		if err != nil {
+			v, ok := i.(int)
-			log.Warn("Failed to parse grpc.clientMaxSendSize, set to default",
+			if !ok {
-				zap.String("role", p.Domain), zap.String("grpc.clientMaxSendSize", valueStr),
+				log.Warn(fmt.Sprintf("Failed to convert int when parsing %s, set to default", key),
-				zap.Error(err))
+					zap.String("role", p.Domain), zap.Any(key, i))
-
 				p.ClientMaxSendSize = DefaultClientMaxSendSize
-		} else {
+				return
-			p.ClientMaxSendSize = value
 			}
-	}
+			p.ClientMaxSendSize = v
-
+		})
-	log.Debug("initClientMaxSendSize",
-		zap.String("role", p.Domain), zap.Int("grpc.clientMaxSendSize", p.ClientMaxSendSize))
 }
 
 func (p *GrpcClientConfig) initClientMaxRecvSize() {
-	var err error
+	funcDesc := "Init client max recv size"
-	valueStr, err := p.Load("grpc.clientMaxRecvSize")
+	key := "grpc.clientMaxRecvSize"
-	if err != nil {
+	p.ParseConfig(funcDesc, key, fmt.Sprintf("%s.%s", p.Domain, key),
-		valueStr, err = p.Load(p.Domain + ".grpc.clientMaxRecvSize")
+		func(s string) (interface{}, error) {
-	}
+			return strconv.Atoi(s)
+		},
+		func(i interface{}, err error) {
 			if err != nil {
 				p.ClientMaxRecvSize = DefaultClientMaxRecvSize
-	} else {
+				return
-		value, err := strconv.Atoi(valueStr)
+			}
-		if err != nil {
+			v, ok := i.(int)
-			log.Warn("Failed to parse grpc.clientMaxRecvSize, set to default",
+			if !ok {
-				zap.String("role", p.Domain), zap.String("grpc.clientMaxRecvSize", valueStr),
+				log.Warn(fmt.Sprintf("Failed to convert int when parsing %s, set to default", key),
-				zap.Error(err))
+					zap.String("role", p.Domain), zap.Any(key, i))
-
 				p.ClientMaxRecvSize = DefaultClientMaxRecvSize
-		} else {
+				return
-			p.ClientMaxRecvSize = value
 			}
-	}
+			p.ClientMaxRecvSize = v
-
+		})
-	log.Debug("initClientMaxRecvSize",
-		zap.String("role", p.Domain), zap.Int("grpc.clientMaxRecvSize", p.ClientMaxRecvSize))
 }
 
 func (p *GrpcClientConfig) initDialTimeout() {
-	var err error
+	funcDesc := "Init dial timeout"
-	valueStr, err := p.Load("grpc.client.dialTimeout")
+	key := "grpc.client.dialTimeout"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.Atoi(s)
+		},
+		func(i interface{}, err error) {
 			if err != nil {
 				p.DialTimeout = DefaultDialTimeout
-	} else {
+				return
-		value, err := strconv.Atoi(valueStr)
+			}
-		if err != nil {
+			v, ok := i.(int)
-			log.Warn("Failed to parse grpc.client.dialTimeout, set to default",
+			if !ok {
-				zap.String("role", p.Domain), zap.String("grpc.client.dialTimeout", valueStr),
+				log.Warn(fmt.Sprintf("Failed to convert int when parsing %s, set to default", key),
-				zap.Error(err))
+					zap.String("role", p.Domain), zap.Any(key, i))
 				p.DialTimeout = DefaultDialTimeout
-		} else {
+				return
-			p.DialTimeout = time.Duration(value) * time.Millisecond
 			}
-	}
+			p.DialTimeout = time.Duration(v) * time.Millisecond
-	log.Debug("Init dial timeout",
+		})
-		zap.String("role", p.Domain), zap.Duration("grpc.log.dialTimeout", p.DialTimeout))
 }
 
 func (p *GrpcClientConfig) initKeepAliveTime() {
-	var err error
+	funcDesc := "Init keep alive time"
-	valueStr, err := p.Load("grpc.client.keepAliveTime")
+	key := "grpc.client.keepAliveTime"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.Atoi(s)
+		},
+		func(i interface{}, err error) {
 			if err != nil {
 				p.KeepAliveTime = DefaultKeepAliveTime
-	} else {
+				return
-		value, err := strconv.Atoi(valueStr)
+			}
-		if err != nil {
+			v, ok := i.(int)
-			log.Warn("Failed to parse grpc.client.keepAliveTime, set to default",
+			if !ok {
-				zap.String("role", p.Domain), zap.String("grpc.client.keepAliveTime", valueStr),
+				log.Warn(fmt.Sprintf("Failed to convert int when parsing %s, set to default", key),
-				zap.Error(err))
+					zap.String("role", p.Domain), zap.Any(key, i))
-
 				p.KeepAliveTime = DefaultKeepAliveTime
-		} else {
+				return
-			p.KeepAliveTime = time.Duration(value) * time.Millisecond
 			}
-	}
+			p.KeepAliveTime = time.Duration(v) * time.Millisecond
-	log.Debug("Init keep alive time",
+		})
-		zap.String("role", p.Domain), zap.Duration("grpc.log.keepAliveTime", p.KeepAliveTime))
 }
 
 func (p *GrpcClientConfig) initKeepAliveTimeout() {
-	var err error
+	funcDesc := "Init keep alive timeout"
-	valueStr, err := p.Load("grpc.client.keepAliveTimeout")
+	key := "grpc.client.keepAliveTimeout"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.Atoi(s)
+		},
+		func(i interface{}, err error) {
 			if err != nil {
 				p.KeepAliveTimeout = DefaultKeepAliveTimeout
-	} else {
+				return
-		value, err := strconv.Atoi(valueStr)
+			}
-		if err != nil {
+			v, ok := i.(int)
-			log.Warn("Failed to parse grpc.client.keepAliveTimeout, set to default",
+			if !ok {
-				zap.String("role", p.Domain), zap.String("grpc.client.keepAliveTimeout", valueStr),
+				log.Warn(fmt.Sprintf("Failed to convert int when parsing %s, set to default", key),
-				zap.Error(err))
+					zap.String("role", p.Domain), zap.Any(key, i))
 				p.KeepAliveTimeout = DefaultKeepAliveTimeout
-		} else {
+				return
-			p.KeepAliveTimeout = time.Duration(value) * time.Millisecond
 			}
+			p.KeepAliveTimeout = time.Duration(v) * time.Millisecond
+		})
 }
-	log.Debug("Init keep alive timeout",
+
-		zap.String("role", p.Domain), zap.Duration("grpc.log.keepAliveTimeout", p.KeepAliveTimeout))
+func (p *GrpcClientConfig) initMaxAttempts() {
+	funcDesc := "Init max attempts"
+	key := "grpc.client.maxMaxAttempts"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.Atoi(s)
+		},
+		func(i interface{}, err error) {
+			if err != nil {
+				p.MaxAttempts = DefaultMaxAttempts
+				return
+			}
+			// This field is required and must be greater than 1.
+			// Any value greater than 5 will be treated as if it were 5.
+			// See: https://github.com/grpc/grpc-proto/blob/master/grpc/service_config/service_config.proto#L138
+			v, ok := i.(int)
+			if !ok {
+				log.Warn(fmt.Sprintf("Failed to convert int when parsing %s, set to default", key),
+					zap.String("role", p.Domain), zap.Any(key, i))
+				p.MaxAttempts = DefaultMaxAttempts
+				return
+			}
+			if v < 2 || v > 5 {
+				log.Warn(fmt.Sprintf("The value of %s should be greater than 1 and less than 6, set to default", key),
+					zap.String("role", p.Domain), zap.Any(key, i))
+				p.MaxAttempts = DefaultMaxAttempts
+				return
+			}
+			p.MaxAttempts = v
+		})
+}
+
+func (p *GrpcClientConfig) initInitialBackoff() {
+	funcDesc := "Init initial back off"
+	key := "grpc.client.initialBackOff"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.ParseFloat(s, 32)
+		},
+		func(i interface{}, err error) {
+			if err != nil {
+				p.InitialBackoff = DefaultInitialBackoff
+				return
+			}
+			v, ok := i.(float64)
+			if !ok {
+				log.Warn(fmt.Sprintf("Failed to convert float64 when parsing %s, set to default", key),
+					zap.String("role", p.Domain), zap.Any(key, i))
+				p.InitialBackoff = DefaultInitialBackoff
+				return
+			}
+			p.InitialBackoff = float32(v)
+		})
+}
+
+func (p *GrpcClientConfig) initMaxBackoff() {
+	funcDesc := "Init max back off"
+	key := "grpc.client.maxBackoff"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.ParseFloat(s, 32)
+		},
+		func(i interface{}, err error) {
+			if err != nil {
+				p.MaxBackoff = DefaultMaxBackoff
+				return
+			}
+			v, ok := i.(float64)
+			if !ok {
+				log.Warn(fmt.Sprintf("Failed to convert float64 when parsing %s, set to default", key),
+					zap.String("role", p.Domain), zap.Any(key, i))
+				p.MaxBackoff = DefaultMaxBackoff
+				return
+			}
+			p.MaxBackoff = float32(v)
+		})
+}
+
+func (p *GrpcClientConfig) initBackoffMultiplier() {
+	funcDesc := "Init back off multiplier"
+	key := "grpc.client.backoffMultiplier"
+	p.ParseConfig(funcDesc, key, "",
+		func(s string) (interface{}, error) {
+			return strconv.ParseFloat(s, 32)
+		},
+		func(i interface{}, err error) {
+			if err != nil {
+				p.BackoffMultiplier = DefaultBackoffMultiplier
+				return
+			}
+			v, ok := i.(float64)
+			if !ok {
+				log.Warn(fmt.Sprintf("Failed to convert float64 when parsing %s, set to default", key),
+					zap.String("role", p.Domain), zap.Any(key, i))
+				p.BackoffMultiplier = DefaultBackoffMultiplier
+				return
+			}
+			p.BackoffMultiplier = float32(v)
+		})
 }

internal/util/paramtable/grpc_param_test.go

@@ -117,6 +117,48 @@ func TestGrpcClientParams(t *testing.T) {
 	Params.initKeepAliveTimeout()
 	assert.Equal(t, Params.KeepAliveTimeout, 500*time.Millisecond)
 
+	Params.initMaxAttempts()
+	assert.Equal(t, Params.MaxAttempts, DefaultMaxAttempts)
+	Params.Save("grpc.client.maxMaxAttempts", "a")
+	Params.initMaxAttempts()
+	assert.Equal(t, Params.MaxAttempts, DefaultMaxAttempts)
+	Params.Save("grpc.client.maxMaxAttempts", "1")
+	Params.initMaxAttempts()
+	assert.Equal(t, Params.MaxAttempts, DefaultMaxAttempts)
+	Params.Save("grpc.client.maxMaxAttempts", "10")
+	Params.initMaxAttempts()
+	assert.Equal(t, Params.MaxAttempts, DefaultMaxAttempts)
+	Params.Save("grpc.client.maxMaxAttempts", "4")
+	Params.initMaxAttempts()
+	assert.Equal(t, Params.MaxAttempts, 4)
+
+	Params.initInitialBackoff()
+	assert.Equal(t, Params.InitialBackoff, DefaultInitialBackoff)
+	Params.Save("grpc.client.initialBackOff", "a")
+	Params.initInitialBackoff()
+	assert.Equal(t, Params.InitialBackoff, DefaultInitialBackoff)
+	Params.Save("grpc.client.initialBackOff", "2.0")
+	Params.initInitialBackoff()
+	assert.Equal(t, Params.InitialBackoff, float32(2.0))
+
+	Params.initMaxBackoff()
+	assert.Equal(t, Params.MaxBackoff, DefaultMaxBackoff)
+	Params.Save("grpc.client.maxBackOff", "a")
+	Params.initMaxBackoff()
+	assert.Equal(t, Params.MaxBackoff, DefaultMaxBackoff)
+	Params.Save("grpc.client.maxBackOff", "50.0")
+	Params.initMaxBackoff()
+	assert.Equal(t, Params.MaxBackoff, float32(50.0))
+
+	Params.initBackoffMultiplier()
+	assert.Equal(t, Params.BackoffMultiplier, DefaultBackoffMultiplier)
+	Params.Save("grpc.client.backoffMultiplier", "a")
+	Params.initBackoffMultiplier()
+	assert.Equal(t, Params.BackoffMultiplier, DefaultBackoffMultiplier)
+	Params.Save("grpc.client.backoffMultiplier", "3.0")
+	Params.initBackoffMultiplier()
+	assert.Equal(t, Params.BackoffMultiplier, float32(3.0))
+
 	Params.Save("common.security.tlsMode", "1")
 	Params.Save("tls.serverPemPath", "/pem")
 	Params.Save("tls.serverKeyPath", "/key")