Support config grpc log level (#16618)

Signed-off-by: xiaofan-luan <xiaofan.luan@zilliz.com>
2022-05-05 10:35:50 +08:00 · 2022-05-05 10:35:50 +08:00 · 333b439a6c
parent fa97f86cda
commit 333b439a6c
17 changed files with 282 additions and 112 deletions
--- a/configs/milvus.yaml
+++ b/configs/milvus.yaml
@ -79,12 +79,6 @@ rootCoord:
  address: localhost
  port: 53100

-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32, Maximum data size received by the server
-    serverMaxSendSize: 2147483647 # math.MaxInt32, Maximum data size sent by the server
-    clientMaxRecvSize: 104857600 # 100 MB, Maximum data size received by the client
-    clientMaxSendSize: 104857600 # 100 MB, Maximum data size sent by the client
-
  dmlChannelNum: 256 # The number of dml channels created at system startup
  maxPartitionNum: 4096 # Maximum number of partitions in a collection
  minSegmentSizeToEnableIndex: 1024 # It's a threshold. When the segment size is less than this value, the segment will not be indexed
@ -116,12 +110,6 @@ proxy:
    readTimeout: 30000 # 30000 ms http read timeout
    writeTimeout: 30000 # 30000 ms http write timeout

-  grpc:
-    serverMaxRecvSize: 536870912 # 512 MB, 512 * 1024 * 1024 Bytes
-    serverMaxSendSize: 536870912 # 512 MB, 512 * 1024 * 1024 Bytes
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
-
  timeTickInterval: 200 # ms, the interval that proxy synchronize the time tick
  msgStream:
    timeTick:
@ -145,24 +133,12 @@ queryCoord:
  balanceIntervalSeconds: 60
  memoryUsageMaxDifferencePercentage: 30

-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32
-    serverMaxSendSize: 2147483647 # math.MaxInt32
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
-
 # Related configuration of queryNode, used to run hybrid search between vector and scalar data.
 queryNode:
  cacheSize: 32 # GB, default 32 GB, `cacheSize` is the memory used for caching data for faster query. The `cacheSize` must be less than system memory size.
  gracefulTime: 0 # Minimum time before the newly inserted data can be searched (in ms)
  port: 21123

-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32
-    serverMaxSendSize: 2147483647 # math.MaxInt32
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
-
  stats:
    publishInterval: 1000 # Interval for querynode to report node information (milliseconds)
  dataSync:
@ -190,30 +166,12 @@ indexCoord:
  address: localhost
  port: 31000

-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32
-    serverMaxSendSize: 2147483647 # math.MaxInt32
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
-
 indexNode:
  port: 21121

-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32
-    serverMaxSendSize: 2147483647 # math.MaxInt32
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
-
 dataCoord:
  address: localhost
  port: 13333
-
-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32
-    serverMaxSendSize: 2147483647 # math.MaxInt32
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
  enableCompaction: true # Enable data segment compression
  enableGarbageCollection: false

@ -234,12 +192,6 @@ dataCoord:
 dataNode:
  port: 21124

-  grpc:
-    serverMaxRecvSize: 2147483647 # math.MaxInt32
-    serverMaxSendSize: 2147483647 # math.MaxInt32
-    clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
-    clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
-
  dataSync:
    flowGraph:
      maxQueueLength: 1024 # Maximum length of task queue in flowgraph
@ -259,6 +211,20 @@ log:
    maxBackups: 20
  format: text # text/json

+grpc:
+  log:
+    level: WARNING
+
+  serverMaxRecvSize: 2147483647 # math.MaxInt32
+  serverMaxSendSize: 2147483647 # math.MaxInt32
+  clientMaxRecvSize: 104857600 # 100 MB, 100 * 1024 * 1024
+  clientMaxSendSize: 104857600 # 100 MB, 100 * 1024 * 1024
+
+  client:
+    dialTimeout:      5000
+    keepAliveTime:    10000
+    keepAliveTimeout: 3000
+
 common:
  # Channel name generation rule: ${namePrefix}-${ChannelIdx}
  chanNamePrefix:
--- a/internal/distributed/datacoord/client/client.go
+++ b/internal/distributed/datacoord/client/client.go
@ -57,6 +57,9 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 		sess: sess,
 	}
--- a/internal/distributed/datanode/client/client.go
+++ b/internal/distributed/datanode/client/client.go
@ -50,6 +50,9 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.DataNodeRole)
--- a/internal/distributed/indexcoord/client/client.go
+++ b/internal/distributed/indexcoord/client/client.go
@ -57,6 +57,9 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 		sess: sess,
 	}
--- a/internal/distributed/indexnode/client/client.go
+++ b/internal/distributed/indexnode/client/client.go
@ -50,6 +50,9 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.IndexNodeRole)
--- a/internal/distributed/proxy/client/client.go
+++ b/internal/distributed/proxy/client/client.go
@ -50,6 +50,9 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.ProxyRole)
--- a/internal/distributed/querycoord/client/client.go
+++ b/internal/distributed/querycoord/client/client.go
@ -56,6 +56,9 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 		sess: sess,
 	}
--- a/internal/distributed/querynode/client/client.go
+++ b/internal/distributed/querynode/client/client.go
@ -51,6 +51,9 @@ func NewClient(ctx context.Context, addr string) (*Client, error) {
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 	}
 	client.grpcClient.SetRole(typeutil.QueryNodeRole)
--- a/internal/distributed/rootcoord/client/client.go
+++ b/internal/distributed/rootcoord/client/client.go
@ -62,6 +62,9 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
 		grpcClient: &grpcclient.ClientBase{
 			ClientMaxRecvSize: ClientParams.ClientMaxRecvSize,
 			ClientMaxSendSize: ClientParams.ClientMaxSendSize,
+			DialTimeout:       ClientParams.DialTimeout,
+			KeepAliveTime:     ClientParams.KeepAliveTime,
+			KeepAliveTimeout:  ClientParams.KeepAliveTimeout,
 		},
 		sess: sess,
 	}
--- a/internal/log/config.go
+++ b/internal/log/config.go
@ -42,6 +42,8 @@ type FileLogConfig struct {
 type Config struct {
 	// Log level.
 	Level string `toml:"level" json:"level"`
+	// grpc log level
+	GrpcLevel string `toml:"grpc-level" json:"grpc-level"`
 	// Log format. one of json, text, or console.
 	Format string `toml:"format" json:"format"`
 	// Disable automatic timestamps in output.
--- a/internal/util/grpcclient/client.go
+++ b/internal/util/grpcclient/client.go
@ -36,12 +36,6 @@ import (
 	"google.golang.org/grpc/keepalive"
 )

-const (
-	dialTimeout      = 5 * time.Second
-	keepAliveTime    = 10 * time.Second
-	keepAliveTimeout = 3 * time.Second
-)
-
 // GrpcClient abstracts client of grpc
 type GrpcClient interface {
 	SetRole(string)
@ -65,6 +59,10 @@ type ClientBase struct {
 	role              string
 	ClientMaxSendSize int
 	ClientMaxRecvSize int
+
+	DialTimeout      time.Duration
+	KeepAliveTime    time.Duration
+	KeepAliveTimeout time.Duration
 }

 // SetRole sets role of client
@ -137,7 +135,7 @@ func (c *ClientBase) connect(ctx context.Context) error {
 	}

 	opts := trace.GetInterceptorOpts()
-	dialContext, cancel := context.WithTimeout(ctx, dialTimeout)
+	dialContext, cancel := context.WithTimeout(ctx, c.DialTimeout)

 	// refer to https://github.com/grpc/grpc-proto/blob/master/grpc/service_config/service_config.proto
 	retryPolicy := `{
@ -166,8 +164,8 @@ func (c *ClientBase) connect(ctx context.Context) error {
 		grpc.WithStreamInterceptor(grpcopentracing.StreamClientInterceptor(opts...)),
 		grpc.WithDefaultServiceConfig(retryPolicy),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{
-			Time:                keepAliveTime,
-			Timeout:             keepAliveTimeout,
+			Time:                c.KeepAliveTime,
+			Timeout:             c.KeepAliveTimeout,
 			PermitWithoutStream: true,
 		}),
 		grpc.WithConnectParams(grpc.ConnectParams{
@ -177,7 +175,7 @@ func (c *ClientBase) connect(ctx context.Context) error {
 				Jitter:     0.2,
 				MaxDelay:   3 * time.Second,
 			},
-			MinConnectTimeout: dialTimeout,
+			MinConnectTimeout: c.DialTimeout,
 		}),
 		grpc.WithPerRPCCredentials(&Token{Value: crypto.Base64Encode(util.MemberCredID)}),
 	)
--- a/internal/util/logutil/logutil.go
+++ b/internal/util/logutil/logutil.go
@ -29,45 +29,73 @@ import (
 	"go.uber.org/zap"
 )

+const (
+	// infoLog indicates Info severity.
+	infoLog int = iota
+	// warningLog indicates Warning severity.
+	warningLog
+	// errorLog indicates Error severity.
+	errorLog
+)
+
 type zapWrapper struct {
-	logger *zap.Logger
+	logger   *zap.Logger
+	logLevel int
 }

 // Info logs a message at InfoLevel.
 func (w *zapWrapper) Info(args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Info(args...)
+	if infoLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Info(args...)
+	}
 }

 func (w *zapWrapper) Infoln(args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Info(args...)
+	if infoLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Info(args...)
+	}
 }

 func (w zapWrapper) Infof(format string, args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Infof(format, args...)
+	if infoLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Infof(format, args...)
+	}
 }

 func (w zapWrapper) Warning(args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Warn(args...)
+	if warningLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Warn(args...)
+	}
 }

 func (w zapWrapper) Warningln(args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Warn(args...)
+	if warningLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Warn(args...)
+	}
 }

 func (w *zapWrapper) Warningf(format string, args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Warnf(format, args...)
+	if warningLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Warnf(format, args...)
+	}
 }

 func (w zapWrapper) Error(args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Error(args...)
+	if errorLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Error(args...)
+	}
 }

 func (w *zapWrapper) Errorln(args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Error(args...)
+	if errorLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Error(args...)
+	}
 }

 func (w zapWrapper) Errorf(format string, args ...interface{}) {
-	w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Errorf(format, args...)
+	if errorLog >= w.logLevel {
+		w.logger.WithOptions(zap.AddCallerSkip(1)).Sugar().Errorf(format, args...)
+	}
 }

 func (w *zapWrapper) Fatal(args ...interface{}) {
@ -88,6 +116,10 @@ func (w *zapWrapper) Fatalf(format string, args ...interface{}) {
 // zap
 // -1=debug, 0=info, 1=warning, 2=error, 3=dpanic, 4=panic, 5=fatal
 func (w *zapWrapper) V(l int) bool {
+	if l < w.logLevel {
+		return false
+	}
+
 	zapLevel := l
 	if l == 3 {
 		zapLevel = 5
@ -116,8 +148,18 @@ func SetupLogger(cfg *log.Config) {
 			log.Fatal("initialize logger error", zap.Error(err))
 		}

-		// initialize grpc and etcd logger
-		wrapper := &zapWrapper{logger}
+		// Initialize grpc log wrapper
+		logLevel := 0
+		switch cfg.GrpcLevel {
+		case "", "ERROR": // If env is unset, set level to ERROR.
+			logLevel = 2
+		case "WARNING":
+			logLevel = 1
+		case "INFO":
+			logLevel = 0
+		}
+
+		wrapper := &zapWrapper{logger, logLevel}
 		grpclog.SetLoggerV2(wrapper)

 		log.Info("Log directory", zap.String("configDir", cfg.File.RootPath))
--- a/internal/util/paramtable/base_table.go
+++ b/internal/util/paramtable/base_table.go
@ -43,10 +43,7 @@ const (
 	DefaultMinioSecretAccessKey = "minioadmin"
 	DefaultMinioUseSSL          = "false"
 	DefaultMinioBucketName      = "a-bucket"
-	DefaultPulsarHost           = "localhost"
-	DefaultPulsarPort           = "6650"
 	DefaultEtcdEndpoints        = "localhost:2379"
-	DefaultRocksmqPath          = "/var/lib/milvus/rdb_data"
 	DefaultInsertBufferSize     = "16777216"
 	DefaultEnvPrefix            = "milvus"
 )
@ -393,6 +390,13 @@ func (gp *BaseTable) InitLogCfg() {
 // SetLogConfig set log config of the base table
 func (gp *BaseTable) SetLogConfig() {
 	gp.LogCfgFunc = func(cfg log.Config) {
+		var err error
+		grpclog, err := gp.Load("grpc.log.level")
+		if err != nil {
+			cfg.GrpcLevel = DefaultLogLevel
+		} else {
+			cfg.GrpcLevel = strings.ToUpper(grpclog)
+		}
 		logutil.SetupLogger(&cfg)
 		defer log.Sync()
 	}
--- a/internal/util/paramtable/base_table_test.go
+++ b/internal/util/paramtable/base_table_test.go
@ -20,6 +20,7 @@ import (

 	"github.com/milvus-io/milvus/internal/proto/commonpb"
 	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/grpclog"
 )

 var baseParams = BaseTable{}
@ -314,4 +315,14 @@ func Test_SetLogger(t *testing.T) {
 		baseParams.SetLogger(UniqueID(0))
 		assert.Equal(t, "datanode-0.log", baseParams.Log.File.Filename)
 	})
+
+	t.Run("TestGrpclog", func(t *testing.T) {
+		baseParams.Save("grpc.log.level", "Warning")
+		baseParams.SetLogConfig()
+
+		baseParams.SetLogger(UniqueID(1))
+		assert.Equal(t, false, grpclog.V(0))
+		assert.Equal(t, true, grpclog.V(1))
+		assert.Equal(t, true, grpclog.V(2))
+	})
 }
--- a/internal/util/paramtable/component_param.go
+++ b/internal/util/paramtable/component_param.go
@ -136,7 +136,6 @@ func (p *commonConfig) init(base *BaseTable) {

 	// must init cluster prefix first
 	p.initClusterPrefix()
-
 	p.initProxySubName()

 	p.initRootCoordTimeTick()
--- a/internal/util/paramtable/grpc_param.go
+++ b/internal/util/paramtable/grpc_param.go
@ -15,6 +15,7 @@ import (
 	"math"
 	"strconv"
 	"sync"
+	"time"

 	"github.com/go-basic/ipv4"
 	"github.com/milvus-io/milvus/internal/log"
@ -33,6 +34,14 @@ const (

 	// DefaultClientMaxRecvSize defines the maximum size of data per grpc request can receive by client side.
 	DefaultClientMaxRecvSize = 100 * 1024 * 1024
+
+	// DefaultLogLevel defines the log level of grpc
+	DefaultLogLevel = "WARNING"
+
+	// Grpc Timeout related configs
+	DefaultDialTimeout      = 5000 * time.Millisecond
+	DefaultKeepAliveTime    = 10000 * time.Millisecond
+	DefaultKeepAliveTimeout = 3000 * time.Millisecond
 )

 ///////////////////////////////////////////////////////////////////////////////
@ -99,20 +108,22 @@ func (p *GrpcServerConfig) init(domain string) {
 func (p *GrpcServerConfig) initServerMaxSendSize() {
 	var err error

-	valueStr, err := p.Load(p.Domain + ".grpc.serverMaxSendSize")
+	valueStr, err := p.Load("grpc.serverMaxSendSize")
 	if err != nil {
-		p.ServerMaxSendSize = DefaultServerMaxSendSize
+		valueStr, err = p.Load(p.Domain + ".grpc.serverMaxSendSize")
 	}
-
-	value, err := strconv.Atoi(valueStr)
 	if err != nil {
-		log.Warn("Failed to parse grpc.serverMaxSendSize, set to default",
-			zap.String("rol", p.Domain), zap.String("grpc.serverMaxSendSize", valueStr),
-			zap.Error(err))
-
 		p.ServerMaxSendSize = DefaultServerMaxSendSize
 	} else {
-		p.ServerMaxSendSize = value
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.serverMaxSendSize, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.serverMaxSendSize", valueStr),
+				zap.Error(err))
+			p.ServerMaxSendSize = DefaultServerMaxSendSize
+		} else {
+			p.ServerMaxSendSize = value
+		}
 	}

 	log.Debug("initServerMaxSendSize",
@ -121,21 +132,22 @@ func (p *GrpcServerConfig) initServerMaxSendSize() {

 func (p *GrpcServerConfig) initServerMaxRecvSize() {
 	var err error
-
-	valueStr, err := p.Load(p.Domain + ".grpc.serverMaxRecvSize")
+	valueStr, err := p.Load("grpc.serverMaxRecvSize")
 	if err != nil {
-		p.ServerMaxRecvSize = DefaultServerMaxRecvSize
+		valueStr, err = p.Load(p.Domain + ".grpc.serverMaxRecvSize")
 	}
-
-	value, err := strconv.Atoi(valueStr)
 	if err != nil {
-		log.Warn("Failed to parse grpc.serverMaxRecvSize, set to default",
-			zap.String("role", p.Domain), zap.String("grpc.serverMaxRecvSize", valueStr),
-			zap.Error(err))
-
 		p.ServerMaxRecvSize = DefaultServerMaxRecvSize
 	} else {
-		p.ServerMaxRecvSize = value
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.serverMaxRecvSize, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.serverMaxRecvSize", valueStr),
+				zap.Error(err))
+			p.ServerMaxRecvSize = DefaultServerMaxRecvSize
+		} else {
+			p.ServerMaxRecvSize = value
+		}
 	}

 	log.Debug("initServerMaxRecvSize",
@ -148,6 +160,10 @@ type GrpcClientConfig struct {

 	ClientMaxSendSize int
 	ClientMaxRecvSize int
+
+	DialTimeout      time.Duration
+	KeepAliveTime    time.Duration
+	KeepAliveTimeout time.Duration
 }

 // InitOnce initialize grpc client config once
@ -162,25 +178,31 @@ func (p *GrpcClientConfig) init(domain string) {

 	p.initClientMaxSendSize()
 	p.initClientMaxRecvSize()
+	p.initDialTimeout()
+	p.initKeepAliveTimeout()
+	p.initKeepAliveTime()
 }

 func (p *GrpcClientConfig) initClientMaxSendSize() {
 	var err error

-	valueStr, err := p.Load(p.Domain + ".grpc.clientMaxSendSize")
+	valueStr, err := p.Load("grpc.clientMaxSendSize")
 	if err != nil {
-		p.ClientMaxSendSize = DefaultClientMaxSendSize
+		valueStr, err = p.Load(p.Domain + ".grpc.clientMaxSendSize")
 	}
-
-	value, err := strconv.Atoi(valueStr)
 	if err != nil {
-		log.Warn("Failed to parse grpc.clientMaxSendSize, set to default",
-			zap.String("role", p.Domain), zap.String("grpc.clientMaxSendSize", valueStr),
-			zap.Error(err))
-
 		p.ClientMaxSendSize = DefaultClientMaxSendSize
 	} else {
-		p.ClientMaxSendSize = value
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.clientMaxSendSize, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.clientMaxSendSize", valueStr),
+				zap.Error(err))
+
+			p.ClientMaxSendSize = DefaultClientMaxSendSize
+		} else {
+			p.ClientMaxSendSize = value
+		}
 	}

 	log.Debug("initClientMaxSendSize",
@ -189,23 +211,86 @@ func (p *GrpcClientConfig) initClientMaxSendSize() {

 func (p *GrpcClientConfig) initClientMaxRecvSize() {
 	var err error
-
-	valueStr, err := p.Load(p.Domain + ".grpc.clientMaxRecvSize")
+	valueStr, err := p.Load("grpc.clientMaxRecvSize")
 	if err != nil {
-		p.ClientMaxRecvSize = DefaultClientMaxRecvSize
+		valueStr, err = p.Load(p.Domain + ".grpc.clientMaxRecvSize")
 	}
-
-	value, err := strconv.Atoi(valueStr)
 	if err != nil {
-		log.Warn("Failed to parse grpc.clientMaxRecvSize, set to default",
-			zap.String("role", p.Domain), zap.String("grpc.clientMaxRecvSize", valueStr),
-			zap.Error(err))
-
 		p.ClientMaxRecvSize = DefaultClientMaxRecvSize
 	} else {
-		p.ClientMaxRecvSize = value
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.clientMaxRecvSize, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.clientMaxRecvSize", valueStr),
+				zap.Error(err))
+
+			p.ClientMaxRecvSize = DefaultClientMaxRecvSize
+		} else {
+			p.ClientMaxRecvSize = value
+		}
 	}

 	log.Debug("initClientMaxRecvSize",
 		zap.String("role", p.Domain), zap.Int("grpc.clientMaxRecvSize", p.ClientMaxRecvSize))
 }
+
+func (p *GrpcClientConfig) initDialTimeout() {
+	var err error
+	valueStr, err := p.Load("grpc.client.dialTimeout")
+	if err != nil {
+		p.DialTimeout = DefaultDialTimeout
+	} else {
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.client.dialTimeout, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.client.dialTimeout", valueStr),
+				zap.Error(err))
+			p.DialTimeout = DefaultDialTimeout
+		} else {
+			p.DialTimeout = time.Duration(value) * 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
+	valueStr, err := p.Load("grpc.client.keepAliveTime")
+	if err != nil {
+		p.KeepAliveTime = DefaultKeepAliveTime
+	} else {
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.client.keepAliveTime, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.client.keepAliveTime", valueStr),
+				zap.Error(err))
+
+			p.KeepAliveTime = DefaultKeepAliveTime
+		} else {
+			p.KeepAliveTime = time.Duration(value) * 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
+	valueStr, err := p.Load("grpc.client.keepAliveTimeout")
+	if err != nil {
+		p.KeepAliveTimeout = DefaultKeepAliveTimeout
+	} else {
+		value, err := strconv.Atoi(valueStr)
+		if err != nil {
+			log.Warn("Failed to parse grpc.client.keepAliveTimeout, set to default",
+				zap.String("role", p.Domain), zap.String("grpc.client.keepAliveTimeout", valueStr),
+				zap.Error(err))
+			p.KeepAliveTimeout = DefaultKeepAliveTimeout
+		} else {
+			p.KeepAliveTimeout = time.Duration(value) * time.Millisecond
+		}
+	}
+	log.Debug("Init keep alive timeout",
+		zap.String("role", p.Domain), zap.Duration("grpc.log.keepAliveTimeout", p.KeepAliveTimeout))
+}
--- a/internal/util/paramtable/grpc_param_test.go
+++ b/internal/util/paramtable/grpc_param_test.go
@ -13,6 +13,7 @@ package paramtable

 import (
 	"testing"
+	"time"

 	"github.com/milvus-io/milvus/internal/util/typeutil"
 	"github.com/stretchr/testify/assert"
@ -68,6 +69,11 @@ func TestGrpcClientParams(t *testing.T) {
 	assert.NotZero(t, Params.ClientMaxRecvSize)
 	t.Logf("ClientMaxRecvSize = %d", Params.ClientMaxRecvSize)

+	Params.Remove("grpc.clientMaxRecvSize")
+	Params.Save(role+".grpc.clientMaxRecvSize", "1000")
+	Params.initClientMaxRecvSize()
+	assert.Equal(t, Params.ClientMaxRecvSize, 1000)
+
 	Params.Remove(role + ".grpc.clientMaxRecvSize")
 	Params.initClientMaxRecvSize()
 	assert.Equal(t, Params.ClientMaxRecvSize, DefaultClientMaxRecvSize)
@ -75,7 +81,40 @@ func TestGrpcClientParams(t *testing.T) {
 	assert.NotZero(t, Params.ClientMaxSendSize)
 	t.Logf("ClientMaxSendSize = %d", Params.ClientMaxSendSize)

+	Params.Remove("grpc.clientMaxSendSize")
+	Params.Save(role+".grpc.clientMaxSendSize", "2000")
+	Params.initClientMaxSendSize()
+	assert.Equal(t, Params.ClientMaxSendSize, 2000)
+
 	Params.Remove(role + ".grpc.clientMaxSendSize")
 	Params.initClientMaxSendSize()
 	assert.Equal(t, Params.ClientMaxSendSize, DefaultClientMaxSendSize)
+
+	Params.initDialTimeout()
+	assert.Equal(t, Params.DialTimeout, DefaultDialTimeout)
+	Params.Save("grpc.client.dialTimeout", "aaa")
+	Params.initDialTimeout()
+	assert.Equal(t, Params.DialTimeout, DefaultDialTimeout)
+	Params.Save("grpc.client.dialTimeout", "100")
+	Params.initDialTimeout()
+	assert.Equal(t, Params.DialTimeout, 100*time.Millisecond)
+
+	Params.initKeepAliveTime()
+	assert.Equal(t, Params.KeepAliveTime, DefaultKeepAliveTime)
+	Params.Save("grpc.client.keepAliveTime", "a")
+	Params.initKeepAliveTime()
+	assert.Equal(t, Params.KeepAliveTime, DefaultKeepAliveTime)
+	Params.Save("grpc.client.keepAliveTime", "200")
+	Params.initKeepAliveTime()
+	assert.Equal(t, Params.KeepAliveTime, 200*time.Millisecond)
+
+	Params.initKeepAliveTimeout()
+	assert.Equal(t, Params.KeepAliveTimeout, DefaultKeepAliveTimeout)
+	Params.Save("grpc.client.keepAliveTimeout", "a")
+	Params.initKeepAliveTimeout()
+	assert.Equal(t, Params.KeepAliveTimeout, DefaultKeepAliveTimeout)
+	Params.Save("grpc.client.keepAliveTimeout", "500")
+	Params.initKeepAliveTimeout()
+	assert.Equal(t, Params.KeepAliveTimeout, 500*time.Millisecond)
+
 }