diff options
| -rw-r--r-- | go.mod | 12 | ||||
| -rw-r--r-- | go.sum | 19 | ||||
| -rw-r--r-- | weed/command/imports.go | 1 | ||||
| -rw-r--r-- | weed/command/scaffold/filer.toml | 11 | ||||
| -rw-r--r-- | weed/filer/ydb/readme.md | 16 | ||||
| -rw-r--r-- | weed/filer/ydb/ydb_store.go | 263 | ||||
| -rw-r--r-- | weed/filer/ydb/ydb_store_kv.go | 100 | ||||
| -rw-r--r-- | weed/filer/ydb/ydb_types.go | 13 | ||||
| -rw-r--r-- | weed/filer/ydb/ydb_types_ydbgen.go | 194 | ||||
| -rw-r--r-- | weed/server/filer_server.go | 1 |
10 files changed, 252 insertions, 378 deletions
@@ -124,15 +124,15 @@ require ( golang.org/x/crypto v0.0.0-20220411220226-7b82a4e95df4 // indirect golang.org/x/exp v0.0.0-20220414153411-bcd21879b8fd golang.org/x/image v0.0.0-20200119044424-58c23975cae1 - golang.org/x/net v0.0.0-20220412020605-290c469a71a5 + golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 golang.org/x/oauth2 v0.0.0-20220411215720-9780585627b5 // indirect - golang.org/x/sys v0.0.0-20220412211240-33da011f77ad + golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32 golang.org/x/text v0.3.7 // indirect golang.org/x/tools v0.1.8-0.20211029000441-d6a9af8af023 golang.org/x/xerrors v0.0.0-20220411194840-2f41105eb62f // indirect google.golang.org/api v0.75.0 google.golang.org/appengine v1.6.7 // indirect - google.golang.org/genproto v0.0.0-20220414192740-2d67ff6cf2b4 // indirect + google.golang.org/genproto v0.0.0-20220429170224-98d788798c3e // indirect google.golang.org/grpc v1.46.0 google.golang.org/protobuf v1.28.0 gopkg.in/inf.v0 v0.9.1 // indirect @@ -203,8 +203,12 @@ require ( github.com/spf13/pflag v1.0.5 // indirect github.com/subosito/gotenv v1.2.0 // indirect github.com/tinylib/msgp v1.1.6 // indirect - github.com/yandex-cloud/ydb-go-sdk/v2 v2.12.0 // indirect + github.com/yandex-cloud/go-genproto v0.0.0-20211115083454-9ca41db5ed9e // indirect github.com/ydb-platform/ydb-go-genproto v0.0.0-20220203104745-929cf9c248bc // indirect + github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.0 // indirect + github.com/ydb-platform/ydb-go-sdk/v3 v3.24.2 // indirect + github.com/ydb-platform/ydb-go-yc v0.6.1 // indirect + github.com/ydb-platform/ydb-go-yc-metadata v0.0.9 // indirect go.etcd.io/etcd/api/v3 v3.5.4 // indirect go.etcd.io/etcd/client/pkg/v3 v3.5.4 // indirect go.uber.org/atomic v1.9.0 // indirect @@ -937,11 +937,23 @@ github.com/xdg-go/scram v1.1.0 h1:d70R37I0HrDLsafRrMBXyrD4lmQbCHE873t00Vr0gm0= github.com/xdg-go/scram v1.1.0/go.mod h1:1WAq6h33pAW+iRreB34OORO2Nf7qel3VV3fjBj+hCSs= github.com/xdg-go/stringprep v1.0.2 h1:6iq84/ryjjeRmMJwxutI51F2GIPlP5BfTvXHeYjyhBc= github.com/xdg-go/stringprep v1.0.2/go.mod h1:8F9zXuvzgwmyT5DUm4GUfZGDdT3W+LCvS6+da4O5kxM= +github.com/yandex-cloud/go-genproto v0.0.0-20210809082946-a97da516c588 h1:Lbz8X5Nre0Lg5QgCblmo0AhScWxeN3CVnX+mZ5Hxksk= github.com/yandex-cloud/go-genproto v0.0.0-20210809082946-a97da516c588/go.mod h1:HEUYX/p8966tMUHHT+TsS0hF/Ca/NYwqprC5WXSDMfE= +github.com/yandex-cloud/go-genproto v0.0.0-20211115083454-9ca41db5ed9e h1:9LPdmD1vqadsDQUva6t2O9MbnyvoOgo8nFNPaOIH5U8= +github.com/yandex-cloud/go-genproto v0.0.0-20211115083454-9ca41db5ed9e/go.mod h1:HEUYX/p8966tMUHHT+TsS0hF/Ca/NYwqprC5WXSDMfE= github.com/yandex-cloud/ydb-go-sdk/v2 v2.12.0 h1:TjMLCV3Poata+0Nw+Oa/ztEFagFrochAkGg91PG49jk= github.com/yandex-cloud/ydb-go-sdk/v2 v2.12.0/go.mod h1:L09Rymwmcpp9QMCbfGAlMtJ4kLFFsbGjkP3qv58r2yY= github.com/ydb-platform/ydb-go-genproto v0.0.0-20220203104745-929cf9c248bc h1:xvTP0fhYNm+Ws+xC34jzF9EdorPUKkucJr0TyybqVSk= github.com/ydb-platform/ydb-go-genproto v0.0.0-20220203104745-929cf9c248bc/go.mod h1:cc138nptTn9eKptCQl/grxP6pBKpo/bnXDiOxuVZtps= +github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.0 h1:74zGbvLn5kwLkkVoicJWzmvWhaIGdIWLUr1tfaMul08= +github.com/ydb-platform/ydb-go-sdk-auth-environ v0.1.0/go.mod h1:9E5eeVy08G/cu5azQVYGHduqxa6hz/dyZzJDjDTE010= +github.com/ydb-platform/ydb-go-sdk/v3 v3.9.0/go.mod h1:/KCidORSzHOsn+j46Iqb8Tf6UXNhPWRMcrMieh+i9Xw= +github.com/ydb-platform/ydb-go-sdk/v3 v3.24.2 h1:hFwtj5icsW5s3+sDXkmEa+rZN6ez9f4FeIOv2AYBrMk= +github.com/ydb-platform/ydb-go-sdk/v3 v3.24.2/go.mod h1:/KCidORSzHOsn+j46Iqb8Tf6UXNhPWRMcrMieh+i9Xw= +github.com/ydb-platform/ydb-go-yc v0.6.1 h1:DBw32JwTOsfFGnMlwri2f7C2joYvFnLNYEQAopID/Qg= +github.com/ydb-platform/ydb-go-yc v0.6.1/go.mod h1:iMalotfQEHibqPDNkwn0oT2UC5ieS5j6teIuGgPzaSE= +github.com/ydb-platform/ydb-go-yc-metadata v0.0.9 h1:jymJK3FVUphDa5q6oyjLODXLc34AR+aFTMiybacMLy0= +github.com/ydb-platform/ydb-go-yc-metadata v0.0.9/go.mod h1:L7zqxXrf3DXY2CWd9T9JBiPVpdooGBxJr4CPtWiMLCg= github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d h1:splanxYIlg+5LfHAM6xpdFEAYOk8iySO56hMFq6uLyA= github.com/youmark/pkcs8 v0.0.0-20181117223130-1be2e3e5546d/go.mod h1:rHwXgn7JulP+udvsHwJoVG1YGAP6VLg4y9I5dyZdqmA= github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= @@ -1129,6 +1141,8 @@ golang.org/x/net v0.0.0-20220325170049-de3da57026de/go.mod h1:CfG3xpIq0wQ8r1q4Su golang.org/x/net v0.0.0-20220401154927-543a649e0bdd/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220412020605-290c469a71a5 h1:bRb386wvrE+oBNdF1d/Xh9mQrfQ4ecYhW5qJ5GvTGT4= golang.org/x/net v0.0.0-20220412020605-290c469a71a5/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= +golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4 h1:HVyaeDAYux4pnY+D/SiwmLOR36ewZ4iGQIIrtnuCjFA= +golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U= golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= @@ -1263,6 +1277,8 @@ golang.org/x/sys v0.0.0-20220328115105-d36c6a25d886/go.mod h1:oPkhp1MJrh7nUepCBc golang.org/x/sys v0.0.0-20220330033206-e17cdc41300f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.0.0-20220412211240-33da011f77ad h1:ntjMns5wyP/fN65tdBD4g8J5w8n015+iIIs9rtjXkY0= golang.org/x/sys v0.0.0-20220412211240-33da011f77ad/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32 h1:Js08h5hqB5xyWR789+QqueR6sDE8mk+YvpETZ+F6X9Y= +golang.org/x/sys v0.0.0-20220429233432-b5fbb4746d32/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= @@ -1504,6 +1520,8 @@ google.golang.org/genproto v0.0.0-20220407144326-9054f6ed7bac/go.mod h1:8w6bsBMX google.golang.org/genproto v0.0.0-20220413183235-5e96e2839df9/go.mod h1:8w6bsBMX6yCPbAVTeqQHvzxW0EIFigd5lZyahWgyfDo= google.golang.org/genproto v0.0.0-20220414192740-2d67ff6cf2b4 h1:myaecH64R0bIEDjNORIel4iXubqzaHU1K2z8ajBwWcM= google.golang.org/genproto v0.0.0-20220414192740-2d67ff6cf2b4/go.mod h1:8w6bsBMX6yCPbAVTeqQHvzxW0EIFigd5lZyahWgyfDo= +google.golang.org/genproto v0.0.0-20220429170224-98d788798c3e h1:gMjH4zLGs9m+dGzR7qHCHaXMOwsJHJKKkHtyXhtOrJk= +google.golang.org/genproto v0.0.0-20220429170224-98d788798c3e/go.mod h1:8w6bsBMX6yCPbAVTeqQHvzxW0EIFigd5lZyahWgyfDo= google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c= google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38= google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM= @@ -1530,6 +1548,7 @@ google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnD google.golang.org/grpc v1.39.1/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE= google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34= google.golang.org/grpc v1.40.1/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34= +google.golang.org/grpc v1.43.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU= google.golang.org/grpc v1.44.0/go.mod h1:k+4IHHFw41K8+bbowsex27ge2rCb65oeWqe4jJ590SU= google.golang.org/grpc v1.45.0/go.mod h1:lN7owxKUQEqMfSyQikvvk5tf/6zMPsrK+ONuO11+0rQ= google.golang.org/grpc v1.46.0 h1:oCjezcn6g6A75TGoKYBPgKmVBLexhYLM6MebdrPApP8= 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/ydb/readme.md b/weed/filer/ydb/readme.md index 2221e13b5..fd6c5c0a2 100644 --- a/weed/filer/ydb/readme.md +++ b/weed/filer/ydb/readme.md @@ -1,6 +1,7 @@ ## YDB database: https://github.com/ydb-platform/ydb + go driver: https://github.com/ydb-platform/ydb-go-sdk options: @@ -8,15 +9,18 @@ options: ``` [ydb] enabled=true +dsn=grpcs://ydb-ru.yandex.net:2135/?database=/ru/home/username/db prefix="seaweedfs" useBucketPrefix=true -coonectionUrl=grpcs://ydb-ru.yandex.net:2135/?database=/ru/home/username/db +poolSizeLimit=50 ``` -get ydb types -``` -ydbgen -dir weed/filer/ydb -``` +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: + * i test using this dev database: `make dev_ydb` diff --git a/weed/filer/ydb/ydb_store.go b/weed/filer/ydb/ydb_store.go index d82fc4da7..4f3477471 100644 --- a/weed/filer/ydb/ydb_store.go +++ b/weed/filer/ydb/ydb_store.go @@ -7,9 +7,12 @@ import ( "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/util" - "github.com/yandex-cloud/ydb-go-sdk/v2" - "github.com/yandex-cloud/ydb-go-sdk/v2/connect" - "github.com/yandex-cloud/ydb-go-sdk/v2/table" + environ "github.com/ydb-platform/ydb-go-sdk-auth-environ" + ydb "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/named" + "github.com/ydb-platform/ydb-go-sdk/v3/table/types" "path" "strings" "time" @@ -24,15 +27,12 @@ var ( table.BeginTx(table.WithOnlineReadOnly()), table.CommitTx(), ) - rwTX = table.TxControl( - table.BeginTx(table.WithSerializableReadWrite()), - table.CommitTx(), - ) + rwTX = table.DefaultTxControl() ) type YdbStore struct { SupportBucketTable bool - DB *connect.Connection + DB ydb.Connection dirBuckets string tablePathPrefix string } @@ -48,34 +48,43 @@ func (store *YdbStore) GetName() string { func (store *YdbStore) Initialize(configuration util.Configuration, prefix string) (err error) { return store.initialize( configuration.GetString("filer.options.buckets_folder"), - configuration.GetString(prefix+"coonectionUrl"), + 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, sqlUrl string, tablePathPrefix string, useBucketPrefix bool, connectionTimeOut int) (err error) { +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 + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() if connectionTimeOut == 0 { connectionTimeOut = defaultConnectionTimeOut } - var cancel context.CancelFunc - connCtx, cancel := context.WithTimeout(context.Background(), time.Duration(connectionTimeOut)*time.Second) - defer cancel() - connParams := connect.MustConnectionString(sqlUrl) - store.DB, err = connect.New(connCtx, connParams) + opts := []ydb.Option{ + environ.WithEnvironCredentials(ctx), + ydb.WithDialTimeout(time.Duration(connectionTimeOut) * time.Second), + } + if poolSizeLimit > 0 { + opts = append(opts, ydb.WithSessionPoolSizeLimit(poolSizeLimit)) + } + store.DB, err = ydb.Open(ctx, dsn, opts...) if err != nil { - store.DB.Close() + _ = store.DB.Close(ctx) store.DB = nil - return fmt.Errorf("can not connect to %s error:%v", sqlUrl, err) + return fmt.Errorf("can not connect to %s error:%v", dsn, err) } - defer store.DB.Close() - - if err = store.DB.EnsurePathExists(connCtx, connParams.Database()); err != nil { - return fmt.Errorf("connect to %s error:%v", sqlUrl, err) + defer func() { _ = store.DB.Close(ctx) }() + 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) } return nil } @@ -92,16 +101,14 @@ func (store *YdbStore) insertOrUpdateEntry(ctx context.Context, entry *filer.Ent } fileMeta := FileMeta{util.HashStringToLong(dir), name, dir, meta} - return table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), query)) - if err != nil { - return err - } - _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters()) - return err - }), - ) + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), query)) + if err != nil { + return fmt.Errorf("Prepare %s : %v", dir, err) + } + _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters()) + return err + }) } func (store *YdbStore) InsertEntry(ctx context.Context, entry *filer.Entry) (err error) { @@ -114,68 +121,71 @@ func (store *YdbStore) UpdateEntry(ctx context.Context, entry *filer.Entry) (err func (store *YdbStore) FindEntry(ctx context.Context, fullpath util.FullPath) (entry *filer.Entry, err error) { dir, name := fullpath.DirAndName() - var res *table.Result - err = table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), findQuery)) - if err != nil { - return err + var data []byte + entryFound := false + err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), findQuery)) + if err != nil { + return fmt.Errorf("Prepare %s : %v", entry.FullPath, err) + } + _, res, err := stmt.Execute(ctx, roTX, table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$name", types.UTF8Value(name)))) + if err != nil { + return fmt.Errorf("Execute %s : %v", entry.FullPath, err) + } + 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) } - _, res, err = stmt.Execute(ctx, roTX, table.NewQueryParameters( - table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))), - table.ValueParam("$name", ydb.UTF8Value(name)))) - return err - }), - ) + entryFound = true + return nil + } + return res.Err() + }) if err != nil { return nil, err } - defer res.Close() - - for res.NextResultSet(ctx) { - for res.NextRow() { - res.SeekItem("meta") - entry.FullPath = fullpath - if err := entry.DecodeAttributesAndChunks(util.MaybeDecompressData(res.String())); err != nil { - return entry, fmt.Errorf("decode %s : %v", entry.FullPath, err) - } - return entry, nil - } + 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 nil, filer_pb.ErrNotFound + return entry, nil } func (store *YdbStore) DeleteEntry(ctx context.Context, fullpath util.FullPath) (err error) { dir, name := fullpath.DirAndName() - return table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteQuery)) - if err != nil { - return err - } - _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters( - table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))), - table.ValueParam("$name", ydb.UTF8Value(name)))) - return err - }), - ) + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteQuery)) + if err != nil { + return fmt.Errorf("Prepare %s : %v", dir, err) + } + _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$name", types.UTF8Value(name)))) + return err + }) } func (store *YdbStore) DeleteFolderChildren(ctx context.Context, fullpath util.FullPath) (err error) { dir, _ := fullpath.DirAndName() - return table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteFolderChildrenQuery)) - if err != nil { - return err - } - _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters( - table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))), - table.ValueParam("$directory", ydb.UTF8Value(dir)))) - return err - }), - ) + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), deleteFolderChildrenQuery)) + if err != nil { + return fmt.Errorf("Prepare %s : %v", dir, err) + } + _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters( + table.ValueParam("$dir_hash", types.Int64Value(util.HashStringToLong(dir))), + table.ValueParam("$directory", types.UTF8Value(dir)))) + return err + }) } func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) (lastFileName string, err error) { @@ -184,62 +194,60 @@ func (store *YdbStore) ListDirectoryEntries(ctx context.Context, dirPath util.Fu 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) - var res *table.Result startFileCompOp := ">" if includeStartFile { startFileCompOp = ">=" } - err = table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), fmt.Sprintf(ListDirectoryQuery, startFileCompOp))) - if err != nil { - return err + err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dir), fmt.Sprintf(ListDirectoryQuery, startFileCompOp))) + if err != nil { + return fmt.Errorf("Prepare %s : %v", dir, err) + } + _, res, err := stmt.Execute(ctx, roTX, 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.Int64Value(limit)), + )) + if err != nil { + return fmt.Errorf("Execute %s : %v", dir, err) + } + 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 { + glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err) + return fmt.Errorf("scan decode %s : %v", entry.FullPath, err) + } + if !eachEntryFunc(entry) { + break + } } - _, res, err = stmt.Execute(ctx, roTX, table.NewQueryParameters( - table.ValueParam("$dir_hash", ydb.Int64Value(util.HashStringToLong(dir))), - table.ValueParam("$directory", ydb.UTF8Value(dir)), - table.ValueParam("$start_name", ydb.UTF8Value(startFileName)), - table.ValueParam("$prefix", ydb.UTF8Value(prefix)), - table.ValueParam("$limit", ydb.Int64Value(limit)), - )) - return err - }), - ) + } + return res.Err() + }) if err != nil { return lastFileName, err } - defer res.Close() - - for res.NextSet() { - for res.NextRow() { - res.SeekItem("name") - name := res.UTF8() - res.SeekItem("meta") - data := res.String() - if res.Err() != nil { - glog.V(0).Infof("scan %s : %v", dirPath, err) - return lastFileName, fmt.Errorf("scan %s: %v", dirPath, err) - } - lastFileName = name - - entry := &filer.Entry{ - FullPath: util.NewFullPath(dir, name), - } - if err = entry.DecodeAttributesAndChunks(util.MaybeDecompressData(data)); err != nil { - glog.V(0).Infof("scan decode %s : %v", entry.FullPath, err) - return lastFileName, fmt.Errorf("scan decode %s : %v", entry.FullPath, err) - } - - if !eachEntryFunc(entry) { - break - } - } - } return lastFileName, nil } func (store *YdbStore) BeginTransaction(ctx context.Context) (context.Context, error) { - session, err := store.DB.Table().Pool().Create(ctx) + session, err := store.DB.Table().CreateSession(ctx) if err != nil { return ctx, err } @@ -251,21 +259,22 @@ func (store *YdbStore) BeginTransaction(ctx context.Context) (context.Context, e } func (store *YdbStore) CommitTransaction(ctx context.Context) error { - if tx, ok := ctx.Value("tx").(*table.Transaction); ok { - return tx.Commit(ctx) + 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 { + if tx, ok := ctx.Value("tx").(table.Transaction); ok { return tx.Rollback(ctx) } return nil } func (store *YdbStore) Shutdown() { - store.DB.Close() + _ = store.DB.Close(context.Background()) } func (store *YdbStore) getPrefix(dir string) string { diff --git a/weed/filer/ydb/ydb_store_kv.go b/weed/filer/ydb/ydb_store_kv.go index 3473d756d..b2a7af74f 100644 --- a/weed/filer/ydb/ydb_store_kv.go +++ b/weed/filer/ydb/ydb_store_kv.go @@ -5,68 +5,74 @@ import ( "fmt" "github.com/chrislusf/seaweedfs/weed/filer" "github.com/chrislusf/seaweedfs/weed/filer/abstract_sql" - "github.com/yandex-cloud/ydb-go-sdk/v2" - "github.com/yandex-cloud/ydb-go-sdk/v2/table" + "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 table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), insertQuery)) - if err != nil { - return fmt.Errorf("kv put: %v", err) - } - _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters()) - return fmt.Errorf("kv put: %v", err) - }), - ) + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), insertQuery)) + if err != nil { + return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, name), err) + } + _, _, err = stmt.Execute(ctx, rwTX, fileMeta.QueryParameters()) + if err != nil { + return fmt.Errorf("kv put %s: %v", util.NewFullPath(dirStr, name), err) + } + return nil + }) } func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { dirStr, dirHash, name := abstract_sql.GenDirAndName(key) - var res *table.Result - err = table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), findQuery)) - if err != nil { - return err - } - _, res, err = stmt.Execute(ctx, roTX, table.NewQueryParameters( - table.ValueParam("$dir_hash", ydb.Int64Value(dirHash)), - table.ValueParam("$name", ydb.UTF8Value(name)))) - return err - }), - ) - if err != nil { - return nil, fmt.Errorf("kv get: %v", err) - } - defer res.Close() - - for res.NextResultSet(ctx) { + valueFound := false + err = store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), findQuery)) + if err != nil { + return fmt.Errorf("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 %s: %v", util.NewFullPath(dirStr, name), err) + } + defer func() { _ = res.Close() }() for res.NextRow() { - if err = res.Scan(&value); err != nil { - return nil, fmt.Errorf("kv get: %v", err) + if err := res.ScanNamed(named.Required("meta", &value)); err != nil { + return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name), err) } - return + valueFound = true + return nil } + return res.Err() + }) + + if !valueFound { + return nil, filer.ErrKvNotFound } - 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 table.Retry(ctx, store.DB.Table().Pool(), - table.OperationFunc(func(ctx context.Context, s *table.Session) (err error) { - stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), deleteQuery)) - if err != nil { - return fmt.Errorf("kv delete: %s", err) - } - _, _, err = stmt.Execute(ctx, rwTX, table.NewQueryParameters( - table.ValueParam("$dir_hash", ydb.Int64Value(dirHash)), - table.ValueParam("$name", ydb.UTF8Value(name)))) - return fmt.Errorf("kv delete: %s", err) - }), - ) + return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) (err error) { + stmt, err := s.Prepare(ctx, store.withPragma(store.getPrefix(dirStr), insertQuery)) + if err != nil { + return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, 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), err) + } + return nil + }) + } diff --git a/weed/filer/ydb/ydb_types.go b/weed/filer/ydb/ydb_types.go index 5bc0d2afd..fb1df3eee 100644 --- a/weed/filer/ydb/ydb_types.go +++ b/weed/filer/ydb/ydb_types.go @@ -1,5 +1,10 @@ package ydb +import ( + "github.com/ydb-platform/ydb-go-sdk/v3/table" + "github.com/ydb-platform/ydb-go-sdk/v3/table/types" +) + //go:generate ydbgen //ydb:gen @@ -12,3 +17,11 @@ type FileMeta struct { //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))) +} diff --git a/weed/filer/ydb/ydb_types_ydbgen.go b/weed/filer/ydb/ydb_types_ydbgen.go deleted file mode 100644 index 8bd1906f9..000000000 --- a/weed/filer/ydb/ydb_types_ydbgen.go +++ /dev/null @@ -1,194 +0,0 @@ -// Code generated by ydbgen; DO NOT EDIT. - -package ydb - -import ( - "strconv" - - "github.com/yandex-cloud/ydb-go-sdk/v2" - "github.com/yandex-cloud/ydb-go-sdk/v2/table" -) - -var ( - _ = strconv.Itoa - _ = ydb.StringValue - _ = table.NewQueryParameters -) - -func (f *FileMeta) Scan(res *table.Result) (err error) { - res.SeekItem("dir_hash") - f.DirHash = res.Int64() - - res.SeekItem("name") - f.Name = res.UTF8() - - res.SeekItem("directory") - f.Directory = res.UTF8() - - return res.Err() -} - -func (f *FileMeta) QueryParameters() *table.QueryParameters { - var v0 ydb.Value - { - vp0 := ydb.Int64Value(f.DirHash) - v0 = vp0 - } - var v1 ydb.Value - { - vp0 := ydb.UTF8Value(f.Name) - v1 = vp0 - } - var v2 ydb.Value - { - vp0 := ydb.UTF8Value(f.Directory) - v2 = vp0 - } - return table.NewQueryParameters( - table.ValueParam("$dir_hash", v0), - table.ValueParam("$name", v1), - table.ValueParam("$directory", v2), - ) -} - -func (f *FileMeta) StructValue() ydb.Value { - var v0 ydb.Value - { - var v1 ydb.Value - { - vp0 := ydb.Int64Value(f.DirHash) - v1 = vp0 - } - var v2 ydb.Value - { - vp0 := ydb.UTF8Value(f.Name) - v2 = vp0 - } - var v3 ydb.Value - { - vp0 := ydb.UTF8Value(f.Directory) - v3 = vp0 - } - v0 = ydb.StructValue( - ydb.StructFieldValue("dir_hash", v1), - ydb.StructFieldValue("name", v2), - ydb.StructFieldValue("directory", v3), - ) - } - return v0 -} - -func (f *FileMeta) StructType() ydb.Type { - var t0 ydb.Type - { - fs0 := make([]ydb.StructOption, 3) - var t1 ydb.Type - { - tp0 := ydb.TypeInt64 - t1 = tp0 - } - fs0[0] = ydb.StructField("dir_hash", t1) - var t2 ydb.Type - { - tp0 := ydb.TypeUTF8 - t2 = tp0 - } - fs0[1] = ydb.StructField("name", t2) - var t3 ydb.Type - { - tp0 := ydb.TypeUTF8 - t3 = tp0 - } - fs0[2] = ydb.StructField("directory", t3) - t0 = ydb.Struct(fs0...) - } - return t0 -} - -func (fs *FileMetas) Scan(res *table.Result) (err error) { - for res.NextRow() { - var x0 FileMeta - res.SeekItem("dir_hash") - x0.DirHash = res.Int64() - - res.SeekItem("name") - x0.Name = res.UTF8() - - res.SeekItem("directory") - x0.Directory = res.UTF8() - - if res.Err() == nil { - *fs = append(*fs, x0) - } - } - return res.Err() -} - -func (fs FileMetas) ListValue() ydb.Value { - var list0 ydb.Value - vs0 := make([]ydb.Value, len(fs)) - for i0, item0 := range fs { - var v0 ydb.Value - { - var v1 ydb.Value - { - var v2 ydb.Value - { - vp0 := ydb.Int64Value(item0.DirHash) - v2 = vp0 - } - var v3 ydb.Value - { - vp0 := ydb.UTF8Value(item0.Name) - v3 = vp0 - } - var v4 ydb.Value - { - vp0 := ydb.UTF8Value(item0.Directory) - v4 = vp0 - } - v1 = ydb.StructValue( - ydb.StructFieldValue("dir_hash", v2), - ydb.StructFieldValue("name", v3), - ydb.StructFieldValue("directory", v4), - ) - } - v0 = v1 - } - vs0[i0] = v0 - } - if len(vs0) == 0 { - var t1 ydb.Type - { - var t2 ydb.Type - { - fs0 := make([]ydb.StructOption, 3) - var t3 ydb.Type - { - tp0 := ydb.TypeInt64 - t3 = tp0 - } - fs0[0] = ydb.StructField("dir_hash", t3) - var t4 ydb.Type - { - tp0 := ydb.TypeUTF8 - t4 = tp0 - } - fs0[1] = ydb.StructField("name", t4) - var t5 ydb.Type - { - tp0 := ydb.TypeUTF8 - t5 = tp0 - } - fs0[2] = ydb.StructField("directory", t5) - t2 = ydb.Struct(fs0...) - } - t1 = t2 - } - t0 := ydb.List(t1) - list0 = ydb.ZeroValue(t0) - } else { - list0 = ydb.ListValue(vs0...) - } - return list0 -} diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index 8779e9ac0..928aad253 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" |