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

Fix unregister node info has no channel (#5699) 

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
pull/5779/head
congqixia 2021-06-09 19:32:48 +08:00 committed by zhenshan.cao
parent f1bb02f724
commit a3788a99c5
2 changed files with 12 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
internal/dataservice/cluster.go
View File

@ -170,7 +170,9 @@ func (c *cluster) register(n *datapb.DataNodeInfo) {
c.dataManager.register(n) c.dataManager.register(n)
cNodes, chanBuffer := c.dataManager.getDataNodes(true) cNodes, chanBuffer := c.dataManager.getDataNodes(true)
var rets []*datapb.DataNodeInfo var rets []*datapb.DataNodeInfo
log.Debug("before register policy applied", zap.Any("n.Channels", n.Channels), zap.Any("buffer", chanBuffer))
rets, chanBuffer = c.registerPolicy.apply(cNodes, n, chanBuffer) rets, chanBuffer = c.registerPolicy.apply(cNodes, n, chanBuffer)
log.Debug("after register policy applied", zap.Any("ret", rets), zap.Any("buffer", chanBuffer))
c.dataManager.updateDataNodes(rets, chanBuffer) c.dataManager.updateDataNodes(rets, chanBuffer)
rets = c.watch(rets) rets = c.watch(rets)
c.dataManager.updateDataNodes(rets, chanBuffer) c.dataManager.updateDataNodes(rets, chanBuffer)
@ -179,9 +181,14 @@ func (c *cluster) register(n *datapb.DataNodeInfo) {
func (c *cluster) unregister(n *datapb.DataNodeInfo) { func (c *cluster) unregister(n *datapb.DataNodeInfo) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
c.sessionManager.releaseSession(n.Address) c.sessionManager.releaseSession(n.Address)
c.dataManager.unregister(n) oldNode := c.dataManager.unregister(n)
if oldNode != nil {
n = oldNode
}
cNodes, chanBuffer := c.dataManager.getDataNodes(true) cNodes, chanBuffer := c.dataManager.getDataNodes(true)
log.Debug("before unregister policy applied", zap.Any("n.Channels", n.Channels), zap.Any("buffer", chanBuffer))
var rets []*datapb.DataNodeInfo var rets []*datapb.DataNodeInfo
if len(cNodes) == 0 { if len(cNodes) == 0 {
for _, chStat := range n.Channels { for _, chStat := range n.Channels {
@ -191,6 +198,7 @@ func (c *cluster) unregister(n *datapb.DataNodeInfo) {
} else { } else {
rets = c.unregisterPolicy.apply(cNodes, n) rets = c.unregisterPolicy.apply(cNodes, n)
} }
log.Debug("after register policy applied", zap.Any("ret", rets), zap.Any("buffer", chanBuffer))
c.dataManager.updateDataNodes(rets, chanBuffer) c.dataManager.updateDataNodes(rets, chanBuffer)
rets = c.watch(rets) rets = c.watch(rets)
c.dataManager.updateDataNodes(rets, chanBuffer) c.dataManager.updateDataNodes(rets, chanBuffer)

5
internal/dataservice/cluster_data_manager.go
View File

@ -158,13 +158,14 @@ func (c *clusterNodeManager) register(n *datapb.DataNodeInfo) {
c.updateMetrics() c.updateMetrics()
} }
func (c *clusterNodeManager) unregister(n *datapb.DataNodeInfo) { func (c *clusterNodeManager) unregister(n *datapb.DataNodeInfo) *datapb.DataNodeInfo {
node, ok := c.dataNodes[n.Address] node, ok := c.dataNodes[n.Address]
if !ok { if !ok {
return return nil
} }
node.status = offline node.status = offline
c.updateMetrics() c.updateMetrics()
return node.info
} }
func (c *clusterNodeManager) updateMetrics() { func (c *clusterNodeManager) updateMetrics() {