From f123aaa7457ad0bd8bc61fffbde9807f1e4b5298 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 12 Nov 2026 18:12:10 -0800 Subject: [PATCH] shell: replicate the downloaded volume before deleting its remote object volume.tier.download downloaded every replica from the remote tier and let the last one in the master's location list delete the shared object, so a replica missing from that list kept reading a deleted object. Download onto one replica, copy it over every other remote-backed replica, re-read the topology so replicas that show up meanwhile are replaced too, and only then delete the last remote-backed replica together with the object before copying the local volume onto it. Co-Authored-By: Claude Opus 5.5 --- weed/shell/command_volume_tier_download.go | 125 +++++++++++++++++---- 1 file changed, 104 insertions(+), 21 deletions(-) diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index 12f4fead3c..4784b64a65 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -43,7 +43,9 @@ func (c *commandVolumeTierDownload) Help() string { volume.tier.download -volumeId=7 volume.tier.download -volumeId=7 -concurrency=1 - This command will download the dat file of a volume from a remote tier to a volume server in local cluster. + This command downloads the dat file of a volume from a remote tier onto one replica, copies + that replica over every other replica, and only then deletes the remote object. A replica on a + volume server that is offline during the command keeps referencing the deleted object. ` } @@ -125,12 +127,6 @@ func collectRemoteVolumes(topoInfo *master_pb.TopologyInfo, collectionPattern st } func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, topoInfo *master_pb.TopologyInfo, collectionPattern string, vid needle.VolumeId, concurrency int) (err error) { - // find volume location - locations, found := commandEnv.MasterClient.GetLocationsClone(uint32(vid)) - if !found { - return fmt.Errorf("volume %d not found", vid) - } - // The volume server verifies the request's collection against the volume's // own, so the pattern must resolve to the actual collection here. collection, found, err := findVolumeCollection(topoInfo, vid, collectionPattern) @@ -140,28 +136,115 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, topoInfo *ma if !found { return fmt.Errorf("volume %d not found in collections matching %q", vid, collectionPattern) } + readTopology := func() (*master_pb.TopologyInfo, error) { + topo, _, topoErr := collectTopologyInfo(commandEnv, 0) + return topo, topoErr + } + return downloadAndReplicateFromRemoteTier(commandEnv.option.GrpcDialOption, writer, readTopology, vid, collection, concurrency) +} - // All replicas point at the same remote object; only the final download may delete - // it. Every earlier replica keeps it so the survivors are not left dangling. - // TODO parallelize this - for i, loc := range locations { - keepRemote := i < len(locations)-1 - // copy the .dat file from remote tier to local - err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote, concurrency) - if err != nil { - // A replica already made local by a prior interrupted run is not a - // failure; skip it so the remaining remote replicas still download. - if strings.Contains(err.Error(), "already on local disk") { - fmt.Fprintf(writer, "volume %d on %s is already on local disk, skipping\n", vid, loc.Url) - continue +// downloadAndReplicateFromRemoteTier downloads the shared remote object onto +// one replica, copies that local volume over every other replica, and only +// then deletes the remote object, so no replica is ever left referencing a +// deleted object. +func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, readTopology func() (*master_pb.TopologyInfo, error), vid needle.VolumeId, collection string, concurrency int) error { + topoInfo, err := readTopology() + if err != nil { + return err + } + remote, local := collectVolumeTierDownloadLocations(topoInfo, vid, collection) + if len(remote) == 0 { + fmt.Fprintf(writer, "volume %d is already on local disk\n", vid) + return nil + } + + source, fresh := pb.ServerAddress(""), true + if len(local) > 0 { + // a previous run already downloaded this replica + source = local[0] + } else { + source, remote, fresh = remote[0], remote[1:], false + // a lone replica is the only reference, so it may delete the object itself + keepRemote := len(remote) > 0 + if err = downloadDatFromRemoteTier(grpcDialOption, writer, vid, collection, source, keepRemote, concurrency); err != nil { + if !strings.Contains(err.Error(), "already on local disk") { + return fmt.Errorf("download dat file for volume %d to %s: %v", vid, source, err) } - return fmt.Errorf("download dat file for volume %d to %s: %v", vid, loc.Url, err) + fmt.Fprintf(writer, "volume %d on %s is already on local disk\n", vid, source) + } + if !keepRemote { + return nil } } + // Copy over all remote-backed replicas but one, re-reading the topology + // so a replica that was offline or created meanwhile is replaced too. The + // last one is deleted together with the remote object and then replaced. + replaced := map[pb.ServerAddress]bool{source: true} + for { + var pending []pb.ServerAddress + for _, server := range remote { + if !replaced[server] { + pending = append(pending, server) + } + } + if len(pending) == 0 { + fmt.Fprintf(writer, "volume %d: no remote-backed replica is listed any more, the remote object is kept\n", vid) + return nil + } + if len(pending) == 1 && fresh { + last := pending[0] + fmt.Fprintf(writer, "delete remote-backed replica of volume %d on %s with its remote object\n", vid, last) + if err = deleteVolume(context.Background(), grpcDialOption, vid, last, false, false, false); err != nil { + return fmt.Errorf("delete volume %d with its remote object on %s: %v", vid, last, err) + } + if err = replicateTierDownloadedVolume(grpcDialOption, writer, vid, source, last); err != nil { + return fmt.Errorf("%v; run volume.fix.replication to restore the replica", err) + } + return nil + } + for _, target := range pending[:len(pending)-1] { + if err = replicateTierDownloadedVolume(grpcDialOption, writer, vid, source, target); err != nil { + return err + } + replaced[target] = true + } + if topoInfo, err = readTopology(); err != nil { + return err + } + remote, _ = collectVolumeTierDownloadLocations(topoInfo, vid, collection) + fresh = true + } +} + +func replicateTierDownloadedVolume(grpcDialOption grpc.DialOption, writer io.Writer, vid needle.VolumeId, source, target pb.ServerAddress) error { + fmt.Fprintf(writer, "replicate downloaded volume %d from %s to %s\n", vid, source, target) + if _, err := copyVolume(context.Background(), grpcDialOption, writer, vid, source, target, "", 0, true); err != nil { + return fmt.Errorf("replicate volume %d from %s to %s: %v", vid, source, target, err) + } return nil } +// collectVolumeTierDownloadLocations splits the replicas of a volume into the +// ones still reading the remote object and the ones already on local disk. +func collectVolumeTierDownloadLocations(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collection string) (remote, local []pb.ServerAddress) { + eachDataNode(topoInfo, func(dc DataCenterId, rack RackId, dn *master_pb.DataNodeInfo) { + for _, disk := range dn.DiskInfos { + for _, vi := range disk.VolumeInfos { + if needle.VolumeId(vi.Id) != vid || vi.Collection != collection { + continue + } + if vi.RemoteStorageName != "" { + remote = append(remote, pb.NewServerAddressFromDataNode(dn)) + } else { + local = append(local, pb.NewServerAddressFromDataNode(dn)) + } + } + } + }) + return +} + func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool, concurrency int) error { err := operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error {