From ac319fe3181c66c5defcd726a9f2a68a13a5a0e6 Mon Sep 17 00:00:00 2001 From: Javier de la Torre Date: Wed, 21 Oct 2026 23:16:02 -0400 Subject: [PATCH] volume: never compact a volume tiered to remote storage (#11895) * volume: never compact a volume tiered to remote storage A tiered volume's data backend is the remote object, but compaction reads the local .dat path. When tiering removed the local .dat, every vacuum of a tiered volume with garbage fails with "no such file or directory", and the master's sweep and the worker's vacuum task keep selecting it. When the local .dat was kept, the compaction succeeds against the stale copy and the commit reloads the volume to read the remote object (old layout) through the compacted index, so live needles stop reading. Mirror the Rust volume server (#11545): - CompactByIndex and CompactByVolumeData refuse a volume with a remote file before creating .cpd/.cpx ("volume N is tiered to remote storage, cannot compact"). - CommitCompact refuses under the data-file locks when the volume became tiered during the copy, and removes .cpd/.cpx. - Store.CheckCompactVolume reports no garbage for a tiered volume, so the master's vacuum and volume.vacuum do not select it. - The worker's vacuum detection skips a volume with a remote copy, as balance and EC detection already do. Tests (each fails on master): TestCompactRefusesTieredVolume, TestCommitCompactRefusesVolumeTieredDuringCopy (on master the commit succeeds and a surviving needle no longer reads), TestPickPrimaryReplicaSkipsTieredVolumes. * volume: skip tiered replicas in vacuum selection, keep their real ratio Reporting a zero garbage ratio for a tiered volume hid it from volume.tier.compact, whose preflight reads that ratio while the volume is still remote, and did not exclude it from a vacuum with a zero threshold. Report the real ratio again, as the Rust volume server does, and have the master's vacuum check skip a replica whose volume info names remote storage, whatever the threshold or request. When CommitCompact refuses a volume tiered during the copy, also close the temporary needle map and remove .cpldb: nothing else would, since no later vacuum selects the volume. The tiered-compaction test no longer depends on random needle sizes reaching a 0.3 garbage ratio. * volume tier: do not attach a copy of a .dat a compaction replaced The upload copies the .dat, then records the remote file in the .vif and switches the backend. A compaction commit in between swaps in a compacted .dat and index; the upload then served the old-layout remote object through the new index, so live needles stopped reading. CommitCompact's tiered check cannot see this: the volume is not tiered until the switch. AttachRemoteDatFile records the remote file and switches the backend under the data-file locks, and refuses when the backend is no longer the file that was copied. The upload then removes the unused remote copy. * volume tier: keep a committed .vif entry when only the dir fsync fails AttachRemoteDatFile rolled back the in-memory remote entry on any SaveVolumeInfo error. A NotCrashDurableError comes after the rename, so the .vif on disk already names the copy; with the entry gone, the upload handler deleted that object, and a reload would serve from a missing remote file. Keep the entry and switch to the copy as the .vif says, and still return the error so the caller keeps the local .dat. * vacuum worker: filter tiered replicas instead of skipping the volume A partially tiered volume is the leftover of an interrupted volume.tier.upload; skipping the whole volume left garbage on its local replicas unreclaimed. Per-replica filtering matches the master's batchVacuumVolumeCheck and is safe: a tiered replica refuses both compaction and commit. --------- Co-authored-by: Javier de la Torre Co-authored-by: Chris Lu --- weed/server/volume_grpc_tier_upload.go | 28 +++-- .../volume_grpc_tier_upload_compact_test.go | 107 +++++++++++++++++ weed/storage/volume_tier.go | 41 ++++++- weed/storage/volume_tier_attach_test.go | 70 +++++++++++ weed/storage/volume_vacuum.go | 29 +++++ weed/storage/volume_vacuum_tiered_test.go | 111 ++++++++++++++++++ weed/topology/topology_vacuum.go | 7 ++ weed/topology/topology_vacuum_test.go | 49 ++++++++ weed/worker/tasks/vacuum/detection.go | 11 +- .../tasks/vacuum/detection_tiered_test.go | 25 ++++ 10 files changed, 465 insertions(+), 13 deletions(-) create mode 100644 weed/server/volume_grpc_tier_upload_compact_test.go create mode 100644 weed/storage/volume_tier_attach_test.go create mode 100644 weed/storage/volume_vacuum_tiered_test.go create mode 100644 weed/worker/tasks/vacuum/detection_tiered_test.go diff --git a/weed/server/volume_grpc_tier_upload.go b/weed/server/volume_grpc_tier_upload.go index 540dd383d0..cd60cae528 100644 --- a/weed/server/volume_grpc_tier_upload.go +++ b/weed/server/volume_grpc_tier_upload.go @@ -5,6 +5,7 @@ import ( "os" "time" + "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" "github.com/seaweedfs/seaweedfs/weed/storage/backend" "github.com/seaweedfs/seaweedfs/weed/storage/needle" @@ -92,8 +93,9 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi return fmt.Errorf("backend %s copy file %s: %v", req.DestinationBackendName, diskFile.Name(), err) } - // save the remote file to volume tier info - v.GetVolumeInfo().Files = append(v.GetVolumeInfo().GetFiles(), &volume_server_pb.RemoteFile{ + // record the remote file and switch to it, unless a compaction commit + // replaced the .dat while it was being copied + if err := v.AttachRemoteDatFile(diskFile, &volume_server_pb.RemoteFile{ BackendType: backendType, BackendId: backendId, Key: key, @@ -101,14 +103,20 @@ func (vs *VolumeServer) VolumeTierMoveDatToRemote(req *volume_server_pb.VolumeTi FileSize: uint64(size), ModifiedTime: uint64(modTime.Unix()), Extension: ".dat", - }) - - if err := v.SaveVolumeInfo(); err != nil { - return fmt.Errorf("volume %d failed to save remote file info: %v", v.Id, err) - } - - if err := v.LoadRemoteFile(); err != nil { - return fmt.Errorf("volume %d failed to load remote file: %v", v.Id, err) + }); err != nil { + // Remove the copy unless the .vif already names it. + recorded := false + for _, f := range v.GetVolumeInfo().GetFiles() { + if f.Key == key && f.BackendType == backendType && f.BackendId == backendId { + recorded = true + } + } + if !recorded { + if delErr := backendStorage.DeleteFile(key); delErr != nil { + glog.Warningf("volume %d: remove unused remote copy %s: %v", v.Id, key, delErr) + } + } + return fmt.Errorf("volume %d failed to attach remote file: %v", v.Id, err) } // Deletes must reach the index now, not only after the next restart. v.MarkTieredDeletable() diff --git a/weed/server/volume_grpc_tier_upload_compact_test.go b/weed/server/volume_grpc_tier_upload_compact_test.go new file mode 100644 index 0000000000..9295ed81e6 --- /dev/null +++ b/weed/server/volume_grpc_tier_upload_compact_test.go @@ -0,0 +1,107 @@ +package weed_server + +import ( + "bytes" + "os" + "path/filepath" + "testing" + + "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" +) + +// afterCopyBackend runs a hook once the .dat has been copied, before the +// upload handler records the remote file. +type afterCopyBackend struct { + tierTimestampTestBackend + afterCopy func() +} + +func (b *afterCopyBackend) CopyFile(file *os.File, fn func(progressed int64, percentage float32) error, concurrency int) (string, int64, error) { + key, size, err := b.tierTimestampTestBackend.CopyFile(file, fn, concurrency) + if err == nil && b.afterCopy != nil { + b.afterCopy() + } + return key, size, err +} + +// A compaction committed after the tier upload copied the .dat, but before the +// upload switched the volume to the remote object, leaves that object with the +// old layout and the volume with the compacted index. The upload must not +// attach it. +func TestTierUploadRefusesADatReplacedByACompactionCommit(t *testing.T) { + root := t.TempDir() + b := &afterCopyBackend{tierTimestampTestBackend: tierTimestampTestBackend{root: root}} + backend.BackendStorages[tierTimestampTestBackendName] = b + t.Cleanup(func() { delete(backend.BackendStorages, tierTimestampTestBackendName) }) + + store := storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "store-id", + []string{t.TempDir()}, []int32{10}, []util.MinFreeSpace{{}}, "", storage.NeedleMapInMemory, + []types.DiskType{types.HardDriveType}, nil, 0, stats.DefaultDiskIOProbeConfig()) + t.Cleanup(store.Close) + if err := store.AddVolume(1, "", storage.NeedleMapInMemory, "000", "", 0, needle.Version3, 0, types.HardDriveType, 0); err != nil { + t.Fatalf("add volume: %v", err) + } + written := map[uint64][]byte{} + for i := uint64(1); i <= 10; i++ { + n := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7, Data: bytes.Repeat([]byte{byte(i)}, 1024)} + n.Checksum = needle.NewCRC(n.Data) + if _, err := store.WriteVolumeNeedle(1, n, false, false); err != nil { + t.Fatalf("write %d: %v", i, err) + } + written[i] = n.Data + } + for i := uint64(1); i <= 5; i++ { + if _, err := store.DeleteVolumeNeedle(1, &needle.Needle{Id: types.Uint64ToNeedleId(i)}); err != nil { + t.Fatalf("delete %d: %v", i, err) + } + } + v := store.GetVolume(1) + if err := v.CompactByIndex(nil); err != nil { + t.Fatalf("compact: %v", err) + } + if err := store.MarkVolumeReadonly(1, false, false); err != nil { + t.Fatalf("mark read-only: %v", err) + } + + committed := false + b.afterCopy = func() { + if err := v.CommitCompact(); err != nil { + t.Errorf("commit: %v", err) + return + } + committed = true + } + vs := &VolumeServer{store: store} + err := vs.VolumeTierMoveDatToRemote(&volume_server_pb.VolumeTierMoveDatToRemoteRequest{ + VolumeId: 1, DestinationBackendName: tierTimestampTestBackendName, + }, &discardServerStream[volume_server_pb.VolumeTierMoveDatToRemoteResponse]{}) + if !committed { + t.Fatal("the compaction did not commit between the copy and the attach") // positive control + } + if err == nil { + t.Errorf("the upload attached a remote copy of a .dat that a compaction had replaced") + } + if v.HasRemoteFile() { + t.Errorf("the volume is tiered to the stale remote copy") + } + if entries, _ := os.ReadDir(root); len(entries) != 0 { + t.Errorf("the abandoned remote copy was left behind: %v", entries) + } + if !util.FileExists(filepath.Join(v.FileName(".dat"))) { + t.Fatalf("the local .dat is gone") + } + for i := uint64(6); i <= 10; i++ { + n := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7} + if _, err := store.ReadVolumeNeedle(1, n, nil, nil); err != nil { + t.Errorf("needle %d does not read after the upload: %v", i, err) + } else if !bytes.Equal(n.Data, written[i]) { + t.Errorf("needle %d reads different data", i) + } + } +} diff --git a/weed/storage/volume_tier.go b/weed/storage/volume_tier.go index 7baa94b766..cd046fdd01 100644 --- a/weed/storage/volume_tier.go +++ b/weed/storage/volume_tier.go @@ -1,6 +1,7 @@ package storage import ( + "errors" "fmt" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -142,6 +143,41 @@ func (v *Volume) loadRemoteFileLocked() error { return nil } +// AttachRemoteDatFile records remoteFile, a copy of the .dat read from source, +// in the volume's .vif and switches the data backend to it. It refuses when +// source is no longer the volume's data backend: a compaction commit since the +// copy replaced the .dat and the index, and the copy has the old layout. The +// check, the .vif write and the switch happen under the data-file locks, so a +// commit either lands before (and is seen here) or after (and refuses a tiered +// volume). +func (v *Volume) AttachRemoteDatFile(source backend.BackendStorageFile, remoteFile *volume_server_pb.RemoteFile) error { + v.dataFileSyncLock.Lock() + defer v.dataFileSyncLock.Unlock() + v.dataFileAccessLock.Lock() + defer v.dataFileAccessLock.Unlock() + if v.DataBackend != source { + return fmt.Errorf("volume %d data file changed during the copy (compacted or reloaded)", v.Id) + } + files := v.volumeInfo.GetFiles() + v.volumeInfo.Files = append(files, remoteFile) + if err := v.SaveVolumeInfo(); err != nil { + // A NotCrashDurableError means the rename committed: the .vif names + // the copy, so keep the entry and serve from the copy, and still + // report the error so the caller keeps the local .dat. Any other + // failure left the old .vif in place. + var ndErr *volume_info.NotCrashDurableError + if !errors.As(err, &ndErr) { + v.volumeInfo.Files = files + return fmt.Errorf("save remote file info: %w", err) + } + if loadErr := v.loadRemoteFileLocked(); loadErr != nil { + return fmt.Errorf("save remote file info: %w; load remote file: %v", err, loadErr) + } + return fmt.Errorf("save remote file info: %w", err) + } + return v.loadRemoteFileLocked() +} + func (v *Volume) SaveVolumeInfo() error { tierFileName := v.FileName(".vif") @@ -149,6 +185,9 @@ func (v *Volume) SaveVolumeInfo() error { v.volumeInfo.ExpireAtSec = expireAtSec } - return volume_info.SaveVolumeInfo(tierFileName, v.volumeInfo) + return saveVolumeInfoFile(tierFileName, v.volumeInfo) } + +// saveVolumeInfoFile writes a .vif; tests replace it to inject failures. +var saveVolumeInfoFile = volume_info.SaveVolumeInfo diff --git a/weed/storage/volume_tier_attach_test.go b/weed/storage/volume_tier_attach_test.go new file mode 100644 index 0000000000..0322c591bb --- /dev/null +++ b/weed/storage/volume_tier_attach_test.go @@ -0,0 +1,70 @@ +package storage + +import ( + "errors" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" +) + +// When the .vif rename commits but the directory fsync fails, the saved .vif +// names the remote copy. AttachRemoteDatFile must keep that entry (the upload +// handler deletes a copy the volume info does not name) and serve from the +// copy the .vif points at, while still reporting the error. +func TestAttachRemoteDatFileKeepsACommittedVifEntry(t *testing.T) { + b := newLocalDirBackend(t) + registerTestBackend(t, b) + + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 86, NeedleMapInMemory, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + defer v.Close() + for i := 1; i <= 5; i++ { + _, _, _, err := v.writeNeedle2(newRandomNeedle(uint64(i)), true, false, false) + require.NoError(t, err) + } + diskFile, ok := v.DataBackend.(*backend.DiskFile) + require.True(t, ok) + key, size, err := b.CopyFile(diskFile.File, nil, 0) + require.NoError(t, err) + + saved := saveVolumeInfoFile + saveVolumeInfoFile = func(fileName string, info *volume_server_pb.VolumeInfo) error { + if err := saved(fileName, info); err != nil { + return err + } + return &volume_info.NotCrashDurableError{FileName: fileName, Err: syscall.EIO} + } + defer func() { saveVolumeInfoFile = saved }() + + bType, bId := backend.BackendNameToTypeId(testBackendName) + err = v.AttachRemoteDatFile(diskFile, &volume_server_pb.RemoteFile{ + BackendType: bType, BackendId: bId, Key: key, FileSize: uint64(size), + ModifiedTime: uint64(time.Now().Unix()), Extension: ".dat", + }) + var ndErr *volume_info.NotCrashDurableError + require.True(t, errors.As(err, &ndErr), "the fsync failure is still reported: %v", err) + + onDisk, _, _, err := volume_info.MaybeLoadVolumeInfo(v.FileName(".vif")) + require.NoError(t, err) + require.Len(t, onDisk.GetFiles(), 1) + require.Equal(t, key, onDisk.GetFiles()[0].Key) + require.Len(t, v.GetVolumeInfo().GetFiles(), 1, "the in-memory entry was rolled back while the .vif names the copy") + require.Equal(t, key, v.GetVolumeInfo().GetFiles()[0].Key) + require.True(t, v.HasRemoteFile()) + _, isDisk := v.DataBackend.(*backend.DiskFile) + require.False(t, isDisk, "the live backend must be the copy the .vif names") + + n := &needle.Needle{Id: types.Uint64ToNeedleId(3)} + _, err = v.readNeedle(n, &ReadOption{}, nil) + require.NoError(t, err) +} diff --git a/weed/storage/volume_vacuum.go b/weed/storage/volume_vacuum.go index f37ce0d1d7..12b86982fc 100644 --- a/weed/storage/volume_vacuum.go +++ b/weed/storage/volume_vacuum.go @@ -62,6 +62,14 @@ func (v *Volume) garbageLevel() float64 { return float64(deletedSize) / float64(fileSize) } +// errTieredVolumeCompact refuses a compaction of a volume whose .dat is in the +// remote tier: the copy reads the local .dat, which tiering removes (or leaves +// behind as a stale copy), and the reload after a commit would read the remote +// object through the compacted index. +func errTieredVolumeCompact(vid needle.VolumeId) error { + return fmt.Errorf("volume %d is tiered to remote storage, cannot compact", vid) +} + type CompactOptions struct { PreallocateBytes int64 MaxBytesPerSecond int64 @@ -87,6 +95,9 @@ func (v *Volume) CompactByVolumeData(opts *CompactOptions) error { return nil } glog.V(3).Infof("Compacting volume %d ...", v.Id) + if v.HasRemoteFile() { + return errTieredVolumeCompact(v.Id) + } //no need to lock for copy on write //v.accessLock.Lock() //defer v.accessLock.Unlock() @@ -132,6 +143,9 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error { return nil } glog.V(3).Infof("Compact2 volume %d ...", v.Id) + if v.HasRemoteFile() { + return errTieredVolumeCompact(v.Id) + } if !v.isCompactionInProgress.CompareAndSwap(false, true) { glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id) @@ -183,6 +197,21 @@ func (v *Volume) CommitCompact() error { defer v.dataFileAccessLock.Unlock() glog.V(3).Infof("Got volume %d committing lock...", v.Id) + if v.HasRemoteFile() { + // The volume was tiered after the copy started: the reload would read + // the remote object through the compacted index. Drop the copy and its + // temporary needle map; no later vacuum of this volume would. + if v.tmpNm != nil { + v.tmpNm.Close() + v.tmpNm = nil + } + for _, ext := range []string{".cpd", ".cpx", ".cpldb"} { + if e := os.RemoveAll(v.FileName(ext)); e != nil { + glog.V(0).Infof("remove %s: %v", v.FileName(ext), e) + } + } + return errTieredVolumeCompact(v.Id) + } if v.nm != nil { v.nm.Close() v.nm = nil diff --git a/weed/storage/volume_vacuum_tiered_test.go b/weed/storage/volume_vacuum_tiered_test.go new file mode 100644 index 0000000000..e39de29385 --- /dev/null +++ b/weed/storage/volume_vacuum_tiered_test.go @@ -0,0 +1,111 @@ +package storage + +import ( + "bytes" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb" + "github.com/seaweedfs/seaweedfs/weed/storage/backend" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/super_block" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// A tiered volume that accepted deletes has garbage, but its .dat is remote: +// the compaction must be refused before any copy, and the store must not +// report the volume as a vacuum candidate. +func TestCompactRefusesTieredVolume(t *testing.T) { + b := newLocalDirBackend(t) + registerTestBackend(t, b) + dir := t.TempDir() + const vid = needle.VolumeId(34) + v, _ := tierUpVolumeLive(t, dir, vid, b) + defer v.Close() + require.True(t, v.HasRemoteFile()) + for i := 1; i <= 4; i++ { + _, err := v.deleteNeedle2(newRandomNeedle(uint64(i))) + require.NoError(t, err) + } + garbage := v.garbageLevel() + require.Greater(t, garbage, 0.0, "precondition: the deletes left garbage") + + require.ErrorContains(t, v.CompactByIndex(nil), "tiered") + require.ErrorContains(t, v.CompactByVolumeData(nil), "tiered") + require.False(t, util.FileExists(v.FileName(".cpd"))) + require.False(t, util.FileExists(v.FileName(".cpx"))) + + loc := &DiskLocation{Directory: dir, volumes: map[needle.VolumeId]*Volume{}} + loc.SetVolume(vid, v) + s := &Store{Locations: []*DiskLocation{loc}} + ratio, _, err := s.CheckCompactVolume(vid) + require.NoError(t, err) + // volume.tier.compact reads this ratio while the volume is still remote to + // decide whether to download and compact it; vacuum selection skips tiered + // volumes by their remote storage name instead. + require.Equal(t, garbage, ratio, "a tiered volume reports its real garbage ratio") +} + +// A volume tiered while its compaction copy ran (keeping the local .dat) must +// not commit: the reload would read the remote object, which has the old +// layout, through the compacted index. +func TestCommitCompactRefusesVolumeTieredDuringCopy(t *testing.T) { + t.Run("memory", func(t *testing.T) { testCommitCompactRefusesVolumeTieredDuringCopy(t, NeedleMapInMemory) }) + t.Run("leveldb", func(t *testing.T) { testCommitCompactRefusesVolumeTieredDuringCopy(t, NeedleMapLevelDb) }) +} + +func testCommitCompactRefusesVolumeTieredDuringCopy(t *testing.T, kind NeedleMapKind) { + b := newLocalDirBackend(t) + registerTestBackend(t, b) + dir := t.TempDir() + v, err := NewVolume(dir, dir, "", 35, kind, &super_block.ReplicaPlacement{}, &needle.TTL{}, 0, needle.GetCurrentVersion(), 0, 0) + require.NoError(t, err) + defer v.Close() + + written := map[uint64][]byte{} + for i := uint64(1); i <= 6; i++ { + n := newRandomNeedle(i) + n.Cookie = 7 + _, _, _, err := v.writeNeedle2(n, true, false, false) + require.NoError(t, err) + written[i] = n.Data + } + for i := uint64(1); i <= 3; i++ { + _, err := v.deleteNeedle2(newEmptyNeedle(i)) + require.NoError(t, err) + } + require.NoError(t, v.CompactByIndex(nil)) + require.True(t, util.FileExists(v.FileName(".cpd"))) + require.NotNil(t, v.tmpNm, "precondition: the copy left its temporary needle map") + if kind == NeedleMapLevelDb { + require.True(t, util.FileExists(v.FileName(".cpldb")), "precondition: the copy left its temporary LevelDB") + } + + // tier the volume up, keeping the local .dat + diskFile, ok := v.DataBackend.(*backend.DiskFile) + require.True(t, ok) + key, size, err := b.CopyFile(diskFile.File, nil, 0) + require.NoError(t, err) + bType, bId := backend.BackendNameToTypeId(testBackendName) + 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", + }) + require.NoError(t, v.SaveVolumeInfo()) + require.NoError(t, v.LoadRemoteFile()) + + require.ErrorContains(t, v.CommitCompact(), "tiered") + require.False(t, util.FileExists(v.FileName(".cpd"))) + require.False(t, util.FileExists(v.FileName(".cpx"))) + require.False(t, util.FileExists(v.FileName(".cpldb")), "the temporary LevelDB is left on disk") + require.Nil(t, v.tmpNm, "the temporary needle map is kept") + for i := uint64(4); i <= 6; i++ { + got := &needle.Needle{Id: types.Uint64ToNeedleId(i), Cookie: 7} + _, err := v.readNeedle(got, nil, nil) + require.NoError(t, err, "needle %d", i) + require.True(t, bytes.Equal(written[i], got.Data), "needle %d data", i) + } +} diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index 7babdceb09..83e1ab6f47 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -39,6 +39,13 @@ func (t *Topology) batchVacuumVolumeCheck(grpcDialOption grpc.DialOption, vid ne defer cancel() for index, dn := range locationlist.list { go func(index int, dn *DataNode, url pb.ServerAddress, vid needle.VolumeId) { + // A replica tiered to remote storage cannot be compacted in place, + // whatever its garbage ratio; volume.tier.compact reclaims it. + if v, lookErr := dn.GetVolumesById(vid); lookErr == nil && v.IsRemote() { + glog.V(1).Infof("skip vacuuming volume %d on %s: tiered to %s", vid, url, v.RemoteStorageName) + ch <- -1 + return + } err := operation.WithVolumeServerClient(false, url, grpcDialOption, func(volumeServerClient volume_server_pb.VolumeServerClient) error { resp, err := volumeServerClient.VacuumVolumeCheck(ctx, &volume_server_pb.VacuumVolumeCheckRequest{ VolumeId: uint32(vid), diff --git a/weed/topology/topology_vacuum_test.go b/weed/topology/topology_vacuum_test.go index 6d96add773..82643f33b2 100644 --- a/weed/topology/topology_vacuum_test.go +++ b/weed/topology/topology_vacuum_test.go @@ -481,3 +481,52 @@ func TestVacuumStalledVolumeServerReleasesGuard(t *testing.T) { }) } } + +// A replica tiered to remote storage is never compacted by a vacuum, not even +// on an explicit request with a zero threshold: its .dat is the remote object. +// volume.tier.compact is the way to reclaim its garbage. +func TestVacuumSkipsTieredReplica(t *testing.T) { + fake := &fakeVacuumServer{ + checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{ + 1: {GarbageRatio: 0.9}, + 2: {GarbageRatio: 0.9}, + }, + } + grpcPort, dialOption := startFakeVolumeServer(t, fake) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, grpcPort, "127.0.0.1", "dn", map[string]uint32{"": 10}) + newVolumeInfo := func(vid needle.VolumeId, remote string) storage.VolumeInfo { + return storage.VolumeInfo{ + Id: vid, + Size: 1 << 20, + Collection: "c", + ReadOnly: remote != "", + ModifiedAtSecond: time.Now().Unix(), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: &super_block.ReplicaPlacement{}, + Ttl: needle.EMPTY_TTL, + RemoteStorageName: remote, + } + } + vl := topo.GetVolumeLayout("c", &super_block.ReplicaPlacement{}, needle.EMPTY_TTL, types.ToDiskType("")) + volumes := []storage.VolumeInfo{newVolumeInfo(1, "s3.default"), newVolumeInfo(2, "")} + dn.UpdateVolumes(volumes) + for _, v := range volumes { + topo.RegisterVolumeLayout(v, dn) + } + c := NewCollection("c", topo.volumeSizeLimit, false) + for _, vid := range []needle.VolumeId{1, 2} { + vl.accessLock.RLock() + ll := vl.vid2location[vid].Copy() + vl.accessLock.RUnlock() + topo.vacuumOneVolumeId(dialOption, vl, c, 0, ll, vid, 0, false) + } + + fake.mu.Lock() + defer fake.mu.Unlock() + if len(fake.committed) != 1 || fake.committed[0] != 2 { + t.Fatalf("committed %v, want only the local volume 2", fake.committed) // volume 2 is the positive control + } +} diff --git a/weed/worker/tasks/vacuum/detection.go b/weed/worker/tasks/vacuum/detection.go index a52cca18f6..bc5eb41d9e 100644 --- a/weed/worker/tasks/vacuum/detection.go +++ b/weed/worker/tasks/vacuum/detection.go @@ -109,10 +109,14 @@ func Detection(metrics []*types.VolumeHealthMetrics, clusterInfo *types.ClusterI } // pickPrimaryReplica returns the eligible replica with the highest garbage -// ratio. Returns nil if no replica satisfies both gates. +// ratio. Returns nil if no replica satisfies both gates. Replicas tiered to +// remote storage are skipped: the volume server cannot compact them. func pickPrimaryReplica(replicas []*types.VolumeHealthMetrics, garbageThreshold float64, minAge time.Duration) *types.VolumeHealthMetrics { var best *types.VolumeHealthMetrics for _, m := range replicas { + if m.HasRemoteCopy { + continue + } if m.GarbageRatio < garbageThreshold { continue } @@ -144,7 +148,7 @@ func recordSkipReasons(replicas []*types.VolumeHealthMetrics, garbageThreshold f } // createVacuumTaskParams creates typed parameters for a vacuum task whose -// Sources list contains every replica of the volume. Worker-side +// Sources list contains every non-tiered replica of the volume. Worker-side // performVacuum iterates over Sources and vacuums each replica. func createVacuumTaskParams(task *types.TaskDetectionResult, replicas []*types.VolumeHealthMetrics, vacuumConfig *Config, clusterInfo *types.ClusterInfo) *worker_pb.TaskParams { garbageThreshold := 0.3 @@ -164,6 +168,9 @@ func createVacuumTaskParams(task *types.TaskDetectionResult, replicas []*types.V sources := make([]*worker_pb.TaskSource, 0, len(replicas)) seen := make(map[string]struct{}, len(replicas)) for _, m := range replicas { + if m.HasRemoteCopy { + continue + } address, err := util.ResolveServerAddress(m.Server, clusterInfo.ActiveTopology) if err != nil { glog.Warningf("Failed to resolve address for server %s for vacuum task on volume %d, dropping replica: %v", m.Server, task.VolumeID, err) diff --git a/weed/worker/tasks/vacuum/detection_tiered_test.go b/weed/worker/tasks/vacuum/detection_tiered_test.go new file mode 100644 index 0000000000..5232b45ea9 --- /dev/null +++ b/weed/worker/tasks/vacuum/detection_tiered_test.go @@ -0,0 +1,25 @@ +package vacuum + +import ( + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/worker/types" +) + +func TestPickPrimaryReplicaSkipsTieredReplicas(t *testing.T) { + m := func(id uint32, server string, remote bool) *types.VolumeHealthMetrics { + return &types.VolumeHealthMetrics{VolumeID: id, Server: server, GarbageRatio: 0.9, Age: 72 * time.Hour, + HasRemoteCopy: remote, IsReadOnly: remote} + } + if p := pickPrimaryReplica([]*types.VolumeHealthMetrics{m(1, "a:8080", false)}, 0.3, 24*time.Hour); p == nil { + t.Fatal("a local volume at 90% garbage must be picked") + } + if p := pickPrimaryReplica([]*types.VolumeHealthMetrics{m(2, "a:8080", true)}, 0.3, 24*time.Hour); p != nil { + t.Fatalf("tiered replica of volume %d picked for vacuum", p.VolumeID) + } + p := pickPrimaryReplica([]*types.VolumeHealthMetrics{m(3, "a:8080", true), m(3, "b:8080", false)}, 0.3, 24*time.Hour) + if p == nil || p.Server != "b:8080" { + t.Fatalf("partially tiered volume should pick its local replica, got %+v", p) + } +}