volume.fix.replication: fix misplaced volumes

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

1 files changed, 56 insertions, 4 deletions

diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go
index c003eea91..4a4ed5f9f 100644
--- a/weed/shell/command_volume_fix_replication.go
+++ b/weed/shell/command_volume_fix_replication.go
@@ -89,7 +89,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 		}
 
 		// find all under replicated volumes
-		var underReplicatedVolumeIds, overReplicatedVolumeIds []uint32
+		var underReplicatedVolumeIds, overReplicatedVolumeIds, misplacedVolumeIds []uint32
 		for vid, replicas := range volumeReplicas {
 			replica := replicas[0]
 			replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
@@ -98,11 +98,20 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 			} else if replicaPlacement.GetCopyCount() < len(replicas) {
 				overReplicatedVolumeIds = append(overReplicatedVolumeIds, vid)
 				fmt.Fprintf(writer, "volume %d replication %s, but over replicated %+d\n", replica.info.Id, replicaPlacement, len(replicas))
+			} else if isMisplaced(replicas, replicaPlacement) {
+				misplacedVolumeIds = append(misplacedVolumeIds, vid)
+				fmt.Fprintf(writer, "volume %d replication %s is not well placed %+v\n", replica.info.Id, replicaPlacement, replicas)
 			}
 		}
 
 		if len(overReplicatedVolumeIds) > 0 {
-			if err := c.fixOverReplicatedVolumes(commandEnv, writer, takeAction, overReplicatedVolumeIds, volumeReplicas, allLocations); err != nil {
+			if err := c.deleteOneVolume(commandEnv, writer, takeAction, overReplicatedVolumeIds, volumeReplicas, allLocations, pickOneReplicaToDelete); err != nil {
+				return err
+			}
+		}
+
+		if len(misplacedVolumeIds) > 0 {
+			if err := c.deleteOneVolume(commandEnv, writer, takeAction, misplacedVolumeIds, volumeReplicas, allLocations, pickOneMisplacedVolume); err != nil {
 				return err
 			}
 		}
@@ -171,12 +180,14 @@ func collectVolumeReplicaLocations(topologyInfo *master_pb.TopologyInfo) (map[ui
 	return volumeReplicas, allLocations
 }
 
-func (c *commandVolumeFixReplication) fixOverReplicatedVolumes(commandEnv *CommandEnv, writer io.Writer, takeAction bool, overReplicatedVolumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location) error {
+type SelectOneVolumeFunc func(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) *VolumeReplica
+
+func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, takeAction bool, overReplicatedVolumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, selectOneVolumeFn SelectOneVolumeFunc) error {
 	for _, vid := range overReplicatedVolumeIds {
 		replicas := volumeReplicas[vid]
 		replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replicas[0].info.ReplicaPlacement))
 
-		replica := pickOneReplicaToDelete(replicas, replicaPlacement)
+		replica := selectOneVolumeFn(replicas, replicaPlacement)
 
 		// check collection name pattern
 		if *c.collectionPattern != "" {
@@ -495,3 +506,44 @@ func pickOneReplicaToDelete(replicas []*VolumeReplica, replicaPlacement *super_b
 	return replicas[0]
 
 }
+
+// check and fix misplaced volumes
+
+func isMisplaced(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) bool {
+
+	for i := 0; i < len(replicas); i++ {
+		others := otherThan(replicas, i)
+		if satisfyReplicaPlacement(replicaPlacement, others, *replicas[i].location) {
+			return false
+		}
+	}
+
+	return true
+
+}
+
+func otherThan(replicas []*VolumeReplica, index int) (others []*VolumeReplica) {
+	for i := 0; i < len(replicas); i++ {
+		if index != i {
+			others = append(others, replicas[i])
+		}
+	}
+	return
+}
+
+func pickOneMisplacedVolume(replicas []*VolumeReplica, replicaPlacement *super_block.ReplicaPlacement) (toDelete *VolumeReplica) {
+
+	var deletionCandidates []*VolumeReplica
+	for i := 0; i < len(replicas); i++ {
+		others := otherThan(replicas, i)
+		if !isMisplaced(others, replicaPlacement) {
+			deletionCandidates = append(deletionCandidates, replicas[i])
+		}
+	}
+	if len(deletionCandidates) > 0 {
+		return pickOneReplicaToDelete(deletionCandidates, replicaPlacement)
+	}
+
+	return pickOneReplicaToDelete(replicas, replicaPlacement)
+
+}