From 744ab586c5b6709466ef3c55c1b1272d745fd2c2 Mon Sep 17 00:00:00 2001 From: Jaehoon Kim Date: Thu, 22 Oct 2026 12:31:42 +0900 Subject: [PATCH] fix: volume.fsck purges live chunks of files renamed during its filer walk (SQL stores) (#11883) * shell: volume.fsck reads the filer from one store snapshot when it can volume.fsck collects the filer's chunk references by walking directory by directory, so each directory is seen at a different time. A file renamed from a directory not yet listed into one already listed is never seen, its chunks look like orphans, and -reallyDeleteFromVolume deletes live data. Checkpoint rotation and write-to-tmp-then-rename hit this. Add an optional filer store capability, EntryScanner, that visits every entry below a directory from one consistent snapshot. The SQL stores implement it with one REPEATABLE READ read-only transaction read in primary-key pages, so a concurrent rename is seen under exactly one of its names. Key-value rows are skipped by their dirhash, a bucket directory whose table is missing fails the scan (from the snapshot it cannot be told from one dropped a moment ago), and hard links are resolved in the same transaction. SQLite offers it only in WAL mode, on its own connection. TraverseBfsMetadata serves the snapshot when asked via request metadata sw-traverse-mode=snapshot and reports the mode in a response header; without the metadata it walks as before, so peer bootstrap and admin export are unchanged. Snapshot responses carry no entry_ts_ns, whose per-entry lookup would cost more than the scan. volume.fsck requests the snapshot and falls back to the walk on an older filer or a store that cannot scan. * filer, shell: fail a scan that cannot start, not fall back to the walk A transient BeginTx failure and a preliminary GetEntry error were treated like a missing capability or missing root, so fsck would take the directory walk - the very path that misses renamed-away entries. Only a genuinely unsupported scan or a confirmed-absent root may fall back. * filer: only snapshot-listed bucket names keep a missing table fatal A cached-only name, whose bucket may have been deleted between copying the cache and opening the snapshot, is skipped when its table is gone. * filer: roll back to a savepoint before skipping a cached-only table On PostgreSQL a failed statement aborts the snapshot transaction, so ignoring a missing-table error without the savepoint still left every later cached-only table unreadable. --------- Co-authored-by: Chris Lu --- weed/filer/abstract_sql/abstract_sql_store.go | 13 +- .../abstract_sql/abstract_sql_store_scan.go | 368 ++++++++++++++++++ .../abstract_sql_store_scan_test.go | 51 +++ weed/filer/filerstore.go | 16 + weed/filer/filerstore_wrapper.go | 29 ++ weed/filer/mysql/mysql_sql_gen.go | 9 + weed/filer/postgres/postgres_sql_gen.go | 8 + weed/filer/sqlite/sqlite_store.go | 34 +- weed/filer/sqlite/sqlite_store_scan_test.go | 288 ++++++++++++++ weed/pb/filer_pb/filer_client.go | 10 + .../server/filer_grpc_server_traverse_meta.go | 97 ++++- ...iler_grpc_server_traverse_snapshot_test.go | 170 ++++++++ weed/shell/command_fs_meta_save.go | 130 +++++++ weed/shell/command_volume_fsck.go | 8 +- 14 files changed, 1221 insertions(+), 10 deletions(-) create mode 100644 weed/filer/abstract_sql/abstract_sql_store_scan.go create mode 100644 weed/filer/abstract_sql/abstract_sql_store_scan_test.go create mode 100644 weed/filer/sqlite/sqlite_store_scan_test.go create mode 100644 weed/server/filer_grpc_server_traverse_snapshot_test.go diff --git a/weed/filer/abstract_sql/abstract_sql_store.go b/weed/filer/abstract_sql/abstract_sql_store.go index e358727743..52f959d80a 100644 --- a/weed/filer/abstract_sql/abstract_sql_store.go +++ b/weed/filer/abstract_sql/abstract_sql_store.go @@ -23,14 +23,20 @@ type SqlGenerator interface { GetSqlDeleteFolderChildren(tableName string) string GetSqlListExclusive(tableName string) string GetSqlListInclusive(tableName string) string + // GetSqlScanPage returns the query and arguments selecting dirhash, name, + // directory and meta of up to limit rows after the (dirhash, name) cursor. + GetSqlScanPage(tableName string, dirhash int64, name string, limit int) (string, []interface{}) GetSqlCreateTable(tableName string) string GetSqlDropTable(tableName string) string } type AbstractSqlStore struct { SqlGenerator - DB *sql.DB - KvDB *sql.DB + DB *sql.DB + KvDB *sql.DB + // ScanDB, when set, is the pool snapshot scans run on (SQLite's main pool + // is a single connection that a long scan must not hold). + ScanDB *sql.DB SupportBucketTable bool SkipDDL bool dbs map[string]bool @@ -457,6 +463,9 @@ func (store *AbstractSqlStore) Shutdown() { if store.KvDB != nil { store.KvDB.Close() } + if store.ScanDB != nil { + store.ScanDB.Close() + } } func isValidBucket(bucket string) bool { diff --git a/weed/filer/abstract_sql/abstract_sql_store_scan.go b/weed/filer/abstract_sql/abstract_sql_store_scan.go new file mode 100644 index 0000000000..2028643755 --- /dev/null +++ b/weed/filer/abstract_sql/abstract_sql_store_scan.go @@ -0,0 +1,368 @@ +package abstract_sql + +import ( + "context" + "database/sql" + "fmt" + "math" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +var _ filer.EntryScanner = (*AbstractSqlStore)(nil) + +// scanTable is one table to read during a scan together with the path its +// directory column is relative to: "" for the default table, whose rows hold +// full directories, and "/buckets/" for a per-bucket table. +type scanTable struct { + table string + base string + // cachedOnly marks a table that only a stale bucket cache remembers: its + // bucket directory was already gone from the snapshot, so a missing table + // means it was dropped meanwhile, not that rows are absent. + cachedOnly bool +} + +// scanPageSize is how many rows one keyset page of a scan fetches. +var scanPageSize = 1000 + +// SetScanPageSizeForTest lets tests drive the keyset cursor across pages. It +// returns the previous size. +func SetScanPageSizeForTest(n int) int { + prev := scanPageSize + scanPageSize = n + return prev +} + +// ScanEntries implements filer.EntryScanner: every table that can hold an +// entry below dirPath is read inside one REPEATABLE READ, read-only +// transaction, in primary-key pages. All rows are seen as of one instant; a +// rename committed meanwhile shows up under exactly one of its names. The +// transaction takes no row locks and does not block writers; it does hold the +// tables it has read against DDL, so a bucket deleted while the scan runs +// drops its table once the scan ends. A page's cursor is closed before its +// entries go to fn, so fn may read the store through the context it receives +// (which carries the transaction) on the same connection. +func (store *AbstractSqlStore) ScanEntries(ctx context.Context, dirPath util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + // Read before the transaction holds a connection: OnBucketCreation keeps + // dbsLock while it waits for a connection of its own, so taking the lock + // while holding a connection could deadlock on an exhausted pool. + cachedBuckets := store.cachedBucketNames() + + tx, err := store.beginSnapshotTx(ctx) + if err != nil { + return err + } + // Read-only: nothing to commit, and a rollback releases the snapshot. + defer tx.Rollback() + + tables, err := store.scanTables(ctx, tx, dirPath, cachedBuckets) + if err != nil { + return err + } + // Reads made through this context (getTxOrDB, kvDB) use the transaction + // and therefore the snapshot. + txCtx := context.WithValue(ctx, "tx", tx) + // The root's own row may live in a table the scan does not read (a + // bucket directory is in the default table, its contents in the bucket's). + if root, err := store.readRowInSnapshot(txCtx, tx, dirPath); err != nil { + return err + } else if root != nil { + if err := fn(txCtx, root); err != nil { + return err + } + } + for _, t := range tables { + if err := store.scanOneTable(txCtx, tx, t, dirPath, fn); err != nil { + return err + } + } + return nil +} + +// readRowInSnapshot reads one entry's row inside the transaction, routing to +// the table getTxOrDB would use but without its bucket-table creation. A +// missing row is nil. +func (store *AbstractSqlStore) readRowInSnapshot(ctx context.Context, tx *sql.Tx, p util.FullPath) (*filer.Entry, error) { + if p == "/" { + return nil, nil + } + parent, name := p.DirAndName() + table, shortDir := DEFAULT_TABLE, parent + if store.SupportBucketTable { + if bucket := bucketTableFor(util.FullPath(parent)); bucket != "" { + table = bucket + if shortDir = strings.TrimPrefix(parent, bucketBase(bucket)); shortDir == "" { + shortDir = "/" + } + } + } + var data []byte + err := tx.QueryRowContext(ctx, store.GetSqlFind(table), util.HashStringToLong(shortDir), name, shortDir).Scan(&data) + if err == sql.ErrNoRows { + return nil, nil + } + if err != nil { + return nil, missingTableOr(err, table, fmt.Sprintf("read %s", p)) + } + entry := &filer.Entry{FullPath: p} + if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil { + return nil, fmt.Errorf("decode %s: %v", p, err) + } + return entry, nil +} + +// beginSnapshotTx starts the read-only REPEATABLE READ transaction that gives +// one snapshot across statements. Nothing weaker is tried: under READ +// COMMITTED each page would see a different snapshot, and the stream would +// still be advertised as one. +func (store *AbstractSqlStore) beginSnapshotTx(ctx context.Context) (*sql.Tx, error) { + db := store.DB + if store.ScanDB != nil { + db = store.ScanDB + } else if db.Stats().MaxOpenConnections == 1 { + // The scan would hold the filer's only connection for its whole run. + return nil, fmt.Errorf("%w: the store's single connection cannot be held for a scan", filer.ErrUnsupportedEntryScan) + } + // A BeginTx failure (a down database) must fail the traversal, not route it + // to the walk: a recovered database lets the walk read renamed-away entries + // as absent, the very loss this scan exists to prevent. + tx, err := db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelRepeatableRead, ReadOnly: true}) + if err != nil { + return nil, fmt.Errorf("begin snapshot: %v", err) + } + return tx, nil +} + +func (store *AbstractSqlStore) cachedBucketNames() []string { + store.dbsLock.Lock() + defer store.dbsLock.Unlock() + names := make([]string, 0, len(store.dbs)) + for name := range store.dbs { + names = append(names, name) + } + return names +} + +// scanTables picks the tables to read for dirPath: the default table alone, +// one bucket's table for a path inside that bucket, or the default table plus +// every bucket table (enumerated from /buckets inside the snapshot). +func (store *AbstractSqlStore) scanTables(ctx context.Context, tx *sql.Tx, dirPath util.FullPath, cachedBuckets []string) ([]scanTable, error) { + if !store.SupportBucketTable { + return []scanTable{{table: DEFAULT_TABLE}}, nil + } + if bucket := bucketTableFor(dirPath); bucket != "" { + return []scanTable{{table: bucket, base: bucketBase(bucket)}}, nil + } + tables := []scanTable{{table: DEFAULT_TABLE}} + if dirPath != "/" && string(dirPath) != bucketsRoot { + // Outside every bucket (e.g. /topics, or an internal /buckets/.system + // folder): everything below dirPath is in the default table. + return tables, nil + } + buckets, err := store.listBucketNames(ctx, tx, cachedBuckets) + if err != nil { + return nil, err + } + for _, bucket := range buckets.names { + tables = append(tables, scanTable{table: bucket, base: bucketBase(bucket), cachedOnly: !buckets.seen[bucket]}) + } + return tables, nil +} + +const bucketsRoot = "/buckets" + +func bucketBase(bucket string) string { + return bucketsRoot + "/" + bucket +} + +// bucketTableFor returns the bucket whose table alone holds everything below +// dirPath, or "" when dirPath is not inside a real bucket (as getTxOrDB routes). +func bucketTableFor(dirPath util.FullPath) string { + p := string(dirPath) + if !strings.HasPrefix(p, bucketsRoot+"/") { + return "" + } + rest := p[len(bucketsRoot)+1:] + bucket := rest + if i := strings.Index(rest, "/"); i >= 0 { + bucket = rest[:i] + } + if bucket == "" || strings.HasPrefix(bucket, ".") || !isValidBucket(bucket) { + return "" + } + return bucket +} + +// bucketNames is the snapshot-listed and cached-only buckets whose tables a +// scan may read. +type bucketNames struct { + names []string + seen map[string]bool // listed under /buckets inside the snapshot +} + +// listBucketNames reads the directories under /buckets within the snapshot +// and keeps those that route to their own table, plus the buckets this +// process has already seen (their directory entry may be gone, their rows not). +func (store *AbstractSqlStore) listBucketNames(ctx context.Context, tx *sql.Tx, cachedBuckets []string) (bucketNames, error) { + var names bucketNames + names.seen = make(map[string]bool) + added := make(map[string]bool) + add := func(name string) { + if added[name] || strings.HasPrefix(name, ".") || !isValidBucket(name) { + return + } + added[name] = true + names.names = append(names.names, name) + } + + const page = 1000 + startFrom := "" + for { + type row struct { + name string + data []byte + } + var rows []row + if err := func() error { + res, err := tx.QueryContext(ctx, store.GetSqlListExclusive(DEFAULT_TABLE), util.HashStringToLong(bucketsRoot), startFrom, bucketsRoot, "%", page) + if err != nil { + return fmt.Errorf("list %s: %v", bucketsRoot, err) + } + defer res.Close() + for res.Next() { + var r row + if err := res.Scan(&r.name, &r.data); err != nil { + return fmt.Errorf("scan %s: %v", bucketsRoot, err) + } + rows = append(rows, r) + } + return res.Err() + }(); err != nil { + return bucketNames{}, err + } + for _, r := range rows { + startFrom = r.name + entry := &filer.Entry{FullPath: util.NewFullPath(bucketsRoot, r.name)} + if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(r.data)); err != nil { + return bucketNames{}, fmt.Errorf("scan decode %s: %v", entry.FullPath, err) + } + if entry.IsDirectory() { + names.seen[r.name] = true + add(r.name) + } + } + if len(rows) < page { + break + } + } + + for _, name := range cachedBuckets { + add(name) + } + return names, nil +} + +// scanOneTable reads one table in primary-key pages and hands the entries +// strictly below dirPath to fn (the root itself is read separately). Key-value +// records share the default table; GenDirAndName stores the raw key bytes as +// their dirhash while entry rows hash the directory, so the mismatch +// identifies them exactly. +func (store *AbstractSqlStore) scanOneTable(ctx context.Context, tx *sql.Tx, t scanTable, dirPath util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + type row struct { + dirHash int64 + name, dir string + data []byte + } + cursorHash, cursorName := int64(math.MinInt64), "" + firstPage := true + for { + var page []row + if t.cachedOnly && firstPage { + // A missing table can only be met by the first page read: once a + // row came back the tx holds the table against a drop. On + // PostgreSQL the failed query aborts the transaction, so the skip + // needs a savepoint to roll back to before the next table's read. + if _, err := tx.ExecContext(ctx, "SAVEPOINT seaweed_scan_bucket"); err != nil { + return fmt.Errorf("scan table %s: %v", t.table, err) + } + } + sqlText, args := store.GetSqlScanPage(t.table, cursorHash, cursorName, scanPageSize) + rows, err := tx.QueryContext(ctx, sqlText, args...) + if err != nil { + if t.cachedOnly && isMissingTableError(err) { + tx.ExecContext(ctx, "ROLLBACK TO SAVEPOINT seaweed_scan_bucket") + return nil // a cached-only name whose table was dropped meanwhile + } + return missingTableOr(err, t.table, fmt.Sprintf("scan table %s", t.table)) + } + firstPage = false + if err := func() error { + defer rows.Close() + for rows.Next() { + var r row + if err := rows.Scan(&r.dirHash, &r.name, &r.dir, &r.data); err != nil { + return fmt.Errorf("scan table %s: %v", t.table, err) + } + page = append(page, r) + } + return rows.Err() + }(); err != nil { + return err + } + + for _, r := range page { + cursorHash, cursorName = r.dirHash, r.name + if r.dirHash != util.HashStringToLong(r.dir) { + continue // a key-value record, see GenDirAndName + } + fullPath := util.NewFullPath(t.base+r.dir, r.name) + if !isStrictlyBelow(fullPath, dirPath) { + continue + } + entry := &filer.Entry{FullPath: fullPath} + if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(r.data)); err != nil { + return fmt.Errorf("scan decode %s: %v", fullPath, err) + } + if err := fn(ctx, entry); err != nil { + return err + } + } + if len(page) < scanPageSize { + return nil + } + } +} + +// missingTableOr names the condition when a bucket's table is not there. The +// scan fails closed: from inside the snapshot a table that was never created +// cannot be told from one dropped a moment ago, whose rows the snapshot still +// counts, and treating either as empty would let fsck purge referenced chunks. +func missingTableOr(err error, table, what string) error { + if table != DEFAULT_TABLE && isMissingTableError(err) { + return fmt.Errorf("bucket %s has a directory entry but no table (never created, or dropped during the scan): %v", table, err) + } + return fmt.Errorf("%s: %v", what, err) +} + +// isMissingTableError recognizes a missing table by message, as InsertEntry +// recognizes duplicates: PostgreSQL `relation "x" does not exist` (SQLSTATE +// 42P01), MySQL error 1146 (SQLSTATE 42S02), SQLite "no such table". +func isMissingTableError(err error) bool { + msg := strings.ToLower(err.Error()) + return strings.Contains(msg, "42p01") || + (strings.Contains(msg, "relation \"") && strings.Contains(msg, "does not exist")) || + strings.Contains(msg, "42s02") || + strings.Contains(msg, "error 1146") || + strings.Contains(msg, "no such table") +} + +// isStrictlyBelow reports whether p is a descendant of dir (not dir itself). +func isStrictlyBelow(p, dir util.FullPath) bool { + if dir == "/" { + return p != "/" + } + return strings.HasPrefix(string(p), string(dir)+"/") +} diff --git a/weed/filer/abstract_sql/abstract_sql_store_scan_test.go b/weed/filer/abstract_sql/abstract_sql_store_scan_test.go new file mode 100644 index 0000000000..e9c7cde6b0 --- /dev/null +++ b/weed/filer/abstract_sql/abstract_sql_store_scan_test.go @@ -0,0 +1,51 @@ +package abstract_sql + +import ( + "errors" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/util" +) + +func TestBucketTableFor(t *testing.T) { + cases := map[string]string{ + "/": "", + "/buckets": "", + "/buckets/": "", + "/buckets/mybucket": "mybucket", + "/buckets/mybucket/a/b": "mybucket", + "/buckets/.system/owners": "", + "/buckets/Invalid_Bucket/x": "", + "/topics/.system/log": "", + "/bucketsfoo/mybucket": "", + } + for path, want := range cases { + if got := bucketTableFor(util.FullPath(path)); got != want { + t.Errorf("bucketTableFor(%s) = %q, want %q", path, got, want) + } + } +} + +// Only a missing table may be skipped; any other schema error must fail the +// scan, or a bucket's live chunks would be reported as orphans. +func TestIsMissingTableError(t *testing.T) { + for _, msg := range []string{ + `ERROR: relation "ghost" does not exist (SQLSTATE 42P01)`, + `Error 1146 (42S02): Table 'seaweedfs.ghost' doesn't exist`, + `SQL logic error: no such table: ghost (1)`, + } { + if !isMissingTableError(errors.New(msg)) { + t.Errorf("not recognized as a missing table: %s", msg) + } + } + for _, msg := range []string{ + `ERROR: column "directory" does not exist (SQLSTATE 42703)`, + `Error 1054 (42S22): Unknown column 'directory' in 'field list'`, + `SQL logic error: no such column: directory (1)`, + "connection refused", + } { + if isMissingTableError(errors.New(msg)) { + t.Errorf("wrongly taken for a missing table: %s", msg) + } + } +} diff --git a/weed/filer/filerstore.go b/weed/filer/filerstore.go index 9689436081..277903249a 100644 --- a/weed/filer/filerstore.go +++ b/weed/filer/filerstore.go @@ -44,6 +44,22 @@ type FilerStore interface { Shutdown() } +// ErrUnsupportedEntryScan is returned by EntryScanner.ScanEntries, before +// any entry is visited, when the store cannot serve the scan. +var ErrUnsupportedEntryScan = errors.New("entry scan not supported by this filer store") + +// EntryScanner is an optional FilerStore capability: visit every entry at and +// below dirPath, in any order, from one consistent snapshot of the store. +// +// A directory walk sees each directory at a different time, so an entry +// renamed from a directory not yet listed into one already listed is never +// seen; volume.fsck then purges its chunks as orphans. A snapshot sees it +// exactly once. The context handed to fn carries the snapshot, so store reads +// made with it (hard-link records) see the same instant. +type EntryScanner interface { + ScanEntries(ctx context.Context, dirPath util.FullPath, fn func(ctx context.Context, entry *Entry) error) error +} + type BucketAware interface { OnBucketCreation(bucket string) OnBucketDeletion(bucket string) diff --git a/weed/filer/filerstore_wrapper.go b/weed/filer/filerstore_wrapper.go index 070bc19fe4..1a17e22cf6 100644 --- a/weed/filer/filerstore_wrapper.go +++ b/weed/filer/filerstore_wrapper.go @@ -2,6 +2,7 @@ package filer import ( "context" + "errors" "fmt" "io" "math" @@ -19,6 +20,7 @@ import ( var ( _ = VirtualFilerStore(&FilerStoreWrapper{}) _ = Debuggable(&FilerStoreWrapper{}) + _ = EntryScanner(&FilerStoreWrapper{}) ) type VirtualFilerStore interface { @@ -364,6 +366,33 @@ func (fsw *FilerStoreWrapper) ListDirectoryPrefixedEntries(ctx context.Context, return lastFileName, err } +// ScanEntries delegates to the default store. With path-specific stores +// configured, entries below dirPath could live in several stores and one +// snapshot per store is not one snapshot, so the scan is declined. Hard links +// are resolved inside the snapshot; a store error fails the scan (the per-path +// copy can be stale), a missing record falls back to it as listings do. +func (fsw *FilerStoreWrapper) ScanEntries(ctx context.Context, dirPath util.FullPath, fn func(ctx context.Context, entry *Entry) error) error { + if fsw.hasPathSpecificStore { + return ErrUnsupportedEntryScan + } + scanner, ok := fsw.defaultStore.(EntryScanner) + if !ok { + return ErrUnsupportedEntryScan + } + stats.FilerStoreCounter.WithLabelValues(fsw.defaultStore.GetName(), "scan").Inc() + start := time.Now() + defer func() { + stats.FilerStoreHistogram.WithLabelValues(fsw.defaultStore.GetName(), "scan").Observe(time.Since(start).Seconds()) + }() + return scanner.ScanEntries(ctx, dirPath, func(scanCtx context.Context, entry *Entry) error { + if err := fsw.maybeReadHardLink(scanCtx, entry); err != nil && !errors.Is(err, ErrKvNotFound) { + return fmt.Errorf("resolve hard link of %s: %w", entry.FullPath, err) + } + filer_pb.AfterEntryDeserialization(entry.GetChunks()) + return fn(scanCtx, entry) + }) +} + func (fsw *FilerStoreWrapper) prefixFilterEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc ListEachEntryFunc) (lastFileName string, err error) { actualStore := fsw.getActualStore(dirPath + "/") diff --git a/weed/filer/mysql/mysql_sql_gen.go b/weed/filer/mysql/mysql_sql_gen.go index 049ea685a0..3566e4e99c 100644 --- a/weed/filer/mysql/mysql_sql_gen.go +++ b/weed/filer/mysql/mysql_sql_gen.go @@ -62,6 +62,15 @@ func (gen *SqlGenMysql) nameExpr() string { return "`name`" } +// GetSqlScanPage spells the cursor as two primary-key ranges: MySQL 5.7 does +// not turn a row-value comparison into a range seek. The name keeps the +// column's own collation (BINARY would filesort every page); the scan promises +// no lexical order. SQLite overrides this method. +func (gen *SqlGenMysql) GetSqlScanPage(tableName string, dirhash int64, name string, limit int) (string, []interface{}) { + return fmt.Sprintf("SELECT `dirhash`, `name`, `directory`, `meta` FROM `%s` WHERE `dirhash` > ? OR (`dirhash` = ? AND `name` > ?) ORDER BY `dirhash` ASC, `name` ASC LIMIT ?", tableName), + []interface{}{dirhash, dirhash, name, limit} +} + func (gen *SqlGenMysql) GetSqlListExclusive(tableName string) string { name := gen.nameExpr() return fmt.Sprintf("SELECT `name`, `meta` FROM `%s` WHERE `dirhash` = ? AND %s > ? AND `directory` = ? AND %s LIKE ? ESCAPE '!' ORDER BY %s ASC LIMIT ?", tableName, name, name, name) diff --git a/weed/filer/postgres/postgres_sql_gen.go b/weed/filer/postgres/postgres_sql_gen.go index 9ff50d96a1..4ba99354cf 100644 --- a/weed/filer/postgres/postgres_sql_gen.go +++ b/weed/filer/postgres/postgres_sql_gen.go @@ -75,6 +75,14 @@ func (gen *SqlGenPostgres) nameExpr() string { return "name" } +// GetSqlScanPage uses a row-value comparison, one primary-key range seek on +// PostgreSQL. The name keeps the column's own collation (COLLATE "C" would +// sort every page); the scan promises no lexical order. +func (gen *SqlGenPostgres) GetSqlScanPage(tableName string, dirhash int64, name string, limit int) (string, []interface{}) { + return fmt.Sprintf(`SELECT dirhash, name, directory, meta FROM "%s" WHERE (dirhash, name) > ($1, $2) ORDER BY dirhash ASC, name ASC LIMIT $3`, tableName), + []interface{}{dirhash, name, limit} +} + func (gen *SqlGenPostgres) GetSqlListExclusive(tableName string) string { name := gen.nameExpr() return fmt.Sprintf(`SELECT NAME, meta FROM "%s" WHERE dirhash=$1 AND %s>$2 AND directory=$3 AND %s like $4 ESCAPE '!' ORDER BY %s ASC LIMIT $5`, tableName, name, name, name) diff --git a/weed/filer/sqlite/sqlite_store.go b/weed/filer/sqlite/sqlite_store.go index 31d696ca73..7a45ef83ca 100644 --- a/weed/filer/sqlite/sqlite_store.go +++ b/weed/filer/sqlite/sqlite_store.go @@ -30,6 +30,31 @@ func (store *SqliteStore) GetName() string { return "sqlite" } +// sqliteSqlGen is the MySQL generator with SQLite's own page cursor: SQLite +// seeks a row-value comparison on the primary key but scans the OR form. +type sqliteSqlGen struct { + *mysql.SqlGenMysql +} + +func (gen sqliteSqlGen) GetSqlScanPage(tableName string, dirhash int64, name string, limit int) (string, []interface{}) { + return fmt.Sprintf("SELECT `dirhash`, `name`, `directory`, `meta` FROM `%s` WHERE (`dirhash`, `name`) > (?, ?) ORDER BY `dirhash` ASC, `name` ASC LIMIT ?", tableName), + []interface{}{dirhash, name, limit} +} + +// ScanEntries offers the snapshot scan only in WAL mode: with a rollback +// journal a long read transaction holds a shared lock and writers fail after +// the busy timeout. +func (store *SqliteStore) ScanEntries(ctx context.Context, dirPath util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + var mode string + if err := store.DB.QueryRowContext(ctx, "PRAGMA journal_mode").Scan(&mode); err != nil { + return fmt.Errorf("%w: journal_mode: %v", filer.ErrUnsupportedEntryScan, err) + } + if !strings.EqualFold(mode, "wal") { + return fmt.Errorf("%w: sqlite journal_mode is %q, the scan needs wal", filer.ErrUnsupportedEntryScan, mode) + } + return store.AbstractSqlStore.ScanEntries(ctx, dirPath, fn) +} + func (store *SqliteStore) Initialize(configuration util.Configuration, prefix string) (err error) { dbFile := configuration.GetString(prefix + "dbFile") createTable := `CREATE TABLE IF NOT EXISTS "%s" ( @@ -69,11 +94,11 @@ func sqliteDSN(dbFile string) string { func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (err error) { store.SupportBucketTable = true - store.SqlGenerator = &mysql.SqlGenMysql{ + store.SqlGenerator = sqliteSqlGen{&mysql.SqlGenMysql{ CreateTableSqlTemplate: createTable, DropTableSqlTemplate: "drop table if exists `%s`", UpsertQueryTemplate: upsertQuery, - } + }} dsn := sqliteDSN(dbFile) @@ -94,6 +119,11 @@ func (store *SqliteStore) initialize(dbFile, createTable, upsertQuery string) (e }, 1, 1, 0); err != nil { return err } + // The main pool is one connection, which a long scan must not hold. + if store.ScanDB, err = sql.Open("sqlite", dsn); err != nil { + return fmt.Errorf("open scan connection to %s: %v", dbFile, err) + } + store.ScanDB.SetMaxOpenConns(1) if err = store.CreateTable(context.Background(), abstract_sql.DEFAULT_TABLE); err != nil { return fmt.Errorf("init table %s: %v", abstract_sql.DEFAULT_TABLE, err) diff --git a/weed/filer/sqlite/sqlite_store_scan_test.go b/weed/filer/sqlite/sqlite_store_scan_test.go new file mode 100644 index 0000000000..b859229171 --- /dev/null +++ b/weed/filer/sqlite/sqlite_store_scan_test.go @@ -0,0 +1,288 @@ +//go:build (linux || darwin || windows) && sqlite + +package sqlite + +import ( + "context" + "errors" + "os" + "path/filepath" + "reflect" + "sort" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/filer/abstract_sql" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// A rename committed while a scan is running is seen under exactly one of its +// names. The scan reads from one snapshot, so unlike a directory walk it can +// neither miss the entry when it moves into a directory already listed, nor +// count it twice when it moves into one not listed yet. +func TestScanEntriesSeesConcurrentRenameExactlyOnce(t *testing.T) { + dbFile := filepath.Join(t.TempDir(), "filer.db") + reader := &SqliteStore{} + if err := reader.initialize(dbFile, createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer reader.Shutdown() + // WAL lets the writer commit while the scan's read transaction is open. + if _, err := reader.DB.Exec("PRAGMA journal_mode=WAL"); err != nil { + t.Fatal(err) + } + writer := &SqliteStore{} + if err := writer.initialize(dbFile, createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer writer.Shutdown() + + ctx := context.Background() + mk := func(p string) *filer.Entry { + return &filer.Entry{FullPath: util.FullPath(p), Attr: filer.Attr{Mode: 0644, Mtime: time.Now(), Crtime: time.Now()}} + } + for _, p := range []string{"/ckpt/a", "/ckpt/b", "/ckpt/tmp/x"} { + if err := reader.InsertEntry(ctx, mk(p)); err != nil { + t.Fatal(err) + } + } + + defer func(n int) { abstract_sql.SetScanPageSizeForTest(n) }(abstract_sql.SetScanPageSizeForTest(2)) + + seen := map[string]int{} + renamed := false + err := reader.ScanEntries(ctx, "/", func(_ context.Context, e *filer.Entry) error { + seen[string(e.FullPath)]++ + if renamed { + return nil + } + renamed = true + // /ckpt/tmp/x -> /ckpt/x, committed by another connection mid-scan. + if err := writer.InsertEntry(ctx, mk("/ckpt/x")); err != nil { + return err + } + return writer.DeleteEntry(ctx, util.FullPath("/ckpt/tmp/x")) + }) + if err != nil { + t.Fatal(err) + } + if !renamed { + t.Fatal("scan yielded nothing") + } + if seen["/ckpt/a"] != 1 || seen["/ckpt/b"] != 1 { + t.Errorf("untouched entries seen %d and %d times, want once each", seen["/ckpt/a"], seen["/ckpt/b"]) + } + if old, new := seen["/ckpt/tmp/x"], seen["/ckpt/x"]; old+new != 1 { + t.Errorf("renamed entry seen %d time(s) under the old name and %d under the new, want exactly one in total", old, new) + } else if old != 1 { + t.Errorf("the rename committed after the snapshot was taken, so the scan must show the old name") + } + if _, err := reader.FindEntry(ctx, util.FullPath("/ckpt/x")); err != nil { + t.Errorf("once the scan is over the new name must be visible: %v", err) + } +} + +// Rows that are not entries, key-value records and a regular file under +// /buckets with a bucket-like name, are stepped over; a scan rooted at a bucket +// yields the bucket's own row (default table) and its contents (bucket table). +func TestScanEntriesRowsAndBucketTables(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + if _, err := store.DB.Exec("PRAGMA journal_mode=WAL"); err != nil { + t.Fatal(err) + } + + ctx := context.Background() + now := time.Now() + // The filer creates bucket tables on bucket creation; the store layer alone does not. + if err := store.CreateTable(ctx, "realbucket"); err != nil { + t.Fatal(err) + } + for _, e := range []*filer.Entry{ + {FullPath: "/buckets", Attr: filer.Attr{Mode: 0755 | os.ModeDir, Mtime: now, Crtime: now}}, + {FullPath: "/buckets/realbucket", Attr: filer.Attr{Mode: 0755 | os.ModeDir, Mtime: now, Crtime: now}}, + {FullPath: "/buckets/realbucket/f1", Attr: filer.Attr{Mode: 0644, Mtime: now, Crtime: now}}, + {FullPath: "/buckets/readme", Attr: filer.Attr{Mode: 0644, Mtime: now, Crtime: now}}, + {FullPath: "/topics/t1", Attr: filer.Attr{Mode: 0644, Mtime: now, Crtime: now}}, + } { + if err := store.InsertEntry(ctx, e); err != nil { + t.Fatal(err) + } + } + for _, key := range []string{"filer.store.id", "/looks/like/a/path", "x"} { + if err := store.KvPut(ctx, []byte(key), []byte("opaque value, not an Entry")); err != nil { + t.Fatal(err) + } + } + + collect := func(root string) []string { + var got []string + if err := store.ScanEntries(ctx, util.FullPath(root), func(_ context.Context, e *filer.Entry) error { + got = append(got, string(e.FullPath)) + return nil + }); err != nil { + t.Fatalf("scan %s: %v", root, err) + } + sort.Strings(got) + return got + } + if got, want := collect("/"), []string{"/buckets", "/buckets/readme", "/buckets/realbucket", "/buckets/realbucket/f1", "/topics/t1"}; !reflect.DeepEqual(got, want) { + t.Errorf("scan /: got %v, want %v", got, want) + } + if got, want := collect("/buckets/realbucket"), []string{"/buckets/realbucket", "/buckets/realbucket/f1"}; !reflect.DeepEqual(got, want) { + t.Errorf("scan /buckets/realbucket: got %v, want %v", got, want) + } + + // A bucket directory whose table does not exist fails the scan: from the + // snapshot it cannot be told from a table dropped a moment ago. + if err := store.InsertEntry(ctx, &filer.Entry{FullPath: "/buckets/ghostbucket", Attr: filer.Attr{Mode: 0755 | os.ModeDir, Mtime: now, Crtime: now}}); err != nil { + t.Fatal(err) + } + err := store.ScanEntries(ctx, "/", func(context.Context, *filer.Entry) error { return nil }) + if err == nil || !strings.Contains(err.Error(), "ghostbucket") { + t.Fatalf("scan over a bucket without a table must fail naming it, got %v", err) + } +} + +// Without WAL the store declines the scan rather than block writers for its +// whole duration. +func TestScanEntriesDeclinedWithoutWAL(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + err := store.ScanEntries(context.Background(), "/", func(context.Context, *filer.Entry) error { return nil }) + if !errors.Is(err, filer.ErrUnsupportedEntryScan) { + t.Fatalf("err = %v, want ErrUnsupportedEntryScan", err) + } +} + +// A store whose pool is one connection, and no scan connection of its own, +// declines rather than hold that connection for the whole scan. +func TestScanEntriesDeclinedOnSingleConnectionPool(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + if _, err := store.DB.Exec("PRAGMA journal_mode=WAL"); err != nil { + t.Fatal(err) + } + store.ScanDB.Close() + store.ScanDB = nil + err := store.ScanEntries(context.Background(), "/", func(context.Context, *filer.Entry) error { return nil }) + if !errors.Is(err, filer.ErrUnsupportedEntryScan) { + t.Fatalf("err = %v, want ErrUnsupportedEntryScan", err) + } +} + +// A hard-linked entry's chunks live in the shared key-value record, not in +// its path row; the wrapper resolves them from inside the snapshot, on the +// single connection the scan holds. +func TestScanEntriesResolvesHardLinks(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + if _, err := store.DB.Exec("PRAGMA journal_mode=WAL"); err != nil { + t.Fatal(err) + } + wrapper := filer.NewFilerStoreWrapper(store) + ctx := context.Background() + now := time.Now() + linked := &filer.Entry{ + FullPath: "/buckets/hlbucket/f0001", + Attr: filer.Attr{Mode: 0644, Mtime: now, Crtime: now}, + HardLinkId: filer.HardLinkId("hardlink-00000001"), + HardLinkCounter: 2, + Chunks: []*filer_pb.FileChunk{{FileId: "3,01637037d6", Size: 4096}}, + } + if err := wrapper.InsertEntry(ctx, linked); err != nil { + t.Fatal(err) + } + + var got *filer.Entry + if err := wrapper.ScanEntries(ctx, "/", func(_ context.Context, e *filer.Entry) error { + if e.FullPath == linked.FullPath { + got = e + } + return nil + }); err != nil { + t.Fatal(err) + } + if got == nil || len(got.GetChunks()) != 1 || got.GetChunks()[0].FileId != "3,01637037d6" || got.HardLinkCounter != 2 { + t.Fatalf("hard-linked entry not resolved from its key-value record: %+v", got) + } + + // Once the record is gone the path row is all there is; the scan hands it + // over instead of failing on that entry every time it runs. + if err := store.KvDelete(ctx, []byte(linked.HardLinkId)); err != nil { + t.Fatal(err) + } + got = nil + if err := wrapper.ScanEntries(ctx, "/", func(_ context.Context, e *filer.Entry) error { + if e.FullPath == linked.FullPath { + got = e + } + return nil + }); err != nil { + t.Fatalf("a missing hard-link record must not fail the scan: %v", err) + } + if got == nil { + t.Fatal("entry with a missing hard-link record not handed over") + } +} + +// The keyset page must be a primary-key seek; an OR-shaped cursor predicate +// makes SQLite scan the table from the start for every page. +func TestScanPageQuerySeeksPrimaryKey(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + sqlText, args := store.GetSqlScanPage("filemeta", int64(-1), "", 10) + rows, err := store.DB.Query("EXPLAIN QUERY PLAN "+sqlText, args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var plan []string + for rows.Next() { + var id, parent, notused int + var detail string + if err := rows.Scan(&id, &parent, ¬used, &detail); err != nil { + t.Fatal(err) + } + plan = append(plan, detail) + } + joined := strings.Join(plan, " | ") + if !strings.Contains(joined, "SEARCH") || strings.Contains(joined, "SCAN") { + t.Fatalf("page query does not seek the primary key: %s", joined) + } +} + +// A bucket name only the process cache remembers may have its table dropped +// meanwhile; the scan skips it instead of failing. +func TestScanEntriesSkipsCachedOnlyBucketTable(t *testing.T) { + store := &SqliteStore{} + if err := store.initialize(filepath.Join(t.TempDir(), "filer.db"), createTableSql, upsertQuerySql); err != nil { + t.Fatal(err) + } + defer store.Shutdown() + if _, err := store.DB.Exec("PRAGMA journal_mode=WAL"); err != nil { + t.Fatal(err) + } + store.OnBucketCreation("gonebucket") // the table is never created + if err := store.ScanEntries(context.Background(), "/", func(context.Context, *filer.Entry) error { return nil }); err != nil { + t.Fatalf("cached-only bucket with a dropped table must not fail the scan: %v", err) + } +} diff --git a/weed/pb/filer_pb/filer_client.go b/weed/pb/filer_pb/filer_client.go index a3903cff48..257349b021 100644 --- a/weed/pb/filer_pb/filer_client.go +++ b/weed/pb/filer_pb/filer_client.go @@ -67,6 +67,16 @@ func GetEntry(ctx context.Context, filerClient FilerClient, fullFilePath util.Fu // consumers would mistake for an entry. const ListSnapshotTsNsTrailerKey = "sw-list-snapshot-ts-ns" +// TraverseModeHeaderKey is gRPC metadata on a TraverseBfsMetadata stream. As +// request metadata, TraverseModeSnapshot asks for the entries from one store +// snapshot (any order, entry_ts_ns left 0) instead of a breadth-first walk; as +// a response header it says what the filer did. An older filer sends none. +const ( + TraverseModeHeaderKey = "sw-traverse-mode" + TraverseModeSnapshot = "snapshot" + TraverseModeBfs = "bfs" +) + type EachEntryFunction func(entry *Entry, isLast bool) error func ReadDirAllEntries(ctx context.Context, filerClient FilerClient, fullDirPath util.FullPath, prefix string, fn EachEntryFunction) (err error) { diff --git a/weed/server/filer_grpc_server_traverse_meta.go b/weed/server/filer_grpc_server_traverse_meta.go index c59d52d79d..7ad1014f81 100644 --- a/weed/server/filer_grpc_server_traverse_meta.go +++ b/weed/server/filer_grpc_server_traverse_meta.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "errors" "fmt" "github.com/seaweedfs/seaweedfs/weed/filer" @@ -9,8 +10,16 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/viant/ptrie" + "google.golang.org/grpc/metadata" ) +// TraverseBfsMetadata streams every entry at and below req.Directory: by +// default a breadth-first walk, parents first, each response stamped with the +// path's commit timestamp. With request metadata TraverseModeHeaderKey=snapshot +// the entries come from one store snapshot instead (filer.EntryScanner), in any +// order, and the response header of the same name says which happened. Snapshot +// responses leave entry_ts_ns at 0: that lookup is one store round trip per +// entry, and the consumers that order by it use the walk. func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequest, stream filer_pb.SeaweedFiler_TraverseBfsMetadataServer) error { glog.V(0).Infof("TraverseBfsMetadata %v", req) @@ -40,12 +49,8 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ // descend into. Queuing the entries themselves would hold the whole // subtree in memory, including every file's chunk list, and can exhaust // the filer on large trees (e.g. during a peer's first-time bootstrap). - sendEntry := func(entry *filer.Entry) error { + send := func(entry *filer.Entry, entryTsNs int64) error { parent, _ := entry.FullPath.DirAndName() - entryTsNs, err := fs.filer.CommitTsNs(ctx, entry.FullPath, entry.Mtime.UnixNano()) - if err != nil { - return fmt.Errorf("commit timestamp of %s: %w", entry.FullPath, err) - } if err := stream.Send(&filer_pb.TraverseBfsMetadataResponse{ Directory: parent, Entry: entry.ToProtoEntry(), @@ -55,6 +60,16 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ } return nil } + sendEntry := func(entry *filer.Entry) error { + entryTsNs, err := fs.filer.CommitTsNs(ctx, entry.FullPath, entry.Mtime.UnixNano()) + if err != nil { + return fmt.Errorf("commit timestamp of %s: %w", entry.FullPath, err) + } + return send(entry, entryTsNs) + } + sendEntryUnstamped := func(entry *filer.Entry) error { + return send(entry, 0) + } dirEntry, err := fs.filer.FindEntry(ctx, util.FullPath(req.Directory)) if err != nil { @@ -63,6 +78,22 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ if isExcluded(dirEntry.FullPath) { return nil } + + if traverseSnapshotRequested(ctx) { + if scanner, ok := fs.filer.Store.(filer.EntryScanner); ok && dirEntry.IsDirectory() { + served, scanErr := traverseFromSnapshot(ctx, scanner, dirEntry, isExcluded, stream.SendHeader, sendEntryUnstamped) + if scanErr != nil { + return scanErr + } + if served { + return nil + } + } + if err := stream.SendHeader(metadata.Pairs(filer_pb.TraverseModeHeaderKey, filer_pb.TraverseModeBfs)); err != nil { + return fmt.Errorf("send traverse mode header: %w", err) + } + } + if err := sendEntry(dirEntry); err != nil { return err } @@ -93,6 +124,62 @@ func (fs *FilerServer) TraverseBfsMetadata(req *filer_pb.TraverseBfsMetadataRequ return nil } +// traverseSnapshotRequested reports whether the client asked for the entries +// from one store snapshot rather than a breadth-first walk. +func traverseSnapshotRequested(ctx context.Context) bool { + md, ok := metadata.FromIncomingContext(ctx) + if !ok { + return false + } + modes := md.Get(filer_pb.TraverseModeHeaderKey) + return len(modes) > 0 && modes[0] == filer_pb.TraverseModeSnapshot +} + +// traverseFromSnapshot serves the traversal from the store's snapshot scan, +// root included: the scan yields the root's own row, so it is as current as +// its descendants (only "/" has no row and is sent as the synthetic root). +// It returns served=false, nothing sent, when the store declines before +// yielding anything; once the first entry is out, any failure is an error. +func traverseFromSnapshot(ctx context.Context, scanner filer.EntryScanner, dirEntry *filer.Entry, isExcluded func(util.FullPath) bool, sendHeader func(metadata.MD) error, sendEntry func(*filer.Entry) error) (served bool, err error) { + started := false + start := func() error { + if started { + return nil + } + started = true + if err := sendHeader(metadata.Pairs(filer_pb.TraverseModeHeaderKey, filer_pb.TraverseModeSnapshot)); err != nil { + return fmt.Errorf("send traverse mode header: %w", err) + } + if dirEntry.FullPath == "/" { + return sendEntry(dirEntry) + } + return nil + } + + scanErr := scanner.ScanEntries(ctx, dirEntry.FullPath, func(_ context.Context, entry *filer.Entry) error { + if isExcluded(entry.FullPath) { + return nil + } + if err := start(); err != nil { + return err + } + return sendEntry(entry) + }) + if scanErr != nil { + if !started && errors.Is(scanErr, filer.ErrUnsupportedEntryScan) { + glog.V(1).Infof("TraverseBfsMetadata %s: %v, walking instead", dirEntry.FullPath, scanErr) + return false, nil + } + glog.V(0).Infof("TraverseBfsMetadata %s: snapshot scan failed: %v", dirEntry.FullPath, scanErr) + return started, fmt.Errorf("snapshot scan of %s: %w", dirEntry.FullPath, scanErr) + } + // Nothing yielded (root gone since the lookup): still commit to the mode. + if err := start(); err != nil { + return true, err + } + return true, nil +} + func (fs *FilerServer) iterateDirectory(ctx context.Context, dirPath util.FullPath, fn func(entry *filer.Entry) error) (err error) { var lastFileName string var listErr error diff --git a/weed/server/filer_grpc_server_traverse_snapshot_test.go b/weed/server/filer_grpc_server_traverse_snapshot_test.go new file mode 100644 index 0000000000..14eb6fdfcc --- /dev/null +++ b/weed/server/filer_grpc_server_traverse_snapshot_test.go @@ -0,0 +1,170 @@ +package weed_server + +import ( + "context" + "sort" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/filer" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/stretchr/testify/assert" + "google.golang.org/grpc/metadata" +) + +// ScanEntries makes the rename test store an EntryScanner: every entry at and +// below dirPath, straight from the map, like a store reading one snapshot. +func (s *renameTestStore) ScanEntries(ctx context.Context, dirPath util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + s.mu.Lock() + var paths []string + for p := range s.entries { + if p == "/" { + continue + } + if dirPath == "/" || p == string(dirPath) || len(p) > len(dirPath) && p[:len(dirPath)+1] == string(dirPath)+"/" { + paths = append(paths, p) + } + } + s.mu.Unlock() + // Deliberately not top-down: children before their parents. + sort.Sort(sort.Reverse(sort.StringSlice(paths))) + for _, p := range paths { + s.mu.Lock() + e := s.entries[p] + s.mu.Unlock() + if err := fn(ctx, e); err != nil { + return err + } + } + return nil +} + +var _ filer.EntryScanner = (*renameTestStore)(nil) + +// headerTraverseStream records the response header along with the entries. +type headerTraverseStream struct { + captureTraverseStream + headers []metadata.MD +} + +func (s *headerTraverseStream) SendHeader(md metadata.MD) error { + s.headers = append(s.headers, md) + return nil +} + +func snapshotRequestCtx() context.Context { + return metadata.NewIncomingContext(context.Background(), metadata.Pairs(filer_pb.TraverseModeHeaderKey, filer_pb.TraverseModeSnapshot)) +} + +func newTraverseTestStore() *renameTestStore { + store := newRenameTestStore() + for _, dir := range []string{"/", "/a", "/a/sub", "/b", "/.system"} { + store.entries[dir] = newDirectoryEntry(dir, 1) + } + for _, file := range []string{"/a/f1", "/a/sub/f2", "/b/f3", "/.system/secret"} { + store.entries[file] = newFileEntry(file, 2) + } + return store +} + +func TestTraverseBfsMetadataSnapshotMode(t *testing.T) { + store := newTraverseTestStore() + server := &FilerServer{filer: newRenameTestFiler(t, store)} + stream := &headerTraverseStream{captureTraverseStream: captureTraverseStream{ctx: snapshotRequestCtx()}} + + err := server.TraverseBfsMetadata(&filer_pb.TraverseBfsMetadataRequest{ + Directory: "/", + ExcludedPrefixes: []string{"/.system"}, + }, stream) + assert.NoError(t, err) + + if assert.Len(t, stream.headers, 1) { + assert.Equal(t, []string{filer_pb.TraverseModeSnapshot}, stream.headers[0].Get(filer_pb.TraverseModeHeaderKey)) + } + // The root comes first so a consumer knows what it asked for; everything + // else is complete but in the store's order, with the exclusion honored. + assert.Equal(t, "/", stream.visited[0]) + assert.ElementsMatch(t, []string{"/", "/a", "/a/f1", "/a/sub", "/a/sub/f2", "/b", "/b/f3"}, stream.visited) + + // No per-entry commit timestamp lookup in snapshot mode: entry_ts_ns is + // left "unrecorded" (0), which consumers already treat as "use the mtime". + for path, tsNs := range stream.entryTsNs { + assert.Zerof(t, tsNs, "entry_ts_ns for %s", path) + } +} + +// Without the request metadata the walk is unchanged: no header, parents first. +func TestTraverseBfsMetadataDefaultStillWalks(t *testing.T) { + store := newTraverseTestStore() + server := &FilerServer{filer: newRenameTestFiler(t, store)} + stream := &headerTraverseStream{captureTraverseStream: captureTraverseStream{ctx: context.Background()}} + + err := server.TraverseBfsMetadata(&filer_pb.TraverseBfsMetadataRequest{Directory: "/"}, stream) + assert.NoError(t, err) + assert.Empty(t, stream.headers) + seen := make(map[string]bool) + for _, path := range stream.visited { + if path != "/" { + parent, _ := util.FullPath(path).DirAndName() + assert.Truef(t, seen[parent], "parent %s of %s not streamed first", parent, path) + } + seen[path] = true + } +} + +// A store that cannot scan (here: path-specific stores are configured, so one +// snapshot per store is not one snapshot) makes the filer walk and say so. +func TestTraverseBfsMetadataSnapshotRequestedButUnsupported(t *testing.T) { + store := newTraverseTestStore() + f := newRenameTestFiler(t, store) + f.Store.AddPathSpecificStore("/elsewhere", "elsewhere", newRenameTestStore()) + server := &FilerServer{filer: f} + stream := &headerTraverseStream{captureTraverseStream: captureTraverseStream{ctx: snapshotRequestCtx()}} + + err := server.TraverseBfsMetadata(&filer_pb.TraverseBfsMetadataRequest{Directory: "/a"}, stream) + assert.NoError(t, err) + if assert.Len(t, stream.headers, 1) { + assert.Equal(t, []string{filer_pb.TraverseModeBfs}, stream.headers[0].Get(filer_pb.TraverseModeHeaderKey)) + } + assert.Equal(t, []string{"/a", "/a/f1", "/a/sub", "/a/sub/f2"}, stream.visited) +} + +type fakeScanner struct { + entries []*filer.Entry + err error // returned after the entries +} + +func (s *fakeScanner) ScanEntries(ctx context.Context, dirPath util.FullPath, fn func(ctx context.Context, entry *filer.Entry) error) error { + for _, e := range s.entries { + if err := fn(ctx, e); err != nil { + return err + } + } + return s.err +} + +type traverseRecorder struct { + headers []metadata.MD + sent []string +} + +func (r *traverseRecorder) sendHeader(md metadata.MD) error { + r.headers = append(r.headers, md) + return nil +} +func (r *traverseRecorder) sendEntry(e *filer.Entry) error { + r.sent = append(r.sent, string(e.FullPath)) + return nil +} + +func noExclusion(util.FullPath) bool { return false } + +// A store that only decides it cannot scan after yielding rows must not make +// the stream fall back to a walk: entries are already out under the snapshot header. +func TestTraverseFromSnapshotLateUnsupportedIsAnError(t *testing.T) { + rec := &traverseRecorder{} + scanner := &fakeScanner{entries: []*filer.Entry{newFileEntry("/ckpt/x", 2)}, err: filer.ErrUnsupportedEntryScan} + served, err := traverseFromSnapshot(context.Background(), scanner, newDirectoryEntry("/ckpt", 1), noExclusion, rec.sendHeader, rec.sendEntry) + assert.Error(t, err) + assert.True(t, served) +} diff --git a/weed/shell/command_fs_meta_save.go b/weed/shell/command_fs_meta_save.go index da88bf34e7..b936252f62 100644 --- a/weed/shell/command_fs_meta_save.go +++ b/weed/shell/command_fs_meta_save.go @@ -20,6 +20,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" ) func init() { @@ -152,6 +155,133 @@ func (c *commandFsMetaSave) Do(args []string, commandEnv *CommandEnv, writer io. // log; volume.fsck and fs.verify must leave it false so the live log-file // chunks are accounted for — otherwise fsck reports them as orphans and could // delete referenced needles. +// doTraverseAndSaving feeds every entry at and below path through genFn and +// saveFn: from one snapshot of the filer store when the filer can provide it +// (so a file renamed during the traversal is not missed), by walking otherwise. +func doTraverseAndSaving(filerClient filer_pb.FilerClient, writer io.Writer, path string, verbose bool, skipSystemLog bool, genFn func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) error, saveFn func(outputChan chan interface{}) error) error { + if used, err := doTraverseSnapshotAndSaving(filerClient, writer, path, verbose, skipSystemLog, genFn, saveFn); used || err != nil { + return err + } + return doTraverseBfsAndSaving(filerClient, writer, path, verbose, skipSystemLog, genFn, saveFn) +} + +// doTraverseSnapshotAndSaving asks the filer to traverse path itself and +// consumes the stream only when the filer serves it from a store snapshot +// (filer_pb.TraverseModeHeaderKey). Otherwise it returns used=false having +// read nothing, and the caller walks as before. +func doTraverseSnapshotAndSaving(filerClient filer_pb.FilerClient, writer io.Writer, path string, verbose bool, skipSystemLog bool, genFn func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) error, saveFn func(outputChan chan interface{}) error) (used bool, err error) { + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + var excludedPrefixes []string + if skipSystemLog { + excludedPrefixes = append(excludedPrefixes, filer.SystemLogDir) + } + + // The walk tolerates a path that does not exist (it saves nothing); the + // filer-side traversal reports it as an error, so let the walk have it. + // Any other failure must stop here: a walk that could then reach the filer + // reads renamed-away entries as absent, and fsck would purge their chunks. + if _, _, _, getErr := filer_pb.GetEntry(ctx, filerClient, util.FullPath(path)); getErr != nil { + if errors.Is(getErr, filer_pb.ErrNotFound) { + return false, nil + } + return false, fmt.Errorf("traverse %s: %w", path, getErr) + } + + err = filerClient.WithFilerClient(true, func(client filer_pb.SeaweedFilerClient) error { + ctx := metadata.AppendToOutgoingContext(ctx, filer_pb.TraverseModeHeaderKey, filer_pb.TraverseModeSnapshot) + stream, err := client.TraverseBfsMetadata(ctx, &filer_pb.TraverseBfsMetadataRequest{ + Directory: path, + ExcludedPrefixes: excludedPrefixes, + }) + if err != nil { + return err + } + header, err := stream.Header() + if err != nil { + if status.Code(err) == codes.Unimplemented { + return nil // a filer without the RPC at all: walk + } + return fmt.Errorf("traverse %s: %w", path, err) + } + if modes := header.Get(filer_pb.TraverseModeHeaderKey); len(modes) == 0 || modes[0] != filer_pb.TraverseModeSnapshot { + // An older filer (no header), a store that cannot scan ("bfs"), or + // a stream that failed before any header, which Header() reports + // as no header and nil: only Recv tells, and a failure is no + // licence to walk instead. + if _, recvErr := stream.Recv(); recvErr != nil && !errors.Is(recvErr, io.EOF) && status.Code(recvErr) != codes.Unimplemented { + return fmt.Errorf("traverse %s: %w", path, recvErr) + } + return nil + } + used = true + + var wg sync.WaitGroup + wg.Add(1) + outputChan := make(chan interface{}, 1024) + saveErrChan := make(chan error, 1) + go func() { + saveErr := saveFn(outputChan) + saveErrChan <- saveErr + if saveErr != nil { + cancel() + } + wg.Done() + }() + + var dirCount, fileCount uint64 + var recvErr error + for { + resp, err := stream.Recv() + if err != nil { + if !errors.Is(err, io.EOF) { + recvErr = fmt.Errorf("traverse %s: %w", path, err) + } + break + } + if resp.Entry == nil { + continue + } + protoMessage := &filer_pb.FullEntry{ + Dir: resp.Directory, + Entry: resp.Entry, + } + if genErr := genFn(ctx, protoMessage, outputChan); genErr != nil { + recvErr = genErr + cancel() + break + } + if resp.Entry.IsDirectory { + dirCount++ + } else { + fileCount++ + } + if verbose { + println(util.NewFullPath(resp.Directory, resp.Entry.Name)) + } + } + + close(outputChan) + wg.Wait() + saveErr := <-saveErrChan + + if recvErr != nil { + return recvErr + } + if saveErr != nil { + return saveErr + } + if writer != nil { + fmt.Fprintf(writer, "collected filer entries from a consistent store snapshot\n") + fmt.Fprintf(writer, "total %d directories, %d files\n", dirCount, fileCount) + } + return nil + }) + return used, err +} + func doTraverseBfsAndSaving(filerClient filer_pb.FilerClient, writer io.Writer, path string, verbose bool, skipSystemLog bool, genFn func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) error, saveFn func(outputChan chan interface{}) error) error { ctx, cancel := context.WithCancel(context.Background()) diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index 63981282b6..e46f01d6df 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -99,6 +99,12 @@ func (c *commandVolumeFsck) Help() string { A newly uploaded chunk may appear as orphan if metadata commit is still pending. 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. + ` } @@ -322,7 +328,7 @@ func (c *commandVolumeFsck) collectFilerFileIdAndPaths(dataNodeVolumeIdToVInfo m } }() - return doTraverseBfsAndSaving(c.env, c.writer, c.getCollectFilerFilePath(), false, false, + return doTraverseAndSaving(c.env, c.writer, c.getCollectFilerFilePath(), false, false, func(ctx context.Context, entry *filer_pb.FullEntry, outputChan chan interface{}) (err error) { if *c.verbose && entry.Entry.IsDirectory { fmt.Fprintf(c.writer, "checking directory %s\n", util.NewFullPath(entry.Dir, entry.Entry.Name))