diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go index c2277d69d..ee89dfc17 100644 --- a/weed/shell/command_volume_fix_replication.go +++ b/weed/shell/command_volume_fix_replication.go @@ -7,6 +7,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "slices" @@ -43,6 +44,10 @@ func (c *commandVolumeFixReplication) Help() string { This command also finds all under-replicated volumes, and finds volume servers with free slots. If the free slots satisfy the replication requirement, the volume content is copied over and mounted. + Misplaced volumes with a surplus replica have a misplaced replica deleted. Without a surplus, a + well-placed replica is added first and the misplaced one is trimmed on a later pass, so the volume + never drops below its intended replica count. + volume.fix.replication # do not take action volume.fix.replication -apply # actually deleting or copying the volume files and mount the volume volume.fix.replication -collectionPattern=important* # fix any collections with prefix "important" @@ -137,14 +142,17 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, fmt.Fprintf(writer, "checking volume %d replication %s has %d replicas [%s]\n", replica.info.Id, replicaPlacement, len(replicas), strings.Join(locations, ", ")) } - switch { - case replicaPlacement.GetCopyCount() > len(replicas) || !satisfyReplicaCurrentLocation(replicaPlacement, replicas): + switch classifyReplicaSet(replicaPlacement, replicas) { + case replicaFixAddOne: underReplicatedVolumeIds = append(underReplicatedVolumeIds, vid) fmt.Fprintf(writer, "volume %d replication %s, but under replicated %+d\n", replica.info.Id, replicaPlacement, len(replicas)) - case isMisplaced(replicas, replicaPlacement): + case replicaFixAddOneBeforeTrim: + underReplicatedVolumeIds = append(underReplicatedVolumeIds, vid) + fmt.Fprintf(writer, "volume %d replication %s is not well placed [%s], adding a well-placed replica before trimming the misplaced one\n", replica.info.Id, replicaPlacement, strings.Join(locations, ", ")) + case replicaFixTrimMisplaced: misplacedVolumeIds = append(misplacedVolumeIds, vid) fmt.Fprintf(writer, "volume %d replication %s is not well placed [%s]\n", replica.info.Id, replicaPlacement, strings.Join(locations, ", ")) - case replicaPlacement.GetCopyCount() < len(replicas): + case replicaFixTrimOver: overReplicatedVolumeIds = append(overReplicatedVolumeIds, vid) fmt.Fprintf(writer, "volume %d replication %s, but over replicated %+d\n", replica.info.Id, replicaPlacement, len(replicas)) } @@ -155,6 +163,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, return fmt.Errorf("lock is lost") } + var deletedVolumeReplicas atomic.Int64 ewg.Reset() ewg.Add(func() error { // find the most underpopulated data nodes @@ -163,14 +172,18 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, }) if *doDelete { ewg.Add(func() error { - return c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, overReplicatedVolumeIds, volumeReplicas, allLocations, pickOneReplicaToDelete) + deleted, err := c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, overReplicatedVolumeIds, volumeReplicas, pickOneReplicaToDelete) + deletedVolumeReplicas.Add(int64(deleted)) + return err }) ewg.Add(func() error { - return c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, misplacedVolumeIds, volumeReplicas, allLocations, pickOneMisplacedVolume) + deleted, err := c.deleteOneVolume(commandEnv, writer, *applyChanges, *doCheck, misplacedVolumeIds, volumeReplicas, pickOneMisplacedVolume) + deletedVolumeReplicas.Add(int64(deleted)) + return err }) } if err := ewg.Wait(); err != nil { - return nil + return err } if !*applyChanges { @@ -206,10 +219,60 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv, } } } + + // Without progress the next pass would reclassify the same volumes and + // loop forever, e.g. when no destination can accept a replica; stop + // instead. Deletions count as progress since they free up slots. + if underReplicatedVolumeIdsCount > 0 && len(fixedVolumeReplicas) == 0 && deletedVolumeReplicas.Load() == 0 { + fmt.Fprintf(writer, "no progress made on %d under replicated volumes, stopping; free up capacity or adjust replica placement, then re-run\n", underReplicatedVolumeIdsCount) + break + } } return nil } +// replicaFix is the single next action volume.fix.replication takes on one +// volume's replica set. +type replicaFix int + +const ( + replicaFixNothing replicaFix = iota + // replicaFixAddOne: the volume lacks a replica, in count or in + // failure-domain spread; copy one to a well-placed destination. + replicaFixAddOne + // replicaFixAddOneBeforeTrim: the replica count is complete but a replica + // is misplaced. Add a well-placed replica first; a later pass trims the + // misplaced one as surplus. Deleting first would drop the volume below its + // intended durability — for good, if no destination can take the + // replacement copy. + replicaFixAddOneBeforeTrim + // replicaFixTrimMisplaced: more replicas than the policy asks for, at + // least one misplaced; delete a misplaced one. + replicaFixTrimMisplaced + // replicaFixTrimOver: more replicas than the policy asks for; delete a + // surplus one. + replicaFixTrimOver +) + +// classifyReplicaSet decides the next action for one volume. Add always wins +// over trim: a volume that lacks a replica and also has a misplaced or surplus +// one gets its missing replica first, and the trim happens on a later pass +// once the new copy registered in the topology. +func classifyReplicaSet(replicaPlacement *super_block.ReplicaPlacement, replicas []*VolumeReplica) replicaFix { + switch { + case replicaPlacement.GetCopyCount() > len(replicas) || !satisfyReplicaCurrentLocation(replicaPlacement, replicas): + return replicaFixAddOne + case isMisplaced(replicas, replicaPlacement): + if len(replicas) <= replicaPlacement.GetCopyCount() { + return replicaFixAddOneBeforeTrim + } + return replicaFixTrimMisplaced + case replicaPlacement.GetCopyCount() < len(replicas): + return replicaFixTrimOver + } + return replicaFixNothing +} + func collectVolumeReplicaLocations(topologyInfo *master_pb.TopologyInfo) (map[uint32][]*VolumeReplica, []location) { volumeReplicas := make(map[uint32][]*VolumeReplica) var allLocations []location @@ -278,10 +341,12 @@ func (c *commandVolumeFixReplication) matchCollectionPattern(collection string) return wildcard.MatchesWildcard(*c.collectionPattern, collection) } -func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, doCheck bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, selectOneVolumeFn SelectOneVolumeFunc) error { +// deleteOneVolume trims one replica from each of the given volumes, and +// reports how many replicas it actually deleted. +func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, doCheck bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, selectOneVolumeFn SelectOneVolumeFunc) (deleted int, err error) { if len(volumeIds) == 0 { // nothing to do - return nil + return 0, nil } for _, vid := range volumeIds { @@ -334,10 +399,12 @@ func (c *commandVolumeFixReplication) deleteOneVolume(commandEnv *CommandEnv, wr if err := deleteVolume(commandEnv.option.GrpcDialOption, needle.VolumeId(replica.info.Id), pb.NewServerAddressFromDataNode(replica.location.dataNode), false, true); err != nil { fmt.Fprintf(writer, "deleting volume %d from %s : %v", replica.info.Id, replica.location.dataNode.Id, err) + } else { + deleted++ } } - return nil + return deleted, nil } func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *CommandEnv, writer io.Writer, applyChanges bool, volumeIds []uint32, volumeReplicas map[uint32][]*VolumeReplica, allLocations []location, retryCount int, volumesPerStep int, maxParallelization int, maxParallelizationPerServer int) (fixedVolumes map[string]int, err error) { diff --git a/weed/shell/command_volume_fix_replication_test.go b/weed/shell/command_volume_fix_replication_test.go index 46d3c008b..5f9d71162 100644 --- a/weed/shell/command_volume_fix_replication_test.go +++ b/weed/shell/command_volume_fix_replication_test.go @@ -646,3 +646,88 @@ func TestSatisfyReplicaCurrentLocation(t *testing.T) { }) } } + +func TestClassifyReplicaSet(t *testing.T) { + + var tests = []struct { + name string + replication string + replicas []*VolumeReplica + expected replicaFix + }{ + { + // The full replica count is present but two copies crowd one rack. + // Deleting a misplaced replica first would drop the volume below its + // intended durability, so add a well-placed replica first. + name: "110 misplaced without surplus adds before trimming", + replication: "110", + replicas: []*VolumeReplica{ + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}}, + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn2"}}}, + {location: &location{"dc2", "r2", &master_pb.DataNodeInfo{Id: "dn3"}}}, + }, + expected: replicaFixAddOneBeforeTrim, + }, + { + // Same misplaced set once the well-placed replica registered: now + // there is a surplus, and the misplaced replica can be trimmed. + name: "110 misplaced with surplus trims", + replication: "110", + replicas: []*VolumeReplica{ + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}}, + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn2"}}}, + {location: &location{"dc1", "r3", &master_pb.DataNodeInfo{Id: "dn4"}}}, + {location: &location{"dc2", "r2", &master_pb.DataNodeInfo{Id: "dn3"}}}, + }, + expected: replicaFixTrimMisplaced, + }, + { + name: "010 missing a replica adds", + replication: "010", + replicas: []*VolumeReplica{ + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}}, + }, + expected: replicaFixAddOne, + }, + { + // Simultaneously over-provisioned in count and lacking rack spread: + // the missing failure domain is filled first, trims come later. + name: "010 surplus count but missing rack spread adds", + replication: "010", + replicas: []*VolumeReplica{ + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}}, + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn2"}}}, + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn3"}}}, + }, + expected: replicaFixAddOne, + }, + { + name: "010 well placed needs nothing", + replication: "010", + replicas: []*VolumeReplica{ + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}}, + {location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}}}, + }, + expected: replicaFixNothing, + }, + { + name: "000 duplicate copy trims", + replication: "000", + replicas: []*VolumeReplica{ + {location: &location{"dc1", "r1", &master_pb.DataNodeInfo{Id: "dn1"}}}, + {location: &location{"dc1", "r2", &master_pb.DataNodeInfo{Id: "dn2"}}}, + }, + expected: replicaFixTrimMisplaced, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + replicaPlacement, _ := super_block.NewReplicaPlacementFromString(tt.replication) + if got := classifyReplicaSet(replicaPlacement, tt.replicas); got != tt.expected { + t.Errorf("%s: expect %v got %v for %v %+v", + tt.name, tt.expected, got, tt.replication, tt.replicas) + } + }) + } +}