From c1ccbe97dd1ca55d1a67f64330e1756588c73073 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 17 Apr 2026 21:21:32 -0700 Subject: [PATCH] feat(filer.backup): -initialSnapshot seeds destination from live tree (#9126) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(filer.backup): -initialSnapshot seeds destination from live tree Replaying the metadata event log on a fresh sync only leaves files that still exist on the source at replay time: any entry that was created and later deleted is replayed as a create/delete pair and never materializes on the destination. Users who wipe the destination and re-run filer.backup therefore see "only new files" instead of a full backup, even when -timeAgo=876000h is passed and the subscription genuinely starts from epoch (ref discussion #8672). Add a -initialSnapshot opt-in flag: when set on a fresh sync (no prior checkpoint, -timeAgo unset), walk the live filer tree under -filerPath via TraverseBfs and seed the destination through sink.CreateEntry, then persist the walk-start timestamp as the checkpoint and subscribe from there. Capturing the timestamp before the walk lets the subscription catch any create/update/delete racing with the walk — sink CreateEntry is idempotent across the builtin sinks so replay is safe. Honors existing -filerExcludePaths / -filerExcludeFileNames / -filerExcludePathPatterns filters and skips /topics/.system/log the same way the subscription path does. Also log "starting from (no prior checkpoint)" instead of a misleading "resuming from 1970-01-01" when the KV has no stored offset. * fix(filer.backup): guard initialSnapshot counters under TraverseBfs workers TraverseBfs fans the callback out across 5 worker goroutines, so the entryCount / byteCount updates and the 5-second progress-log gate in runInitialSnapshot were racing. Switch the counters to atomic.Int64 and protect the lastLog check/update with a short-scoped mutex so the heavy sink.CreateEntry call stays outside the critical section. Flagged by gemini-code-assist on #9126; verified with go test -race. * fix(filer.backup): harden initialSnapshot against transient errors and path edge cases Three review items from CodeRabbit on #9126: 1. getOffset errors no longer leave isFreshSync=true. Before, a transient KV read failure would cause runFilerBackup's retry loop to redo the full -initialSnapshot walk on every retry. Treat any offset-read error as "not fresh" so the snapshot only runs when we've verified there really is no prior checkpoint. 2. initialSnapshotTargetKey now normalizes sourcePath to a trailing- slash base before stripping the prefix, so edge cases where sourceKey equals sourcePath (trailing-slash mismatch or root-entry emission) no longer index past the end. Unit tests cover both forms. 3. Documented the TraverseBfs-enumerates-excluded-subtrees performance characteristic on runInitialSnapshot, since pruning requires a separate change to TraverseBfs itself. * fix(filer.backup): retry setOffset after initialSnapshot to avoid full re-walks If the snapshot walk finishes but the subsequent setOffset fails, the retry loop in runFilerBackup will re-enter doFilerBackup with an empty checkpoint and run the full BFS again — on a multi-million-entry tree that's hours of wasted work over a 100-byte KV write. Retry the write a handful of times with exponential backoff before giving up, and log loudly at the final failure (with snapshotTsNs + sinkId) so operators recognize the symptom instead of guessing at mysterious repeated walks. Nitpick raised by CodeRabbit on #9126. * fix(filer.backup): initialSnapshot ignore404, skew margin, exclude dir-entry itself Three review items from CodeRabbit on #9126: 1. ignore404Error now threads into runInitialSnapshot. If a file is listed by TraverseBfs and then deleted before CreateEntry reads its chunks, the follow path already ignores 404s — the snapshot path was aborting and triggering a full re-walk. Treat an ignorable 404 as "skip this entry, continue." 2. snapshotTsNs now uses `time.Now() - 1min` instead of `time.Now()`. Metadata events are stamped server-side, so a fast backup-host clock could skip events that fire during or right after the walk. Matches the 1-minute margin meta_aggregator.go applies on initial peer traversal; duplicate replay is harmless because CreateEntry is idempotent. 3. Exclude checks now run against the entry's own full path, not just its parent. A walked directory whose full path matches SystemLogDir or -filerExcludePaths was being seeded to the destination; only its descendants were being skipped. Verified with a manual repro where -filerExcludePaths=/data/skipdir now keeps the skipdir entry itself off the destination. * refactor(filer): share destKey helper between buildKey and initialSnapshot Extract destKey(dataSink, targetPath, sourcePath, sourceKey, mTime) from buildKey in filer_sync.go. Both the event-log path (buildKey) and the initialSnapshot walk (initialSnapshotTargetKey) now go through the same helper, so a walk-seeded file and an event-replayed file always resolve to the same destination key. As a bonus, buildKey picks up the defensive trailing-slash normalization that initialSnapshotTargetKey introduced — no more index-past-end risk when sourceKey happens to equal sourcePath. Also tightens the mTime lookup to guard against nil Attributes (caught by an existing test against buildKey when I first moved the lookup out of the incremental branch). --- weed/command/filer_backup.go | 196 +++++++++++++++++++++++++++++- weed/command/filer_backup_test.go | 68 +++++++++++ weed/command/filer_sync.go | 44 +++++-- 3 files changed, 293 insertions(+), 15 deletions(-) diff --git a/weed/command/filer_backup.go b/weed/command/filer_backup.go index 84cbf4828..f3d4b32b6 100644 --- a/weed/command/filer_backup.go +++ b/weed/command/filer_backup.go @@ -1,17 +1,22 @@ package command import ( + "context" "errors" "fmt" nethttp "net/http" "regexp" "strings" + "sync" + "sync/atomic" "time" + "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/replication/repl_util" + "github.com/seaweedfs/seaweedfs/weed/replication/sink" "github.com/seaweedfs/seaweedfs/weed/replication/source" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/util" @@ -35,6 +40,7 @@ type FilerBackupOptions struct { ignore404Error *bool timeAgo *time.Duration retentionDays *int + initialSnapshot *bool } var ( @@ -57,6 +63,7 @@ func init() { filerBackupOptions.retentionDays = cmdFilerBackup.Flag.Int("retentionDays", 0, "incremental backup retention days") filerBackupOptions.disableErrorRetry = cmdFilerBackup.Flag.Bool("disableErrorRetry", false, "disables errors retry, only logs will print") filerBackupOptions.ignore404Error = cmdFilerBackup.Flag.Bool("ignore404Error", true, "ignore 404 errors from filer") + filerBackupOptions.initialSnapshot = cmdFilerBackup.Flag.Bool("initialSnapshot", false, "before subscribing to metadata updates, walk the live filer tree under -filerPath and seed the destination. Only runs on a fresh sync (no prior checkpoint and -timeAgo is 0). After the walk, subscription starts from the walk-start timestamp so concurrent changes are still captured.") } var cmdFilerBackup = &Command{ @@ -70,6 +77,11 @@ var cmdFilerBackup = &Command{ If restarted and "-timeAgo" is not set, the synchronization will resume from the previous checkpoints, persisted every minute. A fresh sync will start from the earliest metadata logs. To reset the checkpoints, just set "-timeAgo" to a high value. + On a fresh sync the metadata event log only re-materializes files that still exist on the source; entries that were + created and later deleted are replayed as a create-then-delete pair and therefore never appear on the destination. + Pass "-initialSnapshot" to walk the live filer tree first and seed the destination with the current tree, then + subscribe from the walk-start timestamp. The walk only runs when there is no prior checkpoint. + `, } @@ -123,17 +135,27 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti // get start time for the data sink startFrom := time.Unix(0, 0) + isFreshSync := true sinkId := util.HashStringToLong(dataSink.GetName() + dataSink.GetSinkToDirectory()) if timeAgo.Milliseconds() == 0 { lastOffsetTsNs, err := getOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, int32(sinkId)) if err != nil { - glog.V(0).Infof("starting from %v", startFrom) - } else { + // A KV read failure is ambiguous — a checkpoint may well exist but the + // source filer is temporarily unreachable. Don't treat that as a fresh + // sync; otherwise runFilerBackup's retry loop would redo the full + // -initialSnapshot walk on every transient error. + isFreshSync = false + glog.V(0).Infof("starting from %v (offset read failed: %v)", startFrom, err) + } else if lastOffsetTsNs > 0 { startFrom = time.Unix(0, lastOffsetTsNs) + isFreshSync = false glog.V(0).Infof("resuming from %v", startFrom) + } else { + glog.V(0).Infof("starting from %v (no prior checkpoint)", startFrom) } } else { startFrom = time.Now().Add(-timeAgo) + isFreshSync = false glog.V(0).Infof("start time is set to %v", startFrom) } @@ -150,6 +172,28 @@ func doFilerBackup(grpcDialOption grpc.DialOption, backupOption *FilerBackupOpti } dataSink.SetSourceFiler(filerSource) + // When the destination has no prior checkpoint and the user opted in to an + // initial snapshot, walk the live filer tree first and seed the destination + // with the current entries. This avoids the "only new files appear" pitfall + // of replaying the metadata event log: entries created-then-deleted before + // the walk leave no trace, so a re-backup after wiping the destination + // reflects what is actually live on the source instead of an empty tree. + if *backupOption.initialSnapshot && isFreshSync { + snapshotTsNs, err := runInitialSnapshot(sourceFiler.ToGrpcAddress(), filerSource, sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.ignore404Error) + if err != nil { + return fmt.Errorf("initial snapshot: %w", err) + } + // The walk can take hours on large trees; retry the tiny KV write a + // handful of times before giving up so a flaky filer KV doesn't force + // the whole walk to repeat on the next retry loop iteration. + if err := persistSnapshotOffset(grpcDialOption, sourceFiler, int32(sinkId), snapshotTsNs); err != nil { + glog.Errorf("initialSnapshot: FAILED to persist offset %d for sinkId %d after retries: %v — the next retry will redo the full walk", snapshotTsNs, sinkId, err) + return fmt.Errorf("persist initial snapshot offset: %w", err) + } + startFrom = time.Unix(0, snapshotTsNs) + glog.V(0).Infof("initialSnapshot done; subscribing from %v", startFrom) + } + var processEventFn func(*filer_pb.SubscribeMetadataResponse) error if *backupOption.ignore404Error { processEventFnGenerated := genProcessFunction(sourcePath, targetPath, excludePaths, reExcludeFileName, excludeFileNames, excludePathPatterns, dataSink, *backupOption.doDeleteFiles, debug) @@ -241,3 +285,151 @@ func isIgnorable404(err error) bool { strings.Contains(errStr, "LookupFileId") || (strings.Contains(errStr, "volume id") && strings.Contains(errStr, "not found")) } + +// persistSnapshotOffset writes the snapshot high-water mark to the source +// filer's KV, retrying a few times with exponential backoff on transient +// errors. Losing this write forces a full re-walk on the next retry loop +// iteration, so a small retry budget here is far cheaper than paying the +// walk cost again on a large tree. +func persistSnapshotOffset(grpcDialOption grpc.DialOption, sourceFiler pb.ServerAddress, sinkId int32, tsNs int64) error { + const attempts = 4 + backoff := 500 * time.Millisecond + var lastErr error + for i := 1; i <= attempts; i++ { + if err := setOffset(grpcDialOption, sourceFiler, BackupKeyPrefix, sinkId, tsNs); err == nil { + return nil + } else { + lastErr = err + if i == attempts { + break + } + glog.V(0).Infof("initialSnapshot: setOffset attempt %d/%d failed: %v (retrying in %v)", i, attempts, err, backoff) + time.Sleep(backoff) + backoff *= 2 + } + } + return lastErr +} + +// runInitialSnapshot walks the live filer tree under sourcePath and seeds the +// destination via the sink. The snapshot timestamp is captured before the walk +// starts so any create/update/delete that races with the walk is still caught +// by the subscription that runs afterward (sink CreateEntry is idempotent for +// all builtin sinks, so replaying a concurrent create from the subscription +// over an entry already written by the walk is safe). +// +// Note on excludes: TraverseBfs enumerates every directory and enqueues its +// children unconditionally before the callback runs, so the exclude filters +// below only prevent *processing* of excluded entries — they do not prune the +// listing RPCs for excluded subtrees. For small system excludes (SystemLogDir) +// that is fine; for large user-supplied excludes (say an archive directory +// with millions of files), the listing cost can dominate the snapshot. A +// proper prune signal through TraverseBfs is a separate change. +func runInitialSnapshot( + grpcAddress string, + filerSource *source.FilerSource, + sourcePath string, + targetPath string, + excludePaths []string, + reExcludeFileName *regexp.Regexp, + excludeFileNames []*wildcard.WildcardMatcher, + excludePathPatterns []*wildcard.WildcardMatcher, + dataSink sink.ReplicationSink, + ignore404Error bool, +) (int64, error) { + // Metadata events are stamped server-side, so the backup host clock may be + // ahead of the filer's. Take `now - 1min` as the subscription watermark so + // a fast client clock can't skip events that fire during/right-after the + // walk. meta_aggregator.go uses the same 1-minute margin on initial peer + // traversal; see the note there on why duplicate replay is harmless. + snapshotTsNs := time.Now().Add(-time.Minute).UnixNano() + glog.V(0).Infof("initialSnapshot: walking %s on %s -> %s (snapshotTsNs=%d, -1m skew margin applied)", sourcePath, grpcAddress, targetPath, snapshotTsNs) + + // TraverseBfs fans the callback out across 5 worker goroutines, so counter + // updates and the progress-log gate need to be safe under concurrent access. + var entryCount, byteCount atomic.Int64 + start := time.Now() + var logMu sync.Mutex + lastLog := start + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + err := filer_pb.TraverseBfs(ctx, filerSource, util.FullPath(sourcePath), func(parentPath util.FullPath, entry *filer_pb.Entry) error { + parent := string(parentPath) + sourceKey := parentPath.Child(entry.Name) + source := string(sourceKey) + // Check exclusion on the entry's own path too — otherwise a walked + // directory whose full path is SystemLogDir or a user-excluded path + // still gets seeded (only its children would be skipped by the parent + // check). + if parent == filer.SystemLogDir || strings.HasPrefix(parent, filer.SystemLogDir+"/") || + source == filer.SystemLogDir || strings.HasPrefix(source, filer.SystemLogDir+"/") { + return nil + } + if matchesExcludePath(parent, excludePaths) || matchesExcludePath(source, excludePaths) { + return nil + } + if isEntryExcluded(parent, entry, reExcludeFileName, excludeFileNames, excludePathPatterns) { + return nil + } + + if !util.IsEqualOrUnder(source, sourcePath) { + return nil + } + + targetKey := initialSnapshotTargetKey(dataSink, targetPath, sourcePath, sourceKey, entry) + if err := dataSink.CreateEntry(targetKey, entry, nil); err != nil { + // A file can be listed by TraverseBfs and then deleted before + // CreateEntry reads its chunks. The follow phase ignores 404s when + // -ignore404Error is set; apply the same policy here so the walk + // doesn't abort (and trigger a full re-walk on retry) just because + // a single entry disappeared mid-snapshot. + if ignore404Error && isIgnorable404(err) { + glog.V(0).Infof("initialSnapshot: source entry %s disappeared, ignore it: %s", sourceKey, err.Error()) + return nil + } + return fmt.Errorf("seed %s: %w", targetKey, err) + } + + curEntries := entryCount.Add(1) + var curBytes int64 + if entry.Attributes != nil { + curBytes = byteCount.Add(int64(entry.Attributes.FileSize)) + } else { + curBytes = byteCount.Load() + } + + now := time.Now() + logMu.Lock() + shouldLog := now.Sub(lastLog) >= 5*time.Second + if shouldLog { + lastLog = now + } + logMu.Unlock() + if shouldLog { + elapsed := now.Sub(start).Seconds() + if elapsed == 0 { + elapsed = 1 + } + glog.V(0).Infof("initialSnapshot: %d entries / %d bytes seeded (%.1f/sec)", curEntries, curBytes, float64(curEntries)/elapsed) + } + return nil + }) + if err != nil { + return 0, err + } + glog.V(0).Infof("initialSnapshot: done — %d entries / %d bytes in %v", entryCount.Load(), byteCount.Load(), time.Since(start)) + return snapshotTsNs, nil +} + +// initialSnapshotTargetKey pulls the entry mtime and delegates to the shared +// destKey helper in filer_sync.go so the walk and the event-log path produce +// the same destination key for the same source entry. +func initialSnapshotTargetKey(dataSink sink.ReplicationSink, targetPath, sourcePath string, sourceKey util.FullPath, entry *filer_pb.Entry) string { + var mTime int64 + if entry.Attributes != nil { + mTime = entry.Attributes.Mtime + } + return destKey(dataSink, targetPath, sourcePath, sourceKey, mTime) +} diff --git a/weed/command/filer_backup_test.go b/weed/command/filer_backup_test.go index b22706d0d..52031f7d0 100644 --- a/weed/command/filer_backup_test.go +++ b/weed/command/filer_backup_test.go @@ -7,7 +7,12 @@ import ( "net/http/httptest" "os" "testing" + "time" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/replication/sink" + "github.com/seaweedfs/seaweedfs/weed/replication/source" + "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" ) @@ -71,3 +76,66 @@ func TestIsIgnorable404_NonIgnorableError(t *testing.T) { t.Errorf("expected not ignorable, got ignorable: %v", genErr) } } + +// stubSink is a minimal ReplicationSink used to exercise initialSnapshotTargetKey +// without standing up a real sink. Only the two methods read by the key builder +// (GetName, IsIncremental) need meaningful behavior; the rest satisfy the interface. +type stubSink struct { + name string + isIncremental bool +} + +func (s *stubSink) GetName() string { return s.name } +func (s *stubSink) Initialize(util.Configuration, string) error { return nil } +func (s *stubSink) DeleteEntry(string, bool, bool, []int32) error { + return nil +} +func (s *stubSink) CreateEntry(string, *filer_pb.Entry, []int32) error { return nil } +func (s *stubSink) UpdateEntry(string, *filer_pb.Entry, string, *filer_pb.Entry, bool, []int32) (bool, error) { + return false, nil +} +func (s *stubSink) GetSinkToDirectory() string { return "" } +func (s *stubSink) SetSourceFiler(*source.FilerSource) {} +func (s *stubSink) IsIncremental() bool { return s.isIncremental } + +var _ sink.ReplicationSink = (*stubSink)(nil) + +func TestInitialSnapshotTargetKey(t *testing.T) { + // Mirror the non-incremental path of buildKey so a refactor of one without + // the other will fail this test. + mirror := &stubSink{name: "mirror", isIncremental: false} + got := initialSnapshotTargetKey(mirror, "/backup", "/data", util.FullPath("/data/sub/file.txt"), &filer_pb.Entry{}) + if got != "/backup/sub/file.txt" { + t.Errorf("mirror sink: got %q, want %q", got, "/backup/sub/file.txt") + } + + // Incremental sinks partition by entry mtime, so the seed must use the same + // YYYY-MM-DD prefix a replayed CreateEntry would produce. buildKey in + // filer_sync.go formats the date in local time, so compute the expected + // key the same way to keep the test timezone-independent. + inc := &stubSink{name: "inc", isIncremental: true} + mtime := int64(1704196800) // 2024-01-02T12:00:00 UTC — unambiguously Jan 2 in nearly all timezones + gotInc := initialSnapshotTargetKey(inc, "/backup", "/data", util.FullPath("/data/sub/file.txt"), &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{Mtime: mtime}, + }) + wantInc := "/backup/" + time.Unix(mtime, 0).Format("2006-01-02") + "/sub/file.txt" + if gotInc != wantInc { + t.Errorf("incremental sink: got %q, want %q", gotInc, wantInc) + } + + // Trailing-slash sourcePath still produces a clean relative key. + gotTrail := initialSnapshotTargetKey(mirror, "/backup", "/data/", util.FullPath("/data/file.txt"), &filer_pb.Entry{}) + if gotTrail != "/backup/file.txt" { + t.Errorf("trailing-slash sourcePath: got %q, want %q", gotTrail, "/backup/file.txt") + } + + // Edge cases CodeRabbit called out: sourceKey equal to sourcePath + // (non-trailing and trailing variants). Real TraverseBfs walks never emit + // the root itself, but the helper must not panic if something else does. + if got := initialSnapshotTargetKey(mirror, "/backup", "/data", util.FullPath("/data"), &filer_pb.Entry{}); got != "/backup" { + t.Errorf("sourceKey == sourcePath (no slash): got %q, want %q", got, "/backup") + } + if got := initialSnapshotTargetKey(mirror, "/backup", "/data/", util.FullPath("/data"), &filer_pb.Entry{}); got != "/backup" { + t.Errorf("sourceKey == sourcePath (trailing slash mismatch): got %q, want %q", got, "/backup") + } +} diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go index 9452f8f80..8463ce0ac 100644 --- a/weed/command/filer_sync.go +++ b/weed/command/filer_sync.go @@ -653,21 +653,39 @@ func genProcessFunction(sourcePath string, targetPath string, excludePaths []str return processEventFn } -func buildKey(dataSink sink.ReplicationSink, message *filer_pb.EventNotification, targetPath string, sourceKey util.FullPath, sourcePath string) (key string) { - if !dataSink.IsIncremental() { - key = util.Join(targetPath, string(sourceKey)[len(sourcePath):]) - } else { - var mTime int64 - if message.NewEntry != nil { - mTime = message.NewEntry.Attributes.Mtime - } else if message.OldEntry != nil { - mTime = message.OldEntry.Attributes.Mtime - } - dateKey := time.Unix(mTime, 0).Format("2006-01-02") - key = util.Join(targetPath, dateKey, string(sourceKey)[len(sourcePath):]) +func buildKey(dataSink sink.ReplicationSink, message *filer_pb.EventNotification, targetPath string, sourceKey util.FullPath, sourcePath string) string { + var mTime int64 + if message.NewEntry != nil && message.NewEntry.Attributes != nil { + mTime = message.NewEntry.Attributes.Mtime + } else if message.OldEntry != nil && message.OldEntry.Attributes != nil { + mTime = message.OldEntry.Attributes.Mtime } + return destKey(dataSink, targetPath, sourcePath, sourceKey, mTime) +} - return escapeKey(key) +// destKey derives the sink-side key for a source entry. Shared between the +// event-log path (buildKey) and the initialSnapshot walk (both paths need the +// same target layout so a walk-seeded file and an event-replayed file resolve +// to the same destination key). Normalizing to a trailing-slash base avoids +// indexing past the end of sourceKey when callers differ on trailing-slash +// conventions or when sourceKey equals sourcePath exactly. +func destKey(dataSink sink.ReplicationSink, targetPath, sourcePath string, sourceKey util.FullPath, mTime int64) string { + base := strings.TrimSuffix(sourcePath, "/") + "/" + sk := string(sourceKey) + var relative string + switch { + case strings.HasPrefix(sk, base): + relative = sk[len(base):] + case sk == strings.TrimSuffix(sourcePath, "/"): + relative = "" + default: + relative = strings.TrimPrefix(sk, "/") + } + if !dataSink.IsIncremental() { + return escapeKey(util.Join(targetPath, relative)) + } + dateKey := time.Unix(mTime, 0).Format("2006-01-02") + return escapeKey(util.Join(targetPath, dateKey, relative)) } // isEntryExcluded checks whether a single side (old or new) of an event is excluded