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 {