diff --git a/master/cluster_balance.go b/master/cluster_balance.go index 862f2a937..bb66a40aa 100644 --- a/master/cluster_balance.go +++ b/master/cluster_balance.go @@ -1083,6 +1083,13 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) { continue } + if checkPlanSourceChanged(mpPlan, mp) { + err = fmt.Errorf("skip rebalance meta partition(%d) because source changed", mpPlan.ID) + log.LogWarnf(err.Error()) + mpPlan.Msg = err.Error() + continue + } + err = c.waitForMetaPartitionReady(mp) if err != nil { log.LogErrorf("waitForMetaPartitionReady err: %s", err.Error()) @@ -1128,6 +1135,13 @@ func (c *Cluster) DoMetaPartitionBalanceTask(plan *proto.ClusterPlan) { return } + if verifyDestinationInMetaReplicas(mp, mrPlan.Destination) { + err = fmt.Errorf("destination %s is in mpid(%d) meta replicas[%v]", mrPlan.Destination, mp.PartitionID, mp.Hosts) + log.LogErrorf(err.Error()) + c.SetMetaReplicaPlanStatusError(plan, mrPlan, err.Error()) + return + } + log.LogDebugf("Start to migrate meta partition(%d) from %s to %s", mpPlan.ID, mrPlan.Source, mrPlan.Destination) err = c.migrateMetaPartition(mrPlan.Source, mrPlan.Destination, mp) if err != nil { @@ -1717,3 +1731,30 @@ func (c *Cluster) GetMpCountByMetaNode(addr string) int { } return ret } + +func checkPlanSourceChanged(mpPlan *proto.MetaBalancePlan, mp *MetaPartition) bool { + for _, item := range mpPlan.Original { + found := false + for _, mr := range mp.Replicas { + if mr.Addr == item.Source { + found = true + break + } + } + if !found { + return true + } + } + + return false +} + +func verifyDestinationInMetaReplicas(mp *MetaPartition, dst string) bool { + for _, item := range mp.Replicas { + if item.Addr == dst { + return true + } + } + + return false +} diff --git a/master/cluster_balance_test.go b/master/cluster_balance_test.go index 0a6ad2306..beaae4abd 100755 --- a/master/cluster_balance_test.go +++ b/master/cluster_balance_test.go @@ -7,6 +7,7 @@ import ( "testing" "github.com/cubefs/cubefs/proto" + "github.com/stretchr/testify/require" ) func TestGetMigrateDestAddr(t *testing.T) { @@ -1881,3 +1882,41 @@ func TestGetMetaNodePressureView(t *testing.T) { t.Errorf("GetMetaNodePressureView returned an unexpected plan") } } + +func TestCheckPlanSourceChanged(t *testing.T) { + mp := &MetaPartition{ + Replicas: []*MetaReplica{ + {Addr: "node1"}, + {Addr: "node2"}, + {Addr: "node3"}, + }, + } + mpPlan := &proto.MetaBalancePlan{ + Original: []*proto.MrBalanceInfo{ + {Source: "node1"}, + {Source: "node2"}, + {Source: "node3"}, + }, + } + ret := checkPlanSourceChanged(mpPlan, mp) + require.False(t, ret) + + mpPlan.Original[0].Source = "node4" + ret = checkPlanSourceChanged(mpPlan, mp) + require.True(t, ret) +} + +func TestVerifyDestinationInMetaReplicas(t *testing.T) { + mp := &MetaPartition{ + Replicas: []*MetaReplica{ + {Addr: "node1"}, + {Addr: "node2"}, + {Addr: "node3"}, + }, + } + ret := verifyDestinationInMetaReplicas(mp, "node1") + require.True(t, ret) + + ret = verifyDestinationInMetaReplicas(mp, "node4") + require.False(t, ret) +} diff --git a/master/cluster_task.go b/master/cluster_task.go index 82176befc..011e38a32 100644 --- a/master/cluster_task.go +++ b/master/cluster_task.go @@ -118,6 +118,12 @@ func (c *Cluster) migrateMetaPartition(srcAddr, targetAddr string, mp *MetaParti return fmt.Errorf("migrateMetaPartition src [%s] is not exist in mp(%d)", srcAddr, mp.PartitionID) } oldHosts = mp.Hosts + if targetAddr != "" && contains(mp.Hosts, targetAddr) { + mp.RUnlock() + err = fmt.Errorf("migrateMetaPartition target [%s] is already exist in mp(%d) hosts:[%v]", targetAddr, mp.PartitionID, mp.Hosts) + log.LogErrorf(err.Error()) + return err + } mp.RUnlock() if err = c.validateDecommissionMetaPartition(mp, srcAddr, false); err != nil {