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):if all replica is unvaliable, mark dp as failed

Browse Source 
Signed-off-by: chihe <chihe@oppo.com>
This commit is contained in:
chihe 2024-06-27 14:51:23 +08:00 committed by AmazingChi
parent 942adaed7d
commit f36c40c07f
3 changed files with 5 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
master/api_service.go
View File

@ -1935,6 +1935,7 @@ func (m *Server) decommissionDataPartition(w http.ResponseWriter, r *http.Reques
sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: rstMsg})
return
}
auditlog.LogMasterOp("DataPartitionDecommission", fmt.Sprintf("decommission dp %v by manual", dp.decommissionInfo()), nil)
rstMsg = fmt.Sprintf(proto.AdminDecommissionDataPartition+" dataPartitionID :%v on node:%v successfully", partitionID, addr)
sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
}

18
master/cluster.go
View File

@ -4473,13 +4473,8 @@ func (c *Cluster) TryDecommissionDisk(disk *DecommissionDisk) {
// mark as failed and set decommission src, make sure it can be included in the calculation of progress
dp.DecommissionSrcAddr = node.Addr
dp.DecommissionSrcDiskPath = disk.DiskPath
if strings.Contains(err.Error(), proto.ErrAllReplicaUnavailable.Error()) {
dp.DecommissionNeedRollbackTimes = defaultDecommissionRollbackLimit
dp.DecommissionNeedRollback = false
} else {
dp.markRollbackFailed(false)
dp.DecommissionErrorMessage = err.Error()
}
dp.markRollbackFailed(false)
dp.DecommissionErrorMessage = err.Error()
dp.DecommissionTerm = disk.DecommissionTerm
log.LogWarnf("action[TryDecommissionDisk] disk(%v) set dp(%v) DecommissionTerm %v",
disk.decommissionInfo(), dp.PartitionID, disk.DecommissionTerm)
@ -5004,13 +4999,8 @@ func (c *Cluster) markDecommissionDataPartition(dp *DataPartition, src *DataNode
if err = dp.MarkDecommissionStatus(addr, "", replica.DiskPath, raftForce, uint64(time.Now().Unix()), migrateType, c, ns); err != nil {
if !strings.Contains(err.Error(), proto.ErrDecommissionDiskErrDPFirst.Error()) {
if strings.Contains(err.Error(), proto.ErrAllReplicaUnavailable.Error()) {
dp.DecommissionNeedRollbackTimes = defaultDecommissionRollbackLimit
dp.DecommissionNeedRollback = false
} else {
dp.markRollbackFailed(false)
dp.DecommissionErrorMessage = err.Error()
}
dp.markRollbackFailed(false)
dp.DecommissionErrorMessage = err.Error()
c.syncUpdateDataPartition(dp)
return
}

2
master/data_partition.go
View File

@ -1082,7 +1082,6 @@ func (partition *DataPartition) MarkDecommissionStatus(srcAddr, dstAddr, srcDisk
if partition.getReplicaDiskErrorNum() == partition.ReplicaNum {
log.LogWarnf("action[MarkDecommissionStatus] dp[%v] all replica is unavaliable, cannot handle in auto decommission mode",
partition.PartitionID)
partition.DecommissionErrorMessage = "all replica is unavailable, cannot handle in auto decommission mode"
return proto.ErrAllReplicaUnavailable
}
raftForce = true
@ -1121,7 +1120,6 @@ func (partition *DataPartition) MarkDecommissionStatus(srcAddr, dstAddr, srcDisk
if partition.getReplicaDiskErrorNum() == partition.ReplicaNum {
log.LogWarnf("action[MarkDecommissionStatus] dp[%v] all replica is unavaliable, cannot handle in manual decommission mode",
partition.PartitionID)
partition.DecommissionErrorMessage = "all replica is unavailable, cannot handle in manual decommission mode"
return proto.ErrAllReplicaUnavailable
}
}