diff --git a/blobstore/access/controller/cluster.go b/blobstore/access/controller/cluster.go index 195b1b234..73188d0ec 100644 --- a/blobstore/access/controller/cluster.go +++ b/blobstore/access/controller/cluster.go @@ -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 diff --git a/blobstore/access/controller/cluster_test.go b/blobstore/access/controller/cluster_test.go index b7faaa7d8..a9fdc943b 100644 --- a/blobstore/access/controller/cluster_test.go +++ b/blobstore/access/controller/cluster_test.go @@ -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, }, diff --git a/blobstore/api/clustermgr/disk.go b/blobstore/api/clustermgr/disk.go index dc1a4ff58..75541f705 100644 --- a/blobstore/api/clustermgr/disk.go +++ b/blobstore/api/clustermgr/disk.go @@ -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"` } diff --git a/blobstore/api/clustermgr/proto.go b/blobstore/api/clustermgr/proto.go index 7881b07ba..b1cfb9319 100644 --- a/blobstore/api/clustermgr/proto.go +++ b/blobstore/api/clustermgr/proto.go @@ -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 { diff --git a/blobstore/cli/access/cluster.go b/blobstore/cli/access/cluster.go index d2fca48ee..0ae906c97 100644 --- a/blobstore/cli/access/cluster.go +++ b/blobstore/cli/access/cluster.go @@ -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 } diff --git a/blobstore/clustermgr/node.go b/blobstore/clustermgr/blobnode.go similarity index 100% rename from blobstore/clustermgr/node.go rename to blobstore/clustermgr/blobnode.go diff --git a/blobstore/clustermgr/disk.go b/blobstore/clustermgr/blobnode_disk.go similarity index 97% rename from blobstore/clustermgr/disk.go rename to blobstore/clustermgr/blobnode_disk.go index 443c195ea..dc39d4a74 100644 --- a/blobstore/clustermgr/disk.go +++ b/blobstore/clustermgr/blobnode_disk.go @@ -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) diff --git a/blobstore/clustermgr/disk_test.go b/blobstore/clustermgr/blobnode_disk_test.go similarity index 100% rename from blobstore/clustermgr/disk_test.go rename to blobstore/clustermgr/blobnode_disk_test.go diff --git a/blobstore/clustermgr/node_test.go b/blobstore/clustermgr/blobnode_test.go similarity index 97% rename from blobstore/clustermgr/node_test.go rename to blobstore/clustermgr/blobnode_test.go index 7981d1449..20687c0bd 100644 --- a/blobstore/clustermgr/node_test.go +++ b/blobstore/clustermgr/blobnode_test.go @@ -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 { diff --git a/blobstore/clustermgr/cluster/alloc.go b/blobstore/clustermgr/cluster/alloc.go index 23a6968d7..a08ea469d 100644 --- a/blobstore/clustermgr/cluster/alloc.go +++ b/blobstore/clustermgr/cluster/alloc.go @@ -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 diff --git a/blobstore/clustermgr/cluster/alloc_test.go b/blobstore/clustermgr/cluster/alloc_test.go index 98a9e5ea0..69b77cb3a 100644 --- a/blobstore/clustermgr/cluster/alloc_test.go +++ b/blobstore/clustermgr/cluster/alloc_test.go @@ -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, diff --git a/blobstore/clustermgr/cluster/blobnode_mgr.go b/blobstore/clustermgr/cluster/blobnode_mgr.go index af3c0f448..1aec90801 100644 --- a/blobstore/clustermgr/cluster/blobnode_mgr.go +++ b/blobstore/clustermgr/cluster/blobnode_mgr.go @@ -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] diff --git a/blobstore/clustermgr/cluster/blobnode_mgr_refresh.go b/blobstore/clustermgr/cluster/blobnode_mgr_refresh.go new file mode 100644 index 000000000..cd320b346 --- /dev/null +++ b/blobstore/clustermgr/cluster/blobnode_mgr_refresh.go @@ -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) + } +} diff --git a/blobstore/clustermgr/cluster/refresh_test.go b/blobstore/clustermgr/cluster/blobnode_mgr_refresh_test.go similarity index 100% rename from blobstore/clustermgr/cluster/refresh_test.go rename to blobstore/clustermgr/cluster/blobnode_mgr_refresh_test.go diff --git a/blobstore/clustermgr/cluster/cluster.go b/blobstore/clustermgr/cluster/cluster.go index c93f6552c..6df40c610 100644 --- a/blobstore/clustermgr/cluster/cluster.go +++ b/blobstore/clustermgr/cluster/cluster.go @@ -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) } diff --git a/blobstore/clustermgr/cluster/metric.go b/blobstore/clustermgr/cluster/metric.go index e5ef3bcc7..d4d2505c2 100644 --- a/blobstore/clustermgr/cluster/metric.go +++ b/blobstore/clustermgr/cluster/metric.go @@ -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++ { diff --git a/blobstore/clustermgr/cluster/proto.go b/blobstore/clustermgr/cluster/proto.go index 2a10fa4f5..eb05c3cc3 100644 --- a/blobstore/clustermgr/cluster/proto.go +++ b/blobstore/clustermgr/cluster/proto.go @@ -10,7 +10,7 @@ import ( type nodeItemInfo struct { clustermgr.NodeInfo - // extraInfo interface{} + extraInfo interface{} } type nodeItem struct { diff --git a/blobstore/clustermgr/cluster/refresh.go b/blobstore/clustermgr/cluster/refresh.go deleted file mode 100644 index b466878ac..000000000 --- a/blobstore/clustermgr/cluster/refresh.go +++ /dev/null @@ -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) - } -} diff --git a/blobstore/clustermgr/cluster/shardnode_mgr.go b/blobstore/clustermgr/cluster/shardnode_mgr.go index 6145ea0be..7905b937b 100644 --- a/blobstore/clustermgr/cluster/shardnode_mgr.go +++ b/blobstore/clustermgr/cluster/shardnode_mgr.go @@ -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), })) } diff --git a/blobstore/clustermgr/cluster/shardnode_mgr_refresh.go b/blobstore/clustermgr/cluster/shardnode_mgr_refresh.go new file mode 100644 index 000000000..f99ed82f6 --- /dev/null +++ b/blobstore/clustermgr/cluster/shardnode_mgr_refresh.go @@ -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) +} diff --git a/blobstore/clustermgr/config_test.go b/blobstore/clustermgr/config_test.go index 1a98aaa33..106d24883 100644 --- a/blobstore/clustermgr/config_test.go +++ b/blobstore/clustermgr/config_test.go @@ -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) diff --git a/blobstore/clustermgr/handler.go b/blobstore/clustermgr/handler.go index 186dd3f9f..eee7fdd87 100644 --- a/blobstore/clustermgr/handler.go +++ b/blobstore/clustermgr/handler.go @@ -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") diff --git a/blobstore/clustermgr/manage.go b/blobstore/clustermgr/manage.go index f4215a726..37127ec6a 100644 --- a/blobstore/clustermgr/manage.go +++ b/blobstore/clustermgr/manage.go @@ -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) diff --git a/blobstore/clustermgr/persistence/normaldb/shardnodedisktbl.go b/blobstore/clustermgr/persistence/normaldb/shardnodedisktbl.go index 48080546e..3d640d91b 100644 --- a/blobstore/clustermgr/persistence/normaldb/shardnodedisktbl.go +++ b/blobstore/clustermgr/persistence/normaldb/shardnodedisktbl.go @@ -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 { diff --git a/blobstore/clustermgr/shardnode.go b/blobstore/clustermgr/shardnode.go new file mode 100644 index 000000000..2da7bf7b3 --- /dev/null +++ b/blobstore/clustermgr/shardnode.go @@ -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)) +} diff --git a/blobstore/clustermgr/shardnode_disk.go b/blobstore/clustermgr/shardnode_disk.go new file mode 100644 index 000000000..1fe36ca6f --- /dev/null +++ b/blobstore/clustermgr/shardnode_disk.go @@ -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 + } +} diff --git a/blobstore/clustermgr/svr.go b/blobstore/clustermgr/svr.go index 7ade6d534..fb80235e3 100644 --- a/blobstore/clustermgr/svr.go +++ b/blobstore/clustermgr/svr.go @@ -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 diff --git a/blobstore/clustermgr/svr_test.go b/blobstore/clustermgr/svr_test.go index 082b9feec..1fb51520f 100644 --- a/blobstore/clustermgr/svr_test.go +++ b/blobstore/clustermgr/svr_test.go @@ -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) diff --git a/blobstore/clustermgr/volume_test.go b/blobstore/clustermgr/volume_test.go index 506c17969..89fd37933 100644 --- a/blobstore/clustermgr/volume_test.go +++ b/blobstore/clustermgr/volume_test.go @@ -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() diff --git a/blobstore/clustermgr/volumemgr/volumemgr.go b/blobstore/clustermgr/volumemgr/volumemgr.go index e1a65a591..2028de920 100644 --- a/blobstore/clustermgr/volumemgr/volumemgr.go +++ b/blobstore/clustermgr/volumemgr/volumemgr.go @@ -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 {