mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-13 23:39:33 +01:00
shell: keep the remote object while replicas are unlisted
Tier download deleted the shared object once the listed replicas were local, even when the replica placement expected more copies than the topology listed. Keep it in that case so a rerun can finish later. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
1 parent
7f83227560
commit
e19993cdb3
2 files changed
+50
-10
No files matched your search
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
||||
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
||||
)
|
||||
|
||||
@@ -45,8 +46,10 @@ func (c *commandVolumeTierDownload) Help() string {
|
||||
volume.tier.download -volumeId=7 -concurrency=1
|
||||
|
||||
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.
|
||||
that replica over every other replica, and only then deletes the remote object. The object is
|
||||
kept while fewer replicas are listed than the replica placement expects; rerun the command once
|
||||
they are back. A replica on an offline volume server that the placement does not account for
|
||||
keeps referencing the deleted object: delete that replica and run volume.fix.replication.
|
||||
|
||||
`
|
||||
}
|
||||
@@ -153,7 +156,8 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
remote, local := collectVolumeTierDownloadLocations(topoInfo, vid, collection)
|
||||
remote, local, copies := collectVolumeTierDownloadLocations(topoInfo, vid, collection)
|
||||
listed := len(remote) + len(local)
|
||||
if len(remote) == 0 {
|
||||
fmt.Fprintf(writer, "volume %d is already on local disk\n", vid)
|
||||
return nil
|
||||
@@ -167,14 +171,17 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
|
||||
} else {
|
||||
source, remote, fresh = remote[0].server, remote[1:], false
|
||||
// a lone replica is the only reference, so it may delete the object itself
|
||||
keepRemote := len(remote) > 0
|
||||
keepRemote := len(remote) > 0 || listed < copies
|
||||
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)
|
||||
}
|
||||
fmt.Fprintf(writer, "volume %d on %s is already on local disk\n", vid, source)
|
||||
}
|
||||
if !keepRemote {
|
||||
if len(remote) == 0 {
|
||||
if keepRemote {
|
||||
fmt.Fprintf(writer, "volume %d: only %d of %d replicas are listed, the remote object is kept\n", vid, listed, copies)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
}
|
||||
@@ -201,6 +208,10 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
|
||||
return nil
|
||||
}
|
||||
if len(pending) == 1 && fresh {
|
||||
if listed < copies {
|
||||
fmt.Fprintf(writer, "volume %d: only %d of %d replicas are listed, the remote object is kept\n", vid, listed, copies)
|
||||
return nil
|
||||
}
|
||||
last := pending[0]
|
||||
fmt.Fprintf(writer, "delete remote-backed replica of volume %d on %s with its remote object\n", vid, last.server)
|
||||
if err = deleteVolume(context.Background(), grpcDialOption, vid, last.server, false, false, false); err != nil {
|
||||
@@ -220,8 +231,8 @@ func downloadAndReplicateFromRemoteTier(grpcDialOption grpc.DialOption, writer i
|
||||
if topoInfo, err = readTopology(); err != nil {
|
||||
return err
|
||||
}
|
||||
remote, _ = collectVolumeTierDownloadLocations(topoInfo, vid, collection)
|
||||
fresh = true
|
||||
remote, local, copies = collectVolumeTierDownloadLocations(topoInfo, vid, collection)
|
||||
listed, fresh = len(remote)+len(local), true
|
||||
}
|
||||
}
|
||||
|
||||
@@ -276,14 +287,18 @@ type tierDownloadReplica struct {
|
||||
}
|
||||
|
||||
// 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 []tierDownloadReplica) {
|
||||
// ones still reading the remote object and the ones already on local disk, and
|
||||
// reports how many replicas the replica placement expects.
|
||||
func collectVolumeTierDownloadLocations(topoInfo *master_pb.TopologyInfo, vid needle.VolumeId, collection string) (remote, local []tierDownloadReplica, copies int) {
|
||||
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 rp, rpErr := super_block.NewReplicaPlacementFromByte(byte(vi.ReplicaPlacement)); rpErr == nil {
|
||||
copies = max(copies, rp.GetCopyCount())
|
||||
}
|
||||
replica := tierDownloadReplica{server: pb.NewServerAddressFromDataNode(dn), diskType: vi.DiskType}
|
||||
if vi.RemoteStorageName != "" {
|
||||
remote = append(remote, replica)
|
||||
|
||||
@@ -41,6 +41,7 @@ type tierDownloadCluster struct {
|
||||
frozen map[pb.ServerAddress]bool
|
||||
writable map[pb.ServerAddress]bool
|
||||
remounts map[pb.ServerAddress]int
|
||||
placement uint32
|
||||
}
|
||||
|
||||
type tierDownloadTestServer struct {
|
||||
@@ -203,7 +204,7 @@ func (c *tierDownloadCluster) readTopology() (*master_pb.TopologyInfo, error) {
|
||||
if !found || hidden[server] {
|
||||
continue
|
||||
}
|
||||
vi := &master_pb.VolumeInformationMessage{Id: 7, Collection: "c", DiskType: c.diskTypes[server]}
|
||||
vi := &master_pb.VolumeInformationMessage{Id: 7, Collection: "c", DiskType: c.diskTypes[server], ReplicaPlacement: c.placement}
|
||||
if state == tierReplicaRemote {
|
||||
vi.RemoteStorageName = "s3.default"
|
||||
vi.RemoteStorageKey = "key"
|
||||
@@ -332,3 +333,27 @@ func TestTierDownloadRestoresSourceAfterBlockingDeletes(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestTierDownloadKeepsRemoteObjectWhileReplicasAreUnlisted(t *testing.T) {
|
||||
for _, tc := range []struct {
|
||||
name string
|
||||
states []string
|
||||
}{
|
||||
{"lone listed replica", []string{tierReplicaRemote, tierReplicaRemote}},
|
||||
{"local source and one remote replica", []string{tierReplicaLocal, tierReplicaRemote, tierReplicaRemote}},
|
||||
{"three remote replicas", []string{tierReplicaRemote, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote}},
|
||||
} {
|
||||
c := newTierDownloadCluster(t, tc.states...)
|
||||
c.placement = uint32(len(tc.states) - 1) // "00n": n+1 copies, one more than listed
|
||||
offline := c.servers[len(c.servers)-1]
|
||||
for read := 0; read < 4; read++ {
|
||||
c.hide(read, offline)
|
||||
}
|
||||
if err := c.run(t); err != nil {
|
||||
t.Fatalf("%s: tier download: %v", tc.name, err)
|
||||
}
|
||||
if !c.objectExists {
|
||||
t.Errorf("%s: remote object deleted while the replica on %s is unlisted", tc.name, offline)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user