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

feat(masterclientpreload): add property CacheDpStorageClass to volume.

Browse Source 
CacheDpStorageClass is used to inform SDK which storageClass to use when access cache dp.

Signed-off-by: true1064 <tangjingyu@oppo.com>
This commit is contained in:
true1064 2023-11-23 03:14:41 +08:00 committed by AmazingChi
parent 92d1d54832
commit 1f9dab0eda
19 changed files with 105 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cli/cmd/fmt.go
View File

@ -194,6 +194,7 @@ func formatSimpleVolView(svv *proto.SimpleVolView) string {
}
sb.WriteString(fmt.Sprintf(" VolStorageClass : %v\n", proto.StorageClassString(svv.VolStorageClass)))
sb.WriteString(fmt.Sprintf(" AllowedStorageClass : %v\n", allowedStorageClassStr))
sb.WriteString(fmt.Sprintf(" CacheDpStorageClass : %v\n", svv.CacheDpStorageClass))
if svv.VolType == 1 {
sb.WriteString(fmt.Sprintf(" ObjBlockSize : %v byte\n", svv.ObjBlockSize))

16
client/fs/super.go
View File

@ -88,8 +88,13 @@ type Super struct {
ebsc *blobstore.BlobStoreClient
sc *SummaryCache
taskPool []common.TaskPool
closeC chan struct{}
taskPool []common.TaskPool
closeC chan struct{}
disableTrash bool
enableVerRead bool
cacheDpStorageClass uint32
}
// Functions that Super needs to implement
@ -215,6 +220,8 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
s.bc = bcache.NewBcacheClient()
}
s.cacheDpStorageClass = opt.CacheDpStorageClass
extentConfig := &stream.ExtentConfig{
Volume: opt.Volname,
Masters: masters,
@ -240,6 +247,7 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
MinWriteAbleDataPartitionCnt: opt.MinWriteAbleDataPartitionCnt,
StreamRetryTimeout: opt.StreamRetryTimeout,
OnRenewalForbiddenMigration: s.mw.RenewalForbiddenMigration,
CacheDpStorageClass: s.cacheDpStorageClass,
}
s.ec, err = stream.NewExtentClient(extentConfig)
@ -284,8 +292,8 @@ func NewSuper(opt *proto.MountOptions) (s *Super, err error) {
atomic.StoreUint32((*uint32)(&s.state), uint32(fs.FSStatRestore))
}
log.LogInfof("NewSuper: cluster(%v) volname(%v) icacheExpiration(%v) LookupValidDuration(%v) AttrValidDuration(%v) state(%v)",
s.cluster, s.volname, inodeExpiration, LookupValidDuration, AttrValidDuration, s.state)
log.LogInfof("NewSuper: cluster(%v) volname(%v) icacheExpiration(%v) LookupValidDuration(%v) AttrValidDuration(%v) state(%v) cacheDpStorageClass(%v)",
s.cluster, s.volname, inodeExpiration, LookupValidDuration, AttrValidDuration, s.state, s.cacheDpStorageClass)
go s.loopSyncMeta()

1
client/fuse.go
View File

@ -1063,6 +1063,7 @@ func loadConfFromMaster(opt *proto.MountOptions) (err error) {
opt.TxTimeout = volumeInfo.TxTimeout
opt.TxConflictRetryNum = volumeInfo.TxConflictRetryNum
opt.TxConflictRetryInterval = volumeInfo.TxConflictRetryInterval
opt.CacheDpStorageClass = volumeInfo.CacheDpStorageClass
var clusterInfo *proto.ClusterInfo
clusterInfo, err = mc.AdminAPI().GetClusterInfo()

1
master/api_args_parse.go
View File

@ -718,6 +718,7 @@ type createVolReq struct {
volStorageClass uint32
allowedStorageClass []uint32 // format is uint32 Separated by commas: "StorageClass1, StorageClass2, ..."
hasMultiReplicaStorageClass bool
cacheDpStorageClass uint32
}
func checkCacheAction(action int) error {

8
master/api_service.go
View File

@ -2484,11 +2484,17 @@ func (m *Server) checkStorageClassForCreateVolReq(req *createVolReq) (err error)
log.LogInfof("[checkStorageClassForCreateVol] create vol(%v), volStorageClass not specified, auto set as: %v",
req.name, proto.StorageClassString(req.volStorageClass))
} else if proto.IsStorageClassBlobStore(req.volStorageClass) {
req.cacheDpStorageClass = m.cluster.GetFastReplicaStorageClassFromCluster(resourceChecker)
log.LogInfof("[checkStorageClassForCreateVol] create vol(%v) volStorageClass(%v) set cacheDpStorageClass: %v",
req.name, proto.StorageClassString(req.volStorageClass), proto.StorageClassString(req.cacheDpStorageClass))
} else if !proto.IsValidStorageClass(req.volStorageClass) {
err = fmt.Errorf("invalid volStorageClass: %v", req.volStorageClass)
log.LogErrorf("[checkStorageClassForCreateVol] create vol(%v) err:%v", req.name, err.Error())
return err
}
log.LogInfof("[checkStorageClassForCreateVol] volStorageClass: %v", proto.StorageClassString(req.volStorageClass))
if len(req.allowedStorageClass) == 0 {
@ -2868,6 +2874,7 @@ func newSimpleView(vol *Vol) (view *proto.SimpleVolView) {
AllowedStorageClass: vol.allowedStorageClass,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
}
view.AllowedStorageClass = make([]uint32, len(vol.allowedStorageClass))
copy(view.AllowedStorageClass, vol.allowedStorageClass)
@ -5369,6 +5376,7 @@ func volStat(vol *Vol, countByMeta bool) (stat *proto.VolStatInfo) {
stat.TrashInterval = vol.TrashInterval
stat.DefaultStorageClass = vol.volStorageClass
stat.CacheDpStorageClass = vol.cacheDpStorageClass
log.LogDebugf("[volStat] vol[%v] total[%v],usedSize[%v] TrashInterval[%v] DefaultStorageClass[%v]",
vol.Name, stat.TotalSize, stat.UsedSize, stat.TrashInterval, stat.DefaultStorageClass)
if proto.IsHot(vol.VolType) {

30
master/cluster.go
View File

@ -1626,21 +1626,19 @@ func (c *Cluster) batchCreatePreLoadDataPartition(vol *Vol, preload *DataPartiti
return fmt.Errorf("vol type is not warm"), nil
}
total := overSoldCap(uint64(preload.preloadCacheCapacity))
reqCreateCount := (total-1)/(util.DefaultDataPartitionSize/util.GB) + 1
chosenStorageClass := c.GetFastReplicaStorageClassFromCluster(nil)
if chosenStorageClass == proto.StorageClass_Unspecified {
log.LogErrorf("cluster has no resource to create preload data partition, vol(%v)", vol.Name)
if vol.cacheDpStorageClass == proto.StorageClass_Unspecified {
err = fmt.Errorf(" has no resource to create preload data partition")
log.LogErrorf("[batchCreatePreLoadDataPartition] vol(%v) err: %v", vol.Name, err.Error())
return err, nil
}
total := overSoldCap(uint64(preload.preloadCacheCapacity))
reqCreateCount := (total-1)/(util.DefaultDataPartitionSize/util.GB) + 1
for i := 0; i < int(reqCreateCount); i++ {
log.LogInfof("create preload data partition (%v) total (%v)", i, reqCreateCount)
var dp *DataPartition
chosenMediaType := proto.GetMediaTypeByStorageClass(chosenStorageClass)
if dp, err = c.createDataPartition(vol.Name, preload, chosenMediaType); err != nil {
if dp, err = c.createDataPartition(vol.Name, preload, vol.cacheDpStorageClass); err != nil {
log.LogErrorf("create preload data partition fail: volume(%v) err(%v)", vol.Name, err)
return err, nil
}
@ -2665,7 +2663,7 @@ func (c *Cluster) migrateDataPartition(srcAddr, targetAddr string, dp *DataParti
if targetAddr != "" {
targetHosts = []string{targetAddr}
} else if targetHosts, _, err = ns.getAvailDataNodeHosts(dp.Hosts, 1, dataNode.MediaType); err != nil {
} else if targetHosts, _, err = ns.getAvailDataNodeHosts(dp.Hosts, 1, dp.MediaType); err != nil {
if _, ok := c.vols[dp.VolName]; !ok {
log.LogWarnf("clusterID[%v] partitionID:%v on node:%v offline failed,PersistenceHosts:[%v]",
c.Name, dp.PartitionID, srcAddr, dp.Hosts)
@ -3696,6 +3694,10 @@ func (c *Cluster) createVol(req *createVolReq) (vol *Vol, err error) {
// NOTE: init data partitions
if proto.IsStorageClassReplica(vol.volStorageClass) && vol.Capacity > 0 {
for _, acs := range req.allowedStorageClass {
if !proto.IsStorageClassReplica(acs) {
continue
}
chosenMediaType := proto.GetMediaTypeByStorageClass(acs)
if readWriteDataPartitions, err = c.initDataPartitionsForCreateVol(vol, req.dpCount, chosenMediaType); err != nil {
goto errHandler
@ -3704,11 +3706,10 @@ func (c *Cluster) createVol(req *createVolReq) (vol *Vol, err error) {
req.name, readWriteDataPartitions, proto.MediaTypeString(chosenMediaType))
}
} else if proto.IsStorageClassBlobStore(vol.volStorageClass) && vol.CacheCapacity > 0 {
chosenStorageClass := c.GetFastReplicaStorageClassFromCluster(nil)
chosenMediaType := proto.GetMediaTypeByStorageClass(chosenStorageClass)
chosenMediaType := proto.GetMediaTypeByStorageClass(vol.cacheDpStorageClass)
log.LogInfof("action[createVol] vol[%v] to create cache dp with storageClass(%v) for blobStore",
req.name, proto.StorageClassString(vol.cacheDpStorageClass))
log.LogInfof("action[createVol] vol[%v] choose cache dp storageClass(%v) for blobStore",
req.name, proto.StorageClassString(chosenStorageClass))
if readWriteDataPartitions, err = c.initDataPartitionsForCreateVol(vol, req.dpCount, chosenMediaType); err != nil {
goto errHandler
}
@ -3793,6 +3794,7 @@ func (c *Cluster) doCreateVol(req *createVolReq) (vol *Vol, err error) {
VolStorageClass: req.volStorageClass,
AllowedStorageClass: req.allowedStorageClass,
CacheDpStorageClass: req.cacheDpStorageClass,
}
log.LogInfof("[doCreateVol] volView, %v", vv)

2
master/cluster_test.go
View File

@ -403,6 +403,7 @@ func TestCreateVolWithDpCount(t *testing.T) {
zoneName: testZone1 + "," + testZone2,
description: "",
qosLimitArgs: &qosArgs{},
volStorageClass: defaultVolStorageClass,
}
_, err := server.cluster.createVol(req)
require.NoError(t, err)
@ -430,6 +431,7 @@ func TestCreateVolWithDpCount(t *testing.T) {
zoneName: testZone1 + "," + testZone2,
description: "",
qosLimitArgs: &qosArgs{},
volStorageClass: defaultVolStorageClass,
}
_, err := server.cluster.createVol(req)
require.Error(t, err)

4
master/metadata_fsm_op.go
View File

@ -336,6 +336,7 @@ type volValue struct {
VolStorageClass uint32
AllowedStorageClass []uint32
CacheDpStorageClass uint32
}
func (v *volValue) Bytes() (raw []byte, err error) {
@ -408,7 +409,8 @@ func newVolValue(vol *Vol) (vv *volValue) {
AccessTimeInterval: vol.AccessTimeValidInterval,
EnablePersistAccessTime: vol.EnablePersistAccessTime,
VolStorageClass: vol.volStorageClass,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
}
vv.AllowedStorageClass = make([]uint32, len(vol.allowedStorageClass))
copy(vv.AllowedStorageClass, vol.allowedStorageClass)

10
master/vol.go
View File

@ -160,6 +160,7 @@ type Vol struct {
// hybrid cloud
allowedStorageClass []uint32 // specifies which storageClasses the vol use, a cluster may have multiple StorageClasses
volStorageClass uint32 // specifies which storageClass is written, unless dirStorageClass is set in file path
cacheDpStorageClass uint32 // for SDK those access cache/preload dp of cold volume
}
func newVol(vv volValue) (vol *Vol) {
@ -234,6 +235,7 @@ func newVol(vv volValue) (vol *Vol) {
vol.allowedStorageClass = make([]uint32, len(vv.AllowedStorageClass))
copy(vol.allowedStorageClass, vv.AllowedStorageClass)
vol.volStorageClass = vv.VolStorageClass
vol.cacheDpStorageClass = vv.CacheDpStorageClass
return
}
@ -1107,8 +1109,12 @@ func (vol *Vol) autoCreateDataPartitions(c *Cluster) {
return
}
cacheStorageClass := c.GetFastReplicaStorageClassFromCluster(nil)
cacheMediaType := proto.GetMediaTypeByStorageClass(cacheStorageClass)
if vol.cacheDpStorageClass == proto.StorageClass_Unspecified {
log.LogErrorf("action[autoCreateDataPartitions] no resource to create cache data partition, vol(%v)", vol.Name)
return
}
cacheMediaType := proto.GetMediaTypeByStorageClass(vol.cacheDpStorageClass)
count := (maxSize-allocSize-1)/vol.dataPartitionSize + 1
log.LogInfof("action[autoCreateDataPartitions] vol[%v] count[%v] volStorageClass[%v], chosenMediaType(%v)",
vol.Name, count, proto.StorageClassString(vol.volStorageClass), proto.MediaTypeString(cacheMediaType))

11
metanode/partition_op_extent.go
View File

@ -145,9 +145,10 @@ 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)
inoParm.StorageClass = req.StorageClass
//TODO:tangjingyu
//if isCachem: inoParm.StorageClass set as inode's old storageClass
if !req.IsCache {
inoParm.StorageClass = req.StorageClass
}
if req.IsCache {
inoParm.Extents.Append(ext)
@ -155,8 +156,8 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe
inoParm.HybridCouldExtents.sortedEks = NewSortedExtents()
inoParm.HybridCouldExtents.sortedEks.(*SortedExtents).Append(ext)
}
log.LogDebugf("ExtentAppendWithCheck: ino(%v) mp(%v) isCache(%v) verSeq (%v)",
req.Inode, req.PartitionID, req.IsCache, mp.verSeq)
log.LogDebugf("ExtentAppendWithCheck: ino(%v) mp(%v) isCache(%v) verSeq (%v) storageClass(%v)",
req.Inode, req.PartitionID, req.IsCache, mp.verSeq, proto.StorageClassString(inoParm.StorageClass))
// Store discard extents right after the append extent key.
if len(req.DiscardExtents) != 0 {

25
preload/sdk/preloadsdk.go
View File

@ -61,6 +61,7 @@ type PreLoadClient struct {
ebsBlockSize int
preloadFileNumTotal int64
preloadFileNumSucceed int64
CacheDpStorageClass uint32
}
type fileInfo struct {
@ -489,15 +490,14 @@ func (c *PreLoadClient) preloadFileWorker(id int64, jobs <-chan fileInfo, wg *sy
log.LogWarnf("Read (%v) wrong size:(%v)", objExtent, n)
continue
}
//TODO:default cache type is ssd? //TODO:tangjingyu: how to get cache storage class from master?
_, err = c.ec.Write(ino, int(objExtent.FileOffset), buf, 0, nil, proto.StorageClass_Replica_SSD, false)
// in preload mode,onece extend_hander set to error, streamer is set to error
// so write should failed immediately
_, err = c.ec.Write(ino, int(objExtent.FileOffset), buf, 0, nil, c.CacheDpStorageClass, false)
//in preload mode,once extend_hander set to error, streamer is set to error
// so write should fail immediately
if err != nil {
subErr = true
log.LogWarnf("preload (%v) to cbfs failed (%v)", job.name, err)
//TODO:default cache type is ssd?
if err = c.ec.GetDataPartitionForWrite(proto.StorageClass_Replica_SSD); err != nil {
if err = c.ec.GetDataPartitionForWrite(c.CacheDpStorageClass); err != nil {
log.LogErrorf("worker %v end for %v", id, err)
noWritableDP = true
}
@ -543,16 +543,25 @@ func (c *PreLoadClient) preloadFile() error {
}
}
func (c *PreLoadClient) CheckColdVolume() bool {
func (c *PreLoadClient) CheckVolumeInfoFromMaster() bool {
var (
err error
view *proto.SimpleVolView
)
if view, err = c.mc.AdminAPI().GetVolumeSimpleInfo(c.vol); err != nil {
log.LogErrorf("getSimpleVolView: get volume simple info fail: volume(%v) err(%v)", c.vol, err)
log.LogErrorf("CheckVolumeInfoFromMaster: get volume simple info fail: volume(%v) err(%v)", c.vol, err)
return false
}
if view.CacheDpStorageClass == proto.StorageClass_Unspecified {
log.LogErrorf("CheckVolumeInfoFromMaster: volume(%v) cacheDpStorageClass not specified", c.vol)
return false
}
c.CacheDpStorageClass = view.CacheDpStorageClass
log.LogInfof("CheckVolumeInfoFromMaster: volume(%v) cacheDpStorageClass: %v",
c.vol, proto.StorageClassString(c.CacheDpStorageClass))
return view.VolType == proto.VolumeTypeCold
}

13
proto/admin_proto.go
View File

@ -1483,3 +1483,16 @@ func GetMediaTypeByStorageClass(storageClass uint32) (mediaType uint32) {
return
}
func GetStorageClassByMediaType(mediaType uint32) (storageClass uint32) {
switch mediaType {
case MediaType_SSD:
storageClass = StorageClass_Replica_SSD
case MediaType_HDD:
storageClass = StorageClass_Replica_HDD
default:
storageClass = StorageClass_Unspecified
}
return
}

1
proto/model.go
View File

@ -281,6 +281,7 @@ type VolStatInfo struct {
DpReadOnlyWhenVolFull bool
TrashInterval int64 `json:"TrashIntervalV2"`
DefaultStorageClass uint32
CacheDpStorageClass uint32
}
// DataPartition represents the structure of storing the file contents.

1
proto/mount_options.go
View File

@ -309,6 +309,7 @@ type MountOptions struct {
TxTimeout int64
TxConflictRetryNum int64
TxConflictRetryInterval int64
CacheDpStorageClass uint32
VolType int
EbsEndpoint string
EbsServicePath string

7
sdk/data/blobstore/reader.go
View File

@ -406,10 +406,9 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent
streamer.WorkAsCache()
}
//TODO:tangjingyu handle cache dp mediaType
////cold volume's cache has only one storage class of datapartitions, so use proto.StorageClass_Unspecified here
reader.ec.Write(reader.ino, int(objExtentKey.FileOffset), buf, proto.FlagsCache, nil, proto.StorageClass_Replica_SSD, false)
log.LogDebugf("TRACE blobStore asyncCache(L2) Exit. cacheKey=%v", cacheKey)
reader.ec.Write(reader.ino, int(objExtentKey.FileOffset), buf, proto.FlagsCache, nil, reader.ec.CacheDpStorageClass, false)
log.LogDebugf("TRACE blobStore asyncCache(L2) Exit. storageClass(%v) cacheKey=%v",
proto.StorageClassString(reader.ec.CacheDpStorageClass), cacheKey)
return
}

6
sdk/data/blobstore/writer.go
View File

@ -507,9 +507,9 @@ func (writer *Writer) asyncCache(ino uint64, offset int, data []byte) {
stat.EndStat("write-async-cache", err, bgTime, 1)
}()
log.LogDebugf("TRACE asyncCache Enter,fileOffset(%v) len(%v)", offset, len(data))
//TODO: default cache type is ssd? //TODO:tangjingyu how to get cache storage class from master
write, err := writer.ec.Write(ino, offset, data, proto.FlagsCache, nil, proto.StorageClass_Replica_SSD, false)
log.LogDebugf("TRACE asyncCache Enter,fileOffset(%v) len(%v), storageClass(%v)",
offset, len(data), proto.StorageClassString(writer.ec.CacheDpStorageClass))
write, err := writer.ec.Write(ino, offset, data, proto.FlagsCache, nil, writer.ec.CacheDpStorageClass, false)
log.LogDebugf("TRACE asyncCache Exit,write(%v) err(%v)", write, err)
}

6
sdk/data/stream/extent_client.go
View File

@ -145,6 +145,8 @@ type ExtentConfig struct {
StreamRetryTimeout int
OnRenewalForbiddenMigration RenewalForbiddenMigrationFunc
CacheDpStorageClass uint32
}
type MultiVerMgr struct {
@ -184,6 +186,7 @@ type ExtentClient struct {
inflightL1BigBlock int32
multiVerMgr *MultiVerMgr
renewalForbiddenMigration RenewalForbiddenMigrationFunc
CacheDpStorageClass uint32
}
func (client *ExtentClient) UidIsLimited(uid uint32) bool {
@ -340,6 +343,9 @@ retry:
log.LogInfof("max streamer limit %d", client.maxStreamerLimit)
client.streamerList = list.New()
client.CacheDpStorageClass = config.CacheDpStorageClass
go client.backgroundEvictStream()
return

10
sdk/data/stream/extent_handler.go
View File

@ -137,7 +137,7 @@ func NewExtentHandler(stream *Streamer, offset int, storeMode int, size int, sto
stop: make(chan struct{}),
meetLimitedIoError: false,
verUpdate: make(chan uint64),
mediaType: storageClass,
mediaType: proto.GetMediaTypeByStorageClass(storageClass),
}
go eh.receiver()
@ -466,7 +466,7 @@ func (eh *ExtentHandler) appendExtentKey() (err error) {
ekey := *eh.key
doAppend := func() (err error) {
discard := eh.stream.extents.Append(&ekey, true)
status, err = eh.stream.client.appendExtentKey(eh.stream.parentInode, eh.inode, ekey, discard, eh.stream.isCache, eh.mediaType)
status, err = eh.stream.client.appendExtentKey(eh.stream.parentInode, eh.inode, ekey, discard, eh.stream.isCache, proto.GetStorageClassByMediaType(eh.mediaType))
if atomic.LoadInt32(&eh.stream.needUpdateVer) > 0 {
if errUpdateExtents := eh.stream.GetExtentsForceRefresh(); errUpdateExtents != nil {
log.LogErrorf("action[appendExtentKey] inode %v GetExtents err %v errUpdateExtents %v", eh.stream.inode, err, errUpdateExtents)
@ -614,12 +614,6 @@ func (eh *ExtentHandler) allocateExtent() (err error) {
for i := 0; i < MaxSelectDataPartitionForWrite; i++ {
if eh.key == nil {
//TODO:tangjinyu ignore mediaType when write to cache dp
//mediaType := eh.mediaType
//if eh.stream.isCache {
// mediaType = proto.MediaType_Unspecified
//}
if dp, err = eh.stream.client.dataWrapper.GetDataPartitionForWrite(exclude, eh.mediaType); err != nil {
log.LogWarnf("allocateExtent: failed to get write data partition, eh(%v) exclude(%v), clear exclude and try again!", eh, exclude)
exclude = make(map[string]struct{})

1
sdk/meta/meta.go
View File

@ -184,6 +184,7 @@ type MetaWrapper struct {
Client wrapper.SimpleClientInfo
IsSnapshotEnabled bool
DefaultStorageClass uint32
CacheDpStorageClass uint32
}
type uniqidRange struct {