From 8c22336944fbcc6a8f4c6c40f6531c0c5de3932a Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 6 Nov 2026 13:32:46 -0800 Subject: [PATCH] filer.meta.backup: stream from the start of an unfinished walk (#12458) * filer.meta.backup: stream from the start of an unfinished walk A walk only inserts entries, so when the first walk is interrupted the next start walks again and streams from the second walk's start: deletes and renames between the two walks never reach the backup store. Keep the first walk's start time in the store until a walk completes and stream from it. Co-Authored-By: Claude Opus 5.5 * filer.meta.backup: start fresh once the stream has passed an earlier walk's start A walk start left in the store by a crash between saving the offset and clearing it, or by an interrupted -restart followed by a plain restart, stops mattering once the stream saves an offset past it. Ignore it then instead of replaying from it on the next walk. Co-Authored-By: Claude Opus 5.5 * filer.meta.backup: fail the walk on an unreadable offset Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- weed/command/filer_meta_backup.go | 83 +++++++++++++---- weed/command/filer_meta_backup_test.go | 120 +++++++++++++++++++++++++ 2 files changed, 184 insertions(+), 19 deletions(-) create mode 100644 weed/command/filer_meta_backup_test.go diff --git a/weed/command/filer_meta_backup.go b/weed/command/filer_meta_backup.go index 33947b3d2a..ff5596fa7c 100644 --- a/weed/command/filer_meta_backup.go +++ b/weed/command/filer_meta_backup.go @@ -2,6 +2,7 @@ package command import ( "context" + "errors" "fmt" "reflect" "strings" @@ -109,17 +110,10 @@ func runFilerMetaBackup(cmd *Command, args []string) bool { } if *metaBackup.restart || missingPreviousBackup { - glog.V(0).Infof("traversing metadata tree...") - startTime := time.Now() - if err := metaBackup.traverseMetadata(); err != nil { - glog.Errorf("traverse meta data: %v", err) + if err := metaBackup.copyMetadata(metaBackup.traverseMetadata); err != nil { + glog.Exitf("%v", err) return true } - if err := metaBackup.setOffset(startTime); err != nil { - glog.Exitf("set metadata backup offset: %v", err) - return true - } - glog.V(0).Infof("metadata copied up to %v", startTime) } for { @@ -162,6 +156,52 @@ func (metaBackup *FilerMetaBackupOptions) shouldInclude(fullpath string) bool { return metaBackup.pathFilter.ShouldInclude(fullpath) } +// copyMetadata walks the tree, then saves the offset the stream resumes from. +// A walk only inserts, so the stream has to replay every delete and rename +// since the first walk that did not finish: its start time stays in the store +// until a walk completes. +func (metaBackup *FilerMetaBackupOptions) copyMetadata(traverse func() error) error { + startTime, err := metaBackup.walkStartTime() + if err != nil { + return err + } + glog.V(0).Infof("traversing metadata tree...") + if err := traverse(); err != nil { + return fmt.Errorf("traverse meta data: %w", err) + } + if err := metaBackup.setOffset(startTime); err != nil { + return fmt.Errorf("set metadata backup offset: %w", err) + } + if err := metaBackup.store.KvDelete(context.Background(), MetaBackupWalkKey); err != nil { + glog.Warningf("clear metadata walk start: %v", err) + } + glog.V(0).Infof("metadata copied up to %v", startTime) + return nil +} + +// walkStartTime is the start of an unfinished earlier walk, unless the stream +// has saved an offset past it since, or else now, saved as this walk's start. +func (metaBackup *FilerMetaBackupOptions) walkStartTime() (time.Time, error) { + earlier, err := metaBackup.getTime(MetaBackupWalkKey) + if err == nil { + offset, offsetErr := metaBackup.getOffset() + if offsetErr != nil && !errors.Is(offsetErr, filer.ErrKvNotFound) { + return time.Time{}, fmt.Errorf("read metadata backup offset: %w", offsetErr) + } + if offsetErr != nil || !offset.After(earlier) { + glog.V(0).Infof("an earlier metadata walk started at %v did not finish; streaming from then", earlier) + return earlier, nil + } + } else if !errors.Is(err, filer.ErrKvNotFound) { + return time.Time{}, fmt.Errorf("read metadata walk start: %w", err) + } + now := time.Now() + if err := metaBackup.setTime(MetaBackupWalkKey, now); err != nil { + return time.Time{}, fmt.Errorf("save metadata walk start: %w", err) + } + return now, nil +} + func (metaBackup *FilerMetaBackupOptions) traverseMetadata() (err error) { return filer_pb.TraverseBfs(context.Background(), metaBackup, util.FullPath(*metaBackup.filerDirectory), func(parentPath util.FullPath, entry *filer_pb.Entry) error { fullpath := string(parentPath.Child(entry.Name)) @@ -178,7 +218,8 @@ func (metaBackup *FilerMetaBackupOptions) traverseMetadata() (err error) { } var ( - MetaBackupKey = []byte("metaBackup") + MetaBackupKey = []byte("metaBackup") + MetaBackupWalkKey = []byte("metaBackupWalk") ) func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup(startTime time.Time) error { @@ -274,23 +315,27 @@ func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup(startTime time.Ti } func (metaBackup *FilerMetaBackupOptions) getOffset() (lastWriteTime time.Time, err error) { - value, err := metaBackup.store.KvGet(context.Background(), MetaBackupKey) + return metaBackup.getTime(MetaBackupKey) +} + +func (metaBackup *FilerMetaBackupOptions) setOffset(lastWriteTime time.Time) error { + return metaBackup.setTime(MetaBackupKey, lastWriteTime) +} + +func (metaBackup *FilerMetaBackupOptions) getTime(key []byte) (time.Time, error) { + value, err := metaBackup.store.KvGet(context.Background(), key) if err != nil { - return + return time.Time{}, err } tsNs := util.BytesToUint64(value) return time.Unix(0, int64(tsNs)), nil } -func (metaBackup *FilerMetaBackupOptions) setOffset(lastWriteTime time.Time) error { +func (metaBackup *FilerMetaBackupOptions) setTime(key []byte, t time.Time) error { valueBuf := make([]byte, 8) - util.Uint64toBytes(valueBuf, uint64(lastWriteTime.UnixNano())) - - if err := metaBackup.store.KvPut(context.Background(), MetaBackupKey, valueBuf); err != nil { - return err - } - return nil + util.Uint64toBytes(valueBuf, uint64(t.UnixNano())) + return metaBackup.store.KvPut(context.Background(), key, valueBuf) } var _ = filer_pb.FilerClient(&FilerMetaBackupOptions{}) diff --git a/weed/command/filer_meta_backup_test.go b/weed/command/filer_meta_backup_test.go new file mode 100644 index 0000000000..35a2d8f8f9 --- /dev/null +++ b/weed/command/filer_meta_backup_test.go @@ -0,0 +1,120 @@ +package command + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" +) + +type kvOnlyStore struct { + filer.FilerStore + kv map[string][]byte + getErr error +} + +func (s *kvOnlyStore) KvGet(_ context.Context, key []byte) ([]byte, error) { + if s.getErr != nil && string(key) == string(MetaBackupKey) { + return nil, s.getErr + } + value, found := s.kv[string(key)] + if !found { + return nil, filer.ErrKvNotFound + } + return value, nil +} + +func (s *kvOnlyStore) KvPut(_ context.Context, key []byte, value []byte) error { + s.kv[string(key)] = value + return nil +} + +func (s *kvOnlyStore) KvDelete(_ context.Context, key []byte) error { + delete(s.kv, string(key)) + return nil +} + +// TestCopyMetadataStreamsFromAnUnfinishedWalk: a walk only inserts, so a walk +// that restarts after an interrupted one must stream from the first walk's +// start, or the deletes between the two walks never reach the backup. +func TestCopyMetadataStreamsFromAnUnfinishedWalk(t *testing.T) { + backup := &FilerMetaBackupOptions{store: &kvOnlyStore{kv: map[string][]byte{}}} + + beforeFirst := time.Now() + interrupted := errors.New("killed") + if err := backup.copyMetadata(func() error { return interrupted }); !errors.Is(err, interrupted) { + t.Fatalf("interrupted walk: err = %v, want %v", err, interrupted) + } + if _, err := backup.getOffset(); !errors.Is(err, filer.ErrKvNotFound) { + t.Fatalf("offset after an interrupted walk: err = %v, want none saved", err) + } + firstStart, err := backup.getTime(MetaBackupWalkKey) + if err != nil || firstStart.Before(beforeFirst) { + t.Fatalf("walk start = %v, %v; want the first walk's start", firstStart, err) + } + + time.Sleep(time.Millisecond) + if err := backup.copyMetadata(func() error { return interrupted }); err == nil { + t.Fatal("second interrupted walk: err = nil") + } + if got, _ := backup.getTime(MetaBackupWalkKey); !got.Equal(firstStart) { + t.Fatalf("walk start after a second interruption = %v, want the first walk's %v", got, firstStart) + } + + if err := backup.copyMetadata(func() error { return nil }); err != nil { + t.Fatalf("finished walk: %v", err) + } + offset, err := backup.getOffset() + if err != nil || !offset.Equal(firstStart) { + t.Fatalf("offset = %v, %v; want the first walk's start %v", offset, err, firstStart) + } + if _, err := backup.getTime(MetaBackupWalkKey); !errors.Is(err, filer.ErrKvNotFound) { + t.Fatalf("walk start after a finished walk: err = %v, want it cleared", err) + } + + beforeRestart := time.Now() + if err := backup.copyMetadata(func() error { return nil }); err != nil { + t.Fatalf("walk after a finished one: %v", err) + } + if offset, _ := backup.getOffset(); offset.Before(beforeRestart) { + t.Fatalf("offset = %v, want a fresh walk's start after %v", offset, beforeRestart) + } +} + +// TestCopyMetadataIgnoresAWalkStartTheStreamPassed: a walk start left behind +// (a crash between saving the offset and clearing it, or an interrupted +// -restart followed by a plain restart) only matters until the stream saves an +// offset past it; after that a new walk starts fresh. +func TestCopyMetadataIgnoresAWalkStartTheStreamPassed(t *testing.T) { + backup := &FilerMetaBackupOptions{store: &kvOnlyStore{kv: map[string][]byte{}}} + walkStart := time.Now().Add(-time.Hour) + if err := backup.setTime(MetaBackupWalkKey, walkStart); err != nil { + t.Fatal(err) + } + + if err := backup.setOffset(walkStart); err != nil { + t.Fatal(err) + } + if got, err := backup.walkStartTime(); err != nil || !got.Equal(walkStart) { + t.Fatalf("walk start with the offset at it = %v, %v; want %v", got, err, walkStart) + } + + if err := backup.setOffset(walkStart.Add(time.Minute)); err != nil { + t.Fatal(err) + } + store := backup.store.(*kvOnlyStore) + store.getErr = errors.New("connection reset") + if _, err := backup.walkStartTime(); !errors.Is(err, store.getErr) { + t.Fatalf("walk start with an unreadable offset: err = %v, want the read error", err) + } + store.getErr = nil + beforeWalk := time.Now() + if err := backup.copyMetadata(func() error { return nil }); err != nil { + t.Fatal(err) + } + if offset, _ := backup.getOffset(); offset.Before(beforeWalk) { + t.Fatalf("offset = %v, want a fresh walk's start after %v", offset, beforeWalk) + } +}