diff --git a/weed/filer/meta_aggregator.go b/weed/filer/meta_aggregator.go index 9f2b7b63f..9c440b281 100644 --- a/weed/filer/meta_aggregator.go +++ b/weed/filer/meta_aggregator.go @@ -49,8 +49,19 @@ type MetaAggregator struct { // whose unflushed events still exist, and de-accounting it at once would // let subscribers advance past them. A re-add clears the mark; a peer // gone past the grace is dropped so it cannot pin the low-watermarks. - peerRemovedAtNs map[pb.ServerAddress]int64 - peerWatermarksLock sync.Mutex + peerRemovedAtNs map[pb.ServerAddress]int64 + // lowWatermarkTsNs and lowFlushWatermarkTsNs are the last minima signalled + // on deliveryAdvanced and flushAdvanced, each closed and replaced when its + // own minimum rises. Held readers park on the one that bounds them: + // arriving data cannot release a watermark hold, only a peer reporting + // the progress that hold waits on. The two are kept apart because a + // delivery advance - one per event a peer streams - cannot release a read + // held at the flush watermark, and waking it costs a whole pass. + lowWatermarkTsNs int64 + lowFlushWatermarkTsNs int64 + deliveryAdvanced chan struct{} + flushAdvanced chan struct{} + peerWatermarksLock sync.Mutex } // MetaAggregator only aggregates data "on the fly". The logs are not re-persisted to disk. @@ -116,6 +127,7 @@ func (ma *MetaAggregator) initPeerWatermark(peer pb.ServerAddress) { ma.peerFlushWatermarks[peer] = 0 } delete(ma.peerRemovedAtNs, peer) + ma.noteLowWatermarksLocked() } func (ma *MetaAggregator) markPeerWatermarkRemoved(peer pb.ServerAddress) { @@ -144,6 +156,8 @@ func (ma *MetaAggregator) dropExpiredRemovedPeersLocked() { delete(ma.peerFlushWatermarks, peer) } } + // Dropping the peer that pinned a minimum raises it, same as it reporting. + ma.noteLowWatermarksLocked() } // advancePeerWatermark records the peer as received-through tsNs. Monotonic, @@ -154,6 +168,7 @@ func (ma *MetaAggregator) advancePeerWatermark(peer pb.ServerAddress, tsNs int64 defer ma.peerWatermarksLock.Unlock() if cur, found := ma.peerWatermarks[peer]; found && tsNs > cur { ma.peerWatermarks[peer] = tsNs + ma.noteLowWatermarksLocked() } } @@ -164,6 +179,7 @@ func (ma *MetaAggregator) advancePeerFlushWatermark(peer pb.ServerAddress, tsNs defer ma.peerWatermarksLock.Unlock() if cur, found := ma.peerFlushWatermarks[peer]; found && tsNs > cur { ma.peerFlushWatermarks[peer] = tsNs + ma.noteLowWatermarksLocked() } } @@ -174,16 +190,7 @@ func (ma *MetaAggregator) PeerLowFlushWatermarkTsNs() int64 { ma.peerWatermarksLock.Lock() defer ma.peerWatermarksLock.Unlock() ma.dropExpiredRemovedPeersLocked() - if len(ma.peerFlushWatermarks) == 0 { - return 0 - } - var low int64 = math.MaxInt64 - for _, tsNs := range ma.peerFlushWatermarks { - if tsNs < low { - low = tsNs - } - } - return low + return lowWatermarkOf(ma.peerFlushWatermarks) } // PeerLowWatermarkTsNs returns the minimum received-through timestamp across @@ -194,11 +201,67 @@ func (ma *MetaAggregator) PeerLowWatermarkTsNs() int64 { ma.peerWatermarksLock.Lock() defer ma.peerWatermarksLock.Unlock() ma.dropExpiredRemovedPeersLocked() - if len(ma.peerWatermarks) == 0 { + return lowWatermarkOf(ma.peerWatermarks) +} + +// DeliveryWatermarkAdvancedChan returns a channel closed the next time the +// delivery low-watermark rises, which is what releases an in-memory read held +// at it. PeerLowFlushWatermarkTsNs has FlushWatermarkAdvancedChan. Callers +// must take the channel before reading the watermark they hold at, so a rise +// in between wakes them instead of being missed. +func (ma *MetaAggregator) DeliveryWatermarkAdvancedChan() <-chan struct{} { + ma.peerWatermarksLock.Lock() + defer ma.peerWatermarksLock.Unlock() + if ma.deliveryAdvanced == nil { + ma.deliveryAdvanced = make(chan struct{}) + } + return ma.deliveryAdvanced +} + +// FlushWatermarkAdvancedChan returns a channel closed the next time the flush +// low-watermark rises, which is what releases a persisted-log read held at it. +func (ma *MetaAggregator) FlushWatermarkAdvancedChan() <-chan struct{} { + ma.peerWatermarksLock.Lock() + defer ma.peerWatermarksLock.Unlock() + if ma.flushAdvanced == nil { + ma.flushAdvanced = make(chan struct{}) + } + return ma.flushAdvanced +} + +// noteLowWatermarksLocked recomputes both minima and wakes the readers parked +// on each one that rose. Caller must hold peerWatermarksLock. +func (ma *MetaAggregator) noteLowWatermarksLocked() { + low, flushLow := lowWatermarkOf(ma.peerWatermarks), lowWatermarkOf(ma.peerFlushWatermarks) + // Falls are recorded too (a joining peer sits at 0 until it signals), so + // the climb back out of one is seen as a rise. + if low > ma.lowWatermarkTsNs { + ma.deliveryAdvanced = closeWatermarkChan(ma.deliveryAdvanced) + } + if flushLow > ma.lowFlushWatermarkTsNs { + ma.flushAdvanced = closeWatermarkChan(ma.flushAdvanced) + } + ma.lowWatermarkTsNs, ma.lowFlushWatermarkTsNs = low, flushLow +} + +// closeWatermarkChan wakes everyone parked on ch and clears it, so the next +// caller parks on a fresh one. +func closeWatermarkChan(ch chan struct{}) chan struct{} { + if ch != nil { + close(ch) + } + return nil +} + +// lowWatermarkOf returns the minimum across the tracked peers, or 0 when none +// are tracked: a peer that has not signalled sits at 0 and pins the minimum +// there, which is what makes completeness unknown. +func lowWatermarkOf(watermarks map[pb.ServerAddress]int64) int64 { + if len(watermarks) == 0 { return 0 } var low int64 = math.MaxInt64 - for _, tsNs := range ma.peerWatermarks { + for _, tsNs := range watermarks { if tsNs < low { low = tsNs } diff --git a/weed/filer/meta_aggregator_testing.go b/weed/filer/meta_aggregator_testing.go new file mode 100644 index 000000000..b7443a28a --- /dev/null +++ b/weed/filer/meta_aggregator_testing.go @@ -0,0 +1,23 @@ +package filer + +import ( + "github.com/seaweedfs/seaweedfs/weed/pb" +) + +// TrackPeerForTesting registers a peer as if the master had announced it, but +// without starting its subscription goroutine, so loop tests can drive the +// low-watermarks by hand. Test support only. +func (ma *MetaAggregator) TrackPeerForTesting(peer pb.ServerAddress) { + ma.peerChansLock.Lock() + ma.peerChans[peer] = make(chan struct{}) + ma.peerChansLock.Unlock() + ma.initPeerWatermark(peer) +} + +// ReportPeerWatermarksForTesting stands in for what a peer's stream reports: +// its delivery watermark (events and idle heartbeats) and its flush +// watermark. Test support only. +func (ma *MetaAggregator) ReportPeerWatermarksForTesting(peer pb.ServerAddress, deliveredTsNs, flushedTsNs int64) { + ma.advancePeerWatermark(peer, deliveredTsNs) + ma.advancePeerFlushWatermark(peer, flushedTsNs) +} diff --git a/weed/filer/meta_aggregator_watermark_test.go b/weed/filer/meta_aggregator_watermark_test.go index bbe10ae9d..f81722adf 100644 --- a/weed/filer/meta_aggregator_watermark_test.go +++ b/weed/filer/meta_aggregator_watermark_test.go @@ -147,3 +147,67 @@ func TestPeerFlushWatermarkBookkeeping(t *testing.T) { } _ = a } + +// closed reports whether ch has been signalled, without blocking. +func closed(ch <-chan struct{}) bool { + select { + case <-ch: + return true + default: + return false + } +} + +// TestWatermarkAdvancedChansAreScoped pins what each wake channel promises. A +// held read re-runs a whole pass - a persisted-log listing in it - so waking +// one whose bound did not move is pure waste, and peers advance their delivery +// watermark on every event they stream. +func TestWatermarkAdvancedChansAreScoped(t *testing.T) { + ma := newTestAggregator() + a, b := pb.ServerAddress("filer-a:8888"), pb.ServerAddress("filer-b:8888") + ma.initPeerWatermark(a) + ma.initPeerWatermark(b) + + delivery, flush := ma.DeliveryWatermarkAdvancedChan(), ma.FlushWatermarkAdvancedChan() + + // One peer alone does not move either minimum: the other still pins both. + ma.advancePeerWatermark(a, 100) + ma.advancePeerFlushWatermark(a, 100) + if closed(delivery) || closed(flush) { + t.Fatalf("a single peer moved a minimum: delivery=%v flush=%v", closed(delivery), closed(flush)) + } + + // Delivery progress on every peer wakes the in-memory holds only. + ma.advancePeerWatermark(b, 50) + if !closed(delivery) { + t.Fatal("delivery low-watermark rose without waking the in-memory holds") + } + if closed(flush) { + t.Fatal("delivery progress woke the persisted-log holds, which it cannot release") + } + + // And flush progress the persisted-log holds only. + delivery = ma.DeliveryWatermarkAdvancedChan() + ma.advancePeerFlushWatermark(b, 50) + if !closed(flush) { + t.Fatal("flush low-watermark rose without waking the persisted-log holds") + } + if closed(delivery) { + t.Fatal("flush progress woke the in-memory holds, which it cannot release") + } + + // A fresh channel is handed out after each signal, so the next park is on + // the next rise rather than on one already consumed. + flush = ma.FlushWatermarkAdvancedChan() + if closed(flush) { + t.Fatal("re-handed a signalled channel") + } + ma.advancePeerFlushWatermark(a, 200) + if closed(flush) { + t.Fatal("one peer above the minimum signalled a rise") + } + ma.advancePeerFlushWatermark(b, 200) + if !closed(flush) { + t.Fatal("the trailing peer catching up did not signal a rise") + } +} diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index 427c94be9..b4c801065 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -50,6 +50,21 @@ const ( // newer. It keeps freshness signals such as filer.sync's sync_offset metric // from looking stuck during read-only periods on the source. idleHeartbeatInterval = 5 * time.Second + + // peerDeliveryClaimInterval paces that heartbeat on the local stream of a + // filer with peers, where it is not a keepalive but a delivery claim: the + // peer aggregator turns it into its delivery low-watermark, and every + // aggregated subscriber in the cluster holds at the minimum across peers. + // So this, not idleHeartbeatInterval, is how far behind live writes a + // quiet filer leaves them. Rounded up to the reader's own poll interval. + peerDeliveryClaimInterval = 200 * time.Millisecond + + // heldWakeFloor coalesces hold releases. Peers advance their watermarks + // per event they stream, and each release costs a whole pass - a log file + // listing included - so releasing on every advance turns a busy cluster + // into a listing storm. It is added to delivery latency, so it stays well + // under the claim interval that already bounds it. + heldWakeFloor = 20 * time.Millisecond ) // metadataStreamSender is satisfied by both gRPC stream types and pipelinedSender. @@ -245,8 +260,9 @@ func memoryHoldsGap(currentTsNs, lastEvictedTsNs int64) bool { // errHeldByPeerWatermark aborts a read at an entry beyond the hold point; the // caller rewinds to the last delivered entry, waits, and re-reads (the -// re-listing is what picks up a late-landing log file). -var errHeldByPeerWatermark = errors.New("held by aggregated peer watermark") +// re-listing is what picks up a late-landing log file). Holding is the normal +// state on a cluster that keeps writing, hence the quiet stop. +var errHeldByPeerWatermark = fmt.Errorf("held by aggregated peer watermark: %w", log_buffer.StopReadingError) // resolveAggReadHoldTsNs bounds how far an aggregated subscriber may read: a // cursor that passes T before every source has provably made T visible loses @@ -654,15 +670,28 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, var diskPassProvenTsNs int64 diskEachLogEntryFn := guardedEachLogEntryFn(func() int64 { return diskPassHoldTsNs }) memEachLogEntryFn := guardedEachLogEntryFn(holdMemTsNs) - // waitHeld pauses a held read until new data or the retry interval (holds - // also release on heartbeats, which do not notify). False: context ended. - waitHeld := func() bool { + // waitHeld pauses a held read until a peer reports further progress, or + // the retry interval elapses (a peer dropped past its grace, or a log file + // landing that no watermark covers). Arriving data is deliberately not a + // wake-up: on a cluster that keeps writing there is always an entry past + // the hold, so waking on it spins the loop without ever releasing the + // hold. The channel is the one for this read's own watermark - a delivery + // advance cannot release a flush-held read - and was taken before the pass + // sampled it, so a rise in between wakes us here instead of being missed. + // False: context ended. + waitHeld := func(scope string, watermarkChan <-chan struct{}) bool { + stats.FilerSubscribeWatermarkHolds.WithLabelValues(scope).Inc() glog.V(3).Infof("held at %v (deliveredUpTo %v, flushLow %v, deliveryLow %v) for %v", time.Unix(0, heldAtTsNs), time.Unix(0, deliveredUpToTsNs), time.Unix(0, fs.filer.MetaAggregator.PeerLowFlushWatermarkTsNs()), time.Unix(0, fs.filer.MetaAggregator.PeerLowWatermarkTsNs()), clientName) select { - case <-aggNotifyChan: + case <-ctx.Done(): + return false + case <-time.After(heldWakeFloor): + } + select { + case <-watermarkChan: case <-ctx.Done(): return false case <-time.After(unflushedGapRetryInterval): @@ -696,6 +725,10 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, glog.V(4).Infof("read on disk %v aggregated subscribe %s from %+v", clientName, req.PathPrefix, lastReadTime) + // Taken before either read samples its watermark, so a rise mid-pass + // cannot land between the sample and the park below. + flushChan := fs.filer.MetaAggregator.FlushWatermarkAdvancedChan() + deliveryChan := fs.filer.MetaAggregator.DeliveryWatermarkAdvancedChan() cursorBeforeDiskTsNs := lastReadTime.Time.UnixNano() // Observe the flush low-watermark before the pass lists files (see @@ -730,7 +763,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, // A hold is not a gap: clear any stale ResumeFromDiskError so the // next pass's disk-miss handling cannot skip past the held entry. readInMemoryLogErr = nil - if !waitHeld() { + if !waitHeld("disk", flushChan) { return nil } continue @@ -836,7 +869,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, // the last delivered entry so nothing in between is skipped. lastReadTime = log_buffer.NewMessagePosition(deliveredUpToTsNs, gapResumeCursorOffset) readInMemoryLogErr = nil - if !waitHeld() { + if !waitHeld("memory", deliveryChan) { return nil } continue @@ -1139,8 +1172,13 @@ func (fs *FilerServer) maybeSendIdleHeartbeat(req *filer_pb.SubscribeMetadataReq // the buffer holds data the subscriber has not reached yet return lastHeartbeatNs } + isLocalStream := fs.filer != nil && logBuffer == fs.filer.LocalMetaLogBuffer + interval := idleHeartbeatInterval + if isLocalStream && fs.filer.MetaAggregator != nil && fs.filer.MetaAggregator.HasRemotePeers() { + interval = peerDeliveryClaimInterval + } now := time.Now().UnixNano() - if now-lastHeartbeatNs < int64(idleHeartbeatInterval) { + if now-lastHeartbeatNs < int64(interval) { return lastHeartbeatNs } // On the local stream the heartbeat is a delivery claim to a peer @@ -1152,7 +1190,7 @@ func (fs *FilerServer) maybeSendIdleHeartbeat(req *filer_pb.SubscribeMetadataReq // (fenced above it), or already appended here - and then the head check // proves this stream has sent it before the heartbeat. heartbeat := &filer_pb.SubscribeMetadataResponse{TsNs: now} - if fs.filer != nil && logBuffer == fs.filer.LocalMetaLogBuffer { + if isLocalStream { heartbeat.TsNs = fs.filer.LocalDeliveredThroughTsNs(now) heartbeat.FlushedTsNs = fs.filer.LocalFlushedThroughTsNs(now) if logBuffer.LastTsNs.Load() > floorTsNs { diff --git a/weed/server/filer_subscribe_agg_test.go b/weed/server/filer_subscribe_agg_test.go new file mode 100644 index 000000000..9d7a422f2 --- /dev/null +++ b/weed/server/filer_subscribe_agg_test.go @@ -0,0 +1,318 @@ +package weed_server + +// End-to-end tests for the aggregated (SubscribeMetadata) loop's peer +// watermark holds. Peers report their progress at their own pace, so on a +// cluster that keeps writing there is almost always an entry newer than the +// low-watermark: a hold is the normal state, and has to be quiet and paced +// rather than logged and retried per arriving event. + +import ( + "context" + "fmt" + "net" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + "google.golang.org/grpc/peer" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/stats" +) + +const ( + testSelfAddress = pb.ServerAddress("self:8888") + testPeerAddress = pb.ServerAddress("peer:8888") +) + +// startAggregator gives the harness a meta aggregator with one remote peer, so +// SubscribeMetadata takes the aggregated path instead of delegating to +// SubscribeLocalMetadata. +func (h *subscribeHarness) startAggregator() *filer.MetaAggregator { + ma := filer.NewMetaAggregator(h.f, testSelfAddress, nil) + ma.TrackPeerForTesting(testSelfAddress) + ma.TrackPeerForTesting(testPeerAddress) + h.f.MetaAggregator = ma + h.t.Cleanup(ma.MetaLogBuffer.ShutdownLogBuffer) + return ma +} + +// reportPeers stands in for both peers' streams reporting through tsNs. +func reportPeers(ma *filer.MetaAggregator, tsNs int64) { + reportPeersAt(ma, tsNs, tsNs) +} + +// reportPeersAt is reportPeers with the two watermarks apart: a peer streams +// an event well before it flushes it, so they advance independently. +func reportPeersAt(ma *filer.MetaAggregator, deliveredTsNs, flushedTsNs int64) { + ma.ReportPeerWatermarksForTesting(testSelfAddress, deliveredTsNs, flushedTsNs) + ma.ReportPeerWatermarksForTesting(testPeerAddress, deliveredTsNs, flushedTsNs) +} + +func (h *subscribeHarness) appendAggregated(tsNs int64) { + if err := h.f.MetaAggregator.MetaLogBuffer.AddLogEntryToBuffer(testEvent(tsNs, fmt.Sprintf("a-%d", tsNs))); err != nil { + h.t.Fatalf("append aggregated: %v", err) + } +} + +func (h *subscribeHarness) subscribeAggregated(sinceNs int64) *runningSubscribe { + // A peer in the context keeps findClientAddress quiet, so an ERROR count + // across the run sees only what the subscribe loop itself writes. + ctx, cancel := context.WithCancel(peer.NewContext(context.Background(), + &peer.Peer{Addr: &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 12345}})) + stream := &fakeSubscribeStream{ctx: ctx} + req := &filer_pb.SubscribeMetadataRequest{ + ClientName: "agg-loop-test", + ClientId: 11, + ClientEpoch: 1, + SinceNs: sinceNs, + } + r := &runningSubscribe{stream: stream, cancel: cancel, done: make(chan error, 1), finished: make(chan struct{})} + go func() { + r.done <- h.fs.SubscribeMetadata(req, stream) + close(r.finished) + }() + h.t.Cleanup(func() { + cancel() + select { + case <-r.finished: + case <-time.After(5 * time.Second): + h.t.Error("aggregated subscribe loop did not exit on cancel") + } + }) + return r +} + +// heldSubscriber starts a subscriber whose peers reported a moment ago and +// then went quiet - the hold sits just behind live writes, exactly where a +// quiet peer's idle heartbeat leaves it - and returns it with the write +// timestamp the peers last reported. +func (h *subscribeHarness) heldSubscriber() (*filer.MetaAggregator, *runningSubscribe, int64) { + ma := h.startAggregator() + frozen := time.Now().Add(-500 * time.Millisecond).UnixNano() + reportPeers(ma, frozen) + return ma, h.subscribeAggregated(frozen), frozen +} + +// writeFor appends an entry every 2ms for d, reporting the peers through the +// PREVIOUS entry each time when report is set: the live-cluster shape, where +// the low-watermark keeps moving but always trails the newest entry, so every +// read ends held. +func (h *subscribeHarness) writeFor(ma *filer.MetaAggregator, d time.Duration, report bool) (written []int64, elapsed time.Duration) { + start := time.Now() + for stop := time.After(d); ; { + select { + case <-stop: + if report && len(written) > 0 { + reportPeers(ma, written[len(written)-1]) + } + return written, time.Since(start) + default: + } + ts := time.Now().UnixNano() + h.appendAggregated(ts) + if report && len(written) > 0 { + reportPeers(ma, written[len(written)-1]) + } + written = append(written, ts) + time.Sleep(2 * time.Millisecond) + } +} + +// heldReads reads back the hold counter the loop keeps in place of the log +// line, which is also how an operator sees a held subscriber now. +func heldReads(scopes ...string) int { + if len(scopes) == 0 { + scopes = []string{"memory", "disk"} + } + var total int + for _, scope := range scopes { + total += int(testutil.ToFloat64(stats.FilerSubscribeWatermarkHolds.WithLabelValues(scope))) + } + return total +} + +// waitForEventsAtLeastOnce asserts every want arrives, in order. The cursor a +// hold rewinds to is inclusive of the last delivered entry, so each cycle +// repeats it - at-least-once, which is the contract; a skip is not. +func waitForEventsAtLeastOnce(t *testing.T, r *runningSubscribe, want []int64, timeout time.Duration) { + t.Helper() + last := want[len(want)-1] + deadline := time.Now().Add(timeout) + var got []int64 + for time.Now().Before(deadline) { + got = eventTimestamps(r.stream.snapshot()) + if len(got) > 0 && got[len(got)-1] >= last { + break + } + time.Sleep(10 * time.Millisecond) + } + seen := make(map[int64]bool, len(got)) + var prev int64 + for _, ts := range got { + if ts < prev { + t.Fatalf("delivered %v after %v: out of order", time.Unix(0, ts), time.Unix(0, prev)) + } + prev, seen[ts] = ts, true + } + for i, ts := range want { + if !seen[ts] { + t.Fatalf("event %d of %d (%v) never delivered", i+1, len(want), time.Unix(0, ts)) + } + } +} + +// TestSubscribeLoop_AggregatedHoldIsQuiet: every held read used to log an +// ERROR, so a busy cluster wrote one line per arriving event. +func TestSubscribeLoop_AggregatedHoldIsQuiet(t *testing.T) { + h := newSubscribeHarness(t) + ma, r, _ := h.heldSubscriber() + + errorsBefore := glog.Stats.Error.Lines() + written, elapsed := h.writeFor(ma, time.Second, false) + errorLines := glog.Stats.Error.Lines() - errorsBefore + + t.Logf("%d writes over %v produced %d ERROR lines", len(written), elapsed, errorLines) + if errorLines > 0 { + t.Fatalf("a held read logged %d ERROR lines over %d writes; holds are control flow", errorLines, len(written)) + } + if got := eventTimestamps(r.stream.snapshot()); len(got) > 0 { + t.Fatalf("delivered %d events past the peers' watermark", len(got)) + } +} + +// TestSubscribeLoop_AggregatedHoldIsPaced is the other half: a held read used +// to wait on the buffer's data channel, which the very next write signalled, +// so the loop re-ran a whole pass - a log file listing in it - per event. +// Arriving data cannot release a hold; only peer progress can. +func TestSubscribeLoop_AggregatedHoldIsPaced(t *testing.T) { + h := newSubscribeHarness(t) + // Long enough that a hold paced by the retry interval alone is far below + // one hold per write, short enough to keep the test quick. + prevRetry := unflushedGapRetryInterval + unflushedGapRetryInterval = 200 * time.Millisecond + t.Cleanup(func() { unflushedGapRetryInterval = prevRetry }) + ma, _, _ := h.heldSubscriber() + + holdsBefore := heldReads() + written, elapsed := h.writeFor(ma, time.Second, false) + holds := heldReads() - holdsBefore + + t.Logf("%d writes over %v produced %d holds", len(written), elapsed, holds) + // The watermarks never moved, so the retry interval alone paces the holds. + if maxHolds := int(elapsed/unflushedGapRetryInterval) + 2; holds > maxHolds { + t.Fatalf("held %d times in %v, want at most %d: a hold must wait for peer progress, not for the next write", + holds, elapsed, maxHolds) + } +} + +// TestSubscribeLoop_AggregatedHoldReleasesOnPeerProgress pins what pacing the +// hold may not cost: once the peers report through the held entry it must be +// delivered, without waiting out the retry interval. +func TestSubscribeLoop_AggregatedHoldReleasesOnPeerProgress(t *testing.T) { + h := newSubscribeHarness(t) + // Far longer than this test runs: only the watermark signal can deliver. + prevRetry := unflushedGapRetryInterval + unflushedGapRetryInterval = 30 * time.Second + t.Cleanup(func() { unflushedGapRetryInterval = prevRetry }) + ma, r, _ := h.heldSubscriber() + + ts := time.Now().UnixNano() + h.appendAggregated(ts) + assertNoEventsFor(t, r, 200*time.Millisecond) + + reportPeers(ma, ts) + waitForEvents(t, r, []int64{ts}, 3*time.Second) +} + +// TestSubscribeLoop_AggregatedHoldCoalescesPeerProgress covers the rest of the +// cost: peers advance their watermark on every event they stream, so releasing +// a hold per advance is the same pass-per-event storm as releasing per write, +// just without the log lines. +func TestSubscribeLoop_AggregatedHoldCoalescesPeerProgress(t *testing.T) { + h := newSubscribeHarness(t) + ma, r, _ := h.heldSubscriber() + + holdsBefore := heldReads() + written, elapsed := h.writeFor(ma, time.Second, true) + holds := heldReads() - holdsBefore + + t.Logf("%d writes over %v produced %d holds", len(written), elapsed, holds) + if maxHolds := int(elapsed/heldWakeFloor) + 2; holds > maxHolds { + t.Fatalf("held %d times in %v, want at most %d: releases must coalesce", holds, elapsed, maxHolds) + } + // Coalescing may not lose anything the peers reported through. + waitForEventsAtLeastOnce(t, r, written, 3*time.Second) +} + +// TestPeerDeliveryClaimPacing pins what a filer with peers owes them: the idle +// heartbeat on its local stream is a delivery claim every aggregated +// subscriber in the cluster holds at, so it refreshes at the claim interval, +// not at the keepalive interval that used to park them ~5s behind live writes. +func TestPeerDeliveryClaimPacing(t *testing.T) { + h := newSubscribeHarness(t) + req := &filer_pb.SubscribeMetadataRequest{ClientSupportsIdleHeartbeat: true} + lb := h.f.LocalMetaLogBuffer + // Claimed a second ago: stale for a delivery claim, fresh for a keepalive. + claimedAtNs := time.Now().Add(-time.Second).UnixNano() + + s := &collectingStream{} + if got := h.fs.maybeSendIdleHeartbeat(req, s, lb, 0, 0, claimedAtNs); got != claimedAtNs || len(s.messages) != 0 { + t.Fatalf("standalone filer sent %d heartbeats early", len(s.messages)) + } + + h.startAggregator() + s = &collectingStream{} + if got := h.fs.maybeSendIdleHeartbeat(req, s, lb, 0, 0, claimedAtNs); got == claimedAtNs || len(s.messages) != 1 { + t.Fatalf("filer with peers sent %d delivery claims, want 1", len(s.messages)) + } + if s.messages[0].TsNs <= claimedAtNs { + t.Fatalf("claim ts %d did not advance past %d", s.messages[0].TsNs, claimedAtNs) + } +} + +// TestSubscribeLoop_AggregatedDiskHoldIgnoresDeliveryProgress pins that each +// hold parks on its own watermark. A persisted-log read is held by what the +// peers have FLUSHED, and peers advance their DELIVERY watermark on every +// event they stream, so waking it on delivery progress would re-list a day of +// log files per event and re-park on the same entry. +func TestSubscribeLoop_AggregatedDiskHoldIgnoresDeliveryProgress(t *testing.T) { + h := newSubscribeHarness(t) + prevRetry := unflushedGapRetryInterval + unflushedGapRetryInterval = 500 * time.Millisecond + t.Cleanup(func() { unflushedGapRetryInterval = prevRetry }) + + // Recent enough that the settled horizon does not take the hold over. + onDisk := time.Now().Add(-30 * time.Second).UnixNano() + h.append(onDisk) + h.f.LocalMetaLogBuffer.ForceFlush() + waitForFlushedFiles(t, h, onDisk) + + ma := h.startAggregator() + // Flushed below the entry on disk, delivered well past it: only the disk + // pass is held. + reportPeersAt(ma, time.Now().UnixNano(), onDisk-int64(time.Second)) + r := h.subscribeAggregated(onDisk - int64(5*time.Second)) + assertNoEventsFor(t, r, 200*time.Millisecond) + + holdsBefore := heldReads("disk") + start := time.Now() + for deadline := start.Add(time.Second); time.Now().Before(deadline); { + reportPeersAt(ma, time.Now().UnixNano(), 0) + time.Sleep(2 * time.Millisecond) + } + holds, elapsed := heldReads("disk")-holdsBefore, time.Since(start) + + t.Logf("%v of delivery-only progress produced %d disk holds", elapsed, holds) + if maxHolds := int(elapsed/unflushedGapRetryInterval) + 2; holds > maxHolds { + t.Fatalf("held on disk %d times in %v, want at most %d: delivery progress cannot release a flush hold", + holds, elapsed, maxHolds) + } + + // The flush watermark reaching the entry is what delivers it. + reportPeersAt(ma, 0, onDisk) + waitForEvents(t, r, []int64{onDisk}, 3*time.Second) +} diff --git a/weed/stats/metrics.go b/weed/stats/metrics.go index a5c0646af..4ef7ad74d 100644 --- a/weed/stats/metrics.go +++ b/weed/stats/metrics.go @@ -242,6 +242,14 @@ var ( Help: "Times a metadata subscriber moved past a log range without proof it was persisted: scope=aggregated means a peer may not have flushed it, scope=local means this filer's own log flush was wedged past the give-up bound.", }, []string{"scope"}) + FilerSubscribeWatermarkHolds = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: Namespace, + Subsystem: subsystemFiler, + Name: "subscribe_watermark_holds", + Help: "Times an aggregated metadata read stopped at an entry newer than the peers' low-watermark and waited for a peer to report further progress: scope=memory held at the delivery watermark, scope=disk at the flush watermark.", + }, []string{"scope"}) + FilerSubscribeGapStalledGauge = prometheus.NewGaugeVec( prometheus.GaugeOpts{ Namespace: Namespace, @@ -907,6 +915,7 @@ func init() { Gather.MustRegister(FilerServerLastSendTsOfSubscribeGauge) Gather.MustRegister(FilerSubscribeGapStalledGauge) Gather.MustRegister(FilerSubscribeUnprovenGapCrossings) + Gather.MustRegister(FilerSubscribeWatermarkHolds) Gather.MustRegister(FilerMetaAggregatorReplayFailures) Gather.MustRegister(FilerObjectSizeBytesHistogram) Gather.MustRegister(collectors.NewGoCollector()) diff --git a/weed/util/log_buffer/log_read.go b/weed/util/log_buffer/log_read.go index 12b839d9a..afbee3e9b 100644 --- a/weed/util/log_buffer/log_read.go +++ b/weed/util/log_buffer/log_read.go @@ -14,6 +14,10 @@ import ( var ( ResumeError = fmt.Errorf("resume") ResumeFromDiskError = fmt.Errorf("resumeFromDisk") + // StopReadingError, wrapped by an eachLogDataFn's error, marks a read the + // callback ended on purpose - control flow, not a failure - so the loop + // hands it back to the caller without logging it. + StopReadingError = fmt.Errorf("stopReading") ) // notificationHealthCheckInterval bounds how long an idle subscriber blocks @@ -297,7 +301,9 @@ func (logBuffer *LogBuffer) LoopProcessLogData(readerName string, startPosition lastReadPosition = NewMessagePosition(logEntry.TsNs, batchIndex) if isDone, err = eachLogDataFn(logEntry); err != nil { - glog.Errorf("LoopProcessLogData: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err) + if !errors.Is(err, StopReadingError) { + glog.Errorf("LoopProcessLogData: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err) + } return } if isDone { @@ -570,7 +576,9 @@ func (logBuffer *LogBuffer) LoopProcessLogDataWithOffset(readerName string, star glog.V(4).Infof("Calling eachLogDataFn for entry at offset %d, next position will be %d", logEntry.Offset, logEntry.Offset+1) if isDone, err = eachLogDataFn(logEntry, logEntry.Offset); err != nil { - glog.Errorf("LoopProcessLogDataWithOffset: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err) + if !errors.Is(err, StopReadingError) { + glog.Errorf("LoopProcessLogDataWithOffset: %s process log entry %d key:%q ts_ns:%d offset:%d size:%d: %v", readerName, batchSize+1, logEntry.Key, logEntry.TsNs, logEntry.Offset, len(logEntry.Data), err) + } return } if isDone {