fix(master): check the migrate plan destination not in mp hosts.#1000222192

Signed-off-by: Wu Huocheng <wuhuocheng@oppo.com>

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
+}

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)
+}

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 {