Set grpc max recv mag as math.MaxInt32

Signed-off-by: Nneza2017 <yefu.chen@zilliz.com>

internal/distributed/datanode/service.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -73,8 +74,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	log.Debug("DataNode address", zap.String("address", addr))
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	datapb.RegisterDataNodeServer(s.grpcServer, s)

internal/distributed/dataservice/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -134,8 +135,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	datapb.RegisterDataServiceServer(s.grpcServer, s)

internal/distributed/indexnode/service.go

@@ -3,6 +3,7 @@ package grpcindexnode
 import (
 	"context"
 	"log"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -61,8 +62,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	indexpb.RegisterIndexNodeServer(s.grpcServer, s)

internal/distributed/indexservice/service.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -133,8 +134,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	indexpb.RegisterIndexServiceServer(s.grpcServer, s)

internal/distributed/masterservice/server.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -209,8 +210,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	masterpb.RegisterMasterServiceServer(s.grpcServer, s)

internal/distributed/proxynode/service.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -83,8 +84,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)),
 		grpc.MaxRecvMsgSize(GRPCMaxMagSize))

internal/distributed/proxyservice/service.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"log"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -116,8 +117,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	proxypb.RegisterProxyServiceServer(s.grpcServer, s)

internal/distributed/querynode/service.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -221,8 +222,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	log.Debug("QueryNode", zap.String("address", addr))
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	querypb.RegisterQueryNodeServer(s.grpcServer, s)

internal/distributed/queryservice/service.go

@@ -2,6 +2,7 @@ package grpcqueryservice
 
 import (
 	"context"
+	"math"
 	"net"
 	"strconv"
 	"sync"
@@ -152,8 +153,11 @@ func (s *Server) startGrpcLoop(grpcPort int) {
 	defer cancel()
 
 	tracer := opentracing.GlobalTracer()
-	s.grpcServer = grpc.NewServer(grpc.UnaryInterceptor(
-		otgrpc.OpenTracingServerInterceptor(tracer)),
+	s.grpcServer = grpc.NewServer(
+		grpc.MaxRecvMsgSize(math.MaxInt32),
+		grpc.MaxSendMsgSize(math.MaxInt32),
+		grpc.UnaryInterceptor(
+			otgrpc.OpenTracingServerInterceptor(tracer)),
 		grpc.StreamInterceptor(
 			otgrpc.OpenTracingStreamServerInterceptor(tracer)))
 	querypb.RegisterQueryServiceServer(s.grpcServer, s)