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 index ce3e7e9ed..33f550600 100644 --- a/docker/compose/local-ydb-compose.yml +++ b/docker/compose/local-ydb-compose.yml @@ -7,10 +7,8 @@ services: - 2135:2135 - 8765:8765 - 2136:2136 - volumes: - - ./seaweedfs.sql:/docker-entrypoint-initdb.d/seaweedfs.sql environment: - - YDB_DEFAULT_LOG_LEVEL=NOTICE + - YDB_DEFAULT_LOG_LEVEL=DEBUG - GRPC_TLS_PORT=2135 - GRPC_PORT=2136 - MON_PORT=8765 @@ -26,5 +24,10 @@ services: 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/filer/ydb/ydb_store.go b/weed/filer/ydb/ydb_store.go index 8271c6e18..8e0815009 100644 --- a/weed/filer/ydb/ydb_store.go +++ b/weed/filer/ydb/ydb_store.go @@ -8,7 +8,7 @@ import ( "github.com/chrislusf/seaweedfs/weed/glog" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/util" - "github.com/ydb-platform/ydb-go-sdk-auth-environ" + 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" @@ -73,8 +73,8 @@ func (store *YdbStore) initialize(dirBuckets string, dsn string, tablePathPrefix connectionTimeOut = defaultConnectionTimeOut } opts := []ydb.Option{ - environ.WithEnvironCredentials(ctx), ydb.WithDialTimeout(time.Duration(connectionTimeOut) * time.Second), + environ.WithEnvironCredentials(ctx), } if poolSizeLimit > 0 { opts = append(opts, ydb.WithSessionPoolSizeLimit(poolSizeLimit)) @@ -84,11 +84,12 @@ func (store *YdbStore) initialize(dirBuckets string, dsn string, tablePathPrefix } store.DB, err = ydb.Open(ctx, dsn, opts...) if err != nil { - _ = store.DB.Close(ctx) - store.DB = nil - return fmt.Errorf("can not connect to %s error:%v", dsn, err) + if store.DB != nil { + _ = store.DB.Close(ctx) + store.DB = nil + } + return fmt.Errorf("can not connect to %s error: %v", dsn, 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 { @@ -98,18 +99,11 @@ func (store *YdbStore) initialize(dirBuckets string, dsn string, tablePathPrefix return fmt.Errorf("MakeRecursive %s : %v", store.tablePathPrefix, err) } - whoAmI, err := store.DB.Discovery().WhoAmI(ctx) - if err != nil { - return fmt.Errorf("connect to %s error:%v", dsn, err) - } - glog.V(0).Infof("connected to ydb: %s", whoAmI.String()) - tablePath := path.Join(store.tablePathPrefix, abstract_sql.DEFAULT_TABLE) - if err := store.createTable(ctx, tablePath); err != nil { + if err = store.createTable(ctx, tablePath); err != nil { glog.Errorf("createTable %s: %v", tablePath, err) } - - return nil + 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) { @@ -309,7 +303,8 @@ func (store *YdbStore) OnBucketCreation(bucket string) { store.dbsLock.Lock() defer store.dbsLock.Unlock() - if err := store.createTable(context.Background(), bucket); err != nil { + if err := store.createTable(context.Background(), + path.Join(store.tablePathPrefix, bucket, abstract_sql.DEFAULT_TABLE)); err != nil { glog.Errorf("createTable %s: %v", bucket, err) } @@ -323,7 +318,8 @@ func (store *YdbStore) OnBucketDeletion(bucket string) { store.dbsLock.Lock() defer store.dbsLock.Unlock() - if err := store.deleteTable(context.Background(), bucket); err != nil { + if err := store.deleteTable(context.Background(), + path.Join(store.tablePathPrefix, bucket, abstract_sql.DEFAULT_TABLE)); err != nil { glog.Errorf("deleteTable %s: %v", bucket, err) } @@ -334,13 +330,6 @@ func (store *YdbStore) OnBucketDeletion(bucket string) { } func (store *YdbStore) createTable(ctx context.Context, prefix string) error { - e, err := store.DB.Scheme().DescribePath(ctx, prefix) - if err != nil { - return fmt.Errorf("describe path %s error:%v", prefix, err) - } - if e.IsTable() { - return nil - } return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { return s.CreateTable(ctx, prefix, createTableOptions()...) }) @@ -351,7 +340,7 @@ func (store *YdbStore) deleteTable(ctx context.Context, prefix string) error { return nil } return store.DB.Table().Do(ctx, func(ctx context.Context, s table.Session) error { - return s.DropTable(ctx, path.Join(prefix, abstract_sql.DEFAULT_TABLE)) + return s.DropTable(ctx, prefix) }) } diff --git a/weed/filer/ydb/ydb_store_kv.go b/weed/filer/ydb/ydb_store_kv.go index 81ac6446b..d679d7246 100644 --- a/weed/filer/ydb/ydb_store_kv.go +++ b/weed/filer/ydb/ydb_store_kv.go @@ -17,11 +17,11 @@ func (store *YdbStore) KvPut(ctx context.Context, key []byte, value []byte) (err 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), err) + return fmt.Errorf("Prepare %s: %v", util.NewFullPath(dirStr, name).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 fmt.Errorf("kv put %s: %v", util.NewFullPath(dirStr, name).Name(), err) } return nil }) @@ -39,12 +39,12 @@ func (store *YdbStore) KvGet(ctx context.Context, key []byte) (value []byte, err 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) + return fmt.Errorf("kv get %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), err) + return fmt.Errorf("scanNamed %s : %v", util.NewFullPath(dirStr, name).Name(), err) } valueFound = true return nil @@ -64,13 +64,13 @@ func (store *YdbStore) KvDelete(ctx context.Context, key []byte) (err error) { 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), err) + 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), err) + 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 index 8c24c4738..4194afbc8 100644 --- a/weed/filer/ydb/ydb_types.go +++ b/weed/filer/ydb/ydb_types.go @@ -29,10 +29,10 @@ func (fm *FileMeta) queryParameters() *table.QueryParameters { func createTableOptions() []options.CreateTableOption { return []options.CreateTableOption{ - options.WithColumn("dir_hash", types.TypeUint64), - options.WithColumn("name", types.TypeUTF8), - options.WithColumn("directory", types.TypeUTF8), - options.WithColumn("meta", types.TypeString), + options.WithColumn("dir_hash", types.Optional(types.TypeUint64)), + 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"), } }