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(meta): hybrid cloud metapartition can start when failed to create blobStore client

Browse Source 
Signed-off-by: true1064 <tangjingyu@oppo.com>
This commit is contained in:
true1064 2024-07-10 19:29:20 +08:00 committed by AmazingChi
parent 5c20dc2109
commit 4d6ee74eb2
4 changed files with 99 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
master/cluster.go
View File

@ -1927,7 +1927,8 @@ func (c *Cluster) createDataPartition(volName string, preload *DataPartitionPreL
return
errHandler:
err = fmt.Errorf("action[createDataPartition],clusterID[%v] vol[%v] Err:%v ", c.Name, volName, err.Error())
err = fmt.Errorf("action[createDataPartition],clusterID[%v] vol[%v] mediaType(%v) Err:%v ",
c.Name, volName, proto.MediaTypeString(mediaType), err.Error())
log.LogError(errors.Stack(err))
Warn(c.Name, err.Error())
return

11
metanode/const.go
View File

@ -254,11 +254,12 @@ const (
intervalToPersistData = time.Minute * 5
intervalToSyncCursor = time.Minute * 1
defaultDelExtentsCnt = 100000
defaultMaxQuotaGoroutine = 5
defaultQuotaSwitch = true
DefaultNameResolveInterval = 1 // minutes
DefaultRaftNumOfLogsToRetain = 20000 * 2
defaultDelExtentsCnt = 100000
defaultMaxQuotaGoroutine = 5
defaultQuotaSwitch = true
DefaultNameResolveInterval = 1 // minutes
DefaultRaftNumOfLogsToRetain = 20000 * 2
DefaultCreateBlobClientIntervalSec = 30
)
const (

95
metanode/partition.go
View File

@ -479,6 +479,66 @@ type OpQuota interface {
getInodeQuota(inode uint64, p *Packet) (err error)
}
type BlobStoreClientWrapper struct {
blobClientLock sync.Mutex
blobClient *blobstore.BlobStoreClient
cfg *access.Config
lastTryCreateTime int64
}
func NewBlobStoreClientWrapper(cfg access.Config) (blobWrapper *BlobStoreClientWrapper, err error) {
blobWrapper = &BlobStoreClientWrapper{
blobClient: nil,
cfg: &cfg,
lastTryCreateTime: 0,
}
if _, _, err = blobWrapper.getBlobStoreClient(); err != nil {
return blobWrapper, err
}
return blobWrapper, nil
}
// will create blobstore client if not created or creation failed earlier
func (ew *BlobStoreClientWrapper) getBlobStoreClient() (blobClient *blobstore.BlobStoreClient, create bool, err error) {
ew.blobClientLock.Lock()
defer ew.blobClientLock.Unlock()
create = false
if ew.blobClient != nil {
return ew.blobClient, create, nil
}
if ew.cfg.Consul.Address == "" {
// TODO:tangjingyu get blobstore addr from master
err = errors.New("addr is empty, can not create blobstore client")
return nil, create, err
}
if time.Now().Unix()-ew.lastTryCreateTime < DefaultCreateBlobClientIntervalSec {
err = fmt.Errorf("addr(%v) create blobstore client failed, wait a while to try create again", ew.cfg.Consul.Address)
return nil, create, err
}
blobClient, err = blobstore.NewEbsClient(*(ew.cfg))
if err != nil {
err = fmt.Errorf("addr(%v) create blobstore client err: %v", ew.cfg.Consul.Address, err.Error())
ew.lastTryCreateTime = time.Now().Unix()
return nil, create, err
} else if blobClient == nil {
err = fmt.Errorf("addr(%v) create blobstore client is nil", ew.cfg.Consul.Address)
ew.lastTryCreateTime = time.Now().Unix()
return nil, create, err
}
log.LogDebugf("[getBlobStoreClient] addr(%v) create blobstore client success", clusterInfo.EbsAddr)
ew.blobClient = blobClient
ew.lastTryCreateTime = 0
create = true
return ew.blobClient, create, nil
}
// metaPartition manages the range of the inode IDs.
// When a new inode is requested, it allocates a new inode id for this inode if possible.
// States:
@ -507,7 +567,8 @@ type metaPartition struct {
vol *Vol
manager *metadataManager
isLoadingMetaPartition bool
ebsClient *blobstore.BlobStoreClient
summaryLock sync.Mutex
blobClientWrapper *BlobStoreClientWrapper
volType int // kept in hybrid cloud for compatibility
isFollowerRead bool
uidManager *UidManager
@ -763,28 +824,24 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) {
mp.volType = volumeInfo.VolType
mp.volStorageClass = volumeInfo.VolStorageClass
if clusterInfo.EbsAddr != "" && proto.IsVolSupportStorageClass(volumeInfo.AllowedStorageClass, proto.StorageClass_BlobStore) {
var ebsClient *blobstore.BlobStoreClient
ebsClient, err = blobstore.NewEbsClient(
access.Config{
ConnMode: access.NoLimitConnMode,
Consul: access.ConsulConfig{
Address: clusterInfo.EbsAddr,
},
MaxSizePutOnce: int64(volumeInfo.ObjBlockSize),
Logger: &access.Logger{Filename: path.Join(log.LogDir, "ebs.log")},
if proto.IsCold(mp.volType) || proto.IsVolSupportStorageClass(volumeInfo.AllowedStorageClass, proto.StorageClass_BlobStore) {
mp.blobClientWrapper, err = NewBlobStoreClientWrapper(access.Config{
ConnMode: access.NoLimitConnMode,
Consul: access.ConsulConfig{
Address: clusterInfo.EbsAddr,
},
)
MaxSizePutOnce: int64(volumeInfo.ObjBlockSize),
Logger: &access.Logger{Filename: path.Join(log.LogDir, "ebs.log")},
})
if err != nil {
log.LogErrorf("action[onStart] err[%v]", err)
return
log.LogWarnf("action[onStart] mp(%v) blobStoreAddr(%v), create blobstore client err[%v], but still start mp and will try create blobstore later",
mp.config.PartitionId, clusterInfo.EbsAddr, err)
// not return err here, blobstore client may be created latter
} else {
log.LogInfof("action[onStart] mp(%v) blobStoreAddr(%v), create blobstore client success",
mp.config.PartitionId, clusterInfo.EbsAddr)
}
if ebsClient == nil {
err = errors.NewErrorf("[onStart] ebsClient is nil")
return
}
mp.ebsClient = ebsClient
}
go mp.startCheckerEvict()

22
metanode/partition_free_list.go
View File

@ -27,6 +27,7 @@ import (
"time"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/sdk/data/blobstore"
"github.com/cubefs/cubefs/util"
"github.com/cubefs/cubefs/util/errors"
"github.com/cubefs/cubefs/util/fileutil"
@ -835,18 +836,25 @@ func (mp *metaPartition) doBatchDeleteObjExtentsInEBS(allInodes []*Inode, isMigr
}
func (mp *metaPartition) deleteObjExtents(oeks []proto.ObjExtentKey) (err error) {
var blobClient *blobstore.BlobStoreClient
var blobCreate bool
blobClient, blobCreate, err = mp.blobClientWrapper.getBlobStoreClient()
if err != nil {
log.LogErrorf("[deleteObjExtents] vol(%v) mp(%v) failed: %v", mp.config.VolName, mp.config.PartitionId, err.Error())
return
}
if blobCreate {
log.LogInfof("[deleteObjExtents] vol(%v) mp(%v) create blob client success", mp.config.VolName, mp.config.PartitionId)
}
total := len(oeks)
for i := 0; i < total; i += maxDelCntOnce {
max := util.Min(i+maxDelCntOnce, total)
if mp.ebsClient == nil {
err = fmt.Errorf("(%v) error ebsClient nil", mp.config.PartitionId)
log.LogErrorf("deleteObjExtents %v", err)
return
}
err = mp.ebsClient.Delete(oeks[i:max])
err = blobClient.Delete(oeks[i:max])
if err != nil {
log.LogErrorf("[deleteObjExtents] vol(%v) mp(%v) delete ebs eks fail, cnt(%d), err(%s)", mp.config.VolName, mp.config.PartitionId, max-i, err.Error())
log.LogErrorf("[deleteObjExtents] vol(%v) mp(%v) delete ebs eks fail, cnt(%d), err(%s)",
mp.config.VolName, mp.config.PartitionId, max-i, err.Error())
return err
}
}