From 048f9ece2da39cd071d401f93693d7ec9059b6e1 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 9 Jun 2026 11:43:12 -0700 Subject: [PATCH] Fix filer metadata-replay OOM under mount reconnect storms (#9901) * fix(filer): propagate multi-filer metadata log read errors A genuine (non not-found) read error in one filer's log stream was logged and skipped, then the merged cursor advanced past the gap, silently dropping that file's events. Abort the whole replay so the subscriber re-reads from the unchanged position; chunk-not-found still skips. * perf(mount): read persisted metadata log chunks directly from volume servers Set LogFileReaderFn so the filer returns log file references and the mount reads the chunk data itself, instead of the filer reading, decoding, and streaming every persisted entry. Keeps a reconnect storm of many mounts from concentrating hundreds of concurrent log replays in filer memory. * perf(filer): pre-size chunk stream reader buffer to view size The chunk size is known up front, so grow the buffer once instead of letting bytes.Buffer double as the streamed pieces arrive (which transiently overshoots to ~2x per reader). * fix(filer): bound concurrent persisted-log replays Each server-side replay holds an open chunk reader per source filer plus a readahead buffer, so a reconnect storm of clients that predate the metadata-chunks offload multiplies into many GB. Gate replays with a semaphore; abort the acquire when the subscriber's stream is gone so cancelled clients do not pile up parked goroutines. --- weed/filer/filer_notify.go | 22 +++++- weed/filer/stream.go | 2 + weed/mount/meta_cache/meta_cache_subscribe.go | 8 ++ weed/pb/filer_pb_direct_read.go | 77 ++++++++++++++++--- weed/pb/filer_pb_direct_read_test.go | 51 ++++++++++++ weed/server/filer_grpc_server_sub_meta.go | 4 +- 6 files changed, 149 insertions(+), 15 deletions(-) diff --git a/weed/filer/filer_notify.go b/weed/filer/filer_notify.go index 311874875..c66235afc 100644 --- a/weed/filer/filer_notify.go +++ b/weed/filer/filer_notify.go @@ -200,7 +200,27 @@ func isChunkNotFoundError(err error) bool { httpNotFoundPattern.MatchString(errMsg) } -func (f *Filer) ReadPersistedLogBuffer(startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) { +// persistedLogReplayLimit caps concurrent legacy replays; each holds a chunk +// reader per source filer, so a reconnect storm of pre-offload clients would +// otherwise pin many GB. Metadata-chunks clients take sendLogFileRefs and never +// reach this path. +const persistedLogReplayLimit = 16 + +var persistedLogReplaySem = make(chan struct{}, persistedLogReplayLimit) + +func (f *Filer) ReadPersistedLogBuffer(ctx context.Context, startPosition log_buffer.MessagePosition, stopTsNs int64, eachLogEntryFn log_buffer.EachLogEntryFuncType) (lastTsNs int64, isDone bool, err error) { + + // Cap concurrent replays; bail if the stream is already gone so cancelled + // clients do not park on the semaphore. + if err := ctx.Err(); err != nil { + return 0, false, err + } + select { + case persistedLogReplaySem <- struct{}{}: + defer func() { <-persistedLogReplaySem }() + case <-ctx.Done(): + return 0, false, ctx.Err() + } visitor, visitErr := f.collectPersistedLogBuffer(startPosition, stopTsNs) if visitErr != nil { diff --git a/weed/filer/stream.go b/weed/filer/stream.go index de53134d1..4acf93862 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -484,6 +484,8 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error { return err } var buffer bytes.Buffer + // pre-size to the known chunk size; avoids bytes.Buffer's doubling regrowth + buffer.Grow(int(chunkView.ViewSize)) var shouldRetry bool jwt := JwtForVolumeServer(chunkView.FileId) for _, urlString := range urlStrings { diff --git a/weed/mount/meta_cache/meta_cache_subscribe.go b/weed/mount/meta_cache/meta_cache_subscribe.go index d2f472468..18b35446c 100644 --- a/weed/mount/meta_cache/meta_cache_subscribe.go +++ b/weed/mount/meta_cache/meta_cache_subscribe.go @@ -2,8 +2,10 @@ package meta_cache import ( "context" + "io" "strings" + "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" @@ -66,6 +68,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil prefix = prefix + "/" } + // Read persisted log chunks directly from volume servers, keeping the replay + // cost off the filer's heap (see LogFileReaderFn below). + lookupFn := filer.LookupFn(client) metadataFollowOption := &pb.MetadataFollowOption{ ClientName: "mount", ClientId: selfSignature, @@ -77,6 +82,9 @@ func SubscribeMetaEvents(mc *MetaCache, selfSignature int32, client filer_pb.Fil StartTsNs: lastTsNs, StopTsNs: 0, EventErrorType: pb.FatalOnError, + LogFileReaderFn: func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + return filer.NewChunkStreamReaderFromLookup(context.Background(), lookupFn, chunks), nil + }, } util.RetryUntil("followMetaUpdates", func() error { metadataFollowOption.ClientEpoch++ diff --git a/weed/pb/filer_pb_direct_read.go b/weed/pb/filer_pb_direct_read.go index 294b78c7a..2a1a78c32 100644 --- a/weed/pb/filer_pb_direct_read.go +++ b/weed/pb/filer_pb_direct_read.go @@ -144,6 +144,23 @@ func readMultiFilersMerged( streams := make([]filerStream, len(filerOrder)) var wg sync.WaitGroup + // A genuine (non chunk-not-found) read error must fail the whole replay: the + // caller advances its cursor only on success, so a swallowed error leaves a + // permanent gap. stop aborts the other readers; fatalErr is read on exit. + stop := make(chan struct{}) + var stopOnce sync.Once + closeStop := func() { stopOnce.Do(func() { close(stop) }) } + var fatalMu sync.Mutex + var fatalErr error + setFatal := func(e error) { + fatalMu.Lock() + if fatalErr == nil { + fatalErr = e + } + fatalMu.Unlock() + closeStop() + } + for i, filerId := range filerOrder { entryCh := make(chan *filer_pb.LogEntry, 512) streams[i] = filerStream{filerId: filerId, entryCh: entryCh} @@ -152,10 +169,20 @@ func readMultiFilersMerged( go func(refs []*filer_pb.LogFileChunkRef, ch chan *filer_pb.LogEntry) { defer wg.Done() defer close(ch) - readFilerFilesToChannel(refs, newReader, startTsNs, stopTsNs, ch) + readFilerFilesToChannel(refs, newReader, startTsNs, stopTsNs, ch, stop, setFatal) }(perFiler[filerId], entryCh) } + // Stop readers, drain channels so none block on a send, then wait for exit. + drainAndWait := func() { + closeStop() + for i := range streams { + for range streams[i].entryCh { + } + } + wg.Wait() + } + // Seed the min-heap with the first entry from each filer pq := &logEntryHeap{} heap.Init(pq) @@ -167,15 +194,23 @@ func readMultiFilersMerged( // Merge loop for pq.Len() > 0 { + // stop is closed only by setFatal here, so a closed stop means a reader + // aborted; lock-free bail on the hot path. + select { + case <-stop: + drainAndWait() + fatalMu.Lock() + fe := fatalErr + fatalMu.Unlock() + return lastTsNs, fe + default: + } + item := heap.Pop(pq).(*logEntryHeapItem) lastTsNs, err = processOneLogEntry(item.entry, filter, processEventFn) if err != nil { - for i := range streams { - for range streams[i].entryCh { - } - } - wg.Wait() + drainAndWait() return } @@ -184,7 +219,13 @@ func readMultiFilersMerged( } } - wg.Wait() + drainAndWait() + fatalMu.Lock() + fe := fatalErr + fatalMu.Unlock() + if fe != nil { + return lastTsNs, fe + } return } @@ -193,6 +234,8 @@ func readFilerFilesToChannel( newReader LogFileReaderFn, startTsNs, stopTsNs int64, ch chan *filer_pb.LogEntry, + stop <-chan struct{}, + setFatal func(error), ) { type prefetchResult struct { entries []*filer_pb.LogEntry @@ -214,7 +257,12 @@ func readFilerFilesToChannel( } for i, ref := range refs { - result := <-pendingCh + var result prefetchResult + select { + case result = <-pendingCh: + case <-stop: + return + } if i+1 < len(refs) { pendingCh = startPrefetch(refs[i+1]) @@ -223,14 +271,19 @@ func readFilerFilesToChannel( if result.err != nil { if isChunkNotFound(result.err) { glog.V(0).Infof("skip log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err) - } else { - glog.Errorf("read log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err) + continue } - continue + glog.Errorf("read log file filer=%s ts=%d: %v", ref.FilerId, ref.FileTsNs, result.err) + setFatal(fmt.Errorf("read log file filer=%s ts=%d: %w", ref.FilerId, ref.FileTsNs, result.err)) + return } for _, entry := range result.entries { - ch <- entry + select { + case ch <- entry: + case <-stop: + return + } } } } diff --git a/weed/pb/filer_pb_direct_read_test.go b/weed/pb/filer_pb_direct_read_test.go index 68f71a306..19a8e20d8 100644 --- a/weed/pb/filer_pb_direct_read_test.go +++ b/weed/pb/filer_pb_direct_read_test.go @@ -278,3 +278,54 @@ func TestDirectReadVsServerSideThroughput(t *testing.T) { t.Logf("Speedup: %.1fx (parallel + prefetch + no gRPC vs server-side sequential)", directRate/serverRate) } } + +// failingReaderFn returns err for the given file key, delegating otherwise. +func failingReaderFn(base LogFileReaderFn, failKey string, err error) LogFileReaderFn { + return func(chunks []*filer_pb.FileChunk) (io.ReadCloser, error) { + if len(chunks) > 0 && chunks[0].FileId == failKey { + return nil, err + } + return base(chunks) + } +} + +// A real (non not-found) read error must fail the whole replay, not silently +// drop the file and advance the cursor. +func TestReadLogFileRefsMultiFilerGenuineErrorAborts(t *testing.T) { + files := newTestLogFiles(3, 2, 10, 0) + failKey := files.refs[2].Chunks[0].FileId // filer01's first file + readerFn := failingReaderFn(files.readerFn(), failKey, fmt.Errorf("failed to locate %s", failKey)) + + var count int64 + _, err := ReadLogFileRefs(files.refs, readerFn, 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { + atomic.AddInt64(&count, 1) + return nil + }) + if err == nil { + t.Fatalf("expected error from genuine read failure, got nil (delivered=%d)", count) + } +} + +// A chunk-not-found error skips only that file (volume gone), not the replay. +func TestReadLogFileRefsMultiFilerNotFoundSkips(t *testing.T) { + files := newTestLogFiles(3, 2, 10, 0) + skipKey := files.refs[2].Chunks[0].FileId // filer01's first file + readerFn := failingReaderFn(files.readerFn(), skipKey, fmt.Errorf("volume not found: %s", skipKey)) + + var count int64 + _, err := ReadLogFileRefs(files.refs, readerFn, 0, 0, + PathFilter{PathPrefix: "/"}, + func(resp *filer_pb.SubscribeMetadataResponse) error { + atomic.AddInt64(&count, 1) + return nil + }) + if err != nil { + t.Fatalf("chunk-not-found should be skipped, got error: %v", err) + } + expected := int64(files.totalEvents() - 10) // one skipped file's events + if count != expected { + t.Fatalf("expected %d events after skipping one file, got %d", expected, count) + } +} diff --git a/weed/server/filer_grpc_server_sub_meta.go b/weed/server/filer_grpc_server_sub_meta.go index 6bec4e2c7..0b5a1c3f8 100644 --- a/weed/server/filer_grpc_server_sub_meta.go +++ b/weed/server/filer_grpc_server_sub_meta.go @@ -215,7 +215,7 @@ func (fs *FilerServer) SubscribeMetadata(req *filer_pb.SubscribeMetadataRequest, if req.ClientSupportsMetadataChunks { processedTsNs, isDone, readPersistedLogErr = fs.sendLogFileRefs(ctx, stream, lastReadTime, req.UntilNs) } else { - processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn) + processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn) } if readPersistedLogErr != nil { return fmt.Errorf("reading from persisted logs: %w", readPersistedLogErr) @@ -367,7 +367,7 @@ func (fs *FilerServer) SubscribeLocalMetadata(req *filer_pb.SubscribeMetadataReq if req.ClientSupportsMetadataChunks { processedTsNs, isDone, readPersistedLogErr = fs.sendLogFileRefs(ctx, stream, lastReadTime, req.UntilNs) } else { - processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(lastReadTime, req.UntilNs, eachLogEntryFn) + processedTsNs, isDone, readPersistedLogErr = fs.filer.ReadPersistedLogBuffer(ctx, lastReadTime, req.UntilNs, eachLogEntryFn) } if readPersistedLogErr != nil { glog.V(0).Infof("read on disk %v local subscribe %s from %+v: %v", clientName, req.PathPrefix, lastReadTime, readPersistedLogErr)