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(meta): fix some issue about compatiable. #22780321

Browse Source 
1. remove useless code.
2. fix compatiable bug abount dataMediatType.

Signed-off-by: Victor1319 <zengxuewei@oppo.com>
This commit is contained in:
Victor1319 2024-11-12 19:09:32 +08:00 committed by AmazingChi
parent 60401d3e87
commit 44d1d95161
10 changed files with 106 additions and 184 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
metanode/datapartition.go
View File

@ -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()

11
metanode/inode.go
View File

@ -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 {

4
metanode/inode_test.go
View File

@ -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))

29
metanode/manager.go
View File

@ -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)

2
metanode/manager_op.go
View File

@ -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()

49
metanode/partition.go
View File

@ -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) {

58
metanode/partition_free_list.go
View File

@ -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

2
metanode/partition_fsm.go
View File

@ -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

100
metanode/partition_op_extent.go
View File

@ -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

20
metanode/partition_op_inode.go
View File

@ -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