aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--docker/Makefile3
-rw-r--r--docker/compose/local-ydb-compose.yml9
-rw-r--r--weed/filer/ydb/ydb_store.go39
-rw-r--r--weed/filer/ydb/ydb_store_kv.go12
-rw-r--r--weed/filer/ydb/ydb_types.go8
5 files changed, 33 insertions, 38 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
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"),
}
}
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-21 05:19:22 +0000