Update component names in config files (#5989)

* rename master/indexService in .yaml

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* rename proxyNode to proxy in config files

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* move master.yaml to root_coord.yaml, move proxy_node.yaml to proxy.yaml

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* move proto/query_service.proto to proto/query_coord.proto

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* move proxy_node_manager.go to proxy_manager.go

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* add query_coord.pb.go

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* update ci

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>
pull/5996/head
Cai Yudong 2021-06-22 19:08:03 +08:00 committed by GitHub
parent 792dba4ae3
commit 7dbf33b788
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
45 changed files with 392 additions and 397 deletions

View File

@ -35,7 +35,7 @@ pipeline {
IMAGE_REPO = "dockerhub-mirror-sh.zilliz.cc/milvusdb"
DOCKER_BUILDKIT = 1
ARTIFACTS = "${env.WORKSPACE}/artifacts"
MILVUS_HELM_BRANCH = "querycoord"
MILVUS_HELM_BRANCH = "rename"
}
stages {
stage('Test') {

View File

@ -44,7 +44,7 @@ pipeline {
DOCKER_CREDENTIALS_ID = "ba070c98-c8cc-4f7c-b657-897715f359fc"
DOKCER_REGISTRY_URL = "registry.zilliz.com"
TARGET_REPO = "${DOKCER_REGISTRY_URL}/milvus"
MILVUS_HELM_BRANCH = "querycoord"
MILVUS_HELM_BRANCH = "rename"
}
stages {
stage('Test') {

View File

@ -9,7 +9,7 @@
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under the License.
proxyNode:
proxy:
timeTickInterval: 200 # ms
msgStream:

View File

@ -34,11 +34,11 @@ pulsar:
port: 6650
maxMessageSize: 5242880 # 5 * 1024 * 1024 Bytes
master:
rootCoord:
address: localhost
port: 53100
proxyNode:
proxy:
port: 19530
queryCoord:
@ -49,7 +49,7 @@ queryNode:
gracefulTime: 1000 # ms, for search
port: 21123
indexService:
indexCoord:
address: localhost
port: 31000

View File

@ -110,7 +110,7 @@ func NewDataNode(ctx context.Context, factory msgstream.Factory) *DataNode {
return node
}
// SetRootCoordInterface sets master service's grpc client, error is returned if repeatedly set.
// SetRootCoordInterface sets RootCoord's grpc client, error is returned if repeatedly set.
func (node *DataNode) SetRootCoordInterface(rc types.RootCoord) error {
switch {
case rc == nil, node.rootCoord != nil:

View File

@ -20,9 +20,9 @@ import (
type ParamTable struct {
paramtable.BaseTable
IP string
Port int
MasterAddress string
IP string
Port int
RootCoordAddress string
}
var Params ParamTable
@ -38,7 +38,7 @@ func (pt *ParamTable) Init() {
}
func (pt *ParamTable) initParams() {
pt.initMasterAddress()
pt.initRootCoordAddress()
pt.initDataCoordAddress()
}
@ -50,12 +50,12 @@ func (pt *ParamTable) initPort() {
pt.Port = pt.ParseInt("dataCoord.port")
}
func (pt *ParamTable) initMasterAddress() {
ret, err := pt.Load("_MasterAddress")
func (pt *ParamTable) initRootCoordAddress() {
ret, err := pt.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
pt.MasterAddress = ret
pt.RootCoordAddress = ret
}
func (pt *ParamTable) initDataCoordAddress() {

View File

@ -23,6 +23,6 @@ func TestParamTable(t *testing.T) {
assert.NotEqual(t, Params.Port, 0)
t.Logf("DataCoord Port:%d", Params.Port)
assert.NotEqual(t, Params.MasterAddress, "")
t.Logf("MasterAddress:%s", Params.MasterAddress)
assert.NotEqual(t, Params.RootCoordAddress, "")
t.Logf("RootCoordAddress:%s", Params.RootCoordAddress)
}

View File

@ -31,14 +31,14 @@ type ParamTable struct {
Port int
listener net.Listener
MasterAddress string
RootCoordAddress string
DataCoordAddress string
}
func (pt *ParamTable) Init() {
once.Do(func() {
pt.BaseTable.Init()
pt.initMasterAddress()
pt.initRootCoordAddress()
pt.initDataCoordAddress()
pt.initPort()
})
@ -64,12 +64,12 @@ func (pt *ParamTable) initPort() {
log.Info("DataNode", zap.Int("port", pt.Port))
}
func (pt *ParamTable) initMasterAddress() {
ret, err := pt.Load("_MasterAddress")
func (pt *ParamTable) initRootCoordAddress() {
ret, err := pt.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
pt.MasterAddress = ret
pt.RootCoordAddress = ret
}
func (pt *ParamTable) initDataCoordAddress() {

View File

@ -33,6 +33,6 @@ func TestParamTable(t *testing.T) {
assert.NotEqual(t, Params.DataCoordAddress, "")
t.Logf("DataCoordAddress:%s", Params.DataCoordAddress)
assert.NotEqual(t, Params.MasterAddress, "")
t.Logf("MasterAddress:%s", Params.MasterAddress)
assert.NotEqual(t, Params.RootCoordAddress, "")
t.Logf("RootCoordAddress:%s", Params.RootCoordAddress)
}

View File

@ -173,9 +173,9 @@ func (s *Server) init() error {
return err
}
// --- Master Server Client ---
// --- RootCoord Client ---
if s.newRootCoordClient != nil {
log.Debug("RootCoord address", zap.String("address", Params.MasterAddress))
log.Debug("RootCoord address", zap.String("address", Params.RootCoordAddress))
log.Debug("Init root coord client ...")
rootCoordClient, err := s.newRootCoordClient()
if err != nil {

View File

@ -40,11 +40,11 @@ func (pt *ParamTable) initParams() {
}
func (pt *ParamTable) initServicePort() {
pt.ServicePort = pt.ParseInt("indexService.port")
pt.ServicePort = pt.ParseInt("indexCoord.port")
}
func (pt *ParamTable) initServiceAddress() {
ret, err := pt.Load("IndexServiceAddress")
ret, err := pt.Load("_IndexCoordAddress")
if err != nil {
panic(err)
}

View File

@ -21,7 +21,7 @@ import (
type ParamTable struct {
paramtable.BaseTable
IndexServerAddress string
IndexCoordAddress string
IP string
Port int
@ -48,16 +48,16 @@ func (pt *ParamTable) LoadFromEnv() {
func (pt *ParamTable) initParams() {
pt.initPort()
pt.initIndexServerAddress()
pt.initIndexCoordAddress()
}
// todo remove and use load from env
func (pt *ParamTable) initIndexServerAddress() {
ret, err := pt.Load("IndexServiceAddress")
func (pt *ParamTable) initIndexCoordAddress() {
ret, err := pt.Load("_IndexCoordAddress")
if err != nil {
panic(err)
}
pt.IndexServerAddress = ret
pt.IndexCoordAddress = ret
}
func (pt *ParamTable) initPort() {

View File

@ -64,7 +64,7 @@ func (c *Client) connect() error {
ctx, cancelFunc := context.WithTimeout(c.ctx, c.timeout)
defer cancelFunc()
opts := trace.GetInterceptorOpts()
log.Debug("ProxyNodeClient try connect ", zap.String("address", c.addr))
log.Debug("ProxyClient try connect ", zap.String("address", c.addr))
conn, err := grpc.DialContext(ctx, c.addr, grpc.WithInsecure(), grpc.WithBlock(),
grpc.WithUnaryInterceptor(
grpc_middleware.ChainUnaryClient(
@ -86,10 +86,10 @@ func (c *Client) connect() error {
err := retry.Retry(c.reconnTry, 500*time.Millisecond, connectGrpcFunc)
if err != nil {
log.Debug("ProxyNodeClient try connect failed", zap.Error(err))
log.Debug("ProxyClient try connect failed", zap.Error(err))
return err
}
log.Debug("ProxyNodeClient connect success")
log.Debug("ProxyClient connect success")
c.grpcClient = proxypb.NewProxyClient(c.conn)
return nil
}

View File

@ -21,9 +21,8 @@ import (
type ParamTable struct {
paramtable.BaseTable
IndexServerAddress string
MasterAddress string
RootCoordAddress string
IndexCoordAddress string
DataCoordAddress string
QueryCoordAddress string
@ -52,28 +51,28 @@ func (pt *ParamTable) LoadFromEnv() {
func (pt *ParamTable) initParams() {
pt.initPort()
pt.initMasterAddress()
pt.initIndexServerAddress()
pt.initRootCoordAddress()
pt.initIndexCoordAddress()
pt.initDataCoordAddress()
pt.initQueryCoordAddress()
}
// todo remove and use load from env
func (pt *ParamTable) initIndexServerAddress() {
ret, err := pt.Load("IndexServiceAddress")
func (pt *ParamTable) initIndexCoordAddress() {
ret, err := pt.Load("_IndexCoordAddress")
if err != nil {
panic(err)
}
pt.IndexServerAddress = ret
pt.IndexCoordAddress = ret
}
// todo remove and use load from env
func (pt *ParamTable) initMasterAddress() {
ret, err := pt.Load("_MasterAddress")
func (pt *ParamTable) initRootCoordAddress() {
ret, err := pt.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
pt.MasterAddress = ret
pt.RootCoordAddress = ret
}
// todo remove and use load from env
@ -95,6 +94,6 @@ func (pt *ParamTable) initQueryCoordAddress() {
}
func (pt *ParamTable) initPort() {
port := pt.ParseInt("proxyNode.port")
port := pt.ParseInt("proxy.port")
pt.Port = port
}

View File

@ -143,9 +143,9 @@ func (s *Server) init() error {
proxy.Params.IP = Params.IP
proxy.Params.NetworkAddress = Params.Address
// for purpose of ID Allocator
proxy.Params.MasterAddress = Params.MasterAddress
proxy.Params.RootCoordAddress = Params.RootCoordAddress
closer := trace.InitTracing(fmt.Sprintf("proxy_node ip: %s, port: %d", Params.IP, Params.Port))
closer := trace.InitTracing(fmt.Sprintf("proxy ip: %s, port: %d", Params.IP, Params.Port))
s.closer = closer
log.Debug("proxy", zap.String("proxy host", Params.IP))
@ -167,7 +167,7 @@ func (s *Server) init() error {
return err
}
rootCoordAddr := Params.MasterAddress
rootCoordAddr := Params.RootCoordAddress
log.Debug("Proxy", zap.String("RootCoord address", rootCoordAddr))
timeout := 3 * time.Second
s.rootCoordClient, err = rcc.NewClient(s.ctx, proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints, timeout)
@ -200,8 +200,8 @@ func (s *Server) init() error {
s.proxy.SetDataCoordClient(s.dataCoordClient)
log.Debug("set data coordinator address ...")
indexServiceAddr := Params.IndexServerAddress
log.Debug("Proxy", zap.String("index coordinator address", indexServiceAddr))
indexCoordAddr := Params.IndexCoordAddress
log.Debug("Proxy", zap.String("index coordinator address", indexCoordAddr))
s.indexCoordClient = grpcindexcoordclient.NewClient(proxy.Params.MetaRootPath, proxy.Params.EtcdEndpoints, timeout)
err = s.indexCoordClient.Init()

View File

@ -24,36 +24,35 @@ type ParamTable struct {
paramtable.BaseTable
Port int
IndexServiceAddress string
MasterAddress string
DataCoordAddress string
IndexCoordAddress string
RootCoordAddress string
DataCoordAddress string
}
func (pt *ParamTable) Init() {
once.Do(func() {
pt.BaseTable.Init()
pt.initPort()
pt.initMasterAddress()
pt.initIndexServiceAddress()
pt.initRootCoordAddress()
pt.initIndexCoordAddress()
pt.initDataCoordAddress()
})
}
func (pt *ParamTable) initMasterAddress() {
ret, err := pt.Load("_MasterAddress")
func (pt *ParamTable) initRootCoordAddress() {
ret, err := pt.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
pt.MasterAddress = ret
pt.RootCoordAddress = ret
}
func (pt *ParamTable) initIndexServiceAddress() {
ret, err := pt.Load("IndexServiceAddress")
func (pt *ParamTable) initIndexCoordAddress() {
ret, err := pt.Load("IndexCoordAddress")
if err != nil {
panic(err)
}
pt.IndexServiceAddress = ret
pt.IndexCoordAddress = ret
}
func (pt *ParamTable) initDataCoordAddress() {

View File

@ -20,13 +20,12 @@ import (
func TestParamTable(t *testing.T) {
Params.Init()
assert.NotEqual(t, Params.IndexServiceAddress, "")
t.Logf("IndexServiceAddress:%s", Params.IndexServiceAddress)
assert.NotEqual(t, Params.IndexCoordAddress, "")
t.Logf("IndexCoordAddress:%s", Params.IndexCoordAddress)
assert.NotEqual(t, Params.DataCoordAddress, "")
t.Logf("DataCoordAddress:%s", Params.DataCoordAddress)
assert.NotEqual(t, Params.MasterAddress, "")
t.Logf("MasterAddress:%s", Params.MasterAddress)
assert.NotEqual(t, Params.RootCoordAddress, "")
t.Logf("RootCoordAddress:%s", Params.RootCoordAddress)
}

View File

@ -106,7 +106,7 @@ func (s *Server) init() error {
}
// --- Master Server Client ---
log.Debug("QueryCoord try to new RootCoord client", zap.Any("RootCoordAddress", Params.MasterAddress))
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, 3*time.Second)
if err != nil {
log.Debug("QueryCoord try to new RootCoord client failed", zap.Error(err))

View File

@ -72,11 +72,11 @@ indexParams:
indexParams["SLICE_SIZE"] = "4"
*/
type MasterServiceMock struct {
type RootCoordMock struct {
Count int
}
func (m *MasterServiceMock) DescribeSegment(in *milvuspb.DescribeSegmentRequest) (*milvuspb.DescribeSegmentResponse, error) {
func (m *RootCoordMock) DescribeSegment(in *milvuspb.DescribeSegmentRequest) (*milvuspb.DescribeSegmentResponse, error) {
if m.Count < 20 {
m.Count++
return nil, errors.New("index not exit")

View File

@ -28,21 +28,20 @@ type ParamTable struct {
QueryNodePort int
QueryNodeID UniqueID
IndexServiceAddress string
MasterAddress string
DataCoordAddress string
QueryCoordAddress string
RootCoordAddress string
IndexCoordAddress string
DataCoordAddress string
QueryCoordAddress string
}
func (pt *ParamTable) Init() {
once.Do(func() {
pt.BaseTable.Init()
pt.initPort()
pt.initMasterAddress()
pt.initIndexServiceAddress()
pt.initRootCoordAddress()
pt.initIndexCoordAddress()
pt.initDataCoordAddress()
pt.initQueryCoordAddress()
})
}
@ -54,20 +53,20 @@ func (pt *ParamTable) LoadFromEnv() {
Params.QueryNodeIP = funcutil.GetLocalIP()
}
func (pt *ParamTable) initMasterAddress() {
ret, err := pt.Load("_MasterAddress")
func (pt *ParamTable) initRootCoordAddress() {
ret, err := pt.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
pt.MasterAddress = ret
pt.RootCoordAddress = ret
}
func (pt *ParamTable) initIndexServiceAddress() {
ret, err := pt.Load("IndexServiceAddress")
func (pt *ParamTable) initIndexCoordAddress() {
ret, err := pt.Load("_IndexCoordAddress")
if err != nil {
panic(err)
}
pt.IndexServiceAddress = ret
pt.IndexCoordAddress = ret
}
func (pt *ParamTable) initDataCoordAddress() {

View File

@ -20,14 +20,14 @@ import (
func TestParamTable(t *testing.T) {
Params.Init()
assert.NotEqual(t, Params.IndexServiceAddress, "")
t.Logf("IndexServiceAddress:%s", Params.IndexServiceAddress)
assert.NotEqual(t, Params.IndexCoordAddress, "")
t.Logf("IndexCoordAddress:%s", Params.IndexCoordAddress)
assert.NotEqual(t, Params.DataCoordAddress, "")
t.Logf("DataCoordAddress:%s", Params.DataCoordAddress)
assert.NotEqual(t, Params.MasterAddress, "")
t.Logf("MasterAddress:%s", Params.MasterAddress)
assert.NotEqual(t, Params.RootCoordAddress, "")
t.Logf("RootCoordAddress:%s", Params.RootCoordAddress)
assert.NotEqual(t, Params.QueryCoordAddress, "")
t.Logf("QueryCoordAddress:%s", Params.QueryCoordAddress)

View File

@ -131,9 +131,9 @@ func (s *Server) init() error {
panic(err)
}
// --- Master Server Client ---
// --- RootCoord Client ---
//ms.Params.Init()
addr := Params.MasterAddress
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, 3*time.Second)
@ -164,7 +164,7 @@ func (s *Server) init() error {
}
// --- IndexCoord ---
log.Debug("Index coord", zap.String("address", Params.IndexServiceAddress))
log.Debug("Index coord", zap.String("address", Params.IndexCoordAddress))
indexCoord := isc.NewClient(qn.Params.MetaRootPath, qn.Params.EtcdEndpoints, 3*time.Second)
if err := indexCoord.Init(); err != nil {

View File

@ -26,29 +26,28 @@ type ParamTable struct {
Address string // ip:port
Port int
IndexServiceAddress string
QueryCoordAddress string
DataCoordAddress string
IndexCoordAddress string
QueryCoordAddress string
DataCoordAddress string
}
func (p *ParamTable) Init() {
once.Do(func() {
p.BaseTable.Init()
err := p.LoadYaml("advanced/master.yaml")
err := p.LoadYaml("advanced/root_coord.yaml")
if err != nil {
panic(err)
}
p.initAddress()
p.initPort()
p.initIndexServiceAddress()
p.initIndexCoordAddress()
p.initQueryCoordAddress()
p.initDataCoordAddress()
})
}
func (p *ParamTable) initAddress() {
ret, err := p.Load("_MasterAddress")
ret, err := p.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
@ -56,15 +55,15 @@ func (p *ParamTable) initAddress() {
}
func (p *ParamTable) initPort() {
p.Port = p.ParseInt("master.port")
p.Port = p.ParseInt("rootCoord.port")
}
func (p *ParamTable) initIndexServiceAddress() {
ret, err := p.Load("IndexServiceAddress")
func (p *ParamTable) initIndexCoordAddress() {
ret, err := p.Load("_IndexCoordAddress")
if err != nil {
panic(err)
}
p.IndexServiceAddress = ret
p.IndexCoordAddress = ret
}
func (p *ParamTable) initQueryCoordAddress() {

View File

@ -26,8 +26,8 @@ func TestParamTable(t *testing.T) {
assert.NotEqual(t, Params.Port, 0)
t.Logf("master port = %d", Params.Port)
assert.NotEqual(t, Params.IndexServiceAddress, "")
t.Logf("IndexServiceAddress:%s", Params.IndexServiceAddress)
assert.NotEqual(t, Params.IndexCoordAddress, "")
t.Logf("IndexCoordAddress:%s", Params.IndexCoordAddress)
assert.NotEqual(t, Params.DataCoordAddress, "")
t.Logf("DataCoordAddress:%s", Params.DataCoordAddress)

View File

@ -159,7 +159,7 @@ func (s *Server) init() error {
s.rootCoord.UpdateStateCode(internalpb.StateCode_Initializing)
log.Debug("RootCoord", zap.Any("State", internalpb.StateCode_Initializing))
s.rootCoord.SetNewProxyClient(
func(s *sessionutil.Session) (types.ProxyNode, error) {
func(s *sessionutil.Session) (types.Proxy, error) {
cli := pnc.NewClient(s.Address, 3*time.Second)
if err := cli.Init(); err != nil {
return nil, err

View File

@ -89,12 +89,12 @@ func GenFlushedSegMsgPack(segID typeutil.UniqueID) *msgstream.MsgPack {
return &msgPack
}
type proxyNodeMock struct {
types.ProxyNode
type proxyMock struct {
types.Proxy
invalidateCollectionMetaCache func(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error)
}
func (p *proxyNodeMock) InvalidateCollectionMetaCache(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) {
func (p *proxyMock) InvalidateCollectionMetaCache(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) {
return p.invalidateCollectionMetaCache(ctx, request)
}
@ -231,8 +231,8 @@ func TestGrpcService(t *testing.T) {
}
collectionMetaCache := make([]string, 0, 16)
pnm := proxyNodeMock{}
core.NewProxyClient = func(*sessionutil.Session) (types.ProxyNode, error) {
pnm := proxyMock{}
core.NewProxyClient = func(*sessionutil.Session) (types.Proxy, error) {
return &pnm, nil
}
pnm.invalidateCollectionMetaCache = func(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) {
@ -851,7 +851,7 @@ func (m *mockCore) Stop() error {
return fmt.Errorf("stop error")
}
func (m *mockCore) SetNewProxyClient(func(sess *sessionutil.Session) (types.ProxyNode, error)) {
func (m *mockCore) SetNewProxyClient(func(sess *sessionutil.Session) (types.Proxy, error)) {
}
type mockDataCoord struct {

View File

@ -27,7 +27,7 @@ type ParamTable struct {
Address string
Port int
MasterAddress string
RootCoordAddress string
EtcdEndpoints []string
KvRootPath string
@ -50,7 +50,7 @@ func (pt *ParamTable) Init() {
pt.BaseTable.Init()
pt.initLogCfg()
pt.initEtcdEndpoints()
pt.initMasterAddress()
pt.initRootCoordAddress()
pt.initMetaRootPath()
pt.initKvRootPath()
pt.initMinIOAddress()
@ -93,12 +93,12 @@ func (pt *ParamTable) initKvRootPath() {
pt.KvRootPath = rootPath + "/" + subPath
}
func (pt *ParamTable) initMasterAddress() {
ret, err := pt.Load("_MasterAddress")
func (pt *ParamTable) initRootCoordAddress() {
ret, err := pt.Load("_RootCoordAddress")
if err != nil {
panic(err)
}
pt.MasterAddress = ret
pt.RootCoordAddress = ret
}
func (pt *ParamTable) initMinIOAddress() {

View File

@ -42,7 +42,7 @@ type ParamTable struct {
NodeID int64
Alias string
MasterAddress string
RootCoordAddress string
EtcdEndpoints []string
MetaRootPath string

View File

@ -31,12 +31,12 @@ var (
*/
var (
// RootCoordProxyNodeLister used to count the num of registered proxy nodes
RootCoordProxyNodeLister = prometheus.NewGaugeVec(
// RootCoordProxyLister used to count the num of registered proxy nodes
RootCoordProxyLister = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: milvusNamespace,
Subsystem: subSystemRootCoord,
Name: "list_of_proxy_node",
Name: "list_of_proxy",
Help: "List of proxy nodes which has register with etcd",
}, []string{"client_id"})
@ -193,7 +193,7 @@ var (
//RegisterRootCoord register RootCoord metrics
func RegisterRootCoord() {
prometheus.MustRegister(RootCoordProxyNodeLister)
prometheus.MustRegister(RootCoordProxyLister)
// for grpc
prometheus.MustRegister(RootCoordCreateCollectionCounter)
@ -525,7 +525,7 @@ var (
}, []string{"status"})
)
//RegisterProxy register ProxyNode metrics
//RegisterProxy register Proxy metrics
func RegisterProxy() {
prometheus.MustRegister(ProxyCreateCollectionCounter)
prometheus.MustRegister(ProxyDropCollectionCounter)

View File

@ -1,5 +1,5 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// source: query_service.proto
// source: query_coord.proto
package querypb
@ -66,7 +66,7 @@ func (x PartitionState) String() string {
}
func (PartitionState) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{0}
return fileDescriptor_aab7cc9a69ed26e8, []int{0}
}
type TriggerCondition int32
@ -97,7 +97,7 @@ func (x TriggerCondition) String() string {
}
func (TriggerCondition) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{1}
return fileDescriptor_aab7cc9a69ed26e8, []int{1}
}
//----------------etcd-----------------
@ -132,7 +132,7 @@ func (x SegmentState) String() string {
}
func (SegmentState) EnumDescriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{2}
return fileDescriptor_aab7cc9a69ed26e8, []int{2}
}
//--------------------query coordinator proto------------------
@ -148,7 +148,7 @@ func (m *RegisterNodeRequest) Reset() { *m = RegisterNodeRequest{} }
func (m *RegisterNodeRequest) String() string { return proto.CompactTextString(m) }
func (*RegisterNodeRequest) ProtoMessage() {}
func (*RegisterNodeRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{0}
return fileDescriptor_aab7cc9a69ed26e8, []int{0}
}
func (m *RegisterNodeRequest) XXX_Unmarshal(b []byte) error {
@ -195,7 +195,7 @@ func (m *RegisterNodeResponse) Reset() { *m = RegisterNodeResponse{} }
func (m *RegisterNodeResponse) String() string { return proto.CompactTextString(m) }
func (*RegisterNodeResponse) ProtoMessage() {}
func (*RegisterNodeResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{1}
return fileDescriptor_aab7cc9a69ed26e8, []int{1}
}
func (m *RegisterNodeResponse) XXX_Unmarshal(b []byte) error {
@ -242,7 +242,7 @@ func (m *ShowCollectionsRequest) Reset() { *m = ShowCollectionsRequest{}
func (m *ShowCollectionsRequest) String() string { return proto.CompactTextString(m) }
func (*ShowCollectionsRequest) ProtoMessage() {}
func (*ShowCollectionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{2}
return fileDescriptor_aab7cc9a69ed26e8, []int{2}
}
func (m *ShowCollectionsRequest) XXX_Unmarshal(b []byte) error {
@ -289,7 +289,7 @@ func (m *ShowCollectionsResponse) Reset() { *m = ShowCollectionsResponse
func (m *ShowCollectionsResponse) String() string { return proto.CompactTextString(m) }
func (*ShowCollectionsResponse) ProtoMessage() {}
func (*ShowCollectionsResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{3}
return fileDescriptor_aab7cc9a69ed26e8, []int{3}
}
func (m *ShowCollectionsResponse) XXX_Unmarshal(b []byte) error {
@ -337,7 +337,7 @@ func (m *ShowPartitionsRequest) Reset() { *m = ShowPartitionsRequest{} }
func (m *ShowPartitionsRequest) String() string { return proto.CompactTextString(m) }
func (*ShowPartitionsRequest) ProtoMessage() {}
func (*ShowPartitionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{4}
return fileDescriptor_aab7cc9a69ed26e8, []int{4}
}
func (m *ShowPartitionsRequest) XXX_Unmarshal(b []byte) error {
@ -391,7 +391,7 @@ func (m *ShowPartitionsResponse) Reset() { *m = ShowPartitionsResponse{}
func (m *ShowPartitionsResponse) String() string { return proto.CompactTextString(m) }
func (*ShowPartitionsResponse) ProtoMessage() {}
func (*ShowPartitionsResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{5}
return fileDescriptor_aab7cc9a69ed26e8, []int{5}
}
func (m *ShowPartitionsResponse) XXX_Unmarshal(b []byte) error {
@ -440,7 +440,7 @@ func (m *LoadCollectionRequest) Reset() { *m = LoadCollectionRequest{} }
func (m *LoadCollectionRequest) String() string { return proto.CompactTextString(m) }
func (*LoadCollectionRequest) ProtoMessage() {}
func (*LoadCollectionRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{6}
return fileDescriptor_aab7cc9a69ed26e8, []int{6}
}
func (m *LoadCollectionRequest) XXX_Unmarshal(b []byte) error {
@ -503,7 +503,7 @@ func (m *ReleaseCollectionRequest) Reset() { *m = ReleaseCollectionReque
func (m *ReleaseCollectionRequest) String() string { return proto.CompactTextString(m) }
func (*ReleaseCollectionRequest) ProtoMessage() {}
func (*ReleaseCollectionRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{7}
return fileDescriptor_aab7cc9a69ed26e8, []int{7}
}
func (m *ReleaseCollectionRequest) XXX_Unmarshal(b []byte) error {
@ -567,7 +567,7 @@ func (m *LoadPartitionsRequest) Reset() { *m = LoadPartitionsRequest{} }
func (m *LoadPartitionsRequest) String() string { return proto.CompactTextString(m) }
func (*LoadPartitionsRequest) ProtoMessage() {}
func (*LoadPartitionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{8}
return fileDescriptor_aab7cc9a69ed26e8, []int{8}
}
func (m *LoadPartitionsRequest) XXX_Unmarshal(b []byte) error {
@ -638,7 +638,7 @@ func (m *ReleasePartitionsRequest) Reset() { *m = ReleasePartitionsReque
func (m *ReleasePartitionsRequest) String() string { return proto.CompactTextString(m) }
func (*ReleasePartitionsRequest) ProtoMessage() {}
func (*ReleasePartitionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{9}
return fileDescriptor_aab7cc9a69ed26e8, []int{9}
}
func (m *ReleasePartitionsRequest) XXX_Unmarshal(b []byte) error {
@ -706,7 +706,7 @@ func (m *CreateQueryChannelRequest) Reset() { *m = CreateQueryChannelReq
func (m *CreateQueryChannelRequest) String() string { return proto.CompactTextString(m) }
func (*CreateQueryChannelRequest) ProtoMessage() {}
func (*CreateQueryChannelRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{10}
return fileDescriptor_aab7cc9a69ed26e8, []int{10}
}
func (m *CreateQueryChannelRequest) XXX_Unmarshal(b []byte) error {
@ -754,7 +754,7 @@ func (m *CreateQueryChannelResponse) Reset() { *m = CreateQueryChannelRe
func (m *CreateQueryChannelResponse) String() string { return proto.CompactTextString(m) }
func (*CreateQueryChannelResponse) ProtoMessage() {}
func (*CreateQueryChannelResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{11}
return fileDescriptor_aab7cc9a69ed26e8, []int{11}
}
func (m *CreateQueryChannelResponse) XXX_Unmarshal(b []byte) error {
@ -810,7 +810,7 @@ func (m *GetPartitionStatesRequest) Reset() { *m = GetPartitionStatesReq
func (m *GetPartitionStatesRequest) String() string { return proto.CompactTextString(m) }
func (*GetPartitionStatesRequest) ProtoMessage() {}
func (*GetPartitionStatesRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{12}
return fileDescriptor_aab7cc9a69ed26e8, []int{12}
}
func (m *GetPartitionStatesRequest) XXX_Unmarshal(b []byte) error {
@ -871,7 +871,7 @@ func (m *PartitionStates) Reset() { *m = PartitionStates{} }
func (m *PartitionStates) String() string { return proto.CompactTextString(m) }
func (*PartitionStates) ProtoMessage() {}
func (*PartitionStates) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{13}
return fileDescriptor_aab7cc9a69ed26e8, []int{13}
}
func (m *PartitionStates) XXX_Unmarshal(b []byte) error {
@ -918,7 +918,7 @@ func (m *GetPartitionStatesResponse) Reset() { *m = GetPartitionStatesRe
func (m *GetPartitionStatesResponse) String() string { return proto.CompactTextString(m) }
func (*GetPartitionStatesResponse) ProtoMessage() {}
func (*GetPartitionStatesResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{14}
return fileDescriptor_aab7cc9a69ed26e8, []int{14}
}
func (m *GetPartitionStatesResponse) XXX_Unmarshal(b []byte) error {
@ -965,7 +965,7 @@ func (m *GetSegmentInfoRequest) Reset() { *m = GetSegmentInfoRequest{} }
func (m *GetSegmentInfoRequest) String() string { return proto.CompactTextString(m) }
func (*GetSegmentInfoRequest) ProtoMessage() {}
func (*GetSegmentInfoRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{15}
return fileDescriptor_aab7cc9a69ed26e8, []int{15}
}
func (m *GetSegmentInfoRequest) XXX_Unmarshal(b []byte) error {
@ -1020,7 +1020,7 @@ func (m *SegmentInfo) Reset() { *m = SegmentInfo{} }
func (m *SegmentInfo) String() string { return proto.CompactTextString(m) }
func (*SegmentInfo) ProtoMessage() {}
func (*SegmentInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{16}
return fileDescriptor_aab7cc9a69ed26e8, []int{16}
}
func (m *SegmentInfo) XXX_Unmarshal(b []byte) error {
@ -1123,7 +1123,7 @@ func (m *GetSegmentInfoResponse) Reset() { *m = GetSegmentInfoResponse{}
func (m *GetSegmentInfoResponse) String() string { return proto.CompactTextString(m) }
func (*GetSegmentInfoResponse) ProtoMessage() {}
func (*GetSegmentInfoResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{17}
return fileDescriptor_aab7cc9a69ed26e8, []int{17}
}
func (m *GetSegmentInfoResponse) XXX_Unmarshal(b []byte) error {
@ -1174,7 +1174,7 @@ func (m *AddQueryChannelRequest) Reset() { *m = AddQueryChannelRequest{}
func (m *AddQueryChannelRequest) String() string { return proto.CompactTextString(m) }
func (*AddQueryChannelRequest) ProtoMessage() {}
func (*AddQueryChannelRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{18}
return fileDescriptor_aab7cc9a69ed26e8, []int{18}
}
func (m *AddQueryChannelRequest) XXX_Unmarshal(b []byte) error {
@ -1245,7 +1245,7 @@ func (m *RemoveQueryChannelRequest) Reset() { *m = RemoveQueryChannelReq
func (m *RemoveQueryChannelRequest) String() string { return proto.CompactTextString(m) }
func (*RemoveQueryChannelRequest) ProtoMessage() {}
func (*RemoveQueryChannelRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{19}
return fileDescriptor_aab7cc9a69ed26e8, []int{19}
}
func (m *RemoveQueryChannelRequest) XXX_Unmarshal(b []byte) error {
@ -1318,7 +1318,7 @@ func (m *WatchDmChannelsRequest) Reset() { *m = WatchDmChannelsRequest{}
func (m *WatchDmChannelsRequest) String() string { return proto.CompactTextString(m) }
func (*WatchDmChannelsRequest) ProtoMessage() {}
func (*WatchDmChannelsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{20}
return fileDescriptor_aab7cc9a69ed26e8, []int{20}
}
func (m *WatchDmChannelsRequest) XXX_Unmarshal(b []byte) error {
@ -1405,7 +1405,7 @@ func (m *SegmentLoadInfo) Reset() { *m = SegmentLoadInfo{} }
func (m *SegmentLoadInfo) String() string { return proto.CompactTextString(m) }
func (*SegmentLoadInfo) ProtoMessage() {}
func (*SegmentLoadInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{21}
return fileDescriptor_aab7cc9a69ed26e8, []int{21}
}
func (m *SegmentLoadInfo) XXX_Unmarshal(b []byte) error {
@ -1483,7 +1483,7 @@ func (m *LoadSegmentsRequest) Reset() { *m = LoadSegmentsRequest{} }
func (m *LoadSegmentsRequest) String() string { return proto.CompactTextString(m) }
func (*LoadSegmentsRequest) ProtoMessage() {}
func (*LoadSegmentsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{22}
return fileDescriptor_aab7cc9a69ed26e8, []int{22}
}
func (m *LoadSegmentsRequest) XXX_Unmarshal(b []byte) error {
@ -1555,7 +1555,7 @@ func (m *ReleaseSegmentsRequest) Reset() { *m = ReleaseSegmentsRequest{}
func (m *ReleaseSegmentsRequest) String() string { return proto.CompactTextString(m) }
func (*ReleaseSegmentsRequest) ProtoMessage() {}
func (*ReleaseSegmentsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{23}
return fileDescriptor_aab7cc9a69ed26e8, []int{23}
}
func (m *ReleaseSegmentsRequest) XXX_Unmarshal(b []byte) error {
@ -1630,7 +1630,7 @@ func (m *DmChannelInfo) Reset() { *m = DmChannelInfo{} }
func (m *DmChannelInfo) String() string { return proto.CompactTextString(m) }
func (*DmChannelInfo) ProtoMessage() {}
func (*DmChannelInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{24}
return fileDescriptor_aab7cc9a69ed26e8, []int{24}
}
func (m *DmChannelInfo) XXX_Unmarshal(b []byte) error {
@ -1678,7 +1678,7 @@ func (m *QueryChannelInfo) Reset() { *m = QueryChannelInfo{} }
func (m *QueryChannelInfo) String() string { return proto.CompactTextString(m) }
func (*QueryChannelInfo) ProtoMessage() {}
func (*QueryChannelInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{25}
return fileDescriptor_aab7cc9a69ed26e8, []int{25}
}
func (m *QueryChannelInfo) XXX_Unmarshal(b []byte) error {
@ -1735,7 +1735,7 @@ func (m *CollectionInfo) Reset() { *m = CollectionInfo{} }
func (m *CollectionInfo) String() string { return proto.CompactTextString(m) }
func (*CollectionInfo) ProtoMessage() {}
func (*CollectionInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{26}
return fileDescriptor_aab7cc9a69ed26e8, []int{26}
}
func (m *CollectionInfo) XXX_Unmarshal(b []byte) error {
@ -1803,7 +1803,7 @@ func (m *HandoffSegments) Reset() { *m = HandoffSegments{} }
func (m *HandoffSegments) String() string { return proto.CompactTextString(m) }
func (*HandoffSegments) ProtoMessage() {}
func (*HandoffSegments) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{27}
return fileDescriptor_aab7cc9a69ed26e8, []int{27}
}
func (m *HandoffSegments) XXX_Unmarshal(b []byte) error {
@ -1856,7 +1856,7 @@ func (m *LoadBalanceSegmentInfo) Reset() { *m = LoadBalanceSegmentInfo{}
func (m *LoadBalanceSegmentInfo) String() string { return proto.CompactTextString(m) }
func (*LoadBalanceSegmentInfo) ProtoMessage() {}
func (*LoadBalanceSegmentInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{28}
return fileDescriptor_aab7cc9a69ed26e8, []int{28}
}
func (m *LoadBalanceSegmentInfo) XXX_Unmarshal(b []byte) error {
@ -1946,7 +1946,7 @@ func (m *LoadBalanceRequest) Reset() { *m = LoadBalanceRequest{} }
func (m *LoadBalanceRequest) String() string { return proto.CompactTextString(m) }
func (*LoadBalanceRequest) ProtoMessage() {}
func (*LoadBalanceRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_5fcb6756dc1afb8d, []int{29}
return fileDescriptor_aab7cc9a69ed26e8, []int{29}
}
func (m *LoadBalanceRequest) XXX_Unmarshal(b []byte) error {
@ -2024,129 +2024,129 @@ func init() {
proto.RegisterType((*LoadBalanceRequest)(nil), "milvus.proto.query.LoadBalanceRequest")
}
func init() { proto.RegisterFile("query_service.proto", fileDescriptor_5fcb6756dc1afb8d) }
func init() { proto.RegisterFile("query_coord.proto", fileDescriptor_aab7cc9a69ed26e8) }
var fileDescriptor_5fcb6756dc1afb8d = []byte{
// 1901 bytes of a gzipped FileDescriptorProto
var fileDescriptor_aab7cc9a69ed26e8 = []byte{
// 1896 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xec, 0x59, 0x4f, 0x73, 0x1c, 0x47,
0x15, 0xd7, 0xec, 0xae, 0x56, 0xda, 0xb7, 0xff, 0x26, 0x6d, 0x5b, 0xac, 0x97, 0x38, 0x71, 0xc6,
0x15, 0xd7, 0xec, 0xae, 0x56, 0xda, 0xb7, 0xff, 0xc6, 0x6d, 0x5b, 0xac, 0x97, 0x38, 0x71, 0xc6,
0x31, 0x76, 0x14, 0x22, 0xa7, 0x94, 0x40, 0x91, 0x03, 0x07, 0x4b, 0x1b, 0x8b, 0x85, 0x44, 0x11,
0x23, 0x13, 0x0a, 0x97, 0x8b, 0x61, 0x76, 0xa6, 0xb5, 0x3b, 0x95, 0x99, 0xee, 0xd5, 0xf4, 0xac,
0x65, 0xfb, 0x00, 0x45, 0x15, 0x07, 0x6e, 0x9c, 0x38, 0xc1, 0x85, 0x0b, 0x55, 0x1c, 0xf8, 0x02,
0x9c, 0xf2, 0x45, 0xa0, 0x8a, 0x22, 0x37, 0x8e, 0x7c, 0x01, 0x6a, 0xba, 0x7b, 0x66, 0xe7, 0x4f,
0xaf, 0xb4, 0x96, 0xec, 0xd8, 0x45, 0x71, 0xdb, 0x7e, 0xfd, 0xfa, 0xfd, 0xef, 0xdf, 0xbc, 0x7e,
0x0b, 0x97, 0x8e, 0x67, 0x38, 0x7c, 0x62, 0x31, 0x1c, 0x3e, 0xf2, 0x1c, 0xbc, 0x35, 0x0d, 0x69,
0x44, 0x11, 0x0a, 0x3c, 0xff, 0xd1, 0x8c, 0x89, 0xd5, 0x16, 0xe7, 0xe8, 0xb7, 0x1c, 0x1a, 0x04,
0x94, 0x08, 0x5a, 0xbf, 0x95, 0xe5, 0xe8, 0x77, 0x3c, 0x12, 0xe1, 0x90, 0xd8, 0x7e, 0xb2, 0xcb,
0x9c, 0x09, 0x0e, 0x6c, 0xb9, 0xd2, 0x5d, 0x3b, 0xb2, 0x2d, 0x87, 0xd2, 0xd0, 0x15, 0x14, 0xe3,
0x57, 0x70, 0xc9, 0xc4, 0x63, 0x8f, 0x45, 0x38, 0xdc, 0xa7, 0x2e, 0x36, 0xf1, 0xf1, 0x0c, 0xb3,
0x08, 0xbd, 0x0f, 0xb5, 0x91, 0xcd, 0x70, 0x4f, 0xbb, 0xae, 0xdd, 0x6e, 0x6e, 0xbf, 0xbe, 0x95,
0xb3, 0x42, 0xaa, 0xff, 0x94, 0x8d, 0x77, 0x6c, 0x86, 0x4d, 0xce, 0x89, 0xbe, 0x0b, 0x6b, 0xb6,
0xeb, 0x86, 0x98, 0xb1, 0x5e, 0xe5, 0x94, 0x43, 0x77, 0x05, 0x8f, 0x99, 0x30, 0x1b, 0xbf, 0xd3,
0xe0, 0x72, 0xde, 0x02, 0x36, 0xa5, 0x84, 0x61, 0xf4, 0x01, 0xd4, 0x59, 0x64, 0x47, 0x33, 0x26,
0x8d, 0xf8, 0xa6, 0x52, 0xde, 0x21, 0x67, 0x31, 0x25, 0x2b, 0xda, 0x81, 0xa6, 0x47, 0xbc, 0xc8,
0x9a, 0xda, 0xa1, 0x1d, 0x24, 0x96, 0xbc, 0x95, 0x3f, 0x99, 0x46, 0x68, 0x48, 0xbc, 0xe8, 0x80,
0x33, 0x9a, 0xe0, 0xa5, 0xbf, 0x8d, 0x9f, 0xc3, 0xc6, 0xe1, 0x84, 0x9e, 0xec, 0x52, 0xdf, 0xc7,
0x4e, 0xe4, 0x51, 0xc2, 0xce, 0x1f, 0x15, 0x04, 0x35, 0x77, 0x34, 0x1c, 0x70, 0x43, 0xaa, 0x26,
0xff, 0x6d, 0x44, 0xf0, 0x8d, 0x92, 0xfc, 0x8b, 0xf8, 0xfc, 0x36, 0xb4, 0x9d, 0x54, 0xd6, 0x70,
0x10, 0x7b, 0x5d, 0xbd, 0x5d, 0x35, 0xf3, 0x44, 0xe3, 0xd7, 0x1a, 0x5c, 0x89, 0xd5, 0x1e, 0xd8,
0x61, 0xe4, 0x3d, 0x7f, 0xaf, 0x90, 0x01, 0xad, 0xac, 0xc2, 0x5e, 0x95, 0xef, 0xe5, 0x68, 0xc6,
0xb1, 0x88, 0x6c, 0xd6, 0x84, 0x8b, 0x38, 0x6e, 0x40, 0x6b, 0x9a, 0x88, 0x9a, 0xfb, 0x9d, 0xa3,
0x19, 0x5f, 0x6a, 0x70, 0xe5, 0x13, 0x6a, 0xbb, 0xf3, 0x68, 0x7f, 0xed, 0x6e, 0xa3, 0xef, 0x43,
0x5d, 0xdc, 0xc2, 0x5e, 0x8d, 0xeb, 0xba, 0x99, 0xd7, 0x25, 0x6f, 0xe8, 0xdc, 0xc2, 0x43, 0x4e,
0x30, 0xe5, 0x21, 0xe3, 0x8f, 0x1a, 0xf4, 0x4c, 0xec, 0x63, 0x9b, 0xe1, 0x97, 0xe9, 0xc5, 0x06,
0xd4, 0x09, 0x75, 0xf1, 0x70, 0xc0, 0xbd, 0xa8, 0x9a, 0x72, 0x65, 0x7c, 0x25, 0x23, 0xfc, 0x12,
0x0b, 0xab, 0x54, 0x09, 0xb5, 0x72, 0x25, 0x64, 0xb2, 0xb0, 0x7a, 0x9e, 0x2c, 0x7c, 0x39, 0xcf,
0xc2, 0xab, 0xee, 0xe9, 0x3c, 0x53, 0xab, 0xb9, 0x4c, 0xfd, 0x0c, 0xae, 0xee, 0x86, 0xd8, 0x8e,
0xf0, 0x8f, 0xe3, 0xcf, 0xc8, 0xee, 0xc4, 0x26, 0x04, 0xfb, 0x89, 0x0b, 0x45, 0xe5, 0x9a, 0x42,
0x79, 0x0f, 0xd6, 0xa6, 0x21, 0x7d, 0xfc, 0x24, 0xb5, 0x3b, 0x59, 0x1a, 0x7f, 0xd2, 0xa0, 0xaf,
0x92, 0x7d, 0x91, 0xeb, 0x7d, 0x0b, 0xba, 0xa1, 0x30, 0xce, 0x72, 0x84, 0x3c, 0xae, 0xb5, 0x61,
0x76, 0x24, 0x59, 0x6a, 0x41, 0x37, 0xa1, 0x13, 0x62, 0x36, 0xf3, 0xe7, 0x7c, 0x55, 0xce, 0xd7,
0x16, 0x54, 0xc9, 0x66, 0xfc, 0x45, 0x83, 0xab, 0x7b, 0x38, 0x4a, 0xb3, 0x17, 0xab, 0xc3, 0xaf,
0x66, 0x0a, 0x8d, 0x00, 0xba, 0x05, 0x3b, 0xd1, 0x75, 0x68, 0x66, 0x58, 0x64, 0x7e, 0xb2, 0x24,
0xf4, 0x3d, 0x58, 0x8d, 0x43, 0x87, 0xb9, 0x45, 0x9d, 0x6d, 0x63, 0xab, 0xdc, 0x3b, 0x6c, 0xe5,
0xa5, 0x9a, 0xe2, 0x80, 0xf1, 0x57, 0x0d, 0xfa, 0xaa, 0xd0, 0x5c, 0x24, 0x7d, 0x0f, 0x60, 0x23,
0x35, 0xce, 0x72, 0x31, 0x73, 0x42, 0x6f, 0xca, 0x2f, 0x0d, 0xc7, 0xe9, 0xe6, 0xf6, 0x8d, 0xb3,
0xcd, 0x63, 0xe6, 0x95, 0x54, 0xc4, 0x20, 0x23, 0xc1, 0xf0, 0xe0, 0xca, 0x1e, 0x8e, 0x0e, 0xf1,
0x38, 0xc0, 0x24, 0x1a, 0x92, 0x23, 0x7a, 0xfe, 0x2c, 0xbe, 0x01, 0xc0, 0xa4, 0x9c, 0xf4, 0x13,
0x92, 0xa1, 0x18, 0x7f, 0xaf, 0x40, 0x33, 0xa3, 0x08, 0xbd, 0x0e, 0x8d, 0x74, 0x57, 0x26, 0x61,
0x4e, 0x28, 0xe5, 0xbf, 0xa2, 0xc8, 0x7f, 0x21, 0x91, 0xd5, 0x72, 0x22, 0x17, 0x40, 0x2d, 0xba,
0x0a, 0xeb, 0x01, 0x0e, 0x2c, 0xe6, 0x3d, 0xc5, 0xf2, 0x6a, 0xaf, 0x05, 0x38, 0x38, 0xf4, 0x9e,
0xe2, 0x78, 0x8b, 0xcc, 0x02, 0x2b, 0xa4, 0x27, 0xac, 0x57, 0x17, 0x5b, 0x64, 0x16, 0x98, 0xf4,
0x84, 0xa1, 0x6b, 0x00, 0x1e, 0x71, 0xf1, 0x63, 0x8b, 0xd8, 0x01, 0xee, 0xad, 0xf1, 0xab, 0xd1,
0xe0, 0x94, 0x7d, 0x3b, 0xc0, 0xf1, 0xa5, 0xe6, 0x8b, 0xe1, 0xa0, 0xb7, 0x2e, 0x0e, 0xca, 0x65,
0xec, 0xaa, 0xbc, 0x50, 0xc3, 0x41, 0xaf, 0x21, 0xce, 0xa5, 0x04, 0xf4, 0x31, 0xb4, 0xa5, 0xdf,
0x96, 0xa8, 0x3a, 0xe0, 0x55, 0x77, 0x5d, 0x95, 0x56, 0x19, 0x40, 0x51, 0x73, 0x2d, 0x96, 0x59,
0x19, 0xbf, 0xd1, 0x60, 0xa3, 0x98, 0xcb, 0x8b, 0x94, 0xdd, 0x77, 0x60, 0xd5, 0x23, 0x47, 0x34,
0xa9, 0xb2, 0x37, 0x4f, 0x31, 0x87, 0x2b, 0x13, 0xdc, 0xc6, 0x3f, 0x34, 0xd8, 0xb8, 0xeb, 0xba,
0x2a, 0x64, 0x7c, 0xf6, 0x9a, 0x9a, 0xe7, 0xaf, 0x92, 0xcb, 0xdf, 0x32, 0xe8, 0xf0, 0x2e, 0xbc,
0x56, 0x40, 0x3d, 0x59, 0x06, 0x0d, 0x53, 0xcf, 0xe3, 0xde, 0x70, 0x80, 0xde, 0x01, 0x3d, 0x8f,
0x7c, 0x12, 0xf3, 0x1b, 0x66, 0x37, 0x87, 0x7d, 0xc3, 0x81, 0xf1, 0x4f, 0x0d, 0xae, 0x9a, 0x38,
0xa0, 0x8f, 0xf0, 0xff, 0xae, 0x8f, 0xff, 0xaa, 0xc0, 0xc6, 0x4f, 0xed, 0xc8, 0x99, 0x0c, 0x02,
0x49, 0x64, 0x2f, 0xc7, 0xc1, 0xc2, 0x15, 0xaf, 0x95, 0xaf, 0x78, 0x5a, 0xa6, 0xab, 0xaa, 0x32,
0x8d, 0x9f, 0x69, 0x5b, 0x9f, 0x27, 0xfe, 0xce, 0xcb, 0x34, 0xd3, 0xc4, 0xd4, 0xcf, 0xd1, 0xc4,
0xa0, 0x5d, 0x68, 0xe3, 0xc7, 0x8e, 0x3f, 0x73, 0xb1, 0x25, 0xb4, 0xaf, 0x71, 0xed, 0x6f, 0x28,
0xb4, 0x67, 0xef, 0x48, 0x4b, 0x1e, 0x1a, 0xf2, 0xab, 0xf2, 0x6f, 0x0d, 0xba, 0x72, 0x37, 0xee,
0xfb, 0x96, 0x40, 0xc5, 0x42, 0x38, 0x2a, 0xe5, 0x70, 0x2c, 0x13, 0xd4, 0xe4, 0x7b, 0x5b, 0xcb,
0x7c, 0x6f, 0xaf, 0x01, 0x1c, 0xf9, 0x33, 0x36, 0xb1, 0x22, 0x2f, 0x48, 0x30, 0xb1, 0xc1, 0x29,
0xf7, 0xbd, 0x00, 0xa3, 0xbb, 0xd0, 0x1a, 0x79, 0xc4, 0xa7, 0x63, 0x6b, 0x6a, 0x47, 0x93, 0x18,
0x19, 0x17, 0xb9, 0x7b, 0xcf, 0xc3, 0xbe, 0xbb, 0xc3, 0x79, 0xcd, 0xa6, 0x38, 0x73, 0x10, 0x1f,
0x31, 0xfe, 0x5c, 0x81, 0x4b, 0xb1, 0x9b, 0xd2, 0xe3, 0x17, 0x50, 0x50, 0x1f, 0x25, 0xa5, 0x50,
0x5d, 0xfc, 0x5d, 0x2c, 0xc4, 0xbb, 0x5c, 0x0e, 0xe7, 0x79, 0x59, 0xa0, 0x1f, 0x41, 0xc7, 0xa7,
0xb6, 0x6b, 0x39, 0x94, 0xb8, 0x3c, 0x13, 0x3c, 0x82, 0x9d, 0xed, 0xb7, 0x55, 0x26, 0xdc, 0x0f,
0xbd, 0xf1, 0x18, 0x87, 0xbb, 0x09, 0xaf, 0xd9, 0xf6, 0xf9, 0xbb, 0x4a, 0x2e, 0x39, 0x82, 0xca,
0x06, 0xf9, 0xc5, 0xc5, 0x2a, 0xa9, 0x81, 0xea, 0x29, 0x3d, 0x57, 0x6d, 0x89, 0x9e, 0x6b, 0x55,
0xd1, 0x36, 0xe7, 0x3b, 0x81, 0x7a, 0xa9, 0x13, 0xb8, 0x0f, 0xed, 0x14, 0x57, 0x78, 0xd1, 0xdf,
0x80, 0xb6, 0x30, 0xcb, 0x8a, 0x23, 0x81, 0xdd, 0xa4, 0x67, 0x16, 0xc4, 0x4f, 0x38, 0x2d, 0x96,
0x9a, 0xe2, 0x96, 0xf8, 0x28, 0x35, 0xcc, 0x0c, 0xc5, 0xf8, 0xbd, 0x06, 0x7a, 0x16, 0x91, 0xb9,
0xe4, 0x65, 0x9a, 0xf1, 0x5b, 0xd0, 0x15, 0x03, 0xa3, 0x39, 0x2c, 0xca, 0xf6, 0xf8, 0x38, 0x2b,
0x6e, 0x80, 0x3e, 0x84, 0x0d, 0xc1, 0x58, 0x82, 0x51, 0xd1, 0x26, 0x5f, 0xe6, 0xbb, 0x66, 0x01,
0x4b, 0x7f, 0x5b, 0x81, 0xce, 0xbc, 0x70, 0x96, 0xb6, 0x6a, 0x89, 0x37, 0x39, 0xba, 0x07, 0x6d,
0x69, 0x83, 0x95, 0x2d, 0xfc, 0xb7, 0x54, 0x55, 0x97, 0x8b, 0xb8, 0xd9, 0xca, 0x40, 0x22, 0x7f,
0x20, 0xc8, 0xf2, 0x4d, 0x0c, 0xe0, 0xb9, 0x5f, 0x37, 0x3b, 0x7e, 0xee, 0xc5, 0x7f, 0xd1, 0xa7,
0xdf, 0x2f, 0xa1, 0xfb, 0x03, 0x9b, 0xb8, 0xf4, 0xe8, 0x28, 0x29, 0xec, 0x73, 0x54, 0xf4, 0x47,
0xf9, 0xbe, 0xe4, 0x19, 0x6e, 0xb9, 0xf1, 0x87, 0x0a, 0x6c, 0xc4, 0xb4, 0x1d, 0xdb, 0xb7, 0x89,
0x83, 0x97, 0xef, 0x46, 0x9f, 0x0f, 0xee, 0xde, 0x80, 0x36, 0xa3, 0xb3, 0xd0, 0xc1, 0x56, 0xae,
0x29, 0x6d, 0x09, 0xe2, 0xbe, 0xb8, 0x98, 0xd7, 0x00, 0x5c, 0x16, 0x59, 0xb9, 0x77, 0x67, 0xc3,
0x65, 0x91, 0xdc, 0x7e, 0x13, 0x9a, 0x52, 0x86, 0x4b, 0x09, 0xe6, 0x1f, 0xaf, 0x75, 0x13, 0x04,
0x69, 0x40, 0x09, 0xef, 0x5f, 0xe3, 0xf3, 0x7c, 0x77, 0x8d, 0xef, 0xae, 0xb9, 0x2c, 0xe2, 0x5b,
0xd7, 0x00, 0x1e, 0xd9, 0xbe, 0xe7, 0xf2, 0x62, 0xe1, 0x3d, 0xea, 0xba, 0xd9, 0xe0, 0x94, 0x38,
0x04, 0xc6, 0xdf, 0x34, 0x40, 0x99, 0xe8, 0x9c, 0x1f, 0x73, 0x6e, 0x42, 0x27, 0xe7, 0x67, 0x3a,
0x48, 0xcb, 0x3a, 0xca, 0x62, 0xd0, 0x1c, 0x09, 0x55, 0x56, 0x88, 0x6d, 0x46, 0x09, 0x0f, 0xda,
0xd2, 0xa0, 0x39, 0x4a, 0xcc, 0x8c, 0x8f, 0x6e, 0x3e, 0x85, 0x4e, 0xfe, 0xc9, 0x83, 0x5a, 0xb0,
0xbe, 0x4f, 0xa3, 0x8f, 0x1f, 0x7b, 0x2c, 0xd2, 0x57, 0x50, 0x07, 0x60, 0x9f, 0x46, 0x07, 0x21,
0x66, 0x98, 0x44, 0xba, 0x86, 0x00, 0xea, 0x9f, 0x91, 0x81, 0xc7, 0xbe, 0xd0, 0x2b, 0xe8, 0x92,
0x7c, 0x23, 0xda, 0xfe, 0x90, 0x7c, 0x8a, 0x03, 0x1a, 0x3e, 0xd1, 0xab, 0xf1, 0xf1, 0x74, 0x55,
0x43, 0x3a, 0xb4, 0x52, 0x96, 0xbd, 0x83, 0x9f, 0xe8, 0xab, 0xa8, 0x01, 0xab, 0xe2, 0x67, 0x7d,
0xf3, 0x33, 0xd0, 0x8b, 0xe6, 0xa1, 0x26, 0xac, 0x4d, 0x44, 0xa9, 0xeb, 0x2b, 0xa8, 0x0b, 0x4d,
0x7f, 0x1e, 0x58, 0x5d, 0x8b, 0x09, 0xe3, 0x70, 0xea, 0xc8, 0x10, 0xeb, 0x95, 0x58, 0x5b, 0x1c,
0xab, 0x01, 0x3d, 0x21, 0x7a, 0x75, 0xf3, 0x87, 0xd0, 0xca, 0x36, 0xfa, 0x68, 0x1d, 0x6a, 0xfb,
0x94, 0x60, 0x7d, 0x25, 0x16, 0xbb, 0x17, 0xd2, 0x13, 0x8f, 0x8c, 0x85, 0x0f, 0xf7, 0x42, 0xfa,
0x14, 0x13, 0xbd, 0x12, 0x6f, 0x30, 0x6c, 0xfb, 0xf1, 0x46, 0x35, 0xde, 0x88, 0x17, 0xd8, 0xd5,
0x6b, 0xdb, 0xff, 0x01, 0x00, 0x01, 0x8b, 0x94, 0x86, 0x2e, 0x9a, 0x02, 0xda, 0xc3, 0xd1, 0x2e,
0x0d, 0xa6, 0x94, 0x24, 0xf2, 0x19, 0x7a, 0x7f, 0xc1, 0x60, 0xb7, 0xcc, 0x2a, 0x4d, 0xee, 0x7f,
0x6b, 0xc1, 0x89, 0x02, 0xbb, 0xb1, 0x82, 0x02, 0xae, 0x31, 0xee, 0x22, 0xee, 0x7b, 0xce, 0x17,
0xc9, 0xa8, 0xe1, 0x14, 0x8d, 0x05, 0xd6, 0x44, 0x63, 0xe1, 0xa2, 0xcb, 0xc5, 0x61, 0x14, 0x7a,
0x64, 0x9c, 0xbc, 0x74, 0x8c, 0x15, 0x74, 0x0c, 0x97, 0xe3, 0x57, 0x50, 0x64, 0x47, 0x1e, 0x8b,
0x3c, 0x87, 0x25, 0x0a, 0xb7, 0x17, 0x2b, 0x2c, 0x31, 0x3f, 0xa3, 0x4a, 0x07, 0x5a, 0xd9, 0xc1,
0x3b, 0xba, 0xa5, 0x2a, 0x60, 0xc5, 0x9f, 0x03, 0xfd, 0xdb, 0x67, 0x33, 0xa6, 0x4a, 0x7c, 0xe8,
0x16, 0x86, 0xdd, 0x68, 0x53, 0x09, 0x7d, 0xca, 0x89, 0x7b, 0xff, 0xdd, 0xa5, 0x78, 0x53, 0x6d,
0x1e, 0x74, 0xf2, 0x03, 0x66, 0xf4, 0xce, 0x22, 0x01, 0xa5, 0x21, 0x5e, 0x7f, 0x73, 0x19, 0xd6,
0x54, 0xd5, 0x03, 0xe8, 0xe4, 0xa7, 0x9e, 0x6a, 0x55, 0xca, 0xc9, 0x68, 0xff, 0xb4, 0x97, 0xac,
0xb1, 0x82, 0x7e, 0x01, 0xaf, 0x95, 0x46, 0x8d, 0xe8, 0xdb, 0xea, 0xa8, 0xab, 0x27, 0x92, 0x67,
0x69, 0x90, 0xd6, 0x67, 0xbe, 0x91, 0x0b, 0xad, 0x2f, 0xcd, 0x9c, 0x97, 0xb7, 0x3e, 0x23, 0xfe,
0x34, 0xeb, 0x9f, 0x59, 0xc3, 0x0c, 0x50, 0x79, 0xd8, 0x88, 0xde, 0x53, 0xa9, 0x58, 0x38, 0xf0,
0xec, 0x6f, 0x2d, 0xcb, 0x9e, 0xa6, 0x7c, 0xc6, 0x21, 0xa1, 0x38, 0x97, 0x53, 0xaa, 0x5d, 0x38,
0x67, 0x54, 0xab, 0x5d, 0x3c, 0x7b, 0x13, 0x45, 0x9d, 0x1f, 0x90, 0xa8, 0x73, 0xa5, 0x1c, 0x88,
0xa9, 0x8b, 0x5a, 0x3d, 0x6f, 0x31, 0x56, 0xb6, 0xbf, 0x5a, 0x87, 0x06, 0x77, 0x9e, 0x03, 0xc2,
0xff, 0x41, 0xf7, 0xf9, 0x83, 0xee, 0x43, 0xe8, 0x16, 0xc6, 0x4c, 0x6a, 0x3c, 0x54, 0xcf, 0xa2,
0xce, 0xba, 0x18, 0x23, 0x40, 0xe5, 0x19, 0x8f, 0xba, 0x42, 0x17, 0xce, 0x82, 0xce, 0xd2, 0xf1,
0x10, 0xba, 0x85, 0x19, 0x8b, 0xda, 0x03, 0xf5, 0x20, 0xe6, 0x2c, 0xe9, 0x9f, 0x43, 0x2b, 0xfb,
0xda, 0x56, 0x7f, 0x94, 0x14, 0xef, 0xf1, 0x97, 0x0f, 0x4a, 0x2f, 0x1e, 0xb4, 0x1f, 0x42, 0xb7,
0xf0, 0xc0, 0x56, 0x47, 0x5e, 0xfd, 0x0a, 0x3f, 0x4b, 0xfa, 0xd7, 0x07, 0x33, 0x3b, 0x1f, 0x3e,
0xd8, 0x1e, 0x7b, 0xd1, 0x64, 0x36, 0x8a, 0x8d, 0xb8, 0x23, 0x4e, 0xbe, 0xe7, 0x51, 0xf9, 0xeb,
0x4e, 0x72, 0xdf, 0xee, 0x70, 0x61, 0x77, 0xb8, 0xb0, 0xe9, 0x68, 0x54, 0xe7, 0xcb, 0x0f, 0xfe,
0x1b, 0x00, 0x00, 0xff, 0xff, 0x36, 0x50, 0xea, 0xbe, 0x28, 0x21, 0x00, 0x00,
0xaf, 0xb4, 0x96, 0xec, 0x38, 0x45, 0x71, 0xdb, 0x7e, 0xfd, 0xfa, 0xfd, 0xef, 0xdf, 0xbc, 0x7e,
0x0b, 0x97, 0x8e, 0x67, 0x38, 0x7c, 0x6a, 0x39, 0x94, 0x86, 0xee, 0xd6, 0x34, 0xa4, 0x11, 0x45,
0x28, 0xf0, 0xfc, 0xc7, 0x33, 0x26, 0x56, 0x5b, 0x7c, 0xbf, 0xdf, 0x72, 0x68, 0x10, 0x50, 0x22,
0x68, 0xfd, 0x56, 0x96, 0xa3, 0xdf, 0xf1, 0x48, 0x84, 0x43, 0x62, 0xfb, 0xc9, 0x2e, 0x73, 0x26,
0x38, 0xb0, 0xe5, 0x4a, 0x77, 0xed, 0xc8, 0xce, 0xca, 0x37, 0x7e, 0x05, 0x97, 0x4d, 0x3c, 0xf6,
0x58, 0x84, 0xc3, 0x7d, 0xea, 0x62, 0x13, 0x1f, 0xcf, 0x30, 0x8b, 0xd0, 0x7b, 0x50, 0x1b, 0xd9,
0x0c, 0xf7, 0xb4, 0x1b, 0xda, 0x9d, 0xe6, 0xf6, 0x6b, 0x5b, 0x39, 0x2b, 0xa4, 0xfa, 0x4f, 0xd8,
0x78, 0xc7, 0x66, 0xd8, 0xe4, 0x9c, 0xe8, 0xbb, 0xb0, 0x66, 0xbb, 0x6e, 0x88, 0x19, 0xeb, 0x55,
0x4e, 0x39, 0x74, 0x4f, 0xf0, 0x98, 0x09, 0xb3, 0xf1, 0x3b, 0x0d, 0xae, 0xe4, 0x2d, 0x60, 0x53,
0x4a, 0x18, 0x46, 0xef, 0x43, 0x9d, 0x45, 0x76, 0x34, 0x63, 0xd2, 0x88, 0x6f, 0x2a, 0xe5, 0x1d,
0x72, 0x16, 0x53, 0xb2, 0xa2, 0x1d, 0x68, 0x7a, 0xc4, 0x8b, 0xac, 0xa9, 0x1d, 0xda, 0x41, 0x62,
0xc9, 0x9b, 0xf9, 0x93, 0x69, 0x84, 0x86, 0xc4, 0x8b, 0x0e, 0x38, 0xa3, 0x09, 0x5e, 0xfa, 0xdb,
0xf8, 0x39, 0x6c, 0x1c, 0x4e, 0xe8, 0xc9, 0x2e, 0xf5, 0x7d, 0xec, 0x44, 0x1e, 0x25, 0xec, 0xfc,
0x51, 0x41, 0x50, 0x73, 0x47, 0xc3, 0x01, 0x37, 0xa4, 0x6a, 0xf2, 0xdf, 0x46, 0x04, 0xdf, 0x28,
0xc9, 0xbf, 0x88, 0xcf, 0x6f, 0x41, 0xdb, 0x49, 0x65, 0x0d, 0x07, 0xb1, 0xd7, 0xd5, 0x3b, 0x55,
0x33, 0x4f, 0x34, 0x7e, 0xad, 0xc1, 0xd5, 0x58, 0xed, 0x81, 0x1d, 0x46, 0xde, 0x8b, 0xf7, 0x0a,
0x19, 0xd0, 0xca, 0x2a, 0xec, 0x55, 0xf9, 0x5e, 0x8e, 0x66, 0x1c, 0x8b, 0xc8, 0x66, 0x4d, 0xb8,
0x88, 0xe3, 0x06, 0xb4, 0xa6, 0x89, 0xa8, 0xb9, 0xdf, 0x39, 0x9a, 0xf1, 0x85, 0x06, 0x57, 0x3f,
0xa6, 0xb6, 0x3b, 0x8f, 0xf6, 0x57, 0xee, 0x36, 0xfa, 0x3e, 0xd4, 0xc5, 0x2d, 0xec, 0xd5, 0xb8,
0xae, 0x5b, 0x79, 0x5d, 0xf2, 0x86, 0xce, 0x2d, 0x3c, 0xe4, 0x04, 0x53, 0x1e, 0x32, 0xfe, 0xa8,
0x41, 0xcf, 0xc4, 0x3e, 0xb6, 0x19, 0x7e, 0x95, 0x5e, 0x6c, 0x40, 0x9d, 0x50, 0x17, 0x0f, 0x07,
0xdc, 0x8b, 0xaa, 0x29, 0x57, 0xc6, 0x97, 0x32, 0xc2, 0xaf, 0xb0, 0xb0, 0x4a, 0x95, 0x50, 0x2b,
0x57, 0x42, 0x26, 0x0b, 0xab, 0xe7, 0xc9, 0xc2, 0x17, 0xf3, 0x2c, 0x7c, 0xdd, 0x3d, 0x9d, 0x67,
0x6a, 0x35, 0x97, 0xa9, 0x9f, 0xc1, 0xb5, 0xdd, 0x10, 0xdb, 0x11, 0xfe, 0x71, 0xfc, 0x19, 0xd9,
0x9d, 0xd8, 0x84, 0x60, 0x3f, 0x71, 0xa1, 0xa8, 0x5c, 0x53, 0x28, 0xef, 0xc1, 0xda, 0x34, 0xa4,
0x4f, 0x9e, 0xa6, 0x76, 0x27, 0x4b, 0xe3, 0x4f, 0x1a, 0xf4, 0x55, 0xb2, 0x2f, 0x72, 0xbd, 0x6f,
0x43, 0x37, 0x14, 0xc6, 0x59, 0x8e, 0x90, 0xc7, 0xb5, 0x36, 0xcc, 0x8e, 0x24, 0x4b, 0x2d, 0xe8,
0x16, 0x74, 0x42, 0xcc, 0x66, 0xfe, 0x9c, 0xaf, 0xca, 0xf9, 0xda, 0x82, 0x2a, 0xd9, 0x8c, 0xbf,
0x68, 0x70, 0x6d, 0x0f, 0x47, 0x69, 0xf6, 0x62, 0x75, 0xf8, 0xeb, 0x99, 0x42, 0x23, 0x80, 0x6e,
0xc1, 0x4e, 0x74, 0x03, 0x9a, 0x19, 0x16, 0x99, 0x9f, 0x2c, 0x09, 0x7d, 0x0f, 0x56, 0xe3, 0xd0,
0x61, 0x6e, 0x51, 0x67, 0xdb, 0xd8, 0x2a, 0xf7, 0x0e, 0x5b, 0x79, 0xa9, 0xa6, 0x38, 0x60, 0xfc,
0x55, 0x83, 0xbe, 0x2a, 0x34, 0x17, 0x49, 0xdf, 0x43, 0xd8, 0x48, 0x8d, 0xb3, 0x5c, 0xcc, 0x9c,
0xd0, 0x9b, 0xf2, 0x4b, 0xc3, 0x71, 0xba, 0xb9, 0x7d, 0xf3, 0x6c, 0xf3, 0x98, 0x79, 0x35, 0x15,
0x31, 0xc8, 0x48, 0x30, 0x3c, 0xb8, 0xba, 0x87, 0xa3, 0x43, 0x3c, 0x0e, 0x30, 0x89, 0x86, 0xe4,
0x88, 0x9e, 0x3f, 0x8b, 0xaf, 0x03, 0x30, 0x29, 0x27, 0xfd, 0x84, 0x64, 0x28, 0xc6, 0xdf, 0x2b,
0xd0, 0xcc, 0x28, 0x42, 0xaf, 0x41, 0x23, 0xdd, 0x95, 0x49, 0x98, 0x13, 0x4a, 0xf9, 0xaf, 0x28,
0xf2, 0x5f, 0x48, 0x64, 0xb5, 0x9c, 0xc8, 0x05, 0x50, 0x8b, 0xae, 0xc1, 0x7a, 0x80, 0x03, 0x8b,
0x79, 0xcf, 0xb0, 0xbc, 0xda, 0x6b, 0x01, 0x0e, 0x0e, 0xbd, 0x67, 0x38, 0xde, 0x22, 0xb3, 0xc0,
0x0a, 0xe9, 0x09, 0xeb, 0xd5, 0xc5, 0x16, 0x99, 0x05, 0x26, 0x3d, 0x61, 0xe8, 0x3a, 0x80, 0x47,
0x5c, 0xfc, 0xc4, 0x22, 0x76, 0x80, 0x7b, 0x6b, 0xfc, 0x6a, 0x34, 0x38, 0x65, 0xdf, 0x0e, 0x70,
0x7c, 0xa9, 0xf9, 0x62, 0x38, 0xe8, 0xad, 0x8b, 0x83, 0x72, 0x19, 0xbb, 0x2a, 0x2f, 0xd4, 0x70,
0xd0, 0x6b, 0x88, 0x73, 0x29, 0x01, 0x7d, 0x04, 0x6d, 0xe9, 0xb7, 0x25, 0xaa, 0x0e, 0x78, 0xd5,
0xdd, 0x50, 0xa5, 0x55, 0x06, 0x50, 0xd4, 0x5c, 0x8b, 0x65, 0x56, 0xc6, 0x6f, 0x34, 0xd8, 0x28,
0xe6, 0xf2, 0x22, 0x65, 0xf7, 0x1d, 0x58, 0xf5, 0xc8, 0x11, 0x4d, 0xaa, 0xec, 0x8d, 0x53, 0xcc,
0xe1, 0xca, 0x04, 0xb7, 0xf1, 0x0f, 0x0d, 0x36, 0xee, 0xb9, 0xae, 0x0a, 0x19, 0x9f, 0xbf, 0xa6,
0xe6, 0xf9, 0xab, 0xe4, 0xf2, 0xb7, 0x0c, 0x3a, 0xbc, 0x03, 0x97, 0x0a, 0xa8, 0x27, 0xcb, 0xa0,
0x61, 0xea, 0x79, 0xdc, 0x1b, 0x0e, 0xd0, 0xdb, 0xa0, 0xe7, 0x91, 0x4f, 0x62, 0x7e, 0xc3, 0xec,
0xe6, 0xb0, 0x6f, 0x38, 0x30, 0xfe, 0xa9, 0xc1, 0x35, 0x13, 0x07, 0xf4, 0x31, 0xfe, 0xdf, 0xf5,
0xf1, 0x5f, 0x15, 0xd8, 0xf8, 0xa9, 0x1d, 0x39, 0x93, 0x41, 0x20, 0x89, 0xec, 0xd5, 0x38, 0x58,
0xb8, 0xe2, 0xb5, 0xf2, 0x15, 0x4f, 0xcb, 0x74, 0x55, 0x55, 0xa6, 0xf1, 0x33, 0x6d, 0xeb, 0xb3,
0xc4, 0xdf, 0x79, 0x99, 0x66, 0x9a, 0x98, 0xfa, 0x39, 0x9a, 0x18, 0xb4, 0x0b, 0x6d, 0xfc, 0xc4,
0xf1, 0x67, 0x2e, 0xb6, 0x84, 0xf6, 0x35, 0xae, 0xfd, 0x75, 0x85, 0xf6, 0xec, 0x1d, 0x69, 0xc9,
0x43, 0x43, 0x7e, 0x55, 0xfe, 0xad, 0x41, 0x57, 0xee, 0xc6, 0x7d, 0xdf, 0x12, 0xa8, 0x58, 0x08,
0x47, 0xa5, 0x1c, 0x8e, 0x65, 0x82, 0x9a, 0x7c, 0x6f, 0x6b, 0x99, 0xef, 0xed, 0x75, 0x80, 0x23,
0x7f, 0xc6, 0x26, 0x56, 0xe4, 0x05, 0x09, 0x26, 0x36, 0x38, 0xe5, 0x81, 0x17, 0x60, 0x74, 0x0f,
0x5a, 0x23, 0x8f, 0xf8, 0x74, 0x6c, 0x4d, 0xed, 0x68, 0x12, 0x23, 0xe3, 0x22, 0x77, 0xef, 0x7b,
0xd8, 0x77, 0x77, 0x38, 0xaf, 0xd9, 0x14, 0x67, 0x0e, 0xe2, 0x23, 0xc6, 0x9f, 0x2b, 0x70, 0x39,
0x76, 0x53, 0x7a, 0xfc, 0x12, 0x0a, 0xea, 0xc3, 0xa4, 0x14, 0xaa, 0x8b, 0xbf, 0x8b, 0x85, 0x78,
0x97, 0xcb, 0xe1, 0x3c, 0x2f, 0x0b, 0xf4, 0x23, 0xe8, 0xf8, 0xd4, 0x76, 0x2d, 0x87, 0x12, 0x97,
0x67, 0x82, 0x47, 0xb0, 0xb3, 0xfd, 0x96, 0xca, 0x84, 0x07, 0xa1, 0x37, 0x1e, 0xe3, 0x70, 0x37,
0xe1, 0x35, 0xdb, 0x3e, 0x7f, 0x57, 0xc9, 0x25, 0x47, 0x50, 0xd9, 0x20, 0xbf, 0xbc, 0x58, 0x25,
0x35, 0x50, 0x3d, 0xa5, 0xe7, 0xaa, 0x2d, 0xd1, 0x73, 0xad, 0x2a, 0xda, 0xe6, 0x7c, 0x27, 0x50,
0x2f, 0x75, 0x02, 0x0f, 0xa0, 0x9d, 0xe2, 0x0a, 0x2f, 0xfa, 0x9b, 0xd0, 0x16, 0x66, 0x59, 0x71,
0x24, 0xb0, 0x9b, 0xf4, 0xcc, 0x82, 0xf8, 0x31, 0xa7, 0xc5, 0x52, 0x53, 0xdc, 0x12, 0x1f, 0xa5,
0x86, 0x99, 0xa1, 0x18, 0xbf, 0xd7, 0x40, 0xcf, 0x22, 0x32, 0x97, 0xbc, 0x4c, 0x33, 0x7e, 0x1b,
0xba, 0x72, 0x5c, 0x94, 0xc2, 0xa2, 0x6c, 0x8f, 0x8f, 0xb3, 0xe2, 0x06, 0xe8, 0x03, 0xd8, 0x10,
0x8c, 0x25, 0x18, 0x15, 0x6d, 0xf2, 0x15, 0xbe, 0x6b, 0x16, 0xb0, 0xf4, 0xb7, 0x15, 0xe8, 0xcc,
0x0b, 0x67, 0x69, 0xab, 0x96, 0x78, 0x93, 0xa3, 0xfb, 0xd0, 0x96, 0x36, 0x58, 0xd9, 0xc2, 0x7f,
0x53, 0x55, 0x75, 0xb9, 0x88, 0x9b, 0xad, 0x0c, 0x24, 0xf2, 0x07, 0x82, 0x2c, 0xdf, 0xc4, 0x00,
0x9e, 0xfb, 0x75, 0xb3, 0xe3, 0xe7, 0x5e, 0xfc, 0x17, 0x7d, 0xfa, 0xfd, 0x12, 0xba, 0x3f, 0xb0,
0x89, 0x4b, 0x8f, 0x8e, 0x92, 0xc2, 0x3e, 0x47, 0x45, 0x7f, 0x98, 0xef, 0x4b, 0x9e, 0xe3, 0x96,
0x1b, 0x7f, 0xa8, 0xc0, 0x46, 0x4c, 0xdb, 0xb1, 0x7d, 0x9b, 0x38, 0x78, 0xf9, 0x6e, 0xf4, 0xc5,
0xe0, 0xee, 0x4d, 0x68, 0x33, 0x3a, 0x0b, 0x1d, 0x6c, 0xe5, 0x9a, 0xd2, 0x96, 0x20, 0xee, 0x8b,
0x8b, 0x79, 0x1d, 0xc0, 0x65, 0x91, 0x95, 0x7b, 0x77, 0x36, 0x5c, 0x16, 0xc9, 0xed, 0x37, 0xa0,
0x29, 0x65, 0xb8, 0x94, 0x60, 0xfe, 0xf1, 0x5a, 0x37, 0x41, 0x90, 0x06, 0x94, 0xf0, 0xfe, 0x35,
0x3e, 0xcf, 0x77, 0xd7, 0xf8, 0xee, 0x9a, 0xcb, 0x22, 0xbe, 0x75, 0x1d, 0xe0, 0xb1, 0xed, 0x7b,
0x2e, 0x2f, 0x16, 0xde, 0xa3, 0xae, 0x9b, 0x0d, 0x4e, 0x89, 0x43, 0x60, 0xfc, 0x4d, 0x03, 0x94,
0x89, 0xce, 0xf9, 0x31, 0xe7, 0x16, 0x74, 0x72, 0x7e, 0xa6, 0x83, 0xb4, 0xac, 0xa3, 0x2c, 0x06,
0xcd, 0x91, 0x50, 0x65, 0x85, 0xd8, 0x66, 0x94, 0xf0, 0xa0, 0x2d, 0x0d, 0x9a, 0xa3, 0xc4, 0xcc,
0xf8, 0xe8, 0xe6, 0x33, 0xe8, 0xe4, 0x9f, 0x3c, 0xa8, 0x05, 0xeb, 0xfb, 0x34, 0xfa, 0xe8, 0x89,
0xc7, 0x22, 0x7d, 0x05, 0x75, 0x00, 0xf6, 0x69, 0x74, 0x10, 0x62, 0x86, 0x49, 0xa4, 0x6b, 0x08,
0xa0, 0xfe, 0x29, 0x19, 0x78, 0xec, 0x73, 0xbd, 0x82, 0x2e, 0xcb, 0x37, 0xa2, 0xed, 0x0f, 0xc9,
0x27, 0x38, 0xa0, 0xe1, 0x53, 0xbd, 0x1a, 0x1f, 0x4f, 0x57, 0x35, 0xa4, 0x43, 0x2b, 0x65, 0xd9,
0x3b, 0xf8, 0x89, 0xbe, 0x8a, 0x1a, 0xb0, 0x2a, 0x7e, 0xd6, 0x37, 0x3f, 0x05, 0xbd, 0x68, 0x1e,
0x6a, 0xc2, 0xda, 0x44, 0x94, 0xba, 0xbe, 0x82, 0xba, 0xd0, 0xf4, 0xe7, 0x81, 0xd5, 0xb5, 0x98,
0x30, 0x0e, 0xa7, 0x8e, 0x0c, 0xb1, 0x5e, 0x89, 0xb5, 0xc5, 0xb1, 0x1a, 0xd0, 0x13, 0xa2, 0x57,
0x37, 0x7f, 0x08, 0xad, 0x6c, 0xa3, 0x8f, 0xd6, 0xa1, 0xb6, 0x4f, 0x09, 0xd6, 0x57, 0x62, 0xb1,
0x7b, 0x21, 0x3d, 0xf1, 0xc8, 0x58, 0xf8, 0x70, 0x3f, 0xa4, 0xcf, 0x30, 0xd1, 0x2b, 0xf1, 0x06,
0xc3, 0xb6, 0x1f, 0x6f, 0x54, 0xe3, 0x8d, 0x78, 0x81, 0x5d, 0xbd, 0xb6, 0xfd, 0x1f, 0x00, 0x10,
0xb0, 0x48, 0x69, 0xe8, 0xa2, 0x29, 0xa0, 0x3d, 0x1c, 0xed, 0xd2, 0x60, 0x4a, 0x49, 0x22, 0x9f,
0xa1, 0xf7, 0x16, 0x0c, 0x76, 0xcb, 0xac, 0xd2, 0xe4, 0xfe, 0xb7, 0x16, 0x9c, 0x28, 0xb0, 0x1b,
0x2b, 0x28, 0xe0, 0x1a, 0xe3, 0x2e, 0xe2, 0x81, 0xe7, 0x7c, 0x9e, 0x8c, 0x1a, 0x4e, 0xd1, 0x58,
0x60, 0x4d, 0x34, 0x16, 0x2e, 0xba, 0x5c, 0x1c, 0x46, 0xa1, 0x47, 0xc6, 0xc9, 0x4b, 0xc7, 0x58,
0x41, 0xc7, 0x70, 0x25, 0x7e, 0x05, 0x45, 0x76, 0xe4, 0xb1, 0xc8, 0x73, 0x58, 0xa2, 0x70, 0x7b,
0xb1, 0xc2, 0x12, 0xf3, 0x73, 0xaa, 0x74, 0xa0, 0x95, 0x1d, 0xbc, 0xa3, 0xdb, 0xaa, 0x02, 0x56,
0xfc, 0x39, 0xd0, 0xbf, 0x73, 0x36, 0x63, 0xaa, 0xc4, 0x87, 0x6e, 0x61, 0xd8, 0x8d, 0x36, 0x95,
0xd0, 0xa7, 0x9c, 0xb8, 0xf7, 0xdf, 0x59, 0x8a, 0x37, 0xd5, 0xe6, 0x41, 0x27, 0x3f, 0x60, 0x46,
0x6f, 0x2f, 0x12, 0x50, 0x1a, 0xe2, 0xf5, 0x37, 0x97, 0x61, 0x4d, 0x55, 0x3d, 0x84, 0x4e, 0x7e,
0xea, 0xa9, 0x56, 0xa5, 0x9c, 0x8c, 0xf6, 0x4f, 0x7b, 0xc9, 0x1a, 0x2b, 0xe8, 0x17, 0x70, 0xa9,
0x34, 0x6a, 0x44, 0xdf, 0x56, 0x47, 0x5d, 0x3d, 0x91, 0x3c, 0x4b, 0x83, 0xb4, 0x3e, 0xf3, 0x8d,
0x5c, 0x68, 0x7d, 0x69, 0xe6, 0xbc, 0xbc, 0xf5, 0x19, 0xf1, 0xa7, 0x59, 0xff, 0xdc, 0x1a, 0x66,
0x80, 0xca, 0xc3, 0x46, 0xf4, 0xae, 0x4a, 0xc5, 0xc2, 0x81, 0x67, 0x7f, 0x6b, 0x59, 0xf6, 0x34,
0xe5, 0x33, 0x0e, 0x09, 0xc5, 0xb9, 0x9c, 0x52, 0xed, 0xc2, 0x39, 0xa3, 0x5a, 0xed, 0xe2, 0xd9,
0x9b, 0x28, 0xea, 0xfc, 0x80, 0x44, 0x9d, 0x2b, 0xe5, 0x40, 0x4c, 0x5d, 0xd4, 0xea, 0x79, 0x8b,
0xb1, 0xb2, 0xfd, 0xe5, 0x3a, 0x34, 0xb8, 0xf3, 0x1c, 0x10, 0xfe, 0x0f, 0xba, 0x2f, 0x1e, 0x74,
0x1f, 0x41, 0xb7, 0x30, 0x66, 0x52, 0xe3, 0xa1, 0x7a, 0x16, 0x75, 0xd6, 0xc5, 0x18, 0x01, 0x2a,
0xcf, 0x78, 0xd4, 0x15, 0xba, 0x70, 0x16, 0x74, 0x96, 0x8e, 0x47, 0xd0, 0x2d, 0xcc, 0x58, 0xd4,
0x1e, 0xa8, 0x07, 0x31, 0x67, 0x49, 0xff, 0x0c, 0x5a, 0xd9, 0xd7, 0xb6, 0xfa, 0xa3, 0xa4, 0x78,
0x8f, 0xbf, 0x7a, 0x50, 0x7a, 0xf9, 0xa0, 0xfd, 0x08, 0xba, 0x85, 0x07, 0xb6, 0x3a, 0xf2, 0xea,
0x57, 0xf8, 0x59, 0xd2, 0xbf, 0x3a, 0x98, 0xd9, 0xf9, 0xe0, 0xe1, 0xf6, 0xd8, 0x8b, 0x26, 0xb3,
0x51, 0x6c, 0xc4, 0x5d, 0x71, 0xf2, 0x5d, 0x8f, 0xca, 0x5f, 0x77, 0x93, 0xfb, 0x76, 0x97, 0x0b,
0xbb, 0xcb, 0x85, 0x4d, 0x47, 0xa3, 0x3a, 0x5f, 0xbe, 0xff, 0xdf, 0x00, 0x00, 0x00, 0xff, 0xff,
0x9e, 0x3d, 0x5b, 0x45, 0x26, 0x21, 0x00, 0x00,
}
// Reference imports to suppress errors if they are not otherwise used.
@ -2658,7 +2658,7 @@ var _QueryCoord_serviceDesc = grpc.ServiceDesc{
},
},
Streams: []grpc.StreamDesc{},
Metadata: "query_service.proto",
Metadata: "query_coord.proto",
}
// QueryNodeClient is the client API for QueryNode service.
@ -3090,5 +3090,5 @@ var _QueryNode_serviceDesc = grpc.ServiceDesc{
},
},
Streams: []grpc.StreamDesc{},
Metadata: "query_service.proto",
Metadata: "query_coord.proto",
}

View File

@ -36,10 +36,10 @@ type ParamTable struct {
NetworkAddress string
Alias string
EtcdEndpoints []string
MetaRootPath string
MasterAddress string
PulsarAddress string
EtcdEndpoints []string
MetaRootPath string
RootCoordAddress string
PulsarAddress string
ProxyID UniqueID
TimeTickInterval time.Duration
@ -65,7 +65,7 @@ var once sync.Once
func (pt *ParamTable) Init() {
once.Do(func() {
pt.BaseTable.Init()
err := pt.LoadYaml("advanced/proxy_node.yaml")
err := pt.LoadYaml("advanced/proxy.yaml")
if err != nil {
panic(err)
}
@ -105,7 +105,7 @@ func (pt *ParamTable) initPulsarAddress() {
}
func (pt *ParamTable) initTimeTickInterval() {
intervalStr, err := pt.Load("proxyNode.timeTickInterval")
intervalStr, err := pt.Load("proxy.timeTickInterval")
if err != nil {
panic(err)
}
@ -134,11 +134,11 @@ func (pt *ParamTable) initProxyTimeTickChannelNames() {
}
func (pt *ParamTable) initMsgStreamTimeTickBufSize() {
pt.MsgStreamTimeTickBufSize = pt.ParseInt64("proxyNode.msgStream.timeTick.bufSize")
pt.MsgStreamTimeTickBufSize = pt.ParseInt64("proxy.msgStream.timeTick.bufSize")
}
func (pt *ParamTable) initMaxNameLength() {
str, err := pt.Load("proxyNode.maxNameLength")
str, err := pt.Load("proxy.maxNameLength")
if err != nil {
panic(err)
}
@ -150,7 +150,7 @@ func (pt *ParamTable) initMaxNameLength() {
}
func (pt *ParamTable) initMaxFieldNum() {
str, err := pt.Load("proxyNode.maxFieldNum")
str, err := pt.Load("proxy.maxFieldNum")
if err != nil {
panic(err)
}
@ -162,7 +162,7 @@ func (pt *ParamTable) initMaxFieldNum() {
}
func (pt *ParamTable) initMaxDimension() {
str, err := pt.Load("proxyNode.maxDimension")
str, err := pt.Load("proxy.maxDimension")
if err != nil {
panic(err)
}
@ -245,7 +245,7 @@ func (pt *ParamTable) initLogCfg() {
panic(err)
}
if len(rootPath) != 0 {
pt.Log.File.Filename = path.Join(rootPath, fmt.Sprintf("proxynode%s.log", pt.Alias))
pt.Log.File.Filename = path.Join(rootPath, fmt.Sprintf("proxy-%s.log", pt.Alias))
} else {
pt.Log.File.Filename = ""
}

View File

@ -55,7 +55,7 @@ func (p *ParamTable) Init() {
once.Do(func() {
// load yaml
p.BaseTable.Init()
err := p.LoadYaml("advanced/master.yaml")
err := p.LoadYaml("advanced/root_coord.yaml")
if err != nil {
panic(err)
}

View File

@ -27,14 +27,14 @@ import (
type proxyClientManager struct {
core *Core
lock sync.Mutex
proxyClient map[int64]types.ProxyNode
proxyClient map[int64]types.Proxy
}
func newProxyClientManager(c *Core) *proxyClientManager {
return &proxyClientManager{
core: c,
lock: sync.Mutex{},
proxyClient: make(map[int64]types.ProxyNode),
proxyClient: make(map[int64]types.Proxy),
}
}

View File

@ -27,7 +27,7 @@ import (
"go.uber.org/zap"
)
type proxyNodeManager struct {
type proxyManager struct {
ctx context.Context
cancel context.CancelFunc
lock sync.Mutex
@ -37,13 +37,13 @@ type proxyNodeManager struct {
delSessions []func(*sessionutil.Session)
}
func newProxyNodeManager(ctx context.Context, etcdEndpoints []string, fns ...func([]*sessionutil.Session)) (*proxyNodeManager, error) {
func newProxyManager(ctx context.Context, etcdEndpoints []string, fns ...func([]*sessionutil.Session)) (*proxyManager, error) {
cli, err := clientv3.New(clientv3.Config{Endpoints: etcdEndpoints})
if err != nil {
return nil, err
}
ctx2, cancel2 := context.WithCancel(ctx)
p := &proxyNodeManager{
p := &proxyManager{
ctx: ctx2,
cancel: cancel2,
lock: sync.Mutex{},
@ -53,19 +53,19 @@ func newProxyNodeManager(ctx context.Context, etcdEndpoints []string, fns ...fun
return p, nil
}
func (p *proxyNodeManager) AddSession(fns ...func(*sessionutil.Session)) {
func (p *proxyManager) AddSession(fns ...func(*sessionutil.Session)) {
p.lock.Lock()
defer p.lock.Unlock()
p.addSessions = append(p.addSessions, fns...)
}
func (p *proxyNodeManager) DelSession(fns ...func(*sessionutil.Session)) {
func (p *proxyManager) DelSession(fns ...func(*sessionutil.Session)) {
p.lock.Lock()
defer p.lock.Unlock()
p.delSessions = append(p.delSessions, fns...)
}
func (p *proxyNodeManager) WatchProxyNode() error {
func (p *proxyManager) WatchProxy() error {
ctx2, cancel := context.WithTimeout(p.ctx, RequestTimeout)
defer cancel()
resp, err := p.etcdCli.Get(
@ -75,7 +75,7 @@ func (p *proxyNodeManager) WatchProxyNode() error {
clientv3.WithSort(clientv3.SortByKey, clientv3.SortAscend),
)
if err != nil {
return fmt.Errorf("proxyNodeManager,watch proxy node failed, error = %w", err)
return fmt.Errorf("proxyManager, watch proxy failed, error = %w", err)
}
sessions := []*sessionutil.Session{}
for _, v := range resp.Kvs {
@ -91,10 +91,10 @@ func (p *proxyNodeManager) WatchProxyNode() error {
f(sessions)
}
for _, s := range sessions {
metrics.RootCoordProxyNodeLister.WithLabelValues(metricProxyNode(s.ServerID)).Set(1)
metrics.RootCoordProxyLister.WithLabelValues(metricProxy(s.ServerID)).Set(1)
}
for _, s := range sessions {
log.Debug("Get proxy node", zap.Int64("node id", s.ServerID), zap.String("node addr", s.Address), zap.String("node name", s.ServerName))
log.Debug("Get proxy", zap.Int64("id", s.ServerID), zap.String("addr", s.Address), zap.String("name", s.ServerName))
}
rch := p.etcdCli.Watch(
@ -114,7 +114,7 @@ func (p *proxyNodeManager) WatchProxyNode() error {
return
case wresp, ok := <-rch:
if !ok {
log.Debug("watch proxy node failed")
log.Debug("watch proxy failed")
return
}
for _, ev := range wresp.Events {
@ -123,7 +123,7 @@ func (p *proxyNodeManager) WatchProxyNode() error {
sess := new(sessionutil.Session)
err := json.Unmarshal(ev.Kv.Value, sess)
if err != nil {
log.Debug("watch proxy node, unmarshal failed", zap.Error(err))
log.Debug("watch proxy, unmarshal failed", zap.Error(err))
continue
}
p.lock.Lock()
@ -131,12 +131,12 @@ func (p *proxyNodeManager) WatchProxyNode() error {
f(sess)
}
p.lock.Unlock()
metrics.RootCoordProxyNodeLister.WithLabelValues(metricProxyNode(sess.ServerID)).Set(1)
metrics.RootCoordProxyLister.WithLabelValues(metricProxy(sess.ServerID)).Set(1)
case mvccpb.DELETE:
sess := new(sessionutil.Session)
err := json.Unmarshal(ev.PrevKv.Value, sess)
if err != nil {
log.Debug("watch proxy node, unmarshal failed", zap.Error(err))
log.Debug("watch proxy, unmarshal failed", zap.Error(err))
continue
}
p.lock.Lock()
@ -144,7 +144,7 @@ func (p *proxyNodeManager) WatchProxyNode() error {
f(sess)
}
p.lock.Unlock()
metrics.RootCoordProxyNodeLister.WithLabelValues(metricProxyNode(sess.ServerID)).Set(0)
metrics.RootCoordProxyLister.WithLabelValues(metricProxy(sess.ServerID)).Set(0)
}
}
}
@ -154,6 +154,6 @@ func (p *proxyNodeManager) WatchProxyNode() error {
return nil
}
func (p *proxyNodeManager) Stop() {
func (p *proxyManager) Stop() {
p.cancel()
}

View File

@ -24,7 +24,7 @@ import (
"go.etcd.io/etcd/clientv3"
)
func TestProxyNodeManager(t *testing.T) {
func TestProxyManager(t *testing.T) {
Params.Init()
cli, err := clientv3.New(clientv3.Config{Endpoints: Params.EtcdEndpoints})
assert.Nil(t, err)
@ -60,7 +60,7 @@ func TestProxyNodeManager(t *testing.T) {
t.Log("get sessions", sess[0], sess[1])
}
pm, err := newProxyNodeManager(ctx, Params.EtcdEndpoints, f1)
pm, err := newProxyManager(ctx, Params.EtcdEndpoints, f1)
assert.Nil(t, err)
fa := func(sess *sessionutil.Session) {
assert.Equal(t, int64(101), sess.ServerID)
@ -73,9 +73,9 @@ func TestProxyNodeManager(t *testing.T) {
pm.AddSession(fa)
pm.DelSession(fd)
err = pm.WatchProxyNode()
err = pm.WatchProxy()
assert.Nil(t, err)
t.Log("======== start watch proxy node ==========")
t.Log("======== start watch proxy ==========")
s2 := sessionutil.Session{
ServerID: 101,

View File

@ -64,7 +64,7 @@ const (
MetricRequestsSuccess = "success"
)
func metricProxyNode(v int64) string {
func metricProxy(v int64) string {
return fmt.Sprintf("client_%d", v)
}
@ -114,7 +114,7 @@ type Core struct {
CallBuildIndexService func(ctx context.Context, binlog []string, field *schemapb.FieldSchema, idxInfo *etcdpb.IndexInfo) (typeutil.UniqueID, error)
CallDropIndexService func(ctx context.Context, indexID typeutil.UniqueID) error
NewProxyClient func(sess *sessionutil.Session) (types.ProxyNode, error)
NewProxyClient func(sess *sessionutil.Session) (types.Proxy, error)
//query service interface, notify query service to release collection
CallReleaseCollectionService func(ctx context.Context, ts typeutil.Timestamp, dbID, collectionID typeutil.UniqueID) error
@ -126,8 +126,8 @@ type Core struct {
//dml channels
dmlChannels *dmlChannels
//ProxyNode manager
proxyNodeManager *proxyNodeManager
//Proxy manager
proxyManager *proxyManager
// proxy clients
proxyClientManager *proxyClientManager
@ -220,7 +220,7 @@ func (c *Core) checkInit() error {
return fmt.Errorf("CallDropIndexService is nil")
}
if c.NewProxyClient == nil {
return fmt.Errorf("NewProxyNodeClient is nil")
return fmt.Errorf("NewProxyClient is nil")
}
if c.CallReleaseCollectionService == nil {
return fmt.Errorf("CallReleaseCollectionService is nil")
@ -465,7 +465,7 @@ func (c *Core) sessionLoop() {
}
}
func (c *Core) watchProxyNodeLoop() {
func (c *Core) watchProxyLoop() {
}
@ -666,7 +666,7 @@ func (c *Core) setMsgStreams() error {
}
//SetNewProxyClient create proxy node by this func
func (c *Core) SetNewProxyClient(f func(sess *sessionutil.Session) (types.ProxyNode, error)) {
func (c *Core) SetNewProxyClient(f func(sess *sessionutil.Session) (types.Proxy, error)) {
if c.NewProxyClient == nil {
c.NewProxyClient = f
} else {
@ -1003,18 +1003,18 @@ func (c *Core) Init() error {
c.dmlChannels.AddProducerChannels(pc...)
c.chanTimeTick = newTimeTickSync(c)
c.chanTimeTick.AddProxyNode(c.session)
c.chanTimeTick.AddProxy(c.session)
c.proxyClientManager = newProxyClientManager(c)
log.Debug("RootCoord, set proxy manager")
c.proxyNodeManager, initError = newProxyNodeManager(
c.proxyManager, initError = newProxyManager(
c.ctx,
Params.EtcdEndpoints,
c.chanTimeTick.GetProxyNodes,
c.chanTimeTick.GetProxy,
c.proxyClientManager.GetProxyClients,
)
c.proxyNodeManager.AddSession(c.chanTimeTick.AddProxyNode, c.proxyClientManager.AddProxyClient)
c.proxyNodeManager.DelSession(c.chanTimeTick.DelProxyNode, c.proxyClientManager.DelProxyClient)
c.proxyManager.AddSession(c.chanTimeTick.AddProxy, c.proxyClientManager.AddProxyClient)
c.proxyManager.DelSession(c.chanTimeTick.DelProxy, c.proxyClientManager.DelProxyClient)
c.ddReqQueue = make(chan reqTask, 1024)
initError = c.setMsgStreams()
@ -1153,8 +1153,8 @@ func (c *Core) Start() error {
log.Debug(typeutil.RootCoordRole, zap.String("time tick channel name", Params.TimeTickChannel))
c.startOnce.Do(func() {
if err := c.proxyNodeManager.WatchProxyNode(); err != nil {
log.Debug("RootCoord Start WatchProxyNode failed", zap.Error(err))
if err := c.proxyManager.WatchProxy(); err != nil {
log.Debug("RootCoord Start WatchProxy failed", zap.Error(err))
return
}
if err := c.reSendDdMsg(c.ctx); err != nil {
@ -1227,7 +1227,7 @@ func (c *Core) GetStatisticsChannel(ctx context.Context) (*milvuspb.StringRespon
}
func (c *Core) CreateCollection(ctx context.Context, in *milvuspb.CreateCollectionRequest) (*commonpb.Status, error) {
metrics.RootCoordCreateCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordCreateCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &commonpb.Status{
@ -1254,7 +1254,7 @@ func (c *Core) CreateCollection(ctx context.Context, in *milvuspb.CreateCollecti
}, nil
}
log.Debug("CreateCollection Success", zap.String("name", in.CollectionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordCreateCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordCreateCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1262,7 +1262,7 @@ func (c *Core) CreateCollection(ctx context.Context, in *milvuspb.CreateCollecti
}
func (c *Core) DropCollection(ctx context.Context, in *milvuspb.DropCollectionRequest) (*commonpb.Status, error) {
metrics.RootCoordDropCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordDropCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &commonpb.Status{
@ -1289,7 +1289,7 @@ func (c *Core) DropCollection(ctx context.Context, in *milvuspb.DropCollectionRe
}, nil
}
log.Debug("DropCollection Success", zap.String("name", in.CollectionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordDropCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordDropCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1297,7 +1297,7 @@ func (c *Core) DropCollection(ctx context.Context, in *milvuspb.DropCollectionRe
}
func (c *Core) HasCollection(ctx context.Context, in *milvuspb.HasCollectionRequest) (*milvuspb.BoolResponse, error) {
metrics.RootCoordHasCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordHasCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.BoolResponse{
@ -1331,7 +1331,7 @@ func (c *Core) HasCollection(ctx context.Context, in *milvuspb.HasCollectionRequ
}, nil
}
log.Debug("HasCollection Success", zap.String("name", in.CollectionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordHasCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordHasCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &milvuspb.BoolResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
@ -1342,7 +1342,7 @@ func (c *Core) HasCollection(ctx context.Context, in *milvuspb.HasCollectionRequ
}
func (c *Core) DescribeCollection(ctx context.Context, in *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) {
metrics.RootCoordDescribeCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordDescribeCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.DescribeCollectionResponse{
@ -1377,7 +1377,7 @@ func (c *Core) DescribeCollection(ctx context.Context, in *milvuspb.DescribeColl
}, nil
}
log.Debug("DescribeCollection Success", zap.String("name", in.CollectionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordDescribeCollectionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordDescribeCollectionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
t.Rsp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1387,7 +1387,7 @@ func (c *Core) DescribeCollection(ctx context.Context, in *milvuspb.DescribeColl
}
func (c *Core) ShowCollections(ctx context.Context, in *milvuspb.ShowCollectionsRequest) (*milvuspb.ShowCollectionsResponse, error) {
metrics.RootCoordShowCollectionsCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordShowCollectionsCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.ShowCollectionsResponse{
@ -1424,7 +1424,7 @@ func (c *Core) ShowCollections(ctx context.Context, in *milvuspb.ShowCollections
}, nil
}
log.Debug("ShowCollections Success", zap.String("dbname", in.DbName), zap.Strings("collection names", t.Rsp.CollectionNames), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordShowCollectionsCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordShowCollectionsCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
t.Rsp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1433,7 +1433,7 @@ func (c *Core) ShowCollections(ctx context.Context, in *milvuspb.ShowCollections
}
func (c *Core) CreatePartition(ctx context.Context, in *milvuspb.CreatePartitionRequest) (*commonpb.Status, error) {
metrics.RootCoordCreatePartitionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordCreatePartitionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &commonpb.Status{
@ -1460,7 +1460,7 @@ func (c *Core) CreatePartition(ctx context.Context, in *milvuspb.CreatePartition
}, nil
}
log.Debug("CreatePartition Success", zap.String("collection name", in.CollectionName), zap.String("partition name", in.PartitionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordCreatePartitionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordCreatePartitionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1468,7 +1468,7 @@ func (c *Core) CreatePartition(ctx context.Context, in *milvuspb.CreatePartition
}
func (c *Core) DropPartition(ctx context.Context, in *milvuspb.DropPartitionRequest) (*commonpb.Status, error) {
metrics.RootCoordDropPartitionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordDropPartitionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &commonpb.Status{
@ -1495,7 +1495,7 @@ func (c *Core) DropPartition(ctx context.Context, in *milvuspb.DropPartitionRequ
}, nil
}
log.Debug("DropPartition Success", zap.String("collection name", in.CollectionName), zap.String("partition name", in.PartitionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordDropPartitionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordDropPartitionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1503,7 +1503,7 @@ func (c *Core) DropPartition(ctx context.Context, in *milvuspb.DropPartitionRequ
}
func (c *Core) HasPartition(ctx context.Context, in *milvuspb.HasPartitionRequest) (*milvuspb.BoolResponse, error) {
metrics.RootCoordHasPartitionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordHasPartitionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.BoolResponse{
@ -1537,7 +1537,7 @@ func (c *Core) HasPartition(ctx context.Context, in *milvuspb.HasPartitionReques
}, nil
}
log.Debug("HasPartition Success", zap.String("collection name", in.CollectionName), zap.String("partition name", in.PartitionName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordHasPartitionCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordHasPartitionCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &milvuspb.BoolResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
@ -1548,7 +1548,7 @@ func (c *Core) HasPartition(ctx context.Context, in *milvuspb.HasPartitionReques
}
func (c *Core) ShowPartitions(ctx context.Context, in *milvuspb.ShowPartitionsRequest) (*milvuspb.ShowPartitionsResponse, error) {
metrics.RootCoordShowPartitionsCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordShowPartitionsCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
log.Debug("ShowPartitionRequest received", zap.String("role", Params.RoleName), zap.Int64("msgID", in.Base.MsgID),
zap.String("collection", in.CollectionName))
code := c.stateCode.Load().(internalpb.StateCode)
@ -1591,7 +1591,7 @@ func (c *Core) ShowPartitions(ctx context.Context, in *milvuspb.ShowPartitionsRe
log.Debug("ShowPartitions succeed", zap.String("role", Params.RoleName), zap.Int64("msgID", t.Req.Base.MsgID),
zap.String("collection name", in.CollectionName), zap.Strings("partition names", t.Rsp.PartitionNames),
zap.Int64s("partition ids", t.Rsp.PartitionIDs))
metrics.RootCoordShowPartitionsCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordShowPartitionsCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
t.Rsp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1600,7 +1600,7 @@ func (c *Core) ShowPartitions(ctx context.Context, in *milvuspb.ShowPartitionsRe
}
func (c *Core) CreateIndex(ctx context.Context, in *milvuspb.CreateIndexRequest) (*commonpb.Status, error) {
metrics.RootCoordCreateIndexCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordCreateIndexCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &commonpb.Status{
@ -1627,7 +1627,7 @@ func (c *Core) CreateIndex(ctx context.Context, in *milvuspb.CreateIndexRequest)
}, nil
}
log.Debug("CreateIndex Success", zap.String("collection name", in.CollectionName), zap.String("field name", in.FieldName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordCreateIndexCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordCreateIndexCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1635,7 +1635,7 @@ func (c *Core) CreateIndex(ctx context.Context, in *milvuspb.CreateIndexRequest)
}
func (c *Core) DescribeIndex(ctx context.Context, in *milvuspb.DescribeIndexRequest) (*milvuspb.DescribeIndexResponse, error) {
metrics.RootCoordDescribeIndexCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordDescribeIndexCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.DescribeIndexResponse{
@ -1676,7 +1676,7 @@ func (c *Core) DescribeIndex(ctx context.Context, in *milvuspb.DescribeIndexRequ
idxNames = append(idxNames, i.IndexName)
}
log.Debug("DescribeIndex Success", zap.String("collection name", in.CollectionName), zap.String("field name", in.FieldName), zap.Strings("index names", idxNames), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordDescribeIndexCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordDescribeIndexCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
if len(t.Rsp.IndexDescriptions) == 0 {
t.Rsp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_IndexNotExist,
@ -1692,7 +1692,7 @@ func (c *Core) DescribeIndex(ctx context.Context, in *milvuspb.DescribeIndexRequ
}
func (c *Core) DropIndex(ctx context.Context, in *milvuspb.DropIndexRequest) (*commonpb.Status, error) {
metrics.RootCoordDropIndexCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordDropIndexCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &commonpb.Status{
@ -1719,7 +1719,7 @@ func (c *Core) DropIndex(ctx context.Context, in *milvuspb.DropIndexRequest) (*c
}, nil
}
log.Debug("DropIndex Success", zap.String("collection name", in.CollectionName), zap.String("field name", in.FieldName), zap.String("index name", in.IndexName), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordDropIndexCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordDropIndexCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1727,7 +1727,7 @@ func (c *Core) DropIndex(ctx context.Context, in *milvuspb.DropIndexRequest) (*c
}
func (c *Core) DescribeSegment(ctx context.Context, in *milvuspb.DescribeSegmentRequest) (*milvuspb.DescribeSegmentResponse, error) {
metrics.RootCoordDescribeSegmentCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordDescribeSegmentCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.DescribeSegmentResponse{
@ -1764,7 +1764,7 @@ func (c *Core) DescribeSegment(ctx context.Context, in *milvuspb.DescribeSegment
}, nil
}
log.Debug("DescribeSegment Success", zap.Int64("collection id", in.CollectionID), zap.Int64("segment id", in.SegmentID), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordDescribeSegmentCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordDescribeSegmentCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
t.Rsp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",
@ -1773,7 +1773,7 @@ func (c *Core) DescribeSegment(ctx context.Context, in *milvuspb.DescribeSegment
}
func (c *Core) ShowSegments(ctx context.Context, in *milvuspb.ShowSegmentsRequest) (*milvuspb.ShowSegmentsResponse, error) {
metrics.RootCoordShowSegmentsCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsTotal).Inc()
metrics.RootCoordShowSegmentsCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsTotal).Inc()
code := c.stateCode.Load().(internalpb.StateCode)
if code != internalpb.StateCode_Healthy {
return &milvuspb.ShowSegmentsResponse{
@ -1810,7 +1810,7 @@ func (c *Core) ShowSegments(ctx context.Context, in *milvuspb.ShowSegmentsReques
}, nil
}
log.Debug("ShowSegments Success", zap.Int64("collection id", in.CollectionID), zap.Int64("partition id", in.PartitionID), zap.Int64s("segments ids", t.Rsp.SegmentIDs), zap.Int64("msgID", in.Base.MsgID))
metrics.RootCoordShowSegmentsCounter.WithLabelValues(metricProxyNode(in.Base.SourceID), MetricRequestsSuccess).Inc()
metrics.RootCoordShowSegmentsCounter.WithLabelValues(metricProxy(in.Base.SourceID), MetricRequestsSuccess).Inc()
t.Rsp.Status = &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: "",

View File

@ -41,13 +41,13 @@ import (
"go.etcd.io/etcd/clientv3"
)
type proxyNodeMock struct {
types.ProxyNode
type proxyMock struct {
types.Proxy
collArray []string
mutex sync.Mutex
}
func (p *proxyNodeMock) InvalidateCollectionMetaCache(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) {
func (p *proxyMock) InvalidateCollectionMetaCache(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error) {
p.mutex.Lock()
defer p.mutex.Unlock()
p.collArray = append(p.collArray, request.CollectionName)
@ -55,7 +55,7 @@ func (p *proxyNodeMock) InvalidateCollectionMetaCache(ctx context.Context, reque
ErrorCode: commonpb.ErrorCode_Success,
}, nil
}
func (p *proxyNodeMock) GetCollArray() []string {
func (p *proxyMock) GetCollArray() []string {
p.mutex.Lock()
defer p.mutex.Unlock()
ret := make([]string, 0, len(p.collArray))
@ -305,11 +305,11 @@ func TestMasterService(t *testing.T) {
_, err = etcdCli.Put(ctx, path.Join(sessKey, typeutil.ProxyRole+"-100"), string(pnb))
assert.Nil(t, err)
pnm := &proxyNodeMock{
pnm := &proxyMock{
collArray: make([]string, 0, 16),
mutex: sync.Mutex{},
}
core.NewProxyClient = func(*sessionutil.Session) (types.ProxyNode, error) {
core.NewProxyClient = func(*sessionutil.Session) (types.Proxy, error) {
return pnm, nil
}
@ -1469,15 +1469,15 @@ func TestMasterService(t *testing.T) {
t.Run("channel timetick", func(t *testing.T) {
const (
proxyNodeIDInvalid = 102
proxyNodeName0 = "proxynode_0"
proxyNodeName1 = "proxynode_1"
chanName0 = "c0"
chanName1 = "c1"
chanName2 = "c2"
ts0 = uint64(100)
ts1 = uint64(120)
ts2 = uint64(150)
proxyIDInvalid = 102
proxyName0 = "proxy_0"
proxyName1 = "proxy_1"
chanName0 = "c0"
chanName1 = "c1"
chanName2 = "c2"
ts0 = uint64(100)
ts1 = uint64(120)
ts2 = uint64(150)
)
p1 := sessionutil.Session{
ServerID: 100,
@ -1529,7 +1529,7 @@ func TestMasterService(t *testing.T) {
msgInvalid := &internalpb.ChannelTimeTickMsg{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_TimeTick,
SourceID: proxyNodeIDInvalid,
SourceID: proxyIDInvalid,
},
ChannelNames: []string{"test"},
Timestamps: []uint64{0},
@ -1539,7 +1539,7 @@ func TestMasterService(t *testing.T) {
time.Sleep(1 * time.Second)
// 2 proxy nodes, 1 master
assert.Equal(t, 3, core.chanTimeTick.GetProxyNodeNum())
assert.Equal(t, 3, core.chanTimeTick.GetProxyNum())
// 3 proxy node channels, 2 master channels
assert.Equal(t, 5, core.chanTimeTick.GetChanNum())
@ -1790,7 +1790,7 @@ func TestMasterService2(t *testing.T) {
err = core.SetQueryCoord(qm)
assert.Nil(t, err)
core.NewProxyClient = func(*sessionutil.Session) (types.ProxyNode, error) {
core.NewProxyClient = func(*sessionutil.Session) (types.Proxy, error) {
return nil, nil
}
@ -1961,7 +1961,7 @@ func TestCheckInit(t *testing.T) {
err = c.checkInit()
assert.NotNil(t, err)
c.NewProxyClient = func(*sessionutil.Session) (types.ProxyNode, error) {
c.NewProxyClient = func(*sessionutil.Session) (types.Proxy, error) {
return nil, nil
}
err = c.checkInit()

View File

@ -101,11 +101,11 @@ func BenchmarkAllocTimestamp(b *testing.B) {
err = core.Register()
assert.Nil(b, err)
pnm := &proxyNodeMock{
pnm := &proxyMock{
collArray: make([]string, 0, 16),
mutex: sync.Mutex{},
}
core.NewProxyClient = func(*sessionutil.Session) (types.ProxyNode, error) {
core.NewProxyClient = func(*sessionutil.Session) (types.Proxy, error) {
return pnm, nil
}

View File

@ -118,13 +118,13 @@ func (t *timetickSync) UpdateTimeTick(in *internalpb.ChannelTimeTickMsg) error {
return nil
}
func (t *timetickSync) AddProxyNode(sess *sessionutil.Session) {
func (t *timetickSync) AddProxy(sess *sessionutil.Session) {
t.lock.Lock()
defer t.lock.Unlock()
t.proxyTimeTick[sess.ServerID] = nil
}
func (t *timetickSync) DelProxyNode(sess *sessionutil.Session) {
func (t *timetickSync) DelProxy(sess *sessionutil.Session) {
t.lock.Lock()
defer t.lock.Unlock()
if _, ok := t.proxyTimeTick[sess.ServerID]; ok {
@ -133,7 +133,7 @@ func (t *timetickSync) DelProxyNode(sess *sessionutil.Session) {
}
}
func (t *timetickSync) GetProxyNodes(sess []*sessionutil.Session) {
func (t *timetickSync) GetProxy(sess []*sessionutil.Session) {
t.lock.Lock()
defer t.lock.Unlock()
for _, s := range sess {
@ -146,7 +146,7 @@ func (t *timetickSync) StartWatch() {
for {
select {
case <-t.core.ctx.Done():
log.Debug("master service context done", zap.Error(t.core.ctx.Err()))
log.Debug("root coord context done", zap.Error(t.core.ctx.Err()))
return
case ptt, ok := <-t.sendChan:
if !ok {
@ -202,8 +202,8 @@ func (t *timetickSync) SendChannelTimeTick(chanName string, ts typeutil.Timestam
return err
}
// GetProxyNodeNum return the num of detected proxy node
func (t *timetickSync) GetProxyNodeNum() int {
// GetProxyNum return the num of detected proxy node
func (t *timetickSync) GetProxyNum() int {
t.lock.Lock()
defer t.lock.Unlock()
return len(t.proxyTimeTick)

View File

@ -111,7 +111,7 @@ type RootCoord interface {
ReleaseDQLMessageStream(ctx context.Context, in *proxypb.ReleaseDQLMessageStreamRequest) (*commonpb.Status, error)
}
// RootCoordComponent is used by grpc server of master service
// RootCoordComponent is used by grpc server of RootCoord
type RootCoordComponent interface {
RootCoord
@ -119,10 +119,10 @@ type RootCoordComponent interface {
SetDataCoord(context.Context, DataCoord) error
SetIndexCoord(IndexCoord) error
SetQueryCoord(QueryCoord) error
SetNewProxyClient(func(sess *sessionutil.Session) (ProxyNode, error))
SetNewProxyClient(func(sess *sessionutil.Session) (Proxy, error))
}
type ProxyNode interface {
type Proxy interface {
Component
InvalidateCollectionMetaCache(ctx context.Context, request *proxypb.InvalidateCollMetaCacheRequest) (*commonpb.Status, error)

View File

@ -119,41 +119,41 @@ func (gp *BaseTable) tryloadFromEnv() {
panic(err)
}
masterAddress := os.Getenv("MASTER_ADDRESS")
if masterAddress == "" {
masterHost, err := gp.Load("master.address")
rootCoordAddress := os.Getenv("ROOT_COORD_ADDRESS")
if rootCoordAddress == "" {
rootCoordHost, err := gp.Load("rootCoord.address")
if err != nil {
panic(err)
}
port, err := gp.Load("master.port")
port, err := gp.Load("rootCoord.port")
if err != nil {
panic(err)
}
masterAddress = masterHost + ":" + port
rootCoordAddress = rootCoordHost + ":" + port
}
err = gp.Save("_MasterAddress", masterAddress)
err = gp.Save("_RootCoordAddress", rootCoordAddress)
if err != nil {
panic(err)
}
indexBuilderAddress := os.Getenv("INDEX_SERVICE_ADDRESS")
if indexBuilderAddress == "" {
indexBuilderHost, err := gp.Load("indexService.address")
indexCoordAddress := os.Getenv("INDEX_COORD_ADDRESS")
if indexCoordAddress == "" {
indexCoordHost, err := gp.Load("indexCoord.address")
if err != nil {
panic(err)
}
port, err := gp.Load("indexService.port")
port, err := gp.Load("indexCoord.port")
if err != nil {
panic(err)
}
indexBuilderAddress = indexBuilderHost + ":" + port
indexCoordAddress = indexCoordHost + ":" + port
}
err = gp.Save("IndexServiceAddress", indexBuilderAddress)
err = gp.Save("_IndexCoordAddress", indexCoordAddress)
if err != nil {
panic(err)
}
queryCoordAddress := os.Getenv("QUERY_SERVICE_ADDRESS")
queryCoordAddress := os.Getenv("QUERY_COORD_ADDRESS")
if queryCoordAddress == "" {
serviceHost, err := gp.Load("queryCoord.address")
if err != nil {
@ -170,7 +170,7 @@ func (gp *BaseTable) tryloadFromEnv() {
panic(err)
}
dataCoordAddress := os.Getenv("DATA_SERVICE_ADDRESS")
dataCoordAddress := os.Getenv("DATA_COORD_ADDRESS")
if dataCoordAddress == "" {
serviceHost, err := gp.Load("dataCoord.address")
if err != nil {

View File

@ -48,7 +48,7 @@ ${protoc} --go_out=plugins=grpc,paths=source_relative:./milvuspb milvus.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./proxypb proxy.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./indexpb index_coord.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./datapb data_coord.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./querypb query_service.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./querypb query_coord.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./planpb plan.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./segcorepb segcore.proto

View File

@ -30,7 +30,7 @@ go test -race -cover "${MILVUS_DIR}/util/trace/..." -failfast
go test -race -cover "${MILVUS_DIR}/util/typeutil/..." -failfast
# TODO: remove to distributed
#go test -race -cover "${MILVUS_DIR}/proxynode/..." -failfast
#go test -race -cover "${MILVUS_DIR}/proxy/..." -failfast
go test -race -cover "${MILVUS_DIR}/datanode/..." -failfast
go test -race -cover "${MILVUS_DIR}/indexnode/..." -failfast
go test -race -cover "${MILVUS_DIR}/querynode/..." -failfast