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

fix(master): logic adapts for decommission parts of the dp on the node. #1000441588

Browse Source 
Signed-off-by: zhumingze <zhumingze@oppo.com>
This commit is contained in:
zhumingze 2025-11-17 19:03:39 +08:00 committed by 郑书强
parent 09e2646fac
commit 09adcf3912
3 changed files with 28 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
master/cluster.go
View File

@ -4792,6 +4792,7 @@ func (c *Cluster) checkDecommissionDataNode() {
// dataNode.resetDecommissionStatus()
// c.syncUpdateDataNode(dataNode)
// }
return true
}
// maybe has decommission failed dp
@ -5190,10 +5191,29 @@ func (c *Cluster) checkDecommissionDisk() {
c.DecommissionDisks.Range(func(key, value interface{}) bool {
disk := value.(*DecommissionDisk)
status := disk.GetDecommissionStatus()
dataNode, err := c.dataNode(disk.SrcAddr)
if err != nil {
if strings.Contains(err.Error(), "not found") {
c.DecommissionDisks.Delete(key)
c.syncDeleteDecommissionDisk(disk)
}
return true
}
if status == DecommissionSuccess || status == DecommissionFail {
partitions := dataNode.badPartitions(disk.DiskPath, c, true)
// if only decommission part of data partitions, do not add disk to success list
if len(partitions) != 0 {
if dataNode.DecommissionLimit != 0 && !dataNode.isBadDisk(disk.DiskPath) {
// can allocate dp again
c.deleteAndSyncDecommissionedDisk(dataNode, disk.DiskPath)
}
return true
}
}
// keep failed decommission disk in list for preventing the reuse of a
// term in future decommissioning operations
if status == DecommissionSuccess {
c.addAndSyncDecommissionSuccessDisk(disk.SrcAddr, disk.DiskPath)
c.addAndSyncDecommissionSuccessDisk(dataNode, disk.DiskPath)
if time.Since(time.Unix(disk.DecommissionCompleteTime, 0)) > (120 * time.Hour) {
if err := c.syncDeleteDecommissionDisk(disk); err != nil {
msg := fmt.Sprintf("action[checkDecommissionDisk],clusterID[%v] node[%v] disk[%v],"+

8
master/data_node.go
View File

@ -638,7 +638,7 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug, persist bo
if successDiskNum+failedDiskNum+cancelDiskNum == totalDisk {
if successDiskNum == totalDisk {
if persist {
dataNode.SetDecommissionStatus(DecommissionSuccess)
dataNode.markDecommissionSuccess(c)
}
return DecommissionSuccess, float64(1)
}
@ -650,7 +650,7 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug, persist bo
}
} else {
if persist {
dataNode.SetDecommissionStatus(DecommissionFail)
dataNode.markDecommissionFail()
} else {
return DecommissionFail, progress / float64(totalDisk)
}
@ -750,6 +750,10 @@ func (dataNode *DataNode) markDecommissionSuccess(c *Cluster) {
func (dataNode *DataNode) markDecommissionFail() {
dataNode.SetDecommissionStatus(DecommissionFail)
// if only decommission part of data partitions, can alloc dp in future
if dataNode.DecommissionLimit != 0 {
dataNode.ToBeOffline = false
}
// dataNode.ToBeOffline = false
// dataNode.DecommissionCompleteTime = time.Now().Unix()
}

8
master/disk_manager.go
View File

@ -277,13 +277,7 @@ func (c *Cluster) deleteAndSyncDecommissionedDisk(dataNode *DataNode, diskPath s
return
}
func (c *Cluster) addAndSyncDecommissionSuccessDisk(addr string, diskPath string) (err error) {
var dataNode *DataNode
if dataNode, err = c.dataNode(addr); err != nil {
log.LogWarnf("action[addAndSyncDecommissionSuccessDisk] cannot find dataNode[%s]", addr)
return
}
func (c *Cluster) addAndSyncDecommissionSuccessDisk(dataNode *DataNode, diskPath string) (err error) {
if exist := dataNode.addDecommissionSuccessDisk(diskPath); exist {
return
}