lsm/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-11-08 18:16:46 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

ec.encode: Fix bug causing source volumes not being deleted after EC conversion. (#6447)

Browse Source 
This logic was originally part of `spreadEcShards()`, which got removed during
the unification effort with `ec.balance` (https://github.com/seaweedfs/seaweedfs/pull/6344),
accidentally breaking functionality in the process.

The commit restores the deletion code for EC'd volumes - with parallelization support.
This commit is contained in:
Lisandro Pin
2025-01-17 10:02:30 +01:00
committed by GitHub
parent c7ae969c06
commit eab2e0e112
1 changed files with 18 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
weed/shell/command_ec_encode.go
View File

@@ -132,6 +132,8 @@ func (c *commandEcEncode) Do(args []string, commandEnv *CommandEnv, writer io.Wr
} }
func doEcEncode(commandEnv *CommandEnv, collection string, vid needle.VolumeId, maxParallelization int) error { func doEcEncode(commandEnv *CommandEnv, collection string, vid needle.VolumeId, maxParallelization int) error {
var ewg *ErrorWaitGroup
if !commandEnv.isLocked() { if !commandEnv.isLocked() {
return fmt.Errorf("lock is lost") return fmt.Errorf("lock is lost")
} }
@@ -143,7 +145,7 @@ func doEcEncode(commandEnv *CommandEnv, collection string, vid needle.VolumeId,
} }
// mark the volume as readonly // mark the volume as readonly
ewg := NewErrorWaitGroup(maxParallelization) ewg = NewErrorWaitGroup(maxParallelization)
for _, location := range locations { for _, location := range locations {
ewg.Add(func() error { ewg.Add(func() error {
if err := markVolumeReplicaWritable(commandEnv.option.GrpcDialOption, vid, location, false, false); err != nil { if err := markVolumeReplicaWritable(commandEnv.option.GrpcDialOption, vid, location, false, false); err != nil {
@@ -161,6 +163,21 @@ func doEcEncode(commandEnv *CommandEnv, collection string, vid needle.VolumeId,
return fmt.Errorf("generate ec shards for volume %d on %s: %v", vid, locations[0].Url, err) return fmt.Errorf("generate ec shards for volume %d on %s: %v", vid, locations[0].Url, err)
} }
// ask the source volume server to delete the original volume
ewg = NewErrorWaitGroup(maxParallelization)
for _, location := range locations {
ewg.Add(func() error {
if err := deleteVolume(commandEnv.option.GrpcDialOption, vid, location.ServerAddress(), false); err != nil {
return fmt.Errorf("deleteVolume %s volume %d: %v", location.Url, vid, err)
}
fmt.Printf("deleted volume %d from %s\n", vid, location.Url)
return nil
})
}
if err := ewg.Wait(); err != nil {
return err
}
return nil return nil
} }