mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
perf(clustermgr): alloc first choose volume which disks have more free space
with: #1000910653 Signed-off-by: JasonHu520 <huzongchao@oppo.com>
This commit is contained in:
JasonHu520
slasher
parent
c170c6c606
commit
f230337013
@ -47,6 +47,7 @@ func TestApplier_Others(t *testing.T) {
|
||||
|
||||
// test flush
|
||||
{
|
||||
testDiskMgr.RegisterDiskUsageCallback(func(proto.DiskID, float64) {})
|
||||
heartbeatInfos := make([]*clustermgr.DiskHeartBeatInfo, 0)
|
||||
for i := 1; i <= 10; i++ {
|
||||
diskInfo, err := testDiskMgr.GetDiskInfo(ctx, proto.DiskID(i))
|
||||
|
||||
@ -57,6 +57,8 @@ type BlobNodeManagerAPI interface {
|
||||
AllocChunks(ctx context.Context, policy AllocPolicy) ([]proto.DiskID, []proto.Vuid, error)
|
||||
// HasEnoughSpace reports whether the cluster currently has enough space to create
|
||||
HasEnoughSpace(ctx context.Context, mode codemode.CodeMode) bool
|
||||
// RegisterDiskUsageCallback registers a callback that is invoked on every disk heartbeat.
|
||||
RegisterDiskUsageCallback(fn func(diskID proto.DiskID, ratio float64))
|
||||
|
||||
NodeManagerAPI
|
||||
persistentHandler
|
||||
@ -147,6 +149,12 @@ type BlobNodeManager struct {
|
||||
|
||||
nodeDiskTable *normaldb.BlobNodeDiskTable
|
||||
blobNodeClient blobnode.StorageAPI
|
||||
|
||||
diskUsageCallback func(diskID proto.DiskID, ratio float64)
|
||||
}
|
||||
|
||||
func (b *BlobNodeManager) RegisterDiskUsageCallback(fn func(diskID proto.DiskID, ratio float64)) {
|
||||
b.diskUsageCallback = fn
|
||||
}
|
||||
|
||||
func (b *BlobNodeManager) Start() {
|
||||
@ -958,6 +966,10 @@ func (b *BlobNodeManager) applyHeartBeatDiskInfo(ctx context.Context, infos []*c
|
||||
return nil
|
||||
})
|
||||
|
||||
if info.Size > 0 && b.diskUsageCallback != nil {
|
||||
b.diskUsageCallback(info.DiskID, float64(info.Used)/float64(info.Size))
|
||||
}
|
||||
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -227,6 +227,7 @@ func TestDiskMgr_Heartbeat(t *testing.T) {
|
||||
defer closeTestDiskMgr()
|
||||
initTestBlobNodeMgrNodes(t, testDiskMgr, 1, 1, testIdcs[0])
|
||||
initTestBlobNodeMgrDisks(t, testDiskMgr, 1, 10, false, testIdcs[0])
|
||||
testDiskMgr.RegisterDiskUsageCallback(func(proto.DiskID, float64) {})
|
||||
_, ctx := trace.StartSpanFromContext(context.Background(), "")
|
||||
|
||||
heartbeatInfos := make([]*clustermgr.DiskHeartBeatInfo, 0)
|
||||
@ -974,6 +975,69 @@ func TestBlobNodeManager_ListNodes(t *testing.T) {
|
||||
require.NotContains(t, normalIDs, proto.NodeID(1)) // node 1 is Dropped
|
||||
}
|
||||
|
||||
func TestBlobNodeMgr_DiskUsageListener(t *testing.T) {
|
||||
mgr, closeMgr := initTestBlobNodeMgr(t)
|
||||
defer closeMgr()
|
||||
_, ctx := trace.StartSpanFromContext(context.Background(), "")
|
||||
|
||||
initTestBlobNodeMgrNodes(t, mgr, 1, 1, testIdcs[0])
|
||||
initTestBlobNodeMgrDisks(t, mgr, 1, 1, true, testIdcs[0])
|
||||
|
||||
const diskID = proto.DiskID(1)
|
||||
const total = int64(100 * 1024 * 1024 * 1024) // 100 GiB
|
||||
|
||||
var lastID proto.DiskID
|
||||
var lastRatio float64
|
||||
mgr.RegisterDiskUsageCallback(func(id proto.DiskID, ratio float64) {
|
||||
lastID = id
|
||||
lastRatio = ratio
|
||||
})
|
||||
const watermark = 0.8
|
||||
|
||||
sendHeartbeat := func(used int64) {
|
||||
infos := []*clustermgr.DiskHeartBeatInfo{{
|
||||
DiskID: diskID,
|
||||
Size: total,
|
||||
Used: used,
|
||||
}}
|
||||
require.NoError(t, mgr.applyHeartBeatDiskInfo(ctx, infos))
|
||||
}
|
||||
|
||||
t.Run("unknown disk skips listener", func(t *testing.T) {
|
||||
lastRatio = 0
|
||||
require.NoError(t, mgr.applyHeartBeatDiskInfo(ctx, []*clustermgr.DiskHeartBeatInfo{
|
||||
{DiskID: proto.DiskID(9999), Size: total, Free: total / 2},
|
||||
}))
|
||||
require.Equal(t, proto.DiskID(0), lastID)
|
||||
})
|
||||
|
||||
t.Run("zero-size disk skips callback", func(t *testing.T) {
|
||||
prevID := lastID
|
||||
require.NoError(t, mgr.applyHeartBeatDiskInfo(ctx, []*clustermgr.DiskHeartBeatInfo{
|
||||
{DiskID: diskID, Size: 0, Used: 0},
|
||||
}))
|
||||
require.Equal(t, prevID, lastID)
|
||||
})
|
||||
|
||||
t.Run("below watermark notifies ratio < threshold", func(t *testing.T) {
|
||||
sendHeartbeat(79 * 1024 * 1024 * 1024) // 79%
|
||||
require.Equal(t, diskID, lastID)
|
||||
require.Less(t, lastRatio, watermark)
|
||||
})
|
||||
|
||||
t.Run("at watermark boundary notifies ratio >= threshold", func(t *testing.T) {
|
||||
sendHeartbeat(80 * 1024 * 1024 * 1024) // exactly 80%
|
||||
require.Equal(t, diskID, lastID)
|
||||
require.GreaterOrEqual(t, lastRatio, watermark)
|
||||
})
|
||||
|
||||
t.Run("above watermark notifies ratio >= threshold", func(t *testing.T) {
|
||||
sendHeartbeat(95 * 1024 * 1024 * 1024) // 95%
|
||||
require.Equal(t, diskID, lastID)
|
||||
require.GreaterOrEqual(t, lastRatio, watermark)
|
||||
})
|
||||
}
|
||||
|
||||
// TestDiskMgr_DroppingNode covers applyDroppingNode/applyDroppedNode corner cases:
|
||||
// node not found, dropping propagation, and dropped without disks.
|
||||
func TestDiskMgr_DroppingNode(t *testing.T) {
|
||||
|
||||
@ -198,6 +198,18 @@ func (mr *MockBlobNodeManagerAPIMockRecorder) HasEnoughSpace(arg0, arg1 interfac
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HasEnoughSpace", reflect.TypeOf((*MockBlobNodeManagerAPI)(nil).HasEnoughSpace), arg0, arg1)
|
||||
}
|
||||
|
||||
// RegisterDiskUsageCallback mocks base method.
|
||||
func (m *MockBlobNodeManagerAPI) RegisterDiskUsageCallback(fn func(proto.DiskID, float64)) {
|
||||
m.ctrl.T.Helper()
|
||||
m.ctrl.Call(m, "RegisterDiskUsageCallback", fn)
|
||||
}
|
||||
|
||||
// RegisterDiskUsageCallback indicates an expected call of RegisterDiskUsageCallback.
|
||||
func (mr *MockBlobNodeManagerAPIMockRecorder) RegisterDiskUsageCallback(fn interface{}) *gomock.Call {
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterDiskUsageCallback", reflect.TypeOf((*MockBlobNodeManagerAPI)(nil).RegisterDiskUsageCallback), fn)
|
||||
}
|
||||
|
||||
// IsDiskWritable mocks base method.
|
||||
func (m *MockBlobNodeManagerAPI) IsDiskWritable(arg0 context.Context, arg1 proto.DiskID) (bool, error) {
|
||||
m.ctrl.T.Helper()
|
||||
|
||||
@ -27,6 +27,7 @@ import (
|
||||
"github.com/cubefs/cubefs/blobstore/clustermgr/base"
|
||||
"github.com/cubefs/cubefs/blobstore/common/trace"
|
||||
mock "github.com/cubefs/cubefs/blobstore/testing/mockclustermgr"
|
||||
"github.com/cubefs/cubefs/blobstore/util/errors"
|
||||
)
|
||||
|
||||
func TestConfigMgr_Others(t *testing.T) {
|
||||
@ -50,10 +51,20 @@ func TestConfigMgr_Others(t *testing.T) {
|
||||
module := configmgr.GetModuleName()
|
||||
require.Equal(t, testModuleName, module)
|
||||
|
||||
err = configmgr.LoadData(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
configmgr.SetRaftServer(nil)
|
||||
|
||||
err = configmgr.Flush(ctx)
|
||||
require.NoError(t, err)
|
||||
|
||||
configmgr.NotifyLeaderChange(ctx, 0, "")
|
||||
|
||||
// cover Get path when key is not in DB or default config
|
||||
mockKvMgr.EXPECT().Get("unknown_key").Return(nil, errors.New("not found"))
|
||||
_, err = configmgr.Get(ctx, "unknown_key")
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
func TestConfigMgr_Apply(t *testing.T) {
|
||||
@ -144,4 +155,28 @@ func TestConfigMgr_Apply(t *testing.T) {
|
||||
|
||||
err = configmgr.Apply(ctx, operTypes, datas, ctxs)
|
||||
require.NoError(t, err)
|
||||
|
||||
// OperTypeSetConfig: Set returns error
|
||||
{
|
||||
ctr2 := gomock.NewController(t)
|
||||
mockKvMgr2 := mock.NewMockKvMgrAPI(ctr2)
|
||||
mockKvMgr2.EXPECT().Set(gomock.Any(), gomock.Any()).Return(errors.New("set error"))
|
||||
cm2, err := New(mockKvMgr2, cfMap)
|
||||
require.NoError(t, err)
|
||||
data, _ := json.Marshal(&clustermgr.ConfigSetArgs{Key: "forbid_sync_config", Value: "true"})
|
||||
err = cm2.Apply(ctx, []int32{OperTypeSetConfig}, [][]byte{data}, []base.ProposeContext{{ReqID: span.TraceID()}})
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
// OperTypeDeleteConfig: Delete returns error
|
||||
{
|
||||
ctr3 := gomock.NewController(t)
|
||||
mockKvMgr3 := mock.NewMockKvMgrAPI(ctr3)
|
||||
mockKvMgr3.EXPECT().Delete(gomock.Any()).Return(errors.New("delete error"))
|
||||
cm3, err := New(mockKvMgr3, cfMap)
|
||||
require.NoError(t, err)
|
||||
data, _ := json.Marshal(&clustermgr.ConfigArgs{Key: "forbid_sync_config"})
|
||||
err = cm3.Apply(ctx, []int32{OperTypeDeleteConfig}, [][]byte{data}, []base.ProposeContext{{ReqID: span.TraceID()}})
|
||||
require.Error(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
@ -43,11 +43,17 @@ func TestConfigMgr(t *testing.T) {
|
||||
"enable_recycle": false,
|
||||
"idc": []interface{}{"idc1", "idc2"},
|
||||
"test_map1": map[string]interface{}{"a": 1, "b": 2},
|
||||
"region": "cn-south",
|
||||
}
|
||||
|
||||
configMgr, err := New(kvMgr, cfMap)
|
||||
require.NoError(t, err)
|
||||
|
||||
// string default values are stored directly
|
||||
regionVal, err := configMgr.Get(ctx, "region")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "cn-south", regionVal)
|
||||
|
||||
ret, _ := configMgr.Get(ctx, "enable_recycle")
|
||||
var enableRec bool
|
||||
json.Unmarshal([]byte(ret), &enableRec)
|
||||
@ -67,4 +73,19 @@ func TestConfigMgr(t *testing.T) {
|
||||
var vv3 map[string]interface{}
|
||||
json.Unmarshal([]byte(idcRet2), &vv3)
|
||||
require.Equal(t, map[string]interface{}{"a": float64(1), "b": float64(2)}, vv3)
|
||||
|
||||
// cover Get path that returns value directly from DB
|
||||
err = configMgr.Set(ctx, "db_only_key", "db_value")
|
||||
require.NoError(t, err)
|
||||
dbVal, err := configMgr.Get(ctx, "db_only_key")
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, "db_value", dbVal)
|
||||
|
||||
// cover Get path that returns error for empty key
|
||||
_, err = configMgr.Get(ctx, "")
|
||||
require.Error(t, err)
|
||||
|
||||
// cover Get path when key is not in DB or default config
|
||||
_, err = configMgr.Get(ctx, "totally_nonexistent_key")
|
||||
require.Error(t, err)
|
||||
}
|
||||
|
||||
@ -39,6 +39,7 @@ type allocConfig struct {
|
||||
allocatableSize uint64
|
||||
codeModes map[codemode.CodeMode]codeModeConf
|
||||
shardNum int
|
||||
diskUsageThreshold float64
|
||||
}
|
||||
|
||||
type idleItem struct {
|
||||
@ -136,16 +137,23 @@ type activeVolumes struct {
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
// volume allocator, use for allocating volume
|
||||
type volumeAllocator struct {
|
||||
// idle volumes
|
||||
idles map[codemode.CodeMode]*idleVolumes
|
||||
// actives volumes
|
||||
actives *activeVolumes
|
||||
// diskHighUsage holds the latest high-watermark state per disk.
|
||||
diskHighUsage sync.Map // map[proto.DiskID]bool
|
||||
|
||||
allocConfig
|
||||
}
|
||||
|
||||
// UpdateDiskHighUsage is called on every disk heartbeat to record whether a
|
||||
// disk currently exceeds the high-usage threshold.
|
||||
func (a *volumeAllocator) UpdateDiskHighUsage(diskID proto.DiskID, ratio float64) {
|
||||
a.diskHighUsage.Store(diskID, ratio > a.diskUsageThreshold)
|
||||
}
|
||||
|
||||
type sortVid []vidLoad
|
||||
|
||||
type vidLoad struct {
|
||||
@ -240,11 +248,11 @@ func (a *volumeAllocator) Insert(v *volume, mode codemode.CodeMode) {
|
||||
a.idles[mode].addAllocatable(v)
|
||||
}
|
||||
|
||||
// PreAlloc select volumes which can alloc
|
||||
// 1. when EnableDiskLoad=false, all volume will range by health, the healthier volume will range in front of the optional head
|
||||
// 2. when EnableDiskLoad=true, if do not hash enough volumes to alloc ,
|
||||
// 1. first add disk's load and retry, each time add one until disk's load equal to diskLoadThreshold will set EnableDiskLoad=false
|
||||
// 2. second minus volume score and retry , each time minus one until volume's score equal to scoreThreshold
|
||||
// PreAlloc pre-allocates up to count volumes for the given code mode.
|
||||
// When a.diskChecker is set, volumes with any high-watermark disk are deferred
|
||||
// to a later retry pass so that volumes on emptier disks are preferred.
|
||||
// The fallback guarantees service continuity: when no low usage disk candidates
|
||||
// remain, high disk usage volumes are included automatically.
|
||||
func (a *volumeAllocator) PreAlloc(ctx context.Context, mode codemode.CodeMode, count int) ([]proto.Vid, int) {
|
||||
span := trace.SpanFromContextSafe(ctx)
|
||||
idleVolumes := a.idles[mode]
|
||||
@ -260,6 +268,9 @@ func (a *volumeAllocator) PreAlloc(ctx context.Context, mode codemode.CodeMode,
|
||||
scoreThreshold := healthiestScore
|
||||
// diskLoadThreshold start half of allocatableDiskLoadThreshold,avoid loop too much times
|
||||
diskLoadThreshold := a.allocatableDiskLoadThreshold / 2
|
||||
// skipWatermark becomes true after disk-load is fully relaxed, allowing
|
||||
// volumes on high-watermark disks to be included as a last-resort fallback.
|
||||
skipDiskUsageCheck := a.diskUsageThreshold <= 0
|
||||
// optionalVids include all volume id which satisfied with our condition(idle/enough free size/health/not over disk load)
|
||||
// all vid will range by health, the healthier volume will range in front of the optional head
|
||||
optionalVids := make([]proto.Vid, 0)
|
||||
@ -281,7 +292,10 @@ RETRY:
|
||||
now := time.Now()
|
||||
for idx, volume := range allIdles {
|
||||
volume.lock.RLock()
|
||||
if volume.canAlloc(a.allocatableSize, scoreThreshold) && (!isEnableDiskLoad || !a.isOverload(volume.vUnits, diskLoadThreshold)) {
|
||||
hasHighUsageDisk := !skipDiskUsageCheck && a.hasHighUsageDisk(volume.vUnits, mode)
|
||||
if volume.canAlloc(a.allocatableSize, scoreThreshold) &&
|
||||
(!isEnableDiskLoad || !a.isOverload(volume.vUnits, diskLoadThreshold)) &&
|
||||
!hasHighUsageDisk {
|
||||
optionalVids = append(optionalVids, volume.vid)
|
||||
// only insufficient free size or unhealthy volume move to temporary head,
|
||||
// ignore over diskLoad volume
|
||||
@ -296,8 +310,10 @@ RETRY:
|
||||
break
|
||||
}
|
||||
|
||||
// go to the end, first retry with high disk load volume
|
||||
// second lower health score volume
|
||||
// Relaxation order on retry:
|
||||
// 1. raise diskLoadThreshold step by step
|
||||
// 2. disable disk-load check entirely and relax disk high usage filter (skipDiskCheck = true)
|
||||
// 3. lower scoreThreshold
|
||||
if isLastShard && idx == len(allIdles)-1 {
|
||||
span.Infof("assignable volume length is %d", len(assignable))
|
||||
if len(assignable) == 0 {
|
||||
@ -307,10 +323,14 @@ RETRY:
|
||||
if isEnableDiskLoad && diskLoadThreshold < a.allocatableDiskLoadThreshold {
|
||||
// When diskLoad exceeds the threshold, retry 3 times at most
|
||||
diskLoadThreshold += int(math.Ceil(float64(a.allocatableDiskLoadThreshold) / 6.0))
|
||||
} else if isEnableDiskLoad {
|
||||
span.Infof("increase diskload to %d", diskLoadThreshold)
|
||||
} else if !skipDiskUsageCheck || isEnableDiskLoad {
|
||||
skipDiskUsageCheck = true
|
||||
isEnableDiskLoad = false
|
||||
span.Info("close diskload and disk usage check")
|
||||
} else if scoreThreshold > allocatableScoreThreshold {
|
||||
scoreThreshold -= 1
|
||||
span.Warnf("lower volume health score %d", scoreThreshold)
|
||||
}
|
||||
allIdles = assignable
|
||||
assignable = assignable[:0]
|
||||
@ -427,6 +447,20 @@ func (a *volumeAllocator) getShardNum(mode codemode.CodeMode) int {
|
||||
return modeConf.tactic.N + modeConf.tactic.M + modeConf.tactic.L
|
||||
}
|
||||
|
||||
// hasHighUsageDisk reports whether any disk in vUnits exceeds the high usage threshold.
|
||||
func (a *volumeAllocator) hasHighUsageDisk(vUnits []*volumeUnit, mode codemode.CodeMode) bool {
|
||||
count := mode.GetShardNum() - mode.T().PutQuorum
|
||||
for _, unit := range vUnits {
|
||||
if v, ok := a.diskHighUsage.Load(unit.vuInfo.DiskID); ok && v.(bool) {
|
||||
count = count - 1
|
||||
if count < 0 {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (a *volumeAllocator) sortVidByHealthAndDiskLoad(mode codemode.CodeMode, vids []proto.Vid) (ret []proto.Vid) {
|
||||
if len(vids) <= 1 {
|
||||
return vids
|
||||
|
||||
@ -54,6 +54,8 @@ type VolumeMgrConfig struct {
|
||||
MinWritableVolumeSpace uint64 `json:"min_writable_volume_space"`
|
||||
AllocatableDiskLoadThreshold int `json:"allocatable_disk_load_threshold"`
|
||||
AllocFactor int `json:"alloc_factor"`
|
||||
|
||||
AllocDiskUsageThreshold float64 `json:"alloc_disk_usage_threshold"`
|
||||
// the volume free size must big than AllocatableSize can alloc
|
||||
AllocatableSize uint64 `json:"allocatable_size"`
|
||||
// the number of volume partitions that can be allocated
|
||||
@ -171,8 +173,11 @@ func NewVolumeMgr(conf VolumeMgrConfig, diskMgr cluster.BlobNodeManagerAPI, scop
|
||||
allocFactor: conf.AllocFactor,
|
||||
allocatableDiskLoadThreshold: conf.AllocatableDiskLoadThreshold,
|
||||
shardNum: conf.ShardNum,
|
||||
diskUsageThreshold: conf.AllocDiskUsageThreshold,
|
||||
}
|
||||
volAllocator := newVolumeAllocator(allocConfig)
|
||||
diskMgr.RegisterDiskUsageCallback(volAllocator.UpdateDiskHighUsage)
|
||||
|
||||
volumeMgr.allocator = volAllocator
|
||||
|
||||
// initial register change status callback func
|
||||
|
||||
@ -103,6 +103,7 @@ func initMockVolumeMgr(t testing.TB) (*VolumeMgr, func()) {
|
||||
mockDiskMgr.EXPECT().Stat(gomock.Any(), proto.DiskTypeHDD).AnyTimes().Return(&clustermgr.SpaceStatInfo{TotalDisk: 35})
|
||||
mockDiskMgr.EXPECT().IsDiskWritable(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(mockIsDiskWritable)
|
||||
mockDiskMgr.EXPECT().GetDiskInfo(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(mockGetDiskInfo)
|
||||
mockDiskMgr.EXPECT().RegisterDiskUsageCallback(gomock.Any()).Times(1)
|
||||
|
||||
mockVolumeMgr, err := NewVolumeMgr(testConfig, mockDiskMgr, mockScopeMgr, mockConfigMgr, volumeDB)
|
||||
require.NoError(t, err)
|
||||
@ -317,6 +318,7 @@ func Test_NewVolumeMgr(t *testing.T) {
|
||||
mockDiskMgr.EXPECT().IsDiskWritable(gomock.Any(), gomock.Any()).AnyTimes().Return(true, nil)
|
||||
mockDiskMgr.EXPECT().GetDiskInfo(gomock.Any(), gomock.Any()).AnyTimes().DoAndReturn(mockGetDiskInfo)
|
||||
mockDiskMgr.EXPECT().HasEnoughSpace(gomock.Any(), gomock.Any()).AnyTimes().Return(true)
|
||||
mockDiskMgr.EXPECT().RegisterDiskUsageCallback(gomock.Any()).Times(1)
|
||||
|
||||
mockVolumeMgr, err := NewVolumeMgr(volConfig, mockDiskMgr, mockScopeMgr, mockConfigMgr, volumeDB)
|
||||
require.NoError(t, err)
|
||||
@ -1201,6 +1203,166 @@ func BenchmarkVolumeMgr_AllocVolume(b *testing.B) {
|
||||
})
|
||||
}
|
||||
|
||||
// TestPreAlloc_HighWatermarkFallback verifies the two-pass watermark behavior:
|
||||
// 1. When allocating 2 volumes from a pool of 4 high-watermark + 2 low-watermark
|
||||
// volumes (all other conditions equal), the 2 low-watermark volumes are returned.
|
||||
// 2. When allocating all 6, the full set is returned via graceful fallback.
|
||||
func TestPreAlloc_HighWatermarkFallback(t *testing.T) {
|
||||
testConfig.checkAndFix()
|
||||
codeModes := make(map[codemode.CodeMode]codeModeConf)
|
||||
for _, policy := range testConfig.CodeModePolicies {
|
||||
cm := policy.ModeName.GetCodeMode()
|
||||
codeModes[cm] = codeModeConf{
|
||||
mode: cm,
|
||||
sizeRatio: policy.SizeRatio,
|
||||
tactic: cm.Tactic(),
|
||||
enable: policy.Enable,
|
||||
}
|
||||
}
|
||||
mode := codemode.EC15P12
|
||||
|
||||
newAllocator := func() *volumeAllocator {
|
||||
a := newVolumeAllocator(allocConfig{
|
||||
codeModes: codeModes,
|
||||
allocatableSize: testConfig.AllocatableSize,
|
||||
allocFactor: testConfig.AllocFactor,
|
||||
allocatableDiskLoadThreshold: testConfig.AllocatableDiskLoadThreshold,
|
||||
shardNum: testConfig.ShardNum,
|
||||
diskUsageThreshold: 0.85,
|
||||
})
|
||||
// vid 1..4: high-watermark disks (ID <= 1000); vid 5..6: low-watermark disks (ID > 1000).
|
||||
const watermarkBoundary = proto.DiskID(1000)
|
||||
allVols := generateVolume(mode, 6, 1)
|
||||
for i, vol := range allVols {
|
||||
for j := range vol.vUnits {
|
||||
if i < 4 {
|
||||
vol.vUnits[j].vuInfo.DiskID = proto.DiskID(j + 1)
|
||||
} else {
|
||||
vol.vUnits[j].vuInfo.DiskID = watermarkBoundary + proto.DiskID(j+1)
|
||||
}
|
||||
}
|
||||
vol.volInfoBase.Status = proto.VolumeStatusIdle
|
||||
a.idles[mode].addAllocatable(vol)
|
||||
}
|
||||
// mark disks with ID <= watermarkBoundary as high-usage via the heartbeat-driven path.
|
||||
for _, vol := range allVols {
|
||||
for _, unit := range vol.vUnits {
|
||||
var ratio float64
|
||||
if unit.vuInfo.DiskID <= watermarkBoundary {
|
||||
ratio = 0.9 // above threshold
|
||||
} else {
|
||||
ratio = 0.5 // below threshold
|
||||
}
|
||||
a.UpdateDiskHighUsage(unit.vuInfo.DiskID, ratio)
|
||||
}
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
_, ctx := trace.StartSpanFromContext(context.Background(), "")
|
||||
lowWatermarkVids := map[proto.Vid]struct{}{5: {}, 6: {}}
|
||||
|
||||
t.Run("prefer low-watermark volumes", func(t *testing.T) {
|
||||
ret, _ := newAllocator().PreAlloc(ctx, mode, 2)
|
||||
require.Equal(t, 2, len(ret))
|
||||
for _, vid := range ret {
|
||||
require.Contains(t, lowWatermarkVids, vid, "expected only low-watermark vids, got vid=%d", vid)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("fallback allocates all volumes", func(t *testing.T) {
|
||||
ret, _ := newAllocator().PreAlloc(ctx, mode, 6)
|
||||
require.Equal(t, 6, len(ret))
|
||||
})
|
||||
|
||||
t.Run("disabled when threshold is zero", func(t *testing.T) {
|
||||
// diskUsageThreshold=0 → skipDiskUsageCheck=true from the start;
|
||||
// even disks at 90% usage are eligible without fallback.
|
||||
a := newVolumeAllocator(allocConfig{
|
||||
codeModes: codeModes,
|
||||
allocatableSize: testConfig.AllocatableSize,
|
||||
allocFactor: testConfig.AllocFactor,
|
||||
allocatableDiskLoadThreshold: testConfig.AllocatableDiskLoadThreshold,
|
||||
shardNum: testConfig.ShardNum,
|
||||
diskUsageThreshold: 0,
|
||||
})
|
||||
vols := generateVolume(mode, 4, 1)
|
||||
for _, vol := range vols {
|
||||
vol.volInfoBase.Status = proto.VolumeStatusIdle
|
||||
a.idles[mode].addAllocatable(vol)
|
||||
for _, unit := range vol.vUnits {
|
||||
a.UpdateDiskHighUsage(unit.vuInfo.DiskID, 0.9)
|
||||
}
|
||||
}
|
||||
ret, _ := a.PreAlloc(ctx, mode, 2)
|
||||
require.Equal(t, 2, len(ret))
|
||||
})
|
||||
|
||||
t.Run("ratio at threshold is not high", func(t *testing.T) {
|
||||
// ratio > threshold is strict; ratio == threshold must NOT be treated as high.
|
||||
const threshold = 0.85
|
||||
a := newVolumeAllocator(allocConfig{
|
||||
codeModes: codeModes,
|
||||
allocatableSize: testConfig.AllocatableSize,
|
||||
allocFactor: testConfig.AllocFactor,
|
||||
allocatableDiskLoadThreshold: testConfig.AllocatableDiskLoadThreshold,
|
||||
shardNum: testConfig.ShardNum,
|
||||
diskUsageThreshold: threshold,
|
||||
})
|
||||
vols := generateVolume(mode, 4, 1)
|
||||
for _, vol := range vols {
|
||||
vol.volInfoBase.Status = proto.VolumeStatusIdle
|
||||
a.idles[mode].addAllocatable(vol)
|
||||
for _, unit := range vol.vUnits {
|
||||
a.UpdateDiskHighUsage(unit.vuInfo.DiskID, threshold)
|
||||
}
|
||||
}
|
||||
ret, _ := a.PreAlloc(ctx, mode, 2)
|
||||
require.Equal(t, 2, len(ret))
|
||||
})
|
||||
|
||||
t.Run("disk usage update overrides previous state", func(t *testing.T) {
|
||||
// EC15P12: count = N+M+L-PutQuorum = 27-24 = 3 (tolerance).
|
||||
// hasHighUsageDisk returns true only when > count (i.e. ≥ 4) disks are high
|
||||
// (count<0 condition), meaning fewer than PutQuorum disks remain available.
|
||||
const threshold = 0.85
|
||||
const (
|
||||
diskA = proto.DiskID(9001)
|
||||
diskB = proto.DiskID(9002)
|
||||
diskC = proto.DiskID(9003)
|
||||
diskD = proto.DiskID(9004)
|
||||
)
|
||||
a := newVolumeAllocator(allocConfig{
|
||||
codeModes: codeModes,
|
||||
allocatableSize: testConfig.AllocatableSize,
|
||||
allocFactor: testConfig.AllocFactor,
|
||||
allocatableDiskLoadThreshold: testConfig.AllocatableDiskLoadThreshold,
|
||||
shardNum: testConfig.ShardNum,
|
||||
diskUsageThreshold: threshold,
|
||||
})
|
||||
units := []*volumeUnit{
|
||||
{vuInfo: &clustermgr.VolumeUnitInfo{DiskID: diskA}},
|
||||
{vuInfo: &clustermgr.VolumeUnitInfo{DiskID: diskB}},
|
||||
{vuInfo: &clustermgr.VolumeUnitInfo{DiskID: diskC}},
|
||||
{vuInfo: &clustermgr.VolumeUnitInfo{DiskID: diskD}},
|
||||
}
|
||||
|
||||
// 3 high disks == count(3): count reaches 0 but not <0, still allowed.
|
||||
a.UpdateDiskHighUsage(diskA, 0.9)
|
||||
a.UpdateDiskHighUsage(diskB, 0.9)
|
||||
a.UpdateDiskHighUsage(diskC, 0.9)
|
||||
require.False(t, a.hasHighUsageDisk(units, mode))
|
||||
|
||||
// 4th disk also high → count goes negative → blocked.
|
||||
a.UpdateDiskHighUsage(diskD, 0.9)
|
||||
require.True(t, a.hasHighUsageDisk(units, mode))
|
||||
|
||||
// Drop diskA below threshold: only 3 of 4 remain high → unblocked.
|
||||
a.UpdateDiskHighUsage(diskA, 0.3)
|
||||
require.False(t, a.hasHighUsageDisk(units, mode))
|
||||
})
|
||||
}
|
||||
|
||||
func BenchmarkVolumeMgr_PreAllocVolume(b *testing.B) {
|
||||
_, ctx := trace.StartSpanFromContext(context.Background(), "")
|
||||
mode := codemode.EC15P12
|
||||
|
||||
@ -192,6 +192,24 @@ func newTestWorkerTaskQueue(cancelPunishDuration, renewDuration time.Duration) *
|
||||
}
|
||||
}
|
||||
|
||||
// mockShardTask implements ShardTask for testing.
|
||||
type mockShardTask struct {
|
||||
source proto.ShardUnitInfoSimple
|
||||
leader proto.ShardUnitInfoSimple
|
||||
destination proto.ShardUnitInfoSimple
|
||||
badDest proto.ShardUnitInfoSimple
|
||||
}
|
||||
|
||||
func (t *mockShardTask) GetSource() proto.ShardUnitInfoSimple { return t.source }
|
||||
func (t *mockShardTask) GetLeader() proto.ShardUnitInfoSimple { return t.leader }
|
||||
func (t *mockShardTask) GetDestination() proto.ShardUnitInfoSimple { return t.destination }
|
||||
func (t *mockShardTask) SetDestination(d proto.ShardUnitInfoSimple) {
|
||||
t.destination = d
|
||||
}
|
||||
func (t *mockShardTask) SetLeader(l proto.ShardUnitInfoSimple) { t.leader = l }
|
||||
func (t *mockShardTask) GetBadDestination() proto.ShardUnitInfoSimple { return t.badDest }
|
||||
func (t *mockShardTask) ToTask() (*proto.Task, error) { return nil, nil }
|
||||
|
||||
func TestWorkerTaskQueue(t *testing.T) {
|
||||
taskID1 := "task_id1"
|
||||
idc := "z0"
|
||||
@ -272,3 +290,190 @@ func TestWorkerTaskQueue(t *testing.T) {
|
||||
_, err = wq.Complete(idc, taskID2, vunits([]proto.Vuid{4, 5, 6}), vunit(4))
|
||||
require.EqualError(t, err, ErrUnmatchedVuids.Error())
|
||||
}
|
||||
|
||||
func TestWorkerTaskQueueMissingPaths(t *testing.T) {
|
||||
idc := "z0"
|
||||
noSuchIdc := "z99"
|
||||
cancelPunishDuration := 100 * time.Millisecond
|
||||
renewDuration := 200 * time.Millisecond
|
||||
taskID := "task_miss_1"
|
||||
task := mockWorkerTask{src: vunits([]proto.Vuid{1, 2, 3}), dst: vunit(4)}
|
||||
|
||||
wq := newTestWorkerTaskQueue(cancelPunishDuration, renewDuration)
|
||||
wq.AddPreparedTask(idc, taskID, &task)
|
||||
|
||||
// Acquire task to put it into doing state
|
||||
_, _, exist := wq.Acquire(idc)
|
||||
require.True(t, exist)
|
||||
|
||||
// Update: success
|
||||
updatedTask := mockWorkerTask{src: vunits([]proto.Vuid{1, 2, 3}), dst: vunit(99)}
|
||||
err := wq.Update(idc, taskID, &updatedTask)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Update: no such IDC
|
||||
err = wq.Update(noSuchIdc, taskID, &updatedTask)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Query: success
|
||||
wt, err := wq.Query(idc, taskID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, vunit(99), wt.GetDestination())
|
||||
|
||||
// Query: no such IDC
|
||||
_, err = wq.Query(noSuchIdc, taskID)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// SetLeaseExpiredS
|
||||
wq.SetLeaseExpiredS(500 * time.Millisecond)
|
||||
require.Equal(t, 500*time.Millisecond, wq.leaseExpiredS)
|
||||
|
||||
// Cancel: no such IDC
|
||||
err = wq.Cancel(noSuchIdc, taskID, task.GetSources(), task.GetDestination())
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Reclaim: no such IDC
|
||||
err = wq.Reclaim(noSuchIdc, taskID, task.GetSources(), task.GetDestination(), vunit(5), 0)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Renewal: no such IDC
|
||||
err = wq.Renewal(noSuchIdc, taskID)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Complete: no such IDC
|
||||
_, err = wq.Complete(noSuchIdc, taskID, task.GetSources(), task.GetDestination())
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
}
|
||||
|
||||
func TestQueueUpdateNotDoing(t *testing.T) {
|
||||
q := NewQueue(200 * time.Millisecond)
|
||||
msgID := "msg_todo"
|
||||
require.NoError(t, q.Push(msgID, "data"))
|
||||
|
||||
// Update on a task in todo state (not doing) should fail
|
||||
err := q.Update(msgID, "new_data")
|
||||
require.EqualError(t, err, errTaskStateNotDoing.Error())
|
||||
}
|
||||
|
||||
func TestShardTaskQueue(t *testing.T) {
|
||||
idc := "z0"
|
||||
noSuchIdc := "z99"
|
||||
cancelPunish := 100 * time.Millisecond
|
||||
|
||||
src := proto.ShardUnitInfoSimple{Suid: proto.EncodeSuid(1, 0, 1), DiskID: 1}
|
||||
dst := proto.ShardUnitInfoSimple{Suid: proto.EncodeSuid(1, 1, 1), DiskID: 2}
|
||||
newDst := proto.ShardUnitInfoSimple{Suid: proto.EncodeSuid(1, 1, 2), DiskID: 3}
|
||||
leader := proto.ShardUnitInfoSimple{Suid: proto.EncodeSuid(1, 2, 1), DiskID: 4}
|
||||
taskID := "shard_task_1"
|
||||
|
||||
task := &mockShardTask{source: src, destination: dst, leader: leader}
|
||||
|
||||
q := NewShardTaskQueue(cancelPunish)
|
||||
|
||||
// AddPreparedTask: first add creates idc queue
|
||||
q.AddPreparedTask(idc, taskID, task)
|
||||
|
||||
// AddPreparedTask: panic on duplicate
|
||||
require.Panics(t, func() {
|
||||
q.AddPreparedTask(idc, taskID, task)
|
||||
})
|
||||
|
||||
// Acquire: no such IDC returns false
|
||||
_, _, exist := q.Acquire(noSuchIdc)
|
||||
require.False(t, exist)
|
||||
|
||||
// Acquire: success
|
||||
id, wt, exist := q.Acquire(idc)
|
||||
require.True(t, exist)
|
||||
require.Equal(t, taskID, id)
|
||||
require.Equal(t, src, wt.GetSource())
|
||||
require.Equal(t, dst, wt.GetDestination())
|
||||
|
||||
// Acquire: queue empty (task is in doing)
|
||||
_, _, exist = q.Acquire(idc)
|
||||
require.False(t, exist)
|
||||
|
||||
// StatsTasks
|
||||
todo, doing := q.StatsTasks()
|
||||
require.Equal(t, 0, todo)
|
||||
require.Equal(t, 1, doing)
|
||||
|
||||
// Cancel: no such IDC
|
||||
err := q.Cancel(noSuchIdc, taskID, src, dst)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Cancel: unmatched suid
|
||||
err = q.Cancel(idc, taskID, src, proto.ShardUnitInfoSimple{DiskID: 99})
|
||||
require.EqualError(t, err, ErrUnmatchedSuid.Error())
|
||||
|
||||
// Cancel: success
|
||||
err = q.Cancel(idc, taskID, src, dst)
|
||||
require.NoError(t, err)
|
||||
|
||||
// wait cancel punish then re-acquire
|
||||
time.Sleep(cancelPunish + 10*time.Millisecond)
|
||||
_, _, exist = q.Acquire(idc)
|
||||
require.True(t, exist)
|
||||
|
||||
// Reclaim: no such IDC
|
||||
err = q.Reclaim(noSuchIdc, taskID, src, dst, newDst, newDst.DiskID)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Reclaim: unmatched suid
|
||||
err = q.Reclaim(idc, taskID, src, proto.ShardUnitInfoSimple{DiskID: 99}, newDst, newDst.DiskID)
|
||||
require.EqualError(t, err, ErrUnmatchedSuid.Error())
|
||||
|
||||
// Reclaim: success (updates destination)
|
||||
err = q.Reclaim(idc, taskID, src, dst, newDst, newDst.DiskID)
|
||||
require.NoError(t, err)
|
||||
|
||||
// re-acquire after reclaim (requeued immediately)
|
||||
_, wt, exist = q.Acquire(idc)
|
||||
require.True(t, exist)
|
||||
require.Equal(t, newDst, wt.GetDestination())
|
||||
|
||||
// Update: no such IDC
|
||||
_, err = q.Update(noSuchIdc, taskID, leader)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Update: success (updates leader)
|
||||
newLeader := proto.ShardUnitInfoSimple{Suid: proto.EncodeSuid(1, 2, 2), DiskID: 5}
|
||||
wt2, err := q.Update(idc, taskID, newLeader)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, newLeader, wt2.GetLeader())
|
||||
|
||||
// Query: no such IDC
|
||||
_, err = q.Query(noSuchIdc, taskID)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Query: success
|
||||
wt3, err := q.Query(idc, taskID)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, newDst, wt3.GetDestination())
|
||||
|
||||
// Renewal: no such IDC
|
||||
err = q.Renewal(noSuchIdc, taskID)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Renewal: success
|
||||
err = q.Renewal(idc, taskID)
|
||||
require.NoError(t, err)
|
||||
|
||||
// Complete: no such IDC
|
||||
_, err = q.Complete(noSuchIdc, taskID, src, newDst)
|
||||
require.EqualError(t, err, errNoSuchIDCQueue.Error())
|
||||
|
||||
// Complete: unmatched suid
|
||||
_, err = q.Complete(idc, taskID, src, proto.ShardUnitInfoSimple{DiskID: 99})
|
||||
require.EqualError(t, err, ErrUnmatchedSuid.Error())
|
||||
|
||||
// Complete: success
|
||||
completed, err := q.Complete(idc, taskID, src, newDst)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, newDst, completed.GetDestination())
|
||||
|
||||
// StatsTasks after complete: empty
|
||||
todo, doing = q.StatsTasks()
|
||||
require.Equal(t, 0, todo)
|
||||
require.Equal(t, 0, doing)
|
||||
}
|
||||
|
||||
@ -118,3 +118,40 @@ func TestAbnormalReport(t *testing.T) {
|
||||
require.True(t, rp.IsVuidReported(vuid))
|
||||
require.False(t, rp.IsVuidReported(vuid+1))
|
||||
}
|
||||
|
||||
func TestTaskStatsMgr_MissingPaths(t *testing.T) {
|
||||
mgr := NewTaskStatsMgr(1, proto.TaskTypeDiskRepair)
|
||||
|
||||
// QueryTaskDetail: not found
|
||||
_, err := mgr.QueryTaskDetail("nonexistent_task")
|
||||
require.Error(t, err)
|
||||
|
||||
// ReportWorkerTaskStats: first call creates entry with StartTime
|
||||
mgr.ReportWorkerTaskStats("task_progress", proto.TaskStatistics{Progress: 50}, 5, 5)
|
||||
info, err := mgr.QueryTaskDetail("task_progress")
|
||||
require.NoError(t, err)
|
||||
require.False(t, info.Completed)
|
||||
|
||||
// ReportWorkerTaskStats: progress >= 100 sets CompleteTime and Completed=true
|
||||
mgr.ReportWorkerTaskStats("task_progress", proto.TaskStatistics{Progress: 100}, 5, 5)
|
||||
info, err = mgr.QueryTaskDetail("task_progress")
|
||||
require.NoError(t, err)
|
||||
require.True(t, info.Completed)
|
||||
require.False(t, info.CompleteTime.IsZero())
|
||||
|
||||
// Second call to NewTaskStatsMgr with same labels hits AlreadyRegisteredError path
|
||||
mgr2 := NewTaskStatsMgr(1, proto.TaskTypeDiskRepair)
|
||||
require.NotNil(t, mgr2)
|
||||
|
||||
// NewCounter: second call with same labels hits AlreadyRegisteredError path
|
||||
c1 := NewCounter(2, "test_type", KindSuccess)
|
||||
c2 := NewCounter(2, "test_type", KindSuccess)
|
||||
require.NotNil(t, c1)
|
||||
require.NotNil(t, c2)
|
||||
|
||||
// NewAbnormalReporter: second call with same labels hits AlreadyRegisteredError path
|
||||
rp1 := NewAbnormalReporter(proto.ClusterID(3), "test_type2", ChunkMissMigrateAbnormal)
|
||||
rp2 := NewAbnormalReporter(proto.ClusterID(3), "test_type2", ChunkMissMigrateAbnormal)
|
||||
require.NotNil(t, rp1)
|
||||
require.NotNil(t, rp2)
|
||||
}
|
||||
|
||||
@ -50,3 +50,13 @@ func (mr *MockAllocVunitMockRecorder) AllocVolumeUnit(arg0, arg1, arg2, arg3 int
|
||||
mr.mock.ctrl.T.Helper()
|
||||
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AllocVolumeUnit", reflect.TypeOf((*MockAllocVunit)(nil).AllocVolumeUnit), arg0, arg1, arg2, arg3)
|
||||
}
|
||||
|
||||
// mockAllocShardUnit is a simple manual mock for IAllocShardUnit.
|
||||
type mockAllocShardUnit struct {
|
||||
ret *client.AllocShardUnitInfo
|
||||
err error
|
||||
}
|
||||
|
||||
func (m *mockAllocShardUnit) AllocShardUnit(_ context.Context, _ proto.Suid, _ []proto.DiskID) (*client.AllocShardUnitInfo, error) {
|
||||
return m.ret, m.err
|
||||
}
|
||||
|
||||
@ -95,3 +95,60 @@ func TestGenTaskID(t *testing.T) {
|
||||
id := GenTaskID(prefix, proto.Vid(1))
|
||||
require.True(t, strings.HasPrefix(id, prefix))
|
||||
}
|
||||
|
||||
func TestSubSuids(t *testing.T) {
|
||||
a := []proto.Suid{1, 2, 3, 4, 5}
|
||||
b := []proto.Suid{1, 2, 3}
|
||||
c := SubSuids(a, b)
|
||||
require.Equal(t, []proto.Suid{4, 5}, c)
|
||||
|
||||
// b is empty
|
||||
require.Equal(t, a, SubSuids(a, nil))
|
||||
|
||||
// a is empty
|
||||
require.Nil(t, SubSuids(nil, b))
|
||||
}
|
||||
|
||||
func TestShouldAllocShardUnitAndRedo(t *testing.T) {
|
||||
require.True(t, ShouldAllocShardUnitAndRedo(errcode.CodeNewSuidNotMatch))
|
||||
require.True(t, ShouldAllocShardUnitAndRedo(errcode.CodeCMGetShardFailed))
|
||||
require.False(t, ShouldAllocShardUnitAndRedo(0))
|
||||
require.False(t, ShouldAllocShardUnitAndRedo(999))
|
||||
}
|
||||
|
||||
func TestAllocShardUnitSafe(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
errMock := errors.New("alloc shard unit error")
|
||||
|
||||
// allocation failed
|
||||
cli := &mockAllocShardUnit{err: errMock}
|
||||
_, err := AllocShardUnitSafe(ctx, cli, proto.ShardUnitInfoSimple{DiskID: 1}, proto.ShardUnitInfoSimple{DiskID: 2}, nil)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// success: newly allocated disk does not overlap with src/dest
|
||||
newSuid := proto.EncodeSuid(1, 0, 1)
|
||||
cli = &mockAllocShardUnit{
|
||||
ret: &client.AllocShardUnitInfo{
|
||||
ShardUnitInfoSimple: proto.ShardUnitInfoSimple{
|
||||
Suid: newSuid,
|
||||
DiskID: 10,
|
||||
},
|
||||
},
|
||||
}
|
||||
ret, err := AllocShardUnitSafe(ctx, cli, proto.ShardUnitInfoSimple{DiskID: 1}, proto.ShardUnitInfoSimple{DiskID: 2}, nil)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, proto.DiskID(10), ret.DiskID)
|
||||
|
||||
// panic: newly allocated disk is the same as src
|
||||
cli = &mockAllocShardUnit{
|
||||
ret: &client.AllocShardUnitInfo{
|
||||
ShardUnitInfoSimple: proto.ShardUnitInfoSimple{
|
||||
Suid: newSuid,
|
||||
DiskID: 1,
|
||||
},
|
||||
},
|
||||
}
|
||||
require.Panics(t, func() {
|
||||
AllocShardUnitSafe(ctx, cli, proto.ShardUnitInfoSimple{DiskID: 1}, proto.ShardUnitInfoSimple{DiskID: 2}, nil)
|
||||
})
|
||||
}
|
||||
|
||||
@ -42,3 +42,22 @@ func TestVolTaskLocker(t *testing.T) {
|
||||
err = mu.TryLock(ctx, 1)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestShardTaskLockerInst(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// singleton: two calls return same instance
|
||||
locker1 := ShardTaskLockerInst()
|
||||
locker2 := ShardTaskLockerInst()
|
||||
require.Equal(t, locker1, locker2)
|
||||
|
||||
// TryLock and Unlock
|
||||
err := locker1.TryLock(ctx, 9999)
|
||||
require.NoError(t, err)
|
||||
err = locker1.TryLock(ctx, 9999)
|
||||
require.EqualError(t, err, ErrVidTaskConflict.Error())
|
||||
locker1.Unlock(ctx, 9999)
|
||||
err = locker1.TryLock(ctx, 9999)
|
||||
require.NoError(t, err)
|
||||
locker1.Unlock(ctx, 9999)
|
||||
}
|
||||
|
||||
@ -673,7 +673,7 @@ func (mgr *BlobDeleteMgr) deleteShard(ctx context.Context, location proto.VunitL
|
||||
|
||||
// has mark delete or delete before and just return
|
||||
if (stageMgr.hasDelete(location.Vuid)) || (markDelete && stageMgr.hasMarkDel(location.Vuid)) {
|
||||
span.Warnf("already delete and return: bid[%d], location[%+v], markDelete[%+v]",
|
||||
span.Debugf("already delete and return: bid[%d], location[%+v], markDelete[%+v]",
|
||||
bid, location, markDelete)
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -41,4 +41,8 @@ func TestBlobnode(t *testing.T) {
|
||||
|
||||
err = cli.Delete(ctx, proto.VunitLocation{}, proto.BlobID(1))
|
||||
require.NoError(t, err)
|
||||
|
||||
client.EXPECT().RepairShard(any, any, any).Return(nil)
|
||||
err = cli.RepairShard(ctx, "127.0.0.1:xxx", proto.ShardRepairTask{})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
@ -31,6 +31,46 @@ import (
|
||||
_ "github.com/cubefs/cubefs/blobstore/testing/nolog"
|
||||
)
|
||||
|
||||
func TestValidMigrateTask(t *testing.T) {
|
||||
taskID := GenMigrateTaskID(proto.TaskTypeDiskRepair, proto.DiskID(1), 1)
|
||||
require.True(t, ValidMigrateTask(proto.TaskTypeDiskRepair, taskID))
|
||||
require.False(t, ValidMigrateTask(proto.TaskTypeBalance, taskID))
|
||||
}
|
||||
|
||||
func TestDiskInfoSimpleIsRepaired(t *testing.T) {
|
||||
disk := &DiskInfoSimple{Status: proto.DiskStatusRepaired}
|
||||
require.True(t, disk.IsRepaired())
|
||||
disk.Status = proto.DiskStatusNormal
|
||||
require.False(t, disk.IsRepaired())
|
||||
}
|
||||
|
||||
func TestShardNodeDiskInfoMethods(t *testing.T) {
|
||||
disk := &ShardNodeDiskInfo{}
|
||||
|
||||
disk.Status = proto.DiskStatusNormal
|
||||
require.True(t, disk.IsHealth())
|
||||
require.False(t, disk.IsBroken())
|
||||
require.False(t, disk.IsDropped())
|
||||
require.False(t, disk.IsRepaired())
|
||||
require.True(t, disk.CanDropped())
|
||||
|
||||
disk.Status = proto.DiskStatusBroken
|
||||
require.False(t, disk.IsHealth())
|
||||
require.True(t, disk.IsBroken())
|
||||
require.False(t, disk.CanDropped())
|
||||
|
||||
disk.Status = proto.DiskStatusDropped
|
||||
require.True(t, disk.IsDropped())
|
||||
require.True(t, disk.CanDropped())
|
||||
|
||||
disk.Status = proto.DiskStatusRepaired
|
||||
require.True(t, disk.IsRepaired())
|
||||
require.True(t, disk.CanDropped())
|
||||
|
||||
disk.Status = proto.DiskStatusRepairing
|
||||
require.False(t, disk.CanDropped())
|
||||
}
|
||||
|
||||
var (
|
||||
defaultVolumeListMarker = proto.Vid(0)
|
||||
defaultDiskListMarker = proto.DiskID(0)
|
||||
@ -552,3 +592,216 @@ func TestClustermgrClient(t *testing.T) {
|
||||
require.NoError(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClustermgrClientExtra(t *testing.T) {
|
||||
cli := NewClusterMgrClient(&cmapi.Config{}).(*clustermgrClient)
|
||||
mockCli := NewMockClusterManager(gomock.NewController(t))
|
||||
cli.client = mockCli
|
||||
|
||||
ctx := context.Background()
|
||||
any := gomock.Any()
|
||||
errMock := errors.New("fake error")
|
||||
|
||||
{
|
||||
// set config
|
||||
mockCli.EXPECT().SetConfig(any, any, any).Return(nil)
|
||||
err := cli.SetConfig(ctx, "key", "value")
|
||||
require.NoError(t, err)
|
||||
|
||||
mockCli.EXPECT().SetConfig(any, any, any).Return(errMock)
|
||||
err = cli.SetConfig(ctx, "key", "value")
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
}
|
||||
{
|
||||
// get service - matching clusterID
|
||||
mockCli.EXPECT().GetService(any, any).Return(cmapi.ServiceInfo{
|
||||
Nodes: []cmapi.ServiceNode{
|
||||
{ClusterID: 1, Host: "127.0.0.1:8080"},
|
||||
{ClusterID: 2, Host: "127.0.0.2:8080"},
|
||||
},
|
||||
}, nil)
|
||||
hosts, err := cli.GetService(ctx, "scheduler", proto.ClusterID(1))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, len(hosts))
|
||||
require.Equal(t, "127.0.0.1:8080", hosts[0])
|
||||
|
||||
mockCli.EXPECT().GetService(any, any).Return(cmapi.ServiceInfo{}, errMock)
|
||||
_, err = cli.GetService(ctx, "scheduler", proto.ClusterID(1))
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
}
|
||||
{
|
||||
// list migrate tasks (public wrapper)
|
||||
task1 := &proto.MigrateTask{TaskID: GenMigrateTaskID(proto.TaskTypeDiskRepair, proto.DiskID(1), 1), TaskType: proto.TaskTypeDiskRepair}
|
||||
task1Bytes, _ := json.Marshal(task1)
|
||||
opts := &cmapi.ListKvOpts{Prefix: GenMigrateTaskPrefix(proto.TaskTypeDiskRepair), Count: 10}
|
||||
mockCli.EXPECT().ListKV(any, any).Return(cmapi.ListKvRet{
|
||||
Kvs: []*cmapi.KeyValue{{Key: task1.TaskID, Value: task1Bytes}}, Marker: "",
|
||||
}, nil)
|
||||
tasks, marker, err := cli.ListMigrateTasks(ctx, proto.TaskTypeDiskRepair, opts)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, len(tasks))
|
||||
require.Equal(t, "", marker)
|
||||
}
|
||||
{
|
||||
// get migrate task - task type mismatch
|
||||
task1, _ := (&proto.MigrateTask{
|
||||
TaskID: GenMigrateTaskID(proto.TaskTypeDiskRepair, proto.DiskID(1), 1),
|
||||
TaskType: proto.TaskTypeDiskRepair,
|
||||
}).ToTask()
|
||||
taskBytes, _ := task1.Marshal()
|
||||
mockCli.EXPECT().GetKV(any, any).Return(cmapi.GetKvRet{Value: taskBytes}, nil)
|
||||
_, err := cli.GetMigrateTask(ctx, proto.TaskTypeBalance, task1.TaskID)
|
||||
require.ErrorIs(t, err, errcode.ErrIllegalTaskType)
|
||||
}
|
||||
{
|
||||
// get migrating disk - task type mismatch
|
||||
diskMeta := &MigratingDiskMeta{Disk: &DiskInfoSimple{DiskID: proto.DiskID(1)}, TaskType: proto.TaskTypeDiskDrop}
|
||||
diskMetaBytes, _ := json.Marshal(diskMeta)
|
||||
mockCli.EXPECT().GetKV(any, any).Return(cmapi.GetKvRet{Value: diskMetaBytes}, nil)
|
||||
_, err := cli.GetMigratingDisk(ctx, proto.TaskTypeBalance, proto.DiskID(1))
|
||||
require.ErrorIs(t, err, errcode.ErrIllegalTaskType)
|
||||
|
||||
// get migrating disk - disk ID mismatch
|
||||
diskMeta2 := &MigratingDiskMeta{Disk: &DiskInfoSimple{DiskID: proto.DiskID(99)}, TaskType: proto.TaskTypeDiskDrop}
|
||||
diskMeta2Bytes, _ := json.Marshal(diskMeta2)
|
||||
mockCli.EXPECT().GetKV(any, any).Return(cmapi.GetKvRet{Value: diskMeta2Bytes}, nil)
|
||||
_, err = cli.GetMigratingDisk(ctx, proto.TaskTypeDiskDrop, proto.DiskID(1))
|
||||
require.ErrorIs(t, err, errcode.ErrIllegalTaskType)
|
||||
}
|
||||
}
|
||||
|
||||
func TestClustermgrClientShardMethods(t *testing.T) {
|
||||
cli := NewClusterMgrClient(&cmapi.Config{}).(*clustermgrClient)
|
||||
mockCli := NewMockClusterManager(gomock.NewController(t))
|
||||
cli.client = mockCli
|
||||
|
||||
ctx := context.Background()
|
||||
any := gomock.Any()
|
||||
errMock := errors.New("fake error")
|
||||
|
||||
{
|
||||
// get shard info - error
|
||||
mockCli.EXPECT().GetShardInfo(any, any).Return(nil, errMock)
|
||||
_, err := cli.GetShardInfo(ctx, proto.ShardID(1))
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// get shard info - success
|
||||
shardInfo := &cmapi.Shard{
|
||||
ShardID: proto.ShardID(10),
|
||||
AppliedIndex: 100,
|
||||
LeaderDiskID: proto.DiskID(2),
|
||||
Units: []cmapi.ShardUnit{
|
||||
{Suid: proto.Suid(1), DiskID: proto.DiskID(1), Host: "127.0.0.1:xxx"},
|
||||
{Suid: proto.Suid(2), DiskID: proto.DiskID(2), Host: "127.0.0.2:xxx"},
|
||||
},
|
||||
}
|
||||
mockCli.EXPECT().GetShardInfo(any, any).Return(shardInfo, nil)
|
||||
ret, err := cli.GetShardInfo(ctx, proto.ShardID(10))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, proto.ShardID(10), ret.ShardID)
|
||||
require.Equal(t, 2, len(ret.ShardUnitInfos))
|
||||
}
|
||||
{
|
||||
// update shard - error
|
||||
mockCli.EXPECT().UpdateShard(any, any).Return(errMock)
|
||||
err := cli.UpdateShard(ctx, &UpdateShardArgs{NewSuid: proto.Suid(1), OldSuid: proto.Suid(2)})
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// update shard - success
|
||||
mockCli.EXPECT().UpdateShard(any, any).Return(nil)
|
||||
err = cli.UpdateShard(ctx, &UpdateShardArgs{NewSuid: proto.Suid(3), OldSuid: proto.Suid(2)})
|
||||
require.NoError(t, err)
|
||||
}
|
||||
{
|
||||
// alloc shard unit - error
|
||||
mockCli.EXPECT().AllocShardUnit(any, any).Return(nil, errMock)
|
||||
_, err := cli.AllocShardUnit(ctx, proto.Suid(1), nil)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// alloc shard unit - success
|
||||
allocRet := &cmapi.AllocShardUnitRet{Suid: proto.Suid(5), DiskID: proto.DiskID(3), Host: "127.0.0.3:xxx"}
|
||||
mockCli.EXPECT().AllocShardUnit(any, any).Return(allocRet, nil)
|
||||
info, err := cli.AllocShardUnit(ctx, proto.Suid(1), nil)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, proto.Suid(5), info.Suid)
|
||||
require.True(t, info.Learner)
|
||||
}
|
||||
{
|
||||
// list disk shard units - error
|
||||
mockCli.EXPECT().ListShardUnit(any, any).Return(nil, errMock)
|
||||
_, err := cli.ListDiskShardUnits(ctx, proto.DiskID(1))
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// list disk shard units - success
|
||||
units := []cmapi.ShardUnitInfo{
|
||||
{Suid: proto.Suid(1), DiskID: proto.DiskID(1), Host: "host1", AppliedIndex: 10},
|
||||
}
|
||||
mockCli.EXPECT().ListShardUnit(any, any).Return(units, nil)
|
||||
ret, err := cli.ListDiskShardUnits(ctx, proto.DiskID(1))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, len(ret))
|
||||
require.Equal(t, proto.Suid(1), ret[0].Suid)
|
||||
}
|
||||
{
|
||||
// list shard - stub, always returns nil
|
||||
shards, _, err := cli.ListShard(ctx, proto.ShardID(0), 10)
|
||||
require.NoError(t, err)
|
||||
require.Nil(t, shards)
|
||||
}
|
||||
{
|
||||
// list shard disk (normal status)
|
||||
disk1 := &cmapi.ShardNodeDiskInfo{
|
||||
DiskInfo: cmapi.DiskInfo{Status: proto.DiskStatusNormal},
|
||||
ShardNodeDiskHeartbeatInfo: cmapi.ShardNodeDiskHeartbeatInfo{DiskID: proto.DiskID(1)},
|
||||
}
|
||||
mockCli.EXPECT().ListShardNodeDisk(any, any).Return(cmapi.ListShardNodeDiskRet{
|
||||
Disks: []*cmapi.ShardNodeDiskInfo{disk1}, Marker: defaultListDiskMarker,
|
||||
}, nil)
|
||||
disks, err := cli.ListShardDisk(ctx)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, len(disks))
|
||||
|
||||
// list broken shard disk - error
|
||||
mockCli.EXPECT().ListShardNodeDisk(any, any).Return(cmapi.ListShardNodeDiskRet{}, errMock)
|
||||
_, err = cli.ListBrokenShardDisk(ctx)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// list repairing shard disk - success
|
||||
mockCli.EXPECT().ListShardNodeDisk(any, any).Return(cmapi.ListShardNodeDiskRet{
|
||||
Disks: []*cmapi.ShardNodeDiskInfo{disk1}, Marker: defaultListDiskMarker,
|
||||
}, nil)
|
||||
disks, err = cli.ListRepairingShardDisk(ctx)
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, 1, len(disks))
|
||||
}
|
||||
{
|
||||
// set shard disk repairing
|
||||
mockCli.EXPECT().SetShardNodeDisk(any, any, any).Return(nil)
|
||||
err := cli.SetShardDiskRepairing(ctx, proto.DiskID(1))
|
||||
require.NoError(t, err)
|
||||
|
||||
// set shard disk repaired
|
||||
mockCli.EXPECT().SetShardNodeDisk(any, any, any).Return(nil)
|
||||
err = cli.SetShardDiskRepaired(ctx, proto.DiskID(1))
|
||||
require.NoError(t, err)
|
||||
}
|
||||
{
|
||||
// get shard disk info - error
|
||||
mockCli.EXPECT().ShardNodeDiskInfo(any, any).Return(nil, errMock)
|
||||
_, err := cli.GetShardDiskInfo(ctx, proto.DiskID(1))
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// get shard disk info - success, also covers ShardNodeDiskInfo.set
|
||||
info := &cmapi.ShardNodeDiskInfo{
|
||||
DiskInfo: cmapi.DiskInfo{Status: proto.DiskStatusNormal, Host: "127.0.0.1:xxx"},
|
||||
ShardNodeDiskHeartbeatInfo: cmapi.ShardNodeDiskHeartbeatInfo{
|
||||
DiskID: proto.DiskID(1), FreeShardCnt: 10, UsedShardCnt: 5,
|
||||
},
|
||||
}
|
||||
mockCli.EXPECT().ShardNodeDiskInfo(any, any).Return(info, nil)
|
||||
disk, err := cli.GetShardDiskInfo(ctx, proto.DiskID(1))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, proto.DiskID(1), disk.DiskID)
|
||||
require.True(t, disk.IsHealth())
|
||||
}
|
||||
}
|
||||
|
||||
@ -31,6 +31,29 @@ import (
|
||||
"github.com/cubefs/cubefs/blobstore/testing/mocks"
|
||||
)
|
||||
|
||||
// newShardMigrateMgrWithCallback creates a manager with loadTaskCallback set.
|
||||
func newShardMigrateMgrWithCallback(t *testing.T, cb loadTaskCallback) *ShardMigrateMgr {
|
||||
ctr := gomock.NewController(t)
|
||||
clusterMgr := NewMockClusterMgrAPI(ctr)
|
||||
taskSwitch := mocks.NewMockSwitcher(ctr)
|
||||
|
||||
conf := &ShardMigrateConfig{
|
||||
ClusterID: 0,
|
||||
TaskCommonConfig: base.TaskCommonConfig{
|
||||
PrepareQueueRetryDelayS: 0,
|
||||
FinishQueueRetryDelayS: 0,
|
||||
CancelPunishDurationS: 0,
|
||||
WorkQueueSize: 3,
|
||||
},
|
||||
loadTaskCallback: cb,
|
||||
}
|
||||
|
||||
mgr := NewShardMigrateMgr(clusterMgr, taskSwitch, conf, proto.TaskTypeShardDiskRepair)
|
||||
m, ok := mgr.(*ShardMigrateMgr)
|
||||
require.True(t, ok)
|
||||
return m
|
||||
}
|
||||
|
||||
var MockMigrateShardInfoMap = map[proto.ShardID]*client.ShardInfoSimple{
|
||||
100: MockGenShardInfo(100, 0),
|
||||
101: MockGenShardInfo(101, 0),
|
||||
@ -468,4 +491,221 @@ func TestShardMigrateMgr_QueryTask(t *testing.T) {
|
||||
_, err := mgr.GetTask(context.Background(), t1.TaskID)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
}
|
||||
{
|
||||
// unmarshal error: GetMigrateTask returns task with invalid data
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().GetMigrateTask(any, any, any).Return(&proto.Task{Data: []byte("invalid-json")}, nil)
|
||||
_, err := mgr.GetTask(context.Background(), t1.TaskID)
|
||||
require.Error(t, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestShardMigrateMgr_SimpleOps(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
mgr := newShardMigrateMgr(t)
|
||||
|
||||
// ReportTask
|
||||
err := mgr.ReportTask(ctx, nil)
|
||||
require.NoError(t, err)
|
||||
|
||||
// StatQueueTaskCnt / Stats with empty queues
|
||||
preparing, doing, finishing := mgr.StatQueueTaskCnt()
|
||||
require.Equal(t, 0, preparing)
|
||||
require.Equal(t, 0, doing)
|
||||
require.Equal(t, 0, finishing)
|
||||
stats := mgr.Stats()
|
||||
require.Equal(t, 0, stats.PreparingCnt)
|
||||
require.Equal(t, 0, stats.WorkerDoingCnt)
|
||||
require.Equal(t, 0, stats.FinishingCnt)
|
||||
|
||||
// StatQueueTaskCnt with tasks in queues
|
||||
t1 := mockGenShardMigrateTask(100, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStateInited, MockMigrateShardInfoMap)
|
||||
mgr.prepareQueue.PushTask(t1.TaskID, t1)
|
||||
t2 := mockGenShardMigrateTask(101, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStatePrepared, MockMigrateShardInfoMap)
|
||||
mgr.workQueue.AddPreparedTask("z0", t2.TaskID, t2)
|
||||
t3 := mockGenShardMigrateTask(102, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStateWorkCompleted, MockMigrateShardInfoMap)
|
||||
mgr.finishQueue.PushTask(t3.TaskID, t3)
|
||||
preparing, doing, finishing = mgr.StatQueueTaskCnt()
|
||||
require.Greater(t, preparing, 0)
|
||||
require.Greater(t, doing, 0)
|
||||
require.Greater(t, finishing, 0)
|
||||
|
||||
// Enabled
|
||||
mgr.taskSwitch.(*mocks.MockSwitcher).EXPECT().Enabled().Return(true)
|
||||
require.True(t, mgr.Enabled())
|
||||
|
||||
// WaitEnable
|
||||
mgr.taskSwitch.(*mocks.MockSwitcher).EXPECT().WaitEnable().Return()
|
||||
mgr.WaitEnable()
|
||||
|
||||
// Done + Close
|
||||
done := mgr.Done()
|
||||
require.NotNil(t, done)
|
||||
mgr.Close()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Close should have closed Done channel")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShardMigrateMgr_QueryTaskFull(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
mgr := newShardMigrateMgr(t)
|
||||
t1 := mockGenShardMigrateTask(100, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStatePrepared, MockMigrateShardInfoMap)
|
||||
t2, err := t1.ToTask()
|
||||
require.NoError(t, err)
|
||||
|
||||
// success
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().GetMigrateTask(any, any, any).Return(t2, nil)
|
||||
ret, err := mgr.QueryTask(ctx, t1.TaskID)
|
||||
require.NoError(t, err)
|
||||
require.NotNil(t, ret)
|
||||
require.Equal(t, t1.TaskType, ret.TaskType)
|
||||
|
||||
// GetTask error
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().GetMigrateTask(any, any, any).Return(nil, errMock)
|
||||
_, err = mgr.QueryTask(ctx, t1.TaskID)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
}
|
||||
|
||||
func TestShardMigrateSuids(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// ListMigratingSuid
|
||||
{
|
||||
mgr := newShardMigrateMgr(t)
|
||||
|
||||
// error
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListAllMigrateTasksByDiskID(any, any, any).Return(nil, errMock)
|
||||
_, err := mgr.ListMigratingSuid(ctx, 1)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// success with task
|
||||
t1 := mockGenShardMigrateTask(100, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStatePrepared, MockMigrateShardInfoMap)
|
||||
protoTask, err := t1.ToTask()
|
||||
require.NoError(t, err)
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListAllMigrateTasksByDiskID(any, any, any).Return([]*proto.Task{protoTask}, nil)
|
||||
suids, err := mgr.ListMigratingSuid(ctx, 4)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, suids, 1)
|
||||
require.Equal(t, t1.Source.Suid, suids[0])
|
||||
|
||||
// success with empty list
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListAllMigrateTasksByDiskID(any, any, any).Return([]*proto.Task{}, nil)
|
||||
suids, err = mgr.ListMigratingSuid(ctx, 4)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, suids, 0)
|
||||
}
|
||||
|
||||
// ListImmigratedSuid
|
||||
{
|
||||
mgr := newShardMigrateMgr(t)
|
||||
|
||||
// error
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListDiskShardUnits(any, any).Return(nil, errMock)
|
||||
_, err := mgr.ListImmigratedSuid(ctx, 1)
|
||||
require.True(t, errors.Is(err, errMock))
|
||||
|
||||
// success with shard units
|
||||
suid := proto.EncodeSuid(100, 0, 1)
|
||||
sunits := []*client.ShardUnitInfoSimple{
|
||||
{Suid: suid, DiskID: 1},
|
||||
}
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListDiskShardUnits(any, any).Return(sunits, nil)
|
||||
suids, err := mgr.ListImmigratedSuid(ctx, 1)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, suids, 1)
|
||||
require.Equal(t, suid, suids[0])
|
||||
|
||||
// success with empty list
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListDiskShardUnits(any, any).Return([]*client.ShardUnitInfoSimple{}, nil)
|
||||
suids, err = mgr.ListImmigratedSuid(ctx, 1)
|
||||
require.NoError(t, err)
|
||||
require.Len(t, suids, 0)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMigratingShardDisks(t *testing.T) {
|
||||
m := newMigratingShardDisks()
|
||||
require.NotNil(t, m)
|
||||
|
||||
diskInfo1 := &client.ShardNodeDiskInfo{}
|
||||
diskInfo2 := &client.ShardNodeDiskInfo{}
|
||||
diskID1 := proto.DiskID(1)
|
||||
diskID2 := proto.DiskID(2)
|
||||
|
||||
require.Equal(t, 0, m.size())
|
||||
|
||||
// add
|
||||
m.add(diskID1, diskInfo1)
|
||||
m.add(diskID2, diskInfo2)
|
||||
require.Equal(t, 2, m.size())
|
||||
|
||||
// get existing
|
||||
got, exist := m.get(diskID1)
|
||||
require.True(t, exist)
|
||||
require.Equal(t, diskInfo1, got)
|
||||
|
||||
// get non-existent
|
||||
_, exist = m.get(proto.DiskID(999))
|
||||
require.False(t, exist)
|
||||
|
||||
// list
|
||||
list := m.list()
|
||||
require.Len(t, list, 2)
|
||||
|
||||
// delete
|
||||
m.delete(diskID1)
|
||||
require.Equal(t, 1, m.size())
|
||||
|
||||
_, exist = m.get(diskID1)
|
||||
require.False(t, exist)
|
||||
}
|
||||
|
||||
func TestShardMigrateLoad_WithCallback(t *testing.T) {
|
||||
callbackDiskIDs := make([]proto.DiskID, 0)
|
||||
cb := func(ctx context.Context, diskID proto.DiskID) {
|
||||
callbackDiskIDs = append(callbackDiskIDs, diskID)
|
||||
}
|
||||
mgr := newShardMigrateMgrWithCallback(t, cb)
|
||||
|
||||
t1 := mockGenShardMigrateTask(100, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStateInited, MockMigrateShardInfoMap)
|
||||
protoTask, _ := t1.ToTask()
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().ListAllMigrateTasks(any, any).Return([]*proto.Task{protoTask}, nil)
|
||||
err := mgr.Load()
|
||||
require.NoError(t, err)
|
||||
require.Len(t, callbackDiskIDs, 1)
|
||||
require.Equal(t, t1.Source.DiskID, callbackDiskIDs[0])
|
||||
}
|
||||
|
||||
func TestDealCancelReason_GetShardInfoError(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
idc := "z0"
|
||||
|
||||
mgr := newShardMigrateMgr(t)
|
||||
t1 := mockGenShardMigrateTask(100, proto.TaskTypeShardDiskRepair, "z0", 4, proto.ShardTaskStatePrepared, MockMigrateShardInfoMap)
|
||||
mgr.workQueue.AddPreparedTask(idc, t1.TaskID, t1)
|
||||
|
||||
// code triggers GetShardInfo, but GetShardInfo fails → early return, no panic
|
||||
mgr.clusterMgrCli.(*MockClusterMgrAPI).EXPECT().GetShardInfo(any, any).Return(nil, errMock)
|
||||
taskArgs := genShardTaskArgs(t1, errcode.ErrShardNodeNotLeader.Error(), errcode.CodeShardNodeNotLeader)
|
||||
err := mgr.CancelTask(ctx, taskArgs)
|
||||
require.NoError(t, err)
|
||||
}
|
||||
|
||||
func TestShardMigrateRun_QueueFull(t *testing.T) {
|
||||
mgr := newShardMigrateMgr(t)
|
||||
mgr.taskSwitch.(*mocks.MockSwitcher).EXPECT().WaitEnable().AnyTimes().Return()
|
||||
mgr.taskSwitch.(*mocks.MockSwitcher).EXPECT().Enabled().AnyTimes().Return(true)
|
||||
|
||||
// fill workQueue to capacity (WorkQueueSize=3) to trigger the "queue full" branch
|
||||
for i := proto.ShardID(200); i < 203; i++ {
|
||||
MockMigrateShardInfoMap[i] = MockGenShardInfo(i, 0)
|
||||
task := mockGenShardMigrateTask(i, proto.TaskTypeShardDiskRepair, "z0", proto.DiskID(i), proto.ShardTaskStatePrepared, MockMigrateShardInfoMap)
|
||||
mgr.workQueue.AddPreparedTask("z0", task.TaskID, task)
|
||||
}
|
||||
|
||||
mgr.Run()
|
||||
// give the loop time to hit the "queue full" branch
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
