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(clustermgr): add shardnode disk and node management

Browse Source 
. #22432395

Signed-off-by: tangdeyi <tangdeyi@oppo.com>
This commit is contained in:
tangdeyi 2024-07-23 19:43:22 +08:00 committed by slasher
parent 6c8ece4ba6
commit f20a06c956
30 changed files with 1601 additions and 651 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
blobstore/access/controller/cluster.go
View File

@ -262,8 +262,8 @@ func (c *clusterControllerImpl) loadWithConfig() error {
}
clusterInfo := &cmapi.ClusterInfo{}
clusterInfo.ClusterID = cs.ClusterID
clusterInfo.Capacity = stat.SpaceStat.TotalSpace
clusterInfo.Available = stat.SpaceStat.WritableSpace
clusterInfo.Capacity = stat.BlobNodeSpaceStat.TotalSpace
clusterInfo.Available = stat.BlobNodeSpaceStat.WritableSpace
clusterInfo.Nodes = cs.Hosts
clusterInfo.Readonly = stat.ReadOnly

2
blobstore/access/controller/cluster_test.go
View File

@ -146,7 +146,7 @@ func stat(w http.ResponseWriter, req *http.Request) {
info := &clustermgr.StatInfo{
LeaderHost: hostAddr,
ReadOnly: false,
SpaceStat: clustermgr.SpaceStatInfo{
BlobNodeSpaceStat: clustermgr.SpaceStatInfo{
TotalSpace: 1 << 40,
WritableSpace: 1 << 20,
},

11
blobstore/api/clustermgr/disk.go
View File

@ -134,9 +134,11 @@ type DiskHeartbeatRet struct {
type DiskStatInfo struct {
IDC string `json:"idc"`
Total int `json:"total"`
TotalChunk int64 `json:"total_chunk"`
TotalFreeChunk int64 `json:"total_free_chunk"`
TotalOversoldFreeChunk int64 `json:"total_oversold_free_chunk"`
TotalChunk int64 `json:"total_chunk,omitempty"`
TotalFreeChunk int64 `json:"total_free_chunk,omitempty"`
TotalOversoldFreeChunk int64 `json:"total_oversold_free_chunk,omitempty"`
TotalShard int32 `json:"total_shard,omitempty"`
TotalFreeShard int32 `json:"total_free_shard,omitempty"`
Available int `json:"available"`
Readonly int `json:"readonly"`
Expired int `json:"expired"`
@ -153,7 +155,8 @@ type SpaceStatInfo struct {
ReadOnlySpace int64 `json:"readonly_space"` // free physical space which is readonly
UsedSpace int64 `json:"used_space"` // used physical space
WritableSpace int64 `json:"writable_space"` // writable logical space
TotalBlobNode int64 `json:"total_blob_node"`
TotalBlobNode int64 `json:"total_blob_node,omitempty"`
TotalShardNode int64 `json:"total_shard_node,omitempty"`
TotalDisk int64 `json:"total_disk"`
DisksStatInfos []DiskStatInfo `json:"disk_stat_infos"`
}

11
blobstore/api/clustermgr/proto.go
View File

@ -37,11 +37,12 @@ type ClusterInfo struct {
}
type StatInfo struct {
LeaderHost string `json:"leader_host"`
ReadOnly bool `json:"read_only"`
RaftStatus interface{} `json:"raft_status"`
SpaceStat SpaceStatInfo `json:"space_stat"`
VolumeStat VolumeStatInfo `json:"volume_stat"`
LeaderHost string `json:"leader_host"`
ReadOnly bool `json:"read_only"`
RaftStatus interface{} `json:"raft_status"`
BlobNodeSpaceStat SpaceStatInfo `json:"space_stat"`
ShardNodeSpaceStat SpaceStatInfo `json:"shard_node_space_stat"`
VolumeStat VolumeStatInfo `json:"volume_stat"`
}
func GetConsulClusterPath(region string) string {

8
blobstore/cli/access/cluster.go
View File

@ -118,8 +118,8 @@ func showClusterWithConfig() error {
clusterInfo := &cmapi.ClusterInfo{}
ClusterID, _ := strconv.ParseUint(clusterID, 10, 32)
clusterInfo.ClusterID = proto.ClusterID(ClusterID)
clusterInfo.Capacity = stat.SpaceStat.TotalSpace
clusterInfo.Available = stat.SpaceStat.WritableSpace
clusterInfo.Capacity = stat.BlobNodeSpaceStat.TotalSpace
clusterInfo.Available = stat.BlobNodeSpaceStat.WritableSpace
clusterInfo.Nodes = hosts
clusterInfo.Readonly = stat.ReadOnly
@ -127,8 +127,8 @@ func showClusterWithConfig() error {
fmt.Println()
fmt.Printf("\tspace in cluster: %s (%s / %s)\n", clusterID,
common.ColorizeInt64(-stat.SpaceStat.WritableSpace, stat.SpaceStat.TotalSpace).Sprint(humanize.IBytes(uint64(stat.SpaceStat.WritableSpace))),
humanize.IBytes(uint64(stat.SpaceStat.TotalSpace)))
common.ColorizeInt64(-stat.BlobNodeSpaceStat.WritableSpace, stat.BlobNodeSpaceStat.TotalSpace).Sprint(humanize.IBytes(uint64(stat.BlobNodeSpaceStat.WritableSpace))),
humanize.IBytes(uint64(stat.BlobNodeSpaceStat.TotalSpace)))
}
return nil
}

0
blobstore/clustermgr/node.go → blobstore/clustermgr/blobnode.go
View File

10
blobstore/clustermgr/disk.go → blobstore/clustermgr/blobnode_disk.go
View File

@ -27,11 +27,11 @@ import (
"github.com/cubefs/cubefs/blobstore/util/errors"
)
func (s *Service) DiskIdAlloc(c *rpc.Context) {
func (s *Service) DiskIDAlloc(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
span.Info("accept DiskIdAlloc request")
span.Info("accept DiskIDAlloc request")
diskID, err := s.BlobNodeMgr.AllocDiskID(ctx)
if err != nil {
span.Error("alloc disk id failed =>", errors.Detail(err))
@ -325,14 +325,14 @@ func (s *Service) DiskHeartbeat(c *rpc.Context) {
ReadOnly: info.Readonly,
}
// filter frequentHeatBeat disk
frequentHeatBeat, err := s.BlobNodeMgr.IsFrequentHeatBeat(args.Disks[i].DiskID, s.HeartbeatNotifyIntervalS)
// filter frequentHeartBeat disk
frequentHeartBeat, err := s.BlobNodeMgr.IsFrequentHeartBeat(args.Disks[i].DiskID, s.HeartbeatNotifyIntervalS)
if err != nil {
span.Errorf("get disk info %d failed, err: %v", args.Disks[i].DiskID, err)
c.RespondError(err)
return
}
if !frequentHeatBeat {
if !frequentHeartBeat {
heartbeatDisks = append(heartbeatDisks, args.Disks[i])
} else {
span.Warnf("disk %d heartbeat too frequent", args.Disks[i].DiskID)

0
blobstore/clustermgr/disk_test.go → blobstore/clustermgr/blobnode_disk_test.go
View File

2
blobstore/clustermgr/node_test.go → blobstore/clustermgr/blobnode_test.go
View File

@ -109,7 +109,7 @@ func TestTopoInfo(t *testing.T) {
var diskSetMaxLen, nodeSetMaxLen int
blobNodeHDDNodeSets := ret.AllNodeSets[proto.DiskTypeHDD.String()]
copySetConf := testService.Config.DiskMgrConfig.CopySetConfigs[proto.NodeRoleBlobNode][proto.DiskTypeHDD]
copySetConf := testService.Config.BlobNodeDiskMgrConfig.CopySetConfigs[proto.DiskTypeHDD]
diskSetCap, nodeSetCap, diskSetIdcCap := copySetConf.DiskSetCap, copySetConf.NodeSetCap, copySetConf.NodeSetIdcCap
for _, nodeSet := range blobNodeHDDNodeSets {
if nodeSet.Number > nodeSetMaxLen {

84
blobstore/clustermgr/cluster/alloc.go
View File

@ -115,7 +115,7 @@ func (a *allocator) Alloc(ctx context.Context, diskType proto.DiskType, mode cod
Disks: _disks,
})
}
// update diskset and nodeset free chunk
// update diskset and nodeset free item
atomic.AddInt64(&diskSetAllocator.weight, -int64(allocCount))
atomic.AddInt64(&nodeSetAllocator.weight, -int64(allocCount))
@ -166,7 +166,7 @@ func (a *allocator) allocNodeSet(ctx context.Context, diskType proto.DiskType, m
return nodeSetAllocator, nil
}
// choose nodeset by free chunk count weight
// choose nodeset by free item count weight
total := len(nodeSetAllocators)
totalWeight := int64(0)
allocatableNodeSets := make([]*nodeSetAllocator, 0, total)
@ -251,7 +251,7 @@ type diskSetAllocator struct {
func (d *diskSetAllocator) alloc(ctx context.Context, count int) (ret []*idcAllocator) {
span := trace.SpanFromContextSafe(ctx)
for _, idcAllocator := range d.idcAllocators {
nodeNum := len(idcAllocator.blobNodeStorages)
nodeNum := len(idcAllocator.nodeStorages)
if idcAllocator.diffHost && nodeNum < count {
span.Errorf("allocate diff host idcAllocator from diskSet: %d failed, allocate num: %d, node num: %d", d.diskSetID, count, nodeNum)
continue
@ -273,16 +273,16 @@ type idcAllocator struct {
diffRack bool
diffHost bool
rackStorages map[string]*rackAllocator
blobNodeStorages []*nodeAllocator
rackStorages map[string]*rackAllocator
nodeStorages []*nodeAllocator
}
// rackAllocator represent an rack storage info
type rackAllocator struct {
rack string
// weight should always read and write by atomic
weight int64
blobNodeStorages []*nodeAllocator
weight int64
nodeStorages []*nodeAllocator
}
// nodeAllocator represent an data node storage info
@ -294,7 +294,7 @@ type nodeAllocator struct {
disks []*diskItem
}
// allocDisk will choose disk by disk free chunk count weight
// allocDisk will choose disk by disk free item count weight
func (d *nodeAllocator) allocDisk(ctx context.Context, excludes map[proto.DiskID]*diskItem) (chosenDisk *diskItem) {
span := trace.SpanFromContextSafe(ctx)
totalWeight := atomic.LoadInt64(&d.weight)
@ -351,7 +351,7 @@ func (s *idcAllocator) alloc(ctx context.Context, count int, excludes map[proto.
ret := make([]proto.DiskID, 0)
totalWeight := atomic.LoadInt64(&s.weight)
span.Debugf("%s idc total free chunk: %d", s.idc, totalWeight)
span.Debugf("%s idc total free item: %d", s.idc, totalWeight)
if totalWeight < int64(count) {
return nil, ErrNoEnoughSpace
}
@ -359,11 +359,11 @@ func (s *idcAllocator) alloc(ctx context.Context, count int, excludes map[proto.
if s.diffRack && s.diffHost {
chosenRacks, chosenDataStorages, chosenDisks = s.allocFromRack(ctx, count, excludes)
} else {
chosenDataStorages, chosenDisks = s.allocFromBlobNodeStorages(ctx, count, totalWeight-defaultAllocTolerateBuff, s.blobNodeStorages, excludes)
chosenDataStorages, chosenDisks = s.allocFromNodeStorages(ctx, count, totalWeight-defaultAllocTolerateBuff, s.nodeStorages, excludes)
}
if len(chosenDisks) < count {
span.Warnf("alloc failed, chosenRacks: %v, chosenBlobNodeStorages: %+v, chosenDisks: %v", chosenRacks, chosenDataStorages, chosenDisks)
span.Warnf("alloc failed, chosenRacks: %v, chosenNodeStorages: %+v, chosenDisks: %v", chosenRacks, chosenDataStorages, chosenDisks)
return nil, ErrNoEnoughSpace
}
@ -385,7 +385,7 @@ func (s *idcAllocator) alloc(ctx context.Context, count int, excludes map[proto.
return ret, nil
}
// 1. alloc rack with free chunk weight
// 1. alloc rack with free item weight
// 2. alloc from rack's data node storage
// 3. if can't meet the alloc count request, then retry with enable same rack
func (s *idcAllocator) allocFromRack(ctx context.Context, count int, excludes map[proto.DiskID]*diskItem) (chosenRacksRet map[string]int, chosenDataStorages map[*nodeAllocator]int, chosenDisks map[proto.DiskID]*diskItem) {
@ -420,11 +420,11 @@ RETRY:
rack := rackStorage.rack
weight := atomic.LoadInt64(&rackStorage.weight)
if weight > 0 && weight >= randNum && chosenRacksNum[rack] <= duplicatedCount &&
(s.diffHost && len(rackStorage.blobNodeStorages) > chosenRacksNum[rack]) {
(s.diffHost && len(rackStorage.nodeStorages) > chosenRacksNum[rack]) {
allocNum := 1
if _, ok := chosenRacksNum[rack]; ok {
// retry with same rack, add all rest num into chosenRacksNum
allocNum = len(rackStorage.blobNodeStorages) - chosenRacksNum[rack]
allocNum = len(rackStorage.nodeStorages) - chosenRacksNum[rack]
chosenRacksNum[rack] += allocNum
} else {
chosenRacks = append(chosenRacks, rack)
@ -454,7 +454,7 @@ RETRY:
}
span.Infof("chosen racks: %v, chosen racks num: %v", chosenRacks, chosenRacksNum)
// shuffle chosen racks, [0-count) will range by rack free chunk weight
// shuffle chosen racks, [0-count) will range by rack free item weight
// [count, total) will be shuffle by random, ensure allocation more evenly
total := len(chosenRacks)
if total > count {
@ -465,14 +465,14 @@ RETRY:
}
}
// alloc chunk from rack's blobNodeStorages
// alloc item from rack's nodeStorages
_count = count
for _, rack := range chosenRacks {
num := chosenRacksNum[rack]
if num > _count {
num = _count
}
dataStorages, disks := s.allocFromBlobNodeStorages(ctx, num, atomic.LoadInt64(&s.rackStorages[rack].weight), s.rackStorages[rack].blobNodeStorages, excludes)
dataStorages, disks := s.allocFromNodeStorages(ctx, num, atomic.LoadInt64(&s.rackStorages[rack].weight), s.rackStorages[rack].nodeStorages, excludes)
for id := range disks {
chosenDisks[id] = disks[id]
chosenRacksRet[rack]++
@ -492,7 +492,7 @@ RETRY:
// 1. copy rack's nodeAllocator pointer array
// 2. alloc from nodeAllocator array
// 3. the alloc result length may not equal to count if there is no enough space or something else
func (s *idcAllocator) allocFromBlobNodeStorages(ctx context.Context, count int, totalWeight int64, srcBlobNodeStorages []*nodeAllocator, excludes map[proto.DiskID]*diskItem) (chosenDataStorages map[*nodeAllocator]int, chosenDisks map[proto.DiskID]*diskItem) {
func (s *idcAllocator) allocFromNodeStorages(ctx context.Context, count int, totalWeight int64, srcNodeStorages []*nodeAllocator, excludes map[proto.DiskID]*diskItem) (chosenDataStorages map[*nodeAllocator]int, chosenDisks map[proto.DiskID]*diskItem) {
span := trace.SpanFromContextSafe(ctx)
excludeHosts := make(map[string]bool)
chosenDisks = make(map[proto.DiskID]*diskItem)
@ -505,22 +505,22 @@ func (s *idcAllocator) allocFromBlobNodeStorages(ctx context.Context, count int,
diskInfo.lock.RUnlock()
}
blobNodeStorages := make([]*nodeAllocator, 0, len(s.blobNodeStorages))
blobNodeStorageNum := 0
// build available blobNodeStorages, filter exclude host or disk
for i := range srcBlobNodeStorages {
nodeStorages := make([]*nodeAllocator, 0, len(s.nodeStorages))
nodeStorageNum := 0
// build available nodeStorages, filter exclude host or disk
for i := range srcNodeStorages {
// not allow same host, then filter exclude host
if s.diffHost && excludeHosts[srcBlobNodeStorages[i].host] {
weight := atomic.LoadInt64(&srcBlobNodeStorages[i].weight)
if s.diffHost && excludeHosts[srcNodeStorages[i].host] {
weight := atomic.LoadInt64(&srcNodeStorages[i].weight)
totalWeight -= weight
continue
}
blobNodeStorages = append(blobNodeStorages, srcBlobNodeStorages[i])
nodeStorages = append(nodeStorages, srcNodeStorages[i])
// allow same host, then exclude target disk. it's quite slowly but alright in test env which enable same host alloc
if !s.diffHost && len(excludes) > 0 {
weight := atomic.LoadInt64(&srcBlobNodeStorages[i].weight)
newDisks := make([]*diskItem, 0, len(srcBlobNodeStorages[i].disks))
for _, disk := range srcBlobNodeStorages[i].disks {
weight := atomic.LoadInt64(&srcNodeStorages[i].weight)
newDisks := make([]*diskItem, 0, len(srcNodeStorages[i].disks))
for _, disk := range srcNodeStorages[i].disks {
if _, ok := excludes[disk.diskID]; ok {
diskWeight := disk.weight()
totalWeight -= diskWeight
@ -529,20 +529,20 @@ func (s *idcAllocator) allocFromBlobNodeStorages(ctx context.Context, count int,
}
newDisks = append(newDisks, disk)
}
blobNodeStorages[blobNodeStorageNum] = &nodeAllocator{
host: srcBlobNodeStorages[i].host,
nodeStorages[nodeStorageNum] = &nodeAllocator{
host: srcNodeStorages[i].host,
weight: weight,
disks: newDisks,
}
}
blobNodeStorageNum += 1
nodeStorageNum += 1
}
span.Debugf("total blobNodeStorages num: %d, excludes host: %v, excludes disk: %v", blobNodeStorageNum, excludeHosts, excludes)
span.Debugf("total nodeStorages num: %d, excludes host: %v, excludes disk: %v", nodeStorageNum, excludeHosts, excludes)
// no available data node after exclude, then return
if blobNodeStorageNum == 0 {
if nodeStorageNum == 0 {
return
}
// no available chunk after exclude, then return
// no available item after exclude, then return
if totalWeight <= 0 {
return
}
@ -550,8 +550,8 @@ func (s *idcAllocator) allocFromBlobNodeStorages(ctx context.Context, count int,
chosenIdx := 0
retryTimes := 0
maxRetryTimes := defaultRetryTimes
// maxRetry times will equal to count when blobNodeStorageNum less than target count
if blobNodeStorageNum < count {
// maxRetry times will equal to count when nodeStorageNum less than target count
if nodeStorageNum < count {
maxRetryTimes = count
}
_totalWeight := totalWeight
@ -564,14 +564,14 @@ RETRY:
} else {
randNum = 0
}
for i := chosenIdx; i < blobNodeStorageNum; i++ {
weight := atomic.LoadInt64(&blobNodeStorages[i].weight)
span.Debugf("total free chunk: %d, blobNode(%s) free chunk: %d, randNum: %d", _totalWeight, blobNodeStorages[i].host, weight, randNum)
for i := chosenIdx; i < nodeStorageNum; i++ {
weight := atomic.LoadInt64(&nodeStorages[i].weight)
span.Debugf("total free item: %d, node(%s) free item: %d, randNum: %d", _totalWeight, nodeStorages[i].host, weight, randNum)
if weight >= randNum {
if selectedDisk := blobNodeStorages[i].allocDisk(ctx, chosenDisks); selectedDisk != nil {
if selectedDisk := nodeStorages[i].allocDisk(ctx, chosenDisks); selectedDisk != nil {
chosenDisks[selectedDisk.diskID] = selectedDisk
chosenDataStorages[blobNodeStorages[i]] += 1
blobNodeStorages[chosenIdx], blobNodeStorages[i] = blobNodeStorages[i], blobNodeStorages[chosenIdx]
chosenDataStorages[nodeStorages[i]] += 1
nodeStorages[chosenIdx], nodeStorages[i] = nodeStorages[i], nodeStorages[chosenIdx]
_totalWeight -= weight
count -= 1
chosenIdx += 1

6
blobstore/clustermgr/cluster/alloc_test.go
View File

@ -47,7 +47,7 @@ var testDiskMgrConfig = DiskMgrConfig{
ChunkSize: 17179869184, // 16G
CodeModes: []codemode.CodeMode{codemode.EC15P12, codemode.EC6P6},
ChunkOversoldRatio: 0.5,
CopySetConfigs: make(map[proto.NodeRole]map[proto.DiskType]CopySetConfig),
CopySetConfigs: make(map[proto.DiskType]CopySetConfig),
}
var (
@ -67,8 +67,8 @@ func initTestDiskMgr(t *testing.T) (d *BlobNodeManager, closeFunc func()) {
ctrl := gomock.NewController(t)
defer ctrl.Finish()
testMockScopeMgr = mock.NewMockScopeMgrAPI(ctrl)
testDiskMgrConfig.CopySetConfigs[proto.NodeRoleBlobNode] = make(map[proto.DiskType]CopySetConfig)
testDiskMgrConfig.CopySetConfigs[proto.NodeRoleBlobNode][proto.DiskTypeHDD] = CopySetConfig{
testDiskMgrConfig.CopySetConfigs = make(map[proto.DiskType]CopySetConfig)
testDiskMgrConfig.CopySetConfigs[proto.DiskTypeHDD] = CopySetConfig{
NodeSetCap: 108,
NodeSetIdcCap: 36,
NodeSetRackCap: 6,

41
blobstore/clustermgr/cluster/blobnode_mgr.go
View File

@ -253,6 +253,10 @@ func (b *BlobNodeManager) AddDisk(ctx context.Context, args *clustermgr.BlobNode
if err := b.CheckDiskInfoDuplicated(ctx, args.DiskID, &args.DiskInfo, &node.info.NodeInfo); err != nil {
return err
}
// disk idc/rack/host uses node one
args.Idc = node.info.Idc
args.Rack = node.info.Rack
args.Host = node.info.Host
return nil
})
if err != nil {
@ -822,7 +826,7 @@ func (b *BlobNodeManager) applyAddDisk(ctx context.Context, info *clustermgr.Blo
if err != nil {
return nil
}
info.DiskSetID = b.topoMgr.AllocDiskSetID(ctx, &info.DiskInfo, &node.info.NodeInfo, b.cfg.CopySetConfigs[node.info.Role][node.info.DiskType])
info.DiskSetID = b.topoMgr.AllocDiskSetID(ctx, &info.DiskInfo, &node.info.NodeInfo, b.cfg.CopySetConfigs[node.info.DiskType])
}
// calculate free and max chunk count
@ -854,41 +858,6 @@ func (b *BlobNodeManager) applyAddDisk(ctx context.Context, info *clustermgr.Blo
return nil
}
// applyAddNode add a new node into cluster, it returns ErrNodeExist if node already exist
func (b *BlobNodeManager) applyAddNode(ctx context.Context, info *clustermgr.BlobNodeInfo) error {
span := trace.SpanFromContextSafe(ctx)
b.metaLock.Lock()
defer b.metaLock.Unlock()
// concurrent double check
_, ok := b.allNodes[info.NodeID]
if ok {
return nil
}
// alloc NodeSetID
if info.NodeSetID == nullNodeSetID {
info.NodeSetID = b.topoMgr.AllocNodeSetID(ctx, &info.NodeInfo, b.cfg.CopySetConfigs[info.Role][info.DiskType], b.cfg.RackAware)
}
info.Status = proto.NodeStatusNormal
ni := &nodeItem{nodeID: info.NodeID, info: nodeItemInfo{NodeInfo: info.NodeInfo}, disks: make(map[proto.DiskID]*diskItem)}
// add node to nodeTbl and nodeSet
err := b.persistentHandler.updateNodeNoLocked(ni)
if err != nil {
span.Error("diskMgr.addNode add node failed: ", err)
return errors.Info(err, "diskMgr.addNode add node failed").Detail(err)
}
b.topoMgr.AddNodeToNodeSet(ni)
b.allNodes[info.NodeID] = ni
b.hostPathFilter.Store(ni.genFilterKey(), ni.nodeID)
return nil
}
func (b *BlobNodeManager) applyAdminUpdateDisk(ctx context.Context, diskInfo *clustermgr.BlobNodeDiskInfo) error {
span := trace.SpanFromContextSafe(ctx)
disk, ok := b.allDisks[diskInfo.DiskID]

210
blobstore/clustermgr/cluster/blobnode_mgr_refresh.go Normal file
View File

@ -0,0 +1,210 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implieb. See the License for the specific language governing
// permissions and limitations under the License.
package cluster
import (
"context"
"encoding/json"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
// maxHeap are used to build max heap, and we can calculate writable space that satisfied with stripe count
type maxHeap []int64
func (h maxHeap) Len() int { return len(h) }
func (h maxHeap) Less(i, j int) bool { return h[i] > h[j] }
func (h maxHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *maxHeap) Push(x interface{}) {
*h = append(*h, x.(int64))
}
func (h *maxHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// refresh use for refreshing storage allocator info and cluster statistic info
func (b *BlobNodeManager) refresh(ctx context.Context) {
span := trace.SpanFromContextSafe(ctx)
// space and disk stat info
spaceStatInfos := make(map[proto.DiskType]*clustermgr.SpaceStatInfo)
// generate diskType -> nodeSet -> diskSet -> idc -> rack -> blobnode storage and statInfo
nodeSetAllocators := make(map[proto.DiskType]nodeSetAllocatorMap)
diskSetAllocators := make(map[proto.DiskType]diskSetAllocatorMap)
ecDiskSet := make(map[proto.DiskType][]*diskItem)
nodeSetsMap := b.topoMgr.GetAllNodeSets(ctx)
for diskType, nodeSets := range nodeSetsMap {
if _, ok := nodeSetAllocators[diskType]; !ok {
nodeSetAllocators[diskType] = make(nodeSetAllocatorMap)
}
if _, ok := diskSetAllocators[diskType]; !ok {
diskSetAllocators[diskType] = make(diskSetAllocatorMap)
}
if _, ok := spaceStatInfos[diskType]; !ok {
spaceStatInfos[diskType] = &clustermgr.SpaceStatInfo{}
}
spaceStatInfo := spaceStatInfos[diskType]
diskStatInfo := make(map[string]*clustermgr.DiskStatInfo)
for i := range b.cfg.IDC {
diskStatInfo[b.cfg.IDC[i]] = &clustermgr.DiskStatInfo{IDC: b.cfg.IDC[i]}
}
for _, nodeSet := range nodeSets {
nodeSetAllocator := newNodeSetAllocator(nodeSet.ID())
for _, diskSet := range nodeSet.GetDiskSets() {
disks := diskSet.GetDisks()
// ecDiskSet[diskType] = append(ecDiskSet[diskType], disks...)
idcAllocators, diskSetFreeChunk := b.generateDiskSetStorage(ctx, disks, spaceStatInfo, diskStatInfo)
diskSetAllocator := newDiskSetAllocator(diskSet.ID(), diskSetFreeChunk, idcAllocators)
diskSetAllocators[diskType][diskSet.ID()] = diskSetAllocator
nodeSetAllocator.addDiskSet(diskSetAllocator)
}
nodeSetAllocators[diskType][nodeSet.ID()] = nodeSetAllocator
}
for idc := range diskStatInfo {
spaceStatInfo.DisksStatInfos = append(spaceStatInfo.DisksStatInfos, *diskStatInfo[idc])
}
spaceStatInfo.TotalBlobNode = int64(b.topoMgr.GetNodeNum(diskType))
}
// compatible
allDisks := b.getAllDisk()
span.Debugf("get all disks, len:%d", len(allDisks))
diskTypeDisks := make(map[proto.DiskType][]*diskItem)
for _, disk := range allDisks {
diskType := b.getDiskType(disk)
diskTypeDisks[diskType] = append(diskTypeDisks[diskType], disk)
}
for diskType := range diskTypeDisks {
if _, ok := nodeSetAllocators[diskType]; !ok {
nodeSetAllocators[diskType] = make(nodeSetAllocatorMap)
}
if _, ok := diskSetAllocators[diskType]; !ok {
diskSetAllocators[diskType] = make(diskSetAllocatorMap)
}
if _, ok := spaceStatInfos[diskType]; !ok {
spaceStatInfos[diskType] = &clustermgr.SpaceStatInfo{}
}
ecDiskSet[diskType] = diskTypeDisks[diskType]
ecSpaceStateInfo := &clustermgr.SpaceStatInfo{}
diskStatInfo := make(map[string]*clustermgr.DiskStatInfo)
for i := range b.cfg.IDC {
diskStatInfo[b.cfg.IDC[i]] = &clustermgr.DiskStatInfo{IDC: b.cfg.IDC[i]}
}
ecIdcAllocators, ecFreeChunk := b.generateDiskSetStorage(ctx, ecDiskSet[diskType], ecSpaceStateInfo, diskStatInfo)
// initial ec allocator
diskSetAllocator := newDiskSetAllocator(ecDiskSetID, ecFreeChunk, ecIdcAllocators)
diskSetAllocators[diskType][ecDiskSetID] = diskSetAllocator
nodeSetAllocator := newNodeSetAllocator(ecNodeSetID)
nodeSetAllocator.addDiskSet(diskSetAllocator)
nodeSetAllocators[diskType][ecNodeSetID] = nodeSetAllocator
span.Debugf("add ec nodeset")
// update space state info
for idc := range diskStatInfo {
ecSpaceStateInfo.DisksStatInfos = append(ecSpaceStateInfo.DisksStatInfos, *diskStatInfo[idc])
}
// set blobnode space info and disk stat info by ec statistic
// TODO: calculate writable space by replicate code mode and ec code mode ratio
spaceStatInfos[diskType] = ecSpaceStateInfo
spaceStatInfos[diskType].TotalBlobNode = int64(b.topoMgr.GetNodeNum(diskType))
}
b.allocator.Store(newAllocator(allocatorConfig{
nodeSets: nodeSetAllocators,
diskSets: diskSetAllocators,
dg: b,
tg: b.topoMgr,
diffHost: b.cfg.HostAware,
diffRack: b.cfg.RackAware,
}))
b.spaceStatInfo.Store(spaceStatInfos)
}
func (b *BlobNodeManager) checkDroppingNode(ctx context.Context) {
if !b.raftServer.IsLeader() {
return
}
span := trace.SpanFromContextSafe(ctx)
droppingNodeDBs, err := b.nodeTbl.GetAllDroppingNode()
if err != nil {
span.Warnf("get dropping nodes failed:%v", err)
return
}
var diskItems []*diskItem
for _, nodeID := range droppingNodeDBs {
node, _ := b.getNode(nodeID)
node.withRLocked(func() error {
// copy diskIDs of node, avoid nested node and disk lock
diskItems = make([]*diskItem, 0, len(node.disks))
for _, di := range node.disks {
diskItems = append(diskItems, di)
}
return nil
})
// check disk status
for _, di := range diskItems {
err = di.withRLocked(func() error {
if di.needFilter() {
return errors.New("node has disk in use")
}
return nil
})
if err != nil {
break
}
}
if err != nil {
span.Debugf("checkDroppingNode node: %d has disk in use", node.nodeID)
continue
}
args := &clustermgr.NodeInfoArgs{NodeID: nodeID}
data, err := json.Marshal(args)
if err != nil {
span.Errorf("checkDroppingNode json marshal failed, args: %v, error: %v", args, err)
return
}
proposeInfo := base.EncodeProposeInfo(b.GetModuleName(), OperTypeDroppedNode, data, base.ProposeContext{ReqID: span.TraceID()})
err = b.raftServer.Propose(ctx, proposeInfo)
if err != nil {
span.Errorf("checkDroppingNode dropped node: %d failed: %v", node.nodeID, err)
return
}
span.Debugf("checkDroppingNode dropped node: %d success", node.nodeID)
}
}

0
blobstore/clustermgr/cluster/refresh_test.go → blobstore/clustermgr/cluster/blobnode_mgr_refresh_test.go
View File

307
blobstore/clustermgr/cluster/cluster.go
View File

@ -15,8 +15,10 @@
package cluster
import (
"container/heap"
"context"
"fmt"
"math"
"sync"
"sync/atomic"
"time"
@ -87,7 +89,7 @@ type NodeManagerAPI interface {
// IsDroppingDisk return true if the specified disk is dropping
IsDroppingDisk(ctx context.Context, id proto.DiskID) (bool, error)
// Stat return disk statistic info of a cluster
Stat(ctx context.Context) *clustermgr.SpaceStatInfo
Stat(ctx context.Context, diskType proto.DiskType) *clustermgr.SpaceStatInfo
// GetHeartbeatChangeDisks return any heartbeat change disks
GetHeartbeatChangeDisks() []HeartbeatEvent
// ValidateNodeInfo validate node info and return any validation error when validate fail
@ -126,7 +128,7 @@ type DiskMgrConfig struct {
HeartbeatExpireIntervalS int `json:"heartbeat_expire_interval_s"`
FlushIntervalS int `json:"flush_interval_s"`
ApplyConcurrency uint32 `json:"apply_concurrency"`
BlobNodeConfig blobnode.Config `json:"blob_node_config"`
BlobNodeConfig blobnode.Config `json:"blob_node_config"` // TODO add ShardNodeConfig
AllocTolerateBuffer int64 `json:"alloc_tolerate_buffer"`
EnsureIndex bool `json:"ensure_index"`
IDC []string `json:"-"`
@ -136,7 +138,7 @@ type DiskMgrConfig struct {
DiskIDScopeName string `json:"-"`
NodeIDScopeName string `json:"-"`
CopySetConfigs map[proto.NodeRole]map[proto.DiskType]CopySetConfig `json:"copy_set_configs"`
CopySetConfigs map[proto.DiskType]CopySetConfig `json:"copy_set_configs"`
}
type CopySetConfig struct {
@ -196,8 +198,8 @@ func (d *manager) AllocDiskID(ctx context.Context) (proto.DiskID, error) {
return proto.DiskID(diskID), nil
}
// IsFrequentHeatBeat judge disk heartbeat interval whether small than HeartbeatNotifyIntervalS
func (d *manager) IsFrequentHeatBeat(id proto.DiskID, HeartbeatNotifyIntervalS int) (bool, error) {
// IsFrequentHeartBeat judge disk heartbeat interval whether small than HeartbeatNotifyIntervalS
func (d *manager) IsFrequentHeartBeat(id proto.DiskID, HeartbeatNotifyIntervalS int) (bool, error) {
diskInfo, ok := d.getDisk(id)
if !ok {
return false, apierrors.ErrCMDiskNotFound
@ -344,9 +346,9 @@ func (d *manager) IsDroppingDisk(ctx context.Context, id proto.DiskID) (bool, er
}
// Stat return disk statistic info of a cluster
func (d *manager) Stat(ctx context.Context) *clustermgr.SpaceStatInfo {
func (d *manager) Stat(ctx context.Context, diskType proto.DiskType) *clustermgr.SpaceStatInfo {
spaceStatInfo := d.spaceStatInfo.Load().(map[proto.DiskType]*clustermgr.SpaceStatInfo)
diskTypeInfo, ok := spaceStatInfo[proto.DiskTypeHDD]
diskTypeInfo, ok := spaceStatInfo[diskType]
if !ok {
return &clustermgr.SpaceStatInfo{}
}
@ -457,6 +459,9 @@ func (d *manager) CheckNodeInfoDuplicated(ctx context.Context, info *clustermgr.
}
func (d *manager) ValidateNodeInfo(ctx context.Context, info *clustermgr.NodeInfo) error {
if !info.Role.IsValid() {
return apierrors.ErrIllegalArguments
}
if !info.DiskType.IsValid() {
return apierrors.ErrIllegalArguments
}
@ -469,6 +474,57 @@ func (d *manager) ValidateNodeInfo(ctx context.Context, info *clustermgr.NodeInf
return nil
}
// applyAddNode add a new node into cluster, it returns ErrNodeExist if node already exist
func (d *manager) applyAddNode(ctx context.Context, info interface{}) error {
span := trace.SpanFromContextSafe(ctx)
var nodeInfo clustermgr.NodeInfo
shardNodeInfo, isShardNode := info.(*clustermgr.ShardNodeInfo)
if isShardNode {
nodeInfo = shardNodeInfo.NodeInfo
}
blobNodeInfo, isBlobNode := info.(*clustermgr.BlobNodeInfo)
if isBlobNode {
nodeInfo = blobNodeInfo.NodeInfo
}
d.metaLock.Lock()
defer d.metaLock.Unlock()
// concurrent double check
_, ok := d.allNodes[nodeInfo.NodeID]
if ok {
return nil
}
// alloc NodeSetID
if nodeInfo.NodeSetID == nullNodeSetID {
nodeInfo.NodeSetID = d.topoMgr.AllocNodeSetID(ctx, &nodeInfo, d.cfg.CopySetConfigs[nodeInfo.DiskType], d.cfg.RackAware)
}
nodeInfo.Status = proto.NodeStatusNormal
ni := &nodeItem{
nodeID: nodeInfo.NodeID,
info: nodeItemInfo{NodeInfo: nodeInfo},
disks: make(map[proto.DiskID]*diskItem),
}
if isShardNode {
ni.info.extraInfo = shardNodeInfo.ShardNodeExtraInfo
}
// add node to nodeTbl and nodeSet
err := d.persistentHandler.updateNodeNoLocked(ni)
if err != nil {
span.Error("ShardNodeManager.addNode add node failed: ", err)
return errors.Info(err, "ShardNodeManager.addNode add node failed").Detail(err)
}
d.topoMgr.AddNodeToNodeSet(ni)
d.allNodes[nodeInfo.NodeID] = ni
d.hostPathFilter.Store(ni.genFilterKey(), ni.nodeID)
return nil
}
// droppingDisk add a dropping disk
func (d *manager) applyDroppingDisk(ctx context.Context, id proto.DiskID, isCommit bool) (bool, error) {
span := trace.SpanFromContextSafe(ctx)
@ -723,6 +779,243 @@ func (d *manager) validateAllocRet(disks []proto.DiskID) error {
return nil
}
func (d *manager) generateDiskSetStorage(ctx context.Context, disks []*diskItem, spaceStatInfo *clustermgr.SpaceStatInfo,
diskStatInfosM map[string]*clustermgr.DiskStatInfo,
) (ret map[string]*idcAllocator, freeChunk int64) {
span := trace.SpanFromContextSafe(ctx)
nodeStgs := make(map[string]*nodeAllocator)
idcFreeItems := make(map[string]int64)
idcRackStgs := make(map[string]map[string]*rackAllocator)
idcNodeStgs := make(map[string][]*nodeAllocator)
rackNodeStgs := make(map[string][]*nodeAllocator)
rackFreeItems := make(map[string]int64)
var (
free, size, diskFreeItem, diskMaxItem int64
idc, rack, host string
)
for _, disk := range disks {
// read one disk info
err := disk.withRLocked(func() error {
idc = disk.info.Idc
rack = disk.info.Rack
host = disk.info.Host
if node, ok := d.getNode(disk.info.NodeID); ok {
idc = node.info.Idc
rack = node.info.Rack
host = node.info.Host
}
blobNodeHeartbeatInfo, isBlobNodeDisk := disk.info.extraInfo.(*clustermgr.DiskHeartBeatInfo)
if isBlobNodeDisk {
free = blobNodeHeartbeatInfo.Free
size = blobNodeHeartbeatInfo.Size
diskFreeItem = blobNodeHeartbeatInfo.FreeChunkCnt
diskMaxItem = blobNodeHeartbeatInfo.MaxChunkCnt
diskStatInfosM[idc].TotalFreeChunk += diskFreeItem
diskStatInfosM[idc].TotalChunk += diskMaxItem
}
shardNodeHeartbeatInfo, isShardNodeDisk := disk.info.extraInfo.(*clustermgr.ShardNodeDiskHeartbeatInfo)
if isShardNodeDisk {
free = shardNodeHeartbeatInfo.Free
size = shardNodeHeartbeatInfo.Size
diskFreeItem = int64(shardNodeHeartbeatInfo.FreeShardCnt)
diskMaxItem = int64(shardNodeHeartbeatInfo.MaxShardCnt)
diskStatInfosM[idc].TotalFreeShard += diskFreeItem
diskStatInfosM[idc].TotalShard += diskMaxItem
}
readonly := disk.info.Readonly
status := disk.info.Status
// rack can be the same in different idc, so we make rack string with idc
rack = idc + "-" + rack
spaceStatInfo.TotalDisk += 1
// idc disk status num calculate
if diskStatInfosM[idc] == nil {
diskStatInfosM[idc] = &clustermgr.DiskStatInfo{IDC: idc}
}
diskStatInfosM[idc].Total += 1
if readonly {
diskStatInfosM[idc].Readonly += 1
}
switch status {
case proto.DiskStatusBroken:
diskStatInfosM[idc].Broken += 1
case proto.DiskStatusRepairing:
diskStatInfosM[idc].Repairing += 1
case proto.DiskStatusRepaired:
diskStatInfosM[idc].Repaired += 1
case proto.DiskStatusDropped:
diskStatInfosM[idc].Dropped += 1
default:
}
if disk.dropping {
diskStatInfosM[idc].Dropping += 1
}
// filter abnormal disk
if disk.info.Status != proto.DiskStatusNormal {
return errors.New("abnormal disk")
}
spaceStatInfo.TotalSpace += size
if readonly { // include dropping disk
spaceStatInfo.ReadOnlySpace += free
return errors.New("readonly disk")
}
spaceStatInfo.FreeSpace += free
diskStatInfosM[idc].Available += 1
// filter expired disk
if disk.isExpire() {
diskStatInfosM[idc].Expired += 1
return errors.New("expired disk")
}
return nil
})
if err != nil {
span.Infof("This is %v, not to build allocator", err)
continue
}
// build for idcRackStorage
if _, ok := idcRackStgs[idc]; !ok {
idcRackStgs[idc] = make(map[string]*rackAllocator)
}
if _, ok := idcRackStgs[idc][rack]; !ok {
idcRackStgs[idc][rack] = &rackAllocator{rack: rack}
}
// build for idcAllocator
if _, ok := idcNodeStgs[idc]; !ok {
idcNodeStgs[idc] = make([]*nodeAllocator, 0)
idcFreeItems[idc] = 0
}
idcFreeItems[idc] += diskFreeItem
// build for rackAllocator
if _, ok := rackNodeStgs[rack]; !ok {
rackNodeStgs[rack] = make([]*nodeAllocator, 0)
rackFreeItems[rack] = 0
}
rackFreeItems[rack] += diskFreeItem
// build for nodeAllocator
if _, ok := nodeStgs[host]; !ok {
nodeStgs[host] = &nodeAllocator{host: host, disks: make([]*diskItem, 0)}
// append idc data node
idcNodeStgs[idc] = append(idcNodeStgs[idc], nodeStgs[host])
// append rack data node
rackNodeStgs[rack] = append(rackNodeStgs[rack], nodeStgs[host])
}
nodeStgs[host].disks = append(nodeStgs[host].disks, disk)
nodeStgs[host].weight += diskFreeItem
nodeStgs[host].free += free
}
span.Debugf("all nodeStgs: %+v", nodeStgs)
for _, rackStgs := range idcRackStgs {
for rack := range rackStgs {
rackStgs[rack].weight = rackFreeItems[rack]
rackStgs[rack].nodeStorages = rackNodeStgs[rack]
}
}
for idc := range idcNodeStgs {
span.Infof("%s idcNodeStgs length: %d", idc, len(idcNodeStgs[idc]))
}
spaceStatInfo.UsedSpace = spaceStatInfo.TotalSpace - spaceStatInfo.FreeSpace - spaceStatInfo.ReadOnlySpace
if len(idcRackStgs) > 0 {
ret = make(map[string]*idcAllocator)
for i := range d.cfg.IDC {
ret[d.cfg.IDC[i]] = &idcAllocator{
idc: d.cfg.IDC[i],
weight: idcFreeItems[d.cfg.IDC[i]],
diffRack: d.cfg.RackAware,
diffHost: d.cfg.HostAware,
rackStorages: idcRackStgs[d.cfg.IDC[i]],
nodeStorages: idcNodeStgs[d.cfg.IDC[i]],
}
freeChunk += idcFreeItems[d.cfg.IDC[i]]
}
spaceStatInfo.WritableSpace += d.calculateWritable(idcNodeStgs)
}
return
}
func (d *manager) calculateWritable(nodeStgs map[string][]*nodeAllocator) int64 {
// writable space statistic
codeMode, suCount := d.getMaxSuCount()
idcSuCount := suCount / len(d.cfg.IDC)
if d.cfg.HostAware && len(nodeStgs) > 0 {
// calculate minimum idc writable chunk num
calIDCWritableFunc := func(stgs []*nodeAllocator) int64 {
stripe := make([]int64, idcSuCount)
lefts := make(maxHeap, 0)
n := int64(0)
for _, v := range stgs {
count := v.free / d.cfg.ChunkSize
if count > 0 {
lefts = append(lefts, count)
}
}
heap.Init(&lefts)
for {
if lefts.Len() < idcSuCount {
break
}
for i := 0; i < idcSuCount; i++ {
stripe[i] = heap.Pop(&lefts).(int64)
}
// set minimum stripe count to 10 with more random selection, optimize writable space accuracy
min := int64(10)
n += min
for i := 0; i < idcSuCount; i++ {
stripe[i] -= min
if stripe[i] > 0 {
heap.Push(&lefts, stripe[i])
}
}
}
return n
}
minimumStripeCount := int64(math.MaxInt64)
for idc := range nodeStgs {
n := calIDCWritableFunc(nodeStgs[idc])
if n < minimumStripeCount {
minimumStripeCount = n
}
}
return minimumStripeCount * int64(codeMode.Tactic().N) * d.cfg.ChunkSize
}
if len(nodeStgs) > 0 {
minimumChunkNum := int64(math.MaxInt64)
for idc := range nodeStgs {
idcChunkNum := int64(0)
for i := range nodeStgs[idc] {
idcChunkNum += nodeStgs[idc][i].free / d.cfg.ChunkSize
}
if idcChunkNum < minimumChunkNum {
minimumChunkNum = idcChunkNum
}
}
return minimumChunkNum / int64(idcSuCount) * int64(codeMode.Tactic().N) * d.cfg.ChunkSize
}
return 0
}
func (d *manager) getMaxSuCount() (codemode.CodeMode, int) {
suCount := 0
idx := 0
for i := range d.cfg.CodeModes {
codeModeInfo := d.cfg.CodeModes[i].Tactic()
if codeModeInfo.N+codeModeInfo.M+codeModeInfo.L > suCount {
idx = i
suCount = codeModeInfo.N + codeModeInfo.M + codeModeInfo.L
}
}
return d.cfg.CodeModes[idx], suCount
}
func fmtApplyContextKey(opType, id string) string {
return fmt.Sprintf("%s-%s", opType, id)
}

2
blobstore/clustermgr/cluster/metric.go
View File

@ -62,7 +62,7 @@ func init() {
func (d *manager) Report(ctx context.Context, region string, clusterID proto.ClusterID, isLeader string) {
vec := spaceStatInfoMetric
vec.Reset()
spaceStatInfo := d.Stat(ctx)
spaceStatInfo := d.Stat(ctx, proto.DiskTypeHDD)
reflectTyes := reflect.TypeOf(*spaceStatInfo)
reflectVals := reflect.ValueOf(*spaceStatInfo)
for i := 0; i < reflectTyes.NumField(); i++ {

2
blobstore/clustermgr/cluster/proto.go
View File

@ -10,7 +10,7 @@ import (
type nodeItemInfo struct {
clustermgr.NodeInfo
// extraInfo interface{}
extraInfo interface{}
}
type nodeItem struct {

439
blobstore/clustermgr/cluster/refresh.go
View File

@ -1,439 +0,0 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implieb. See the License for the specific language governing
// permissions and limitations under the License.
package cluster
import (
"container/heap"
"context"
"encoding/json"
"math"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
"github.com/cubefs/cubefs/blobstore/common/codemode"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
// maxHeap are use to build max heap, and we can calculate writable space that satisfied with stripe count
type maxHeap []int64
func (h maxHeap) Len() int { return len(h) }
func (h maxHeap) Less(i, j int) bool { return h[i] > h[j] }
func (h maxHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *maxHeap) Push(x interface{}) {
*h = append(*h, x.(int64))
}
func (h *maxHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// refresh use for refreshing storage allocator info and cluster statistic info
func (b *BlobNodeManager) refresh(ctx context.Context) {
span := trace.SpanFromContextSafe(ctx)
// space and disk stat info
spaceStatInfos := make(map[proto.DiskType]*clustermgr.SpaceStatInfo)
// generate diskType -> nodeSet -> diskSet -> idc -> rack -> blobnode storage and statInfo
nodeSetAllocators := make(map[proto.DiskType]nodeSetAllocatorMap)
diskSetAllocators := make(map[proto.DiskType]diskSetAllocatorMap)
ecDiskSet := make(map[proto.DiskType][]*diskItem)
nodeSetsMap := b.topoMgr.GetAllNodeSets(ctx)
for diskType, nodeSets := range nodeSetsMap {
if _, ok := nodeSetAllocators[diskType]; !ok {
nodeSetAllocators[diskType] = make(nodeSetAllocatorMap)
}
if _, ok := diskSetAllocators[diskType]; !ok {
diskSetAllocators[diskType] = make(diskSetAllocatorMap)
}
if _, ok := spaceStatInfos[diskType]; !ok {
spaceStatInfos[diskType] = &clustermgr.SpaceStatInfo{}
}
spaceStatInfo := spaceStatInfos[diskType]
diskStatInfo := make(map[string]*clustermgr.DiskStatInfo)
for i := range b.cfg.IDC {
diskStatInfo[b.cfg.IDC[i]] = &clustermgr.DiskStatInfo{IDC: b.cfg.IDC[i]}
}
for _, nodeSet := range nodeSets {
nodeSetAllocator := newNodeSetAllocator(nodeSet.ID())
for _, diskSet := range nodeSet.GetDiskSets() {
disks := diskSet.GetDisks()
// ecDiskSet[diskType] = append(ecDiskSet[diskType], disks...)
idcAllocators, diskSetFreeChunk := b.generateDiskSetStorage(ctx, disks, spaceStatInfo, diskStatInfo)
diskSetAllocator := newDiskSetAllocator(diskSet.ID(), diskSetFreeChunk, idcAllocators)
diskSetAllocators[diskType][diskSet.ID()] = diskSetAllocator
nodeSetAllocator.addDiskSet(diskSetAllocator)
}
nodeSetAllocators[diskType][nodeSet.ID()] = nodeSetAllocator
}
for idc := range diskStatInfo {
spaceStatInfo.DisksStatInfos = append(spaceStatInfo.DisksStatInfos, *diskStatInfo[idc])
}
spaceStatInfo.TotalBlobNode = int64(b.topoMgr.GetNodeNum(diskType))
}
// compatible
allDisks := b.getAllDisk()
span.Debugf("get all disks, len:%d", len(allDisks))
diskTypeDisks := make(map[proto.DiskType][]*diskItem)
for _, disk := range allDisks {
diskType := b.getDiskType(disk)
diskTypeDisks[diskType] = append(diskTypeDisks[diskType], disk)
}
for diskType := range diskTypeDisks {
if _, ok := nodeSetAllocators[diskType]; !ok {
nodeSetAllocators[diskType] = make(nodeSetAllocatorMap)
}
if _, ok := diskSetAllocators[diskType]; !ok {
diskSetAllocators[diskType] = make(diskSetAllocatorMap)
}
if _, ok := spaceStatInfos[diskType]; !ok {
spaceStatInfos[diskType] = &clustermgr.SpaceStatInfo{}
}
ecDiskSet[diskType] = diskTypeDisks[diskType]
ecSpaceStateInfo := &clustermgr.SpaceStatInfo{}
diskStatInfo := make(map[string]*clustermgr.DiskStatInfo)
for i := range b.cfg.IDC {
diskStatInfo[b.cfg.IDC[i]] = &clustermgr.DiskStatInfo{IDC: b.cfg.IDC[i]}
}
ecIdcAllocators, ecFreeChunk := b.generateDiskSetStorage(ctx, ecDiskSet[diskType], ecSpaceStateInfo, diskStatInfo)
// initial ec allocator
diskSetAllocator := newDiskSetAllocator(ecDiskSetID, ecFreeChunk, ecIdcAllocators)
diskSetAllocators[diskType][ecDiskSetID] = diskSetAllocator
nodeSetAllocator := newNodeSetAllocator(ecNodeSetID)
nodeSetAllocator.addDiskSet(diskSetAllocator)
nodeSetAllocators[diskType][ecNodeSetID] = nodeSetAllocator
span.Debugf("add ec nodeset")
// update space state info
for idc := range diskStatInfo {
ecSpaceStateInfo.DisksStatInfos = append(ecSpaceStateInfo.DisksStatInfos, *diskStatInfo[idc])
}
// set blobnode space info and disk stat info by ec statistic
// TODO: calculate writable space by replicate code mode and ec code mode ratio
spaceStatInfos[diskType] = ecSpaceStateInfo
}
b.allocator.Store(newAllocator(allocatorConfig{
nodeSets: nodeSetAllocators,
diskSets: diskSetAllocators,
dg: b,
tg: b.topoMgr,
diffHost: b.cfg.HostAware,
diffRack: b.cfg.RackAware,
}))
b.spaceStatInfo.Store(spaceStatInfos)
}
func (b *BlobNodeManager) generateDiskSetStorage(ctx context.Context, disks []*diskItem, spaceStatInfo *clustermgr.SpaceStatInfo,
diskStatInfosM map[string]*clustermgr.DiskStatInfo,
) (ret map[string]*idcAllocator, freeChunk int64) {
span := trace.SpanFromContextSafe(ctx)
blobNodeStgs := make(map[string]*nodeAllocator)
idcFreeChunks := make(map[string]int64)
idcRackStgs := make(map[string]map[string]*rackAllocator)
idcBlobNodeStgs := make(map[string][]*nodeAllocator)
rackBlobNodeStgs := make(map[string][]*nodeAllocator)
rackFreeChunks := make(map[string]int64)
var (
free, diskFreeChunk int64
idc, rack, host string
)
for _, disk := range disks {
// read one disk info
err := disk.withRLocked(func() error {
idc = disk.info.Idc
rack = disk.info.Rack
host = disk.info.Host
if node, ok := b.getNode(disk.info.NodeID); ok {
idc = node.info.Idc
rack = node.info.Rack
host = node.info.Host
}
heartbeatInfo := disk.info.extraInfo.(*clustermgr.DiskHeartBeatInfo)
diskFreeChunk = heartbeatInfo.FreeChunkCnt
maxChunk := heartbeatInfo.MaxChunkCnt
readonly := disk.info.Readonly
size := heartbeatInfo.Size
free = heartbeatInfo.Free
status := disk.info.Status
// rack can be the same in different idc, so we make rack string with idc
rack = idc + "-" + rack
spaceStatInfo.TotalDisk += 1
// idc disk status num calculate
if diskStatInfosM[idc] == nil {
diskStatInfosM[idc] = &clustermgr.DiskStatInfo{IDC: idc}
}
diskStatInfosM[idc].Total += 1
diskStatInfosM[idc].TotalChunk += maxChunk
diskStatInfosM[idc].TotalFreeChunk += diskFreeChunk
if readonly {
diskStatInfosM[idc].Readonly += 1
}
switch status {
case proto.DiskStatusBroken:
diskStatInfosM[idc].Broken += 1
case proto.DiskStatusRepairing:
diskStatInfosM[idc].Repairing += 1
case proto.DiskStatusRepaired:
diskStatInfosM[idc].Repaired += 1
case proto.DiskStatusDropped:
diskStatInfosM[idc].Dropped += 1
default:
}
if disk.dropping {
diskStatInfosM[idc].Dropping += 1
}
// filter abnormal disk
if disk.info.Status != proto.DiskStatusNormal {
return errors.New("abnormal disk")
}
spaceStatInfo.TotalSpace += size
if readonly { // include dropping disk
spaceStatInfo.ReadOnlySpace += free
return errors.New("readonly disk")
}
spaceStatInfo.FreeSpace += free
diskStatInfosM[idc].Available += 1
// filter expired disk
if disk.isExpire() {
diskStatInfosM[idc].Expired += 1
return errors.New("expired disk")
}
return nil
})
if err != nil {
span.Infof("This is %v, not to build allocator", err)
continue
}
// build for idcRackStorage
if _, ok := idcRackStgs[idc]; !ok {
idcRackStgs[idc] = make(map[string]*rackAllocator)
}
if _, ok := idcRackStgs[idc][rack]; !ok {
idcRackStgs[idc][rack] = &rackAllocator{rack: rack}
}
// build for idcAllocator
if _, ok := idcBlobNodeStgs[idc]; !ok {
idcBlobNodeStgs[idc] = make([]*nodeAllocator, 0)
idcFreeChunks[idc] = 0
}
idcFreeChunks[idc] += diskFreeChunk
// build for rackAllocator
if _, ok := rackBlobNodeStgs[rack]; !ok {
rackBlobNodeStgs[rack] = make([]*nodeAllocator, 0)
rackFreeChunks[rack] = 0
}
rackFreeChunks[rack] += diskFreeChunk
// build for nodeAllocator
if _, ok := blobNodeStgs[host]; !ok {
blobNodeStgs[host] = &nodeAllocator{host: host, disks: make([]*diskItem, 0)}
// append idc data node
idcBlobNodeStgs[idc] = append(idcBlobNodeStgs[idc], blobNodeStgs[host])
// append rack data node
rackBlobNodeStgs[rack] = append(rackBlobNodeStgs[rack], blobNodeStgs[host])
}
blobNodeStgs[host].disks = append(blobNodeStgs[host].disks, disk)
blobNodeStgs[host].weight += diskFreeChunk
blobNodeStgs[host].free += free
}
span.Debugf("all blobNodeStgs: %+v", blobNodeStgs)
for _, rackStgs := range idcRackStgs {
for rack := range rackStgs {
rackStgs[rack].weight = rackFreeChunks[rack]
rackStgs[rack].blobNodeStorages = rackBlobNodeStgs[rack]
}
}
for idc := range idcBlobNodeStgs {
span.Infof("%s idcBlobNodeStgs length: %d", idc, len(idcBlobNodeStgs[idc]))
}
spaceStatInfo.UsedSpace = spaceStatInfo.TotalSpace - spaceStatInfo.FreeSpace - spaceStatInfo.ReadOnlySpace
if len(idcRackStgs) > 0 {
ret = make(map[string]*idcAllocator)
for i := range b.cfg.IDC {
spaceStatInfo.TotalBlobNode += int64(len(idcBlobNodeStgs[b.cfg.IDC[i]]))
ret[b.cfg.IDC[i]] = &idcAllocator{
idc: b.cfg.IDC[i],
weight: idcFreeChunks[b.cfg.IDC[i]],
diffRack: b.cfg.RackAware,
diffHost: b.cfg.HostAware,
rackStorages: idcRackStgs[b.cfg.IDC[i]],
blobNodeStorages: idcBlobNodeStgs[b.cfg.IDC[i]],
}
freeChunk += idcFreeChunks[b.cfg.IDC[i]]
}
spaceStatInfo.WritableSpace += b.calculateWritable(idcBlobNodeStgs)
}
return
}
func (b *BlobNodeManager) calculateWritable(idcBlobNodeStgs map[string][]*nodeAllocator) int64 {
// writable space statistic
codeMode, suCount := b.getMaxSuCount()
idcSuCount := suCount / len(b.cfg.IDC)
if b.cfg.HostAware && len(idcBlobNodeStgs) > 0 {
// calculate minimum idc writable chunk num
calIDCWritableFunc := func(stgs []*nodeAllocator) int64 {
stripe := make([]int64, idcSuCount)
lefts := make(maxHeap, 0)
n := int64(0)
for _, v := range stgs {
count := v.free / b.cfg.ChunkSize
if count > 0 {
lefts = append(lefts, count)
}
}
heap.Init(&lefts)
for {
if lefts.Len() < idcSuCount {
break
}
for i := 0; i < idcSuCount; i++ {
stripe[i] = heap.Pop(&lefts).(int64)
}
// set minimum stripe count to 10 with more random selection, optimize writable space accuracy
min := int64(10)
n += min
for i := 0; i < idcSuCount; i++ {
stripe[i] -= min
if stripe[i] > 0 {
heap.Push(&lefts, stripe[i])
}
}
}
return n
}
minimumStripeCount := int64(math.MaxInt64)
for idc := range idcBlobNodeStgs {
n := calIDCWritableFunc(idcBlobNodeStgs[idc])
if n < minimumStripeCount {
minimumStripeCount = n
}
}
return minimumStripeCount * int64(codeMode.Tactic().N) * b.cfg.ChunkSize
}
if len(idcBlobNodeStgs) > 0 {
minimumChunkNum := int64(math.MaxInt64)
for idc := range idcBlobNodeStgs {
idcChunkNum := int64(0)
for i := range idcBlobNodeStgs[idc] {
idcChunkNum += idcBlobNodeStgs[idc][i].free / b.cfg.ChunkSize
}
if idcChunkNum < minimumChunkNum {
minimumChunkNum = idcChunkNum
}
}
return minimumChunkNum / int64(idcSuCount) * int64(codeMode.Tactic().N) * b.cfg.ChunkSize
}
return 0
}
func (b *BlobNodeManager) getMaxSuCount() (codemode.CodeMode, int) {
suCount := 0
idx := 0
for i := range b.cfg.CodeModes {
codeModeInfo := b.cfg.CodeModes[i].Tactic()
if codeModeInfo.N+codeModeInfo.M+codeModeInfo.L > suCount {
idx = i
suCount = codeModeInfo.N + codeModeInfo.M + codeModeInfo.L
}
}
return b.cfg.CodeModes[idx], suCount
}
func (b *BlobNodeManager) checkDroppingNode(ctx context.Context) {
if !b.raftServer.IsLeader() {
return
}
span := trace.SpanFromContextSafe(ctx)
droppingNodeDBs, err := b.nodeTbl.GetAllDroppingNode()
if err != nil {
span.Warnf("get dropping nodes failed:%v", err)
return
}
var diskItems []*diskItem
for _, nodeID := range droppingNodeDBs {
node, _ := b.getNode(nodeID)
node.withRLocked(func() error {
// copy diskIDs of node, avoid nested node and disk lock
diskItems = make([]*diskItem, 0, len(node.disks))
for _, di := range node.disks {
diskItems = append(diskItems, di)
}
return nil
})
// check disk status
for _, di := range diskItems {
err = di.withRLocked(func() error {
if di.needFilter() {
return errors.New("node has disk in use")
}
return nil
})
if err != nil {
break
}
}
if err != nil {
span.Debugf("checkDroppingNode node: %d has disk in use", node.nodeID)
continue
}
args := &clustermgr.NodeInfoArgs{NodeID: nodeID}
data, err := json.Marshal(args)
if err != nil {
span.Errorf("checkDroppingNode json marshal failed, args: %v, error: %v", args, err)
return
}
proposeInfo := base.EncodeProposeInfo(b.GetModuleName(), OperTypeDroppedNode, data, base.ProposeContext{ReqID: span.TraceID()})
err = b.raftServer.Propose(ctx, proposeInfo)
if err != nil {
span.Errorf("checkDroppingNode dropped node: %d failed: %v", node.nodeID, err)
return
}
span.Debugf("checkDroppingNode dropped node: %d success", node.nodeID)
}
}

415
blobstore/clustermgr/cluster/shardnode_mgr.go
View File

@ -6,31 +6,34 @@ import (
"fmt"
"math/rand"
"sync"
"sync/atomic"
"time"
"github.com/cubefs/cubefs/blobstore/util/defaulter"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/api/shardnode"
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
"github.com/cubefs/cubefs/blobstore/clustermgr/persistence/normaldb"
"github.com/cubefs/cubefs/blobstore/clustermgr/scopemgr"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/errors"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
apierrors "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/sharding"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/defaulter"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
const (
shardNodeDiskIDScopeName = "sn-diskid"
shardNodeIDScopeName = "sn-nodeid"
ShardNodeDiskIDScopeName = "sn-diskid"
ShardNodeIDScopeName = "sn-nodeid"
)
type ShardNodeManagerAPI interface {
// GetNodeInfo return node info with specified node id, it return ErrCMNodeNotFound if node not found
// GetNodeInfo return node info with specified node id, it returns ErrCMNodeNotFound if node not found
GetNodeInfo(ctx context.Context, nodeID proto.NodeID) (*clustermgr.ShardNodeInfo, error)
// GetDiskInfo return disk info, it return ErrDiskNotFound if disk not found
GetDiskInfo(ctx context.Context, id proto.DiskID) (*clustermgr.ShardNodeDiskInfo, error)
// AddDisk add shardNode disk to CM
AddDisk(ctx context.Context, args *clustermgr.ShardNodeDiskInfo) error
// ListDroppingDisk return all dropping disk info
ListDroppingDisk(ctx context.Context) ([]*clustermgr.ShardNodeDiskInfo, error)
// ListDiskInfo return disk list with list option
@ -43,10 +46,10 @@ type ShardNodeManagerAPI interface {
}
func NewShardNodeMgr(scopeMgr scopemgr.ScopeMgrAPI, db *normaldb.NormalDB, cfg DiskMgrConfig) (*ShardNodeManager, error) {
_, ctx := trace.StartSpanFromContext(context.Background(), "NewDiskMgr")
_, ctx := trace.StartSpanFromContext(context.Background(), "NewShardNodeMgr")
cfg.NodeIDScopeName = shardNodeIDScopeName
cfg.DiskIDScopeName = shardNodeDiskIDScopeName
cfg.NodeIDScopeName = ShardNodeIDScopeName
cfg.DiskIDScopeName = ShardNodeDiskIDScopeName
defaulter.LessOrEqual(&cfg.RefreshIntervalS, defaultRefreshIntervalS)
defaulter.LessOrEqual(&cfg.HeartbeatExpireIntervalS, defaultHeartbeatExpireIntervalS)
defaulter.LessOrEqual(&cfg.FlushIntervalS, defaultFlushIntervalS)
@ -55,8 +58,8 @@ func NewShardNodeMgr(scopeMgr scopemgr.ScopeMgrAPI, db *normaldb.NormalDB, cfg D
defaultAllocTolerateBuff = cfg.AllocTolerateBuffer
}
if len(cfg.CodeModes) == 0 {
return nil, errors.New("code mode can not be nil")
if len(cfg.CodeModes) != 1 {
return nil, errors.New("shardnode code mode length must be 1")
}
if len(cfg.IDC) == 0 {
return nil, errors.New("idc can not be nil")
@ -72,7 +75,7 @@ func NewShardNodeMgr(scopeMgr scopemgr.ScopeMgrAPI, db *normaldb.NormalDB, cfg D
return nil, errors.Info(err, "open node table failed").Detail(err)
}
bm := &ShardNodeManager{
sm := &ShardNodeManager{
diskTbl: diskTbl,
nodeTbl: nodeTbl,
}
@ -81,21 +84,21 @@ func NewShardNodeMgr(scopeMgr scopemgr.ScopeMgrAPI, db *normaldb.NormalDB, cfg D
topoMgr: newTopoMgr(),
taskPool: base.NewTaskDistribution(int(cfg.ApplyConcurrency), 1),
scopeMgr: scopeMgr,
persistentHandler: bm,
persistentHandler: sm,
closeCh: make(chan interface{}),
cfg: cfg,
}
bm.manager = m
sm.manager = m
// initial load data
err = bm.LoadData(ctx)
err = sm.LoadData(ctx)
if err != nil {
return nil, err
}
_, ctxNew := trace.StartSpanFromContext(context.Background(), "")
bm.refresh(ctxNew)
sm.refresh(ctxNew)
ticker := time.NewTicker(time.Duration(cfg.RefreshIntervalS) * time.Second)
go func() {
@ -103,17 +106,28 @@ func NewShardNodeMgr(scopeMgr scopemgr.ScopeMgrAPI, db *normaldb.NormalDB, cfg D
for {
select {
case <-ticker.C:
bm.refresh(ctxNew)
case <-bm.closeCh:
sm.refresh(ctxNew)
case <-sm.closeCh:
return
}
}
}()
return bm, nil
return sm, nil
}
type AllocShardsPolicy struct{}
type AllocShardsPolicy struct {
DiskType proto.DiskType
Suids []proto.Suid
Range sharding.Range
RouteVersion proto.RouteVersion
ExcludeDiskSets []proto.DiskSetID
RepairUnits []clustermgr.ShardUnit
ExcludeDisks []proto.DiskID
DiskSetID proto.DiskSetID
Idc string
}
type ShardNodeManager struct {
*manager
@ -123,7 +137,19 @@ type ShardNodeManager struct {
}
func (s *ShardNodeManager) GetDiskInfo(ctx context.Context, id proto.DiskID) (*clustermgr.ShardNodeDiskInfo, error) {
return nil, nil
disk, ok := s.getDisk(id)
if !ok {
return nil, apierrors.ErrCMDiskNotFound
}
var diskInfo clustermgr.ShardNodeDiskInfo
// need to copy before return, or the higher level may change the disk info by the pointer
disk.withRLocked(func() error {
diskInfo.DiskInfo = disk.info.DiskInfo
diskInfo.ShardNodeDiskHeartbeatInfo = *(disk.info.extraInfo.(*clustermgr.ShardNodeDiskHeartbeatInfo))
return nil
})
return &(diskInfo), nil
}
func (s *ShardNodeManager) ListDroppingDisk(ctx context.Context) ([]*clustermgr.ShardNodeDiskInfo, error) {
@ -132,19 +158,226 @@ func (s *ShardNodeManager) ListDroppingDisk(ctx context.Context) ([]*clustermgr.
// ListDiskInfo return disk info with specified query condition
func (s *ShardNodeManager) ListDiskInfo(ctx context.Context, opt *clustermgr.ListOptionArgs) (disks []*clustermgr.ShardNodeDiskInfo, marker proto.DiskID, err error) {
return
if opt == nil {
return nil, 0, nil
}
span := trace.SpanFromContextSafe(ctx)
if opt.Count > defaultListDiskMaxCount {
opt.Count = defaultListDiskMaxCount
}
diskInfoDBs, err := s.diskTbl.ListDisk(opt)
if err != nil {
span.Error("shardNodeManager ListDiskInfo failed, err: %v", err)
return nil, 0, errors.Info(err, "shardNodeManager ListDiskInfo failed").Detail(err)
}
if len(diskInfoDBs) > 0 {
marker = diskInfoDBs[len(diskInfoDBs)-1].DiskID
}
for i := range diskInfoDBs {
diskInfo := s.diskInfoRecordToDiskInfo(diskInfoDBs[i])
disk, _ := s.getDisk(diskInfo.DiskID)
disk.withRLocked(func() error {
heartbeatInfo := disk.info.extraInfo.(*clustermgr.ShardNodeDiskHeartbeatInfo)
diskInfo.FreeShardCnt = heartbeatInfo.FreeShardCnt
diskInfo.UsedShardCnt = heartbeatInfo.UsedShardCnt
diskInfo.MaxShardCnt = heartbeatInfo.MaxShardCnt
diskInfo.Free = heartbeatInfo.Free
diskInfo.Used = heartbeatInfo.Used
diskInfo.Size = heartbeatInfo.Size
return nil
})
disks = append(disks, diskInfo)
}
if len(disks) == 0 {
marker = proto.InvalidDiskID
}
return disks, marker, nil
}
func (s *ShardNodeManager) AddDisk(ctx context.Context, args *clustermgr.ShardNodeDiskInfo) error {
span := trace.SpanFromContextSafe(ctx)
node, ok := s.getNode(args.NodeID)
if !ok {
span.Warnf("node not exist, disk info: %v", args)
return apierrors.ErrCMNodeNotFound
}
err := node.withRLocked(func() error {
if node.info.Status == proto.NodeStatusDropped {
span.Warnf("node is dropped, disk info: %v", args)
return apierrors.ErrCMNodeNotFound
}
if node.dropping {
span.Warnf("node is dropping, disk info: %v", args)
return apierrors.ErrCMNodeIsDropping
}
if err := s.CheckDiskInfoDuplicated(ctx, args.DiskID, &args.DiskInfo, &node.info.NodeInfo); err != nil {
return err
}
// disk idc/rack/host uses node one
args.Idc = node.info.Idc
args.Rack = node.info.Rack
args.Host = node.info.Host
return nil
})
if err != nil {
return err
}
data, err := json.Marshal(args)
if err != nil {
span.Errorf("json marshal failed, disk info: %v, error: %v", args, err)
return errors.Info(apierrors.ErrUnexpected).Detail(err)
}
pendingKey := fmtApplyContextKey("disk-add", args.DiskID.ToString())
s.pendingEntries.Store(pendingKey, nil)
defer s.pendingEntries.Delete(pendingKey)
proposeInfo := base.EncodeProposeInfo(s.GetModuleName(), OperTypeAddDisk, data, base.ProposeContext{ReqID: span.TraceID()})
err = s.raftServer.Propose(ctx, proposeInfo)
if err != nil {
span.Error(err)
return apierrors.ErrRaftPropose
}
if v, _ := s.manager.pendingEntries.Load(pendingKey); v != nil {
return v.(error)
}
return nil
}
func (s *ShardNodeManager) GetNodeInfo(ctx context.Context, nodeID proto.NodeID) (*clustermgr.ShardNodeInfo, error) {
return nil, nil
node, ok := s.getNode(nodeID)
if !ok {
return nil, apierrors.ErrCMNodeNotFound
}
// need to copy before return, or the higher level may change the node info by pointer
nodeInfo := &clustermgr.ShardNodeInfo{}
node.withRLocked(func() error {
nodeInfo.NodeInfo = node.info.NodeInfo
nodeInfo.ShardNodeExtraInfo = node.info.extraInfo.(clustermgr.ShardNodeExtraInfo)
return nil
})
return nodeInfo, nil
}
func (s *ShardNodeManager) AllocShards(ctx context.Context, policy AllocShardsPolicy) ([]proto.DiskID, error) {
return nil, nil
// AllocShards not retry when alloc one shard failed, so the caller can implement the retry logic as needed.
func (s *ShardNodeManager) AllocShards(ctx context.Context, policy AllocShardsPolicy) ([]proto.DiskID, proto.DiskSetID, error) {
span, ctx := trace.StartSpanFromContextWithTraceID(ctx, "AllocShards", trace.SpanFromContextSafe(ctx).TraceID())
var (
err error
addShardLock sync.Mutex
excludesDiskSetID proto.DiskSetID
allocator = s.allocator.Load().(*allocator)
// to make sure return disks order match with policy.Suids
suidIndexMap = make(map[proto.Suid]int)
suidDiskMap = make(map[proto.Suid]proto.DiskID, len(policy.Suids))
units = make([]clustermgr.ShardUnit, s.cfg.CodeModes[0].GetShardNum())
retDiskIDs = make([]proto.DiskID, len(policy.Suids))
)
// repair shard case
if len(policy.ExcludeDisks) > 0 {
ret, err := allocator.ReAlloc(ctx, reAllocPolicy{
diskType: policy.DiskType,
diskSetID: policy.DiskSetID,
idc: policy.Idc,
count: len(policy.Suids),
excludes: policy.ExcludeDisks,
})
if err != nil {
return nil, nullDiskSetID, err
}
units = policy.RepairUnits
for idx, diskID := range ret {
suid := policy.Suids[idx]
suidIndexMap[suid] = idx
suidDiskMap[suid] = ret[idx]
units = append(units, clustermgr.ShardUnit{
Suid: suid,
DiskID: diskID,
Learner: true,
})
}
} else {
// create shard case
tactic := s.cfg.CodeModes[0].Tactic()
idcIndexes := tactic.GetECLayoutByAZ()
rand.Seed(time.Now().UnixNano())
rand.Shuffle(len(idcIndexes), func(i, j int) {
idcIndexes[i], idcIndexes[j] = idcIndexes[j], idcIndexes[i]
})
span.Debugf("idcIndexes is %#v", idcIndexes)
// alloc disks in one diskSet
ret, err := allocator.Alloc(ctx, policy.DiskType, s.cfg.CodeModes[0], policy.ExcludeDiskSets)
if err != nil {
span.Errorf("create shard alloc disks failed, err: %s", err.Error())
return nil, nullDiskSetID, err
}
for idcIdx, r := range ret {
if err := s.validateAllocRet(r.Disks); err != nil {
return nil, nullDiskSetID, err
}
for diskIDIdx, suidIdx := range idcIndexes[idcIdx] {
suid := policy.Suids[suidIdx]
suidIndexMap[suid] = suidIdx
suidDiskMap[suid] = r.Disks[diskIDIdx]
units[suidIdx] = clustermgr.ShardUnit{
Suid: suid,
DiskID: r.Disks[diskIDIdx],
Learner: false,
}
}
}
}
wg := sync.WaitGroup{}
wg.Add(len(suidDiskMap))
for suid, diskID := range suidDiskMap {
disk, _ := s.getDisk(diskID)
disk.lock.RLock()
host := disk.info.Host
diskSetID := disk.info.DiskSetID
disk.lock.RUnlock()
go func(_suid proto.Suid, _diskID proto.DiskID) {
defer wg.Done()
addShardArgs := &shardnode.AddShardArgs{
DiskID: _diskID,
Suid: _suid,
Range: policy.Range,
Units: units,
RouteVersion: policy.RouteVersion,
}
shardNodeErr := s.shardNodeClient.AddShard(ctx, addShardArgs)
if shardNodeErr != nil {
atomic.StoreUint32((*uint32)(&excludesDiskSetID), uint32(diskSetID))
span.Errorf("alloc shard failed, diskID: %d, diskSetID: %d, host: %s, err: %v", _diskID, diskSetID, host, shardNodeErr)
return
}
addShardLock.Lock()
retDiskIDs[suidIndexMap[_suid]] = _diskID
addShardLock.Unlock()
}(suid, diskID)
}
wg.Wait()
if excludesDiskSetID != nullDiskSetID {
return nil, excludesDiskSetID, ErrShardNodeCreateShardFailed
}
// validate alloc ret again to avoid abnormal situations
if err := s.validateAllocRet(retDiskIDs); err != nil {
return nil, nullDiskSetID, err
}
return retDiskIDs, nullDiskSetID, err
}
func (s *ShardNodeManager) LoadData(ctx context.Context) error {
@ -375,24 +608,131 @@ func (s *ShardNodeManager) Apply(ctx context.Context, operTypes []int32, datas [
// heartBeatDiskInfo process disk's heartbeat
func (s *ShardNodeManager) applyHeartBeatDiskInfo(ctx context.Context, infos []clustermgr.ShardNodeDiskHeartbeatInfo) error {
span := trace.SpanFromContextSafe(ctx)
expireTime := time.Now().Add(time.Duration(s.cfg.HeartbeatExpireIntervalS) * time.Second)
for i := range infos {
info := infos[i]
disk, ok := s.getDisk(info.DiskID)
// sometimes, we may not find disk from allDisks
// it was happened when disk register and heartbeat request very close
if !ok {
span.Warnf("disk not found in all disk, diskID: %d", info.DiskID)
continue
}
// modify disk heartbeat memory info, dump into db timely
disk.withLocked(func() error {
heartbeatInfo := disk.info.extraInfo.(*clustermgr.ShardNodeDiskHeartbeatInfo)
heartbeatInfo.Free = info.Free
heartbeatInfo.Size = info.Size
heartbeatInfo.Used = info.Used
heartbeatInfo.UsedShardCnt = info.UsedShardCnt
// calculate free and max shard count
heartbeatInfo.MaxShardCnt = int32(info.Size / proto.MaxShardSize)
heartbeatInfo.FreeShardCnt = heartbeatInfo.MaxShardCnt - heartbeatInfo.UsedShardCnt
freeShardCnt := int32(info.Free / proto.MaxShardSize)
// use the minimum value as free shard count
if freeShardCnt < heartbeatInfo.FreeShardCnt {
span.Debugf("use minimum free shard count, disk id[%d], free shard[%d]", disk.diskID, freeShardCnt)
heartbeatInfo.FreeShardCnt = freeShardCnt
}
if heartbeatInfo.FreeShardCnt < 0 {
heartbeatInfo.FreeShardCnt = 0
}
disk.lastExpireTime = disk.expireTime
disk.expireTime = expireTime
return nil
})
}
return nil
}
// applyAddDisk add a new disk into cluster, it return ErrDiskExist if disk already exist
func (s *ShardNodeManager) applyAddDisk(ctx context.Context, info *clustermgr.ShardNodeDiskInfo) error {
return nil
}
span := trace.SpanFromContextSafe(ctx)
s.metaLock.Lock()
defer s.metaLock.Unlock()
// concurrent double check
_, ok := s.allDisks[info.DiskID]
if ok {
return nil
}
node, ok := s.allNodes[info.NodeID]
if !ok {
return ErrNodeNotExist
}
// return err by pendingEntries
err := node.withRLocked(func() error {
if node.info.Status == proto.NodeStatusDropped || node.dropping {
span.Warnf("node is dropped or dropping, disk info: %v", info)
pendingKey := fmtApplyContextKey("disk-add", info.DiskID.ToString())
if _, ok := s.pendingEntries.Load(pendingKey); ok {
s.pendingEntries.Store(pendingKey, apierrors.ErrCMNodeNotFound)
}
return apierrors.ErrCMNodeNotFound
}
return nil
})
if err != nil {
return nil
}
info.DiskSetID = s.topoMgr.AllocDiskSetID(ctx, &info.DiskInfo, &node.info.NodeInfo, s.cfg.CopySetConfigs[node.info.DiskType])
// calculate free and max chunk count
info.MaxShardCnt = int32(info.Size / proto.MaxShardSize)
info.FreeShardCnt = info.MaxShardCnt - info.UsedShardCnt
err = s.diskTbl.AddDisk(s.diskInfoToDiskInfoRecord(info))
if err != nil {
span.Error("ShardNodeManager.addDisk add disk failed: ", err)
return errors.Info(err, "ShardNodeManager.addDisk add disk failed").Detail(err)
}
disk := &diskItem{
diskID: info.DiskID,
info: diskItemInfo{DiskInfo: info.DiskInfo, extraInfo: &info.ShardNodeDiskHeartbeatInfo},
weightGetter: shardNodeDiskWeightGetter,
weightDecrease: shardNodeDiskWeightDecrease,
expireTime: time.Now().Add(time.Duration(s.cfg.HeartbeatExpireIntervalS) * time.Second),
}
node.withLocked(func() error {
node.disks[info.DiskID] = disk
return nil
})
s.topoMgr.AddDiskToDiskSet(node.info.DiskType, node.info.NodeSetID, disk)
s.allDisks[info.DiskID] = disk
s.hostPathFilter.Store(disk.genFilterKey(), 1)
// applyAddNode add a new node into cluster, it returns ErrNodeExist if node already exist
func (s *ShardNodeManager) applyAddNode(ctx context.Context, info *clustermgr.ShardNodeInfo) error {
return nil
}
func (s *ShardNodeManager) applyAdminUpdateDisk(ctx context.Context, diskInfo *clustermgr.ShardNodeDiskInfo) error {
return nil
}
span := trace.SpanFromContextSafe(ctx)
disk, ok := s.allDisks[diskInfo.DiskID]
if !ok {
span.Errorf("admin update shardnode disk, diskId:%d not exist", diskInfo.DiskID)
return ErrDiskNotExist
}
func (s *ShardNodeManager) refresh(ctx context.Context) {
return disk.withLocked(func() error {
if diskInfo.Status.IsValid() {
disk.info.Status = diskInfo.Status
}
heartbeatInfo := disk.info.extraInfo.(*clustermgr.ShardNodeDiskHeartbeatInfo)
if diskInfo.MaxShardCnt > 0 {
heartbeatInfo.MaxShardCnt = diskInfo.MaxShardCnt
}
if diskInfo.FreeShardCnt > 0 {
heartbeatInfo.FreeShardCnt = diskInfo.FreeShardCnt
}
diskRecord := s.diskInfoToDiskInfoRecord(&clustermgr.ShardNodeDiskInfo{DiskInfo: disk.info.DiskInfo, ShardNodeDiskHeartbeatInfo: *heartbeatInfo})
return s.diskTbl.UpdateDisk(diskInfo.DiskID, diskRecord)
})
}
func (s *ShardNodeManager) diskInfoToDiskInfoRecord(info *clustermgr.ShardNodeDiskInfo) *normaldb.ShardNodeDiskInfoRecord {
@ -507,7 +847,8 @@ func (s *shardNodePersistentHandler) addDiskNoLocked(di *diskItem) error {
func (s *shardNodePersistentHandler) updateNodeNoLocked(n *nodeItem) error {
return s.nodeTbl.UpdateNode(s.nodeInfoToNodeInfoRecord(&clustermgr.ShardNodeInfo{
NodeInfo: n.info.NodeInfo,
NodeInfo: n.info.NodeInfo,
ShardNodeExtraInfo: n.info.extraInfo.(clustermgr.ShardNodeExtraInfo),
}))
}

77
blobstore/clustermgr/cluster/shardnode_mgr_refresh.go Normal file
View File

@ -0,0 +1,77 @@
// Copyright 2022 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implieb. See the License for the specific language governing
// permissions and limitations under the License.
package cluster
import (
"context"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/common/proto"
)
// refresh use for refreshing storage allocator info and cluster statistic info
func (s *ShardNodeManager) refresh(ctx context.Context) {
// space and disk stat info
spaceStatInfos := make(map[proto.DiskType]*clustermgr.SpaceStatInfo)
// generate diskType -> nodeSet -> diskSet -> idc -> rack -> shardnode storage and statInfo
nodeSetAllocators := make(map[proto.DiskType]nodeSetAllocatorMap)
diskSetAllocators := make(map[proto.DiskType]diskSetAllocatorMap)
nodeSetsMap := s.topoMgr.GetAllNodeSets(ctx)
for diskType, nodeSets := range nodeSetsMap {
if _, ok := nodeSetAllocators[diskType]; !ok {
nodeSetAllocators[diskType] = make(nodeSetAllocatorMap)
}
if _, ok := diskSetAllocators[diskType]; !ok {
diskSetAllocators[diskType] = make(diskSetAllocatorMap)
}
if _, ok := spaceStatInfos[diskType]; !ok {
spaceStatInfos[diskType] = &clustermgr.SpaceStatInfo{}
}
spaceStatInfo := spaceStatInfos[diskType]
diskStatInfo := make(map[string]*clustermgr.DiskStatInfo)
for i := range s.cfg.IDC {
diskStatInfo[s.cfg.IDC[i]] = &clustermgr.DiskStatInfo{IDC: s.cfg.IDC[i]}
}
for _, nodeSet := range nodeSets {
nodeSetAllocator := newNodeSetAllocator(nodeSet.ID())
for _, diskSet := range nodeSet.GetDiskSets() {
disks := diskSet.GetDisks()
idcAllocators, diskSetFreeShard := s.generateDiskSetStorage(ctx, disks, spaceStatInfo, diskStatInfo)
diskSetAllocator := newDiskSetAllocator(diskSet.ID(), int64(diskSetFreeShard), idcAllocators)
diskSetAllocators[diskType][diskSet.ID()] = diskSetAllocator
nodeSetAllocator.addDiskSet(diskSetAllocator)
}
nodeSetAllocators[diskType][nodeSet.ID()] = nodeSetAllocator
}
for idc := range diskStatInfo {
spaceStatInfo.DisksStatInfos = append(spaceStatInfo.DisksStatInfos, *diskStatInfo[idc])
}
spaceStatInfo.TotalShardNode = int64(s.topoMgr.GetNodeNum(diskType))
}
s.allocator.Store(newAllocator(allocatorConfig{
nodeSets: nodeSetAllocators,
diskSets: diskSetAllocators,
dg: s,
tg: s.topoMgr,
diffHost: s.cfg.HostAware,
diffRack: s.cfg.RackAware,
}))
s.spaceStatInfo.Store(spaceStatInfos)
}

2
blobstore/clustermgr/config_test.go
View File

@ -54,7 +54,7 @@ func TestConfig(t *testing.T) {
codeModePolicies := make([]codemode.Policy, 0)
err = json.Unmarshal([]byte(rawCodeModePolicies), &codeModePolicies)
require.NoError(t, err)
require.Equal(t, testServiceCfg.CodeModePolicies, codeModePolicies)
require.Equal(t, testServiceCfg.VolumeCodeModePolicies, codeModePolicies)
rawChunkSize, err := testClusterClient.GetConfig(context.Background(), proto.VolumeChunkSizeKey)
require.NoError(t, err)

28
blobstore/clustermgr/handler.go
View File

@ -30,11 +30,11 @@ func NewHandler(service *Service) *rpc.Router {
rpc.POST("/config/delete", service.ConfigDelete, rpc.OptArgsQuery())
//==================disk==========================
//==================blobnode disk==========================
rpc.RegisterArgsParser(&clustermgr.DiskInfoArgs{}, "json")
rpc.RegisterArgsParser(&clustermgr.ListOptionArgs{}, "json")
rpc.POST("/diskid/alloc", service.DiskIdAlloc)
rpc.POST("/diskid/alloc", service.DiskIDAlloc)
rpc.GET("/disk/info", service.DiskInfo, rpc.OptArgsQuery())
@ -56,7 +56,7 @@ func NewHandler(service *Service) *rpc.Router {
rpc.POST("/admin/disk/update", service.AdminDiskUpdate, rpc.OptArgsBody())
//=====================node==========================
//=====================blobnode==========================
rpc.RegisterArgsParser(&clustermgr.NodeInfoArgs{}, "json")
rpc.POST("/node/add", service.NodeAdd, rpc.OptArgsBody())
@ -67,6 +67,28 @@ func NewHandler(service *Service) *rpc.Router {
rpc.GET("/topo/info", service.TopoInfo)
//==================shardnode disk==========================
rpc.POST("/shardnode/diskid/alloc", service.ShardNodeDiskIDAlloc)
rpc.GET("/shardnode/disk/info", service.ShardNodeDiskInfo, rpc.OptArgsQuery())
rpc.POST("/shardnode/disk/add", service.ShardNodeDiskAdd, rpc.OptArgsBody())
rpc.POST("/shardnode/disk/set", service.ShardNodeDiskSet, rpc.OptArgsBody())
rpc.GET("/shardnode/disk/list", service.ShardNodeDiskList, rpc.OptArgsQuery())
rpc.POST("/shardnode/disk/heartbeat", service.ShardNodeDiskHeartbeat, rpc.OptArgsBody())
rpc.POST("/admin/shardnode/disk/update", service.AdminShardNodeDiskUpdate, rpc.OptArgsBody())
//=====================shardnode==========================
rpc.POST("/shardnode/add", service.ShardNodeAdd, rpc.OptArgsBody())
rpc.GET("/shardnode/info", service.ShardNodeInfo, rpc.OptArgsQuery())
rpc.GET("/shardnode/topo/info", service.ShardNodeTopoInfo)
//==================service==========================
rpc.RegisterArgsParser(&clustermgr.GetServiceArgs{}, "json")

4
blobstore/clustermgr/manage.go
View File

@ -20,6 +20,7 @@ import (
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
apierrors "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/raftserver"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/trace"
@ -121,7 +122,8 @@ func (s *Service) Stat(c *rpc.Context) {
ret := new(clustermgr.StatInfo)
ret.RaftStatus = s.raftNode.Status()
ret.LeaderHost = s.raftNode.GetLeaderHost()
ret.SpaceStat = *(s.BlobNodeMgr.Stat(ctx))
ret.BlobNodeSpaceStat = *(s.BlobNodeMgr.Stat(ctx, proto.DiskTypeHDD))
ret.ShardNodeSpaceStat = *(s.ShardNodeMgr.Stat(ctx, proto.DiskTypeNVMeSSD))
ret.VolumeStat = s.VolumeMgr.Stat(ctx)
ret.ReadOnly = s.Readonly
c.RespondJSON(ret)

4
blobstore/clustermgr/persistence/normaldb/shardnodedisktbl.go
View File

@ -27,7 +27,8 @@ func OpenShardNodeDiskTable(db kvstore.KVStore, ensureIndex bool) (*ShardNodeDis
}
table := &ShardNodeDiskTable{
diskTable: &diskTable{
diskTbl: db.Table(shardNodeDiskCF),
diskTbl: db.Table(shardNodeDiskCF),
droppedDiskTbl: db.Table(shardNodeDiskDropCF),
indexes: map[string]indexItem{
diskStatusIndex: {indexNames: []string{diskStatusIndex}, tbl: db.Table(shardNodeDiskStatusIndexCF)},
diskHostIndex: {indexNames: []string{diskHostIndex}, tbl: db.Table(shardNodeDiskHostIndexCF)},
@ -36,6 +37,7 @@ func OpenShardNodeDiskTable(db kvstore.KVStore, ensureIndex bool) (*ShardNodeDis
},
},
}
table.diskTable.rd = table
// ensure index
if ensureIndex {

127
blobstore/clustermgr/shardnode.go Normal file
View File

@ -0,0 +1,127 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package clustermgr
import (
"encoding/json"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
"github.com/cubefs/cubefs/blobstore/clustermgr/cluster"
apierrors "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
func (s *Service) ShardNodeAdd(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.ShardNodeInfo)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept ShardNodeAdd request, args: %v", args)
if nodeID, ok := s.ShardNodeMgr.CheckNodeInfoDuplicated(ctx, &args.NodeInfo); ok {
span.Warnf("node already exist, no need to create again, node info: %v", args)
c.RespondJSON(&clustermgr.NodeIDAllocRet{NodeID: nodeID})
return
}
if args.ClusterID != s.ClusterID {
span.Warn("invalid clusterID")
c.RespondError(apierrors.ErrIllegalArguments)
return
}
for i := range s.IDC {
if args.Idc == s.IDC[i] {
break
}
if i == len(s.IDC)-1 {
span.Warnf("invalid idc %s, service idc: %v", args.Idc, s.IDC)
c.RespondError(apierrors.ErrIllegalArguments)
return
}
}
if err := s.ShardNodeMgr.ValidateNodeInfo(ctx, &args.NodeInfo); err != nil {
span.Warn("invalid nodeinfo")
c.RespondError(err)
return
}
nodeID, err := s.ShardNodeMgr.AllocNodeID(ctx)
if err != nil {
span.Errorf("alloc node id failed =>", errors.Detail(err))
c.RespondError(err)
return
}
args.NodeID = nodeID
data, err := json.Marshal(args)
if err != nil {
span.Errorf("json marshal failed, node info: %v, error: %v", args, err)
c.RespondError(errors.Info(apierrors.ErrUnexpected).Detail(err))
return
}
proposeInfo := base.EncodeProposeInfo(s.ShardNodeMgr.GetModuleName(), cluster.OperTypeAddNode, data, base.ProposeContext{ReqID: span.TraceID()})
err = s.raftNode.Propose(ctx, proposeInfo)
if err != nil {
span.Error(err)
c.RespondError(apierrors.ErrRaftPropose)
return
}
c.RespondJSON(&clustermgr.NodeIDAllocRet{NodeID: nodeID})
}
func (s *Service) ShardNodeInfo(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.NodeInfoArgs)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept ShardNodeInfo request, args: %v", args)
// linear read
if err := s.raftNode.ReadIndex(ctx); err != nil {
span.Errorf("node info read index error: %v", err)
c.RespondError(apierrors.ErrRaftReadIndex)
return
}
ret, err := s.ShardNodeMgr.GetNodeInfo(ctx, args.NodeID)
if err != nil {
span.Warnf("node not found: %d", args.NodeID)
c.RespondError(err)
return
}
c.RespondJSON(ret)
}
func (s *Service) ShardNodeTopoInfo(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
span.Info("accept ShardNodeTopoInfo request")
// linear read
if err := s.raftNode.ReadIndex(ctx); err != nil {
span.Errorf("topo info read index error: %v", err)
c.RespondError(apierrors.ErrRaftReadIndex)
return
}
c.RespondJSON(s.ShardNodeMgr.GetTopoInfo(ctx))
}

282
blobstore/clustermgr/shardnode_disk.go Normal file
View File

@ -0,0 +1,282 @@
// Copyright 2024 The CubeFS Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
// implied. See the License for the specific language governing
// permissions and limitations under the License.
package clustermgr
import (
"encoding/json"
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
"github.com/cubefs/cubefs/blobstore/clustermgr/cluster"
apierrors "github.com/cubefs/cubefs/blobstore/common/errors"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc"
"github.com/cubefs/cubefs/blobstore/common/trace"
"github.com/cubefs/cubefs/blobstore/util/errors"
)
func (s *Service) ShardNodeDiskIDAlloc(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
span.Info("accept ShardNodeDiskIDAlloc request")
diskID, err := s.ShardNodeMgr.AllocDiskID(ctx)
if err != nil {
span.Error("alloc disk id failed =>", errors.Detail(err))
c.RespondError(err)
return
}
c.RespondJSON(&clustermgr.DiskIDAllocRet{DiskID: diskID})
}
func (s *Service) ShardNodeDiskAdd(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.ShardNodeDiskInfo)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept ShardNodeDiskAdd request, args: %v", args)
if args.ClusterID != s.ClusterID {
span.Warn("invalid clusterID")
c.RespondError(apierrors.ErrIllegalArguments)
return
}
curDiskID := s.ScopeMgr.GetCurrent(cluster.ShardNodeDiskIDScopeName)
if proto.DiskID(curDiskID) < args.DiskID {
span.Warn("invalid disk_id")
c.RespondError(apierrors.ErrIllegalArguments)
return
}
curNodeID := s.ScopeMgr.GetCurrent(cluster.ShardNodeIDScopeName)
if proto.NodeID(curNodeID) < args.NodeID {
span.Warn("invalid node_id")
c.RespondError(apierrors.ErrIllegalArguments)
return
}
err := s.ShardNodeMgr.AddDisk(ctx, args)
if err != nil {
c.RespondError(err)
return
}
}
func (s *Service) ShardNodeDiskInfo(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.DiskInfoArgs)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept ShardNodeDiskInfo request, args: %v", args)
// linear read
if err := s.raftNode.ReadIndex(ctx); err != nil {
span.Errorf("info read index error: %v", err)
c.RespondError(apierrors.ErrRaftReadIndex)
return
}
ret, err := s.ShardNodeMgr.GetDiskInfo(ctx, args.DiskID)
if err != nil {
span.Warnf("disk not found: %d", args.DiskID)
c.RespondError(err)
return
}
c.RespondJSON(ret)
}
func (s *Service) ShardNodeDiskList(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.ListOptionArgs)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept ShardNodeDiskList request, args: %v", args)
if err := s.raftNode.ReadIndex(ctx); err != nil {
span.Errorf("list read index error: %v", err)
c.RespondError(apierrors.ErrRaftReadIndex)
return
}
// idc can not be nil when rack param set
if args.Rack != "" && args.Idc == "" {
span.Warnf("can not list disk by rack only")
c.RespondError(apierrors.ErrIllegalArguments)
return
}
if args.Marker != proto.InvalidDiskID {
if _, err := s.ShardNodeMgr.GetDiskInfo(ctx, args.Marker); err != nil {
span.Warnf("invalid marker, marker disk not exist")
err = apierrors.ErrIllegalArguments
c.RespondError(err)
return
}
}
if args.Count == 0 {
args.Count = 10
}
disks, marker, err := s.ShardNodeMgr.ListDiskInfo(ctx, args)
if err != nil {
span.Errorf("list disk info failed =>", errors.Detail(err))
err = errors.Info(apierrors.ErrUnexpected).Detail(err)
c.RespondError(err)
return
}
ret := &clustermgr.ListShardNodeDiskRet{
Disks: disks,
Marker: marker,
}
c.RespondJSON(ret)
}
func (s *Service) ShardNodeDiskSet(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.DiskSetArgs)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept ShardNodeDiskSet request, args: %v", args)
// not allow to set disk dropped in this API
if args.Status < proto.DiskStatusNormal || args.Status >= proto.DiskStatusDropped {
c.RespondError(apierrors.ErrInvalidStatus)
return
}
diskInfo, err := s.ShardNodeMgr.GetDiskInfo(ctx, args.DiskID)
if err != nil {
c.RespondError(err)
return
}
if diskInfo.Status == args.Status {
return
}
err = s.ShardNodeMgr.SetStatus(ctx, args.DiskID, args.Status, false)
if err != nil {
span.Errorf("disk set failed =>", errors.Detail(err))
c.RespondError(err)
return
}
data, err := json.Marshal(args)
if err != nil {
span.Errorf("set args: %v, error: %v", args, err)
c.RespondError(errors.Info(apierrors.ErrUnexpected).Detail(err))
return
}
proposeInfo := base.EncodeProposeInfo(s.ShardNodeMgr.GetModuleName(), cluster.OperTypeSetDiskStatus, data, base.ProposeContext{ReqID: span.TraceID()})
err = s.raftNode.Propose(ctx, proposeInfo)
if err != nil {
span.Error(err)
c.RespondError(apierrors.ErrRaftPropose)
return
}
if args.Status == proto.DiskStatusBroken {
err = s.CatalogMgr.UpdateShardUnitStatus(ctx, args.DiskID)
if err != nil {
span.Error(errors.Detail(err))
return
}
}
}
func (s *Service) ShardNodeDiskHeartbeat(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.ShardNodeDisksHeartbeatArgs)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
heartbeatDisks := make([]clustermgr.ShardNodeDiskHeartbeatInfo, 0)
for i := range args.Disks {
// filter frequentHeartBeat disk
frequentHeartBeat, err := s.ShardNodeMgr.IsFrequentHeartBeat(args.Disks[i].DiskID, s.HeartbeatNotifyIntervalS)
if err != nil {
span.Errorf("filter frequent heartbeat disk %d failed, err: %v", args.Disks[i].DiskID, err)
c.RespondError(err)
return
}
if !frequentHeartBeat {
heartbeatDisks = append(heartbeatDisks, args.Disks[i])
} else {
span.Warnf("disk %d heartbeat too frequent", args.Disks[i].DiskID)
}
}
if len(heartbeatDisks) == 0 {
return
}
args.Disks = heartbeatDisks
data, err := json.Marshal(args)
span.Debugf("heartbeat params: %s", string(data))
if err != nil {
span.Errorf("heartbeat args: %v, error: %v", args, err)
err = errors.Info(apierrors.ErrUnexpected).Detail(err)
c.RespondError(err)
return
}
proposeInfo := base.EncodeProposeInfo(s.ShardNodeMgr.GetModuleName(), cluster.OperTypeHeartbeatDiskInfo, data, base.ProposeContext{ReqID: span.TraceID()})
err = s.raftNode.Propose(ctx, proposeInfo)
if err != nil {
span.Error(err)
c.RespondError(apierrors.ErrRaftPropose)
return
}
}
func (s *Service) AdminShardNodeDiskUpdate(c *rpc.Context) {
ctx := c.Request.Context()
span := trace.SpanFromContextSafe(ctx)
args := new(clustermgr.ShardNodeDiskInfo)
if err := c.ParseArgs(args); err != nil {
c.RespondError(err)
return
}
span.Infof("accept AdminShardNodeDiskUpdate request, args: %v", args)
_, err := s.ShardNodeMgr.GetDiskInfo(ctx, args.DiskID)
if err != nil {
span.Errorf("admin update disk:%d not exist", args.DiskID)
c.RespondError(err)
return
}
data, err := json.Marshal(args)
if err != nil {
span.Errorf("update args: %v, error: %v", args, err)
c.RespondError(errors.Info(apierrors.ErrUnexpected).Detail(err))
return
}
proposeInfo := base.EncodeProposeInfo(s.ShardNodeMgr.GetModuleName(), cluster.OperTypeAdminUpdateDisk, data, base.ProposeContext{ReqID: span.TraceID()})
err = s.raftNode.Propose(ctx, proposeInfo)
if err != nil {
span.Error(err)
c.RespondError(apierrors.ErrRaftPropose)
return
}
}

129
blobstore/clustermgr/svr.go
View File

@ -81,10 +81,15 @@ const (
defaultMetricReportIntervalM = 2
defaultCheckConsistentIntervalM = 360
defaultNodeSetCap = 108
defaultNodeSetRackCap = 6
defaultDiskSetCap = 2160
defaultDiskCountPerNodeInDiskSet = 20
defaultBlobNodeSetCap = 24
defaultBlobNodeSetRackCap = 6
defaultBlobNodeDiskSetCap = 120
defaultDiskCountPerBlobNodeInDiskSet = 10
defaultShardNodeSetCap = 12
defaultShardNodeSetRackCap = 3
defaultShardNodeDiskSetCap = 36
defaultDiskCountPerShardNodeInDiskSet = 3
)
var (
@ -103,10 +108,12 @@ type Config struct {
DBCacheSize uint64 `json:"db_cache_size"`
NormalDBPath string `json:"normal_db_path"`
KvDBPath string `json:"kv_db_path"`
CodeModePolicies []codemode.Policy `json:"code_mode_policies"`
VolumeCodeModePolicies []codemode.Policy `json:"volume_code_mode_policies"`
ShardCodeModeName codemode.CodeModeName `json:"shard_code_mode_name"`
ClusterCfg map[string]interface{} `json:"cluster_config"`
RaftConfig RaftConfig `json:"raft_config"`
DiskMgrConfig cluster.DiskMgrConfig `json:"disk_mgr_config"`
BlobNodeDiskMgrConfig cluster.DiskMgrConfig `json:"blob_node_disk_mgr_config"`
ShardNodeDiskMgrConfig cluster.DiskMgrConfig `json:"shard_node_disk_mgr_config"`
ClusterReportIntervalS int `json:"cluster_report_interval_s"`
ConsulAgentAddr string `json:"consul_agent_addr"`
ConsulToken string `json:"consul_token"`
@ -133,9 +140,10 @@ type Service struct {
ServiceMgr *servicemgr.ServiceMgr
// Note: BlobNodeMgr should always list before volumeMgr
// cause BlobNodeMgr applier LoadData should be call first, or VolumeMgr LoadData may return error with disk not found
BlobNodeMgr *cluster.BlobNodeManager
VolumeMgr *volumemgr.VolumeMgr
KvMgr *kvmgr.KvMgr
BlobNodeMgr *cluster.BlobNodeManager
ShardNodeMgr *cluster.ShardNodeManager
VolumeMgr *volumemgr.VolumeMgr
KvMgr *kvmgr.KvMgr
dbs map[string]base.SnapshotDB
// status indicate service's current state, like normal/snapshot
@ -234,7 +242,7 @@ func New(cfg *Config) (*Service, error) {
if err != nil {
log.Fatalf("new scopeMgr failed, err: %v", err)
}
blobNodeMgr, err := cluster.NewBlobNodeMgr(scopeMgr, normalDB, cfg.DiskMgrConfig)
blobNodeMgr, err := cluster.NewBlobNodeMgr(scopeMgr, normalDB, cfg.BlobNodeDiskMgrConfig)
if err != nil {
log.Fatalf("new blobNodeMgr failed, err: %v", err)
}
@ -262,6 +270,13 @@ func New(cfg *Config) (*Service, error) {
service.BlobNodeMgr = blobNodeMgr
service.ServiceMgr = serviceMgr
service.ScopeMgr = scopeMgr
if len(cfg.ShardCodeModeName) != 0 {
shardNodeMgr, err := cluster.NewShardNodeMgr(scopeMgr, normalDB, cfg.ShardNodeDiskMgrConfig)
if err != nil {
log.Fatalf("new shardNodeMgr failed, err: %v", err)
}
service.ShardNodeMgr = shardNodeMgr
}
// raft server initial
applyIndex := uint64(0)
@ -307,13 +322,21 @@ func New(cfg *Config) (*Service, error) {
scopeMgr.SetRaftServer(raftServer)
volumeMgr.SetRaftServer(raftServer)
configMgr.SetRaftServer(raftServer)
if len(cfg.ShardCodeModeName) != 0 {
service.ShardNodeMgr.SetRaftServer(raftServer)
}
// wait for raft start
service.waitForRaftStart()
// start volumeMgr task
volumeMgr.Start()
// refresh disk expire time after all ready
blobNodeMgr.RefreshExpireTime()
if len(cfg.ShardCodeModeName) != 0 {
service.ShardNodeMgr.RefreshExpireTime()
}
// start raft node background progress
go raftNode.Start()
@ -413,24 +436,25 @@ func (c *Config) checkAndFix() (err error) {
return errors.New("ChunkOversoldRatio must be between (1-VolumeOverboughtRatioKey)^2/VolumeOverboughtRatioKey and (1-VolumeOverboughtRatioKey)/VolumeOverboughtRatioKey")
}
c.VolumeMgrConfig.VolumeOverboughtRatio = volumeOverboughtRatio
c.DiskMgrConfig.ChunkOversoldRatio = chunkOversoldRatio
c.BlobNodeDiskMgrConfig.ChunkOversoldRatio = chunkOversoldRatio
}
c.VolumeMgrConfig.ChunkSize = c.ChunkSize
c.DiskMgrConfig.ChunkSize = int64(c.ChunkSize)
c.BlobNodeDiskMgrConfig.ChunkSize = int64(c.ChunkSize)
c.ClusterCfg[proto.VolumeChunkSizeKey] = c.ChunkSize
c.ClusterCfg[proto.CodeModeConfigKey] = c.CodeModePolicies
c.ClusterCfg[proto.CodeModeConfigKey] = c.VolumeCodeModePolicies
if len(c.CodeModePolicies) == 0 {
return errors.New("invalid code mode config")
if len(c.VolumeCodeModePolicies) == 0 {
return errors.New("invalid volume code mode config")
}
sort.Slice(c.CodeModePolicies, func(i, j int) bool {
return c.CodeModePolicies[i].MinSize < c.CodeModePolicies[j].MinSize
sort.Slice(c.VolumeCodeModePolicies, func(i, j int) bool {
return c.VolumeCodeModePolicies[i].MinSize < c.VolumeCodeModePolicies[j].MinSize
})
sortedPolicies := make([]codemode.Policy, 0)
for i := range c.CodeModePolicies {
if c.CodeModePolicies[i].Enable {
sortedPolicies = append(sortedPolicies, c.CodeModePolicies[i])
for i := range c.VolumeCodeModePolicies {
if c.VolumeCodeModePolicies[i].Enable {
sortedPolicies = append(sortedPolicies, c.VolumeCodeModePolicies[i])
}
}
if len(sortedPolicies) > 0 {
@ -438,9 +462,9 @@ func (c *Config) checkAndFix() (err error) {
return errors.New("min size range must be started with 0")
}
} else {
for _, modePolicy := range c.CodeModePolicies {
for _, modePolicy := range c.VolumeCodeModePolicies {
codeMode := modePolicy.ModeName.GetCodeMode()
c.DiskMgrConfig.CodeModes = append(c.DiskMgrConfig.CodeModes, codeMode)
c.BlobNodeDiskMgrConfig.CodeModes = append(c.BlobNodeDiskMgrConfig.CodeModes, codeMode)
}
}
for i := 0; i < len(sortedPolicies)-1; i++ {
@ -460,16 +484,19 @@ func (c *Config) checkAndFix() (err error) {
if c.UnavailableIDC == "" && codeMode.Tactic().AZCount != len(c.IDC) {
return errors.New("idc count not match modeTactic AZCount")
}
c.DiskMgrConfig.CodeModes = append(c.DiskMgrConfig.CodeModes, codeMode)
c.BlobNodeDiskMgrConfig.CodeModes = append(c.BlobNodeDiskMgrConfig.CodeModes, codeMode)
}
c.VolumeMgrConfig.CodeModePolicies = c.CodeModePolicies
c.VolumeMgrConfig.CodeModePolicies = c.VolumeCodeModePolicies
c.DiskMgrConfig.IDC = c.IDC
c.BlobNodeDiskMgrConfig.IDC = c.IDC
c.VolumeMgrConfig.IDC = c.IDC
c.VolumeMgrConfig.UnavailableIDC = c.UnavailableIDC
c.VolumeMgrConfig.Region = c.Region
c.VolumeMgrConfig.ClusterID = c.ClusterID
if c.UnavailableIDC == "" && c.ShardCodeModeName.Tactic().AZCount != len(c.IDC) {
return errors.New("idc count not match shardNode modeTactic AZCount")
}
if c.RaftConfig.SnapshotPatchNum == 0 {
c.RaftConfig.SnapshotPatchNum = 64
}
@ -487,24 +514,44 @@ func (c *Config) checkAndFix() (err error) {
c.KvDBPath = c.DBPath + "/kvdb"
}
copySetConfs := c.DiskMgrConfig.CopySetConfigs
if copySetConfs == nil {
copySetConfs = make(map[proto.NodeRole]map[proto.DiskType]cluster.CopySetConfig)
c.DiskMgrConfig.CopySetConfigs = copySetConfs
blobNodeCopySetConfs := c.BlobNodeDiskMgrConfig.CopySetConfigs
if blobNodeCopySetConfs == nil {
blobNodeCopySetConfs = make(map[proto.DiskType]cluster.CopySetConfig)
c.BlobNodeDiskMgrConfig.CopySetConfigs = blobNodeCopySetConfs
}
if copySetConfs[proto.NodeRoleBlobNode] == nil {
copySetConfs[proto.NodeRoleBlobNode] = make(map[proto.DiskType]cluster.CopySetConfig)
blobNodeHDDCopySetConf := blobNodeCopySetConfs[proto.DiskTypeHDD]
defaulter.Equal(&blobNodeHDDCopySetConf.NodeSetCap, defaultBlobNodeSetCap)
defaulter.Equal(&blobNodeHDDCopySetConf.NodeSetRackCap, defaultBlobNodeSetRackCap)
defaulter.Equal(&blobNodeHDDCopySetConf.DiskSetCap, defaultBlobNodeDiskSetCap)
defaulter.Equal(&blobNodeHDDCopySetConf.DiskCountPerNodeInDiskSet, defaultDiskCountPerBlobNodeInDiskSet)
blobNodeCopySetConfs[proto.DiskTypeHDD] = blobNodeHDDCopySetConf
for diskType, copySetConf := range blobNodeCopySetConfs {
copySetConf.NodeSetIdcCap = (copySetConf.NodeSetCap + len(c.IDC) - 1) / len(c.IDC)
blobNodeCopySetConfs[diskType] = copySetConf
}
blobNodeHDDCopySetConf := copySetConfs[proto.NodeRoleBlobNode][proto.DiskTypeHDD]
defaulter.Equal(&blobNodeHDDCopySetConf.NodeSetCap, defaultNodeSetCap)
defaulter.Equal(&blobNodeHDDCopySetConf.NodeSetRackCap, defaultNodeSetRackCap)
defaulter.Equal(&blobNodeHDDCopySetConf.DiskSetCap, defaultDiskSetCap)
defaulter.Equal(&blobNodeHDDCopySetConf.DiskCountPerNodeInDiskSet, defaultDiskCountPerNodeInDiskSet)
copySetConfs[proto.NodeRoleBlobNode][proto.DiskTypeHDD] = blobNodeHDDCopySetConf
for _, copySetConfOfRole := range copySetConfs {
for diskType, copySetConf := range copySetConfOfRole {
if len(c.ShardCodeModeName) != 0 {
shardNodeTactic := c.ShardCodeModeName.GetCodeMode().Tactic()
if !shardNodeTactic.IsReplicateMode() {
return errors.New("invalid shard code mode config")
}
c.ShardNodeDiskMgrConfig.CodeModes = append(c.ShardNodeDiskMgrConfig.CodeModes, c.ShardCodeModeName.GetCodeMode())
c.ShardNodeDiskMgrConfig.IDC = c.IDC
shardNodeCopySetConfs := c.ShardNodeDiskMgrConfig.CopySetConfigs
if shardNodeCopySetConfs == nil {
shardNodeCopySetConfs = make(map[proto.DiskType]cluster.CopySetConfig)
c.ShardNodeDiskMgrConfig.CopySetConfigs = shardNodeCopySetConfs
}
shardNodeNVMeCopySetConf := shardNodeCopySetConfs[proto.DiskTypeNVMeSSD]
defaulter.Equal(&shardNodeNVMeCopySetConf.NodeSetCap, defaultShardNodeSetCap)
defaulter.Equal(&shardNodeNVMeCopySetConf.NodeSetRackCap, defaultShardNodeSetRackCap)
defaulter.Equal(&shardNodeNVMeCopySetConf.DiskSetCap, defaultShardNodeDiskSetCap)
defaulter.Equal(&shardNodeNVMeCopySetConf.DiskCountPerNodeInDiskSet, defaultDiskCountPerShardNodeInDiskSet)
shardNodeCopySetConfs[proto.DiskTypeNVMeSSD] = shardNodeNVMeCopySetConf
for diskType, copySetConf := range shardNodeCopySetConfs {
copySetConf.NodeSetIdcCap = (copySetConf.NodeSetCap + len(c.IDC) - 1) / len(c.IDC)
copySetConfOfRole[diskType] = copySetConf
shardNodeCopySetConfs[diskType] = copySetConf
}
}
@ -597,7 +644,7 @@ func (s *Service) loop() {
Readonly: s.Readonly,
Nodes: make([]string, 0),
}
spaceStatInfo := s.BlobNodeMgr.Stat(ctx)
spaceStatInfo := s.BlobNodeMgr.Stat(ctx, proto.DiskTypeHDD)
clusterInfo.Capacity = spaceStatInfo.TotalSpace
clusterInfo.Available = spaceStatInfo.WritableSpace
// filter learner node

27
blobstore/clustermgr/svr_test.go
View File

@ -48,7 +48,7 @@ var testServiceCfg = &Config{
ClusterID: 1,
Readonly: false,
DBPath: "/tmp/tmpsvrdb-" + randID(),
CodeModePolicies: []codemode.Policy{
VolumeCodeModePolicies: []codemode.Policy{
{
ModeName: codemode.EC15P12.Name(),
MinSize: 1048577,
@ -64,6 +64,7 @@ var testServiceCfg = &Config{
Enable: true,
},
},
ShardCodeModeName: codemode.Replica3.Name(),
ClusterCfg: map[string]interface{}{},
ClusterReportIntervalS: 1,
MetricReportIntervalM: 1,
@ -81,7 +82,12 @@ var testServiceCfg = &Config{
TickIntervalMs: 20,
},
},
DiskMgrConfig: cluster.DiskMgrConfig{
BlobNodeDiskMgrConfig: cluster.DiskMgrConfig{
RefreshIntervalS: 300,
RackAware: false,
HostAware: true,
},
ShardNodeDiskMgrConfig: cluster.DiskMgrConfig{
RefreshIntervalS: 300,
RackAware: false,
HostAware: true,
@ -126,13 +132,20 @@ func initTestService(t testing.TB) (*Service, func()) {
cfg.RaftConfig.ServerConfig.Members = []raftserver.Member{
{NodeID: 1, Host: fmt.Sprintf("127.0.0.1:%d", GetFreePort()), Learner: false},
}
copySetConfs := make(map[proto.NodeRole]map[proto.DiskType]cluster.CopySetConfig)
copySetConfs[proto.NodeRoleBlobNode] = make(map[proto.DiskType]cluster.CopySetConfig)
blobNodeHDDCopySetConf := copySetConfs[proto.NodeRoleBlobNode][proto.DiskTypeHDD]
blobNodeCopySetConfs := make(map[proto.DiskType]cluster.CopySetConfig)
blobNodeHDDCopySetConf := blobNodeCopySetConfs[proto.DiskTypeHDD]
blobNodeHDDCopySetConf.NodeSetCap = 3
blobNodeHDDCopySetConf.DiskSetCap = 6
copySetConfs[proto.NodeRoleBlobNode][proto.DiskTypeHDD] = blobNodeHDDCopySetConf
cfg.DiskMgrConfig.CopySetConfigs = copySetConfs
blobNodeCopySetConfs[proto.DiskTypeHDD] = blobNodeHDDCopySetConf
cfg.BlobNodeDiskMgrConfig.CopySetConfigs = blobNodeCopySetConfs
shardNodeCopySetConfs := make(map[proto.DiskType]cluster.CopySetConfig)
shardNodeNVMeCopySetConf := shardNodeCopySetConfs[proto.DiskTypeNVMeSSD]
shardNodeNVMeCopySetConf.NodeSetCap = 3
shardNodeNVMeCopySetConf.DiskSetCap = 6
shardNodeNVMeCopySetConf.DiskCountPerNodeInDiskSet = 6
shardNodeCopySetConfs[proto.DiskTypeNVMeSSD] = shardNodeNVMeCopySetConf
cfg.ShardNodeDiskMgrConfig.CopySetConfigs = shardNodeCopySetConfs
os.Mkdir(cfg.DBPath, 0o755)
testService, err := New(&cfg)
require.NoError(t, err)

16
blobstore/clustermgr/volume_test.go
View File

@ -42,7 +42,7 @@ func initServiceWithData() (*Service, func()) {
cfg.DBPath = os.TempDir() + "/" + uuid.NewString() + strconv.FormatInt(rand.Int63n(math.MaxInt64), 10)
cfg.VolumeMgrConfig.FlushIntervalS = 600
cfg.VolumeMgrConfig.MinAllocableVolumeCount = 0
cfg.DiskMgrConfig.HeartbeatExpireIntervalS = 600
cfg.BlobNodeDiskMgrConfig.HeartbeatExpireIntervalS = 600
cfg.ClusterReportIntervalS = 3
cfg.ClusterCfg[proto.VolumeReserveSizeKey] = "20000000"
cfg.RaftConfig.ServerConfig.ListenPort = GetFreePort()
@ -68,22 +68,22 @@ func initServiceWithData() (*Service, func()) {
func TestService_CreateVolume(t *testing.T) {
testServiceCfg.UnavailableIDC = "z0"
for i := range testServiceCfg.CodeModePolicies {
testServiceCfg.CodeModePolicies[i].Enable = false
for i := range testServiceCfg.VolumeCodeModePolicies {
testServiceCfg.VolumeCodeModePolicies[i].Enable = false
}
testServiceCfg.CodeModePolicies = append(testServiceCfg.CodeModePolicies,
testServiceCfg.VolumeCodeModePolicies = append(testServiceCfg.VolumeCodeModePolicies,
codemode.Policy{ModeName: codemode.EC4P4L2.Name(), Enable: true})
testService, _ := initServiceWithData()
cleanTestService(testService) // waiting closed
cleanWG.Done()
// set EC4P4L2 enable=false
for i := range testServiceCfg.CodeModePolicies {
if testServiceCfg.CodeModePolicies[i].ModeName == codemode.EC4P4L2.Name() {
testServiceCfg.CodeModePolicies[i].Enable = false
for i := range testServiceCfg.VolumeCodeModePolicies {
if testServiceCfg.VolumeCodeModePolicies[i].ModeName == codemode.EC4P4L2.Name() {
testServiceCfg.VolumeCodeModePolicies[i].Enable = false
} else {
testServiceCfg.CodeModePolicies[i].Enable = true
testServiceCfg.VolumeCodeModePolicies[i].Enable = true
}
}
_, clean := initServiceWithData()

2
blobstore/clustermgr/volumemgr/volumemgr.go
View File

@ -750,7 +750,7 @@ func (v *VolumeMgr) loop() {
span_.Infof("leader node start create volume")
allocatableVolCounts := v.allocator.StatAllocatable()
diskNums := v.diskMgr.Stat(ctx_).TotalDisk
diskNums := v.diskMgr.Stat(ctx_, proto.DiskTypeHDD).TotalDisk
CREATE:
for _, modeConfig := range v.codeMode {