diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 268aeab10f..69f1bdcc56 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1234,12 +1234,24 @@ impl VolumeServer for VolumeGrpcService { DiskType::from_string(&vol_info.disk_type) }; + let has_remote_dat = vol_info + .volume_info + .as_ref() + .map(|vi| !vi.files.is_empty()) + .unwrap_or(false); + // a remote-backed volume only lands its .idx/.vif locally; the .dat stays in the tier + let needed_space = if has_remote_dat { + vol_info.idx_file_size + } else { + vol_info.dat_file_size + }; + // Find a free disk location using Go's Store.FindFreeLocation semantics. let (data_base, idx_base, selected_disk_type) = { let store = self.state.store.read().unwrap(); let Some(loc_idx) = store.find_free_location_predicate(|loc| { loc.disk_type == requested_disk_type - && loc.available_space.load(Ordering::Relaxed) > vol_info.dat_file_size + && loc.available_space.load(Ordering::Relaxed) > needed_space }) else { return Err(Status::internal(format!( "no space left {}", @@ -1266,12 +1278,6 @@ impl VolumeServer for VolumeGrpcService { Status::internal(format!("write .note for volume {}: {}", vid, e)) })?; - let has_remote_dat = vol_info - .volume_info - .as_ref() - .map(|vi| !vi.files.is_empty()) - .unwrap_or(false); - let (tx, rx) = tokio::sync::mpsc::channel::>(16); let state = self.state.clone(); diff --git a/weed/server/volume_grpc_copy.go b/weed/server/volume_grpc_copy.go index 0d372982f3..47119b59b2 100644 --- a/weed/server/volume_grpc_copy.go +++ b/weed/server/volume_grpc_copy.go @@ -74,9 +74,15 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre if req.DiskType != "" { diskType = req.DiskType } + hasRemoteDatFile = volFileInfoResp.VolumeInfo != nil && len(volFileInfoResp.VolumeInfo.Files) > 0 + // a remote-backed volume only lands its .idx/.vif locally; the .dat stays in the tier + neededSpace := volFileInfoResp.DatFileSize + if hasRemoteDatFile { + neededSpace = volFileInfoResp.IdxFileSize + } location := vs.store.FindFreeLocation(func(location *storage.DiskLocation) bool { return location.DiskType == types.ToDiskType(diskType) && - location.AvailableSpace.Load() > volFileInfoResp.DatFileSize + location.AvailableSpace.Load() > neededSpace }) if location == nil { return fmt.Errorf("%s %s", util.ErrVolumeNoSpaceLeft, types.ToDiskType(diskType).ReadableString()) @@ -84,7 +90,6 @@ func (vs *VolumeServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, stre dataBaseFileName = storage.VolumeFileName(location.Directory, volFileInfoResp.Collection, int(req.VolumeId)) indexBaseFileName = storage.VolumeFileName(location.IdxDirectory, volFileInfoResp.Collection, int(req.VolumeId)) - hasRemoteDatFile = volFileInfoResp.VolumeInfo != nil && len(volFileInfoResp.VolumeInfo.Files) > 0 // The .note marks the copy as in-progress; a leftover note fails the // volume load on restart, so a write failure must abort the copy. diff --git a/weed/shell/command_volume_tier_upload.go b/weed/shell/command_volume_tier_upload.go index 38ee1c3cf4..68678d264d 100644 --- a/weed/shell/command_volume_tier_upload.go +++ b/weed/shell/command_volume_tier_upload.go @@ -58,6 +58,9 @@ func (c *commandVolumeTierUpload) Help() string { The index file is still local, and the same O(1) disk read is applied to the remote file. + Each replica keeps its own local index pointing at the same remote object, + so the volume keeps its replica count for reads after tiering. + ` } @@ -119,22 +122,7 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str return fmt.Errorf("collect topology info: %v", err) } - var existingLocations []wdclient.Location - 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 && (collection == "" || vi.Collection == collection) { - fmt.Printf("find volume %d from Url:%s, GrpcPort:%d, DC:%s\n", vid, dn.Id, dn.GrpcPort, string(dc)) - existingLocations = append(existingLocations, wdclient.Location{ - Url: dn.Id, - PublicUrl: dn.Id, - GrpcPort: int(dn.GrpcPort), - DataCenter: string(dc), - }) - } - } - } - }) + existingLocations := collectVolumeTierUploadLocations(topoInfo, vid, collection, writer) if len(existingLocations) == 0 { if collection == "" { @@ -157,25 +145,51 @@ func doVolumeTierUpload(commandEnv *CommandEnv, writer io.Writer, collection str if keepLocalDatFile { return nil } - // now the first replica has the .idx and .vif files. - // ask replicas on other volume server to delete its own local copy + // Re-copy the uploaded replica's .idx/.vif onto the other replicas instead + // of deleting them: the remote object key lives only in the .vif, so losing + // the single server holding it would orphan the volume. for i, location := range existingLocations { if i == 0 { continue } - fmt.Printf("delete volume %d from Url:%s\n", vid, location.Url) - // Other replicas were never tier-uploaded; their .vif (if any) points - // to the same cloud key just written for this volume. Keep the remote - // object so the tier-uploaded replica still references valid data. - err = deleteVolume(commandEnv.option.GrpcDialOption, vid, location.ServerAddress(), false, true) + fmt.Fprintf(writer, "replicate remote volume %d metadata from %s to %s\n", vid, existingLocations[0].Url, location.Url) + err = replicateVolumeToServer(commandEnv.option.GrpcDialOption, writer, vid, existingLocations[0].ServerAddress(), location.ServerAddress(), "") if err != nil { - return fmt.Errorf("deleteVolume %s volume %d: %v", location.Url, vid, err) + return fmt.Errorf("replicate volume %d from %s to %s: %v", vid, existingLocations[0].Url, location.Url, err) } } return nil } +// collectVolumeTierUploadLocations lists the replica locations of a volume, +// putting an already-tiered replica first so a rerun after a partial failure +// reuses its remote object instead of uploading a second copy. +func collectVolumeTierUploadLocations(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collection string, writer io.Writer) []wdclient.Location { + var tiered, local []wdclient.Location + 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 && (collection == "" || vi.Collection == collection) { + fmt.Fprintf(writer, "find volume %d from Url:%s, GrpcPort:%d, DC:%s\n", vid, dn.Id, dn.GrpcPort, string(dc)) + loc := wdclient.Location{ + Url: dn.Id, + PublicUrl: dn.Id, + GrpcPort: int(dn.GrpcPort), + DataCenter: string(dc), + } + if vi.RemoteStorageKey != "" { + tiered = append(tiered, loc) + } else { + local = append(local, loc) + } + } + } + } + }) + return append(tiered, local...) +} + func uploadDatToRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, sourceVolumeServer pb.ServerAddress, dest string, keepLocalDatFile bool) error { err := operation.WithVolumeServerClient(true, sourceVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { diff --git a/weed/shell/command_volume_tier_upload_test.go b/weed/shell/command_volume_tier_upload_test.go new file mode 100644 index 0000000000..ac618700b5 --- /dev/null +++ b/weed/shell/command_volume_tier_upload_test.go @@ -0,0 +1,56 @@ +package shell + +import ( + "io" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +func tierUploadTopo(nodes map[string]*master_pb.VolumeInformationMessage) *master_pb.TopologyInfo { + var dns []*master_pb.DataNodeInfo + for _, id := range []string{"n1", "n2", "n3"} { + vi, found := nodes[id] + if !found { + continue + } + dns = append(dns, &master_pb.DataNodeInfo{ + Id: id, + DiskInfos: map[string]*master_pb.DiskInfo{"": {VolumeInfos: []*master_pb.VolumeInformationMessage{vi}}}, + }) + } + return &master_pb.TopologyInfo{ + DataCenterInfos: []*master_pb.DataCenterInfo{{ + RackInfos: []*master_pb.RackInfo{{DataNodeInfos: dns}}, + }}, + } +} + +// A replica that is already tiered must become the upload source, so a rerun +// after a partial failure reuses its remote object instead of uploading a +// second copy under a new key. +func TestCollectVolumeTierUploadLocationsPrefersTieredReplica(t *testing.T) { + topo := tierUploadTopo(map[string]*master_pb.VolumeInformationMessage{ + "n1": {Id: 1, Collection: "c"}, + "n2": {Id: 1, Collection: "c", RemoteStorageName: "s3.default", RemoteStorageKey: "some-key"}, + "n3": {Id: 1, Collection: "c"}, + }) + locations := collectVolumeTierUploadLocations(topo, needle.VolumeId(1), "c", io.Discard) + if len(locations) != 3 { + t.Fatalf("want 3 locations, got %d", len(locations)) + } + if locations[0].Url != "n2" { + t.Fatalf("want tiered replica n2 first, got %s", locations[0].Url) + } +} + +func TestCollectVolumeTierUploadLocationsFiltersCollection(t *testing.T) { + topo := tierUploadTopo(map[string]*master_pb.VolumeInformationMessage{ + "n1": {Id: 1, Collection: "c"}, + "n2": {Id: 1, Collection: "other"}, + }) + if locations := collectVolumeTierUploadLocations(topo, needle.VolumeId(1), "c", io.Discard); len(locations) != 1 || locations[0].Url != "n1" { + t.Fatalf("want only n1, got %+v", locations) + } +}