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

feat(data): refactor tiny extent delete logic. #1000089221

Browse Source 
Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2025-04-29 15:23:33 +08:00 committed by zhumingze1108
parent a5467a3d58
commit 36936cafd8
11 changed files with 56 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
datanode/data_partition_repair.go
View File

@ -41,6 +41,13 @@ type RepairExtentInfo struct {
Source string `json:"src"`
}
func (ei *RepairExtentInfo) String() (m string) {
if ei == nil {
return "repair extent info nil"
}
return fmt.Sprintf("%s_%s", ei.ExtentInfo.String(), ei.Source)
}
// DataPartitionRepairTask defines the repair task for the data partition.
type DataPartitionRepairTask struct {
TaskType uint8

3
datanode/partition.go
View File

@ -1291,7 +1291,8 @@ func (dp *DataPartition) doStreamFixTinyDeleteRecord(repairTask *DataPartitionRe
DeleteLimiterWait()
dp.disk.allocCheckLimit(proto.IopsWriteType, 1)
// log.LogInfof("doStreamFixTinyDeleteRecord Delete PartitionID(%v)_Extent(%v)_Offset(%v)_Size(%v)", dp.partitionID, extentID, offset, size)
store.MarkDelete(extentID, int64(offset), int64(size))
// store.MarkDelete(extentID, int64(offset), int64(size))
store.RecordTinyDelete(extentID, int64(offset), int64(size))
if store.IsClosed() {
log.LogWarnf("dp %v in doStreamFixTinyDeleteRecord exit due to store is closed", dp.partitionID)
return

4
datanode/repl/packet.go
View File

@ -122,6 +122,8 @@ func (p *FollowerPacket) identificationErrorResultCode(errLog string, errMsg str
p.ResultCode = proto.OpDiskNoSpaceErr
} else if strings.Contains(errMsg, storage.LimitedIoError.Error()) {
p.ResultCode = proto.OpLimitedIoErr
} else if strings.Contains(errMsg, storage.TinyRecoverError.Error()) {
p.ResultCode = proto.OpTinyRecoverErr
} else if strings.Contains(errMsg, storage.TryAgainError.Error()) {
p.ResultCode = proto.OpAgain
} else if strings.Contains(errMsg, raft.ErrNotLeader.Error()) {
@ -452,6 +454,8 @@ func (p *Packet) identificationErrorResultCode(errLog string, errMsg string) {
p.ResultCode = proto.OpDiskNoSpaceErr
} else if strings.Contains(errMsg, storage.LimitedIoError.Error()) {
p.ResultCode = proto.OpLimitedIoErr
} else if strings.Contains(errMsg, storage.TinyRecoverError.Error()) {
p.ResultCode = proto.OpTinyRecoverErr
} else if strings.Contains(errMsg, storage.TryAgainError.Error()) {
p.ResultCode = proto.OpAgain
} else if strings.Contains(errMsg, raft.ErrNotLeader.Error()) {

2
datanode/repl/repl_protocol.go
View File

@ -432,7 +432,7 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
reply.StartT, fmt.Errorf(string(reply.Data[:reply.Size]))))
if reply.IsWriteOpOfPacketProtoVerForbidden() {
log.LogDebugf(err.Error())
} else if reply.ResultCode == proto.OpNotExistErr || reply.ResultCode == proto.ErrCodeVersionOpError {
} else if reply.ResultCode == proto.OpNotExistErr || reply.ResultCode == proto.ErrCodeVersionOpError || reply.ResultCode == proto.OpTinyRecoverErr {
log.LogInfof(err.Error())
} else if (reply.ResultCode == proto.OpTryOtherAddr && reply.Opcode == proto.OpWrite) ||
reply.Opcode == proto.OpReadTinyDeleteRecord ||

1
datanode/storage/error.go
View File

@ -29,6 +29,7 @@ var (
ForbiddenMetaPartitionError = errors.New("meta partition is forbidden")
TryAgainError = errors.New("try again")
LimitedIoError = errors.New("limited io error")
TinyRecoverError = errors.New("tiny extent recovering error")
CrcMismatchError = errors.New("packet Crc is incorrect")
NoLeaderError = errors.New("no raft leader")
ExtentNotFoundError = errors.New("extent does not exist")

5
datanode/storage/extent_store.go
View File

@ -814,9 +814,12 @@ func (s *ExtentStore) punchDelete(extentID uint64, offset, size int64) (err erro
if err != nil {
return nil
}
if offset+size > e.dataSize {
return
log.LogWarnf("punchDelete: tiny extent maybe recovering, retry later. extId %d, offset %d, size %d, eSize %d", extentID, offset, size, e.dataSize)
return TinyRecoverError
}
var hasDelete bool
if hasDelete, err = e.punchDelete(offset, size); err != nil {
return

16
datanode/wrap_operator.go
View File

@ -109,7 +109,7 @@ func (s *DataNode) OperatePacket(p *repl.Packet, c net.Conn) (err error) {
p.LogMessage(p.GetOpMsg(), c.RemoteAddr().String(), start, err))
if p.IsWriteOpOfPacketProtoVerForbidden() || strings.Contains(logContent, raft.ErrNotLeader.Error()) || p.Opcode == proto.OpReadTinyDeleteRecord {
log.LogWarnf(logContent)
} else if isColdVolExtentDelErr(p) {
} else if isColdVolExtentDelErr(p) || p.ResultCode == proto.OpTinyRecoverErr || p.ResultCode == proto.OpLimitedIoErr {
log.LogInfof(logContent)
} else {
log.LogErrorf(logContent)
@ -801,7 +801,12 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) {
var err error
defer func() {
if err != nil {
log.LogErrorf(fmt.Sprintf("(%v) error(%v).", p.GetUniqueLogId(), err))
msg := fmt.Sprintf("(%v) error(%v).", p.GetUniqueLogId(), err)
if err == storage.LimitedIoError || err == storage.TinyRecoverError {
log.LogInfo(msg)
} else {
log.LogError(msg)
}
p.PackErrorBody(ActionBatchMarkDelete, err.Error())
} else {
p.PacketOkReply()
@ -854,7 +859,12 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) {
err = partition.ExtentStore().MarkDelete(ext.ExtentId, 0, 0)
}
if err != nil {
log.LogErrorf("action[handleBatchMarkDeletePacket]: failed to mark delete normalExtent extent(%v), offset(%v) err %v", ext.ExtentId, ext.FileOffset, err)
msg := fmt.Sprintf("action[handleBatchMarkDeletePacket]: failed to mark delete normalExtent extent(%v), offset(%v) err %v", ext.ExtentId, ext.FileOffset, err)
if err == storage.TinyRecoverError {
log.LogInfo(msg)
} else {
log.LogError(msg)
}
}
})

9
metanode/manager_op.go
View File

@ -136,9 +136,12 @@ func (m *metadataManager) opMasterHeartbeat(conn net.Conn, p *Packet,
log.LogWarnf("[opMasterHeartbeat] change GOGC, old(%v) new(%v)", oldGOGC, req.MetaNodeGOGC)
}
}
m.fileStatsConfig.fileStatsEnable = req.FileStatsEnable
if len(req.FileStatsThresholds) != 0 {
m.updateFileStatsConfig(req.FileStatsThresholds)
if m.fileStatsConfig != nil {
m.fileStatsConfig.fileStatsEnable = req.FileStatsEnable
if len(req.FileStatsThresholds) != 0 {
m.updateFileStatsConfig(req.FileStatsThresholds)
}
}
log.LogDebugf("metaNode.raftPartitionCanUsingDifferentPort from %v to %v", m.metaNode.raftPartitionCanUsingDifferentPort, req.RaftPartitionCanUsingDifferentPortEnabled)

4
metanode/partition_delete_extents.go
View File

@ -213,7 +213,7 @@ func (mp *metaPartition) batchDeleteExtentsByDp(dpId uint64, extents []*proto.De
err = mp.doBatchDeleteExtentsByPartition(dpId, batchEk)
if err != nil {
msg := fmt.Sprintf("[batchDeleteExtentsByDp] vol(%v) mp(%v) failed to delete dp(%v) extents cnt(%v), err %s", mp.GetVolName(), mp.config.PartitionId, dp.PartitionID, len(batchEk), err.Error())
if strings.Contains(msg, "OpLimitedIoErr") {
if strings.Contains(msg, "OpLimitedIoErr") || strings.Contains(msg, "OpTinyRecoverErr") {
log.LogInfo(msg)
} else {
log.LogError(msg)
@ -426,7 +426,7 @@ func (mp *metaPartition) deleteExtentsFromList(fileList *synclist.SyncList) {
retry, err = mp.batchDeleteExtentsByDp(dpId, eks)
if err != nil {
msg := fmt.Sprintf("[deleteExtentsFromList] mp(%v) failed to delete dp(%v) err(%v)", mp.config.PartitionId, dpId, err)
if strings.Contains(msg, "OpLimitedIoErr") {
if strings.Contains(msg, "OpLimitedIoErr") || strings.Contains(msg, "OpTinyRecoverErr") {
log.LogInfo(msg)
} else {
log.LogError(msg)

18
metanode/partition_free_list.go
View File

@ -297,14 +297,22 @@ func (mp *metaPartition) batchDeleteExtentsByPartition(partitionDeleteExtents ma
}
}
extents.Range(func(_ int, ek proto.ExtentKey) bool {
if occurErrors[ek.PartitionId] == nil {
successDeleteExtentCnt++
return true
} else {
log.LogWarnf("batchDeleteExtentsByPartition: deleteInode Inode(%v) error(%v)", inode.Inode, occurErrors[ek.PartitionId])
dpErr := occurErrors[ek.PartitionId]
if dpErr != nil {
msg := fmt.Sprintf("batchDeleteExtentsByPartition: deleteInode Inode(%v) error(%v)", inode.Inode, dpErr.Error())
if strings.Contains(msg, "OpLimitedIoErr") || strings.Contains(msg, "OpTinyRecoverErr") {
log.LogInfo(msg)
} else {
log.LogWarn(msg)
}
return false
}
successDeleteExtentCnt++
return true
})
if successDeleteExtentCnt == extents.Len() {
shouldCommit = append(shouldCommit, inode)
log.LogDebugf("action[batchDeleteExtentsByPartition]: delete vol(%v) mp(%v) inode(%v) success", mp.config.VolName, mp.config.PartitionId, inode.Inode)

3
proto/packet.go
View File

@ -280,6 +280,7 @@ const (
OpMetaInodeAccessTimeGet uint8 = 0xB2
OpReachMaxExtentsErr uint8 = 0xB3
OpTinyRecoverErr uint8 = 0xB4
// hybirdCloud
OpMismatchStorageClass uint8 = 0x82
@ -821,6 +822,8 @@ func (p *Packet) GetResultMsg() (m string) {
m = "OpForbidErr"
case OpLimitedIoErr:
m = "OpLimitedIoErr"
case OpTinyRecoverErr:
m = "OpTinyRecoverErr"
case OpStoreClosed:
return "OpStoreClosed"
case OpReachMaxExtentsErr: