diff --git a/test/volume_server/grpc/move_persist_freeze_test.go b/test/volume_server/grpc/move_persist_freeze_test.go new file mode 100644 index 0000000000..987069fad0 --- /dev/null +++ b/test/volume_server/grpc/move_persist_freeze_test.go @@ -0,0 +1,137 @@ +package volume_server_grpc_test + +import ( + "context" + "net/http" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/test/volume_server/framework" + "github.com/seaweedfs/seaweedfs/test/volume_server/matrix" + "github.com/seaweedfs/seaweedfs/weed/operation/volume_move" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" +) + +func moverServerAddress(cluster framework.MultiCluster, index int) pb.ServerAddress { + return pb.ServerAddress(cluster.VolumeAdminAddress(index) + "." + strings.Split(cluster.VolumeGRPCAddress(index), ":")[1]) +} + +func reloadVolume(t *testing.T, client volume_server_pb.VolumeServerClient, volumeID uint32) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if _, err := client.VolumeUnmount(ctx, &volume_server_pb.VolumeUnmountRequest{VolumeId: volumeID}); err != nil { + t.Fatalf("unmount volume %d: %v", volumeID, err) + } + if _, err := client.VolumeMount(ctx, &volume_server_pb.VolumeMountRequest{VolumeId: volumeID}); err != nil { + t.Fatalf("mount volume %d: %v", volumeID, err) + } +} + +func uploadStatus(t *testing.T, httpClient *http.Client, volumeURL string, volumeID uint32, key uint64, data string) int { + t.Helper() + resp := framework.UploadBytes(t, httpClient, volumeURL, framework.NewFileID(volumeID, key, 0x5EED0000+uint32(key)), []byte(data)) + framework.ReadAllAndClose(t, resp) + return resp.StatusCode +} + +func TestLiveMoveFreezeSurvivesSourceReload(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + cluster := framework.StartMultiVolumeClusterAuto(t, matrix.P1(), 2) + sourceConn, sourceClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0)) + defer sourceConn.Close() + targetConn, targetClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(1)) + defer targetConn.Close() + + const volumeID = uint32(977) + framework.AllocateVolume(t, sourceClient, volumeID, "") + httpClient := framework.NewHTTPClient() + if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 1, "before-move"); code != http.StatusCreated { + t.Fatalf("initial upload: status %d", code) + } + + var reloadedSourceWrite, targetWrite int + mover := volume_move.NewMover(grpc.WithTransportCredentials(insecure.NewCredentials())) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + err := mover.LiveMoveVolume(ctx, needle.VolumeId(volumeID), moverServerAddress(cluster, 0), moverServerAddress(cluster, 1), volume_move.VolumeMoveOptions{ + IdleTimeout: time.Second, + Progress: func(percent float64, stage string) { + switch percent { + case 20: + // frozen, not yet copied: a reload must not reopen the source + reloadVolume(t, sourceClient, volumeID) + reloadedSourceWrite = uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 2, "lost-after-move") + case 70: + // copied, not yet tailed: the copy must not inherit the freeze + targetWrite = uploadStatus(t, httpClient, cluster.VolumeAdminURL(1), volumeID, 3, "on-target") + } + }, + }) + if err != nil { + t.Fatalf("LiveMoveVolume: %v", err) + } + if reloadedSourceWrite != http.StatusConflict { + t.Fatalf("reloaded source write: status %d, want %d", reloadedSourceWrite, http.StatusConflict) + } + if targetWrite != http.StatusCreated { + t.Fatalf("copied target refused a write before the tail: status %d", targetWrite) + } + + // The writable mark on the target is persisted too. + reloadVolume(t, targetClient, volumeID) + if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(1), volumeID, 4, "after-reload"); code != http.StatusCreated { + t.Fatalf("moved volume refused a write after a reload: status %d", code) + } + for key, want := range map[uint64]string{1: "before-move", 3: "on-target", 4: "after-reload"} { + resp := framework.ReadBytes(t, httpClient, cluster.VolumeAdminURL(1), framework.NewFileID(volumeID, key, 0x5EED0000+uint32(key))) + body := framework.ReadAllAndClose(t, resp) + if resp.StatusCode != http.StatusOK || string(body) != want { + t.Fatalf("read key %d from target: status %d body %q, want %q", key, resp.StatusCode, body, want) + } + } +} + +func TestLiveMoveAbortClearsPersistedFreeze(t *testing.T) { + if testing.Short() { + t.Skip("skipping integration test in short mode") + } + + cluster := framework.StartMultiVolumeClusterAuto(t, matrix.P1(), 2) + sourceConn, sourceClient := framework.DialVolumeServer(t, cluster.VolumeGRPCAddress(0)) + defer sourceConn.Close() + + const volumeID = uint32(978) + framework.AllocateVolume(t, sourceClient, volumeID, "") + httpClient := framework.NewHTTPClient() + if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 1, "before-move"); code != http.StatusCreated { + t.Fatalf("initial upload: status %d", code) + } + + mover := volume_move.NewMover(grpc.WithTransportCredentials(insecure.NewCredentials())) + ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) + defer cancel() + err := mover.LiveMoveVolume(ctx, needle.VolumeId(volumeID), moverServerAddress(cluster, 0), moverServerAddress(cluster, 1), volume_move.VolumeMoveOptions{ + Progress: func(percent float64, stage string) { + if percent == 20 { + cancel() + } + }, + }) + if err == nil { + t.Fatalf("LiveMoveVolume succeeded with a cancelled context") + } + + reloadVolume(t, sourceClient, volumeID) + if code := uploadStatus(t, httpClient, cluster.VolumeAdminURL(0), volumeID, 2, "after-abort"); code != http.StatusCreated { + t.Fatalf("aborted move left the source readonly after a reload: status %d", code) + } +} diff --git a/weed/operation/volume_move/fake_client_test.go b/weed/operation/volume_move/fake_client_test.go index 7d1ba5c6fa..fceac21086 100644 --- a/weed/operation/volume_move/fake_client_test.go +++ b/weed/operation/volume_move/fake_client_test.go @@ -40,9 +40,10 @@ type fakeCluster struct { lastAppendAtNs uint64 - copyReqs []*volume_server_pb.VolumeCopyRequest - deleteReqs []*volume_server_pb.VolumeDeleteRequest - ecCopyReqs []*volume_server_pb.VolumeEcShardsCopyRequest + copyReqs []*volume_server_pb.VolumeCopyRequest + readonlyReqs []*volume_server_pb.VolumeMarkReadonlyRequest + deleteReqs []*volume_server_pb.VolumeDeleteRequest + ecCopyReqs []*volume_server_pb.VolumeEcShardsCopyRequest } func newFakeCluster() *fakeCluster { @@ -94,6 +95,9 @@ func (f *fakeClient) VolumeMarkReadonly(ctx context.Context, req *volume_server_ if err := f.cluster.record(f.addr, "VolumeMarkReadonly"); err != nil { return nil, err } + f.cluster.mu.Lock() + f.cluster.readonlyReqs = append(f.cluster.readonlyReqs, req) + f.cluster.mu.Unlock() return &volume_server_pb.VolumeMarkReadonlyResponse{}, nil } diff --git a/weed/operation/volume_move/volume_move.go b/weed/operation/volume_move/volume_move.go index e195a0f845..21f3eb169a 100644 --- a/weed/operation/volume_move/volume_move.go +++ b/weed/operation/volume_move/volume_move.go @@ -65,7 +65,7 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so } opts.Progress(10, fmt.Sprintf("marking volume %d readonly on %s", volumeId, source)) - sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, true) + sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, true, true) // The source stays authoritative until its delete succeeds; on any earlier // failure undo the freeze this move added and clean up the target copy. // Installed before the error check: the marking RPC itself can fail after @@ -130,8 +130,9 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so m.restoreVolumeWritable(volumeId, source) } else { // The hard freeze added on top of a readonly-reporting status is - // kept: it is indistinguishable from an operator's mark. It is - // not persisted, so a volume server restart clears it. + // kept: it is indistinguishable from an operator's mark. Only a + // freeze of a writable volume is persisted, so a volume server + // restart clears this one. fmt.Fprintf(opts.Writer, "volume %d on %s keeps the readonly mark the move added; volume.mark -writable clears it if the prior readonly state was transient\n", volumeId, source) } } @@ -139,6 +140,9 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so if err != nil { return fmt.Errorf("mark volume %d readonly on %s: %v", volumeId, source, err) } + if sourceWasWritable { + fmt.Fprintf(opts.Writer, "volume %d on %s stays readonly across restarts until the move ends; if the move is interrupted and %s has no copy of it, volume.mark -node %s -volumeId %d -writable reopens it, otherwise delete one copy before reopening the other\n", volumeId, source, target, source, volumeId) + } targetHadVolume, targetStateKnown = m.probeVolume(ctx, volumeId, target) if !sourceWasWritable { @@ -168,6 +172,13 @@ func (m *Mover) LiveMoveVolume(ctx context.Context, volumeId needle.VolumeId, so return fmt.Errorf("copy volume %d from %s to %s: %v", volumeId, source, target, err) } copyCompleted = true + if sourceWasWritable { + // The copy carries the source's persisted freeze; the target must take + // the tailed writes and serve as the moved volume. + if err = m.MarkVolumeWritable(ctx, volumeId, target, true, false); err != nil { + return fmt.Errorf("mark volume %d writable on %s: %v", volumeId, target, err) + } + } opts.Progress(70, fmt.Sprintf("tailing volume %d from %s to %s", volumeId, source, target)) fmt.Fprintf(opts.Writer, "tailing volume %d from %s to %s\n", volumeId, source, target) @@ -273,7 +284,7 @@ func (m *Mover) CopyVolume(ctx context.Context, volumeId needle.VolumeId, source // is (force=false): missed concurrent deletes only make the new replica // trivially stale, and hard-marking here could pin a transiently readonly // (e.g. low-disk) source readonly with nothing to gate on it. - sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, false) + sourceWasWritable, err := m.ensureVolumeReadonly(ctx, volumeId, source, false, false) defer func() { if sourceWasWritable && (err != nil || restoreWritable) { m.restoreVolumeWritable(volumeId, source) @@ -293,7 +304,9 @@ func (m *Mover) CopyVolume(ctx context.Context, volumeId needle.VolumeId, source // readonly-but-can-delete flag, and neither blocks needle deletes, so only the // hard mark makes the volume immutable — required before deleting the source // of a move. Without force, a readonly-reporting volume is left untouched. -func (m *Mover) ensureVolumeReadonly(ctx context.Context, volumeId needle.VolumeId, server pb.ServerAddress, force bool) (wasWritable bool, err error) { +// persist keeps the freeze of a writable volume across a volume reload or +// server restart; a mark over an already-readonly volume is never persisted. +func (m *Mover) ensureVolumeReadonly(ctx context.Context, volumeId needle.VolumeId, server pb.ServerAddress, force, persist bool) (wasWritable bool, err error) { err = m.withClient(false, server, func(client volume_server_pb.VolumeServerClient) error { resp, statusErr := client.VolumeStatus(ctx, &volume_server_pb.VolumeStatusRequest{ VolumeId: uint32(volumeId), @@ -307,7 +320,7 @@ func (m *Mover) ensureVolumeReadonly(ctx context.Context, volumeId needle.Volume } _, readonlyErr := client.VolumeMarkReadonly(ctx, &volume_server_pb.VolumeMarkReadonlyRequest{ VolumeId: uint32(volumeId), - Persist: false, + Persist: persist && wasWritable, }) return readonlyErr }) @@ -331,7 +344,8 @@ func (m *Mover) probeVolume(ctx context.Context, volumeId needle.VolumeId, serve return false, false } -// restoreVolumeWritable undoes a readonly mark after a failed move. It runs on +// restoreVolumeWritable undoes a readonly mark after a failed move, including +// a persisted one: the volume servers always persist a writable mark. It runs on // its own deadline so an abort via cancelled context still restores the source, // and only logs a failure — the caller is already returning the move error. func (m *Mover) restoreVolumeWritable(volumeId needle.VolumeId, server pb.ServerAddress) { diff --git a/weed/operation/volume_move/volume_move_test.go b/weed/operation/volume_move/volume_move_test.go index 0b41fcbe77..b4285a277e 100644 --- a/weed/operation/volume_move/volume_move_test.go +++ b/weed/operation/volume_move/volume_move_test.go @@ -147,12 +147,19 @@ func TestLiveMoveVolumeSequence(t *testing.T) { "src:8080 VolumeMarkReadonly", "dst:8080 ReadVolumeFileStatus", "dst:8080 VolumeCopy", + "dst:8080 VolumeMarkWritable", "dst:8080 VolumeTailReceiver", "src:8080 ReadVolumeFileStatus", "dst:8080 ReadVolumeFileStatus", "src:8080 VolumeDelete", }) + // The freeze survives a reload of the source; the copy inherits it, so the + // target is reopened before the tail writes to it. + if !cluster.readonlyReqs[0].Persist { + t.Errorf("freeze of a writable source not persisted: %+v", cluster.readonlyReqs[0]) + } + copyReq := cluster.copyReqs[0] if copyReq.DiskType != "ssd" || copyReq.IoBytePerSecond != 42 || copyReq.SourceDataNode != string(srcAddr) { t.Errorf("copy request not propagated: %+v", copyReq) @@ -631,3 +638,59 @@ func TestCopyVolumeRestoreWritable(t *testing.T) { } } } + +func TestLiveMoveVolumeReadonlySourceFreezeNotPersisted(t *testing.T) { + // A readonly-reporting source may be only transiently so (e.g. low disk); + // the hard mark over it must not outlive a restart, and the copy keeps + // whatever state the source had. + cluster := newFakeCluster() + cluster.readonly[string(srcAddr)] = true + cluster.status[string(srcAddr)] = volumeStatus(1000, 100, 10) + cluster.status[string(dstAddr)] = volumeStatus(1000, 100, 10) + cluster.statusFailures[string(dstAddr)] = 1 + + if err := cluster.mover().LiveMoveVolume(context.Background(), 7, srcAddr, dstAddr, VolumeMoveOptions{}); err != nil { + t.Fatalf("LiveMoveVolume: %v", err) + } + if cluster.readonlyReqs[0].Persist { + t.Errorf("freeze of an already-readonly source persisted: %+v", cluster.readonlyReqs[0]) + } + for _, call := range cluster.callList() { + if call == "dst:8080 VolumeMarkWritable" { + t.Fatalf("copy of a readonly source made writable: %v", cluster.callList()) + } + } +} + +func TestLiveMoveVolumeTargetReopenFailureRestores(t *testing.T) { + // A target left with the copied freeze would refuse the tailed writes; + // abort, drop the copy and reopen the source. + cluster := newFakeCluster() + cluster.status[string(srcAddr)] = volumeStatus(1000, 100, 10) + cluster.errs["dst:8080 VolumeMarkWritable"] = errors.New("target unreachable") + + err := cluster.mover().LiveMoveVolume(context.Background(), 7, srcAddr, dstAddr, VolumeMoveOptions{}) + if err == nil || !strings.Contains(err.Error(), "mark volume 7 writable on dst:8080") { + t.Fatalf("expected target reopen error, got: %v", err) + } + calls := cluster.callList() + for _, call := range calls { + if call == "dst:8080 VolumeTailReceiver" { + t.Fatalf("tailed into a target that is still frozen: %v", calls) + } + } + if fmt.Sprint(calls[len(calls)-2:]) != fmt.Sprint([]string{"dst:8080 VolumeDelete", "src:8080 VolumeMarkWritable"}) { + t.Fatalf("expected target cleanup then source restore, got: %v", calls) + } +} + +func TestCopyVolumeFreezeNotPersisted(t *testing.T) { + // A copy leaves the source serving, so its freeze must not outlive a crash. + cluster := newFakeCluster() + if _, err := cluster.mover().CopyVolume(context.Background(), 7, srcAddr, dstAddr, "", 0, false, nil); err != nil { + t.Fatalf("CopyVolume: %v", err) + } + if cluster.readonlyReqs[0].Persist { + t.Errorf("copy freeze persisted: %+v", cluster.readonlyReqs[0]) + } +}