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): when rolling back special replica dp.

Browse Source 
only need to delete decommission src if new replica is recovered

Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-07-12 08:56:42 +08:00 committed by AmazingChi
parent e07aeac619
commit 44f9fce264
8 changed files with 41 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
master/admin_task_manager.go
View File

@ -17,6 +17,7 @@ package master
import (
"encoding/json"
"fmt"
"github.com/google/uuid"
"net"
"sync"
"time"
@ -108,7 +109,9 @@ func (sender *AdminTaskManager) getToBeDeletedTasks() (delTasks []*proto.AdminTa
}
func (sender *AdminTaskManager) doSendTasks() {
tasks := sender.getToDoTasks()
id := uuid.New()
log.LogDebugf("doSendTasks %v", id.String())
tasks := sender.getToDoTasks(id.String())
if len(tasks) == 0 {
return
}
@ -252,7 +255,7 @@ func (sender *AdminTaskManager) AddTask(t *proto.AdminTask) {
}
}
func (sender *AdminTaskManager) getToDoTasks() (tasks []*proto.AdminTask) {
func (sender *AdminTaskManager) getToDoTasks(id string) (tasks []*proto.AdminTask) {
sender.RLock()
defer sender.RUnlock()
tasks = make([]*proto.AdminTask, 0)
@ -262,6 +265,7 @@ func (sender *AdminTaskManager) getToDoTasks() (tasks []*proto.AdminTask) {
if t.IsHeartbeatTask() && t.CheckTaskNeedSend() {
tasks = append(tasks, t)
t.SendTime = time.Now().Unix()
log.LogDebugf("getToDoTasks get heartbeatTask %v %v", t.RequestID, id)
}
}
// send urgent task immediately

2
master/api_service.go
View File

@ -1635,7 +1635,6 @@ func (m *Server) addDataReplica(w http.ResponseWriter, r *http.Request) {
if !dp.setRestoreReplicaForbidden() {
retry++
if retry > defaultDecommissionRetryLimit {
} else {
err = errors.NewErrorf("set RestoreReplicaMetaForbidden failed")
sendErrReply(w, r, newErrHTTPReply(err))
return
@ -5680,6 +5679,7 @@ func (m *Server) queryDecommissionToken(w http.ResponseWriter, r *http.Request)
CurTokenNum: s.CurTokenNum,
MaxTokenNum: s.MaxTokenNum,
RunningDp: s.RunningDp,
TotalDP: s.TotalDP,
})
}
}

7
master/cluster.go
View File

@ -766,10 +766,15 @@ func (c *Cluster) checkLeaderAddr() {
func (c *Cluster) checkDataNodeHeartbeat() {
tasks := make([]*proto.AdminTask, 0)
id := uuid.New()
log.LogDebugf("checkDataNodeHeartbeat start %v", id.String())
c.dataNodes.Range(func(addr, dataNode interface{}) bool {
node := dataNode.(*DataNode)
node.checkLiveness()
log.LogDebugf("checkDataNodeHeartbeat checkLiveness for data node %v %v", node.Addr, id.String())
task := node.createHeartbeatTask(c.masterAddr(), c.diskQosEnable)
log.LogDebugf("checkDataNodeHeartbeat createHeartbeatTask for data node %v task %v %v", node.Addr,
task.RequestID, id.String())
hbReq := task.Request.(*proto.HeartBeatRequest)
c.volMutex.RLock()
defer c.volMutex.RUnlock()
@ -784,7 +789,9 @@ func (c *Cluster) checkDataNodeHeartbeat() {
tasks = append(tasks, task)
return true
})
log.LogDebugf("checkDataNodeHeartbeat add task %v", id.String())
c.addDataNodeTasks(tasks)
log.LogDebugf("checkDataNodeHeartbeat end %v", id.String())
}
func (c *Cluster) checkMetaNodeHeartbeat() {

12
master/data_node.go
View File

@ -417,6 +417,10 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug bool) (uint
successDiskNum = 0
failedDiskNum = 0
cancelDiskNum = 0
markDisks = make([]string, 0)
successDisks = make([]string, 0)
failedDisks = make([]string, 0)
cancelDisks = make([]string, 0)
progress float64
)
if dataNode.GetDecommissionStatus() == DecommissionInitial {
@ -457,12 +461,16 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug bool) (uint
status := dd.GetDecommissionStatus()
if status == DecommissionSuccess {
successDiskNum++
successDisks = append(successDisks, dd.DiskPath)
} else if status == markDecommission {
markDiskNum++
markDisks = append(markDisks, dd.DiskPath)
} else if status == DecommissionFail {
failedDiskNum++
failedDisks = append(failedDisks, dd.DiskPath)
} else if status == DecommissionCancel {
cancelDiskNum++
cancelDisks = append(cancelDisks, dd.DiskPath)
}
_, diskProgress := dd.updateDecommissionStatus(c, debug)
progress += diskProgress
@ -492,9 +500,9 @@ func (dataNode *DataNode) updateDecommissionStatus(c *Cluster, debug bool) (uint
if debug {
log.LogInfof("action[updateDecommissionStatus] dataNode[%v] progress[%v] DecommissionDiskNum[%v] "+
"DecommissionDisks %v markDiskNum[%v] successDiskNum[%v] failedDiskNum[%v] cancelDiskNum[%v]",
"DecommissionDisks %v markDiskNum[%v] %v successDiskNum[%v] %v failedDiskNum[%v] %v cancelDiskNum[%v] %v",
dataNode.Addr, progress/float64(totalDisk), len(dataNode.DecommissionDiskList), dataNode.DecommissionDiskList, markDiskNum,
successDiskNum, failedDiskNum, cancelDiskNum)
markDisks, successDiskNum, successDisks, failedDiskNum, failedDisks, cancelDiskNum, cancelDisks)
}
return dataNode.GetDecommissionStatus(), progress / float64(totalDisk)
}

10
master/data_partition.go
View File

@ -1875,10 +1875,16 @@ func (partition *DataPartition) needRollback(c *Cluster) bool {
log.LogWarnf("action[rollback]dp[%v] rollback to del from bad dataPartitionIDs failed:%v", partition.PartitionID, err)
}
partition.DecommissionNeedRollback = false
err = partition.removeReplicaByForce(c, partition.DecommissionDstAddr)
removeAddr := partition.DecommissionDstAddr
// when special replica partition enter SpecialDecommissionWaitAddResFin, new replica is recoverd, so only
// need to delete DecommissionSrcAddr
if partition.isSpecialReplicaCnt() && partition.GetSpecialReplicaDecommissionStep() >= SpecialDecommissionWaitAddResFin {
removeAddr = partition.DecommissionSrcAddr
}
err = partition.removeReplicaByForce(c, removeAddr)
if err != nil {
log.LogWarnf("action[needRollback]dp[%v] remove decommission dst replica %v failed: %v",
partition.PartitionID, partition.DecommissionDstAddr, err)
partition.PartitionID, removeAddr, err)
}
c.syncUpdateDataPartition(partition)
auditlog.LogMasterOp("DataPartitionDecommissionRollback",

1
master/disk_manager.go
View File

@ -527,6 +527,7 @@ func (dd *DecommissionDisk) cancelDecommission(cluster *Cluster, ns *nodeSet) (e
msg := fmt.Sprintf("dp(%v) cancel decommission", dp.decommissionInfo())
dp.ResetDecommissionStatus()
dp.setRestoreReplicaStop()
cluster.syncUpdateDataPartition(dp)
auditlog.LogMasterOp("CancelDataPartitionDecommission", msg, nil)
}
dd.SetDecommissionStatus(DecommissionCancel)

12
master/topology.go
View File

@ -960,6 +960,7 @@ type nodeSetDecommissionParallelStatus struct {
CurTokenNum int32
MaxTokenNum int32
RunningDp []uint64
TotalDP int
}
func newNodeSet(c *Cluster, id uint64, cap int, zoneName string) *nodeSet {
@ -1085,7 +1086,7 @@ func (ns *nodeSet) UpdateMaxParallel(maxParallel int32) {
atomic.StoreInt32(&ns.decommissionParallelLimit, maxParallel)
}
func (ns *nodeSet) getDecommissionParallelStatus() (int32, int32, []uint64) {
func (ns *nodeSet) getDecommissionParallelStatus() (int32, int32, []uint64, int) {
return ns.decommissionDataPartitionList.getDecommissionParallelStatus()
}
@ -1867,12 +1868,13 @@ func (zone *Zone) queryDecommissionParallelStatus() (err error, stats []nodeSetD
}
for _, ns := range nodeSets {
curToken, maxToken, dps := ns.getDecommissionParallelStatus()
curToken, maxToken, dps, total := ns.getDecommissionParallelStatus()
stat := nodeSetDecommissionParallelStatus{
ID: ns.ID,
CurTokenNum: curToken,
MaxTokenNum: maxToken,
RunningDp: dps,
TotalDP: total,
}
stats = append(stats, stat)
}
@ -2005,15 +2007,15 @@ func (l *DecommissionDataPartitionList) Remove(value *DataPartition) {
}
}
func (l *DecommissionDataPartitionList) getDecommissionParallelStatus() (int32, int32, []uint64) {
func (l *DecommissionDataPartitionList) getDecommissionParallelStatus() (int32, int32, []uint64, int) {
l.mu.Lock()
defer l.mu.Unlock()
dps := make([]uint64, 0)
for id := range l.runningMap {
dps = append(dps, id)
}
return atomic.LoadInt32(&l.curParallel), atomic.LoadInt32(&l.parallelLimit), dps
total := l.decommissionList.Len()
return atomic.LoadInt32(&l.curParallel), atomic.LoadInt32(&l.parallelLimit), dps, total
}
func (l *DecommissionDataPartitionList) updateMaxParallel(maxParallel int32) {

1
proto/model.go
View File

@ -413,6 +413,7 @@ type DecommissionTokenStatus struct {
CurTokenNum int32
MaxTokenNum int32
RunningDp []uint64
TotalDP int
}
type VolVersionInfo struct {