Function TestMigrateGenTasklets

blobstore/blobnode/task_runner_migrate_test.go:30–162

(t *testing.T)

)

func TestMigrateGenTasklets(t *testing.T) {
	mode := codemode.EC6P10L2
	replicas := genMockVol(100, codemode.CodeMode(mode))
	badi := 10
	godi := 11
	balanceTask := &proto.MigrateTask{
		TaskID:      "mock_balance_task_id",
		TaskType:    proto.TaskTypeBalance,
		CodeMode:    codemode.CodeMode(mode),
		Sources:     replicas,
		Destination: replicas[badi],
		SourceVuid:  replicas[badi].Vuid,
	}
	bids := []proto.BlobID{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
	sizes := []int64{1024, 2048, 0, 512, 23, 65, 12, 50, 100, 2047}
	markDeleteBidIndex := 9
	bidsMap := make(map[proto.BlobID]int64, len(bids))
	for idx := range bids {
		bidsMap[bids[idx]] = sizes[idx]
	}

	workutils.TaskBufPool = workutils.NewBufPool(&workutils.BufConfig{
		MigrateBufSize:     2 * 1024,
		MigrateBufCapacity: 10,
		RepairBufSize:      1,
		RepairBufCapacity:  1,
	})
	getter := NewMockGetterWithBids(replicas, codemode.CodeMode(mode), bids, sizes)
	w := NewMigrateWorker(MigrateTaskEx{taskInfo: balanceTask, blobNodeCli: getter, downloadShardConcurrency: 1})

	tasklets, _ := w.GenTasklets(context.Background())
	require.Equal(t, 0, len(tasklets))

	getter.setFail(replicas[badi].Vuid, errors.New("fake error"))
	_, err := w.GenTasklets(context.Background())
	require.Equal(t, DstErr, err.errType)

	{
		getter.setWell(replicas[badi].Vuid)
		shards, _ := getter.ListShards(context.Background(), replicas[badi])
		for _, shard := range shards {
			getter.Delete(context.Background(), balanceTask.SourceVuid, shard.Bid)
		}
		tasklets, _ = w.GenTasklets(context.Background())
		require.Equal(t, 4, len(tasklets))
		var bids []*ShardInfoSimple
		for _, tasklet := range tasklets {
			var size int64 = 0
			for _, bid := range tasklet.bids {
				size += bid.Size
			}
			bids = append(bids, tasklet.bids...)
			require.LessOrEqual(t, size, int64(workutils.TaskBufPool.GetMigrateBufSize()))
		}

		for _, bid := range bids {
			require.Equal(t, bidsMap[bid.Bid], bid.Size)
		}