From 44d1d951613df882b4530b07f1d93a99d179fb8e Mon Sep 17 00:00:00 2001 From: Victor1319 Date: Tue, 12 Nov 2024 19:09:32 +0800 Subject: [PATCH] fix(meta): fix some issue about compatiable. #22780321 1. remove useless code. 2. fix compatiable bug abount dataMediatType. Signed-off-by: Victor1319 --- metanode/datapartition.go | 15 +++++ metanode/inode.go | 11 +++- metanode/inode_test.go | 4 ++ metanode/manager.go | 29 ++++----- metanode/manager_op.go | 2 +- metanode/partition.go | 49 +++++++--------- metanode/partition_free_list.go | 58 +----------------- metanode/partition_fsm.go | 2 +- metanode/partition_op_extent.go | 100 ++++++++++++-------------------- metanode/partition_op_inode.go | 20 +++---- 10 files changed, 106 insertions(+), 184 deletions(-) diff --git a/metanode/datapartition.go b/metanode/datapartition.go index bc2150c9e..daa12d5fb 100644 --- a/metanode/datapartition.go +++ b/metanode/datapartition.go @@ -51,6 +51,7 @@ type Vol struct { sync.RWMutex dataPartitionView map[uint64]*DataPartition volDeleteLockTime int64 + info *proto.SimpleVolView } // NewVol returns a new volume instance. @@ -75,6 +76,20 @@ func (v *Vol) UpdatePartitions(partitions *DataPartitionsView) { } } +func (v *Vol) SetVolView(info *proto.SimpleVolView) { + v.Lock() + defer v.Unlock() + + v.info = info +} + +func (v *Vol) GetVolView() *proto.SimpleVolView { + v.Lock() + defer v.Unlock() + + return v.info +} + func (v *Vol) replaceOrInsert(partition *DataPartition) { v.Lock() defer v.Unlock() diff --git a/metanode/inode.go b/metanode/inode.go index 4d9967261..20f10846e 100644 --- a/metanode/inode.go +++ b/metanode/inode.go @@ -1061,11 +1061,13 @@ func (i *Inode) UnmarshalInodeValue(buff *bytes.Buffer) (err error) { if err, _ = extents.UnmarshalBinary(buff.Bytes(), false); err != nil { return } - i.HybridCouldExtents.sortedEks = extents + if extents.Len() > 0 { + i.HybridCouldExtents.sortedEks = extents + } i.StorageClass = legacyReplicaStorageClass if i.StorageClass == proto.StorageClass_Unspecified && isFile && extents.Len() > 0 { - return fmt.Errorf("UnmarshalInodeValue: legacyReplicaStorageClass not set in config") + return fmt.Errorf("UnmarshalInodeValue: legacyReplicaStorageClass not set in config, ino %d", i.Inode) } return } @@ -1139,6 +1141,9 @@ func (i *Inode) UnmarshalInodeValue(buff *bytes.Buffer) (err error) { } } i.StorageClass = legacyReplicaStorageClass + if !proto.IsValidStorageClass(i.StorageClass) { + i.StorageClass = proto.StorageClass_BlobStore + } } if v3 { @@ -1172,7 +1177,7 @@ func (i *Inode) UnmarshalInodeValue(buff *bytes.Buffer) (err error) { } if i.StorageClass == proto.StorageClass_Unspecified && isFile { - return fmt.Errorf("UnmarshalInodeValue: legacyReplicaStorageClass not set in config") + i.StorageClass = proto.StorageClass_BlobStore } if i.Reserved&V4MigrationExtentsFlag > 0 { diff --git a/metanode/inode_test.go b/metanode/inode_test.go index c86e7ecd3..754fbfb05 100644 --- a/metanode/inode_test.go +++ b/metanode/inode_test.go @@ -244,6 +244,10 @@ func TestCompitableWithOldVersion(t *testing.T) { err = oldIno2.Unmarshal(data) require.NoError(t, err) require.True(t, oldIno2.EqualNew(newIno)) + + err = newIno.Unmarshal(data) + require.NoError(t, err) + require.True(t, oldIno2.EqualNew(newIno)) } oldIno := NewOldVersionInode(1024, uint32(os.ModeDir)) diff --git a/metanode/manager.go b/metanode/manager.go index 9960a3e76..6af4ab6bf 100644 --- a/metanode/manager.go +++ b/metanode/manager.go @@ -116,7 +116,7 @@ func (m *metadataManager) GetAllVolumes() (volumes *util.Set) { } func (m *metadataManager) getDataPartitions(volName string) (view *proto.DataPartitionsView, err error) { - view, err = masterClient.ClientAPI().GetDataPartitions(volName) + view, err = masterClient.ClientAPI().EncodingGzip().GetDataPartitions(volName) if err != nil { log.LogErrorf("action[getDataPartitions]: failed to get data partitions for volume %v", volName) } @@ -669,28 +669,20 @@ func (m *metadataManager) loadPartitions() (err error) { return } -func (m *metadataManager) forceUpdatePartitionVolume(partition MetaPartition) { +func (m *metadataManager) forceUpdatePartitionVolume(partition MetaPartition) error { volName := partition.GetBaseConfig().VolName // NOTE: maybe add a cache will be better? dataView, volView, err := m.getVolumeUpdateInfo(volName) if err != nil { log.LogErrorf("action[registerPartition]: failed to get info of volume %v", volName) - return + return err } partition.UpdateVolumeView(dataView, volView) -} - -func (m *metadataManager) registerPartition(id uint64, partition MetaPartition) { - func() { - m.mu.Lock() - defer m.mu.Unlock() - m.partitions[id] = partition - }() - m.forceUpdatePartitionVolume(partition) + return nil } func (m *metadataManager) attachPartition(id uint64, partition MetaPartition) (err error) { - syslog.Printf("start load metaPartition %v\n", id) + syslog.Printf("start load metaPartition %v", id) partition.ForceSetMetaPartitionToLoadding() if err = partition.Start(false); err != nil { msg := fmt.Sprintf("load meta partition %v fail: %v", id, err) @@ -698,7 +690,11 @@ func (m *metadataManager) attachPartition(id uint64, partition MetaPartition) (e syslog.Println(msg) return } - m.registerPartition(id, partition) + + m.mu.Lock() + m.partitions[id] = partition + m.mu.Unlock() + msg := fmt.Sprintf("load meta partition %v success", id) log.LogInfof(msg) syslog.Println(msg) @@ -739,10 +735,6 @@ func (m *metadataManager) createPartition(request *proto.CreateMetaPartitionRequ } partition := NewMetaPartition(mpc, m) - if partition == nil { - err = errors.NewErrorf("[createPartition] partition is nil") - return - } if err = partition.RenameStaleMetadata(); err != nil { err = errors.NewErrorf("[createPartition]->%s", err.Error()) @@ -772,7 +764,6 @@ func (m *metadataManager) createPartition(request *proto.CreateMetaPartitionRequ } m.partitions[request.PartitionID] = partition }() - m.forceUpdatePartitionVolume(partition) log.LogInfof("load meta partition %v success", request.PartitionID) diff --git a/metanode/manager_op.go b/metanode/manager_op.go index 11101dea6..f429cf869 100644 --- a/metanode/manager_op.go +++ b/metanode/manager_op.go @@ -1488,7 +1488,7 @@ func (m *metadataManager) opDeleteMetaPartition(conn net.Conn, if err != nil { p.PacketOkReply() m.respondToClientWithVer(conn, p) - return + return nil } // Ack the master request conf := mp.GetBaseConfig() diff --git a/metanode/partition.go b/metanode/partition.go index ffe6a42fc..23fe10fe9 100644 --- a/metanode/partition.go +++ b/metanode/partition.go @@ -572,7 +572,6 @@ type metaPartition struct { vol *Vol manager *metadataManager isLoadingMetaPartition bool - summaryLock sync.Mutex blobClientWrapper *BlobStoreClientWrapper volType int // kept in hybrid cloud for compatibility isFollowerRead bool @@ -584,17 +583,14 @@ type metaPartition struct { uniqChecker *uniqChecker verSeq uint64 multiVersionList *proto.VolVersionInfoList - versionLock sync.Mutex verUpdateChan chan []byte enableAuditLog bool recycleInodeDelFileFlag atomicutil.Flag enablePersistAccessTime bool accessTimeValidInterval uint64 - storageTypes []uint32 statByStorageClass []*proto.StatOfStorageClass statByMigrateStorageClass []*proto.StatOfStorageClass fmList *forbiddenMigrationList - volStorageClass uint32 syncAtimeCh chan uint64 } @@ -606,6 +602,10 @@ func (mp *metaPartition) SetForbidden(status bool) { mp.config.Forbidden = status } +func (mp *metaPartition) GetVolStorageClass() uint32 { + return mp.vol.GetVolView().VolStorageClass +} + func (mp *metaPartition) IsForbidWriteOpOfProtoVer0() bool { return mp.config.ForbidWriteOpOfProtoVer0 } @@ -850,10 +850,9 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) { return } - var volumeInfo *proto.SimpleVolView retryCnt := 0 for ; retryCnt < 200; retryCnt++ { - if volumeInfo, err = masterClient.AdminAPI().GetVolumeSimpleInfo(mp.config.VolName); err != nil { + if err = mp.manager.forceUpdatePartitionVolume(mp); err != nil { log.LogWarnf("[onStart] vol(%v) mpId(%d) retryCnt(%v), GetVolumeSimpleInfo err[%v]", mp.config.VolName, mp.config.PartitionId, retryCnt, err) time.Sleep(3 * time.Second) @@ -866,40 +865,29 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) { return } - mp.vol.volDeleteLockTime = volumeInfo.DeleteLockTime + volInfo := mp.vol.GetVolView() + mp.vol.volDeleteLockTime = volInfo.DeleteLockTime if mp.manager.metaNode.clusterEnableSnapshot { go mp.runVersionOp() } - mp.volType = volumeInfo.VolType - if proto.IsValidStorageClass(volumeInfo.VolStorageClass) { - mp.volStorageClass = volumeInfo.VolStorageClass - log.LogInfof("[onStart] vol(%v) mpId(%v), from master VolStorageClass(%v)", - mp.config.VolName, mp.config.PartitionId, proto.StorageClassString(mp.volStorageClass)) - } else if volumeInfo.VolStorageClass == proto.StorageClass_Unspecified { - // handle compatibility with old version master which has no field VolStorageClass - if proto.IsValidStorageClass(legacyReplicaStorageClass) { - mp.volStorageClass = legacyReplicaStorageClass - log.LogWarnf("[onStart] vol(%v) mpId(%v), use conf legacyReplicaStorageClass(%v)", - mp.config.VolName, mp.config.PartitionId, proto.StorageClassString(legacyReplicaStorageClass)) - } else { - err = errors.NewErrorf("[onStart] vol(%v) mpId(%d), master invalid volStorageClass(%v) and conf legacyReplicaStorageClass not set", - mp.config.VolName, mp.config.PartitionId, volumeInfo.VolStorageClass) - return - } - } else { + mp.volType = volInfo.VolType + if !proto.IsValidStorageClass(volInfo.VolStorageClass) { err = errors.NewErrorf("[onStart] vol(%v) mpId(%d), get from master invalid volStorageClass(%v)", - mp.config.VolName, mp.config.PartitionId, volumeInfo.VolStorageClass) + mp.config.VolName, mp.config.PartitionId, volInfo.VolStorageClass) return } - if proto.IsCold(mp.volType) || proto.IsVolSupportStorageClass(volumeInfo.AllowedStorageClass, proto.StorageClass_BlobStore) { + log.LogInfof("[onStart] vol(%v) mpId(%v), from master VolStorageClass(%v)", + mp.config.VolName, mp.config.PartitionId, proto.StorageClassString(mp.GetVolStorageClass())) + + if proto.IsCold(mp.volType) || proto.IsVolSupportStorageClass(volInfo.AllowedStorageClass, proto.StorageClass_BlobStore) { mp.blobClientWrapper, err = NewBlobStoreClientWrapper(access.Config{ ConnMode: access.NoLimitConnMode, Consul: access.ConsulConfig{ Address: gClusterInfo.EbsAddr, // gClusterInfo is fetched from master in register procedure }, - MaxSizePutOnce: int64(volumeInfo.ObjBlockSize), + MaxSizePutOnce: int64(volInfo.ObjBlockSize), Logger: &access.Logger{Filename: path.Join(log.LogDir, "ebs.log")}, }) @@ -1292,7 +1280,12 @@ func (mp *metaPartition) load(isCreate bool) (err error) { } - return mp.LoadSnapshot(snapshotPath) + err = mp.LoadSnapshot(snapshotPath) + if err != nil { + return err + } + + return nil } func (mp *metaPartition) store(sm *storeMsg) (err error) { diff --git a/metanode/partition_free_list.go b/metanode/partition_free_list.go index 96c60e886..5f08efd19 100644 --- a/metanode/partition_free_list.go +++ b/metanode/partition_free_list.go @@ -59,7 +59,7 @@ func (mp *metaPartition) startFreeList() (err error) { return } - go mp.updateVolWorker() + // go mp.updateVolWorker() go mp.deleteWorker() go mp.startRecycleInodeDelFile() mp.startToDeleteExtents() @@ -88,6 +88,7 @@ func (mp *metaPartition) UpdateVolumeView(dataView *proto.DataPartitionsView, vo return newView } mp.vol.UpdatePartitions(convert(dataView)) + mp.vol.SetVolView(volumeView) mp.vol.volDeleteLockTime = volumeView.DeleteLockTime mp.enablePersistAccessTime = volumeView.EnablePersistAccessTime if volumeView.AccessTimeInterval <= proto.MinAccessTimeValidInterval { @@ -96,61 +97,6 @@ func (mp *metaPartition) UpdateVolumeView(dataView *proto.DataPartitionsView, vo atomic.StoreUint64(&mp.accessTimeValidInterval, uint64(volumeView.AccessTimeInterval)) } -func (mp *metaPartition) updateVolView(convert func(view *proto.DataPartitionsView) *DataPartitionsView) (err error) { - volName := mp.config.VolName - dataView, err := masterClient.ClientAPI().EncodingGzip().GetDataPartitions(volName) - if err != nil { - err = fmt.Errorf("updateVolWorker: get data partitions view fail: volume(%v) err(%v)", - volName, err) - log.LogErrorf(err.Error()) - return - } - mp.vol.UpdatePartitions(convert(dataView)) - - volView, err := masterClient.AdminAPI().GetVolumeSimpleInfo(volName) - if err != nil { - err = fmt.Errorf("updateVolWorker: get volumeinfo fail: volume(%v) err(%v)", volName, err) - log.LogErrorf(err.Error()) - return - } - mp.vol.volDeleteLockTime = volView.DeleteLockTime - return nil -} - -func (mp *metaPartition) updateVolWorker() { - t := time.NewTicker(UpdateVolTicket) - convert := func(view *proto.DataPartitionsView) *DataPartitionsView { - newView := &DataPartitionsView{ - DataPartitions: make([]*DataPartition, len(view.DataPartitions)), - } - for i := 0; i < len(view.DataPartitions); i++ { - if len(view.DataPartitions[i].Hosts) < 1 { - log.LogErrorf("updateVolWorker dp id(%v) is invalid, DataPartitionResponse detail[%v]", - view.DataPartitions[i].PartitionID, view.DataPartitions[i]) - continue - } - newView.DataPartitions[i] = &DataPartition{ - PartitionID: view.DataPartitions[i].PartitionID, - Status: view.DataPartitions[i].Status, - Hosts: view.DataPartitions[i].Hosts, - ReplicaNum: view.DataPartitions[i].ReplicaNum, - IsDiscard: view.DataPartitions[i].IsDiscard, - } - } - return newView - } - mp.updateVolView(convert) - for { - select { - case <-mp.stopC: - t.Stop() - return - case <-t.C: - mp.updateVolView(convert) - } - } -} - const ( MinDeleteBatchCounts = 100 MaxSleepCnt = 10 diff --git a/metanode/partition_fsm.go b/metanode/partition_fsm.go index eb3a1369a..9575f5ea2 100644 --- a/metanode/partition_fsm.go +++ b/metanode/partition_fsm.go @@ -1003,7 +1003,7 @@ func (mp *metaPartition) HandleLeaderChange(leader uint64) { exporter.Warning(fmt.Sprintf("[HandleLeaderChange] pid %v init root inode id: %s.", mp.config.PartitionId, err.Error())) } ino := NewInode(id, proto.Mode(os.ModePerm|os.ModeDir)) - ino.StorageClass = mp.volStorageClass + ino.StorageClass = mp.GetVolStorageClass() go mp.initInode(ino) } // refresh forbidden migration list diff --git a/metanode/partition_op_extent.go b/metanode/partition_op_extent.go index fbeaced9a..3282bdfcd 100644 --- a/metanode/partition_op_extent.go +++ b/metanode/partition_op_extent.go @@ -20,7 +20,6 @@ import ( "fmt" "os" "sort" - "sync/atomic" "time" "github.com/cubefs/cubefs/util/timeutil" @@ -106,8 +105,8 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe return } if req.IsCache && req.IsMigration { - log.LogErrorf("ExtentAppendWithCheck parameter IsCache and IsMigration can not be ture at the same time") err = errors.New("ExtentAppendWithCheck parameter IsCache and IsMigration can not be ture at the same time") + log.LogError(err) reply := []byte(err.Error()) p.PacketErrorWithBody(status, reply) return @@ -137,7 +136,6 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe // TODO:hechi: if storage type is not ssd , update extent key by CacheExtentAppendWithCheck // check volume's Type: if volume's type is cold, cbfs' extent can be modify/add only when objextent exist - // if proto.IsCold(mp.volType) { if req.IsCache { i.RLock() if i.HybridCouldExtents.sortedEks == nil { @@ -165,7 +163,6 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe // extent key verSeq not set value since marshal will not include verseq // use inode verSeq instead inoParm.setVer(mp.verSeq) - if !req.IsCache { inoParm.StorageClass = req.StorageClass } @@ -453,14 +450,6 @@ func (mp *metaPartition) ExtentsList(req *proto.GetExtentsRequest, p *Packet) (e status = retMsg.Status ) - // if !proto.IsStorageClassReplica(ino.StorageClass) && (req.IsCache != true && req.IsMigration != true) { - // status = proto.OpMismatchStorageClass - // err = fmt.Errorf("ino(%v) storageClass(%v) IsCache(%v) IsMigration(%v) do not support ExtentsList", - // ino.Inode, ino.StorageClass, req.IsCache, req.IsMigration) - // p.PacketErrorWithBody(status, []byte(err.Error())) - // return - // } - if status != proto.OpOk { p.PacketErrorWithBody(status, reply) return @@ -470,7 +459,7 @@ func (mp *metaPartition) ExtentsList(req *proto.GetExtentsRequest, p *Packet) (e log.LogInfof("action[ExtentsList] inode[%v] request verseq [%v] ino ver [%v] extent size %v ino.Size %v ino[%v] hist len %v", req.Inode, req.VerSeq, ino.getVer(), len(ino.Extents.eks), ino.Size, ino, ino.getLayerLen()) - resp.WriteGeneration = atomic.LoadUint64(&ino.WriteGeneration) + resp.WriteGeneration = ino.WriteGeneration if req.VerSeq > 0 && ino.getVer() > 0 && (req.VerSeq < ino.getVer() || isInitSnapVer(req.VerSeq)) { mp.GetExtentByVer(ino, req, resp) vIno := ino.Copy().(*Inode) @@ -541,21 +530,6 @@ func (mp *metaPartition) ExtentsList(req *proto.GetExtentsRequest, p *Packet) (e reply = []byte(err.Error()) } else if !req.InnerReq { mp.persistInodeAccessTime(ino.Inode, p) - - // if inode is required for writing, mark it as forbidden migration - if req.OpenForWrite { - var val []byte - val, err = ino.Marshal() - if err != nil { - p.PacketErrorWithBody(proto.OpErr, []byte(err.Error())) - return - } - _, err = mp.submit(opFSMForbiddenMigrationInode, val) - if err != nil { - status = proto.OpErr - reply = []byte(err.Error()) - } - } } // mark inode as ForbiddenMigration p.PacketErrorWithBody(status, reply) @@ -572,44 +546,44 @@ func (mp *metaPartition) ObjExtentsList(req *proto.GetExtentsRequest, p *Packet) reply []byte status = retMsg.Status ) - if status == proto.OpOk { - if ino.StorageClass != proto.StorageClass_BlobStore { - status = proto.OpMismatchStorageClass - reply = []byte(fmt.Sprintf("Dismatch storage type, current storage type is %s", - proto.StorageClassString(ino.StorageClass))) - p.PacketErrorWithBody(status, reply) - return - } - resp := &proto.GetObjExtentsResponse{} - ino.DoReadFunc(func() { - resp.Generation = ino.Generation - resp.Size = ino.Size - // cache ek - ino.Extents.Range(func(_ int, ek proto.ExtentKey) bool { - resp.Extents = append(resp.Extents, ek) + + if status != proto.OpOk { + p.PacketErrorWithBody(status, reply) + return + } + + if ino.StorageClass != proto.StorageClass_BlobStore && ino.Size > 0 { + status = proto.OpMismatchStorageClass + reply = []byte(fmt.Sprintf("Dismatch storage type, current storage type is %s", + proto.StorageClassString(ino.StorageClass))) + p.PacketErrorWithBody(status, reply) + return + } + resp := &proto.GetObjExtentsResponse{} + ino.DoReadFunc(func() { + resp.Generation = ino.Generation + resp.Size = ino.Size + // cache ek + ino.Extents.Range(func(_ int, ek proto.ExtentKey) bool { + resp.Extents = append(resp.Extents, ek) + return true + }) + // from SortedHybridCloudExtents + if ino.HybridCouldExtents.sortedEks != nil { + objEks := ino.HybridCouldExtents.sortedEks.(*SortedObjExtents) + objEks.Range(func(ek proto.ObjExtentKey) bool { + resp.ObjExtents = append(resp.ObjExtents, ek) return true }) - // from SortedHybridCloudExtents - if ino.HybridCouldExtents.sortedEks != nil { - objEks := ino.HybridCouldExtents.sortedEks.(*SortedObjExtents) - objEks.Range(func(ek proto.ObjExtentKey) bool { - resp.ObjExtents = append(resp.ObjExtents, ek) - return true - }) - } - //ino.ObjExtents.Range(func(ek proto.ObjExtentKey) bool { - // resp.ObjExtents = append(resp.ObjExtents, ek) - // return true - //}) - }) - - reply, err = json.Marshal(resp) - if err != nil { - status = proto.OpErr - reply = []byte(err.Error()) - } else if !req.InnerReq { - mp.persistInodeAccessTime(ino.Inode, p) } + }) + + reply, err = json.Marshal(resp) + if err != nil { + status = proto.OpErr + reply = []byte(err.Error()) + } else if !req.InnerReq { + mp.persistInodeAccessTime(ino.Inode, p) } p.PacketErrorWithBody(status, reply) return diff --git a/metanode/partition_op_inode.go b/metanode/partition_op_inode.go index 5dcfba690..710943ebd 100644 --- a/metanode/partition_op_inode.go +++ b/metanode/partition_op_inode.go @@ -131,7 +131,7 @@ func txReplyInfo(inode *Inode, txInfo *proto.TransactionInfo, quotaInfos map[uin } // for compatibility: handle req from old version client without filed StorageType -func (mp *metaPartition) checkCreateInoStorageClassForCompatibility(reqStorageClass uint32, inodeId uint64) (err error, resultStorageClass uint32) { +func (mp *metaPartition) checkCreateInoStorageClassForCompatibility(reqStorageClass uint32, inodeId uint64) (resultStorageClass uint32, err error) { if proto.IsValidStorageClass(reqStorageClass) { resultStorageClass = reqStorageClass return @@ -143,26 +143,20 @@ func (mp *metaPartition) checkCreateInoStorageClassForCompatibility(reqStorageCl } if proto.IsHot(mp.volType) { - if !proto.IsValidStorageClass(legacyReplicaStorageClass) { + if !proto.IsValidStorageClass(mp.GetVolStorageClass()) { err = fmt.Errorf("CreateInode req without StorageClass but metanode config legacyReplicaStorageClass not set") return } - resultStorageClass = legacyReplicaStorageClass + resultStorageClass = mp.GetVolStorageClass() log.LogDebugf("legacy CreateInode req, hot vol(%v), mpId(%v), ino(%v), set ino storageClass as config legacyReplicaStorageClass(%v)", - mp.config.VolName, mp.config.PartitionId, inodeId, proto.StorageClassString(legacyReplicaStorageClass)) + mp.config.VolName, mp.config.PartitionId, inodeId, proto.StorageClassString(resultStorageClass)) } else { resultStorageClass = proto.StorageClass_BlobStore log.LogDebugf("legacy CreateInode req, cold vol(%v), mpId(%v), ino(%v), set ino storageClass as blobstore", 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 } @@ -182,7 +176,7 @@ func (mp *metaPartition) CreateInode(req *CreateInoReq, p *Packet, remoteAddr st auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.GetFullPath(), err, time.Since(start).Milliseconds(), inoID, 0) }() } - if err, requiredStorageClass = mp.checkCreateInoStorageClassForCompatibility(req.StorageType, inoID); err != nil { + if requiredStorageClass, err = mp.checkCreateInoStorageClassForCompatibility(req.StorageType, inoID); err != nil { p.PacketErrorWithBody(proto.OpErr, []byte(err.Error())) log.LogErrorf("[CreateInode] %v, req(%+v)", err.Error(), req) return @@ -252,7 +246,7 @@ func (mp *metaPartition) QuotaCreateInode(req *proto.QuotaCreateInodeRequest, p auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.GetFullPath(), err, time.Since(start).Milliseconds(), inoID, 0) }() } - if err, requiredStorageClass = mp.checkCreateInoStorageClassForCompatibility(req.StorageType, inoID); err != nil { + if requiredStorageClass, err = mp.checkCreateInoStorageClassForCompatibility(req.StorageType, inoID); err != nil { p.PacketErrorWithBody(proto.OpErr, []byte(err.Error())) log.LogErrorf("[QuotaCreateInode] %v, req(%+v)", err.Error(), req) return @@ -978,7 +972,7 @@ func (mp *metaPartition) TxCreateInode(req *proto.TxCreateInodeRequest, p *Packe auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.GetFullPath(), err, time.Since(start).Milliseconds(), inoID, 0) }() } - if err, requiredStorageClass = mp.checkCreateInoStorageClassForCompatibility(req.StorageType, inoID); err != nil { + if requiredStorageClass, err = mp.checkCreateInoStorageClassForCompatibility(req.StorageType, inoID); err != nil { p.PacketErrorWithBody(proto.OpErr, []byte(err.Error())) log.LogErrorf("[QuotaCreateInode] %v, req(%+v)", err.Error(), req) return