gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix(clustermgr): use actual range count for shard init done check

Browse Source 
with: #1000979225

Signed-off-by: xiejian <xiejian3@oppo.com>
This commit is contained in:
xiejian 2026-05-18 19:02:24 +08:00
parent 612b656595
commit bb7e06259b
4 changed files with 118 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
blobstore/clustermgr/catalog/catalog.go
View File

@ -61,12 +61,13 @@ var (
)
type CatalogMgr struct {
module string
allShards *concurrentShards
allSpaces *concurrentSpaces
dirty atomic.Value
initShardDone int32
applyTaskPool *base.TaskDistribution
module string
allShards *concurrentShards
allSpaces *concurrentSpaces
dirty atomic.Value
initShardDone int32
actualInitShardNum int // actual range count generated by InitShardingRange, may exceed InitShardNum due to power-of-2 rounding
applyTaskPool *base.TaskDistribution
catalogTbl *catalogdb.CatalogTable
transitedTbl *catalogdb.TransitedTable

11
blobstore/clustermgr/catalog/createshard.go
View File

@ -43,12 +43,15 @@ func (c *CatalogMgr) createShard(ctx context.Context) error {
span := trace.SpanFromContextSafe(ctx)
// alloc shardID and shard range, and save initial shard and unit info into transited tbl
createShardCtxs := make([]createShardCtx, 0, c.InitShardNum)
minShardID, _, err := c.scopeMgr.Alloc(ctx, shardIDScopeName, c.InitShardNum)
ranges := sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, c.InitShardNum)
if len(ranges) != c.InitShardNum {
span.Warnf("InitShardNum %d expanded to %d due to sharding algorithm power-of-2 rounding", c.InitShardNum, len(ranges))
}
createShardCtxs := make([]createShardCtx, 0, len(ranges))
minShardID, _, err := c.scopeMgr.Alloc(ctx, shardIDScopeName, len(ranges))
if err != nil {
return errors.Info(err, "scope alloc shardID failed")
}
ranges := sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, c.InitShardNum)
unitCount := c.CodeMode.GetShardNum()
for i := 0; i < len(ranges); i++ {
shardID := proto.ShardID(minShardID + uint64(i))
@ -234,7 +237,7 @@ func (c *CatalogMgr) applyCreateShard(ctx context.Context, shard *shardItem) err
c.routeMgr.InsertRouteItems(ctx, []*base.RouteItem{route})
shard.info.RouteVersion = proto.RouteVersion(routeVersion)
if c.allShards.getShardNum() == c.InitShardNum {
if c.allShards.getShardNum() == c.actualInitShardNum {
err := c.kvMgr.Set(proto.ShardInitDoneKey, []byte("1"))
if err != nil {
return errors.Info(err, "put shard init done key to kv failed")

95
blobstore/clustermgr/catalog/createshard_test.go
View File

@ -17,7 +17,12 @@ package catalog
import (
"context"
"encoding/json"
"fmt"
"math/rand"
"os"
"path"
"testing"
"time"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/require"
@ -25,9 +30,11 @@ import (
"github.com/cubefs/cubefs/blobstore/api/clustermgr"
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
"github.com/cubefs/cubefs/blobstore/clustermgr/cluster"
"github.com/cubefs/cubefs/blobstore/clustermgr/persistence/catalogdb"
"github.com/cubefs/cubefs/blobstore/common/codemode"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/raftserver"
"github.com/cubefs/cubefs/blobstore/common/sharding"
"github.com/cubefs/cubefs/blobstore/common/trace"
mock "github.com/cubefs/cubefs/blobstore/testing/mockclustermgr"
"github.com/cubefs/cubefs/blobstore/testing/mocks"
@ -192,3 +199,91 @@ func TestCatalogMgr_applyCreateShard(t *testing.T) {
require.NoError(t, err)
require.Equal(t, expectedRouteVersion, shardRecord.RouteVersion)
}
// TestCatalogMgr_applyCreateShard_InitShardDone verifies that initShardDone is set only
// when the number of applied shards reaches actualInitShardNum (the real range count produced
// by InitShardingRange), not InitShardNum which may be smaller due to power-of-2 rounding.
func TestCatalogMgr_applyCreateShard_InitShardDone(t *testing.T) {
// Use InitShardNum=3 (odd): InitShardingRange rounds it up to 4 (next power of 2).
// Before the fix, initShardDone would trigger after 3 shards instead of 4.
conf := testConfig
conf.InitShardNum = 3
dir := path.Join(os.TempDir(), fmt.Sprintf("catalogmgr-initdone-%d-%010d", time.Now().Unix(), rand.Intn(100000000)))
catalogDBPath := path.Join(dir, "sharddb")
defer os.RemoveAll(dir)
catalogDB, err := catalogdb.Open(catalogDBPath)
require.NoError(t, err)
ctr := gomock.NewController(t)
mockDiskMgr := cluster.NewMockShardNodeManagerAPI(ctr)
mockScopeMgr := mock.NewMockScopeMgrAPI(ctr)
mockKvMgr := mock.NewMockKvMgrAPI(ctr)
mockDiskMgr.EXPECT().GetDiskInfo(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(mockGetDiskInfo)
// ShardInitDoneKey not found: initShardDone must NOT be set on startup
mockKvMgr.EXPECT().Get(gomock.Any()).Return(nil, fmt.Errorf("key not found"))
// Set is called exactly once when actualInitShardNum shards have been applied
mockKvMgr.EXPECT().Set(gomock.Any(), gomock.Any()).Times(1).Return(nil)
catalogMgr, err := NewCatalogMgr(conf, mockDiskMgr, mockScopeMgr, mockKvMgr, catalogDB)
require.NoError(t, err)
defer catalogMgr.Close()
_, ctx := trace.StartSpanFromContext(context.Background(), "")
// Initially, initShardDone must NOT be set
require.False(t, catalogMgr.IsShardInitDone(ctx))
// actualInitShardNum equals the real range count, which is greater than InitShardNum for odd values
actualNum := catalogMgr.actualInitShardNum
ranges := sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, conf.InitShardNum)
require.Equal(t, actualNum, len(ranges))
require.Greater(t, actualNum, conf.InitShardNum, "InitShardingRange should expand odd shardCount")
unitCount := conf.CodeMode.GetShardNum()
// Apply actualNum-1 shards: initShardDone must NOT be set at any intermediate step
for i := 0; i < actualNum-1; i++ {
shard := newTestShardItem(proto.ShardID(i+1), ranges[i], unitCount)
require.NoError(t, catalogMgr.applyCreateShard(ctx, shard))
require.False(t, catalogMgr.IsShardInitDone(ctx),
"initShardDone must not be set after only %d shards (need %d)", i+1, actualNum)
}
// Apply the last shard: initShardDone must now be set
lastShard := newTestShardItem(proto.ShardID(actualNum), ranges[actualNum-1], unitCount)
require.NoError(t, catalogMgr.applyCreateShard(ctx, lastShard))
require.True(t, catalogMgr.IsShardInitDone(ctx),
"initShardDone must be set after all %d shards are applied", actualNum)
}
// newTestShardItem constructs a minimal valid shardItem for use in tests.
func newTestShardItem(shardID proto.ShardID, rng *sharding.Range, unitCount int) *shardItem {
unitEpochs := make([]*shardUnitEpoch, unitCount)
units := make([]clustermgr.ShardUnit, unitCount)
for j := 0; j < unitCount; j++ {
suid := proto.EncodeSuid(shardID, uint8(j), proto.MinEpoch)
unitEpochs[j] = &shardUnitEpoch{
suidPrefix: suid.SuidPrefix(),
epoch: proto.MinEpoch,
nextEpoch: proto.MinEpoch,
}
units[j] = clustermgr.ShardUnit{
Suid: suid,
DiskID: proto.DiskID(j + 1),
}
}
return &shardItem{
shardID: shardID,
unitEpochs: unitEpochs,
info: shardInfoBase{
Shard: clustermgr.Shard{
ShardID: shardID,
Range: *rng,
Units: units,
},
},
}
}

13
blobstore/clustermgr/catalog/startup.go
View File

@ -28,6 +28,7 @@ import (
"github.com/cubefs/cubefs/blobstore/common/kvstore"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/raftserver"
"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"
@ -75,12 +76,16 @@ func NewCatalogMgr(conf Config, diskMgr cluster.ShardNodeManagerAPI, scopeMgr sc
return nil, errors.Info(err, "open transited table failed").Detail(err)
}
// Compute actual range count; InitShardingRange may round up to the next power of 2.
actualInitShardNum := len(sharding.InitShardingRange(sharding.RangeType_RangeTypeHash, 2, conf.InitShardNum))
// initial catalogMgr
catalogMgr := &CatalogMgr{
allShards: newConcurrentShards(conf.ShardConcurrentMapNum),
allSpaces: newConcurrentSpaces(conf.SpaceConcurrentMapNum),
catalogTbl: catalogTable,
transitedTbl: transitedTable,
allShards: newConcurrentShards(conf.ShardConcurrentMapNum),
allSpaces: newConcurrentSpaces(conf.SpaceConcurrentMapNum),
catalogTbl: catalogTable,
transitedTbl: transitedTable,
actualInitShardNum: actualInitShardNum,
applyTaskPool: base.NewTaskDistribution(int(conf.ApplyConcurrency), 1),
scopeMgr: scopeMgr,