admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-02-02 01:06:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Check and reset if grpc client serverID mismatch with session (#26448)

Browse Source 
Signed-off-by: wayblink <anyang.wang@zilliz.com>
This commit is contained in:
wayblink 2023-08-18 22:50:20 +08:00 committed by GitHub
parent 79ca04b995
commit 71133e5eaa
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 38 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
internal/distributed/datacoord/client/client.go
View File

@ -80,6 +80,7 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
client.grpcClient.SetRole(typeutil.DataCoordRole)
client.grpcClient.SetGetAddrFunc(client.getDataCoordAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetSession(sess)
return client, nil
}

1
internal/distributed/querycoord/client/client.go
View File

@ -73,6 +73,7 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
client.grpcClient.SetRole(typeutil.QueryCoordRole)
client.grpcClient.SetGetAddrFunc(client.getQueryCoordAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetSession(sess)
return client, nil
}

1
internal/distributed/rootcoord/client/client.go
View File

@ -80,6 +80,7 @@ func NewClient(ctx context.Context, metaRoot string, etcdCli *clientv3.Client) (
client.grpcClient.SetRole(typeutil.RootCoordRole)
client.grpcClient.SetGetAddrFunc(client.getRootCoordAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetSession(sess)
return client, nil
}

33
internal/util/grpcclient/client.go
View File

@ -36,6 +36,7 @@ import (
"google.golang.org/grpc/keepalive"
"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/tracer"
"github.com/milvus-io/milvus/pkg/util"
@ -62,6 +63,7 @@ type GrpcClient[T interface {
Close() error
SetNodeID(int64)
GetNodeID() int64
SetSession(sess *sessionutil.Session)
}
// ClientBase is a base of grpc client
@ -90,6 +92,7 @@ type ClientBase[T interface {
MaxBackoff float32
BackoffMultiplier float32
NodeID atomic.Int64
sess *sessionutil.Session
sf singleflight.Group
}
@ -294,11 +297,6 @@ func (c *ClientBase[T]) callOnce(ctx context.Context, caller func(client T) (any
return ret, nil
}
if !funcutil.CheckCtxValid(ctx) {
// start bg check in case of https://github.com/milvus-io/milvus/issues/22435
go c.bgHealthCheck(client)
return generic.Zero[T](), err
}
if IsCrossClusterRoutingErr(err) {
log.Warn("CrossClusterRoutingErr, start to reset connection", zap.Error(err))
c.resetConnection(client)
@ -309,6 +307,26 @@ func (c *ClientBase[T]) callOnce(ctx context.Context, caller func(client T) (any
c.resetConnection(client)
return ret, err
}
if !funcutil.CheckCtxValid(ctx) {
// check if server ID matches coord session, if not, reset connection
if c.sess != nil {
sessions, _, getSessionErr := c.sess.GetSessions(c.GetRole())
if getSessionErr != nil {
// Only log but not handle this error as it is an auxiliary logic
log.Warn("Fail to GetSessions", zap.Error(getSessionErr))
}
if coordSess, exist := sessions[c.GetRole()]; exist {
if c.GetNodeID() != coordSess.ServerID {
log.Warn("Server ID mismatch, may connected to a old server, start to reset connection", zap.Error(err))
c.resetConnection(client)
return ret, err
}
}
}
// start bg check in case of https://github.com/milvus-io/milvus/issues/22435
go c.bgHealthCheck(client)
return generic.Zero[T](), err
}
if !funcutil.IsGrpcErr(err) {
log.Warn("ClientBase:isNotGrpcErr", zap.Error(err))
return generic.Zero[T](), err
@ -397,6 +415,11 @@ func (c *ClientBase[T]) GetNodeID() int64 {
return c.NodeID.Load()
}
// SetSession set session role of client
func (c *ClientBase[T]) SetSession(sess *sessionutil.Session) {
c.sess = sess
}
func IsCrossClusterRoutingErr(err error) bool {
// GRPC utilizes `status.Status` to encapsulate errors,
// hence it is not viable to employ the `errors.Is` for assessment.

8
internal/util/mock/grpcclient.go
View File

@ -21,11 +21,12 @@ import (
"fmt"
"sync"
"github.com/milvus-io/milvus/pkg/tracer"
"go.uber.org/zap"
"google.golang.org/grpc"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/tracer"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/generic"
"github.com/milvus-io/milvus/pkg/util/retry"
@ -41,6 +42,7 @@ type GRPCClientBase[T any] struct {
GetGrpcClientErr error
role string
nodeID int64
sess *sessionutil.Session
}
func (c *GRPCClientBase[T]) SetGetAddrFunc(f func() (string, error)) {
@ -163,3 +165,7 @@ func (c *GRPCClientBase[T]) GetNodeID() int64 {
func (c *GRPCClientBase[T]) SetNodeID(nodeID int64) {
c.nodeID = nodeID
}
func (c *GRPCClientBase[T]) SetSession(sess *sessionutil.Session) {
c.sess = sess
}