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): modify alignment logic for new replica decommission progress to improve accuracy.

Browse Source 
close:#1000121688

Signed-off-by: shuqiang-zheng <zhengshuqiang@oppo.com>
This commit is contained in:
shuqiang-zheng 2025-05-27 15:02:49 +08:00 committed by zhumingze1108
parent 72089c6f0c
commit 36118447cd
1 changed files with 40 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
datanode/partition_raft.go
View File

@ -18,6 +18,7 @@ import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/cubefs/cubefs/datanode/storage"
syslog "log"
"net"
"os"
@ -285,6 +286,22 @@ func (dp *DataPartition) StartRaftLoggingSchedule() {
}
}
func compareExtentsBySize(toCompareExtents, baseExtents []*storage.ExtentInfo) bool {
for _, extent := range toCompareExtents {
found := false
for _, base := range baseExtents {
if base.Size == extent.Size {
found = true
break
}
}
if !found {
return false
}
}
return true
}
// StartRaftAfterRepair starts the raft after repairing a partition.
// It can only happens after all the extent files are repaired by the leader.
// When the repair is finished, the local dp.partitionSize is same as the leader's dp.partitionSize.
@ -295,6 +312,8 @@ func (dp *DataPartition) StartRaftAfterRepair(isLoad bool) {
initPartitionSize, initMaxExtentID uint64
currLeaderPartitionSize uint64
currLeaderRealUsedSize uint64
currLeaderPartitionNormalExtents []*storage.ExtentInfo
localPartitionNormalExtents []*storage.ExtentInfo
err error
)
timer := time.NewTicker(5 * time.Second)
@ -355,6 +374,22 @@ func (dp *DataPartition) StartRaftAfterRepair(isLoad bool) {
continue
}
currLeaderPartitionNormalExtents, err = dp.getLeaderPartitionNormalExtentInfo()
if err != nil {
log.LogErrorf("action[StartRaftAfterRepair] PartitionID(%v) get leader normal extents err(%v)", dp.partitionID, err)
continue
}
localPartitionNormalExtents, _, err = dp.getLocalExtentInfo(proto.NormalExtentType, nil)
if err != nil {
log.LogErrorf("action[StartRaftAfterRepair] PartitionID(%v) get local normal extents err(%v)", dp.partitionID, err)
continue
}
if !compareExtentsBySize(currLeaderPartitionNormalExtents, localPartitionNormalExtents) {
log.LogWarnf("action[StartRaftAfterRepair] PartitionID(%v) leader normal extent incomplete match local normal extent, wait snapshot recover", dp.partitionID)
continue
}
if err := dp.StartRaft(isLoad); err != nil {
log.LogErrorf("action[StartRaftAfterRepair] dp(%v) start raft err(%v)", dp.info(), err)
dp.DataPartitionCreateType = proto.NormalCreateDataPartition
@ -689,6 +724,11 @@ func (dp *DataPartition) findMaxAppliedID(allAppliedIDs []uint64) (maxAppliedID
return maxAppliedID, index
}
func (dp *DataPartition) getLeaderPartitionNormalExtentInfo() (extentInfos []*storage.ExtentInfo, err error) {
target := dp.getReplicaAddr(0)
return dp.getRemoteExtentInfo(proto.NormalExtentType, nil, target)
}
// Get the partition size from the leader.
func (dp *DataPartition) getLeaderPartitionSize(maxExtentID uint64) (logicSize, realUsedSize uint64, err error) {
var conn *net.TCPConn