milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix createCollection failed occasionally (#28592) (#28712) 

fix: create collection seldom failure #28592

Signed-off-by: MrPresent-Han <chun.han@zilliz.com>
pull/28750/head
MrPresent-Han 2023-11-27 11:10:25 +08:00 committed by GitHub
parent f13ef1665a
commit fc30d291be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 52 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
internal/distributed/datanode/client/client.go
View File

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"go.uber.org/zap"
"google.golang.org/grpc"
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@ -27,6 +28,8 @@ import (
"github.com/milvus-io/milvus/internal/proto/datapb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/grpcclient"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/commonpbutil"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/paramtable"
@ -38,6 +41,7 @@ var Params *paramtable.ComponentParam = paramtable.Get()
// Client is the grpc client for DataNode
type Client struct {
grpcClient grpcclient.GrpcClient[datapb.DataNodeClient]
sess *sessionutil.Session
addr string
}
@ -46,16 +50,24 @@ func NewClient(ctx context.Context, addr string, nodeID int64) (*Client, error)
if addr == "" {
return nil, fmt.Errorf("address is empty")
}
sess := sessionutil.NewSession(ctx)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
log.Debug("DataNodeClient New Etcd Session failed", zap.Error(err))
return nil, err
}
config := &Params.DataNodeGrpcClientCfg
client := &Client{
addr: addr,
grpcClient: grpcclient.NewClientBase[datapb.DataNodeClient](config, "milvus.proto.data.DataNode"),
sess: sess,
}
// node shall specify node id
client.grpcClient.SetRole(fmt.Sprintf("%s-%d", typeutil.DataNodeRole, nodeID))
client.grpcClient.SetGetAddrFunc(client.getAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetNodeID(nodeID)
client.grpcClient.SetSession(sess)
return client, nil
}

12
internal/distributed/indexnode/client/client.go
View File

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"go.uber.org/zap"
"google.golang.org/grpc"
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@ -27,6 +28,8 @@ import (
"github.com/milvus-io/milvus/internal/proto/indexpb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/grpcclient"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/commonpbutil"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/paramtable"
@ -39,6 +42,7 @@ var Params *paramtable.ComponentParam = paramtable.Get()
type Client struct {
grpcClient grpcclient.GrpcClient[indexpb.IndexNodeClient]
addr string
sess *sessionutil.Session
}
// NewClient creates a new IndexNode client.
@ -46,16 +50,24 @@ func NewClient(ctx context.Context, addr string, nodeID int64, encryption bool)
if addr == "" {
return nil, fmt.Errorf("address is empty")
}
sess := sessionutil.NewSession(ctx)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
log.Debug("IndexNodeClient New Etcd Session failed", zap.Error(err))
return nil, err
}
config := &Params.IndexNodeGrpcClientCfg
client := &Client{
addr: addr,
grpcClient: grpcclient.NewClientBase[indexpb.IndexNodeClient](config, "milvus.proto.index.IndexNode"),
sess: sess,
}
// node shall specify node id
client.grpcClient.SetRole(fmt.Sprintf("%s-%d", typeutil.IndexNodeRole, nodeID))
client.grpcClient.SetGetAddrFunc(client.getAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetNodeID(nodeID)
client.grpcClient.SetSession(sess)
if encryption {
client.grpcClient.EnableEncryption()
}

12
internal/distributed/proxy/client/client.go
View File

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"go.uber.org/zap"
"google.golang.org/grpc"
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@ -27,6 +28,8 @@ import (
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/proxypb"
"github.com/milvus-io/milvus/internal/util/grpcclient"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/commonpbutil"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/paramtable"
@ -39,6 +42,7 @@ var Params *paramtable.ComponentParam = paramtable.Get()
type Client struct {
grpcClient grpcclient.GrpcClient[proxypb.ProxyClient]
addr string
sess *sessionutil.Session
}
// NewClient creates a new client instance
@ -46,16 +50,24 @@ func NewClient(ctx context.Context, addr string, nodeID int64) (*Client, error)
if addr == "" {
return nil, fmt.Errorf("address is empty")
}
sess := sessionutil.NewSession(ctx)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
log.Debug("Proxy client new session failed", zap.Error(err))
return nil, err
}
config := &Params.ProxyGrpcClientCfg
client := &Client{
addr: addr,
grpcClient: grpcclient.NewClientBase[proxypb.ProxyClient](config, "milvus.proto.proxy.Proxy"),
sess: sess,
}
// node shall specify node id
client.grpcClient.SetRole(fmt.Sprintf("%s-%d", typeutil.ProxyRole, nodeID))
client.grpcClient.SetGetAddrFunc(client.getAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetNodeID(nodeID)
client.grpcClient.SetSession(sess)
return client, nil
}

6
internal/distributed/proxy/service.go
View File

@ -319,14 +319,14 @@ func (s *Server) startInternalGrpc(grpcPort int, errChan chan error) {
Timeout: 10 * time.Second, // Wait 10 second for the ping ack before assuming the connection is dead
}
log.Debug("Proxy internal server listen on tcp", zap.Int("port", grpcPort))
log.Info("Proxy internal server listen on tcp", zap.Int("port", grpcPort))
lis, err := net.Listen("tcp", ":"+strconv.Itoa(grpcPort))
if err != nil {
log.Warn("Proxy internal server failed to listen on", zap.Error(err), zap.Int("port", grpcPort))
errChan <- err
return
}
log.Debug("Proxy internal server already listen on tcp", zap.Int("port", grpcPort))
log.Info("Proxy internal server already listen on tcp", zap.Int("port", grpcPort))
opts := tracer.GetInterceptorOpts()
s.grpcInternalServer = grpc.NewServer(
@ -358,7 +358,7 @@ func (s *Server) startInternalGrpc(grpcPort int, errChan chan error) {
grpc_health_v1.RegisterHealthServer(s.grpcInternalServer, s)
errChan <- nil
log.Debug("create Proxy internal grpc server",
log.Info("create Proxy internal grpc server",
zap.Any("enforcement policy", kaep),
zap.Any("server parameters", kasp))

12
internal/distributed/querynode/client/client.go
View File

@ -20,6 +20,7 @@ import (
"context"
"fmt"
"go.uber.org/zap"
"google.golang.org/grpc"
"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@ -27,6 +28,8 @@ import (
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/querypb"
"github.com/milvus-io/milvus/internal/util/grpcclient"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/pkg/log"
"github.com/milvus-io/milvus/pkg/util/commonpbutil"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/paramtable"
@ -37,6 +40,7 @@ import (
type Client struct {
grpcClient grpcclient.GrpcClient[querypb.QueryNodeClient]
addr string
sess *sessionutil.Session
}
// NewClient creates a new QueryNode client.
@ -44,16 +48,24 @@ func NewClient(ctx context.Context, addr string, nodeID int64) (*Client, error)
if addr == "" {
return nil, fmt.Errorf("addr is empty")
}
sess := sessionutil.NewSession(ctx)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
log.Debug("QueryNodeClient NewClient failed", zap.Error(err))
return nil, err
}
config := &paramtable.Get().QueryNodeGrpcClientCfg
client := &Client{
addr: addr,
grpcClient: grpcclient.NewClientBase[querypb.QueryNodeClient](config, "milvus.proto.query.QueryNode"),
sess: sess,
}
// node shall specify node id
client.grpcClient.SetRole(fmt.Sprintf("%s-%d", typeutil.QueryNodeRole, nodeID))
client.grpcClient.SetGetAddrFunc(client.getAddr)
client.grpcClient.SetNewGrpcClientFunc(client.newGrpcClient)
client.grpcClient.SetNodeID(nodeID)
client.grpcClient.SetSession(sess)
return client, nil
}

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

@ -379,7 +379,7 @@ func (c *ClientBase[T]) checkGrpcErr(ctx context.Context, err error) (needRetry,
return false, false, merr.WrapErrServiceUnimplemented(err)
case IsServerIDMismatchErr(err):
if ok, err := c.checkNodeSessionExist(ctx); !ok {
// if session doesn't exist, no need to retry for datanode/indexnode/querynode
// if session doesn't exist, no need to retry for datanode/indexnode/querynode/proxy
return false, false, err
}
return true, true, err