recreate grpc connections if too many errors

address https://github.com/chrislusf/seaweedfs/issues/2098

1 files changed, 15 insertions, 8 deletions

diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go
index edb60e4fa..5490339b3 100644
--- a/weed/pb/grpc_client_server.go
+++ b/weed/pb/grpc_client_server.go
@@ -31,7 +31,8 @@ var (
 
 type versionedGrpcClient struct {
 	*grpc.ClientConn
-	version int
+	version  int
+	errCount int
 }
 
 func init() {
@@ -103,6 +104,7 @@ func getOrCreateConnection(address string, opts ...grpc.DialOption) (*versionedG
 	vgc := &versionedGrpcClient{
 		grpcConnection,
 		rand.Int(),
+		0,
 	}
 	grpcClients[address] = vgc
 
@@ -116,15 +118,20 @@ func WithCachedGrpcClient(fn func(*grpc.ClientConn) error, address string, opts
 		return fmt.Errorf("getOrCreateConnection %s: %v", address, err)
 	}
 	executionErr := fn(vgc.ClientConn)
-	if executionErr != nil && strings.Contains(executionErr.Error(), "transport") {
-		grpcClientsLock.Lock()
-		if t, ok := grpcClients[address]; ok {
-			if t.version == vgc.version {
-				vgc.Close()
-				delete(grpcClients, address)
+	if executionErr != nil {
+		vgc.errCount++
+		if vgc.errCount > 3 ||
+			strings.Contains(executionErr.Error(), "transport") ||
+			strings.Contains(executionErr.Error(), "connection closed") {
+			grpcClientsLock.Lock()
+			if t, ok := grpcClients[address]; ok {
+				if t.version == vgc.version {
+					vgc.Close()
+					delete(grpcClients, address)
+				}
 			}
+			grpcClientsLock.Unlock()
 		}
-		grpcClientsLock.Unlock()
 	}
 
 	return executionErr