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: handle dataPartition raft apply error

Browse Source 
Signed-off-by: wanghongyan <wanghongyan@oppo.com>
This commit is contained in:
wanghongyan 2021-01-13 08:51:21 +00:00 committed by Shuoran Liu
parent 8f5bbf4850
commit 7a056735ec
3 changed files with 26 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
datanode/partition.go
View File

@ -549,6 +549,14 @@ func (dp *DataPartition) checkIsDiskError(err error) (diskError bool) {
return
}
func newRaftApplyError(err error) error {
return errors.NewErrorf("[Custom Error]: unhandled raft apply error, err(%s)", err)
}
func isRaftApplyError(errMsg string) bool {
return strings.Contains(errMsg, "[Custom Error]: unhandled raft apply error")
}
// String returns the string format of the data partition information.
func (dp *DataPartition) String() (m string) {
return fmt.Sprintf(DataPartitionPrefix+"_%v_%v", dp.partitionID, dp.partitionSize)

12
datanode/partition_op_by_raft.go
View File

@ -220,14 +220,10 @@ func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (r
resp = proto.OpOk
dp.uploadApplyID(raftApplyID)
} else {
// only stop myself partition
if !IsDiskErr(err.Error()) {
dp.stopRaft()
}
err = fmt.Errorf("[ApplyRandomWrite] ApplyID(%v) Partition(%v)_Extent(%v)_ExtentOffset(%v)_Size(%v) apply err(%v) retry[20]", raftApplyID, dp.partitionID, opItem.extentID, opItem.offset, opItem.size, err)
exporter.Warning(err.Error())
resp = proto.OpDiskErr
panic(newRaftApplyError(err))
}
}()
if dp.IsRejectWrite() {
@ -244,12 +240,12 @@ func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (r
for i := 0; i < 20; i++ {
err = dp.ExtentStore().Write(opItem.extentID, opItem.offset, opItem.size, opItem.data, opItem.crc, storage.RandomWriteType, opItem.opcode == proto.OpSyncRandomWrite)
if dp.checkIsDiskError(err) {
return
}
if err == nil {
break
}
if IsDiskErr(err.Error()) {
panic(newRaftApplyError(err))
}
if strings.Contains(err.Error(), storage.ExtentNotFoundError.Error()) {
err = nil
return

21
datanode/partition_raftfsm.go
View File

@ -45,13 +45,9 @@ func (dp *DataPartition) ApplyMemberChange(confChange *raftproto.ConfChange, ind
if err == nil {
dp.uploadApplyID(index)
} else {
// only stop myself partition
if !IsDiskErr(err.Error()) {
dp.stopRaft()
}
err = fmt.Errorf("[ApplyMemberChange] ApplyID(%v) Partition(%v) apply err(%v)]", index, dp.partitionID, err)
exporter.Warning(err.Error())
panic(newRaftApplyError(err))
}
}(index)
@ -94,13 +90,18 @@ func (dp *DataPartition) ApplyMemberChange(confChange *raftproto.ConfChange, ind
}
if err != nil {
log.LogErrorf("action[ApplyMemberChange] dp(%v) type(%v) err(%v).", dp.partitionID, confChange.Type, err)
if IsDiskErr(err.Error()) {
panic(newRaftApplyError(err))
}
return
}
if isUpdated {
dp.DataPartitionCreateType = proto.NormalCreateDataPartition
if err = dp.PersistMetadata(); err != nil {
dp.checkIsDiskError(err)
log.LogErrorf("action[ApplyMemberChange] dp(%v) PersistMetadata err(%v).", dp.partitionID, err)
if IsDiskErr(err.Error()) {
panic(newRaftApplyError(err))
}
return
}
}
@ -126,7 +127,13 @@ func (dp *DataPartition) ApplySnapshot(peers []raftproto.Peer, iterator raftprot
// HandleFatalEvent notifies the application when panic happens.
func (dp *DataPartition) HandleFatalEvent(err *raft.FatalError) {
log.LogFatalf("action[HandleFatalEvent] err(%v).", err)
if isRaftApplyError(err.Err.Error()) {
dp.stopRaft()
dp.checkIsDiskError(err.Err)
log.LogCriticalf("action[HandleFatalEvent] err(%v).", err)
} else {
log.LogFatalf("action[HandleFatalEvent] err(%v).", err)
}
}
// HandleLeaderChange notifies the application when the raft leader has changed.