diff options
25 files changed, 658 insertions, 18 deletions
diff --git a/docker/Makefile b/docker/Makefile index 76cdf75c8..b46dcedf1 100644 --- a/docker/Makefile +++ b/docker/Makefile @@ -49,6 +49,9 @@ dev_replicate: build dev_auditlog: build docker-compose -f compose/local-auditlog-compose.yml -p seaweedfs up +dev_ydb: build + docker-compose -f compose/local-ydb-compose.yml -p seaweedfs up + cluster: build docker-compose -f compose/local-cluster-compose.yml -p seaweedfs up diff --git a/docker/compose/local-ydb-compose.yml b/docker/compose/local-ydb-compose.yml new file mode 100644 index 000000000..33f550600 --- /dev/null +++ b/docker/compose/local-ydb-compose.yml @@ -0,0 +1,33 @@ +version: '2' + +services: + ydb: + image: cr.yandex/yc/yandex-docker-local-ydb + ports: + - 2135:2135 + - 8765:8765 + - 2136:2136 + environment: + - YDB_DEFAULT_LOG_LEVEL=DEBUG + - GRPC_TLS_PORT=2135 + - GRPC_PORT=2136 + - MON_PORT=8765 + server: + image: chrislusf/seaweedfs:local + ports: + - 9333:9333 + - 19333:19333 + - 8084:8080 + - 18084:18080 + - 8888:8888 + - 18888:18888 + command: "server -ip=server -filer -volume.max=0 -master.volumeSizeLimitMB=1024 -volume.preStopSeconds=1" + volumes: + - ./master-cloud.toml:/etc/seaweedfs/master.toml + environment: + - WEED_LEVELDB2_ENABLED=false + - WEED_YDB_ENABLED=true + - WEED_YDB_DSN=grpc://ydb:2136/?database=local + - YDB_ANONYMOUS_CREDENTIALS=1 + depends_on: + - ydb
\ No newline at end of file diff --git a/weed/command/imports.go b/weed/command/imports.go index 5b3195907..04079b162 100644 --- a/weed/command/imports.go +++ b/weed/command/imports.go @@ -32,4 +32,5 @@ import ( _ "github.com/chrislusf/seaweedfs/weed/filer/redis2" _ "github.com/chrislusf/seaweedfs/weed/filer/redis3" _ "github.com/chrislusf/seaweedfs/weed/filer/sqlite" + _ "github.com/chrislusf/seaweedfs/weed/filer/ydb" ) diff --git a/weed/command/scaffold/filer.toml b/weed/command/scaffold/filer.toml index 0a505bbdc..3a920d80b 100644 --- a/weed/command/scaffold/filer.toml +++ b/weed/command/scaffold/filer.toml @@ -295,6 +295,17 @@ password="" # skip tls cert validation insecure_skip_verify = true +[ydb] +enabled = false +useBucketPrefix=true +dsn="grpc://localhost:2136?database=/local" +prefix="en" +poolSizeLimit=50 +# Authenticate produced with one of next environment variables: +# YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS=<path/to/sa_key_file> — used service account key file by path +# YDB_ANONYMOUS_CREDENTIALS="1" — used for authenticate with anonymous access. Anonymous access needs for connect to testing YDB installation +# YDB_METADATA_CREDENTIALS="1" — used metadata service for authenticate to YDB from yandex cloud virtual machine or from yandex function +# YDB_ACCESS_TOKEN_CREDENTIALS=<access_token> — used for authenticate to YDB with short-life access token. For example, access token may be IAM token ########################## ########################## diff --git a/weed/filer/abstract_sql/abstract_sql_store.go b/weed/filer/abstract_sql/abstract_sql_store.go index 4bf9b16fa..13268b944 100644 --- a/weed/filer/abstract_sql/abstract_sql_store.go +++ b/weed/filer/abstract_sql/abstract_sql_store.go @@ -156,7 +156,7 @@ func (store *AbstractSqlStore) InsertEntry(ctx context.Context, entry *filer.Ent return fmt.Errorf("encode %s: %s", entry.FullPath, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { meta = util.MaybeGzipData(meta) } diff --git a/weed/filer/abstract_sql/abstract_sql_store_kv.go b/weed/filer/abstract_sql/abstract_sql_store_kv.go index 03b016c76..aaf1c196c 100644 --- a/weed/filer/abstract_sql/abstract_sql_store_kv.go +++ b/weed/filer/abstract_sql/abstract_sql_store_kv.go @@ -18,7 +18,7 @@ func (store *AbstractSqlStore) KvPut(ctx context.Context, key []byte, value []by return fmt.Errorf("findDB: %v", err) } - dirStr, dirHash, name := genDirAndName(key) + dirStr, dirHash, name := GenDirAndName(key) res, err := db.ExecContext(ctx, store.GetSqlInsert(DEFAULT_TABLE), dirHash, name, dirStr, value) if err == nil { @@ -53,7 +53,7 @@ func (store *AbstractSqlStore) KvGet(ctx context.Context, key []byte) (value []b return nil, fmt.Errorf("findDB: %v", err) } - dirStr, dirHash, name := genDirAndName(key) + dirStr, dirHash, name := GenDirAndName(key) row := db.QueryRowContext(ctx, store.GetSqlFind(DEFAULT_TABLE), dirHash, name, dirStr) err = row.Scan(&value) @@ -76,7 +76,7 @@ func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err er return fmt.Errorf("findDB: %v", err) } - dirStr, dirHash, name := genDirAndName(key) + dirStr, dirHash, name := GenDirAndName(key) res, err := db.ExecContext(ctx, store.GetSqlDelete(DEFAULT_TABLE), dirHash, name, dirStr) if err != nil { @@ -92,7 +92,7 @@ func (store *AbstractSqlStore) KvDelete(ctx context.Context, key []byte) (err er } -func genDirAndName(key []byte) (dirStr string, dirHash int64, name string) { +func GenDirAndName(key []byte) (dirStr string, dirHash int64, name string) { for len(key) < 8 { key = append(key, 0) } diff --git a/weed/filer/arangodb/arangodb_store.go b/weed/filer/arangodb/arangodb_store.go index 9fd1fffb3..13d14b2b0 100644 --- a/weed/filer/arangodb/arangodb_store.go +++ b/weed/filer/arangodb/arangodb_store.go @@ -157,7 +157,7 @@ func (store *ArangodbStore) InsertEntry(ctx context.Context, entry *filer.Entry) return fmt.Errorf("encode %s: %s", entry.FullPath, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { meta = util.MaybeGzipData(meta) } model := &Model{ @@ -196,7 +196,7 @@ func (store *ArangodbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) return fmt.Errorf("encode %s: %s", entry.FullPath, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { meta = util.MaybeGzipData(meta) } model := &Model{ diff --git a/weed/filer/cassandra/cassandra_store.go b/weed/filer/cassandra/cassandra_store.go index fb61b0771..d8c094a45 100644 --- a/weed/filer/cassandra/cassandra_store.go +++ b/weed/filer/cassandra/cassandra_store.go @@ -100,7 +100,7 @@ func (store *CassandraStore) InsertEntry(ctx context.Context, entry *filer.Entry return fmt.Errorf("encode %s: %s", entry.FullPath, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { meta = util.MaybeGzipData(meta) } diff --git a/weed/filer/etcd/etcd_store.go b/weed/filer/etcd/etcd_store.go index 2a5dfc926..4146a3899 100644 --- a/weed/filer/etcd/etcd_store.go +++ b/weed/filer/etcd/etcd_store.go @@ -82,7 +82,7 @@ func (store *EtcdStore) InsertEntry(ctx context.Context, entry *filer.Entry) (er return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { meta = weed_util.MaybeGzipData(meta) } diff --git a/weed/filer/filerstore.go b/weed/filer/filerstore.go index a092ee456..260945b33 100644 --- a/weed/filer/filerstore.go +++ b/weed/filer/filerstore.go @@ -7,6 +7,8 @@ import ( "io" ) +const CountEntryChunksForGzip = 50 + var ( ErrUnsupportedListDirectoryPrefixed = errors.New("unsupported directory prefix listing") ErrUnsupportedSuperLargeDirectoryListing = errors.New("unsupported super large directory listing") diff --git a/weed/filer/hbase/hbase_store.go b/weed/filer/hbase/hbase_store.go index e0d878ca7..c5d6eb48c 100644 --- a/weed/filer/hbase/hbase_store.go +++ b/weed/filer/hbase/hbase_store.go @@ -75,7 +75,7 @@ func (store *HbaseStore) InsertEntry(ctx context.Context, entry *filer.Entry) er if err != nil { return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = util.MaybeGzipData(value) } diff --git a/weed/filer/leveldb/leveldb_store.go b/weed/filer/leveldb/leveldb_store.go index 73d757e62..6abb37f99 100644 --- a/weed/filer/leveldb/leveldb_store.go +++ b/weed/filer/leveldb/leveldb_store.go @@ -86,7 +86,7 @@ func (store *LevelDBStore) InsertEntry(ctx context.Context, entry *filer.Entry) return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = weed_util.MaybeGzipData(value) } diff --git a/weed/filer/leveldb2/leveldb2_store.go b/weed/filer/leveldb2/leveldb2_store.go index 966686ed9..d68493bd7 100644 --- a/weed/filer/leveldb2/leveldb2_store.go +++ b/weed/filer/leveldb2/leveldb2_store.go @@ -88,7 +88,7 @@ func (store *LevelDB2Store) InsertEntry(ctx context.Context, entry *filer.Entry) return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = weed_util.MaybeGzipData(value) } diff --git a/weed/filer/leveldb3/leveldb3_store.go b/weed/filer/leveldb3/leveldb3_store.go index e448f0093..d21515bd4 100644 --- a/weed/filer/leveldb3/leveldb3_store.go +++ b/weed/filer/leveldb3/leveldb3_store.go @@ -177,7 +177,7 @@ func (store *LevelDB3Store) InsertEntry(ctx context.Context, entry *filer.Entry) return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = weed_util.MaybeGzipData(value) } diff --git a/weed/filer/mongodb/mongodb_store.go b/weed/filer/mongodb/mongodb_store.go index c12354ad6..83686bfe7 100644 --- a/weed/filer/mongodb/mongodb_store.go +++ b/weed/filer/mongodb/mongodb_store.go @@ -107,7 +107,7 @@ func (store *MongodbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) return fmt.Errorf("encode %s: %s", entry.FullPath, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { meta = util.MaybeGzipData(meta) } diff --git a/weed/filer/redis/universal_redis_store.go b/weed/filer/redis/universal_redis_store.go index 0cdf58d7f..89684647b 100644 --- a/weed/filer/redis/universal_redis_store.go +++ b/weed/filer/redis/universal_redis_store.go @@ -40,7 +40,7 @@ func (store *UniversalRedisStore) InsertEntry(ctx context.Context, entry *filer. return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = util.MaybeGzipData(value) } diff --git a/weed/filer/redis2/universal_redis_store.go b/weed/filer/redis2/universal_redis_store.go index deccf8922..7a34092a0 100644 --- a/weed/filer/redis2/universal_redis_store.go +++ b/weed/filer/redis2/universal_redis_store.go @@ -52,7 +52,7 @@ func (store *UniversalRedis2Store) InsertEntry(ctx context.Context, entry *filer return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = util.MaybeGzipData(value) } diff --git a/weed/filer/redis3/universal_redis_store.go b/weed/filer/redis3/universal_redis_store.go index f04ee493d..10a87e2a4 100644 --- a/weed/filer/redis3/universal_redis_store.go +++ b/weed/filer/redis3/universal_redis_store.go @@ -40,7 +40,7 @@ func (store *UniversalRedis3Store) InsertEntry(ctx context.Context, entry *filer return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = util.MaybeGzipData(value) } diff --git a/weed/filer/redis_lua/universal_redis_store.go b/weed/filer/redis_lua/universal_redis_store.go index 9674ac03f..0ab0f2f24 100644 --- a/weed/filer/redis_lua/universal_redis_store.go +++ b/weed/filer/redis_lua/universal_redis_store.go @@ -53,7 +53,7 @@ func (store *UniversalRedisLuaStore) InsertEntry(ctx context.Context, entry *fil return fmt.Errorf("encoding %s %+v: %v", entry.FullPath, entry.Attr, err) } - if len(entry.Chunks) > 50 { + if len(entry.Chunks) > filer.CountEntryChunksForGzip { value = util.MaybeGzipData(value) } diff --git a/weed/filer/ydb/readme.md b/weed/filer/ydb/readme.md new file mode 100644 index 000000000..fd6c5c0a2 --- /dev/null +++ b/weed/filer/ydb/readme.md @@ -0,0 +1,26 @@ +## YDB + +database: https://github.com/ydb-platform/ydb + +go driver: https://github.com/ydb-platform/ydb-go-sdk + +options: + +``` +[ydb] +enabled=true +dsn=grpcs://ydb-ru.yandex.net:2135/?database=/ru/home/username/db +prefix="seaweedfs" +useBucketPrefix=true +poolSizeLimit=50 +``` + +Authenticate produced with one of next environment variables: + * `YDB_SERVICE_ACCOUNT_KEY_FILE_CREDENTIALS=<path/to/sa_key_file>` — used service account key file by path + * `YDB_ANONYMOUS_CREDENTIALS="1"` — used for authenticate with anonymous access. Anonymous access needs for connect to testing YDB installation + * `YDB_METADATA_CREDENTIALS="1"` — used metadata service for authenticate to YDB from yandex cloud virtual machine or from yandex function + * `YDB_ACCESS_TOKEN_CREDENTIALS=<access_token>` — used for authenticate to YDB with short-life access token. For example, access token may be IAM token + * `YDB_CONNECTION_STRING="grpcs://endpoint/?database=database"` + + * i test using this dev database: +`make dev_ydb` diff --git a/weed/filer/ydb/ydb_queries.go b/weed/filer/ydb/ydb_queries.go new file mode 100644 index 000000000..659cc2158 --- /dev/null +++ b/weed/filer/ydb/ydb_queries.go @@ -0,0 +1,64 @@ +package ydb + +import asql "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql" + +const ( + insertQuery = ` + DECLARE $dir_hash AS int64; + DECLARE $name AS Utf8; + DECLARE $directory AS Utf8; + DECLARE $meta AS String; + + UPSERT INTO ` + asql.DEFAULT_TABLE + ` + (dir_hash, name, directory, meta) + VALUES + ($dir_hash, $name, $directory, $meta);` + + updateQuery = ` + DECLARE $dir_hash AS int64; + DECLARE $name AS Utf8; + DECLARE $directory AS Utf8; + DECLARE $meta AS String; + + REPLACE INTO ` + asql.DEFAULT_TABLE + ` + (dir_hash, name, directory, meta) + VALUES + ($dir_hash, $name, $directory, $meta) + COMMIT;` + + deleteQuery = ` + DECLARE $dir_hash AS int64; + DECLARE $name AS Utf8; + + DELETE FROM ` + asql.DEFAULT_TABLE + ` + WHERE dir_hash == $dir_hash AND name == $name; + COMMIT;` + + findQuery = ` + DECLARE $dir_hash AS int64; + DECLARE $name AS Utf8; + + SELECT meta + FROM ` + asql.DEFAULT_TABLE + ` + WHERE dir_hash == $dir_hash AND name == $name;` + + deleteFolderChildrenQuery = ` + DECLARE $dir_hash AS int64; + DECLARE $directory AS Utf8; + + DELETE FROM ` + asql.DEFAULT_TABLE + ` + WHERE dir_hash == $dir_hash AND directory == $directory; + COMMIT;` + + listDirectoryQuery = ` + DECLARE $dir_hash AS int64; + DECLARE $directory AS Utf8; + DECLARE $start_name AS Utf8; + DECLARE $prefix AS Utf8; + DECLARE $limit AS Uint64; + + SELECT name, meta + FROM ` + asql.DEFAULT_TABLE + ` + WHERE dir_hash == $dir_hash AND directory == $directory and name %s $start_name and name LIKE $prefix + ORDER BY name ASC LIMIT $limit;` +) diff --git a/weed/filer/ydb/ydb_store.go b/weed/filer/ydb/ydb_store.go new file mode 100644 index 000000000..6d6389dde --- /dev/null +++ b/weed/filer/ydb/ydb_store.go @@ -0,0 +1,380 @@ +package ydb + +import ( + "context" + "fmt" + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql" + "github.com/chrislusf/seaweedfs/weed/glog" + "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" + "github.com/chrislusf/seaweedfs/weed/util" + environ "github.com/ydb-platform/ydb-go-sdk-auth-environ" + "github.com/ydb-platform/ydb-go-sdk/v3" + "github.com/ydb-platform/ydb-go-sdk/v3/sugar" + "github.com/ydb-platform/ydb-go-sdk/v3/table" + "github.com/ydb-platform/ydb-go-sdk/v3/table/result" + "github.com/ydb-platform/ydb-go-sdk/v3/table/result/named" + "github.com/ydb-platform/ydb-go-sdk/v3/table/types" + "os" + "path" + "strings" + "sync" + "time" +) + +const ( + defaultConnectionTimeOut = 10 +) + +var ( + roTX = table.TxControl( + table.BeginTx(table.WithOnlineReadOnly()), + table.CommitTx(), + ) + rwTX = table.DefaultTxControl() +) + +type YdbStore struct { + DB ydb.Connection + dirBuckets string + tablePathPrefix string + SupportBucketTable bool + dbs map[string]bool + dbsLock sync.Mutex +} + +func init() { + filer.Stores = append(filer.Stores, &YdbStore{}) +} + +func (store *YdbStore) GetName() string { + return "ydb" +} + +func (store *YdbStore) Initialize(configuration util.Configuration, prefix string) (err error) { + return store.initialize( + configuration.GetString("filer.options.buckets_folder"), + configuration.GetString(prefix+"dsn"), + configuration.GetString(prefix+"tablePathPrefix"), + configuration.GetBool(prefix+"useBucketPrefix"), + configuration.GetInt(prefix+"connectionTimeOut"), + configuration.GetInt(prefix+"poolSizeLimit"), + ) +} + +func (store *YdbStore) initialize(dirBuckets string, dsn string, tablePathPrefix string, useBucketPrefix bool, connectionTimeOut int, poolSizeLimit int) (err error) { + store.dirBuckets = dirBuckets + store.tablePathPrefix = tablePathPrefix + store.SupportBucketTable = useBucketPrefix + store.dbs = make(map[string]bool) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if connectionTimeOut == 0 { + connectionTimeOut = defaultConnectionTimeOut + } + opts := []ydb.Option{ + ydb.WithDialTimeout(time.Duration(connectionTimeOut) * time.Second), + environ.WithEnvironCredentials(ctx), + } + if poolSizeLimit > 0 { + opts = append(opts, ydb.WithSessionPoolSizeLimit(poolSizeLimit)) + } + if dsn == "" { + dsn = os.Getenv("YDB_CONNECTION_STRING") + } + store.DB, err = ydb.Open(ctx, dsn, opts...) + if err != nil { + if store.DB != nil { + _ = store.DB.Close(ctx) + store.DB = nil + } + return fmt.Errorf("can not connect to %s error: %v", dsn, err) + } + + store.tablePathPrefix = path.Join(store.DB.Name(), tablePathPrefix) + if err = sugar.RemoveRecursive(ctx, store.DB, store.tablePathPrefix); err != nil { + return fmt.Errorf("RemoveRecursive %s : %v", store.tablePathPrefix, err) + } + if err = sugar.MakeRecursive(ctx, store.DB, store.tablePathPrefix); err != nil { + return fmt.Errorf("MakeRecursive %s : %v", store.tablePathPrefix, err) + } + + tablePath := path.Join(store.tablePathPrefix, abstract_sql.DEFAULT_TABLE) + if err = store.createTable(ctx, tablePath); err != nil { + glog.Errorf("createTable %s: %v", tablePath, err) + } + return err +} + +func (store *YdbStore) doTxOrDB(ctx context.Context, query *string, params *table.QueryParameters, tc *table.TransactionControl, processResultFunc func(res result.Result) error) (err error) { + var res result.Result + if tx, ok := ctx.Value("tx").(table.Transaction); ok { + res, err = tx.Execute(ctx, *query, params) + if err != nil { + return fmt.Errorf("execute transaction: %v", err) + } + } else { + err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, *query) + if err != nil { + return fmt.Errorf("prepare: %v", err) + } + _, res, err = stmt.Execute(ctx, tc, params) + if err != nil { + return fmt.Errorf("execute statement: %v", err) + } + return nil + }) + } + if err != nil && processResultFunc != nil && res != nil { + if err = processResultFunc(res); err != nil { + return fmt.Errorf("process resul: %v", err) + } + } + return err +} + +func (store *YdbStore) insertOrUpdateEntry(ctx context.Context, entry *filer.Entry, query string) (err error) { + dir, name := entry.FullPath.DirAndName() + meta, err := entry.EncodeAttributesAndChunks() + if err != nil { + return fmt.Errorf("encode %s: %s", entry.FullPath, err) + } + + if len(entry.Chunks) > filer.CountEntryChunksForGzip { + meta = util.MaybeGzipData(meta) + } + queryWithPragma := withPragma(store.getPrefix(ctx, dir), query) + fileMeta := FileMeta{util.HashStringToLong(dir), name, dir, meta} + return store.doTxOrDB(ctx, &queryWithPragma, fileMeta.queryParameters(), rwTX, nil) +} + +func (store *YdbStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { + return store.insertOrUpdateEntry(ctx, entry, insertQuery) +} + +func (store *YdbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err error) { + return store.insertOrUpdateEntry(ctx, entry, updateQuery) +} + +func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) { + dir, name := fullpath.DirAndName() + var data []byte + entryFound := false + queryWithPragma := withPragma(store.getPrefix(ctx, dir), findQuery) + queryParams := table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$name", types.UTF8Value(name))) + + err = store.doTxOrDB(ctx, &queryWithPragma, queryParams, roTX, func(res result.Result) error { + defer func() { + _ = res.Close() + }() + for res.NextRow() { + if err := res.ScanNamed(named.Required("meta", &data)); err != nil { + return fmt.Errorf("scanNamed %s : %v", entry.FullPath, err) + } + entryFound = true + return nil + } + return res.Err() + }) + if err != nil { + return nil, err + } + if !entryFound { + return nil, filer_pb.ErrNotFound + } + entry.FullPath = fullpath + if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil { + return nil, fmt.Errorf("decode %s : %v", entry.FullPath, err) + } + + return entry, nil +} + +func (store *YdbStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) { + dir, name := fullpath.DirAndName() + queryWithPragma := withPragma(store.getPrefix(ctx, dir), deleteQuery) + queryParams := table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$name", types.UTF8Value(name))) + + return store.doTxOrDB(ctx, &queryWithPragma, queryParams, rwTX, nil) +} + +func (store *YdbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) { + dir, _ := fullpath.DirAndName() + queryWithPragma := withPragma(store.getPrefix(ctx, dir), deleteFolderChildrenQuery) + queryParams := table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$directory", types.UTF8Value(dir))) + + return store.doTxOrDB(ctx, &queryWithPragma, queryParams, rwTX, nil) +} + +func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { + return store.ListDirectoryPrefixedEntries(ctx, dirPath, startFileName, includeStartFile, limit, "", nil) +} + +func (store *YdbStore) ListDirectoryPrefixedEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, prefix string, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { + dir := string(dirPath) + startFileCompOp := ">" + if includeStartFile { + startFileCompOp = ">=" + } + queryWithPragma := withPragma(store.getPrefix(ctx, dir), fmt.Sprintf(listDirectoryQuery, startFileCompOp)) + queryParams := table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$directory", types.UTF8Value(dir)), + table.ValueParam("$start_name", types.UTF8Value(startFileName)), + table.ValueParam("$prefix", types.UTF8Value(prefix+"%")), + table.ValueParam("$limit", types.Uint64Value(uint64(limit))), + ) + err = store.doTxOrDB(ctx, &queryWithPragma, queryParams, roTX, func(res result.Result) error { + defer func() { + _ = res.Close() + }() + for res.NextResultSet(ctx) { + for res.NextRow() { + var name string + var data []byte + if err := res.ScanNamed( + named.Required("name", &name), + named.Required("meta", &data)); err != nil { + return fmt.Errorf("scanNamed %s : %v", dir, err) + } + lastFileName = name + entry := &filer.Entry{ + FullPath: util.NewFullPath(dir, name), + } + if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil { + return fmt.Errorf("scan decode %s : %v", entry.FullPath, err) + } + if !eachEntryFunc(entry) { + break + } + } + } + return res.Err() + }) + if err != nil { + return lastFileName, err + } + return lastFileName, nil +} + +func (store *YdbStore) BeginTransaction(ctx context.Context) (context.Context, error) { + session, err := store.DB.Table().CreateSession(ctx) + if err != nil { + return ctx, err + } + tx, err := session.BeginTransaction(ctx, table.TxSettings(table.WithSerializableReadWrite())) + if err != nil { + return ctx, err + } + return context.WithValue(ctx, "tx", tx), nil +} + +func (store *YdbStore) CommitTransaction(ctx context.Context) error { + if tx, ok := ctx.Value("tx").(table.Transaction); ok { + _, err := tx.CommitTx(ctx) + return err + } + return nil +} + +func (store *YdbStore) RollbackTransaction(ctx context.Context) error { + if tx, ok := ctx.Value("tx").(table.Transaction); ok { + return tx.Rollback(ctx) + } + return nil +} + +func (store *YdbStore) Shutdown() { + _ = store.DB.Close(context.Background()) +} + +func (store *YdbStore) CanDropWholeBucket() bool { + return store.SupportBucketTable +} + +func (store *YdbStore) OnBucketCreation(bucket string) { + store.dbsLock.Lock() + defer store.dbsLock.Unlock() + + if err := store.createTable(context.Background(), + path.Join(store.tablePathPrefix, bucket, abstract_sql.DEFAULT_TABLE)); err != nil { + glog.Errorf("createTable %s: %v", bucket, err) + } + + if store.dbs == nil { + return + } + store.dbs[bucket] = true +} + +func (store *YdbStore) OnBucketDeletion(bucket string) { + store.dbsLock.Lock() + defer store.dbsLock.Unlock() + + if err := store.deleteTable(context.Background(), + path.Join(store.tablePathPrefix, bucket, abstract_sql.DEFAULT_TABLE)); err != nil { + glog.Errorf("deleteTable %s: %v", bucket, err) + } + + if store.dbs == nil { + return + } + delete(store.dbs, bucket) +} + +func (store *YdbStore) createTable(ctx context.Context, prefix string) error { + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + return s.CreateTable(ctx, prefix, createTableOptions()...) + }) +} + +func (store *YdbStore) deleteTable(ctx context.Context, prefix string) error { + if !store.SupportBucketTable { + return nil + } + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + return s.DropTable(ctx, prefix) + }) +} + +func (store *YdbStore) getPrefix(ctx context.Context, dir string) (tablePathPrefix string) { + tablePathPrefix = store.tablePathPrefix + if !store.SupportBucketTable { + return + } + + prefixBuckets := store.dirBuckets + "/" + if strings.HasPrefix(dir, prefixBuckets) { + // detect bucket + bucketAndDir := dir[len(prefixBuckets):] + t := strings.Index(bucketAndDir, "/") + if t < 0 { + return + } + bucket := bucketAndDir[:t] + + if bucket != "" { + return + } + store.dbsLock.Lock() + defer store.dbsLock.Unlock() + + if _, found := store.dbs[bucket]; !found { + if err := store.createTable(ctx, + path.Join(store.tablePathPrefix, bucket, abstract_sql.DEFAULT_TABLE)); err == nil { + store.dbs[bucket] = true + } else { + glog.Errorf("createTable %s: %v", bucket, err) + } + } + tablePathPrefix = path.Join(store.tablePathPrefix, bucket) + } + return +} diff --git a/weed/filer/ydb/ydb_store_kv.go b/weed/filer/ydb/ydb_store_kv.go new file mode 100644 index 000000000..069c35224 --- /dev/null +++ b/weed/filer/ydb/ydb_store_kv.go @@ -0,0 +1,78 @@ +package ydb + +import ( + "context" + "fmt" + "github.com/chrislusf/seaweedfs/weed/filer" + "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql" + "github.com/chrislusf/seaweedfs/weed/util" + "github.com/ydb-platform/ydb-go-sdk/v3/table" + "github.com/ydb-platform/ydb-go-sdk/v3/table/result/named" + "github.com/ydb-platform/ydb-go-sdk/v3/table/types" +) + +func (store *YdbStore) KvPut(ctx context.Context, key []byte, value []byte) (err error) { + dirStr, dirHash, name := abstract_sql.GenDirAndName(key) + fileMeta := FileMeta{dirHash, name, dirStr, value} + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, withPragma(store.getPrefix(ctx, dirStr), insertQuery)) + if err != nil { + return fmt.Errorf("kv put prepare %s: %v", util.NewFullPath(dirStr, name).Name(), err) + } + _, _, err = stmt.Execute(ctx, rwTX, fileMeta.queryParameters()) + if err != nil { + return fmt.Errorf("kv put execute %s: %v", util.NewFullPath(dirStr, name).Name(), err) + } + return nil + }) +} + +func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { + dirStr, dirHash, name := abstract_sql.GenDirAndName(key) + valueFound := false + err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + stmt, err := s.Prepare(ctx, withPragma(store.getPrefix(ctx, dirStr), findQuery)) + if err != nil { + return fmt.Errorf("kv get prepare %s: %v", util.NewFullPath(dirStr, name), err) + } + _, res, err := stmt.Execute(ctx, roTX, table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(dirHash)), + table.ValueParam("$name", types.UTF8Value(name)))) + if err != nil { + return fmt.Errorf("kv get execute %s: %v", util.NewFullPath(dirStr, name).Name(), err) + } + defer func() { _ = res.Close() }() + for res.NextRow() { + if err := res.ScanNamed(named.Required("meta", &value)); err != nil { + return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name).Name(), err) + } + valueFound = true + return nil + } + return res.Err() + }) + + if !valueFound { + return nil, filer.ErrKvNotFound + } + + return value, nil +} + +func (store *YdbStore) KvDelete(ctx context.Context, key []byte) (err error) { + dirStr, dirHash, name := abstract_sql.GenDirAndName(key) + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, withPragma(store.getPrefix(ctx, dirStr), insertQuery)) + if err != nil { + return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, name).Name(), err) + } + _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(dirHash)), + table.ValueParam("$name", types.UTF8Value(name)))) + if err != nil { + return fmt.Errorf("kv delete %s: %v", util.NewFullPath(dirStr, name).Name(), err) + } + return nil + }) + +} diff --git a/weed/filer/ydb/ydb_types.go b/weed/filer/ydb/ydb_types.go new file mode 100644 index 000000000..07e69e6a5 --- /dev/null +++ b/weed/filer/ydb/ydb_types.go @@ -0,0 +1,41 @@ +package ydb + +import ( + "github.com/ydb-platform/ydb-go-sdk/v3/table" + "github.com/ydb-platform/ydb-go-sdk/v3/table/options" + "github.com/ydb-platform/ydb-go-sdk/v3/table/types" +) + +//go:generate ydbgen + +//ydb:gen +type FileMeta struct { + DirHash int64 `ydb:"type:int64"` + Name string `ydb:"type:utf8"` + Directory string `ydb:"type:utf8"` + Meta []byte `ydb:"type:string"` +} + +//ydb:gen scan,value +type FileMetas []FileMeta + +func (fm *FileMeta) queryParameters() *table.QueryParameters { + return table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(fm.DirHash)), + table.ValueParam("$name", types.UTF8Value(fm.Name)), + table.ValueParam("$directory", types.UTF8Value(fm.Directory)), + table.ValueParam("$meta", types.StringValue(fm.Meta))) +} + +func createTableOptions() []options.CreateTableOption { + return []options.CreateTableOption{ + options.WithColumn("dir_hash", types.Optional(types.TypeInt64)), + options.WithColumn("name", types.Optional(types.TypeUTF8)), + options.WithColumn("directory", types.Optional(types.TypeUTF8)), + options.WithColumn("meta", types.Optional(types.TypeString)), + options.WithPrimaryKeyColumn("dir_hash", "name"), + } +} +func withPragma(prefix string, query string) string { + return `PRAGMA TablePathPrefix("` + prefix + `");` + query +} diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 29baf46ea..82b15084d 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -38,6 +38,7 @@ import ( _ "github.com/chrislusf/seaweedfs/weed/filer/redis2" _ "github.com/chrislusf/seaweedfs/weed/filer/redis3" _ "github.com/chrislusf/seaweedfs/weed/filer/sqlite" + _ "github.com/chrislusf/seaweedfs/weed/filer/ydb" "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/notification" _ "github.com/chrislusf/seaweedfs/weed/notification/aws_sqs" |