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))