diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index 12f4fead3c..f028fe43a7 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -2,6 +2,7 @@ package shell import ( "context" + "errors" "flag" "fmt" "io" @@ -15,6 +16,8 @@ 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" ) func init() { @@ -43,7 +46,11 @@ 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. 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. ` } @@ -125,12 +132,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 +141,202 @@ 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) (err error) { + topoInfo, err := readTopology() + if err != nil { + return err + } + remote, local, copies := collectVolumeTierDownloadLocations(topoInfo, vid, collection) + listed := len(remote) + len(local) + if len(remote) == 0 { + if len(local) == 0 { + return fmt.Errorf("volume %d is not listed on any volume server", vid) + } + fmt.Fprintf(writer, "volume %d is already on local disk\n", vid) + return nil + } + + var source pb.ServerAddress + fresh := true + if len(local) > 0 { + // a previous run already downloaded this replica + source = local[0].server + } 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 || 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) } - 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 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 } } + restore, err := freezeTierDownloadSource(grpcDialOption, vid, source) + if restore != nil { + defer func() { + if restoreErr := restore(); restoreErr != nil { + err = errors.Join(err, fmt.Errorf("restore volume %d state on %s: %w", vid, source, restoreErr)) + } + }() + } + if err != nil { + return fmt.Errorf("block deletes on volume %d at %s: %v", vid, source, err) + } + + // 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 []tierDownloadReplica + for _, replica := range remote { + if !replaced[replica.server] { + pending = append(pending, replica) + } + } + 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 { + 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 + } + if server, found := serverListedTwice(remote, local); found { + fmt.Fprintf(writer, "volume %d is listed twice on %s, the remote object is kept\n", vid, server) + 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 { + return fmt.Errorf("delete volume %d with its remote object on %s: %v; if the delete was applied, run volume.fix.replication to restore the replica", vid, last.server, 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.server] = true + } + if topoInfo, err = readTopology(); err != nil { + return err + } + remote, local, copies = collectVolumeTierDownloadLocations(topoInfo, vid, collection) + listed, fresh = len(remote)+len(local), true + } +} + +// freezeTierDownloadSource blocks deletes on the copy source: a delete applied +// after VolumeCopy read the source index would be lost on the replaced replica. +// restore brings back the writable mark, or reloads the persisted state. It is +// also returned when the mark fails, since the server may have applied it. +func freezeTierDownloadSource(grpcDialOption grpc.DialOption, vid needle.VolumeId, source pb.ServerAddress) (restore func() error, err error) { + var wasWritable, statusRead bool + err = operation.WithVolumeServerClient(false, source, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + resp, statusErr := client.VolumeStatus(context.Background(), &volume_server_pb.VolumeStatusRequest{VolumeId: uint32(vid)}) + if statusErr != nil { + return statusErr + } + wasWritable, statusRead = !resp.IsReadOnly, true + _, markErr := client.VolumeMarkReadonly(context.Background(), &volume_server_pb.VolumeMarkReadonlyRequest{VolumeId: uint32(vid)}) + return markErr + }) + if !statusRead { + return nil, err + } + restore = func() error { + return operation.WithVolumeServerClient(false, source, grpcDialOption, func(client volume_server_pb.VolumeServerClient) error { + if wasWritable { + _, markErr := client.VolumeMarkWritable(context.Background(), &volume_server_pb.VolumeMarkWritableRequest{VolumeId: uint32(vid)}) + return markErr + } + return remountVolume(client, vid) + }) + } + return +} + +func remountVolume(client volume_server_pb.VolumeServerClient, vid needle.VolumeId) error { + if _, err := client.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{VolumeId: uint32(vid)}); err != nil { + return err + } + _, err := client.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{VolumeId: uint32(vid)}) + return err +} + +func replicateTierDownloadedVolume(grpcDialOption grpc.DialOption, writer io.Writer, vid needle.VolumeId, source pb.ServerAddress, target tierDownloadReplica) error { + fmt.Fprintf(writer, "replicate downloaded volume %d from %s to %s\n", vid, source, target.server) + if _, err := copyVolume(context.Background(), grpcDialOption, writer, vid, source, target.server, types.ToDiskType(target.diskType).ReadableString(), 0, true); err != nil { + return fmt.Errorf("replicate volume %d from %s to %s: %v", vid, source, target.server, err) + } return nil } +func serverListedTwice(remote, local []tierDownloadReplica) (pb.ServerAddress, bool) { + seen := make(map[pb.ServerAddress]bool) + for _, replica := range append(append([]tierDownloadReplica{}, remote...), local...) { + if seen[replica.server] { + return replica.server, true + } + seen[replica.server] = true + } + return "", false +} + +type tierDownloadReplica struct { + server pb.ServerAddress + diskType string +} + +// collectVolumeTierDownloadLocations splits the replicas of a volume into the +// 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) + } else { + local = append(local, replica) + } + } + } + }) + 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 { @@ -193,21 +368,7 @@ func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, return downloadErr } - _, unmountErr := volumeServerClient.VolumeUnmount(context.Background(), &volume_server_pb.VolumeUnmountRequest{ - VolumeId: uint32(volumeId), - }) - if unmountErr != nil { - return unmountErr - } - - _, mountErr := volumeServerClient.VolumeMount(context.Background(), &volume_server_pb.VolumeMountRequest{ - VolumeId: uint32(volumeId), - }) - if mountErr != nil { - return mountErr - } - - return nil + return remountVolume(volumeServerClient, volumeId) }) return err diff --git a/weed/shell/command_volume_tier_download_test.go b/weed/shell/command_volume_tier_download_test.go new file mode 100644 index 0000000000..20136e3181 --- /dev/null +++ b/weed/shell/command_volume_tier_download_test.go @@ -0,0 +1,430 @@ +package shell + +import ( + "context" + "fmt" + "io" + "net" + "strings" + "sync" + "testing" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "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/types" +) + +const ( + tierReplicaRemote = "remote" + tierReplicaLocal = "local" +) + +// tierDownloadCluster models the replicas of one tiered volume across fake +// volume servers sharing one remote object. +type tierDownloadCluster struct { + mu sync.Mutex + servers []pb.ServerAddress + ports map[pb.ServerAddress]int + replicas map[pb.ServerAddress]string + diskTypes map[pb.ServerAddress]string + objectExists bool + downloads int + dangling []string + // hiddenOnRead hides servers from the n-th topology read (0-based) + hiddenOnRead map[int]map[pb.ServerAddress]bool + reads int + frozen map[pb.ServerAddress]bool + writable map[pb.ServerAddress]bool + remounts map[pb.ServerAddress]int + placement uint32 + // staleTwin lists a remote-backed twin of the volume on another disk + staleTwin map[pb.ServerAddress]bool + // markFails applies the readonly mark, then fails as a lost master notification does + markFails map[pb.ServerAddress]bool + mountFails map[pb.ServerAddress]bool +} + +type tierDownloadTestServer struct { + volume_server_pb.UnimplementedVolumeServerServer + cluster *tierDownloadCluster + addr pb.ServerAddress +} + +func (c *tierDownloadCluster) deleteObjectLocked(deleter pb.ServerAddress) { + c.objectExists = false + for server, state := range c.replicas { + if server != deleter && state == tierReplicaRemote { + c.dangling = append(c.dangling, string(server)) + } + } +} + +func (s *tierDownloadTestServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.VolumeTierMoveDatFromRemoteRequest, _ volume_server_pb.VolumeServer_VolumeTierMoveDatFromRemoteServer) error { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + if c.replicas[s.addr] != tierReplicaRemote { + return fmt.Errorf("volume %d is already on local disk", req.VolumeId) + } + if !c.objectExists { + return fmt.Errorf("remote object missing") + } + c.downloads++ + c.replicas[s.addr] = tierReplicaLocal + if !req.KeepRemoteDatFile { + c.deleteObjectLocked(s.addr) + } + return nil +} + +func (s *tierDownloadTestServer) VolumeUnmount(context.Context, *volume_server_pb.VolumeUnmountRequest) (*volume_server_pb.VolumeUnmountResponse, error) { + return &volume_server_pb.VolumeUnmountResponse{}, nil +} + +func (s *tierDownloadTestServer) VolumeMount(context.Context, *volume_server_pb.VolumeMountRequest) (*volume_server_pb.VolumeMountResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + if c.mountFails[s.addr] { + return nil, fmt.Errorf("mount volume: disk error") + } + c.frozen[s.addr] = false + c.remounts[s.addr]++ + return &volume_server_pb.VolumeMountResponse{}, nil +} + +func (s *tierDownloadTestServer) VolumeStatus(context.Context, *volume_server_pb.VolumeStatusRequest) (*volume_server_pb.VolumeStatusResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + return &volume_server_pb.VolumeStatusResponse{IsReadOnly: !c.writable[s.addr] || c.frozen[s.addr]}, nil +} + +func (s *tierDownloadTestServer) VolumeMarkReadonly(_ context.Context, req *volume_server_pb.VolumeMarkReadonlyRequest) (*volume_server_pb.VolumeMarkReadonlyResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + c.frozen[s.addr] = !req.CanDelete + if c.markFails[s.addr] { + return nil, fmt.Errorf("notify master: leader changed") + } + return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil +} + +func (s *tierDownloadTestServer) VolumeMarkWritable(context.Context, *volume_server_pb.VolumeMarkWritableRequest) (*volume_server_pb.VolumeMarkWritableResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + c.frozen[s.addr] = false + c.writable[s.addr] = true + return &volume_server_pb.VolumeMarkWritableResponse{}, nil +} + +func (s *tierDownloadTestServer) VolumeCopy(req *volume_server_pb.VolumeCopyRequest, _ volume_server_pb.VolumeServer_VolumeCopyServer) error { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + // an empty disk type means the source's, as on the real servers + want, got := types.ToDiskType(c.diskTypes[s.addr]), types.ToDiskType(c.diskTypes[pb.ServerAddress(req.SourceDataNode)]) + if req.DiskType != "" { + got = types.ToDiskType(req.DiskType) + } + if got != want { + return fmt.Errorf("copy volume %d onto disk type %q, replica is on %q", req.VolumeId, got, want) + } + if !c.frozen[pb.ServerAddress(req.SourceDataNode)] { + return fmt.Errorf("volume %d on %s still accepts deletes during the copy", req.VolumeId, req.SourceDataNode) + } + switch c.replicas[pb.ServerAddress(req.SourceDataNode)] { + case tierReplicaLocal: + c.replicas[s.addr] = tierReplicaLocal + case tierReplicaRemote: + c.replicas[s.addr] = tierReplicaRemote + default: + return fmt.Errorf("volume %d not found on %s", req.VolumeId, req.SourceDataNode) + } + return nil +} + +func (s *tierDownloadTestServer) VolumeDelete(_ context.Context, req *volume_server_pb.VolumeDeleteRequest) (*volume_server_pb.VolumeDeleteResponse, error) { + c := s.cluster + c.mu.Lock() + defer c.mu.Unlock() + if c.replicas[s.addr] == tierReplicaRemote && !req.KeepRemoteData { + c.deleteObjectLocked(s.addr) + } + delete(c.replicas, s.addr) + return &volume_server_pb.VolumeDeleteResponse{}, nil +} + +func newTierDownloadCluster(t *testing.T, states ...string) *tierDownloadCluster { + t.Helper() + c := &tierDownloadCluster{ + ports: make(map[pb.ServerAddress]int), + replicas: make(map[pb.ServerAddress]string), + diskTypes: make(map[pb.ServerAddress]string), + objectExists: true, + hiddenOnRead: make(map[int]map[pb.ServerAddress]bool), + frozen: make(map[pb.ServerAddress]bool), + writable: make(map[pb.ServerAddress]bool), + remounts: make(map[pb.ServerAddress]int), + staleTwin: make(map[pb.ServerAddress]bool), + markFails: make(map[pb.ServerAddress]bool), + mountFails: make(map[pb.ServerAddress]bool), + } + for _, state := range states { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + port := lis.Addr().(*net.TCPAddr).Port + addr := pb.NewServerAddress("127.0.0.1", port, port) + srv := grpc.NewServer() + volume_server_pb.RegisterVolumeServerServer(srv, &tierDownloadTestServer{cluster: c, addr: addr}) + go srv.Serve(lis) + t.Cleanup(srv.Stop) + c.servers = append(c.servers, addr) + c.ports[addr] = port + c.replicas[addr] = state + if len(c.servers)%2 == 0 { + c.diskTypes[addr] = "ssd" + } + } + return c +} + +func (c *tierDownloadCluster) hide(read int, servers ...pb.ServerAddress) { + if c.hiddenOnRead[read] == nil { + c.hiddenOnRead[read] = make(map[pb.ServerAddress]bool) + } + for _, server := range servers { + c.hiddenOnRead[read][server] = true + } +} + +func (c *tierDownloadCluster) readTopology() (*master_pb.TopologyInfo, error) { + c.mu.Lock() + defer c.mu.Unlock() + hidden := c.hiddenOnRead[c.reads] + c.reads++ + var dns []*master_pb.DataNodeInfo + for _, server := range c.servers { + state, found := c.replicas[server] + if !found || hidden[server] { + continue + } + 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" + } + disks := map[string]*master_pb.DiskInfo{"": {VolumeInfos: []*master_pb.VolumeInformationMessage{vi}}} + if c.staleTwin[server] { + disks["twin"] = &master_pb.DiskInfo{VolumeInfos: []*master_pb.VolumeInformationMessage{{ + Id: 7, Collection: "c", RemoteStorageName: "s3.default", RemoteStorageKey: "key", ReplicaPlacement: c.placement, + }}} + } + dns = append(dns, &master_pb.DataNodeInfo{ + Id: fmt.Sprintf("127.0.0.1:%d", c.ports[server]), + GrpcPort: uint32(c.ports[server]), + DiskInfos: disks, + }) + } + return &master_pb.TopologyInfo{DataCenterInfos: []*master_pb.DataCenterInfo{{ + RackInfos: []*master_pb.RackInfo{{DataNodeInfos: dns}}, + }}}, nil +} + +func (c *tierDownloadCluster) run(t *testing.T) error { + t.Helper() + return downloadAndReplicateFromRemoteTier(grpc.WithTransportCredentials(insecure.NewCredentials()), io.Discard, c.readTopology, needle.VolumeId(7), "c", 0) +} + +func (c *tierDownloadCluster) assertAllLocal(t *testing.T) { + t.Helper() + c.mu.Lock() + defer c.mu.Unlock() + for _, server := range c.servers { + if state := c.replicas[server]; state != tierReplicaLocal { + t.Errorf("replica on %s: want local, got %q", server, state) + } + } + if len(c.dangling) > 0 { + t.Errorf("remote object deleted while still referenced by %s", strings.Join(c.dangling, ",")) + } +} + +func TestTierDownloadReplicatesBeforeDeletingRemoteObject(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote) + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + c.assertAllLocal(t) + if c.objectExists { + t.Errorf("remote object should be deleted once every replica is local") + } + if c.downloads != 1 { + t.Errorf("want one download from the remote tier, got %d", c.downloads) + } +} + +func TestTierDownloadReplacesReplicaMissingFromFirstTopology(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote) + c.hide(0, c.servers[2]) + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + c.assertAllLocal(t) + if c.objectExists { + t.Errorf("remote object should be deleted once every replica is local") + } +} + +func TestTierDownloadKeepsRemoteObjectWhenLastReplicaVanishes(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote) + c.hide(1, c.servers[2]) + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + if !c.objectExists { + t.Fatalf("remote object must be kept while an unlisted replica may still reference it") + } + if len(c.dangling) > 0 { + t.Errorf("remote object deleted while still referenced by %v", c.dangling) + } +} + +func TestTierDownloadSingleReplicaDeletesRemoteObject(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaRemote) + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + c.assertAllLocal(t) + if c.objectExists { + t.Errorf("remote object should be deleted") + } +} + +func TestTierDownloadRerunReusesLocalReplica(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote) + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + c.assertAllLocal(t) + if c.objectExists { + t.Errorf("remote object should be deleted once every replica is local") + } + if c.downloads != 0 { + t.Errorf("want no download when a replica is already local, got %d", c.downloads) + } +} + +func TestTierDownloadCopiesFromSsdSourceOntoHddReplica(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaRemote, tierReplicaRemote, tierReplicaRemote) + c.hide(0, c.servers[0]) + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + c.assertAllLocal(t) +} + +func TestTierDownloadRestoresSourceAfterBlockingDeletes(t *testing.T) { + for _, writable := range []bool{false, true} { + c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote) + source := c.servers[0] + c.writable[source] = writable + if err := c.run(t); err != nil { + t.Fatalf("tier download (writable=%v): %v", writable, err) + } + c.assertAllLocal(t) + if c.frozen[source] { + t.Errorf("writable=%v: source still blocks deletes after the command", writable) + } + if c.writable[source] != writable { + t.Errorf("writable=%v: source writable state changed to %v", writable, c.writable[source]) + } + if !writable && c.remounts[source] != 1 { + t.Errorf("readonly source should reload its persisted state once, got %d remounts", c.remounts[source]) + } + } +} + +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) + } + } +} + +func TestTierDownloadFailsWhenNoReplicaIsListed(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaRemote) + c.hide(0, c.servers[0]) + if err := c.run(t); err == nil { + t.Fatalf("tier download should fail when no replica is listed") + } +} + +func TestTierDownloadKeepsRemoteObjectForStaleTwinOnSource(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote) + c.staleTwin[c.servers[0]] = true + if err := c.run(t); err != nil { + t.Fatalf("tier download: %v", err) + } + if !c.objectExists { + t.Errorf("remote object deleted while a twin on %s still references it", c.servers[0]) + } +} + +func TestTierDownloadRestoresSourceWhenMarkFailsAfterApplying(t *testing.T) { + for _, writable := range []bool{false, true} { + c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote) + source := c.servers[0] + c.writable[source] = writable + c.markFails[source] = true + if err := c.run(t); err == nil { + t.Fatalf("writable=%v: tier download should fail when the readonly mark fails", writable) + } + if c.frozen[source] { + t.Errorf("writable=%v: source still blocks deletes after the failed mark", writable) + } + if c.writable[source] != writable { + t.Errorf("writable=%v: source writable state changed to %v", writable, c.writable[source]) + } + if !c.objectExists { + t.Errorf("writable=%v: remote object deleted although no replica was replaced", writable) + } + } +} + +func TestTierDownloadReportsSourceRestoreFailure(t *testing.T) { + c := newTierDownloadCluster(t, tierReplicaLocal, tierReplicaRemote, tierReplicaRemote) + source := c.servers[0] + c.mountFails[source] = true + err := c.run(t) + if err == nil || !strings.Contains(err.Error(), "restore volume 7 state") { + t.Fatalf("tier download should report the failed source restore, got %v", err) + } + c.assertAllLocal(t) +}