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 set delayDelMinute after migration

Browse Source 
Signed-off-by: zhaochenyang <zhaochenyang@oppo.com>
This commit is contained in:
zhaochenyang 2024-02-26 14:53:27 +08:00 committed by AmazingChi
parent de7d0693b5
commit cf5db8b6b6
7 changed files with 26 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
lcnode/const.go
View File

@ -33,6 +33,7 @@ const (
configLcNodeTaskCountLimit = "lcNodeTaskCountLimit"
configEnableDebugService = "enableDebugService"
configDelayDelMinute = "delayDelMinute"
)
// Default of configuration value
@ -52,6 +53,7 @@ const (
defaultUnboundedChanInitCapacity = 10000
defaultLcNodeTaskCountLimit = 1
maxLcNodeTaskCountLimit = 20
defaultDelayDelMinute = 1440 // default retention min(1 day) of old eks after migration
MaxSizePutOnce = int64(1) << 23
)
@ -70,4 +72,5 @@ var (
maxDirChanNum = 1000000
enableDebugService bool
delayDelMinute uint64
)

4
lcnode/lc_scanner.go
View File

@ -396,7 +396,7 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) {
log.LogErrorf("migrate err: %v, dentry: %+v, skip it", err, dentry)
return
}
err = s.mw.UpdateExtentKeyAfterMigration(dentry.Inode, proto.OpTypeToStorageType(dentry.Op), nil, dentry.WriteGen, dentry.Path)
err = s.mw.UpdateExtentKeyAfterMigration(dentry.Inode, proto.OpTypeToStorageType(dentry.Op), nil, dentry.WriteGen, delayDelMinute, dentry.Path)
if err != nil {
atomic.AddInt64(&s.currentStat.ErrorSkippedNum, 1)
log.LogErrorf("update extent key err: %v, dentry: %+v, skip it", err, dentry)
@ -419,7 +419,7 @@ func (s *LcScanner) handleFile(dentry *proto.ScanDentry) {
log.LogErrorf("migrateToEbs err: %v, dentry: %+v, skip it", err, dentry)
return
}
err = s.mw.UpdateExtentKeyAfterMigration(dentry.Inode, proto.OpTypeToStorageType(dentry.Op), oeks, dentry.WriteGen, dentry.Path)
err = s.mw.UpdateExtentKeyAfterMigration(dentry.Inode, proto.OpTypeToStorageType(dentry.Op), oeks, dentry.WriteGen, delayDelMinute, dentry.Path)
if err != nil {
atomic.AddInt64(&s.currentStat.ErrorSkippedNum, 1)
log.LogErrorf("update extent key err: %v, dentry: %+v, skip it", err, dentry)

2
lcnode/meta.go
View File

@ -23,7 +23,7 @@ type MetaWrapper interface {
BatchInodeGet(inodes []uint64) []*proto.InodeInfo
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
UpdateExtentKeyAfterMigration(inode uint64, storageType uint32, extentKeys []proto.ObjExtentKey, writeGen uint64, delayDelMinute uint64, fullPath string) error
DeleteMigrationExtentKey(inode uint64, fullPath string) error
ReadDirLimit_ll(parentID uint64, from string, limit uint64) ([]proto.Dentry, error)
Close() error

2
lcnode/meta_test.go
View File

@ -71,7 +71,7 @@ func (*MockMetaWrapper) Evict(inode uint64, fullPath string) error {
return nil
}
func (*MockMetaWrapper) UpdateExtentKeyAfterMigration(inode uint64, storageType uint32, extentKeys []proto.ObjExtentKey, writeGen uint64, fullPath string) error {
func (*MockMetaWrapper) UpdateExtentKeyAfterMigration(inode uint64, storageType uint32, extentKeys []proto.ObjExtentKey, writeGen uint64, delayDelMinute uint64, fullPath string) error {
return nil
}

15
lcnode/server.go
View File

@ -221,6 +221,21 @@ func (l *LcNode) parseConfig(cfg *config.Config) (err error) {
}
log.LogInfof("loadConfig: setup config: %v(%v)", configLcNodeTaskCountLimit, lcNodeTaskCountLimit)
// parse delayDelMinute
var delay int64
delayStr := cfg.GetString(configDelayDelMinute)
if delayStr != "" {
if delay, err = strconv.ParseInt(delayStr, 10, 64); err != nil {
return fmt.Errorf("%v,err:%v", proto.ErrInvalidCfg, err.Error())
}
}
if delay <= 0 {
delayDelMinute = defaultDelayDelMinute
} else {
delayDelMinute = uint64(delay)
}
log.LogInfof("loadConfig: setup config: %v(%v)", configDelayDelMinute, delayDelMinute)
enableDebugService = cfg.GetBool(configEnableDebugService)
log.LogInfof("loadConfig: setup config: %v(%v)", configEnableDebugService, enableDebugService)

4
sdk/meta/api.go
View File

@ -2767,12 +2767,12 @@ func (mw *MetaWrapper) RenewalForbiddenMigration(inode uint64) error {
}
func (mw *MetaWrapper) UpdateExtentKeyAfterMigration(inode uint64, storageType uint32, objExtentKeys []proto.ObjExtentKey,
writeGen uint64, fullPath string) error {
writeGen uint64, delayDelMinute uint64, fullPath string) error {
mp := mw.getPartitionByInode(inode)
if mp == nil {
return syscall.ENOENT
}
status, err := mw.updateExtentKeyAfterMigration(mp, inode, storageType, objExtentKeys, writeGen, fullPath)
status, err := mw.updateExtentKeyAfterMigration(mp, inode, storageType, objExtentKeys, writeGen, delayDelMinute, fullPath)
if err != nil || status != statusOK {
log.LogErrorf("UpdateExtentKeyAfterMigration: inode(%v) storageType(%v) extentKeys(%v) writeGen(%v) err(%v) status(%v)",
inode, storageType, objExtentKeys, writeGen, err, status)

3
sdk/meta/operation.go
View File

@ -3019,7 +3019,7 @@ func (mw *MetaWrapper) renewalForbiddenMigration(mp *MetaPartition, inode uint64
}
func (mw *MetaWrapper) updateExtentKeyAfterMigration(mp *MetaPartition, inode uint64, storageType uint32,
extentKeys []proto.ObjExtentKey, writeGen uint64, fullPath string) (status int, err error) {
extentKeys []proto.ObjExtentKey, writeGen uint64, delayDelMinute uint64, fullPath string) (status int, err error) {
bgTime := stat.BeginStat()
defer func() {
stat.EndStat("updateExtentKeyAfterMigration", err, bgTime, 1)
@ -3031,6 +3031,7 @@ func (mw *MetaWrapper) updateExtentKeyAfterMigration(mp *MetaPartition, inode ui
NewObjExtentKeys: extentKeys,
WriteGen: writeGen,
}
req.DelayDeleteMinute = delayDelMinute
req.FullPaths = []string{fullPath}
packet := proto.NewPacketReqID()
packet.Opcode = proto.OpMetaUpdateExtentKeyAfterMigration