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

Change retry times (#6083) 

Signed-off-by: godchen <qingxiang.chen@zilliz.com>
pull/6092/head
godchen 2021-06-24 19:05:06 +08:00 committed by GitHub
parent 10e5fc3382
commit 3785bb6330
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
22 changed files with 104 additions and 135 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
internal/allocator/id.go
View File

@ -23,7 +23,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/commonpb"
"github.com/milvus-io/milvus/internal/proto/rootcoordpb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/typeutil"
)
@ -72,7 +71,7 @@ func NewIDAllocator(ctx context.Context, metaRoot string, etcdEndpoints []string
func (ia *IDAllocator) Start() error {
var err error
ia.rootCoordClient, err = rcc.NewClient(ia.Ctx, ia.metaRoot, ia.etcdEndpoints, retry.Attempts(300))
ia.rootCoordClient, err = rcc.NewClient(ia.Ctx, ia.metaRoot, ia.etcdEndpoints)
if err != nil {
panic(err)
}

3
internal/datacoord/cluster_session_manager.go
View File

@ -15,7 +15,6 @@ import (
"sync"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/retry"
)
const retryTimes = 2
@ -43,7 +42,7 @@ func newClusterSessionManager(ctx context.Context, dataClientCreator dataNodeCre
// lock acquired
func (m *clusterSessionManager) createSession(addr string) (types.DataNode, error) {
cli, err := m.dataClientCreator(m.ctx, addr, retry.Attempts(300))
cli, err := m.dataClientCreator(m.ctx, addr)
if err != nil {
return nil, err
}

14
internal/datacoord/server.go
View File

@ -48,8 +48,8 @@ type (
Timestamp = typeutil.Timestamp
)
type dataNodeCreatorFunc func(ctx context.Context, addr string, retryOptions ...retry.Option) (types.DataNode, error)
type rootCoordCreatorFunc func(ctx context.Context, metaRootPath string, etcdEndpoints []string, retryOptions ...retry.Option) (types.RootCoord, error)
type dataNodeCreatorFunc func(ctx context.Context, addr string) (types.DataNode, error)
type rootCoordCreatorFunc func(ctx context.Context, metaRootPath string, etcdEndpoints []string) (types.RootCoord, error)
type Server struct {
ctx context.Context
@ -91,12 +91,12 @@ func CreateServer(ctx context.Context, factory msgstream.Factory) (*Server, erro
return s, nil
}
func defaultDataNodeCreatorFunc(ctx context.Context, addr string, retryOptions ...retry.Option) (types.DataNode, error) {
return datanodeclient.NewClient(ctx, addr, retryOptions...)
func defaultDataNodeCreatorFunc(ctx context.Context, addr string) (types.DataNode, error) {
return datanodeclient.NewClient(ctx, addr)
}
func defaultRootCoordCreatorFunc(ctx context.Context, metaRootPath string, etcdEndpoints []string, retryOptions ...retry.Option) (types.RootCoord, error) {
return rootcoordclient.NewClient(ctx, metaRootPath, etcdEndpoints, retryOptions...)
func defaultRootCoordCreatorFunc(ctx context.Context, metaRootPath string, etcdEndpoints []string) (types.RootCoord, error) {
return rootcoordclient.NewClient(ctx, metaRootPath, etcdEndpoints)
}
// Register register data service at etcd
@ -450,7 +450,7 @@ func (s *Server) handleFlushingSegments(ctx context.Context) {
func (s *Server) initRootCoordClient() error {
var err error
if s.rootCoordClient, err = s.rootCoordClientCreator(s.ctx, Params.MetaRootPath, Params.EtcdEndpoints, retry.Attempts(300)); err != nil {
if s.rootCoordClient, err = s.rootCoordClientCreator(s.ctx, Params.MetaRootPath, Params.EtcdEndpoints); err != nil {
return err
}
if err = s.rootCoordClient.Init(); err != nil {

6
internal/datacoord/server_test.go
View File

@ -636,7 +636,7 @@ func TestGetRecoveryInfo(t *testing.T) {
svr := newTestServer(t, nil)
defer closeTestServer(t, svr)
svr.rootCoordClientCreator = func(ctx context.Context, metaRootPath string, etcdEndpoints []string, retryOptions ...retry.Option) (types.RootCoord, error) {
svr.rootCoordClientCreator = func(ctx context.Context, metaRootPath string, etcdEndpoints []string) (types.RootCoord, error) {
return newMockRootCoordService(), nil
}
@ -773,10 +773,10 @@ func newTestServer(t *testing.T, receiveCh chan interface{}) *Server {
svr, err := CreateServer(context.TODO(), factory)
assert.Nil(t, err)
svr.dataClientCreator = func(ctx context.Context, addr string, retryOptions ...retry.Option) (types.DataNode, error) {
svr.dataClientCreator = func(ctx context.Context, addr string) (types.DataNode, error) {
return newMockDataNodeClient(0, receiveCh)
}
svr.rootCoordClientCreator = func(ctx context.Context, metaRootPath string, etcdEndpoints []string, retryOptions ...retry.Option) (types.RootCoord, error) {
svr.rootCoordClientCreator = func(ctx context.Context, metaRootPath string, etcdEndpoints []string) (types.RootCoord, error) {
return newMockRootCoordService(), nil
}
assert.Nil(t, err)

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

@ -42,8 +42,6 @@ type Client struct {
sess *sessionutil.Session
addr string
retryOptions []retry.Option
}
func getDataCoordAddress(sess *sessionutil.Session) (string, error) {
@ -61,7 +59,7 @@ func getDataCoordAddress(sess *sessionutil.Session) (string, error) {
return ms.Address, nil
}
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) (*Client, error) {
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string) (*Client, error) {
sess := sessionutil.NewSession(ctx, metaRoot, etcdEndpoints)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
@ -70,18 +68,17 @@ func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, ret
}
ctx, cancel := context.WithCancel(ctx)
return &Client{
ctx: ctx,
cancel: cancel,
sess: sess,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
sess: sess,
}, nil
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
var err error
connectDataCoordFn := func() error {
c.addr, err = getDataCoordAddress(c.sess)
@ -111,7 +108,7 @@ func (c *Client) connect() error {
return nil
}
err = retry.Do(c.ctx, connectDataCoordFn, c.retryOptions...)
err = retry.Do(c.ctx, connectDataCoordFn, retryOptions...)
if err != nil {
log.Debug("DataCoord try reconnect failed", zap.Error(err))
return err

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

@ -59,10 +59,10 @@ func NewClient(ctx context.Context, addr string, retryOptions ...retry.Option) (
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
connectGrpcFunc := func() error {
opts := trace.GetInterceptorOpts()
log.Debug("DataNode connect ", zap.String("address", c.addr))
@ -86,7 +86,7 @@ func (c *Client) connect() error {
return nil
}
err := retry.Do(c.ctx, connectGrpcFunc, c.retryOptions...)
err := retry.Do(c.ctx, connectGrpcFunc, retryOptions...)
if err != nil {
log.Debug("DataNodeClient try connect failed", zap.Error(err))
return err

5
internal/distributed/datanode/datanode_test.go
View File

@ -25,7 +25,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/milvuspb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/stretchr/testify/assert"
)
@ -111,10 +110,10 @@ func TestRun(t *testing.T) {
Params.Init()
dnServer.newRootCoordClient = func(string, []string, ...retry.Option) (types.RootCoord, error) {
dnServer.newRootCoordClient = func(string, []string) (types.RootCoord, error) {
return &mockRootCoord{}, nil
}
dnServer.newDataCoordClient = func(string, []string, ...retry.Option) (types.DataCoord, error) {
dnServer.newDataCoordClient = func(string, []string) (types.DataCoord, error) {
return &mockDataCoord{}, nil
}

17
internal/distributed/datanode/service.go
View File

@ -38,7 +38,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/milvuspb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/trace"
)
@ -55,8 +54,8 @@ type Server struct {
rootCoord types.RootCoord
dataCoord types.DataCoord
newRootCoordClient func(string, []string, ...retry.Option) (types.RootCoord, error)
newDataCoordClient func(string, []string, ...retry.Option) (types.DataCoord, error)
newRootCoordClient func(string, []string) (types.RootCoord, error)
newDataCoordClient func(string, []string) (types.DataCoord, error)
closer io.Closer
}
@ -69,11 +68,11 @@ func NewServer(ctx context.Context, factory msgstream.Factory) (*Server, error)
cancel: cancel,
msFactory: factory,
grpcErrChan: make(chan error),
newRootCoordClient: func(etcdMetaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) (types.RootCoord, error) {
return rcc.NewClient(ctx1, etcdMetaRoot, etcdEndpoints, retryOptions...)
newRootCoordClient: func(etcdMetaRoot string, etcdEndpoints []string) (types.RootCoord, error) {
return rcc.NewClient(ctx1, etcdMetaRoot, etcdEndpoints)
},
newDataCoordClient: func(etcdMetaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) (types.DataCoord, error) {
return dsc.NewClient(ctx1, etcdMetaRoot, etcdEndpoints, retryOptions...)
newDataCoordClient: func(etcdMetaRoot string, etcdEndpoints []string) (types.DataCoord, error) {
return dsc.NewClient(ctx1, etcdMetaRoot, etcdEndpoints)
},
}
@ -178,7 +177,7 @@ func (s *Server) init() error {
if s.newRootCoordClient != nil {
log.Debug("RootCoord address", zap.String("address", Params.RootCoordAddress))
log.Debug("Init root coord client ...")
rootCoordClient, err := s.newRootCoordClient(dn.Params.MetaRootPath, dn.Params.EtcdEndpoints, retry.Attempts(300))
rootCoordClient, err := s.newRootCoordClient(dn.Params.MetaRootPath, dn.Params.EtcdEndpoints)
if err != nil {
log.Debug("DataNode newRootCoordClient failed", zap.Error(err))
panic(err)
@ -206,7 +205,7 @@ func (s *Server) init() error {
if s.newDataCoordClient != nil {
log.Debug("Data service address", zap.String("address", Params.DataCoordAddress))
log.Debug("DataNode Init data service client ...")
dataCoordClient, err := s.newDataCoordClient(dn.Params.MetaRootPath, dn.Params.EtcdEndpoints, retry.Attempts(300))
dataCoordClient, err := s.newDataCoordClient(dn.Params.MetaRootPath, dn.Params.EtcdEndpoints)
if err != nil {
log.Debug("DataNode newDataCoordClient failed", zap.Error(err))
panic(err)

17
internal/distributed/indexcoord/client/client.go
View File

@ -43,8 +43,6 @@ type Client struct {
addr string
sess *sessionutil.Session
retryOptions []retry.Option
}
func getIndexCoordAddr(sess *sessionutil.Session) (string, error) {
@ -63,7 +61,7 @@ func getIndexCoordAddr(sess *sessionutil.Session) (string, error) {
return ms.Address, nil
}
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) (*Client, error) {
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string) (*Client, error) {
sess := sessionutil.NewSession(ctx, metaRoot, etcdEndpoints)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
@ -72,18 +70,17 @@ func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, ret
}
ctx, cancel := context.WithCancel(ctx)
return &Client{
ctx: ctx,
cancel: cancel,
sess: sess,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
sess: sess,
}, nil
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
var err error
connectIndexCoordaddrFn := func() error {
c.addr, err = getIndexCoordAddr(c.sess)
@ -113,7 +110,7 @@ func (c *Client) connect() error {
return nil
}
err = retry.Do(c.ctx, connectIndexCoordaddrFn, c.retryOptions...)
err = retry.Do(c.ctx, connectIndexCoordaddrFn, retryOptions...)
if err != nil {
log.Debug("IndexCoordClient try connect failed", zap.Error(err))
return err

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

@ -39,29 +39,26 @@ type Client struct {
conn *grpc.ClientConn
addr string
retryOptions []retry.Option
}
func NewClient(ctx context.Context, addr string, retryOptions ...retry.Option) (*Client, error) {
func NewClient(ctx context.Context, addr string) (*Client, error) {
if addr == "" {
return nil, fmt.Errorf("address is empty")
}
ctx, cancel := context.WithCancel(ctx)
return &Client{
ctx: ctx,
cancel: cancel,
addr: addr,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
addr: addr,
}, nil
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
connectGrpcFunc := func() error {
opts := trace.GetInterceptorOpts()
log.Debug("IndexNodeClient try connect ", zap.String("address", c.addr))
@ -85,7 +82,7 @@ func (c *Client) connect() error {
return nil
}
err := retry.Do(c.ctx, connectGrpcFunc, c.retryOptions...)
err := retry.Do(c.ctx, connectGrpcFunc, retryOptions...)
if err != nil {
log.Debug("IndexNodeClient try connect failed", zap.Error(err))
return err

3
internal/distributed/indexnode/service.go
View File

@ -32,7 +32,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/milvuspb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/trace"
"google.golang.org/grpc"
)
@ -136,7 +135,7 @@ func (s *Server) init() error {
return err
}
s.indexCoordClient, err = grpcindexcoordclient.NewClient(s.loopCtx, indexnode.Params.MetaRootPath, indexnode.Params.EtcdEndpoints, retry.Attempts(300))
s.indexCoordClient, err = grpcindexcoordclient.NewClient(s.loopCtx, indexnode.Params.MetaRootPath, indexnode.Params.EtcdEndpoints)
if err != nil {
log.Debug("New indexCoordeClient failed", zap.Error(err))
return err

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

@ -38,29 +38,26 @@ type Client struct {
conn *grpc.ClientConn
addr string
retryOptions []retry.Option
}
func NewClient(ctx context.Context, addr string, retryOptions ...retry.Option) (*Client, error) {
func NewClient(ctx context.Context, addr string) (*Client, error) {
if addr == "" {
return nil, fmt.Errorf("address is empty")
}
ctx, cancel := context.WithCancel(ctx)
return &Client{
ctx: ctx,
cancel: cancel,
addr: addr,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
addr: addr,
}, nil
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
connectGrpcFunc := func() error {
opts := trace.GetInterceptorOpts()
log.Debug("ProxyClient try connect ", zap.String("address", c.addr))
@ -84,7 +81,7 @@ func (c *Client) connect() error {
return nil
}
err := retry.Do(c.ctx, connectGrpcFunc, c.retryOptions...)
err := retry.Do(c.ctx, connectGrpcFunc, retryOptions...)
if err != nil {
log.Debug("ProxyClient try connect failed", zap.Error(err))
return err

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

@ -38,7 +38,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/proxypb"
"github.com/milvus-io/milvus/internal/proxy"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/trace"
"github.com/opentracing/opentracing-go"
)
@ -170,7 +169,7 @@ func (s *Server) init() error {
rootCoordAddr := Params.RootCoordAddress
log.Debug("Proxy", zap.String("RootCoord address", rootCoordAddr))
s.rootCoordClient, err = rcc.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints, retry.Attempts(300))
s.rootCoordClient, err = rcc.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints)
if err != nil {
log.Debug("Proxy new rootCoordClient failed ", zap.Error(err))
return err
@ -190,7 +189,7 @@ func (s *Server) init() error {
dataCoordAddr := Params.DataCoordAddress
log.Debug("Proxy", zap.String("data coordinator address", dataCoordAddr))
s.dataCoordClient, err = grpcdatacoordclient.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints, retry.Attempts(300))
s.dataCoordClient, err = grpcdatacoordclient.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints)
if err != nil {
log.Debug("Proxy new dataCoordClient failed ", zap.Error(err))
return err
@ -206,7 +205,7 @@ func (s *Server) init() error {
indexCoordAddr := Params.IndexCoordAddress
log.Debug("Proxy", zap.String("index coordinator address", indexCoordAddr))
s.indexCoordClient, err = grpcindexcoordclient.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints, retry.Attempts(300))
s.indexCoordClient, err = grpcindexcoordclient.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints)
if err != nil {
log.Debug("Proxy new indexCoordClient failed ", zap.Error(err))
return err
@ -222,7 +221,7 @@ func (s *Server) init() error {
queryCoordAddr := Params.QueryCoordAddress
log.Debug("Proxy", zap.String("query coordinator address", queryCoordAddr))
s.queryCooedClient, err = grpcquerycoordclient.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints, retry.Attempts(300))
s.queryCooedClient, err = grpcquerycoordclient.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints)
if err != nil {
return err
}

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

@ -42,8 +42,6 @@ type Client struct {
sess *sessionutil.Session
addr string
retryOptions []retry.Option
}
func getQueryCoordAddress(sess *sessionutil.Session) (string, error) {
@ -62,7 +60,7 @@ func getQueryCoordAddress(sess *sessionutil.Session) (string, error) {
}
// NewClient creates a client for QueryCoord grpc call.
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) (*Client, error) {
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string) (*Client, error) {
sess := sessionutil.NewSession(ctx, metaRoot, etcdEndpoints)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
@ -71,18 +69,17 @@ func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, ret
}
ctx, cancel := context.WithCancel(ctx)
return &Client{
ctx: ctx,
cancel: cancel,
sess: sess,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
sess: sess,
}, nil
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
var err error
connectQueryCoordAddressFn := func() error {
c.addr, err = getQueryCoordAddress(c.sess)
@ -112,7 +109,7 @@ func (c *Client) connect() error {
return nil
}
err = retry.Do(c.ctx, connectQueryCoordAddressFn, c.retryOptions...)
err = retry.Do(c.ctx, connectQueryCoordAddressFn, retryOptions...)
if err != nil {
log.Debug("QueryCoordClient try reconnect failed", zap.Error(err))
return err

5
internal/distributed/querycoord/service.go
View File

@ -28,7 +28,6 @@ import (
qc "github.com/milvus-io/milvus/internal/querycoord"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/trace"
"go.uber.org/zap"
"google.golang.org/grpc"
@ -108,7 +107,7 @@ func (s *Server) init() error {
// --- Master Server Client ---
log.Debug("QueryCoord try to new RootCoord client", zap.Any("RootCoordAddress", Params.RootCoordAddress))
rootCoord, err := rcc.NewClient(s.loopCtx, qc.Params.MetaRootPath, qc.Params.EtcdEndpoints, retry.Attempts(300))
rootCoord, err := rcc.NewClient(s.loopCtx, qc.Params.MetaRootPath, qc.Params.EtcdEndpoints)
if err != nil {
log.Debug("QueryCoord try to new RootCoord client failed", zap.Error(err))
panic(err)
@ -139,7 +138,7 @@ func (s *Server) init() error {
// --- Data service client ---
log.Debug("QueryCoord try to new DataCoord client", zap.Any("DataCoordAddress", Params.DataCoordAddress))
dataCoord, err := dsc.NewClient(s.loopCtx, qc.Params.MetaRootPath, qc.Params.EtcdEndpoints, retry.Attempts(300))
dataCoord, err := dsc.NewClient(s.loopCtx, qc.Params.MetaRootPath, qc.Params.EtcdEndpoints)
if err != nil {
log.Debug("QueryCoord try to new DataCoord client failed", zap.Error(err))
panic(err)

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

@ -39,29 +39,26 @@ type Client struct {
conn *grpc.ClientConn
addr string
retryOptions []retry.Option
}
func NewClient(ctx context.Context, addr string, retryOptions ...retry.Option) (*Client, error) {
func NewClient(ctx context.Context, addr string) (*Client, error) {
if addr == "" {
return nil, fmt.Errorf("addr is empty")
}
ctx, cancel := context.WithCancel(ctx)
return &Client{
ctx: ctx,
cancel: cancel,
addr: addr,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
addr: addr,
}, nil
}
func (c *Client) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *Client) connect() error {
func (c *Client) connect(retryOptions ...retry.Option) error {
connectGrpcFunc := func() error {
opts := trace.GetInterceptorOpts()
log.Debug("QueryNodeClient try connect ", zap.String("address", c.addr))
@ -85,7 +82,7 @@ func (c *Client) connect() error {
return nil
}
err := retry.Do(c.ctx, connectGrpcFunc, c.retryOptions...)
err := retry.Do(c.ctx, connectGrpcFunc, retryOptions...)
if err != nil {
log.Debug("QueryNodeClient try connect failed", zap.Error(err))
return err

8
internal/distributed/querynode/service.go
View File

@ -103,7 +103,7 @@ func (s *Server) init() error {
}
// --- QueryCoord ---
log.Debug("QueryNode start to new QueryCoordClient", zap.Any("QueryCoordAddress", Params.QueryCoordAddress))
queryCoord, err := qcc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints, retry.Attempts(300))
queryCoord, err := qcc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints)
if err != nil {
log.Debug("QueryNode new QueryCoordClient failed", zap.Error(err))
panic(err)
@ -136,7 +136,7 @@ func (s *Server) init() error {
addr := Params.RootCoordAddress
log.Debug("QueryNode start to new RootCoordClient", zap.Any("QueryCoordAddress", addr))
rootCoord, err := rcc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints, retry.Attempts(300))
rootCoord, err := rcc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints)
if err != nil {
log.Debug("QueryNode new RootCoordClient failed", zap.Error(err))
panic(err)
@ -165,7 +165,7 @@ func (s *Server) init() error {
// --- IndexCoord ---
log.Debug("Index coord", zap.String("address", Params.IndexCoordAddress))
indexCoord, err := isc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints, retry.Attempts(300))
indexCoord, err := isc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints)
if err != nil {
log.Debug("QueryNode new IndexCoordClient failed", zap.Error(err))
@ -196,7 +196,7 @@ func (s *Server) init() error {
// --- DataCoord ---
log.Debug("QueryNode start to new DataCoordClient", zap.Any("DataCoordAddress", Params.DataCoordAddress))
dataCoord, err := dsc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints, retry.Attempts(300))
dataCoord, err := dsc.NewClient(s.ctx, qn.Params.MetaRootPath, qn.Params.EtcdEndpoints)
if err != nil {
log.Debug("QueryNode new DataCoordClient failed", zap.Error(err))
panic(err)

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

@ -43,8 +43,6 @@ type GrpcClient struct {
sess *sessionutil.Session
addr string
retryOptions []retry.Option
}
func getRootCoordAddr(sess *sessionutil.Session) (string, error) {
@ -68,7 +66,7 @@ func getRootCoordAddr(sess *sessionutil.Session) (string, error) {
// metaRoot is the path in etcd for root coordinator registration
// etcdEndpoints are the address list for etcd end points
// timeout is default setting for each grpc call
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) (*GrpcClient, error) {
func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string) (*GrpcClient, error) {
sess := sessionutil.NewSession(ctx, metaRoot, etcdEndpoints)
if sess == nil {
err := fmt.Errorf("new session error, maybe can not connect to etcd")
@ -78,18 +76,17 @@ func NewClient(ctx context.Context, metaRoot string, etcdEndpoints []string, ret
ctx, cancel := context.WithCancel(ctx)
return &GrpcClient{
ctx: ctx,
cancel: cancel,
sess: sess,
retryOptions: retryOptions,
ctx: ctx,
cancel: cancel,
sess: sess,
}, nil
}
func (c *GrpcClient) Init() error {
return c.connect()
return c.connect(retry.Attempts(300))
}
func (c *GrpcClient) connect() error {
func (c *GrpcClient) connect(retryOptions ...retry.Option) error {
var err error
connectRootCoordAddrFn := func() error {
c.addr, err = getRootCoordAddr(c.sess)
@ -119,7 +116,7 @@ func (c *GrpcClient) connect() error {
return nil
}
err = retry.Do(c.ctx, connectRootCoordAddrFn, c.retryOptions...)
err = retry.Do(c.ctx, connectRootCoordAddrFn, retryOptions...)
if err != nil {
log.Debug("RootCoordClient try reconnect failed", zap.Error(err))
return err

27
internal/distributed/rootcoord/service.go
View File

@ -38,7 +38,6 @@ import (
"github.com/milvus-io/milvus/internal/rootcoord"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/retry"
"github.com/milvus-io/milvus/internal/util/sessionutil"
"github.com/milvus-io/milvus/internal/util/trace"
)
@ -58,9 +57,9 @@ type Server struct {
indexCoord types.IndexCoord
queryCoord types.QueryCoord
newIndexCoordClient func(string, []string, ...retry.Option) types.IndexCoord
newDataCoordClient func(string, []string, ...retry.Option) types.DataCoord
newQueryCoordClient func(string, []string, ...retry.Option) types.QueryCoord
newIndexCoordClient func(string, []string) types.IndexCoord
newDataCoordClient func(string, []string) types.DataCoord
newQueryCoordClient func(string, []string) types.QueryCoord
closer io.Closer
}
@ -84,8 +83,8 @@ func NewServer(ctx context.Context, factory msgstream.Factory) (*Server, error)
func (s *Server) setClient() {
ctx := context.Background()
s.newDataCoordClient = func(etcdMetaRoot string, etcdEndpoints []string, retryOptions ...retry.Option) types.DataCoord {
dsClient, err := dsc.NewClient(s.ctx, etcdMetaRoot, etcdEndpoints, retryOptions...)
s.newDataCoordClient = func(etcdMetaRoot string, etcdEndpoints []string) types.DataCoord {
dsClient, err := dsc.NewClient(s.ctx, etcdMetaRoot, etcdEndpoints)
if err != nil {
panic(err)
}
@ -100,8 +99,8 @@ func (s *Server) setClient() {
}
return dsClient
}
s.newIndexCoordClient = func(metaRootPath string, etcdEndpoints []string, retryOptions ...retry.Option) types.IndexCoord {
isClient, err := isc.NewClient(s.ctx, metaRootPath, etcdEndpoints, retryOptions...)
s.newIndexCoordClient = func(metaRootPath string, etcdEndpoints []string) types.IndexCoord {
isClient, err := isc.NewClient(s.ctx, metaRootPath, etcdEndpoints)
if err != nil {
panic(err)
}
@ -113,8 +112,8 @@ func (s *Server) setClient() {
}
return isClient
}
s.newQueryCoordClient = func(metaRootPath string, etcdEndpoints []string, retryOptions ...retry.Option) types.QueryCoord {
qsClient, err := qsc.NewClient(s.ctx, metaRootPath, etcdEndpoints, retryOptions...)
s.newQueryCoordClient = func(metaRootPath string, etcdEndpoints []string) types.QueryCoord {
qsClient, err := qsc.NewClient(s.ctx, metaRootPath, etcdEndpoints)
if err != nil {
panic(err)
}
@ -165,7 +164,7 @@ func (s *Server) init() error {
log.Debug("RootCoord", zap.Any("State", internalpb.StateCode_Initializing))
s.rootCoord.SetNewProxyClient(
func(se *sessionutil.Session) (types.Proxy, error) {
cli, err := pnc.NewClient(s.ctx, se.Address, retry.Attempts(300))
cli, err := pnc.NewClient(s.ctx, se.Address)
if err != nil {
return nil, err
}
@ -181,7 +180,7 @@ func (s *Server) init() error {
if s.newDataCoordClient != nil {
log.Debug("RootCoord start to create DataCoord client")
dataCoord := s.newDataCoordClient(rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints, retry.Attempts(300))
dataCoord := s.newDataCoordClient(rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints)
if err := s.rootCoord.SetDataCoord(s.ctx, dataCoord); err != nil {
panic(err)
}
@ -189,7 +188,7 @@ func (s *Server) init() error {
}
if s.newIndexCoordClient != nil {
log.Debug("RootCoord start to create IndexCoord client")
indexCoord := s.newIndexCoordClient(rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints, retry.Attempts(300))
indexCoord := s.newIndexCoordClient(rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints)
if err := s.rootCoord.SetIndexCoord(indexCoord); err != nil {
panic(err)
}
@ -197,7 +196,7 @@ func (s *Server) init() error {
}
if s.newQueryCoordClient != nil {
log.Debug("RootCoord start to create QueryCoord client")
queryCoord := s.newQueryCoordClient(rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints, retry.Attempts(300))
queryCoord := s.newQueryCoordClient(rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints)
if err := s.rootCoord.SetQueryCoord(queryCoord); err != nil {
panic(err)
}

8
internal/distributed/rootcoord/service_test.go
View File

@ -258,7 +258,7 @@ func TestGrpcService(t *testing.T) {
svr.rootCoord.UpdateStateCode(internalpb.StateCode_Healthy)
cli, err := rcc.NewClient(context.Background(), rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints, retry.Attempts(300))
cli, err := rcc.NewClient(context.Background(), rootcoord.Params.MetaRootPath, rootcoord.Params.EtcdEndpoints)
assert.Nil(t, err)
err = cli.Init()
@ -925,13 +925,13 @@ func TestRun(t *testing.T) {
assert.NotNil(t, err)
assert.EqualError(t, err, "listen tcp: address 1000000: invalid port")
svr.newDataCoordClient = func(string, []string, ...retry.Option) types.DataCoord {
svr.newDataCoordClient = func(string, []string) types.DataCoord {
return &mockDataCoord{}
}
svr.newIndexCoordClient = func(string, []string, ...retry.Option) types.IndexCoord {
svr.newIndexCoordClient = func(string, []string) types.IndexCoord {
return &mockIndex{}
}
svr.newQueryCoordClient = func(string, []string, ...retry.Option) types.QueryCoord {
svr.newQueryCoordClient = func(string, []string) types.QueryCoord {
return &mockQuery{}
}

3
internal/indexcoord/node_mgr.go
View File

@ -22,7 +22,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/commonpb"
"github.com/milvus-io/milvus/internal/proto/indexpb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/util/retry"
)
func (i *IndexCoord) removeNode(nodeID UniqueID) {
@ -44,7 +43,7 @@ func (i *IndexCoord) addNode(nodeID UniqueID, req *indexpb.RegisterNodeRequest)
}
nodeAddress := req.Address.Ip + ":" + strconv.FormatInt(req.Address.Port, 10)
nodeClient, err := grpcindexnodeclient.NewClient(context.TODO(), nodeAddress, retry.Attempts(300))
nodeClient, err := grpcindexnodeclient.NewClient(context.TODO(), nodeAddress)
if err != nil {
return err
}

3
internal/querycoord/querynode.go
View File

@ -26,7 +26,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/querypb"
"github.com/milvus-io/milvus/internal/proto/schemapb"
"github.com/milvus-io/milvus/internal/types"
"github.com/milvus-io/milvus/internal/util/retry"
)
type queryNode struct {
@ -42,7 +41,7 @@ type queryNode struct {
}
func newQueryNode(ctx context.Context, address string, id UniqueID, kv *etcdkv.EtcdKV) (*queryNode, error) {
client, err := nodeclient.NewClient(ctx, address, retry.Attempts(300))
client, err := nodeclient.NewClient(ctx, address)
if err != nil {
return nil, err
}