shell: volume.tier.upload clean up replicated copies

fix https://github.com/seaweedfs/seaweedfs/issues/3804

1 files changed, 18 insertions, 5 deletions

diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go
index 83b67b7f9..9def3a176 100644
--- a/weed/shell/command_volume_tier_upload.go
+++ b/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) {
 	// find volume location
-	locations, found := commandEnv.MasterClient.GetLocations(uint32(vid))
+	existingLocations, found := commandEnv.MasterClient.GetLocations(uint32(vid))
 	if !found {
 		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 {
-		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
-	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 {
-		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