Browse Source

shell: volume.tier.upload clean up replicated copies

fix https://github.com/seaweedfs/seaweedfs/issues/3804
pull/3865/head
chrislu 2 years ago
parent
commit
52e0a88a15
  1. 23
      weed/shell/command_volume_tier_upload.go

23
weed/shell/command_volume_tier_upload.go

@ -97,20 +97,33 @@ func (c *commandVolumeTierUpload) Do(args []string, commandEnv *CommandEnv, writ
func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool) (err error) { func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection string, vid needle.VolumeId, dest string, keepLocalDatFile bool) (err error) {
// find volume location // find volume location
locations, found := commandEnv.MasterClient.GetLocations(uint32(vid))
existingLocations, found := commandEnv.MasterClient.GetLocations(uint32(vid))
if !found { if !found {
return fmt.Errorf("volume %d not found", vid) return fmt.Errorf("volume %d not found", vid)
} }
err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, needle.VolumeId(vid), locations, false)
err = markVolumeReplicasWritable(commandEnv.option.GrpcDialOption, vid, existingLocations, false)
if err != nil { if err != nil {
return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, locations[0].Url, err)
return fmt.Errorf("mark volume %d as readonly on %s: %v", vid, existingLocations[0].Url, err)
} }
// copy the .dat file to remote tier // copy the .dat file to remote tier
err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, locations[0].ServerAddress(), dest, keepLocalDatFile)
err = uploadDatToRemoteTier(commandEnv.option.GrpcDialOption, writer, vid, collection, existingLocations[0].ServerAddress(), dest, keepLocalDatFile)
if err != nil { if err != nil {
return fmt.Errorf("copy dat file for volume %d on %s to %s: %v", vid, locations[0].Url, dest, err)
return fmt.Errorf("copy dat file for volume %d on %s to %s: %v", vid, existingLocations[0].Url, dest, err)
}
// now the first replica has the .idx and .vif files.
// ask replicas on other volume server to delete its own local copy
for i, location := range existingLocations {
if i == 0 {
break
}
fmt.Printf("delete volume %d from %s\n", vid, location.Url)
err = deleteVolume(commandEnv.option.GrpcDialOption, vid, location.ServerAddress())
if err != nil {
return fmt.Errorf("deleteVolume %s volume %d: %v", location.Url, vid, err)
}
} }
return nil return nil

Loading…
Cancel
Save