gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

refactor: vol checkstatus task, delete data partition and meta partition synchronously

Browse Source 
Signed-off-by: zhuhyc <zzhniy.163.niy@163.com>
This commit is contained in:
zhuhyc 2019-12-30 16:10:54 +08:00
parent ad4d476eee
commit 77d92b16e0
1 changed files with 58 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
master/vol.go
View File

@ -473,11 +473,67 @@ func (vol *Vol) checkStatus(c *Cluster) {
log.LogInfof("action[volCheckStatus] vol[%v],status[%v]", vol.Name, vol.Status)
metaTasks := vol.getTasksToDeleteMetaPartitions()
dataTasks := vol.getTasksToDeleteDataPartitions()
if len(metaTasks) == 0 && len(dataTasks) == 0 {
vol.deleteVolFromStore(c)
}
c.addMetaNodeTasks(metaTasks)
c.addDataNodeTasks(dataTasks)
go func() {
for _, metaTask := range metaTasks {
vol.deleteMetaPartitionFromMetaNode(c, metaTask)
}
for _, dataTask := range dataTasks {
vol.deleteDataPartitionFromDataNode(c, dataTask)
}
}()
return
}
func (vol *Vol) deleteMetaPartitionFromMetaNode(c *Cluster, task *proto.AdminTask) {
mp, err := vol.metaPartition(task.PartitionID)
if err != nil {
return
}
metaNode, err := c.metaNode(task.OperatorAddr)
if err != nil {
return
}
_, err = metaNode.Sender.syncSendAdminTask(task)
if err != nil {
log.LogErrorf("action[deleteMetaPartition] vol[%v],meta partition[%v],err[%v]", mp.volName, mp.PartitionID, err)
return
}
mp.Lock()
mp.removeReplicaByAddr(metaNode.Addr)
mp.removeMissingReplica(metaNode.Addr)
mp.Unlock()
return
}
func (vol *Vol) deleteDataPartitionFromDataNode(c *Cluster, task *proto.AdminTask) {
dp, err := vol.getDataPartitionByID(task.PartitionID)
if err != nil {
return
}
dataNode, err := c.dataNode(task.OperatorAddr)
if err != nil {
return
}
_, err = dataNode.TaskManager.syncSendAdminTask(task)
if err != nil {
log.LogErrorf("action[deleteDataReplica] vol[%v],data partition[%v],err[%v]", dp.VolName, dp.PartitionID, err)
return
}
dp.Lock()
dp.removeReplicaByAddr(dataNode.Addr)
dp.checkAndRemoveMissReplica(dataNode.Addr)
if err = dp.update("deleteDataReplica", dp.VolName, dp.Peers, dp.Hosts, c); err != nil {
dp.Unlock()
return
}
dp.Unlock()
return
}