mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
refactor(meta): hybrid cloud add storageClass validation check when inode.storageClass changes
##22542807 Signed-off-by: true1064 <tangjingyu@oppo.com>
This commit is contained in:
true1064
AmazingChi
parent
282bb4bde2
commit
ade3d22670
@ -206,8 +206,8 @@ const (
|
||||
opFSMUpdateExtentKeyAfterMigration = 88
|
||||
opFSMInternalFreeInodeMigrationExtentKey = 89
|
||||
opFSMSetInodeCreateTime = 90 // for debug
|
||||
opFSMDeleteMigrationExtentKey = 91
|
||||
opFSMInternalDeleteMigrationExtentKey = 93
|
||||
opFSMSetDeleteMigrationExtentKey = 91
|
||||
opFSMInternalSetDeleteMigrationExtentKey = 93
|
||||
)
|
||||
|
||||
var (
|
||||
|
||||
@ -2498,6 +2498,13 @@ func (i *Inode) updateStorageClass(storageClass uint32, isCache, isMigration boo
|
||||
if isCache {
|
||||
return nil
|
||||
}
|
||||
|
||||
if !proto.IsValidStorageClass(storageClass) {
|
||||
err := fmt.Errorf("[updateStorageClass] isCache(%v) isMigration(%v): invalid storageClass(%v)",
|
||||
isCache, isMigration, storageClass)
|
||||
panic(err.Error())
|
||||
}
|
||||
|
||||
if isMigration {
|
||||
if i.HybridCouldExtentsMigration.storageClass == proto.StorageClass_Unspecified {
|
||||
i.HybridCouldExtentsMigration.storageClass = storageClass
|
||||
|
||||
@ -496,23 +496,23 @@ func (mp *metaPartition) Apply(command []byte, index uint64) (resp interface{},
|
||||
}
|
||||
err = mp.fsmSetCreateTime(req)
|
||||
case opFSMInternalFreeInodeMigrationExtentKey:
|
||||
err = mp.internalDeleteMigrationExtentKey(msg.V)
|
||||
case opFSMDeleteMigrationExtentKey:
|
||||
err = mp.fsmInternalFreeMigrationExtentKey(msg.V)
|
||||
case opFSMSetDeleteMigrationExtentKey:
|
||||
ino := NewInode(0, 0)
|
||||
if err = ino.Unmarshal(msg.V); err != nil {
|
||||
log.LogWarnf("[Apply] mp(%v) opFSMDeleteMigrationExtentKey Unmarshal inode failed: %v",
|
||||
log.LogWarnf("[Apply] mp(%v) opFSMSetDeleteMigrationExtentKey Unmarshal inode failed: %v",
|
||||
mp.config.PartitionId, err.Error())
|
||||
return
|
||||
}
|
||||
resp = mp.fsmDeleteMigrationExtentKey(ino)
|
||||
case opFSMInternalDeleteMigrationExtentKey:
|
||||
resp = mp.fsmSetDeleteMigrationExtentKey(ino)
|
||||
case opFSMInternalSetDeleteMigrationExtentKey:
|
||||
ino := NewInode(0, 0)
|
||||
if err = ino.Unmarshal(msg.V); err != nil {
|
||||
log.LogWarnf("[Apply] mp(%v) opFSMInternalDeleteMigrationExtentKey Unmarshal inode failed: %v",
|
||||
log.LogWarnf("[Apply] mp(%v) opFSMInternalSetDeleteMigrationExtentKey Unmarshal inode failed: %v",
|
||||
mp.config.PartitionId, err.Error())
|
||||
return
|
||||
}
|
||||
resp = mp.fsmInternalDeleteMigrationExtentKey(ino)
|
||||
resp = mp.fsmInternalSetDeleteMigrationExtentKey(ino)
|
||||
default:
|
||||
// do nothing
|
||||
case opFSMSyncInodeAccessTime:
|
||||
|
||||
@ -1144,6 +1144,11 @@ func (mp *metaPartition) fsmUpdateExtentKeyAfterMigration(inoParam *Inode) (resp
|
||||
log.LogInfof("action[fsmUpdateExtentKeyAfterMigration] mp(%v) inode(%v) migration ek will be deleted at %v",
|
||||
mp.config.PartitionId, i.Inode, time.Unix(i.HybridCouldExtentsMigration.expiredTime, 0).Format("2006-01-02 15:04:05"))
|
||||
mp.freeList.Push(i.Inode)
|
||||
if !proto.IsValidStorageClass(i.StorageClass) {
|
||||
panicMsg := fmt.Sprintf("[fsmUpdateExtentKeyAfterMigration] mp(%v) inode(%v): invalid storageClass(%v)",
|
||||
mp.config.PartitionId, i.Inode, i.StorageClass)
|
||||
panic(panicMsg)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@ -1178,7 +1183,7 @@ func (mp *metaPartition) fsmSetCreateTime(req *SetCreateTimeRequest) (err error)
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) internalDeleteMigrationExtentKey(val []byte) (err error) {
|
||||
func (mp *metaPartition) fsmInternalFreeMigrationExtentKey(val []byte) (err error) {
|
||||
if len(val) == 0 {
|
||||
return
|
||||
}
|
||||
@ -1193,7 +1198,7 @@ func (mp *metaPartition) internalDeleteMigrationExtentKey(val []byte) (err error
|
||||
}
|
||||
return
|
||||
}
|
||||
log.LogDebugf("internalDeleteMigrationExtentKey: partitionID(%v) inode(%v)",
|
||||
log.LogDebugf("fsmInternalFreeMigrationExtentKey: partitionID(%v) inode(%v)",
|
||||
mp.config.PartitionId, inoParam.Inode)
|
||||
err = mp.internalDeleteInodeMigrationExtentKey(inoParam)
|
||||
if err != nil {
|
||||
@ -1221,7 +1226,7 @@ func (mp *metaPartition) internalDeleteInodeMigrationExtentKey(inoParam *Inode)
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) fsmDeleteMigrationExtentKey(inoParam *Inode) (resp *InodeResponse) {
|
||||
func (mp *metaPartition) fsmSetDeleteMigrationExtentKey(inoParam *Inode) (resp *InodeResponse) {
|
||||
resp = NewInodeResponse()
|
||||
resp.Status = proto.OpOk
|
||||
item := mp.inodeTree.CopyGet(inoParam)
|
||||
@ -1231,12 +1236,12 @@ func (mp *metaPartition) fsmDeleteMigrationExtentKey(inoParam *Inode) (resp *Ino
|
||||
}
|
||||
i := item.(*Inode)
|
||||
i.SetDeleteMigrationExtentKeyImmediately()
|
||||
log.LogInfof("action[fsmDeleteMigrationExtentKey] inode %v migration ek will be deleted immediately", i.Inode)
|
||||
log.LogInfof("action[fsmSetDeleteMigrationExtentKey] inode %v migration ek will be deleted immediately", i.Inode)
|
||||
mp.freeList.Push(i.Inode)
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) fsmInternalDeleteMigrationExtentKey(inoParam *Inode) (resp *InodeResponse) {
|
||||
func (mp *metaPartition) fsmInternalSetDeleteMigrationExtentKey(inoParam *Inode) (resp *InodeResponse) {
|
||||
resp = NewInodeResponse()
|
||||
resp.Status = proto.OpOk
|
||||
item := mp.inodeTree.CopyGet(inoParam)
|
||||
@ -1247,14 +1252,15 @@ func (mp *metaPartition) fsmInternalDeleteMigrationExtentKey(inoParam *Inode) (r
|
||||
i := item.(*Inode)
|
||||
// may be triggered by updateMigrationExtentKey
|
||||
if proto.IsStorageClassBlobStore(inoParam.HybridCouldExtentsMigration.storageClass) {
|
||||
log.LogDebugf("action[fsmInternalDeleteMigrationExtentKey] inode %v migration ek replaced,"+
|
||||
log.LogDebugf("action[fsmInternalSetDeleteMigrationExtentKey] inode %v migration ek replaced,"+
|
||||
" old migrationStorageClass(%v)", i.Inode, i.HybridCouldExtentsMigration.storageClass)
|
||||
i.HybridCouldExtentsMigration.sortedEks = inoParam.HybridCouldExtentsMigration.sortedEks
|
||||
i.HybridCouldExtentsMigration.storageClass = inoParam.HybridCouldExtentsMigration.storageClass
|
||||
}
|
||||
|
||||
i.SetDeleteMigrationExtentKeyImmediately()
|
||||
log.LogInfof("action[fsmInternalDeleteMigrationExtentKey] inode %v migration ek will be deleted immediately", i.Inode)
|
||||
log.LogWarnf("action[fsmInternalSetDeleteMigrationExtentKey] mp(%v) inode(%v migration ek will be deleted immediately",
|
||||
mp.config.PartitionId, i.Inode)
|
||||
mp.freeList.Push(i.Inode)
|
||||
return
|
||||
}
|
||||
|
||||
@ -54,7 +54,7 @@ func (mp *metaPartition) CheckQuota(inodeId uint64, p *Packet) (iParm *Inode, in
|
||||
|
||||
mp.uidManager.acLock.Lock()
|
||||
if mp.uidManager.getUidAcl(inode.Uid) {
|
||||
log.LogWarnf("CheckQuota UidSpace.volname [%v] mp[%v] uid %v be set full", mp.uidManager.mpID, mp.uidManager.volName, inode.Uid)
|
||||
log.LogWarnf("CheckQuota UidSpace.volName[%v] mp[%v] uid %v be set full", mp.uidManager.mpID, mp.uidManager.volName, inode.Uid)
|
||||
mp.uidManager.acLock.Unlock()
|
||||
status = proto.OpNoSpaceErr
|
||||
err = errors.New("CheckQuota UidSpace is over quota")
|
||||
|
||||
@ -157,6 +157,12 @@ func (mp *metaPartition) checkCreateInoStorageClassForCompatibility(reqStorageCl
|
||||
mp.config.VolName, mp.config.PartitionId, inodeId)
|
||||
}
|
||||
|
||||
if !proto.IsValidStorageClass(resultStorageClass) {
|
||||
panicMsg := fmt.Sprintf("[checkCreateInoStorageClassForCompatibility] mp(%v) inode(%v): invalid resultStorageClass(%v)",
|
||||
mp.config.PartitionId, inodeId, resultStorageClass)
|
||||
panic(panicMsg)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@ -1127,26 +1133,26 @@ func (mp *metaPartition) RenewalForbiddenMigration(req *proto.RenewalForbiddenMi
|
||||
|
||||
func (mp *metaPartition) UpdateExtentKeyAfterMigration(req *proto.UpdateExtentKeyAfterMigrationRequest, p *Packet,
|
||||
remoteAddr string) (err error) {
|
||||
ino := NewInode(req.Inode, 0)
|
||||
inoParm := NewInode(req.Inode, 0)
|
||||
var item BtreeItem
|
||||
if item = mp.inodeTree.Get(ino); item == nil {
|
||||
err = fmt.Errorf("mp(%v) can not find inode(%v)", mp.config.PartitionId, ino.Inode)
|
||||
if item = mp.inodeTree.Get(inoParm); item == nil {
|
||||
err = fmt.Errorf("mp(%v) can not find inode(%v)", mp.config.PartitionId, inoParm.Inode)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err)
|
||||
p.PacketErrorWithBody(proto.OpNotExistErr, []byte(err.Error()))
|
||||
return
|
||||
} else {
|
||||
ino.UpdateHybridCloudParams(item.(*Inode))
|
||||
inoParm.UpdateHybridCloudParams(item.(*Inode))
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
if mp.IsEnableAuditLog() {
|
||||
defer func() {
|
||||
auditlog.LogMigrationOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.GetFullPath(), err, time.Since(start).Milliseconds(), req.Inode, ino.StorageClass, req.StorageClass)
|
||||
auditlog.LogMigrationOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.GetFullPath(), err, time.Since(start).Milliseconds(), req.Inode, inoParm.StorageClass, req.StorageClass)
|
||||
}()
|
||||
}
|
||||
|
||||
if proto.IsDir(ino.Type) && req.NewObjExtentKeys != nil {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) is dir, but request NewObjExtentKeys is not nil", mp.config.PartitionId, ino.Inode)
|
||||
if proto.IsDir(inoParm.Type) && req.NewObjExtentKeys != nil {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) is dir, but request NewObjExtentKeys is not nil", mp.config.PartitionId, inoParm.Inode)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err)
|
||||
p.PacketErrorWithBody(proto.OpArgMismatchErr, []byte(err.Error()))
|
||||
return
|
||||
@ -1170,44 +1176,44 @@ func (mp *metaPartition) UpdateExtentKeyAfterMigration(req *proto.UpdateExtentKe
|
||||
mp.internalNotifyFollowerToDeleteExtentKey(delMigrationIno)
|
||||
}()
|
||||
|
||||
if atomic.LoadUint32(&ino.ForbiddenMigration) == ForbiddenToMigration {
|
||||
if atomic.LoadUint32(&inoParm.ForbiddenMigration) == ForbiddenToMigration {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) is forbidden to migration for lease is occupied by others",
|
||||
mp.config.PartitionId, ino.Inode)
|
||||
mp.config.PartitionId, inoParm.Inode)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err)
|
||||
p.PacketErrorWithBody(proto.OpLeaseOccupiedByOthers, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
writeGen := atomic.LoadUint64(&ino.WriteGeneration)
|
||||
writeGen := atomic.LoadUint64(&inoParm.WriteGeneration)
|
||||
if writeGen > req.WriteGen {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) write generation not match, curent(%v) request(%v)",
|
||||
mp.config.PartitionId, ino.Inode, writeGen, req.WriteGen)
|
||||
mp.config.PartitionId, inoParm.Inode, writeGen, req.WriteGen)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err)
|
||||
p.PacketErrorWithBody(proto.OpLeaseGenerationNotMatch, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
// wal logs for UpdateExtentKeyAfterMigration is persisted, but return no leader later,
|
||||
// and request is send to meta node by retry
|
||||
if ino.StorageClass == req.StorageClass {
|
||||
if inoParm.StorageClass == req.StorageClass {
|
||||
msg := fmt.Sprintf("mp(%v) inode(%v) storageClass(%v) is same with req, may be migrated before",
|
||||
mp.config.PartitionId, ino.Inode, ino.StorageClass)
|
||||
mp.config.PartitionId, inoParm.Inode, inoParm.StorageClass)
|
||||
log.LogWarnf("action[UpdateExtentKeyAfterMigration] %v", msg)
|
||||
p.PacketErrorWithBody(proto.OpNotPerm, []byte(msg))
|
||||
return
|
||||
}
|
||||
// store ek after migration in HybridCouldExtentsMigration
|
||||
ino.HybridCouldExtentsMigration.storageClass = req.StorageClass
|
||||
ino.HybridCouldExtentsMigration.expiredTime = time.Now().Add(time.Duration(req.DelayDeleteMinute) * time.Minute).Unix()
|
||||
inoParm.HybridCouldExtentsMigration.storageClass = req.StorageClass
|
||||
inoParm.HybridCouldExtentsMigration.expiredTime = time.Now().Add(time.Duration(req.DelayDeleteMinute) * time.Minute).Unix()
|
||||
if req.StorageClass == proto.StorageClass_BlobStore {
|
||||
ino.HybridCouldExtentsMigration.sortedEks = NewSortedObjExtentsFromObjEks(req.NewObjExtentKeys)
|
||||
inoParm.HybridCouldExtentsMigration.sortedEks = NewSortedObjExtentsFromObjEks(req.NewObjExtentKeys)
|
||||
} else if req.StorageClass == proto.StorageClass_Replica_HDD {
|
||||
if item.(*Inode).HybridCouldExtentsMigration.sortedEks == nil &&
|
||||
item.(*Inode).HybridCouldExtentsMigration.storageClass == proto.StorageClass_Unspecified {
|
||||
log.LogDebugf("action[UpdateExtentKeyAfterMigration] ino %v has no migration data", ino.Inode)
|
||||
ino.HybridCouldExtentsMigration.sortedEks = NewSortedExtents()
|
||||
log.LogDebugf("action[UpdateExtentKeyAfterMigration] inoParm %v has no migration data", inoParm.Inode)
|
||||
inoParm.HybridCouldExtentsMigration.sortedEks = NewSortedExtents()
|
||||
} else {
|
||||
if item.(*Inode).HybridCouldExtentsMigration.storageClass != proto.StorageClass_Replica_HDD {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) storageClass(%v) migrateStorageClass(%v): inode is migrating or migrated from (%v), can not migrate to %v",
|
||||
mp.config.PartitionId, ino.Inode, proto.StorageClassString(ino.StorageClass),
|
||||
mp.config.PartitionId, inoParm.Inode, proto.StorageClassString(inoParm.StorageClass),
|
||||
proto.StorageClassString(item.(*Inode).HybridCouldExtentsMigration.storageClass),
|
||||
proto.StorageClassString(item.(*Inode).HybridCouldExtentsMigration.storageClass),
|
||||
proto.StorageClassString(proto.StorageClass_Replica_HDD))
|
||||
@ -1215,19 +1221,19 @@ func (mp *metaPartition) UpdateExtentKeyAfterMigration(req *proto.UpdateExtentKe
|
||||
p.PacketErrorWithBody(proto.OpArgMismatchErr, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
ino.HybridCouldExtentsMigration.sortedEks = item.(*Inode).HybridCouldExtentsMigration.sortedEks
|
||||
inoParm.HybridCouldExtentsMigration.sortedEks = item.(*Inode).HybridCouldExtentsMigration.sortedEks
|
||||
}
|
||||
} else {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) unknown migration storageClass(%v)",
|
||||
mp.config.PartitionId, ino.Inode, req.StorageClass)
|
||||
mp.config.PartitionId, inoParm.Inode, req.StorageClass)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err)
|
||||
p.PacketErrorWithBody(proto.OpArgMismatchErr, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
val, err := ino.Marshal()
|
||||
val, err := inoParm.Marshal()
|
||||
if err != nil {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) Marshal inner err: %v",
|
||||
mp.config.PartitionId, ino.Inode, err.Error())
|
||||
mp.config.PartitionId, inoParm.Inode, err.Error())
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err.Error())
|
||||
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
|
||||
return
|
||||
@ -1235,14 +1241,14 @@ func (mp *metaPartition) UpdateExtentKeyAfterMigration(req *proto.UpdateExtentKe
|
||||
fsmResp, submitErr := mp.submit(opFSMUpdateExtentKeyAfterMigration, val)
|
||||
if submitErr != nil {
|
||||
if submitErr == raft.ErrNotLeader {
|
||||
err = fmt.Errorf("mp(%v) inode(%v), not leader when submit raft", mp.config.PartitionId, ino.Inode)
|
||||
err = fmt.Errorf("mp(%v) inode(%v), not leader when submit raft", mp.config.PartitionId, inoParm.Inode)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err.Error())
|
||||
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
err = fmt.Errorf("mp(%v) inode(%v) submit raft inner err: %v",
|
||||
mp.config.PartitionId, ino.Inode, err.Error())
|
||||
mp.config.PartitionId, inoParm.Inode, err.Error())
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] %v", err.Error())
|
||||
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
|
||||
return
|
||||
@ -1250,7 +1256,7 @@ func (mp *metaPartition) UpdateExtentKeyAfterMigration(req *proto.UpdateExtentKe
|
||||
fsmRespStatus := fsmResp.(*InodeResponse).Status
|
||||
if fsmRespStatus != proto.OpOk {
|
||||
err = fmt.Errorf("mp(%v) inode(%v) storageClass(%v), raft resp inner err status(%v)",
|
||||
mp.config.PartitionId, ino.Inode, ino.StorageClass, fsmRespStatus)
|
||||
mp.config.PartitionId, inoParm.Inode, inoParm.StorageClass, fsmRespStatus)
|
||||
log.LogErrorf("action[UpdateExtentKeyAfterMigration] req(%v), err: %v", req, err.Error())
|
||||
p.PacketErrorWithBody(fsmRespStatus, []byte(err.Error()))
|
||||
return
|
||||
@ -1413,7 +1419,7 @@ func (mp *metaPartition) DeleteMigrationExtentKey(req *proto.DeleteMigrationExte
|
||||
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
resp, err := mp.submit(opFSMDeleteMigrationExtentKey, val)
|
||||
resp, err := mp.submit(opFSMSetDeleteMigrationExtentKey, val)
|
||||
if err != nil {
|
||||
p.PacketErrorWithBody(proto.OpAgain, []byte(err.Error()))
|
||||
return
|
||||
@ -1426,14 +1432,16 @@ func (mp *metaPartition) DeleteMigrationExtentKey(req *proto.DeleteMigrationExte
|
||||
func (mp *metaPartition) internalNotifyFollowerToDeleteExtentKey(ino *Inode) {
|
||||
val, err := ino.Marshal()
|
||||
if err != nil {
|
||||
log.LogErrorf("action[internalNotifyFollowerToDeleteExtentKey] ino %v marshal failed:%v", ino.Inode, err)
|
||||
log.LogErrorf("action[internalNotifyFollowerToDeleteExtentKey] mpId(%v) ino(%v) marshal failed: %v",
|
||||
mp.config.PartitionId, ino.Inode, err)
|
||||
return
|
||||
}
|
||||
_, err = mp.submit(opFSMInternalDeleteMigrationExtentKey, val)
|
||||
_, err = mp.submit(opFSMInternalSetDeleteMigrationExtentKey, val)
|
||||
if err != nil {
|
||||
log.LogErrorf("action[internalNotifyFollowerToDeleteExtentKey] ino %v submit opFSMDeleteMigrationExtentKey "+
|
||||
"failed:%v", ino.Inode, err)
|
||||
log.LogErrorf("action[internalNotifyFollowerToDeleteExtentKey] mpId(%v) ino(%v) submit failed: %v",
|
||||
mp.config.PartitionId, ino.Inode, err)
|
||||
return
|
||||
}
|
||||
log.LogWarnf("action[internalNotifyFollowerToDeleteExtentKey] submit ino %v", ino.Inode)
|
||||
log.LogWarnf("action[internalNotifyFollowerToDeleteExtentKey] mpId(%v) ino(%v) storageClass(%v) migrateStorageClass(%v) submit",
|
||||
mp.config.PartitionId, ino.Inode, ino.StorageClass, ino.HybridCouldExtentsMigration.storageClass)
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
