diff --git a/weed/server/volume_grpc_tier_download.go b/weed/server/volume_grpc_tier_download.go index afde9a44f..af07f80d6 100644 --- a/weed/server/volume_grpc_tier_download.go +++ b/weed/server/volume_grpc_tier_download.go @@ -3,10 +3,13 @@ package weed_server import ( "fmt" "os" + "path/filepath" + "runtime" "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) @@ -60,10 +63,11 @@ func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.Volume ProcessedPercentage: percentage, }) } - // copy the data file - _, err := backendStorage.DownloadFile(v.FileName(".dat"), storageKey, fn) + // copy the data file (DownloadFile opens, fsyncs, and closes the .dat internally) + datFileName := v.FileName(".dat") + _, err := backendStorage.DownloadFile(datFileName, storageKey, fn) if err != nil { - return fmt.Errorf("backend %s copy file %s: %v", storageName, v.FileName(".dat"), err) + return fmt.Errorf("backend %s copy file %s: %v", storageName, datFileName, err) } if remoteFileModifiedTime > 0 { modifiedTime := time.Unix(int64(remoteFileModifiedTime), 0) @@ -74,23 +78,70 @@ func (vs *VolumeServer) VolumeTierMoveDatFromRemote(req *volume_server_pb.Volume } } - if req.KeepRemoteDatFile { - return nil + // fsync the containing directory so the new .dat and the about-to-be-rewritten + // .vif are durably linked before we touch the shared remote object. + if err := fsyncDir(filepath.Dir(datFileName)); err != nil { + return fmt.Errorf("volume %d fsync dir for %s: %v", v.Id, datFileName, err) } - // remove remote file - if err := backendStorage.DeleteFile(storageKey); err != nil { - return fmt.Errorf("volume %d failed to delete remote file %s: %v", v.Id, storageKey, err) - } - - // forget remote file + // Trim the remote file reference and persist the .vif (util.WriteFile fsyncs it) + // BEFORE deleting the remote object. After this point hasRemoteFile is false, so a + // crash before DeleteFile merely leaks the remote object while the volume reloads + // its local .dat. The volume must NEVER be left with a .vif referencing the remote + // object while that object is deleted. v.GetVolumeInfo().Files = v.GetVolumeInfo().Files[1:] if err := v.SaveVolumeInfo(); err != nil { return fmt.Errorf("volume %d failed to save remote file info: %v", v.Id, err) } - v.DataBackend.Close() - v.DataBackend = nil + // fsync the directory again so the rewritten .vif is durable. + if err := fsyncDir(filepath.Dir(datFileName)); err != nil { + return fmt.Errorf("volume %d fsync dir after saving volume info: %v", v.Id, err) + } + + // Swap the data backend from the remote storage to the now-local .dat on BOTH + // paths, so a KeepRemoteDatFile=true download still leaves the replica serving + // from local disk (hasRemoteFile=false) rather than the shared remote object. + if err := swapToLocalDatBackend(v, datFileName); err != nil { + return fmt.Errorf("volume %d failed to open local dat file %s: %v", v.Id, datFileName, err) + } + + if req.KeepRemoteDatFile { + // Surviving replicas still reference this object; keep it intact. + return nil + } + + // remove remote file: only the last replica to download deletes the shared object. + if err := backendStorage.DeleteFile(storageKey); err != nil { + return fmt.Errorf("volume %d failed to delete remote file %s: %v", v.Id, storageKey, err) + } return nil } + +// swapToLocalDatBackend closes the remote data backend and opens the downloaded +// local .dat as a DiskFile so reads are served from local disk. +func swapToLocalDatBackend(v *storage.Volume, datFileName string) error { + dataFile, err := os.OpenFile(datFileName, os.O_RDWR, 0644) + if err != nil { + return err + } + // Swap under the volume's data lock so concurrent reads never see a closed + // or half-swapped backend. + v.SwapDataBackend(backend.NewDiskFile(dataFile)) + return nil +} + +// fsyncDir flushes a directory entry so renamed/created files within it survive +// a crash. Directory fsync is unsupported on Windows, so it is skipped there. +func fsyncDir(dir string) error { + if runtime.GOOS == "windows" { + return nil + } + d, err := os.Open(dir) + if err != nil { + return err + } + defer d.Close() + return d.Sync() +} diff --git a/weed/server/volume_grpc_tier_download_test.go b/weed/server/volume_grpc_tier_download_test.go new file mode 100644 index 000000000..43401ca26 --- /dev/null +++ b/weed/server/volume_grpc_tier_download_test.go @@ -0,0 +1,313 @@ +package weed_server + +import ( + "bytes" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" + + "google.golang.org/grpc" +) + +const tierTestBackendName = "tier_test_local_dir.default" + +// tierTestBackend is a fake BackendStorage backed by files in a temp dir. It +// records DeleteFile calls so a test can assert that a shared remote object is +// (or is not) removed. +type tierTestBackend struct { + root string + + mu sync.Mutex + deletes []string +} + +func (b *tierTestBackend) ToProperties() map[string]string { return map[string]string{"root": b.root} } + +func (b *tierTestBackend) NewStorageFile(key string, tierInfo *volume_server_pb.VolumeInfo) backend.BackendStorageFile { + return &tierTestBackendFile{backend: b, key: key, tierInfo: tierInfo} +} + +func (b *tierTestBackend) CopyFile(f *os.File, fn func(progressed int64, percentage float32) error) (key string, size int64, err error) { + key = fmt.Sprintf("obj-%d", time.Now().UnixNano()) + dst := filepath.Join(b.root, key) + out, err := os.Create(dst) + if err != nil { + return "", 0, err + } + defer out.Close() + if _, err = f.Seek(0, io.SeekStart); err != nil { + return "", 0, err + } + written, err := io.Copy(out, f) + if err != nil { + return "", 0, err + } + return key, written, nil +} + +func (b *tierTestBackend) DownloadFile(fileName string, key string, fn func(progressed int64, percentage float32) error) (size int64, err error) { + in, err := os.Open(filepath.Join(b.root, key)) + if err != nil { + return 0, err + } + defer in.Close() + out, err := os.OpenFile(fileName, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644) + if err != nil { + return 0, err + } + written, err := io.Copy(out, in) + if err != nil { + out.Close() + return 0, err + } + // mirror the real backends: fsync the .dat before close so the caller can + // trim the .vif and delete the remote object without risking a torn write. + if syncErr := out.Sync(); syncErr != nil { + out.Close() + return 0, syncErr + } + if closeErr := out.Close(); closeErr != nil { + return 0, closeErr + } + return written, nil +} + +func (b *tierTestBackend) DeleteFile(key string) error { + b.mu.Lock() + b.deletes = append(b.deletes, key) + b.mu.Unlock() + return os.Remove(filepath.Join(b.root, key)) +} + +func (b *tierTestBackend) deleteHistory() []string { + b.mu.Lock() + defer b.mu.Unlock() + out := make([]string, len(b.deletes)) + copy(out, b.deletes) + return out +} + +func (b *tierTestBackend) objectExists(key string) bool { + _, err := os.Stat(filepath.Join(b.root, key)) + return err == nil +} + +type tierTestBackendFile struct { + backend *tierTestBackend + key string + tierInfo *volume_server_pb.VolumeInfo +} + +func (f *tierTestBackendFile) ReadAt(p []byte, off int64) (int, error) { + in, err := os.Open(filepath.Join(f.backend.root, f.key)) + if err != nil { + return 0, err + } + defer in.Close() + return in.ReadAt(p, off) +} +func (f *tierTestBackendFile) WriteAt(p []byte, off int64) (int, error) { panic("not implemented") } +func (f *tierTestBackendFile) Truncate(off int64) error { panic("not implemented") } +func (f *tierTestBackendFile) Close() error { return nil } +func (f *tierTestBackendFile) Name() string { return f.key } +func (f *tierTestBackendFile) Sync() error { return nil } +func (f *tierTestBackendFile) GetStat() (int64, time.Time, error) { + files := f.tierInfo.GetFiles() + if len(files) == 0 { + return 0, time.Time{}, fmt.Errorf("remote file info not found") + } + return int64(files[0].FileSize), time.Unix(int64(files[0].ModifiedTime), 0), nil +} + +// fakeTierStream is a no-op server stream for the tier-download RPC. +type fakeTierStream struct { + grpc.ServerStream +} + +func (s *fakeTierStream) Send(*volume_server_pb.VolumeTierMoveDatFromRemoteResponse) error { return nil } + +func newTierTestStore(t *testing.T, dir string) *storage.Store { + t.Helper() + diskIOProbeConfig := stats.DefaultDiskIOProbeConfig() + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "", + []string{dir}, []int32{100}, []util.MinFreeSpace{{}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, nil, 3, diskIOProbeConfig) + + done := make(chan bool) + go func() { + for { + select { + case <-store.NewVolumesChan: + case <-store.DeletedVolumesChan: + case <-done: + return + } + } + }() + t.Cleanup(func() { + store.Close() + close(done) + }) + return store +} + +// tierUpVolumeOnDisk creates a real local volume in dir, writes a few needles, +// uploads the .dat to the fake backend, rewrites the .vif to remote mode, and +// removes the local .dat — mirroring volume_grpc_tier_upload.go. It returns the +// remote object key and the original local .dat bytes for later comparison. +func tierUpVolumeOnDisk(t *testing.T, dir string, vid needle.VolumeId, b *tierTestBackend) (key string, localDat []byte) { + t.Helper() + store := newTierTestStore(t, dir) + if err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + for i := 1; i <= 5; i++ { + n := new(needle.Needle) + n.Id = types.Uint64ToNeedleId(uint64(i)) + n.Data = []byte(fmt.Sprintf("payload-%d-localdisk", i)) + n.Checksum = needle.NewCRC(n.Data) + if _, err := store.WriteVolumeNeedle(vid, n, true, false); err != nil { + t.Fatalf("write needle %d: %v", i, err) + } + } + + v := store.GetVolume(vid) + if v == nil { + t.Fatal("volume not found after add") + } + diskFile, ok := v.DataBackend.(*backend.DiskFile) + if !ok { + t.Fatalf("expected on-disk backend before tier-up, got %T", v.DataBackend) + } + datPath := v.FileName(".dat") + key, size, err := b.CopyFile(diskFile.File, nil) + if err != nil { + t.Fatalf("upload to fake backend: %v", err) + } + + bType, bId := backend.BackendNameToTypeId(tierTestBackendName) + v.GetVolumeInfo().Files = append(v.GetVolumeInfo().GetFiles(), &volume_server_pb.RemoteFile{ + BackendType: bType, + BackendId: bId, + Key: key, + FileSize: uint64(size), + ModifiedTime: uint64(time.Now().Unix()), + Extension: ".dat", + }) + if err := v.SaveVolumeInfo(); err != nil { + t.Fatalf("save volume info: %v", err) + } + + localDat, err = os.ReadFile(datPath) + if err != nil { + t.Fatalf("read local dat: %v", err) + } + if err := store.UnmountVolume(vid); err != nil { + t.Fatalf("unmount: %v", err) + } + if err := os.Remove(datPath); err != nil { + t.Fatalf("remove local dat: %v", err) + } + return key, localDat +} + +// TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal reproduces the +// multi-replica tier-download data loss: when KeepRemoteDatFile=true (every +// replica except the last), the replica must end up served from its local .dat +// with the shared remote object intact. Before the fix the keep-remote path +// returned right after the download without trimming the .vif or swapping the +// data backend, so the replica stayed remote-backed and a subsequent delete of +// the shared object on the final replica bricked it. +func TestTierMoveDatFromRemote_KeepRemote_LeavesReplicaLocal(t *testing.T) { + b := &tierTestBackend{root: t.TempDir()} + backend.BackendStorages[tierTestBackendName] = b + t.Cleanup(func() { delete(backend.BackendStorages, tierTestBackendName) }) + + dir := t.TempDir() + const vid = needle.VolumeId(71) + key, localDat := tierUpVolumeOnDisk(t, dir, vid, b) + + if !b.objectExists(key) { + t.Fatal("remote object missing after tier-up") + } + + // Mount the tiered volume the way a volume server does at startup. + store := newTierTestStore(t, dir) + v := store.GetVolume(vid) + if v == nil { + t.Fatal("tiered volume not loaded by store") + } + if !v.HasRemoteFile() { + t.Fatal("volume should load in remote mode before download") + } + + vs := &VolumeServer{store: store} + req := &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{ + VolumeId: uint32(vid), + Collection: "", + KeepRemoteDatFile: true, + } + if err := vs.VolumeTierMoveDatFromRemote(req, &fakeTierStream{}); err != nil { + t.Fatalf("VolumeTierMoveDatFromRemote: %v", err) + } + + // keepRemote: the shared object must survive untouched. + if len(b.deleteHistory()) != 0 { + t.Fatalf("KeepRemoteDatFile=true must not delete the shared remote object, deletes: %v", b.deleteHistory()) + } + if !b.objectExists(key) { + t.Fatal("remote object must still exist after a keep-remote download") + } + + // The in-memory volume must now be served from a local DiskFile, not the + // remote backend — this is the assertion that fails on the pre-fix code. + if _, ok := v.DataBackend.(*backend.DiskFile); !ok { + t.Fatalf("after keep-remote download the data backend must be local DiskFile, got %T", v.DataBackend) + } + + // Remount and confirm the volume is no longer remote-backed and reads come + // from the local .dat (matching the bytes uploaded before tiering). + if err := store.UnmountVolume(vid); err != nil { + t.Fatalf("unmount after download: %v", err) + } + if err := store.MountVolume(vid); err != nil { + t.Fatalf("remount after download: %v", err) + } + v2 := store.GetVolume(vid) + if v2 == nil { + t.Fatal("volume missing after remount") + } + if v2.HasRemoteFile() { + t.Fatal("after a keep-remote download the remounted replica must not be remote-backed") + } + if _, ok := v2.DataBackend.(*backend.DiskFile); !ok { + t.Fatalf("remounted replica must read from local DiskFile, got %T", v2.DataBackend) + } + + // The .vif must no longer reference the remote object. + storageName, storageKey := v2.RemoteStorageNameKey() + if storageName != "" || storageKey != "" { + t.Fatalf("trimmed .vif must not reference remote object, got %q/%q", storageName, storageKey) + } + + // The local .dat must hold the downloaded content (the original bytes). + gotDat, err := os.ReadFile(v2.FileName(".dat")) + if err != nil { + t.Fatalf("read local dat after download: %v", err) + } + if !bytes.Equal(gotDat, localDat) { + t.Fatalf("local .dat content mismatch: got %d bytes, want %d", len(gotDat), len(localDat)) + } +} diff --git a/weed/shell/command_volume_tier_compact.go b/weed/shell/command_volume_tier_compact.go index 1dd501833..f26af60a6 100644 --- a/weed/shell/command_volume_tier_compact.go +++ b/weed/shell/command_volume_tier_compact.go @@ -213,9 +213,9 @@ func doVolumeTierCompact(commandEnv *CommandEnv, writer io.Writer, rv remoteVolu fmt.Fprintf(writer, "volume %d garbage ratio %.4f, starting compaction...\n", rv.vid, garbageRatio) // step 2: download .dat from remote to local - // this deletes the remote file and reloads the volume as local + // this deletes the remote file and reloads the volume as local, then re-uploads below fmt.Fprintf(writer, " downloading volume %d from %s to local...\n", rv.vid, rv.remoteStorageName) - err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress) + err = downloadDatFromRemoteTier(grpcDialOption, writer, rv.vid, rv.collection, rv.serverAddress, false) if err != nil { return fmt.Errorf("download volume %d from remote: %v", rv.vid, err) } diff --git a/weed/shell/command_volume_tier_download.go b/weed/shell/command_volume_tier_download.go index 4626bd383..d261747c3 100644 --- a/weed/shell/command_volume_tier_download.go +++ b/weed/shell/command_volume_tier_download.go @@ -5,6 +5,7 @@ import ( "flag" "fmt" "io" + "strings" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -124,11 +125,20 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s return fmt.Errorf("volume %d not found", vid) } + // 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 _, loc := range locations { + 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()) + err = downloadDatFromRemoteTier(commandEnv.option.GrpcDialOption, writer, needle.VolumeId(vid), collection, loc.ServerAddress(), keepRemote) 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 + } return fmt.Errorf("download dat file for volume %d to %s: %v", vid, loc.Url, err) } } @@ -136,12 +146,13 @@ func doVolumeTierDownload(commandEnv *CommandEnv, writer io.Writer, collection s return nil } -func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress) error { +func downloadDatFromRemoteTier(grpcDialOption grpc.DialOption, writer io.Writer, volumeId needle.VolumeId, collection string, targetVolumeServer pb.ServerAddress, keepRemote bool) error { err := operation.WithVolumeServerClient(true, targetVolumeServer, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { stream, downloadErr := volumeServerClient.VolumeTierMoveDatFromRemote(context.Background(), &volume_server_pb.VolumeTierMoveDatFromRemoteRequest{ - VolumeId: uint32(volumeId), - Collection: collection, + VolumeId: uint32(volumeId), + Collection: collection, + KeepRemoteDatFile: keepRemote, }) var lastProcessed int64 diff --git a/weed/shell/command_volume_tier_move.go b/weed/shell/command_volume_tier_move.go index f12fcb222..0d098756f 100644 --- a/weed/shell/command_volume_tier_move.go +++ b/weed/shell/command_volume_tier_move.go @@ -251,7 +251,11 @@ func (c *commandVolumeTierMove) doVolumeTierMove(commandEnv *CommandEnv, writer } // a replica already on the target tier (e.g. left by an interrupted earlier run) - // anchors the move: skip the copy, only fulfill replication and clean up old replicas + // anchors the move: skip the copy, only fulfill replication and clean up old replicas. + // The anchor must match both the target disk type AND (when set) the target data + // center, so a bare presence elsewhere never short-circuits a cross-DC move. + // No replica sync is needed here: remote-tiered replicas share one cloud object, + // so their content is byte-identical and there is no local divergence to reconcile. for _, r := range replicas { if types.ToDiskType(r.info.DiskType) != toDiskType || (toDataCenter != "" && r.location.dc != toDataCenter) { continue @@ -372,7 +376,9 @@ func (c *commandVolumeTierMove) doMoveOneVolume(commandEnv *CommandEnv, writer i if preserveServers[loc.Url] { continue } - if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false, false); err != nil { + // keepRemoteData=true: remote-tiered replicas share one cloud object, so + // deleting a replica must not delete the object the survivors still point at. + if err = deleteVolume(commandEnv.option.GrpcDialOption, vid, loc.ServerAddress(), false, true); err != nil { fmt.Fprintf(writer, "failed to delete volume %d on %s: %v\n", vid, loc.Url, err) } } diff --git a/weed/storage/backend/rclone_backend/rclone_backend.go b/weed/storage/backend/rclone_backend/rclone_backend.go index 24e8620e5..1fe7aca76 100644 --- a/weed/storage/backend/rclone_backend/rclone_backend.go +++ b/weed/storage/backend/rclone_backend/rclone_backend.go @@ -190,6 +190,13 @@ func downloadViaRclone(fs fs.Fs, filename string, key string, fn func(progressed tr := accounting.NewStats(ctx).NewTransfer(obj, fs) defer func() { + // fsync the .dat before closing so its content is durable before the caller + // trims the remote reference and deletes the shared remote object. + if syncer, ok := file.(interface{ Sync() error }); ok { + if syncErr := syncer.Sync(); err == nil && syncErr != nil { + err = syncErr + } + } if closeErr := file.Close(); err == nil && closeErr != nil { err = closeErr } diff --git a/weed/storage/backend/s3_backend/s3_download.go b/weed/storage/backend/s3_backend/s3_download.go index af6f7b4f3..82d73d71b 100644 --- a/weed/storage/backend/s3_backend/s3_download.go +++ b/weed/storage/backend/s3_backend/s3_download.go @@ -50,6 +50,12 @@ func downloadFromS3(sess s3iface.S3API, destFileName string, sourceBucket string return fileSize, fmt.Errorf("failed to download /buckets/%s/%s to %s: %v", sourceBucket, sourceKey, destFileName, err) } + // fsync the downloaded .dat so its content is durable before the caller trims the + // remote reference and deletes the shared remote object. + if syncErr := f.Sync(); syncErr != nil { + return fileSize, fmt.Errorf("failed to fsync %s: %v", destFileName, syncErr) + } + glog.V(1).Infof("downloaded file %s\n", destFileName) return diff --git a/weed/storage/volume.go b/weed/storage/volume.go index 10249482f..b326f9ab3 100644 --- a/weed/storage/volume.go +++ b/weed/storage/volume.go @@ -310,6 +310,19 @@ func (v *Volume) Close() { v.doClose() } +// SwapDataBackend atomically replaces the data backend (e.g. swapping a +// remote-tier backend for a freshly downloaded local .dat), closing the old +// one. Held under dataFileAccessLock so a concurrent read/write never observes +// a half-swapped or closed backend. +func (v *Volume) SwapDataBackend(newBackend backend.BackendStorageFile) { + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + if v.DataBackend != nil { + v.DataBackend.Close() + } + v.DataBackend = newBackend +} + func (v *Volume) doClose() { if v.nm != nil { if err := v.nm.Sync(); err != nil {