Function TestClusterMigrateData

server/api/cluster_test.go:215–305

(t *testing.T)

}

func TestClusterMigrateData(t *testing.T) {
	ns := "test-ns"
	clusterName := "test-cluster"
	clusterStore := store.NewClusterStore(engine.NewMock())
	handler := &ClusterHandler{s: clusterStore}

	ctx := context.Background()
	nodeAddrs := []string{"127.0.0.1:7770", "127.0.0.1:7771"}
	sourceRedisClient := redis.NewClient(&redis.Options{Addr: nodeAddrs[0]})
	targetRedisClient := redis.NewClient(&redis.Options{Addr: nodeAddrs[1]})

	cluster, err := store.NewCluster(clusterName, nodeAddrs, 1)
	require.NoError(t, err)
	require.NoError(t, cluster.Reset(ctx))
	defer func() {
		require.NoError(t, cluster.Reset(ctx))
	}()
	require.NoError(t, cluster.SyncToNodes(ctx))
	clusterStore.CreateCluster(ctx, ns, cluster)

	sendRequest := func(t *testing.T, ns, cluster string, slotRange store.SlotRange) {
		recorder := httptest.NewRecorder()
		reqCtx := GetTestContext(recorder)
		reqCtx.Set(consts.ContextKeyStore, handler.s)
		reqCtx.Params = []gin.Param{{Key: "namespace", Value: ns}, {Key: "cluster", Value: cluster}}
		body, err := json.Marshal(&MigrateSlotRequest{Target: 1, Slot: slotRange})
		require.NoError(t, err)
		reqCtx.Request.Body = io.NopCloser(bytes.NewBuffer(body))

		middleware.RequiredCluster(reqCtx)
		handler.MigrateSlot(reqCtx)
		require.Equal(t, http.StatusOK, recorder.Code)
	}

	runController := func(t *testing.T) *controller.Controller {
		ctrl, err := controller.New(clusterStore, &config.ControllerConfig{
			FailOver: &config.FailOverConfig{
				PingIntervalSeconds: 1,
				MaxPingCount:        3,
			},
		})
		require.NoError(t, err)
		require.NoError(t, ctrl.Start(ctx))
		ctrl.WaitForReady()
		return ctrl
	}

	t.Run("migrate slot(s) from one shard to another", func(t *testing.T) {
		for _, slotRange := range []store.SlotRange{
			{Start: 10, Stop: 10},
			{Start: 11, Stop: 14},
		} {
			for i := slotRange.Start; i <= slotRange.Stop; i++ {
				require.NoError(t, sourceRedisClient.Set(ctx, util.SlotTable[i], "test-value", 0).Err())
			}
			sendRequest(t, ns, "test-cluster", slotRange)
			gotCluster, err := clusterStore.GetCluster(ctx, ns, clusterName)
			require.NoError(t, err)