diff --git a/weed/filer/filer_remote_tombstone.go b/weed/filer/filer_remote_tombstone.go index fb6e2be13f..b07a05ff7a 100644 --- a/weed/filer/filer_remote_tombstone.go +++ b/weed/filer/filer_remote_tombstone.go @@ -10,19 +10,26 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/remote_storage" + "github.com/seaweedfs/seaweedfs/weed/stats" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/util/log_buffer" "google.golang.org/protobuf/proto" ) const ( - // remoteDeletionTombstoneTTL bounds a tombstone when no write-back sync - // offset ever confirms the remote delete. A remote object re-created + // defaultRemoteDeletionTombstoneTTL bounds a tombstone when no write-back + // sync offset ever confirms the remote delete. A remote object re-created // outside the filer at a deleted path stays hidden for this long. - remoteDeletionTombstoneTTL = 24 * time.Hour - // remoteDeletionTombstoneLimit bounds tracked paths; past it new - // tombstones are dropped after an expired sweep still leaves no room. - remoteDeletionTombstoneLimit = 1 << 16 + defaultRemoteDeletionTombstoneTTL = 24 * time.Hour + // defaultRemoteDeletionTombstoneLimit bounds tracked paths; at the cap a + // new delete is not recorded and the lazy fetch stays off instead. + defaultRemoteDeletionTombstoneLimit = 1 << 16 +) + +// Variables for tests that fill or age the table. +var ( + remoteDeletionTombstoneTTL = defaultRemoteDeletionTombstoneTTL + remoteDeletionTombstoneLimit = defaultRemoteDeletionTombstoneLimit ) // remoteDeletionTombstones tracks paths deleted under a remote mount whose @@ -49,7 +56,92 @@ type remoteDeletionTombstones struct { mu sync.Mutex files map[string]remoteDeletionTombstone // file path -> delete event dirs map[string]remoteDeletionTombstone // deleted directory path -> event; covers its subtree - pending map[string]bool // tombstone path recorded ahead of its event + pending map[string]bool // tombstone path recorded ahead of its event + // full is set when a delete could not be recorded: from then on an + // unrecorded path may be a pending delete, so every lazy fetch stays off. + // It lifts once the table has room and the newest unrecorded delete is + // past the TTL (skippedThroughTsNs): an unrecorded delete has no event + // to confirm it, so only the TTL makes it safe, as for a pending one. + full bool + skippedThroughTsNs int64 + // drainedOffsets records, per mount, the write-back offset the last + // drain released through, so a read repeats the sweep only once the + // offset has advanced. + drainedOffsets map[string]int64 +} + +// isFull reports whether an unrecorded delete still keeps the lazy fetch off. +func (t *remoteDeletionTombstones) isFull() bool { + if t == nil { + return false + } + t.mu.Lock() + defer t.mu.Unlock() + return t.full +} + +// drain, while the table is full, evicts expired tombstones and releases the +// mount's tombstones its write-back offset has passed, then re-evaluates +// the full state. It sweeps only when that offset advanced since the last +// drain for the mount or the unrecorded deletes have aged past the TTL, so +// a listing of many missing children does not rescan the table per child. +// haveOffset is false when the mount's persisted offset could not be read. +func (t *remoteDeletionTombstones) drain(mountDir string, nowNs, offsetTsNs int64, haveOffset bool) { + if t == nil { + return + } + t.mu.Lock() + defer t.mu.Unlock() + if t.drainedOffsets == nil { + t.drainedOffsets = make(map[string]int64) + } + advanced := haveOffset && offsetTsNs > t.drainedOffsets[mountDir] + aged := t.skippedThroughTsNs != 0 && nowNs-t.skippedThroughTsNs >= int64(remoteDeletionTombstoneTTL) + if !advanced && !aged { + return + } + t.evictExpiredLocked(nowNs) + if advanced { + t.releaseConsumedLocked(mountDir, offsetTsNs) + t.drainedOffsets[mountDir] = offsetTsNs + } + t.observeLocked() +} + +// releaseConsumedLocked drops the mount's tombstones whose event the mount's +// write-back offset has passed; a pending one has no event timestamp yet +// and stays, and another mount's tombstones answer to their own offset. +// Caller must hold t.mu. +func (t *remoteDeletionTombstones) releaseConsumedLocked(mountDir string, offsetTsNs int64) { + prefix := strings.TrimSuffix(mountDir, "/") + "/" + under := func(p string) bool { return p == mountDir || strings.HasPrefix(p, prefix) } + for p, cur := range t.files { + if under(p) && !t.pending[p] && cur.streamTsNs <= offsetTsNs { + delete(t.files, p) + } + } + for p, cur := range t.dirs { + if under(p) && !t.pending[p] && cur.streamTsNs <= offsetTsNs { + delete(t.dirs, p) + } + } +} + +// observeLocked publishes the table size and lifts the full state once a +// delete fits again and the newest unrecorded delete is past the TTL. +// Caller must hold t.mu. +func (t *remoteDeletionTombstones) observeLocked() { + size := len(t.files) + len(t.dirs) + if size < remoteDeletionTombstoneLimit && (t.skippedThroughTsNs == 0 || time.Now().UnixNano()-t.skippedThroughTsNs >= int64(remoteDeletionTombstoneTTL)) { + t.full = false + t.skippedThroughTsNs = 0 + } + stats.FilerRemoteDeletionTombstones.Set(float64(size)) + if t.full { + stats.FilerRemoteDeletionTombstonesFull.Set(1) + } else { + stats.FilerRemoteDeletionTombstonesFull.Set(0) + } } func newRemoteDeletionTombstones() *remoteDeletionTombstones { @@ -116,7 +208,14 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, commitTsNs, s if len(t.files)+len(t.dirs) >= remoteDeletionTombstoneLimit { t.evictExpiredLocked(time.Now().UnixNano()) if len(t.files)+len(t.dirs) >= remoteDeletionTombstoneLimit { - glog.V(0).Infof("remote deletion tombstones full (%d), skipping %s", remoteDeletionTombstoneLimit, path) + if !t.full { + glog.V(0).Infof("remote deletion tombstones full (%d), lazy remote fetch off until the table has room and the unrecorded deletes age out; first skipped %s", remoteDeletionTombstoneLimit, path) + } + t.full = true + if commitTsNs > t.skippedThroughTsNs { + t.skippedThroughTsNs = commitTsNs + } + t.observeLocked() return } } @@ -127,6 +226,7 @@ func (t *remoteDeletionTombstones) upsert(path string, isDir bool, commitTsNs, s if isDir { t.dropCoveredLocked(path, commitTsNs) } + t.observeLocked() } // dropCoveredLocked removes descendant tombstones a new directory tombstone @@ -165,6 +265,7 @@ func (t *remoteDeletionTombstones) drop(path string, tsNs int64) { delete(t.files, path) delete(t.pending, path) } + t.observeLocked() } // clear drops a file tombstone when a write at the path is at least as new as @@ -179,6 +280,7 @@ func (t *remoteDeletionTombstones) clear(path string, tsNs int64) { delete(t.files, path) delete(t.pending, path) } + t.observeLocked() } // blockedSince returns the tombstone governing path — its own file tombstone @@ -230,6 +332,7 @@ func (t *remoteDeletionTombstones) releaseThrough(path string, tsNs int64) { } p = p[:i] } + t.observeLocked() } // releaseExpired drops the tombstones governing path that outlived the TTL — @@ -255,6 +358,7 @@ func (t *remoteDeletionTombstones) releaseExpired(path string, nowNs int64) { } p = p[:i] } + t.observeLocked() } func (t *remoteDeletionTombstones) evictExpiredLocked(nowNs int64) { @@ -270,6 +374,7 @@ func (t *remoteDeletionTombstones) evictExpiredLocked(nowNs int64) { delete(t.pending, p) } } + t.observeLocked() } // noteRemoteDeletion records a delete of path under a remote mount so lazy @@ -300,6 +405,16 @@ func (f *Filer) isRemoteDeletionPending(ctx context.Context, p util.FullPath, mo if f.remoteTombstones == nil { return false } + // With deletes unrecorded, any path may be one: no lazy fetch resurrects + // it until the table has room again and the unrecorded deletes aged out. + // Reads keep the table draining meanwhile. + if f.remoteTombstones.isFull() { + offset, err := f.readRemoteSyncOffset(ctx, mountDir) + f.remoteTombstones.drain(string(mountDir), time.Now().UnixNano(), offset, err == nil) + if f.remoteTombstones.isFull() { + return true + } + } streamTsNs, commitTsNs, pending := f.remoteTombstones.blockedSince(string(p)) if commitTsNs == 0 { return false diff --git a/weed/filer/filer_remote_tombstone_test.go b/weed/filer/filer_remote_tombstone_test.go index 66aa7f87a1..f30f729172 100644 --- a/weed/filer/filer_remote_tombstone_test.go +++ b/weed/filer/filer_remote_tombstone_test.go @@ -165,6 +165,107 @@ func TestRemoteDeletionTombstones_RaisedLocalTombstoneIsPendingAgain(t *testing. assert.False(t, pending) } +func TestMaybeLazyFetchFromRemote_FullTableKeepsTheFetchOff(t *testing.T) { + const storageType = "stub_tomb_full" + stub := &countingRemoteClient{ + stubRemoteClient: stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 11}, + }, + } + f, _ := newMountedTestFiler(t, storageType, stub, 0) + remoteDeletionTombstoneLimit = 2 + t.Cleanup(func() { remoteDeletionTombstoneLimit = defaultRemoteDeletionTombstoneLimit }) + + now := time.Now().UnixNano() + f.noteRemoteDeletion("/buckets/mybucket/dir/a.txt", false, now) + f.noteRemoteDeletion("/buckets/mybucket/dir/b.txt", false, now) + assert.False(t, f.remoteTombstones.isFull()) + + // the third delete does not fit: from here every path may be a pending + // delete, so no lookup miss consults the remote + f.noteRemoteDeletion("/buckets/mybucket/dir/c.txt", false, now) + assert.True(t, f.remoteTombstones.isFull()) + entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt") + require.NoError(t, err) + assert.Nil(t, entry) + assert.Equal(t, 0, stub.statCalls) + + // room again once a recorded delete lifts, but the unrecorded delete of + // c.txt has no event to confirm it: the fetch stays off until the TTL + f.remoteTombstones.drop("/buckets/mybucket/dir/a.txt", now) + assert.True(t, f.remoteTombstones.isFull()) + entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt") + require.NoError(t, err) + assert.Nil(t, entry) + assert.Equal(t, 0, stub.statCalls) + + remoteDeletionTombstoneTTL = 0 + t.Cleanup(func() { remoteDeletionTombstoneTTL = defaultRemoteDeletionTombstoneTTL }) + entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt") + require.NoError(t, err) + require.NotNil(t, entry) + assert.NotZero(t, stub.statCalls) + assert.False(t, f.remoteTombstones.isFull()) +} + +func TestMaybeLazyFetchFromRemote_FullTableDrainsOnReads(t *testing.T) { + const storageType = "stub_tomb_drain" + stub := &countingRemoteClient{ + stubRemoteClient: stubRemoteClient{ + statResult: &filer_pb.RemoteEntry{RemoteMtime: 1700000000, RemoteSize: 11}, + }, + } + f, store := newMountedTestFiler(t, storageType, stub, 0) + remoteDeletionTombstoneLimit = 2 + t.Cleanup(func() { remoteDeletionTombstoneLimit = defaultRemoteDeletionTombstoneLimit }) + now := time.Now().UnixNano() + tsA, tsB, tsC := now-3*int64(time.Second), now-2*int64(time.Second), now-int64(time.Second) + + // a confirmed delete on another mount, then two on this one fill the + // table; a third on this mount is unrecorded and closes the fetch + remoteDeletionTombstoneLimit = 3 + f.remoteTombstones.addFromEvent("/buckets/other/dir/z.txt", false, tsA, tsA) + f.remoteTombstones.addFromEvent("/buckets/mybucket/dir/a.txt", false, tsA, tsA) + f.remoteTombstones.addFromEvent("/buckets/mybucket/dir/b.txt", false, tsB, tsB) + f.remoteTombstones.addFromEvent("/buckets/mybucket/dir/c.txt", false, tsC, tsC) + assert.True(t, f.remoteTombstones.isFull()) + assert.Len(t, f.remoteTombstones.files, 3) + + // the write-back offset behind the recorded deletes: a read drains + // nothing and the fetch stays off + putRemoteSyncOffset(t, store, "/buckets/mybucket", tsA-1) + entry, err := f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt") + require.NoError(t, err) + assert.Nil(t, entry) + assert.Equal(t, 0, stub.statCalls) + assert.Len(t, f.remoteTombstones.files, 3) + + // the offset past both recorded deletes of this mount: the read drains + // them without any mutation; the other mount's tombstone answers to its + // own offset and stays; the unrecorded delete is still inside the TTL, + // so the fetch stays off + putRemoteSyncOffset(t, store, "/buckets/mybucket", tsB) + entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt") + require.NoError(t, err) + assert.Nil(t, entry) + assert.Equal(t, 0, stub.statCalls) + assert.Len(t, f.remoteTombstones.files, 1) + _, otherKept := f.remoteTombstones.files["/buckets/other/dir/z.txt"] + assert.True(t, otherKept, "another mount's tombstone must not be released by this mount's offset") + assert.True(t, f.remoteTombstones.isFull()) + + // once the unrecorded delete has aged past the TTL the next read lifts + // the gate + f.remoteTombstones.mu.Lock() + f.remoteTombstones.skippedThroughTsNs = now - int64(remoteDeletionTombstoneTTL) - 1 + f.remoteTombstones.mu.Unlock() + entry, err = f.maybeLazyFetchFromRemote(context.Background(), "/buckets/mybucket/dir/other.txt") + require.NoError(t, err) + require.NotNil(t, entry) + assert.NotZero(t, stub.statCalls) + assert.False(t, f.remoteTombstones.isFull()) +} + func TestMaybeLazyFetchFromRemote_SkipsTombstonedPath(t *testing.T) { const storageType = "stub_tomb_fetch" stub := &countingRemoteClient{ diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index a7d2e75b96..b39e9c410f 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -300,6 +300,22 @@ var ( Help: "Counter of filer store requests.", }, []string{"store", "type"}) + FilerRemoteDeletionTombstones = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemFiler, + Name: "remote_deletion_tombstones", + Help: "Paths deleted under a remote mount whose remote delete the write-back daemon has not consumed yet.", + }) + + FilerRemoteDeletionTombstonesFull = prometheus.NewGauge( + prometheus.GaugeOpts{ + Namespace: Namespace, + Subsystem: subsystemFiler, + Name: "remote_deletion_tombstones_full", + Help: "1 from a delete the full table could not record until the table has room and that delete is past the TTL; lazy remote fetches are off meanwhile. 0 otherwise.", + }) + FilerStoreHistogram = prometheus.NewHistogramVec( prometheus.HistogramOpts{ Namespace: Namespace, @@ -1037,6 +1053,8 @@ func init() { Gather.MustRegister(FilerInFlightUploadCountGauge) Gather.MustRegister(FilerStoreCounter) Gather.MustRegister(FilerStoreHistogram) + Gather.MustRegister(FilerRemoteDeletionTombstones) + Gather.MustRegister(FilerRemoteDeletionTombstonesFull) Gather.MustRegister(FilerSyncOffsetGauge) Gather.MustRegister(FilerSyncEventsReceivedCounter) Gather.MustRegister(FilerSyncEventsProcessedCounter)