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