From 7f185767982016a6808d4857fe78e9795b8455df Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Wed, 11 Nov 2026 10:11:21 -0800 Subject: [PATCH] filer/etcd: serve volume.fsck from one etcd revision (#12555) * filer/etcd: scan entries from one revision volume.fsck walked etcd one directory at a time, so a file moved from a directory not yet listed into one already listed was seen under neither name and its chunks were reported as orphans. The etcd store now implements filer.EntryScanner: every page is read at the revision of the first read, and a compacted revision fails the scan instead of falling back to the walk. Co-Authored-By: Claude Opus 5.5 * filer/etcd: read hard links at the scan revision Hard-link records resolved during a scan were read at the current revision, so a link updated meanwhile mixed two instants. Reads made with the scan's context now use its revision. Co-Authored-By: Claude Opus 5.5 * shell: list etcd among the stores volume.fsck reads from a snapshot Co-Authored-By: Claude Opus 5.5 * filer/etcd: skip hard-link records in a scan Hard-link records live beside the entries under their random ids, so a root scan could read one as an entry with a path made of random bytes, which the traversal stream cannot send. Skip hard-link ids and keys whose directory is not valid UTF-8. Co-Authored-By: Claude Opus 5.5 * filer/etcd: recognize a hard-link record by the id it stores A hard-link id can be valid UTF-8, so the key shape alone let some records through a root scan as phantom files. The record's value is the shared entry, whose HardLinkId is the record's own key; no entry key can equal its own hard-link id. Co-Authored-By: Claude Opus 5.5 * filer/etcd: scan entries under an invalid-UTF-8 directory The filer sanitizes names but stores directories as given, so a path written over HTTP can leave entries the listing never reaches. Skipping them let volume.fsck count their chunks as orphans. Send them under the sanitized directory instead, the substitution the filer uses for names. Co-Authored-By: Claude Opus 5.5 * filer/etcd: fail a scan whose sanitized path names another entry An entry under an invalid-UTF-8 directory is reported under the sanitized directory. When that path is also another stored entry at the scan revision, or another sanitized entry in the same scan, acting on the reported path (volume.fsck deleting it) would reach the other file, so the scan fails and names the stored path instead. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- go.mod | 2 +- weed/filer/etcd/etcd_store.go | 2 +- weed/filer/etcd/etcd_store_kv.go | 2 +- weed/filer/etcd/etcd_store_scan.go | 131 +++++++++++ weed/filer/etcd/etcd_store_scan_test.go | 290 ++++++++++++++++++++++++ weed/shell/command_volume_fsck.go | 8 +- 6 files changed, 428 insertions(+), 7 deletions(-) create mode 100644 weed/filer/etcd/etcd_store_scan.go create mode 100644 weed/filer/etcd/etcd_store_scan_test.go diff --git a/go.mod b/go.mod index be14669c10..70888e645b 100644 --- a/go.mod +++ b/go.mod @@ -84,6 +84,7 @@ require ( github.com/xdg-go/scram v1.2.0 github.com/xdg-go/stringprep v1.0.4 // indirect github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect + go.etcd.io/etcd/api/v3 v3.7.1 go.etcd.io/etcd/client/v3 v3.7.1 go.mongodb.org/mongo-driver v1.17.9 go.opencensus.io v0.24.0 // indirect @@ -479,7 +480,6 @@ require ( github.com/zeebo/blake3 v0.2.4 // indirect github.com/zeebo/errs v1.4.0 // indirect go.etcd.io/bbolt v1.5.0 // indirect - go.etcd.io/etcd/api/v3 v3.7.1 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect go.opentelemetry.io/contrib/detectors/gcp v1.46.0 // indirect go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.71.0 // indirect diff --git a/weed/filer/etcd/etcd_store.go b/weed/filer/etcd/etcd_store.go index acbeb755a0..6cf7dd880d 100644 --- a/weed/filer/etcd/etcd_store.go +++ b/weed/filer/etcd/etcd_store.go @@ -138,7 +138,7 @@ func (store *EtcdStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (er func (store *EtcdStore) FindEntry(ctx context.Context, fullpath weed_util.FullPath) (entry *filer.Entry, err error) { key := genKey(fullpath.DirAndName()) - resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key)) + resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...) if err != nil { return nil, fmt.Errorf("get %s : %v", fullpath, err) } diff --git a/weed/filer/etcd/etcd_store_kv.go b/weed/filer/etcd/etcd_store_kv.go index 8a9ed68543..15bcb131c2 100644 --- a/weed/filer/etcd/etcd_store_kv.go +++ b/weed/filer/etcd/etcd_store_kv.go @@ -20,7 +20,7 @@ func (store *EtcdStore) KvPut(ctx context.Context, key []byte, value []byte) (er func (store *EtcdStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { - resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key)) + resp, err := store.client.Get(ctx, store.etcdKeyPrefix+string(key), atScanRevision(ctx)...) if err != nil { return nil, fmt.Errorf("kv get: %w", err) diff --git a/weed/filer/etcd/etcd_store_scan.go b/weed/filer/etcd/etcd_store_scan.go new file mode 100644 index 0000000000..a87ee9c392 --- /dev/null +++ b/weed/filer/etcd/etcd_store_scan.go @@ -0,0 +1,131 @@ +package etcd + +import ( + "bytes" + "context" + "errors" + "fmt" + + "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" + clientv3 "go.etcd.io/etcd/client/v3" + + "github.com/seaweedfs/seaweedfs/weed/filer" + weed_util "github.com/seaweedfs/seaweedfs/weed/util" +) + +var _ filer.EntryScanner = (*EtcdStore)(nil) + +var scanPageSize int64 = 1000 + +type scanRevisionKey struct{} + +// atScanRevision pins a read made with a ScanEntries context to the scan's +// revision, so hard-link records resolve as of the same instant. +func atScanRevision(ctx context.Context, opts ...clientv3.OpOption) []clientv3.OpOption { + if rev, ok := ctx.Value(scanRevisionKey{}).(int64); ok { + opts = append(opts, clientv3.WithRev(rev)) + } + return opts +} + +// ScanEntries implements filer.EntryScanner. Every page is read at the +// revision of the first read, so a rename committed meanwhile shows up under +// exactly one of its names. If etcd compacts that revision before the scan +// ends, the scan fails rather than mixing revisions. +func (store *EtcdStore) ScanEntries(ctx context.Context, dirPath weed_util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + return scanEntries(ctx, store.client, store.etcdKeyPrefix, dirPath, fn) +} + +func scanEntries(ctx context.Context, kv clientv3.KV, keyPrefix string, dirPath weed_util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + resp, err := kv.Get(ctx, keyPrefix+string(genKey(dirPath.DirAndName()))) + if err != nil { + return fmt.Errorf("scan %s: %v", dirPath, err) + } + rev := resp.Header.Revision + scanCtx := context.WithValue(ctx, scanRevisionKey{}, rev) + if dirPath != "/" && len(resp.Kvs) > 0 { + root := &filer.Entry{FullPath: dirPath} + if err := root.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(resp.Kvs[0].Value)); err != nil { + return fmt.Errorf("scan decode %s: %v", dirPath, err) + } + if err := fn(scanCtx, root); err != nil { + return err + } + } + + // Children of dirPath, then everything deeper; "/" covers both. + prefixes := []string{string(dirPath)} + if dirPath != "/" { + prefixes = []string{string(genDirectoryKeyPrefix(dirPath, "")), string(dirPath) + "/"} + } + sanitized := map[weed_util.FullPath]struct{}{} + for _, prefix := range prefixes { + if err := scanPrefix(scanCtx, kv, keyPrefix, prefix, dirPath, rev, sanitized, fn); err != nil { + return err + } + } + return nil +} + +func scanPrefix(ctx context.Context, kv clientv3.KV, keyPrefix, prefix string, dirPath weed_util.FullPath, rev int64, sanitized map[weed_util.FullPath]struct{}, fn func(ctx context.Context, entry *filer.Entry) error) error { + end := clientv3.GetPrefixRangeEnd(keyPrefix + prefix) + for start := keyPrefix + prefix; ; { + resp, err := kv.Get(ctx, start, clientv3.WithRange(end), clientv3.WithLimit(scanPageSize), clientv3.WithRev(rev)) + if errors.Is(err, rpctypes.ErrCompacted) { + return fmt.Errorf("scan %s: etcd compacted revision %d before the scan finished; rerun it, or keep more revisions (etcd --auto-compaction-retention): %w", dirPath, rev, err) + } + if err != nil { + return fmt.Errorf("scan %s: %w", dirPath, err) + } + for _, item := range resp.Kvs { + key := item.Key[len(keyPrefix):] + name := getNameFromKey(key) + if name == "" || len(name) == len(key) { + continue + } + // Names are sanitized when stored, directories are not; the stream + // needs valid UTF-8, and fsck must still count the chunks. + rawDir := string(key[:len(key)-len(name)-1]) + dir := weed_util.SanitizeUTF8Name(rawDir) + entry := &filer.Entry{FullPath: weed_util.NewFullPath(dir, name)} + if err := entry.DecodeAttributesAndChunks(weed_util.MaybeDecompressData(item.Value)); err != nil { + return fmt.Errorf("scan decode %s: %v", entry.FullPath, err) + } + // A hard-link record is stored under its own id (filer.NewHardLinkId). + if bytes.Equal(entry.HardLinkId, key) { + continue + } + if dir != rawDir { + if err := claimSanitizedPath(ctx, kv, keyPrefix, rev, entry.FullPath, weed_util.NewFullPath(rawDir, name), sanitized); err != nil { + return err + } + } + if err := fn(ctx, entry); err != nil { + return err + } + } + if !resp.More || len(resp.Kvs) == 0 { + return nil + } + start = string(resp.Kvs[len(resp.Kvs)-1].Key) + "\x00" + } +} + +// claimSanitizedPath fails the scan when the path an entry is reported under +// names a different entry, so acting on it (fsck deleting it) cannot reach +// another file. +func claimSanitizedPath(ctx context.Context, kv clientv3.KV, keyPrefix string, rev int64, p, rawPath weed_util.FullPath, sanitized map[weed_util.FullPath]struct{}) error { + _, taken := sanitized[p] + if !taken { + resp, err := kv.Get(ctx, keyPrefix+string(genKey(p.DirAndName())), clientv3.WithRev(rev), clientv3.WithKeysOnly()) + if err != nil { + return fmt.Errorf("scan lookup %s: %w", p, err) + } + taken = len(resp.Kvs) > 0 + } + if taken { + return fmt.Errorf("scan: entry %q has a directory that is not valid UTF-8 and would be reported as %s, which names another entry; rename or delete one of them", string(rawPath), p) + } + sanitized[p] = struct{}{} + return nil +} diff --git a/weed/filer/etcd/etcd_store_scan_test.go b/weed/filer/etcd/etcd_store_scan_test.go new file mode 100644 index 0000000000..3a7376bc00 --- /dev/null +++ b/weed/filer/etcd/etcd_store_scan_test.go @@ -0,0 +1,290 @@ +package etcd + +import ( + "context" + "errors" + "os" + "sort" + "testing" + + "go.etcd.io/etcd/api/v3/etcdserverpb" + "go.etcd.io/etcd/api/v3/mvccpb" + "go.etcd.io/etcd/api/v3/v3rpc/rpctypes" + clientv3 "go.etcd.io/etcd/client/v3" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +type fakeVersion struct { + rev int64 + value []byte // nil marks a delete +} + +// fakeMVCC keeps every revision of every key, as etcd does until compaction. +type fakeMVCC struct { + clientv3.KV + rev int64 + compacted int64 + keys map[string][]fakeVersion + gets int + beforeGet func(gets int) +} + +func newFakeMVCC() *fakeMVCC { + return &fakeMVCC{keys: map[string][]fakeVersion{}} +} + +func (m *fakeMVCC) put(key string, value []byte) { + m.rev++ + m.keys[key] = append(m.keys[key], fakeVersion{m.rev, value}) +} + +func (m *fakeMVCC) del(key string) { + m.rev++ + m.keys[key] = append(m.keys[key], fakeVersion{m.rev, nil}) +} + +func (m *fakeMVCC) Get(ctx context.Context, key string, opts ...clientv3.OpOption) (*clientv3.GetResponse, error) { + m.gets++ + if m.beforeGet != nil { + m.beforeGet(m.gets) + } + op := clientv3.OpGet(key, opts...) + rev := op.Rev() + if rev == 0 { + rev = m.rev + } + if rev < m.compacted { + return nil, rpctypes.ErrCompacted + } + end := string(op.RangeBytes()) + var names []string + for k := range m.keys { + if k == key || (end != "" && k > key && k < end) { + names = append(names, k) + } + } + sort.Strings(names) + resp := &clientv3.GetResponse{Header: &etcdserverpb.ResponseHeader{Revision: rev}} + for _, k := range names { + var value []byte + for _, v := range m.keys[k] { + if v.rev <= rev { + value = v.value + } + } + if value == nil { + continue + } + if limit := op.Limit(); limit > 0 && int64(len(resp.Kvs)) == limit { + resp.More = true + break + } + resp.Kvs = append(resp.Kvs, &mvccpb.KeyValue{Key: []byte(k), Value: value}) + } + return resp, nil +} + +const testKeyPrefix = "pfx" + +func (m *fakeMVCC) putEntry(t *testing.T, p util.FullPath, isDir bool) { + t.Helper() + entry := &filer.Entry{FullPath: p} + if isDir { + entry.Attr.Mode = os.ModeDir | 0755 + } else { + entry.Chunks = []*filer_pb.FileChunk{{FileId: "3,01" + p.Name(), Size: 1}} + } + meta, err := entry.EncodeAttributesAndChunks() + if err != nil { + t.Fatal(err) + } + m.put(entryKey(p), meta) +} + +func entryKey(p util.FullPath) string { + return testKeyPrefix + string(genKey(p.DirAndName())) +} + +func scanPaths(t *testing.T, m *fakeMVCC, dirPath util.FullPath) map[util.FullPath]int { + t.Helper() + seen := map[util.FullPath]int{} + if err := scanEntries(context.Background(), m, testKeyPrefix, dirPath, func(_ context.Context, entry *filer.Entry) error { + seen[entry.FullPath]++ + return nil + }); err != nil { + t.Fatalf("scan %s: %v", dirPath, err) + } + return seen +} + +func withScanPageSize(t *testing.T, n int64) { + prev := scanPageSize + scanPageSize = n + t.Cleanup(func() { scanPageSize = prev }) +} + +func TestScanEntriesSeesFileMovedDuringScan(t *testing.T) { + withScanPageSize(t, 2) + m := newFakeMVCC() + for _, d := range []util.FullPath{"/buckets", "/buckets/b", "/buckets/b/aaa", "/buckets/b/pad", "/buckets/b/zzz", "/buckets/b/zzz/late", "/buckets/bc"} { + m.putEntry(t, d, true) + } + for _, f := range []util.FullPath{"/buckets/b/pad/p0", "/buckets/b/pad/p1", "/buckets/b/pad/p2", "/buckets/b/zzz/late/f1", "/buckets/b/zzz/late/f2", "/buckets/bc/x"} { + m.putEntry(t, f, false) + } + m.put(testKeyPrefix+"/buckets/b/hardlink-without-separator", []byte("not an entry")) + + // Pages run root, children of /buckets/b, then deeper paths in key order: + // the move lands after aaa's subtree was read and before late's. + m.beforeGet = func(gets int) { + if gets == 5 { + from, to := util.FullPath("/buckets/b/zzz/late/f1"), util.FullPath("/buckets/b/aaa/f1") + value := m.keys[entryKey(from)][len(m.keys[entryKey(from)])-1].value + m.del(entryKey(from)) + m.put(entryKey(to), value) + } + } + + seen := scanPaths(t, m, "/buckets/b") + want := []util.FullPath{"/buckets/b", "/buckets/b/aaa", "/buckets/b/pad", "/buckets/b/zzz", "/buckets/b/zzz/late", + "/buckets/b/pad/p0", "/buckets/b/pad/p1", "/buckets/b/pad/p2", "/buckets/b/zzz/late/f1", "/buckets/b/zzz/late/f2"} + for _, p := range want { + if seen[p] != 1 { + t.Errorf("%s seen %d times, want once", p, seen[p]) + } + } + if len(seen) != len(want) { + t.Errorf("seen %v, want exactly %v", seen, want) + } + if m.gets < 5 { + t.Fatalf("the move never ran: %d reads", m.gets) + } + if _, found := m.keys[entryKey("/buckets/b/aaa/f1")]; !found { + t.Fatal("the move did not land") + } +} + +func TestScanEntriesFromRoot(t *testing.T) { + withScanPageSize(t, 3) + m := newFakeMVCC() + m.putEntry(t, "/a", true) + m.putEntry(t, "/a/x", false) + m.putEntry(t, "/top", false) + m.put(testKeyPrefix+"filer.store.id", []byte{1, 2, 3, 4}) + m.put(testKeyPrefix+"/\x00", []byte("root marker")) + for _, id := range []string{"/\x00\xff\xfe\x80abcdefghijk\x01", "/\x00abcdefghijklmn\x01"} { + link, err := (&filer.Entry{FullPath: "/top", HardLinkId: filer.HardLinkId(id), HardLinkCounter: 2}).EncodeAttributesAndChunks() + if err != nil { + t.Fatal(err) + } + m.put(testKeyPrefix+id, link) + } + m.putEntry(t, "/\xc3\x28xyz/name", false) + + seen := scanPaths(t, m, "/") + if len(seen) != 4 || seen["/a"] != 1 || seen["/a/x"] != 1 || seen["/top"] != 1 || seen["/_(xyz/name"] != 1 { + t.Fatalf("seen %v", seen) + } +} + +func TestScanEntriesReadsHardLinkAtScanRevision(t *testing.T) { + m := newFakeMVCC() + m.putEntry(t, "/d", true) + m.putEntry(t, "/d/f", false) + m.put(testKeyPrefix+"link", []byte("v1")) + m.beforeGet = func(gets int) { + if gets == 2 { + m.put(testKeyPrefix+"link", []byte("v2")) + } + } + var got []string + err := scanEntries(context.Background(), m, testKeyPrefix, "/d", func(ctx context.Context, entry *filer.Entry) error { + resp, err := m.Get(ctx, testKeyPrefix+"link", atScanRevision(ctx)...) + if err != nil { + return err + } + got = append(got, string(resp.Kvs[0].Value)) + return nil + }) + if err != nil { + t.Fatal(err) + } + for _, v := range got { + if v != "v1" { + t.Fatalf("hard link read %q inside the scan, want the scan revision's v1", got) + } + } + if len(got) != 2 { + t.Fatalf("visited %d entries, want 2", len(got)) + } +} + +func TestScanEntriesFailsWhenRevisionCompacted(t *testing.T) { + withScanPageSize(t, 1) + m := newFakeMVCC() + m.putEntry(t, "/d", true) + m.putEntry(t, "/d/a", false) + m.putEntry(t, "/d/b", false) + m.beforeGet = func(gets int) { + if gets == 3 { + m.putEntry(t, "/d/c", false) + m.compacted = m.rev + } + } + err := scanEntries(context.Background(), m, testKeyPrefix, "/d", func(context.Context, *filer.Entry) error { return nil }) + if !errors.Is(err, rpctypes.ErrCompacted) { + t.Fatalf("got %v, want a compaction error", err) + } + if errors.Is(err, filer.ErrUnsupportedEntryScan) { + t.Fatalf("a compacted scan must fail, not fall back to the walk: %v", err) + } +} + +func TestScanEntriesRejectsSanitizedPathOfAnotherEntry(t *testing.T) { + for _, tc := range []struct { + name string + paths []util.FullPath + lateDup bool + wantErr bool + }{ + {name: "free", paths: []util.FullPath{"/\xc3\x28xyz/name"}}, + {name: "stored entry", paths: []util.FullPath{"/\xc3\x28xyz/name", "/_(xyz/name"}, wantErr: true}, + {name: "two invalid directories", paths: []util.FullPath{"/a\xff/b", "/a\xfe/b"}, wantErr: true}, + {name: "created after the scan revision", paths: []util.FullPath{"/\xc3\x28xyz/name"}, lateDup: true}, + } { + t.Run(tc.name, func(t *testing.T) { + withScanPageSize(t, 1) + m := newFakeMVCC() + for _, p := range tc.paths { + m.putEntry(t, p, false) + } + if tc.lateDup { + m.beforeGet = func(gets int) { + if gets == 2 { + m.putEntry(t, "/_(xyz/name", false) + } + } + } + var seen []util.FullPath + err := scanEntries(context.Background(), m, testKeyPrefix, "/", func(_ context.Context, entry *filer.Entry) error { + seen = append(seen, entry.FullPath) + return nil + }) + if tc.wantErr { + if err == nil { + t.Fatalf("scan succeeded with %v, want a collision error", seen) + } + return + } + if err != nil { + t.Fatal(err) + } + if len(seen) != 1 || seen[0] != "/_(xyz/name" { + t.Fatalf("seen %v", seen) + } + }) + } +} diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index c826615997..20c15e6cd4 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -100,10 +100,10 @@ func (c *commandVolumeFsck) Help() string { The default 5h cutoff provides sufficient buffer for metadata commits. Step 2 reads the filer from one consistent snapshot when its store supports - that (the SQL stores do), so a file renamed while the filer is being read is - still counted and its chunks are not mistaken for orphans. Other stores are - walked directory by directory; there, avoid running with -reallyDeleteFromVolume - while files are being renamed or moved. + that (the SQL stores and etcd do), so a file renamed while the filer is being + read is still counted and its chunks are not mistaken for orphans. Other + stores are walked directory by directory; there, avoid running with + -reallyDeleteFromVolume while files are being renamed or moved. With -findMissingChunksInFiler, -volumeId may name a volume the master no longer lists, to find the filer entries that still reference it. A volume is