-
Notifications
You must be signed in to change notification settings - Fork 3k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
enhance: remove the rpc layer of coordinator when enabling standalone…
… or mixcoord (#38207) issue: #37764 pr: #37815 also see: #38259 - add a local client to call local server directly for querycoord/rootcoord/datacoord. - enable local client if milvus is running mixcoord or standalone mode. - after removing rpc layer from mixcoord, the querycoord at standby mode will be blocked forever of deployment rolling --------- Signed-off-by: chyezh <[email protected]>
- Loading branch information
Showing
18 changed files
with
548 additions
and
65 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,162 @@ | ||
| package coordclient | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "go.uber.org/zap" | ||
|
|
||
| dcc "github.com/milvus-io/milvus/internal/distributed/datacoord/client" | ||
| qcc "github.com/milvus-io/milvus/internal/distributed/querycoord/client" | ||
| rcc "github.com/milvus-io/milvus/internal/distributed/rootcoord/client" | ||
| "github.com/milvus-io/milvus/internal/proto/datapb" | ||
| "github.com/milvus-io/milvus/internal/proto/querypb" | ||
| "github.com/milvus-io/milvus/internal/proto/rootcoordpb" | ||
| "github.com/milvus-io/milvus/internal/types" | ||
| "github.com/milvus-io/milvus/internal/util/grpcclient" | ||
| "github.com/milvus-io/milvus/pkg/log" | ||
| "github.com/milvus-io/milvus/pkg/util/paramtable" | ||
| "github.com/milvus-io/milvus/pkg/util/syncutil" | ||
| "github.com/milvus-io/milvus/pkg/util/typeutil" | ||
| ) | ||
|
|
||
| // localClient is a client that can access local server directly | ||
| type localClient struct { | ||
| queryCoordClient *syncutil.Future[types.QueryCoordClient] | ||
| dataCoordClient *syncutil.Future[types.DataCoordClient] | ||
| rootCoordClient *syncutil.Future[types.RootCoordClient] | ||
| } | ||
|
|
||
| var ( | ||
| enableLocal *LocalClientRoleConfig // a global map to store all can be local accessible roles. | ||
| glocalClient *localClient // !!! WARNING: local client will ignore all interceptor of grpc client and server. | ||
| ) | ||
|
|
||
| func init() { | ||
| enableLocal = &LocalClientRoleConfig{} | ||
| glocalClient = &localClient{ | ||
| queryCoordClient: syncutil.NewFuture[types.QueryCoordClient](), | ||
| dataCoordClient: syncutil.NewFuture[types.DataCoordClient](), | ||
| rootCoordClient: syncutil.NewFuture[types.RootCoordClient](), | ||
| } | ||
| } | ||
|
|
||
| type LocalClientRoleConfig struct { | ||
| ServerType string | ||
| EnableQueryCoord bool | ||
| EnableDataCoord bool | ||
| EnableRootCoord bool | ||
| } | ||
|
|
||
| // EnableLocalClientRole init localable roles | ||
| func EnableLocalClientRole(cfg *LocalClientRoleConfig) { | ||
| if !paramtable.Get().CommonCfg.LocalRPCEnabled.GetAsBool() { | ||
| return | ||
| } | ||
| if cfg.ServerType != typeutil.StandaloneRole && cfg.ServerType != typeutil.MixtureRole { | ||
| return | ||
| } | ||
| enableLocal = cfg | ||
| } | ||
|
|
||
| // RegisterQueryCoordServer register query coord server | ||
| func RegisterQueryCoordServer(server querypb.QueryCoordServer) { | ||
| if !enableLocal.EnableQueryCoord { | ||
| return | ||
| } | ||
| newLocalClient := grpcclient.NewLocalGRPCClient(&querypb.QueryCoord_ServiceDesc, server, querypb.NewQueryCoordClient) | ||
| glocalClient.queryCoordClient.Set(&nopCloseQueryCoordClient{newLocalClient}) | ||
| log.Info("register query coord server", zap.Any("enableLocalClient", enableLocal)) | ||
| } | ||
|
|
||
| // RegsterDataCoordServer register data coord server | ||
| func RegisterDataCoordServer(server datapb.DataCoordServer) { | ||
| if !enableLocal.EnableDataCoord { | ||
| return | ||
| } | ||
| newLocalClient := grpcclient.NewLocalGRPCClient(&datapb.DataCoord_ServiceDesc, server, datapb.NewDataCoordClient) | ||
| glocalClient.dataCoordClient.Set(&nopCloseDataCoordClient{newLocalClient}) | ||
| log.Info("register data coord server", zap.Any("enableLocalClient", enableLocal)) | ||
| } | ||
|
|
||
| // RegisterRootCoordServer register root coord server | ||
| func RegisterRootCoordServer(server rootcoordpb.RootCoordServer) { | ||
| if !enableLocal.EnableRootCoord { | ||
| return | ||
| } | ||
| newLocalClient := grpcclient.NewLocalGRPCClient(&rootcoordpb.RootCoord_ServiceDesc, server, rootcoordpb.NewRootCoordClient) | ||
| glocalClient.rootCoordClient.Set(&nopCloseRootCoordClient{newLocalClient}) | ||
| log.Info("register root coord server", zap.Any("enableLocalClient", enableLocal)) | ||
| } | ||
|
|
||
| // GetQueryCoordClient return query coord client | ||
| func GetQueryCoordClient(ctx context.Context) types.QueryCoordClient { | ||
| var client types.QueryCoordClient | ||
| var err error | ||
| if enableLocal.EnableQueryCoord { | ||
| client, err = glocalClient.queryCoordClient.GetWithContext(ctx) | ||
| } else { | ||
| // TODO: we should make a singleton here. but most unittest rely on a dedicated client. | ||
| client, err = qcc.NewClient(ctx) | ||
| } | ||
| if err != nil { | ||
| panic(fmt.Sprintf("get query coord client failed: %v", err)) | ||
| } | ||
| return client | ||
| } | ||
|
|
||
| // GetDataCoordClient return data coord client | ||
| func GetDataCoordClient(ctx context.Context) types.DataCoordClient { | ||
| var client types.DataCoordClient | ||
| var err error | ||
| if enableLocal.EnableDataCoord { | ||
| client, err = glocalClient.dataCoordClient.GetWithContext(ctx) | ||
| } else { | ||
| // TODO: we should make a singleton here. but most unittest rely on a dedicated client. | ||
| client, err = dcc.NewClient(ctx) | ||
| } | ||
| if err != nil { | ||
| panic(fmt.Sprintf("get data coord client failed: %v", err)) | ||
| } | ||
| return client | ||
| } | ||
|
|
||
| // GetRootCoordClient return root coord client | ||
| func GetRootCoordClient(ctx context.Context) types.RootCoordClient { | ||
| var client types.RootCoordClient | ||
| var err error | ||
| if enableLocal.EnableRootCoord { | ||
| client, err = glocalClient.rootCoordClient.GetWithContext(ctx) | ||
| } else { | ||
| // TODO: we should make a singleton here. but most unittest rely on a dedicated client. | ||
| client, err = rcc.NewClient(ctx) | ||
| } | ||
| if err != nil { | ||
| panic(fmt.Sprintf("get root coord client failed: %v", err)) | ||
| } | ||
| return client | ||
| } | ||
|
|
||
| type nopCloseQueryCoordClient struct { | ||
| querypb.QueryCoordClient | ||
| } | ||
|
|
||
| func (n *nopCloseQueryCoordClient) Close() error { | ||
| return nil | ||
| } | ||
|
|
||
| type nopCloseDataCoordClient struct { | ||
| datapb.DataCoordClient | ||
| } | ||
|
|
||
| func (n *nopCloseDataCoordClient) Close() error { | ||
| return nil | ||
| } | ||
|
|
||
| type nopCloseRootCoordClient struct { | ||
| rootcoordpb.RootCoordClient | ||
| } | ||
|
|
||
| func (n *nopCloseRootCoordClient) Close() error { | ||
| return nil | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| package coordclient | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/stretchr/testify/assert" | ||
|
|
||
| "github.com/milvus-io/milvus/internal/proto/datapb" | ||
| "github.com/milvus-io/milvus/internal/proto/querypb" | ||
| "github.com/milvus-io/milvus/internal/proto/rootcoordpb" | ||
| "github.com/milvus-io/milvus/pkg/util/paramtable" | ||
| "github.com/milvus-io/milvus/pkg/util/typeutil" | ||
| ) | ||
|
|
||
| func TestRegistry(t *testing.T) { | ||
| paramtable.Init() | ||
| paramtable.Get().Save(paramtable.Get().CommonCfg.LocalRPCEnabled.Key, "true") | ||
|
|
||
| assert.False(t, enableLocal.EnableQueryCoord) | ||
| assert.False(t, enableLocal.EnableDataCoord) | ||
| assert.False(t, enableLocal.EnableRootCoord) | ||
|
|
||
| EnableLocalClientRole(&LocalClientRoleConfig{ | ||
| ServerType: typeutil.RootCoordRole, | ||
| EnableQueryCoord: true, | ||
| EnableDataCoord: true, | ||
| EnableRootCoord: true, | ||
| }) | ||
| assert.False(t, enableLocal.EnableQueryCoord) | ||
| assert.False(t, enableLocal.EnableDataCoord) | ||
| assert.False(t, enableLocal.EnableRootCoord) | ||
|
|
||
| RegisterRootCoordServer(&rootcoordpb.UnimplementedRootCoordServer{}) | ||
| RegisterDataCoordServer(&datapb.UnimplementedDataCoordServer{}) | ||
| RegisterQueryCoordServer(&querypb.UnimplementedQueryCoordServer{}) | ||
| assert.False(t, glocalClient.dataCoordClient.Ready()) | ||
| assert.False(t, glocalClient.queryCoordClient.Ready()) | ||
| assert.False(t, glocalClient.rootCoordClient.Ready()) | ||
|
|
||
| enableLocal = &LocalClientRoleConfig{} | ||
|
|
||
| EnableLocalClientRole(&LocalClientRoleConfig{ | ||
| ServerType: typeutil.StandaloneRole, | ||
| EnableQueryCoord: true, | ||
| EnableDataCoord: true, | ||
| EnableRootCoord: true, | ||
| }) | ||
| assert.True(t, enableLocal.EnableDataCoord) | ||
| assert.True(t, enableLocal.EnableQueryCoord) | ||
| assert.True(t, enableLocal.EnableRootCoord) | ||
|
|
||
| RegisterRootCoordServer(&rootcoordpb.UnimplementedRootCoordServer{}) | ||
| RegisterDataCoordServer(&datapb.UnimplementedDataCoordServer{}) | ||
| RegisterQueryCoordServer(&querypb.UnimplementedQueryCoordServer{}) | ||
| assert.True(t, glocalClient.dataCoordClient.Ready()) | ||
| assert.True(t, glocalClient.queryCoordClient.Ready()) | ||
| assert.True(t, glocalClient.rootCoordClient.Ready()) | ||
|
|
||
| enableLocal = &LocalClientRoleConfig{} | ||
|
|
||
| EnableLocalClientRole(&LocalClientRoleConfig{ | ||
| ServerType: typeutil.MixtureRole, | ||
| EnableQueryCoord: true, | ||
| EnableDataCoord: true, | ||
| EnableRootCoord: true, | ||
| }) | ||
| assert.True(t, enableLocal.EnableDataCoord) | ||
| assert.True(t, enableLocal.EnableQueryCoord) | ||
| assert.True(t, enableLocal.EnableRootCoord) | ||
|
|
||
| assert.NotNil(t, GetQueryCoordClient(context.Background())) | ||
| assert.NotNil(t, GetDataCoordClient(context.Background())) | ||
| assert.NotNil(t, GetRootCoordClient(context.Background())) | ||
| GetQueryCoordClient(context.Background()).Close() | ||
| GetDataCoordClient(context.Background()).Close() | ||
| GetRootCoordClient(context.Background()).Close() | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.