mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-29 08:20:42 +01:00
shell: volume.tier.upload keeps volume replicas (#10314)
* volume: copying a remote-backed volume only needs space for the index VolumeCopy sized its target-location check by the source .dat even when that .dat lives in a cloud tier and only .idx/.vif land locally, so re-replicating a tiered volume demanded the full remote size in free disk. Require the index size instead. * shell: volume.tier.upload keeps volume replicas Tiering a replicated volume deleted every replica but the upload source, leaving one server holding the only .idx and the only .vif that knows the remote object key — losing that server orphaned the volume even though its data sat intact in the cloud. Replicate the uploaded .idx/.vif onto the other replica servers instead (VolumeCopy skips the .dat for remote-backed volumes), so all replicas serve reads from the same remote object and the volume keeps its replica count. An already-tiered replica is preferred as the upload source, so a rerun after a partial failure reuses the existing remote object instead of uploading a second copy under a new key. * shell: group tier upload locations instead of re-prepending * rust volume: copying a remote-backed volume only needs space for the index Mirror the Go VolumeCopy change: size the free-location check by the source .idx when the .dat lives in a cloud tier, since only .idx/.vif land locally.
This commit is contained in:
1 parent
40ee4a08c5
commit
c1a1e3c1e3
4 files changed
+114
-33
No files matched your search
@@ -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::<Result<volume_server_pb::VolumeCopyResponse, Status>>(16);
|
||||
let state = self.state.clone();
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user