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) + } +}