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(lcnode): hybrid cloud DeleteMigrationExtentKey before migration

Browse Source 
Signed-off-by: zhaochenyang <zhaochenyang@oppo.com>
This commit is contained in:
zhaochenyang 2024-02-23 09:59:07 +08:00 committed by AmazingChi
parent 698311248a
commit 5b867906d2
5 changed files with 20 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
lcnode/lc_scanner.go
View File

@ -384,6 +384,12 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) {
case proto.OpTypeStorageClassHDD:
s.limiter.Wait(context.Background())
if dentry.HasMek {
if err := s.mw.DeleteMigrationExtentKey(dentry.Inode, dentry.Path); err != nil {
log.LogErrorf("DeleteMigrationExtentKey err: %v, dentry: %+v", err, dentry)
}
return
}
err := s.transitionMgr.migrate(dentry)
if err != nil {
atomic.AddInt64(&s.currentStat.ErrorSkippedNum, 1)
@ -401,6 +407,12 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) {
case proto.OpTypeStorageClassEBS:
s.limiter.Wait(context.Background())
if dentry.HasMek {
if err := s.mw.DeleteMigrationExtentKey(dentry.Inode, dentry.Path); err != nil {
log.LogErrorf("DeleteMigrationExtentKey err: %v, dentry: %+v", err, dentry)
}
return
}
oeks, err := s.transitionMgr.migrateToEbs(dentry)
if err != nil {
atomic.AddInt64(&s.currentStat.ErrorSkippedNum, 1)
@ -437,6 +449,7 @@ func (s *LcScanner) batchHandleFile() {
d.Size = info.Size
d.StorageClass = info.StorageClass
d.WriteGen = info.WriteGen
d.HasMek = info.HasMigrationEk
s.fileChan.In <- d
}
}

1
lcnode/meta.go
View File

@ -24,6 +24,7 @@ type MetaWrapper interface {
DeleteWithCond_ll(parentID, cond uint64, name string, isDir bool, fullPath string) (inode *proto.InodeInfo, err error)
Evict(inode uint64, fullPath string) error
UpdateExtentKeyAfterMigration(inode uint64, storageType uint32, extentKeys []proto.ObjExtentKey, writeGen uint64, fullPath string) error
DeleteMigrationExtentKey(inode uint64, fullPath string) error
ReadDirLimit_ll(parentID uint64, from string, limit uint64) ([]proto.Dentry, error)
Close() error
}

4
lcnode/meta_test.go
View File

@ -75,6 +75,10 @@ func (*MockMetaWrapper) UpdateExtentKeyAfterMigration(inode uint64, storageType
return nil
}
func (*MockMetaWrapper) DeleteMigrationExtentKey(inode uint64, fullPath string) error {
return nil
}
func (*MockMetaWrapper) ReadDirLimit_ll(parentID uint64, from string, limit uint64) ([]proto.Dentry, error) {
// for handleDirLimitDepthFirst
if parentID == 4 {

2
metanode/partition_fsmop_inode.go
View File

@ -1079,7 +1079,7 @@ func (mp *metaPartition) fsmUpdateExtentKeyAfterMigration(inoParam *Inode) (resp
func logCurrentExtentKeys(storageClass uint32, sortedEks interface{}, inode uint64) {
if sortedEks == nil {
log.LogInfof("action[fsmUpdateExtentKeyAfterMigration] inode %v current ek empty", inode)
log.LogErrorf("action[fsmUpdateExtentKeyAfterMigration] inode %v current ek empty", inode)
} else {
if proto.IsStorageClassReplica(storageClass) {
log.LogInfof("action[fsmUpdateExtentKeyAfterMigration] inode %v current ek %v",

1
proto/lifecycle.go
View File

@ -330,6 +330,7 @@ type ScanDentry struct {
Size uint64 `json:"size"` // for migrate: size of the current dentry
StorageClass uint32 `json:"sc"` // for migrate: storage class of the current dentry
WriteGen uint64 `json:"gen"` // for migrate: used to determine whether a file is modified
HasMek bool `json:"mek"` // for migrate: if HasMek, call DeleteMigrationExtentKey instead of migrating
}
type BatchDentries struct {