diff --git a/weed/command/filer_meta_backup.go b/weed/command/filer_meta_backup.go index ff5596fa7c..9691ea755c 100644 --- a/weed/command/filer_meta_backup.go +++ b/weed/command/filer_meta_backup.go @@ -11,6 +11,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/spf13/viper" + "golang.org/x/time/rate" "google.golang.org/grpc" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -31,6 +32,7 @@ type FilerMetaBackupOptions struct { excludePrefixes *string restart *bool backupFilerConfig *string + entriesPerSecond *int pathFilter *util.PathPrefixFilter store filer.FilerStore @@ -46,11 +48,12 @@ func init() { metaBackup.excludePrefixes = cmdFilerMetaBackup.Flag.String("excludePrefixes", "", "comma-separated path prefixes to exclude from backup") metaBackup.restart = cmdFilerMetaBackup.Flag.Bool("restart", false, "copy the full metadata before async incremental backup") metaBackup.backupFilerConfig = cmdFilerMetaBackup.Flag.String("config", "", "path to filer.toml specifying backup filer store") + metaBackup.entriesPerSecond = cmdFilerMetaBackup.Flag.Int("entriesPerSecond", 0, "limit how many entries per second the full metadata copy reads from the filer, 0 means unlimited") metaBackup.clientId = util.RandomInt32() } var cmdFilerMetaBackup = &Command{ - UsageLine: "filer.meta.backup [-filer=localhost:8888] [-filerDir=/] [-includePrefixes=...] [-excludePrefixes=...] [-restart] -config=/path/to/backup_filer.toml", + UsageLine: "filer.meta.backup [-filer=localhost:8888] [-filerDir=/] [-includePrefixes=...] [-excludePrefixes=...] [-restart] [-entriesPerSecond=0] -config=/path/to/backup_filer.toml", Short: "continuously backup filer meta data changes to anther filer store specified in a backup_filer.toml", Long: `continuously backup filer meta data changes. The backup writes to another filer store specified in a backup_filer.toml. @@ -61,6 +64,9 @@ The backup writes to another filer store specified in a backup_filer.toml. The -includePrefixes and -excludePrefixes flags accept comma-separated path prefixes. Paths must be absolute (start with '/'). Matching is at directory boundaries. When both match, the deeper prefix wins. + +An interrupted metadata copy continues from where it stopped. Use +-entriesPerSecond to lighten the load the copy puts on the filer. `, } @@ -175,6 +181,9 @@ func (metaBackup *FilerMetaBackupOptions) copyMetadata(traverse func() error) er if err := metaBackup.store.KvDelete(context.Background(), MetaBackupWalkKey); err != nil { glog.Warningf("clear metadata walk start: %v", err) } + if err := metaBackup.store.KvDelete(context.Background(), MetaBackupWalkPathKey); err != nil { + glog.Warningf("clear metadata walk progress: %v", err) + } glog.V(0).Infof("metadata copied up to %v", startTime) return nil } @@ -195,6 +204,9 @@ func (metaBackup *FilerMetaBackupOptions) walkStartTime() (time.Time, error) { } else if !errors.Is(err, filer.ErrKvNotFound) { return time.Time{}, fmt.Errorf("read metadata walk start: %w", err) } + if err := metaBackup.store.KvDelete(context.Background(), MetaBackupWalkPathKey); err != nil && !errors.Is(err, filer.ErrKvNotFound) { + return time.Time{}, fmt.Errorf("clear metadata walk progress: %w", err) + } now := time.Now() if err := metaBackup.setTime(MetaBackupWalkKey, now); err != nil { return time.Time{}, fmt.Errorf("save metadata walk start: %w", err) @@ -202,24 +214,47 @@ func (metaBackup *FilerMetaBackupOptions) walkStartTime() (time.Time, error) { return now, nil } +// traverseMetadata continues from the last path an unfinished walk saved. 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 { + ctx := context.Background() + resumeAfter, err := metaBackup.store.KvGet(ctx, MetaBackupWalkPathKey) + if err != nil && !errors.Is(err, filer.ErrKvNotFound) { + return fmt.Errorf("read metadata walk progress: %w", err) + } + limiter := rate.NewLimiter(rate.Inf, 0) + if *metaBackup.entriesPerSecond > 0 { + limiter = rate.NewLimiter(rate.Limit(*metaBackup.entriesPerSecond), *metaBackup.entriesPerSecond) + } + lastSaved := time.Now() + return filer_pb.TraverseDfs(ctx, metaBackup, util.FullPath(*metaBackup.filerDirectory), util.FullPath(resumeAfter), func(parentPath util.FullPath, entry *filer_pb.Entry) error { + if err := limiter.Wait(ctx); err != nil { + return err + } fullpath := string(parentPath.Child(entry.Name)) if !metaBackup.shouldInclude(fullpath) { return nil } println("+", fullpath) - if err := metaBackup.store.InsertEntry(context.Background(), filer.FromPbEntry(string(parentPath), entry)); err != nil { + if err := metaBackup.store.InsertEntry(ctx, filer.FromPbEntry(string(parentPath), entry)); err != nil { return fmt.Errorf("insert entry error: %w", err) } return nil + }, func(walked util.FullPath) { + if time.Since(lastSaved) < 3*time.Second { + return + } + lastSaved = time.Now() + if err := metaBackup.store.KvPut(ctx, MetaBackupWalkPathKey, []byte(walked)); err != nil { + glog.Warningf("save metadata walk progress: %v", err) + } }) } var ( - MetaBackupKey = []byte("metaBackup") - MetaBackupWalkKey = []byte("metaBackupWalk") + MetaBackupKey = []byte("metaBackup") + MetaBackupWalkKey = []byte("metaBackupWalk") + MetaBackupWalkPathKey = []byte("metaBackupWalkPath") ) func (metaBackup *FilerMetaBackupOptions) streamMetadataBackup(startTime time.Time) error { diff --git a/weed/command/filer_meta_backup_test.go b/weed/command/filer_meta_backup_test.go index 35a2d8f8f9..ed53401c62 100644 --- a/weed/command/filer_meta_backup_test.go +++ b/weed/command/filer_meta_backup_test.go @@ -118,3 +118,52 @@ func TestCopyMetadataIgnoresAWalkStartTheStreamPassed(t *testing.T) { t.Fatalf("offset = %v, want a fresh walk's start after %v", offset, beforeWalk) } } + +// TestCopyMetadataResumesOnlyItsOwnWalk: the saved walk path belongs to the +// walk whose start the stream replays from; a fresh walk must not skip ahead. +func TestCopyMetadataResumesOnlyItsOwnWalk(t *testing.T) { + backup := &FilerMetaBackupOptions{store: &kvOnlyStore{kv: map[string][]byte{}}} + walkedTo := func() string { + value, err := backup.store.KvGet(context.Background(), MetaBackupWalkPathKey) + if errors.Is(err, filer.ErrKvNotFound) { + return "" + } + if err != nil { + t.Fatal(err) + } + return string(value) + } + + interrupted := errors.New("killed") + if err := backup.copyMetadata(func() error { + return errors.Join(backup.store.KvPut(context.Background(), MetaBackupWalkPathKey, []byte("/a/b")), interrupted) + }); !errors.Is(err, interrupted) { + t.Fatalf("interrupted walk: err = %v", err) + } + if err := backup.copyMetadata(func() error { + if got := walkedTo(); got != "/a/b" { + t.Fatalf("resumed walk sees %q, want /a/b", got) + } + return nil + }); err != nil { + t.Fatal(err) + } + if got := walkedTo(); got != "" { + t.Fatalf("walk path after a finished walk = %q, want it cleared", got) + } + + if err := backup.setTime(MetaBackupWalkKey, time.Now().Add(-time.Hour)); err != nil { + t.Fatal(err) + } + if err := backup.store.KvPut(context.Background(), MetaBackupWalkPathKey, []byte("/x")); err != nil { + t.Fatal(err) + } + if err := backup.copyMetadata(func() error { + if got := walkedTo(); got != "" { + t.Fatalf("fresh walk sees %q left by a walk the stream passed", got) + } + return nil + }); err != nil { + t.Fatal(err) + } +} diff --git a/weed/pb/filer_pb/filer_client_dfs.go b/weed/pb/filer_pb/filer_client_dfs.go new file mode 100644 index 0000000000..af7a597258 --- /dev/null +++ b/weed/pb/filer_pb/filer_client_dfs.go @@ -0,0 +1,124 @@ +package filer_pb + +import ( + "context" + "fmt" + "slices" + "strings" + "sync" + + "golang.org/x/sync/errgroup" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +var traverseDfsPageSize uint32 = 1024 + +// TraverseDfs walks the tree under parentPath depth first in listing order, so +// one path tells how far it got: every entry before it was visited, and so was +// the subtree of a directory it names. progress gets these paths in walk +// order, and a walk given one as resumeAfter skips everything up to it. fn may +// run concurrently for the files of one directory. +func TraverseDfs(ctx context.Context, filerClient FilerClient, parentPath, resumeAfter util.FullPath, fn func(parentPath util.FullPath, entry *Entry) error, progress func(walked util.FullPath)) error { + parentPath = util.NormalizePath(string(parentPath)) + var resume []string + if resumeAfter != "" && resumeAfter != parentPath && resumeAfter.IsUnder(parentPath) { + glog.V(0).Infof("traverse %s after %s", parentPath, resumeAfter) + resume = util.FullPath(strings.TrimPrefix(string(resumeAfter), strings.TrimSuffix(string(parentPath), "/"))).Split() + } + return traverseDfsDirectory(ctx, filerClient, parentPath, resume, fn, progress) +} + +func traverseDfsDirectory(ctx context.Context, filerClient FilerClient, dir util.FullPath, resume []string, fn func(parentPath util.FullPath, entry *Entry) error, progress func(walked util.FullPath)) error { + startFrom, inclusive := "", false + if len(resume) > 0 { + startFrom, inclusive = resume[0], true + } + for { + var page []*Entry + if err := List(ctx, filerClient, string(dir), "", func(entry *Entry, isLast bool) error { + page = append(page, entry) + return nil + }, startFrom, inclusive, traverseDfsPageSize); err != nil { + return err + } + + listed := page + // the entry the walk stopped at: done with its subtree when it ends the + // path, else a directory to resume inside + var resumeIn []string + if inclusive && len(page) > 0 && page[0].Name == startFrom { + if len(resume) == 1 { + page = page[1:] + } else if page[0].IsDirectory { + resumeIn = resume[1:] + } + } + + if resumeIn != nil { + child := dir.Child(page[0].Name) + if err := traverseDfsDirectory(ctx, filerClient, child, resumeIn, fn, progress); err != nil { + return err + } + progress(child) + page = page[1:] + } + for len(page) > 0 { + run := page + if i := slices.IndexFunc(page, func(entry *Entry) bool { return entry.IsDirectory }); i >= 0 { + run = page[:i+1] + } + if err := visitDfsRun(ctx, dir, run, fn, progress); err != nil { + return err + } + if last := run[len(run)-1]; last.IsDirectory { + child := dir.Child(last.Name) + if err := traverseDfsDirectory(ctx, filerClient, child, nil, fn, progress); err != nil { + return err + } + progress(child) + } + page = page[len(run):] + } + + if uint32(len(listed)) < traverseDfsPageSize { + return nil + } + last := listed[len(listed)-1].Name + if !inclusive && last == startFrom { + return fmt.Errorf("list %s: pagination stuck at %q", dir, startFrom) + } + startFrom, inclusive = last, false + } +} + +// visitDfsRun calls fn concurrently on entries that end at the first +// directory, and reports each file once every entry before it is done. The +// caller reports the directory after its subtree. +func visitDfsRun(ctx context.Context, dir util.FullPath, run []*Entry, fn func(parentPath util.FullPath, entry *Entry) error, progress func(walked util.FullPath)) error { + var mu sync.Mutex + done := make([]bool, len(run)) + next := 0 + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(5) + for i, entry := range run { + g.Go(func() error { + if err := gctx.Err(); err != nil { + return err + } + if err := fn(dir, entry); err != nil { + return err + } + mu.Lock() + defer mu.Unlock() + done[i] = true + for next < len(run) && done[next] && !run[next].IsDirectory { + progress(dir.Child(run[next].Name)) + next++ + } + return nil + }) + } + return g.Wait() +} diff --git a/weed/pb/filer_pb/filer_client_dfs_test.go b/weed/pb/filer_pb/filer_client_dfs_test.go new file mode 100644 index 0000000000..628de7fe69 --- /dev/null +++ b/weed/pb/filer_pb/filer_client_dfs_test.go @@ -0,0 +1,159 @@ +package filer_pb + +import ( + "context" + "fmt" + "reflect" + "slices" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func startTreeFiler(t *testing.T, tree map[string][]string) *stubFilerClient { + t.Helper() + server := &stubFiler{} + server.listings = func(req *ListEntriesRequest, send func(*Entry) error) error { + names := append([]string(nil), tree[req.Directory]...) + sort.Strings(names) + sent := uint32(0) + for _, name := range names { + if req.StartFromFileName != "" && (name < req.StartFromFileName || name == req.StartFromFileName && !req.InclusiveStartFrom) { + continue + } + if sent == req.Limit { + break + } + _, isDir := tree[string(util.FullPath(req.Directory).Child(name))] + if err := send(&Entry{Name: name, IsDirectory: isDir}); err != nil { + return err + } + sent++ + } + return nil + } + return startStubFiler(t, server) +} + +func traverseDfsForTest(t *testing.T, filerClient FilerClient, root, resumeAfter util.FullPath) (visited map[string]int, progressed []string) { + t.Helper() + var mu sync.Mutex + visited = make(map[string]int) + err := TraverseDfs(context.Background(), filerClient, root, resumeAfter, func(parentPath util.FullPath, entry *Entry) error { + mu.Lock() + visited[string(parentPath.Child(entry.Name))]++ + mu.Unlock() + return nil + }, func(walked util.FullPath) { + progressed = append(progressed, string(walked)) + }) + if err != nil { + t.Fatal(err) + } + return visited, progressed +} + +func TestTraverseDfsResumesAfterEachProgressPath(t *testing.T) { + saved := traverseDfsPageSize + traverseDfsPageSize = 2 + t.Cleanup(func() { traverseDfsPageSize = saved }) + + filerClient := startTreeFiler(t, map[string][]string{ + "/": {"a", "b", "c", "d", "e"}, + "/a": {"x", "y", "z"}, + "/a/y": {"1", "2", "3"}, + "/c": {}, + "/e": {"f"}, + }) + + visited, progressed := traverseDfsForTest(t, filerClient, "/", "") + want := []string{"/a/x", "/a/y/1", "/a/y/2", "/a/y/3", "/a/y", "/a/z", "/a", "/b", "/c", "/d", "/e/f", "/e"} + if !reflect.DeepEqual(progressed, want) { + t.Fatalf("progress %v, want %v", progressed, want) + } + if len(visited) != len(want) { + t.Fatalf("visited %v, want each of %v", visited, want) + } + for _, p := range want { + if visited[p] != 1 { + t.Fatalf("visited %s %d times, want once", p, visited[p]) + } + } + + for i, resumeAfter := range want { + visited, progressed := traverseDfsForTest(t, filerClient, "/", util.FullPath(resumeAfter)) + if !slices.Equal(progressed, want[i+1:]) { + t.Fatalf("after %s: progress %v, want %v", resumeAfter, progressed, want[i+1:]) + } + for _, p := range want[i+1:] { + wantVisits := 1 + if strings.HasPrefix(resumeAfter, p+"/") { + wantVisits = 0 + } + if visited[p] != wantVisits { + t.Fatalf("after %s: visited %s %d times, want %d", resumeAfter, p, visited[p], wantVisits) + } + delete(visited, p) + } + if len(visited) != 0 { + t.Fatalf("after %s: visited %v again", resumeAfter, visited) + } + } +} + +func TestTraverseDfsResumeAfterAMissingOrForeignPath(t *testing.T) { + filerClient := startTreeFiler(t, map[string][]string{ + "/": {"a", "b"}, + "/a": {"x", "y"}, + "/a/y": {"1", "2", "3"}, + }) + + _, progressed := traverseDfsForTest(t, filerClient, "/a/", "/a/y/15") + if want := []string{"/a/y/2", "/a/y/3", "/a/y"}; !reflect.DeepEqual(progressed, want) { + t.Fatalf("after a deleted path: progress %v, want %v", progressed, want) + } + + _, progressed = traverseDfsForTest(t, filerClient, "/a", "/b") + if want := []string{"/a/x", "/a/y/1", "/a/y/2", "/a/y/3", "/a/y"}; !reflect.DeepEqual(progressed, want) { + t.Fatalf("after a path outside the walk: progress %v, want %v", progressed, want) + } +} + +func TestTraverseDfsReportsFilesBeforeTheirPageFinishes(t *testing.T) { + filerClient := startTreeFiler(t, map[string][]string{ + "/": {"a", "b", "c", "d"}, + "/d": {"x"}, + }) + + var mu sync.Mutex + var progressed []string + reachedB := make(chan struct{}) + err := TraverseDfs(context.Background(), filerClient, "/", "", func(parentPath util.FullPath, entry *Entry) error { + if entry.Name != "c" { + return nil + } + select { + case <-reachedB: + return nil + case <-time.After(5 * time.Second): + return fmt.Errorf("no progress while %s was in flight", entry.Name) + } + }, func(walked util.FullPath) { + mu.Lock() + defer mu.Unlock() + progressed = append(progressed, string(walked)) + if walked == "/b" { + close(reachedB) + } + }) + if err != nil { + t.Fatal(err) + } + if want := []string{"/a", "/b", "/c", "/d/x", "/d"}; !reflect.DeepEqual(progressed, want) { + t.Fatalf("progress %v, want %v", progressed, want) + } +}