Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-11-13 23:39:33 +01:00
Code Issues Packages Projects Releases Wiki Activity

shell: replicate the downloaded volume before deleting its remote object

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Chris LuandClaude Opus 5.5 committed 2026-11-12 18:12:10 -08:00
1 parent 917cf534db
commit f123aaa745
1 file changed
+104 -21
Download Patch File Download Diff File Expand all files Collapse all files
weed/shell/command_volume_tier_download.go
+104 -21
View File
@@ -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 {