shell: ensure dc and rack does not change for replicated volumes

fix https://github.com/chrislusf/seaweedfs/issues/1203

1 files changed, 17 insertions, 11 deletions

diff --git a/weed/shell/command_volume_balance.go b/weed/shell/command_volume_balance.go
index bed4f4306..488beb998 100644
--- a/weed/shell/command_volume_balance.go
+++ b/weed/shell/command_volume_balance.go
@@ -109,14 +109,7 @@ func (c *commandVolumeBalance) Do(args []string, commandEnv *CommandEnv, writer
 	return nil
 }
 
-func balanceVolumeServers(commandEnv *CommandEnv, dataNodeInfos []*master_pb.DataNodeInfo, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
-
-	var nodes []*Node
-	for _, dn := range dataNodeInfos {
-		nodes = append(nodes, &Node{
-			info: dn,
-		})
-	}
+func balanceVolumeServers(commandEnv *CommandEnv, nodes []*Node, volumeSizeLimit uint64, collection string, applyBalancing bool) error {
 
 	// balance writable volumes
 	for _, n := range nodes {
@@ -151,15 +144,19 @@ func balanceVolumeServers(commandEnv *CommandEnv, dataNodeInfos []*master_pb.Dat
 	return nil
 }
 
-func collectVolumeServersByType(t *master_pb.TopologyInfo, selectedDataCenter string) (typeToNodes map[uint64][]*master_pb.DataNodeInfo) {
-	typeToNodes = make(map[uint64][]*master_pb.DataNodeInfo)
+func collectVolumeServersByType(t *master_pb.TopologyInfo, selectedDataCenter string) (typeToNodes map[uint64][]*Node) {
+	typeToNodes = make(map[uint64][]*Node)
 	for _, dc := range t.DataCenterInfos {
 		if selectedDataCenter != "" && dc.Id != selectedDataCenter {
 			continue
 		}
 		for _, r := range dc.RackInfos {
 			for _, dn := range r.DataNodeInfos {
-				typeToNodes[dn.MaxVolumeCount] = append(typeToNodes[dn.MaxVolumeCount], dn)
+				typeToNodes[dn.MaxVolumeCount] = append(typeToNodes[dn.MaxVolumeCount], &Node{
+					info: dn,
+					dc:   dc.Id,
+					rack: r.Id,
+				})
 			}
 		}
 	}
@@ -169,6 +166,8 @@ func collectVolumeServersByType(t *master_pb.TopologyInfo, selectedDataCenter st
 type Node struct {
 	info            *master_pb.DataNodeInfo
 	selectedVolumes map[uint32]*master_pb.VolumeInformationMessage
+	dc              string
+	rack            string
 }
 
 func sortWritableVolumes(volumes []*master_pb.VolumeInformationMessage) {
@@ -210,6 +209,13 @@ func balanceSelectedVolume(commandEnv *CommandEnv, nodes []*Node, sortCandidates
 			sortCandidatesFn(candidateVolumes)
 
 			for _, v := range candidateVolumes {
+				if v.ReplicaPlacement > 0 {
+					if fullNode.dc != emptyNode.dc && fullNode.rack != emptyNode.rack {
+						// TODO this logic is too simple, but should work most of the time
+						// Need a correct algorithm to handle all different cases
+						continue
+					}
+				}
 				if _, found := emptyNode.selectedVolumes[v.Id]; !found {
 					if err := moveVolume(commandEnv, v, fullNode, emptyNode, applyBalancing); err == nil {
 						delete(fullNode.selectedVolumes, v.Id)