Fix the migrating slot didn't compatible with old format (#310)

bseto · web-flow · commit 2c834b4af8da · 2025-05-13T12:10:57.000+08:00
diff --git a/cmd/client/command/helper.go b/cmd/client/command/helper.go
@@ -51,7 +51,7 @@ func printCluster(cluster *store.Cluster) {
 				role = strings.ToUpper(store.RoleMaster)
 			}
 			migratingStatus := "NO"
-			if shard.MigratingSlot != nil {
+			if shard.IsMigrating() {
 				migratingStatus = fmt.Sprintf("%s --> %d", shard.MigratingSlot, shard.TargetShardIndex)
 			}
 			columns := []string{fmt.Sprintf("%d", i), node.ID(), node.Addr(), role, migratingStatus}
diff --git a/controller/cluster.go b/controller/cluster.go
@@ -318,19 +318,12 @@ func (c *ClusterChecker) tryUpdateMigrationStatus(ctx context.Context, clonedClu
 		if !shard.IsMigrating() {
 			continue
 		}
-
 		sourceNodeClusterInfo, err := shard.GetMasterNode().GetClusterInfo(ctx)
 		if err != nil {
 			log.Error("Failed to get the cluster info from the source node", zap.Error(err))
 			return
 		}
-		if sourceNodeClusterInfo.MigratingSlot == nil {
-			log.Error("The source migration slot is empty",
-				zap.String("migrating_slot", shard.MigratingSlot.String()),
-			)
-			return
-		}
-		if !sourceNodeClusterInfo.MigratingSlot.Equal(shard.MigratingSlot) {
+		if !sourceNodeClusterInfo.MigratingSlot.Equal(shard.MigratingSlot.SlotRange) {
 			log.Error("Mismatch migrating slot",
 				zap.String("source_migrating_slot", sourceNodeClusterInfo.MigratingSlot.String()),
 				zap.String("migrating_slot", shard.MigratingSlot.String()),
@@ -355,9 +348,9 @@ func (c *ClusterChecker) tryUpdateMigrationStatus(ctx context.Context, clonedClu
 			c.updateCluster(clonedCluster)
 			log.Warn("Failed to migrate the slot", zap.String("slot", migratingSlot.String()))
 		case "success":
-			clonedCluster.Shards[i].SlotRanges = store.RemoveSlotFromSlotRanges(clonedCluster.Shards[i].SlotRanges, *shard.MigratingSlot)
+			clonedCluster.Shards[i].SlotRanges = store.RemoveSlotFromSlotRanges(clonedCluster.Shards[i].SlotRanges, shard.MigratingSlot.SlotRange)
 			clonedCluster.Shards[shard.TargetShardIndex].SlotRanges = store.AddSlotToSlotRanges(
-				clonedCluster.Shards[shard.TargetShardIndex].SlotRanges, *shard.MigratingSlot,
+				clonedCluster.Shards[shard.TargetShardIndex].SlotRanges, shard.MigratingSlot.SlotRange,
 			)
 			migratedSlot := shard.MigratingSlot
 			clonedCluster.Shards[i].ClearMigrateState()
diff --git a/controller/cluster_test.go b/controller/cluster_test.go
@@ -109,7 +109,7 @@ func TestCluster_FailureCount(t *testing.T) {
 				mockNode0, mockNode1, mockNode2, mockNode3,
 			},
 			SlotRanges:       []store.SlotRange{{Start: 0, Stop: 16383}},
-			MigratingSlot:    nil,
+			MigratingSlot:    &store.MigratingSlot{IsMigrating: false},
 			TargetShardIndex: -1,
 		}},
 	}
@@ -221,7 +221,7 @@ func TestCluster_MigrateSlot(t *testing.T) {
 	}()
 	slotRange, err := store.NewSlotRange(0, 0)
 	require.NoError(t, err)
-	require.NoError(t, cluster.MigrateSlot(ctx, *slotRange, 1, false))
+	require.NoError(t, cluster.MigrateSlot(ctx, slotRange, 1, false))
 
 	s := NewMockClusterStore()
 	require.NoError(t, s.CreateCluster(ctx, ns, cluster))
diff --git a/server/api/cluster.go b/server/api/cluster.go
@@ -33,7 +33,7 @@ import (
 
 type MigrateSlotRequest struct {
 	Target   int             `json:"target" validate:"required"`
-	Slot     store.SlotRange `json:"slot" validate:"required"`
+	Slot     store.SlotRange `json:"slot" validate:"required"` // we don't use store.MigratingSlot here because we expect a valid SlotRange
 	SlotOnly bool            `json:"slot_only"`
 }
 
diff --git a/server/api/cluster_test.go b/server/api/cluster_test.go
@@ -131,7 +131,7 @@ func TestClusterBasics(t *testing.T) {
 		slotRange, err := store.NewSlotRange(3, 3)
 		require.NoError(t, err)
 		testMigrateReq := &MigrateSlotRequest{
-			Slot:     *slotRange,
+			Slot:     slotRange,
 			SlotOnly: true,
 			Target:   1,
 		}
@@ -271,7 +271,7 @@ func TestClusterMigrateData(t *testing.T) {
 			currentVersion := gotCluster.Version.Load()
 			sourceSlotRanges := gotCluster.Shards[0].SlotRanges
 			targetSlotRanges := gotCluster.Shards[1].SlotRanges
-			require.EqualValues(t, slotRange, *gotCluster.Shards[0].MigratingSlot)
+			require.EqualValues(t, slotRange, gotCluster.Shards[0].MigratingSlot.SlotRange)
 			require.EqualValues(t, 1, gotCluster.Shards[0].TargetShardIndex)
 
 			// Run the controller to check and update the migration status
diff --git a/store/cluster.go b/store/cluster.go
@@ -181,7 +181,7 @@ func (cluster *Cluster) findShardIndexBySlot(slot SlotRange) (int, error) {
 	for i := 0; i < len(cluster.Shards); i++ {
 		slotRanges := cluster.Shards[i].SlotRanges
 		for _, slotRange := range slotRanges {
-			if slotRange.HasOverlap(&slot) {
+			if slotRange.HasOverlap(slot) {
 				if sourceShardIdx != -1 {
 					return sourceShardIdx, consts.ErrSlotRangeBelongsToMultipleShards
 				}
@@ -226,7 +226,7 @@ func (cluster *Cluster) MigrateSlot(ctx context.Context, slot SlotRange, targetS
 	}
 
 	// Will start the data migration in the background
-	cluster.Shards[sourceShardIdx].MigratingSlot = &slot
+	cluster.Shards[sourceShardIdx].MigratingSlot = FromSlotRange(slot)
 	cluster.Shards[sourceShardIdx].TargetShardIndex = targetShardIdx
 	return nil
 }
diff --git a/store/cluster_node.go b/store/cluster_node.go
@@ -85,9 +85,9 @@ type ClusterNode struct {
 }
 
 type ClusterInfo struct {
-	CurrentEpoch   int64      `json:"cluster_current_epoch"`
-	MigratingSlot  *SlotRange `json:"migrating_slot"`
-	MigratingState string     `json:"migrating_state"`
+	CurrentEpoch   int64          `json:"cluster_current_epoch"`
+	MigratingSlot  *MigratingSlot `json:"migrating_slot"`
+	MigratingState string         `json:"migrating_state"`
 }
 
 type ClusterNodeInfo struct {
@@ -195,10 +195,11 @@ func (n *ClusterNode) GetClusterInfo(ctx context.Context) (*ClusterInfo, error)
 			}
 		case "migrating_slot", "migrating_slot(s)":
 			// TODO(@git-hulk): handle multiple migrating slots
-			clusterInfo.MigratingSlot, err = ParseSlotRange(fields[1])
+			slotRange, err := ParseSlotRange(fields[1])
 			if err != nil {
 				return nil, err
 			}
+			clusterInfo.MigratingSlot = FromSlotRange(*slotRange)
 		case "migrating_state":
 			clusterInfo.MigratingState = fields[1]
 		}
diff --git a/store/cluster_shard.go b/store/cluster_shard.go
@@ -33,11 +33,17 @@ import (
 	"github.com/apache/kvrocks-controller/consts"
 )
 
+const (
+	// the old migrating slot was denoted by an int and -1 was
+	// used to denote a non migrating slot
+	NotMigratingInt = -1
+)
+
 type Shard struct {
-	Nodes            []Node      `json:"nodes"`
-	SlotRanges       []SlotRange `json:"slot_ranges"`
-	TargetShardIndex int         `json:"target_shard_index"`
-	MigratingSlot    *SlotRange  `json:"migrating_slot"`
+	Nodes            []Node         `json:"nodes"`
+	SlotRanges       []SlotRange    `json:"slot_ranges"`
+	TargetShardIndex int            `json:"target_shard_index"`
+	MigratingSlot    *MigratingSlot `json:"migrating_slot"`
 }
 
 type Shards []*Shard
@@ -112,7 +118,7 @@ func (shard *Shard) addNode(addr, role, password string) error {
 }
 
 func (shard *Shard) IsMigrating() bool {
-	return shard.MigratingSlot != nil && shard.TargetShardIndex != -1
+	return shard.MigratingSlot != nil && shard.MigratingSlot.IsMigrating && shard.TargetShardIndex != -1
 }
 
 func (shard *Shard) GetMasterNode() Node {
@@ -206,7 +212,7 @@ func (shard *Shard) promoteNewMaster(ctx context.Context, masterNodeID, preferre
 	return preferredNewMasterNode.ID(), nil
 }
 
-func (shard *Shard) HasOverlap(slotRange *SlotRange) bool {
+func (shard *Shard) HasOverlap(slotRange SlotRange) bool {
 	for _, shardSlotRange := range shard.SlotRanges {
 		if shardSlotRange.HasOverlap(slotRange) {
 			return true
@@ -261,7 +267,7 @@ func (shard *Shard) UnmarshalJSON(bytes []byte) error {
 	var data struct {
 		SlotRanges       []SlotRange    `json:"slot_ranges"`
 		TargetShardIndex int            `json:"target_shard_index"`
-		MigratingSlot    *SlotRange     `json:"migrating_slot"`
+		MigratingSlot    *MigratingSlot `json:"migrating_slot"`
 		Nodes            []*ClusterNode `json:"nodes"`
 	}
 	if err := json.Unmarshal(bytes, &data); err != nil {
diff --git a/store/cluster_shard_test.go b/store/cluster_shard_test.go
@@ -29,11 +29,11 @@ import (
 
 func TestShard_HasOverlap(t *testing.T) {
 	shard := NewShard()
-	slotRange := &SlotRange{Start: 0, Stop: 100}
-	shard.SlotRanges = append(shard.SlotRanges, *slotRange)
+	slotRange := SlotRange{Start: 0, Stop: 100}
+	shard.SlotRanges = append(shard.SlotRanges, slotRange)
 	require.True(t, shard.HasOverlap(slotRange))
-	require.True(t, shard.HasOverlap(&SlotRange{Start: 50, Stop: 150}))
-	require.False(t, shard.HasOverlap(&SlotRange{Start: 101, Stop: 150}))
+	require.True(t, shard.HasOverlap(SlotRange{Start: 50, Stop: 150}))
+	require.False(t, shard.HasOverlap(SlotRange{Start: 101, Stop: 150}))
 }
 
 func TestShard_Sort(t *testing.T) {
@@ -56,17 +56,22 @@ func TestShard_Sort(t *testing.T) {
 func TestShard_IsServicing(t *testing.T) {
 	var err error
 	shard := NewShard()
+	shard.TargetShardIndex = 0
+	shard.MigratingSlot = &MigratingSlot{IsMigrating: false}
+	require.False(t, shard.IsServicing())
+
 	shard.TargetShardIndex = 0
 	shard.MigratingSlot = nil
 	require.False(t, shard.IsServicing())
 
 	shard.TargetShardIndex = 0
-	shard.MigratingSlot, err = NewSlotRange(1, 1)
+	slotRange, err := NewSlotRange(1, 1)
 	require.Nil(t, err)
+	shard.MigratingSlot = FromSlotRange(slotRange)
 	require.True(t, shard.IsServicing())
 
 	shard.TargetShardIndex = -1
-	shard.MigratingSlot = nil
+	shard.MigratingSlot = &MigratingSlot{IsMigrating: false}
 	shard.SlotRanges = []SlotRange{{Start: 0, Stop: 100}}
 	require.True(t, shard.IsServicing())
 
diff --git a/store/cluster_test.go b/store/cluster_test.go
@@ -44,19 +44,19 @@ func TestCluster_FindIndexShardBySlot(t *testing.T) {
 
 	slotRange, err := NewSlotRange(0, 0)
 	require.NoError(t, err)
-	shard, err := cluster.findShardIndexBySlot(*slotRange)
+	shard, err := cluster.findShardIndexBySlot(slotRange)
 	require.NoError(t, err)
 	require.Equal(t, 0, shard)
 
 	slotRange, err = NewSlotRange(MaxSlotID/3+1, MaxSlotID/3+1)
 	require.NoError(t, err)
-	shard, err = cluster.findShardIndexBySlot(*slotRange)
+	shard, err = cluster.findShardIndexBySlot(slotRange)
 	require.NoError(t, err)
 	require.Equal(t, 1, shard)
 
 	slotRange, err = NewSlotRange(MaxSlotID, MaxSlotID)
 	require.NoError(t, err)
-	shard, err = cluster.findShardIndexBySlot(*slotRange)
+	shard, err = cluster.findShardIndexBySlot(slotRange)
 	require.NoError(t, err)
 	require.Equal(t, 2, shard)
 }
diff --git a/store/slot.go b/store/slot.go
@@ -44,24 +44,26 @@ type SlotRange struct {
 
 type SlotRanges []SlotRange
 
-func NewSlotRange(start, stop int) (*SlotRange, error) {
+type MigratingSlot struct {
+	SlotRange
+	IsMigrating bool
+}
+
+func NewSlotRange(start, stop int) (SlotRange, error) {
 	if start > stop {
-		return nil, errors.New("start was larger than stop")
+		return SlotRange{}, errors.New("start was larger than stop")
 	}
 	if (start < MinSlotID || start > MaxSlotID) ||
 		(stop < MinSlotID || stop > MaxSlotID) {
-		return nil, ErrSlotOutOfRange
+		return SlotRange{}, ErrSlotOutOfRange
 	}
-	return &SlotRange{
+	return SlotRange{
 		Start: start,
 		Stop:  stop,
 	}, nil
 }
 
-func (slotRange *SlotRange) Equal(that *SlotRange) bool {
-	if that == nil {
-		return false
-	}
+func (slotRange *SlotRange) Equal(that SlotRange) bool {
 	if slotRange.Start != that.Start {
 		return false
 	}
@@ -71,7 +73,7 @@ func (slotRange *SlotRange) Equal(that *SlotRange) bool {
 	return true
 }
 
-func (slotRange *SlotRange) HasOverlap(that *SlotRange) bool {
+func (slotRange *SlotRange) HasOverlap(that SlotRange) bool {
 	return slotRange.Stop >= that.Start && slotRange.Start <= that.Stop
 }
 
@@ -156,13 +158,84 @@ func (SlotRanges *SlotRanges) Contains(slot int) bool {
 
 func (SlotRanges *SlotRanges) HasOverlap(slotRange SlotRange) bool {
 	for _, slotRange := range *SlotRanges {
-		if slotRange.HasOverlap(&slotRange) {
+		if slotRange.HasOverlap(slotRange) {
 			return true
 		}
 	}
 	return false
 }
 
+func (s *SlotRange) Reset() {
+	s.Start = 0
+	s.Stop = 0
+}
+
+// FromSlotRange will return a MigratingSlot with the IsMigrating field set to true.
+// IsMigrating field would probably only be set to false from an unmarshal, like when
+// reading from the topology string
+func FromSlotRange(slotRange SlotRange) *MigratingSlot {
+	return &MigratingSlot{
+		SlotRange:   slotRange,
+		IsMigrating: true,
+	}
+}
+
+func (s *MigratingSlot) UnmarshalJSON(data []byte) error {
+	var slotsString any
+	if err := json.Unmarshal(data, &slotsString); err != nil {
+		return err
+	}
+	switch t := slotsString.(type) {
+	case string:
+		slotRange := SlotRange{}
+		err := json.Unmarshal(data, &slotRange)
+		if err != nil {
+			s.Reset()
+			return err
+		}
+		s.SlotRange = slotRange
+		s.IsMigrating = true
+	case float64:
+		// We use integer to represent the slot because we don't support the slot range
+		// in the past. So we need to support the integer type for backward compatibility.
+		// But the number in JSON is float64, so we need to convert it to int here.
+		if t == NotMigratingInt {
+			s.Reset()
+			return nil
+		}
+		if t < MinSlotID || t > MaxSlotID {
+			s.Reset()
+			return ErrSlotOutOfRange
+		}
+		slotID := int(t)
+		s.Start = slotID
+		s.Stop = slotID
+		s.IsMigrating = true
+	default:
+		s.Reset()
+		return fmt.Errorf("invalid slot range type: %T", slotsString)
+	}
+	return nil
+}
+
+func (s *MigratingSlot) MarshalJSON() ([]byte, error) {
+	if !s.IsMigrating {
+		// backwards compatibility. When we read from an old cluster that had `-1`
+		// denoting !isMigrating. The MigratingSlot field will not be nil. So when
+		// this field is marshal'd back into JSON format, we can keep it as it was
+		// which was `-1`.
+		// The only case this turns back to null is if a migration happens on this
+		// shard, and the function `ClearMigrateState()` is called on the shard.
+		return json.Marshal(NotMigratingInt)
+	}
+	return json.Marshal(s.String())
+}
+
+func (s *MigratingSlot) Reset() {
+	s.SlotRange.Reset()
+	s.IsMigrating = false
+}
+
 // CanMerge will return true if the given SlotRanges are adjacent with each other
 func CanMerge(a, b SlotRange) bool {
 	// Ensure a starts before b for easier comparison
@@ -218,7 +291,7 @@ func RemoveSlotFromSlotRanges(source SlotRanges, slot SlotRange) SlotRanges {
 	result := make([]SlotRange, 0, len(source))
 	for _, slotRange := range source {
 		// if no overlap, keep original range
-		if !slotRange.HasOverlap(&slot) {
+		if !slotRange.HasOverlap(slot) {
 			result = append(result, slotRange)
 			continue
 		}
diff --git a/store/slot_test.go b/store/slot_test.go

Original file line number	Diff line number	Diff line change
`@@ -51,7 +51,7 @@ func printCluster(cluster *store.Cluster) {`
`51`	`51`	`role = strings.ToUpper(store.RoleMaster)`
`52`	`52`	`}`
`53`	`53`	`migratingStatus := "NO"`
`54`		`- if shard.MigratingSlot != nil {`
	`54`	`+ if shard.IsMigrating() {`
`55`	`55`	`migratingStatus = fmt.Sprintf("%s --> %d", shard.MigratingSlot, shard.TargetShardIndex)`
`56`	`56`	`}`
`57`	`57`	`columns := []string{fmt.Sprintf("%d", i), node.ID(), node.Addr(), role, migratingStatus}`
Original file line number	Diff line number	Diff line change
`@@ -33,7 +33,7 @@ import (`
`33`	`33`
`34`	`34`	`type MigrateSlotRequest struct {`
`35`	`35`	Target int `json:"target" validate:"required"`
`36`		- Slot store.SlotRange `json:"slot" validate:"required"`
	`36`	+ Slot store.SlotRange `json:"slot" validate:"required"` // we don't use store.MigratingSlot here because we expect a valid SlotRange
`37`	`37`	SlotOnly bool `json:"slot_only"`
`38`	`38`	`}`
`39`	`39`
Original file line number	Diff line number	Diff line change
`@@ -181,7 +181,7 @@ func (cluster *Cluster) findShardIndexBySlot(slot SlotRange) (int, error) {`
`181`	`181`	`for i := 0; i < len(cluster.Shards); i++ {`
`182`	`182`	`slotRanges := cluster.Shards[i].SlotRanges`
`183`	`183`	`for _, slotRange := range slotRanges {`
`184`		`- if slotRange.HasOverlap(&slot) {`
	`184`	`+ if slotRange.HasOverlap(slot) {`
`185`	`185`	`if sourceShardIdx != -1 {`
`186`	`186`	`return sourceShardIdx, consts.ErrSlotRangeBelongsToMultipleShards`
`187`	`187`	`}`
`@@ -226,7 +226,7 @@ func (cluster *Cluster) MigrateSlot(ctx context.Context, slot SlotRange, targetS`
`226`	`226`	`}`
`227`	`227`
`228`	`228`	`// Will start the data migration in the background`
`229`		`- cluster.Shards[sourceShardIdx].MigratingSlot = &slot`
	`229`	`+ cluster.Shards[sourceShardIdx].MigratingSlot = FromSlotRange(slot)`
`230`	`230`	`cluster.Shards[sourceShardIdx].TargetShardIndex = targetShardIdx`
`231`	`231`	`return nil`
`232`	`232`	`}`
Original file line number	Diff line number	Diff line change
`@@ -85,9 +85,9 @@ type ClusterNode struct {`
`85`	`85`	`}`
`86`	`86`
`87`	`87`	`type ClusterInfo struct {`
`88`		- CurrentEpoch int64 `json:"cluster_current_epoch"`
`89`		- MigratingSlot *SlotRange `json:"migrating_slot"`
`90`		- MigratingState string `json:"migrating_state"`
	`88`	+ CurrentEpoch int64 `json:"cluster_current_epoch"`
	`89`	+ MigratingSlot *MigratingSlot `json:"migrating_slot"`
	`90`	+ MigratingState string `json:"migrating_state"`
`91`	`91`	`}`
`92`	`92`
`93`	`93`	`type ClusterNodeInfo struct {`
`@@ -195,10 +195,11 @@ func (n ClusterNode) GetClusterInfo(ctx context.Context) (ClusterInfo, error)`
`195`	`195`	`}`
`196`	`196`	`case "migrating_slot", "migrating_slot(s)":`
`197`	`197`	`// TODO(@git-hulk): handle multiple migrating slots`
`198`		`- clusterInfo.MigratingSlot, err = ParseSlotRange(fields[1])`
	`198`	`+ slotRange, err := ParseSlotRange(fields[1])`
`199`	`199`	`if err != nil {`
`200`	`200`	`return nil, err`
`201`	`201`	`}`
	`202`	`+ clusterInfo.MigratingSlot = FromSlotRange(*slotRange)`
`202`	`203`	`case "migrating_state":`
`203`	`204`	`clusterInfo.MigratingState = fields[1]`
`204`	`205`	`}`