mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
fix(master): fix decommission priority problems.
close:#1000197987 Signed-off-by: shuqiang-zheng <zhengshuqiang@oppo.com>
This commit is contained in:
shuqiang-zheng
zhumingze1108
parent
bde40fb283
commit
fd7976866c
@ -2300,6 +2300,157 @@ func (l *DecommissionDataPartitionList) startTraverse() {
|
||||
l.start <- struct{}{}
|
||||
}
|
||||
|
||||
func updateDecommissionWeight(dps []*DataPartition, c *Cluster) {
|
||||
for _, dp := range dps {
|
||||
if dp.IsDiscard {
|
||||
dp.SetDecommissionStatus(DecommissionSuccess)
|
||||
log.LogWarnf("action[DecommissionListTraverse] skip dp(%v) discard(%v)", dp.PartitionID, dp.IsDiscard)
|
||||
continue
|
||||
}
|
||||
diskErrReplicaNum := dp.getReplicaDiskErrorNum()
|
||||
if diskErrReplicaNum == dp.ReplicaNum || diskErrReplicaNum == uint8(len(dp.Peers)) {
|
||||
log.LogWarnf("action[DecommissionListTraverse] dp[%v] all live replica is unavaliable", dp.decommissionInfo())
|
||||
err := proto.ErrAllReplicaUnavailable
|
||||
dp.DecommissionErrorMessage = err.Error()
|
||||
dp.markRollbackFailed(false)
|
||||
continue
|
||||
}
|
||||
if dp.DecommissionType == AutoDecommission && dp.IsMarkDecommission() {
|
||||
if dp.lostLeader(c) && !dp.DecommissionRaftForce {
|
||||
dp.DecommissionRaftForce = true
|
||||
log.LogWarnf("action[DecommissionListTraverse] change dp[%v] decommission raftForce from false to true", dp.decommissionInfo())
|
||||
}
|
||||
diskErrReplicas := dp.getAllDiskErrorReplica()
|
||||
if isReplicasContainsHost(diskErrReplicas, dp.DecommissionSrcAddr) {
|
||||
if dp.ReplicaNum == 3 {
|
||||
if (diskErrReplicaNum == 2 && len(dp.Hosts) == 3) || (diskErrReplicaNum == 1 && len(dp.Hosts) == 2) {
|
||||
dp.DecommissionWeight = highestPriorityDecommissionWeight
|
||||
} else if diskErrReplicaNum == 1 && len(dp.Hosts) == 3 {
|
||||
dp.DecommissionWeight = highPriorityDecommissionWeight
|
||||
}
|
||||
} else if dp.ReplicaNum == 2 {
|
||||
if diskErrReplicaNum == 1 && len(dp.Hosts) == 2 {
|
||||
dp.DecommissionWeight = highPriorityDecommissionWeight
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *DecommissionDataPartitionList) handleDpTraverseToReleaseToken(dp *DataPartition, c *Cluster) {
|
||||
if dp.IsDecommissionSuccess() {
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
dp.ReleaseDecommissionToken(c)
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
msg := fmt.Sprintf("ns %v(%p) dp %v decommission success, cost %v",
|
||||
l.nsId, l, dp.decommissionInfo(), time.Since(dp.RecoverStartTime))
|
||||
delete(dp.DecommissionDiskRetryMap, dp.DecommissionSrcAddr+"_"+dp.DecommissionSrcDiskPath)
|
||||
dp.ResetDecommissionStatus()
|
||||
dp.setRestoreReplicaStop()
|
||||
err := c.syncUpdateDataPartition(dp)
|
||||
if err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) Remove success dp[%v] failed for %v",
|
||||
l.nsId, l, dp.PartitionID, err)
|
||||
} else {
|
||||
log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for success",
|
||||
l.nsId, l, dp.PartitionID)
|
||||
}
|
||||
auditlog.LogMasterOp("TraverseDataPartition", msg, err)
|
||||
} else if dp.IsDecommissionFailed() {
|
||||
remove := false
|
||||
needRollback, needSkip := dp.tryRollback(c)
|
||||
if !needRollback {
|
||||
log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for fail",
|
||||
l.nsId, l, dp.PartitionID)
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
key := dp.DecommissionSrcAddr + "_" + dp.DecommissionSrcDiskPath
|
||||
if dp.DecommissionDiskRetryMap[key] >= math.MaxInt {
|
||||
dp.DecommissionDiskRetryMap[key] = 0
|
||||
} else {
|
||||
dp.DecommissionDiskRetryMap[key]++
|
||||
}
|
||||
// if dp is not removed from decommission list, do not reset RestoreReplica
|
||||
dp.setRestoreReplicaStop()
|
||||
remove = true
|
||||
}
|
||||
if needSkip {
|
||||
continue
|
||||
}
|
||||
// rollback fail/success need release token
|
||||
dp.ReleaseDecommissionToken(c)
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
c.syncUpdateDataPartition(dp)
|
||||
msg := fmt.Sprintf("ns %v(%p) dp %v decommission failed, remove %v", l.nsId, l, dp.decommissionInfo(), remove)
|
||||
auditlog.LogMasterOp("TraverseDataPartition", msg, nil)
|
||||
} else if dp.IsDecommissionPaused() {
|
||||
log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for paused ",
|
||||
l.nsId, l, dp.PartitionID)
|
||||
dp.ReleaseDecommissionToken(c)
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
dp.setRestoreReplicaStop()
|
||||
c.syncUpdateDataPartition(dp)
|
||||
} else if dp.IsDecommissionInitial() { // fixed done ,not release token
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
dp.ResetDecommissionStatus()
|
||||
c.syncUpdateDataPartition(dp)
|
||||
}
|
||||
}
|
||||
|
||||
func (l *DecommissionDataPartitionList) handleDpTraverseToDecommission(dp *DataPartition, c *Cluster) {
|
||||
if dp.IsMarkDecommission() {
|
||||
if time.Since(dp.DecommissionRetryTime) < defaultDecommissionRetryInternal {
|
||||
log.LogWarnf("[traverse] dp %v should wait for decommissionRetry,lastDecommissionRetryTime %v", dp.PartitionID, dp.DecommissionRetryTime)
|
||||
return
|
||||
}
|
||||
if dp.AcquireDecommissionFirstHostToken(c) {
|
||||
if dp.TryAcquireDecommissionToken(c) {
|
||||
go func(dp *DataPartition) {
|
||||
dp.TryToDecommission(c)
|
||||
}(dp) // special replica cnt cost some time from prepare to running
|
||||
} else {
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *DecommissionDataPartitionList) traverseDps(dps []*DataPartition, c *Cluster, isReleaseToken bool) {
|
||||
for _, dp := range dps {
|
||||
select {
|
||||
case <-l.done:
|
||||
log.LogWarnf("ns %v(%p) traverse exit!", l.nsId, l)
|
||||
l.Clear()
|
||||
return
|
||||
case <-c.stopc:
|
||||
log.LogWarnf("ns %v(%p) cluster stopped! traverse exit!", l.nsId, l)
|
||||
l.Clear()
|
||||
return
|
||||
default:
|
||||
// process decommission
|
||||
}
|
||||
log.LogDebugf("[DecommissionListTraverse]ns %v(%p) traverse dp(%v)", l.nsId, l, dp.decommissionInfo())
|
||||
if isReleaseToken {
|
||||
l.handleDpTraverseToReleaseToken(dp, c)
|
||||
} else {
|
||||
l.handleDpTraverseToDecommission(dp, c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (l *DecommissionDataPartitionList) traverse(c *Cluster) {
|
||||
t := time.NewTicker(DecommissionInterval)
|
||||
// wait for loading all ap when reload metadata
|
||||
@ -2328,146 +2479,17 @@ func (l *DecommissionDataPartitionList) traverse(c *Cluster) {
|
||||
log.LogWarnf("ns %v(%p) Leader changed, stop traverseDataPartition!", l.nsId, l)
|
||||
return
|
||||
}
|
||||
|
||||
allDecommissionDP := l.GetAllDecommissionDataPartitions()
|
||||
updateDecommissionWeight(allDecommissionDP, c)
|
||||
l.traverseDps(allDecommissionDP, c, true)
|
||||
|
||||
for _, dp := range allDecommissionDP {
|
||||
if dp.IsDiscard {
|
||||
dp.SetDecommissionStatus(DecommissionSuccess)
|
||||
log.LogWarnf("action[DecommissionListTraverse] skip dp(%v) discard(%v)", dp.PartitionID, dp.IsDiscard)
|
||||
continue
|
||||
}
|
||||
diskErrReplicaNum := dp.getReplicaDiskErrorNum()
|
||||
if diskErrReplicaNum == dp.ReplicaNum || diskErrReplicaNum == uint8(len(dp.Peers)) {
|
||||
log.LogWarnf("action[DecommissionListTraverse] dp[%v] all live replica is unavaliable", dp.decommissionInfo())
|
||||
err := proto.ErrAllReplicaUnavailable
|
||||
dp.DecommissionErrorMessage = err.Error()
|
||||
dp.markRollbackFailed(false)
|
||||
continue
|
||||
}
|
||||
if dp.DecommissionType == AutoDecommission && dp.IsMarkDecommission() {
|
||||
if dp.lostLeader(c) && !dp.DecommissionRaftForce {
|
||||
dp.DecommissionRaftForce = true
|
||||
log.LogWarnf("action[DecommissionListTraverse] change dp[%v] decommission raftForce from false to true", dp.decommissionInfo())
|
||||
}
|
||||
diskErrReplicas := dp.getAllDiskErrorReplica()
|
||||
if isReplicasContainsHost(diskErrReplicas, dp.DecommissionSrcAddr) {
|
||||
if dp.ReplicaNum == 3 {
|
||||
if (diskErrReplicaNum == 2 && len(dp.Hosts) == 3) || (diskErrReplicaNum == 1 && len(dp.Hosts) == 2) {
|
||||
dp.DecommissionWeight = highestPriorityDecommissionWeight
|
||||
} else if diskErrReplicaNum == 1 && len(dp.Hosts) == 3 {
|
||||
dp.DecommissionWeight = highPriorityDecommissionWeight
|
||||
}
|
||||
} else if dp.ReplicaNum == 2 {
|
||||
if diskErrReplicaNum == 1 && len(dp.Hosts) == 2 {
|
||||
dp.DecommissionWeight = highPriorityDecommissionWeight
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
allDecommissionDP = l.GetAllDecommissionDataPartitions()
|
||||
sort.Slice(allDecommissionDP, func(i, j int) bool {
|
||||
return allDecommissionDP[i].DecommissionWeight > allDecommissionDP[j].DecommissionWeight
|
||||
})
|
||||
log.LogDebugf("[DecommissionListTraverse]ns %v(%p) traverse dp len (%v)", l.nsId, l, len(allDecommissionDP))
|
||||
for _, dp := range allDecommissionDP {
|
||||
select {
|
||||
case <-l.done:
|
||||
log.LogWarnf("ns %v(%p) traverse exit!", l.nsId, l)
|
||||
l.Clear()
|
||||
return
|
||||
case <-c.stopc:
|
||||
log.LogWarnf("ns %v(%p) cluster stopped! traverse exit!", l.nsId, l)
|
||||
l.Clear()
|
||||
return
|
||||
default:
|
||||
// process decommission
|
||||
}
|
||||
log.LogDebugf("[DecommissionListTraverse]ns %v(%p) traverse dp(%v)", l.nsId, l, dp.decommissionInfo())
|
||||
if dp.IsDecommissionSuccess() {
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
dp.ReleaseDecommissionToken(c)
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
msg := fmt.Sprintf("ns %v(%p) dp %v decommission success, cost %v",
|
||||
l.nsId, l, dp.decommissionInfo(), time.Since(dp.RecoverStartTime))
|
||||
delete(dp.DecommissionDiskRetryMap, dp.DecommissionSrcAddr+"_"+dp.DecommissionSrcDiskPath)
|
||||
dp.ResetDecommissionStatus()
|
||||
dp.setRestoreReplicaStop()
|
||||
err := c.syncUpdateDataPartition(dp)
|
||||
if err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) Remove success dp[%v] failed for %v",
|
||||
l.nsId, l, dp.PartitionID, err)
|
||||
} else {
|
||||
log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for success",
|
||||
l.nsId, l, dp.PartitionID)
|
||||
}
|
||||
auditlog.LogMasterOp("TraverseDataPartition", msg, err)
|
||||
} else if dp.IsDecommissionFailed() {
|
||||
remove := false
|
||||
needRollback, needSkip := dp.tryRollback(c)
|
||||
if !needRollback {
|
||||
log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for fail",
|
||||
l.nsId, l, dp.PartitionID)
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
key := dp.DecommissionSrcAddr + "_" + dp.DecommissionSrcDiskPath
|
||||
if dp.DecommissionDiskRetryMap[key] >= math.MaxInt {
|
||||
dp.DecommissionDiskRetryMap[key] = 0
|
||||
} else {
|
||||
dp.DecommissionDiskRetryMap[key]++
|
||||
}
|
||||
// if dp is not removed from decommission list, do not reset RestoreReplica
|
||||
dp.setRestoreReplicaStop()
|
||||
remove = true
|
||||
}
|
||||
if needSkip {
|
||||
continue
|
||||
}
|
||||
// rollback fail/success need release token
|
||||
dp.ReleaseDecommissionToken(c)
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
c.syncUpdateDataPartition(dp)
|
||||
msg := fmt.Sprintf("ns %v(%p) dp %v decommission failed, remove %v", l.nsId, l, dp.decommissionInfo(), remove)
|
||||
auditlog.LogMasterOp("TraverseDataPartition", msg, nil)
|
||||
} else if dp.IsDecommissionPaused() {
|
||||
log.LogDebugf("action[DecommissionListTraverse]ns %v(%p) Remove dp[%v] for paused ",
|
||||
l.nsId, l, dp.PartitionID)
|
||||
dp.ReleaseDecommissionToken(c)
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
dp.setRestoreReplicaStop()
|
||||
c.syncUpdateDataPartition(dp)
|
||||
} else if dp.IsDecommissionInitial() { // fixed done ,not release token
|
||||
if err := c.setDpRepairingStatus(dp, false); err != nil {
|
||||
log.LogWarnf("action[DecommissionListTraverse]ns %v(%p) dp[%v] set repairStatus to false failed, err %v", l.nsId, l, dp.decommissionInfo(), err)
|
||||
}
|
||||
l.Remove(dp)
|
||||
dp.ResetDecommissionStatus()
|
||||
c.syncUpdateDataPartition(dp)
|
||||
} else if dp.IsMarkDecommission() {
|
||||
if time.Since(dp.DecommissionRetryTime) < defaultDecommissionRetryInternal {
|
||||
log.LogWarnf("[traverse] dp %v should wait for decommissionRetry,lastDecommissionRetryTime %v", dp.PartitionID, dp.DecommissionRetryTime)
|
||||
continue
|
||||
}
|
||||
if dp.AcquireDecommissionFirstHostToken(c) {
|
||||
if dp.TryAcquireDecommissionToken(c) {
|
||||
go func(dp *DataPartition) {
|
||||
dp.TryToDecommission(c)
|
||||
}(dp) // special replica cnt cost some time from prepare to running
|
||||
} else {
|
||||
dp.ReleaseDecommissionFirstHostToken(c)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
l.traverseDps(allDecommissionDP, c, false)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
