From ed1c54dae90d853a4ce61daebd27e6d465c6105b Mon Sep 17 00:00:00 2001 From: MaratKarimov Date: Tue, 21 Jul 2026 03:35:20 +0300 Subject: [PATCH] Bump Tarantool client library from 2.4.2 to 3.0.0 (#10377) * Bump Tarantool client library from 2.4.2 to 3.0.0 * Fix review comments --------- Co-authored-by: Marat Karimov --- docker/Dockerfile.tarantool.dev_env | 6 ++ go.mod | 3 +- go.sum | 6 +- weed/filer/tarantool/tarantool_slog.go | 75 ++++++++++++++++++++ weed/filer/tarantool/tarantool_store.go | 57 ++++++++------- weed/filer/tarantool/tarantool_store_kv.go | 31 ++++---- weed/filer/tarantool/tarantool_store_test.go | 2 +- 7 files changed, 132 insertions(+), 48 deletions(-) create mode 100644 weed/filer/tarantool/tarantool_slog.go diff --git a/docker/Dockerfile.tarantool.dev_env b/docker/Dockerfile.tarantool.dev_env index 4ce0fc9af..d9fad40b5 100644 --- a/docker/Dockerfile.tarantool.dev_env +++ b/docker/Dockerfile.tarantool.dev_env @@ -15,3 +15,9 @@ COPY tarantool /opt/tarantool/app # build app RUN tt build app +# the base image's default healthcheck assumes a single-instance layout and +# checks a control socket path that doesn't exist for this multi-instance +# app, so it never reports healthy; check instance status instead. +HEALTHCHECK --interval=5s --timeout=3s --start-period=15s --retries=6 \ + CMD tt status app | awk 'NR>2{if ($2!="RUNNING") exit 1} END{if (NR<=2) exit 1}' + diff --git a/go.mod b/go.mod index b4e346814..67715345a 100644 --- a/go.mod +++ b/go.mod @@ -150,7 +150,8 @@ require ( github.com/schollz/progressbar/v3 v3.19.1 github.com/seaweedfs/go-fuse/v2 v2.9.3 github.com/shirou/gopsutil/v4 v4.26.5 - github.com/tarantool/go-tarantool/v2 v2.4.2 + github.com/tarantool/go-option v1.1.0 + github.com/tarantool/go-tarantool/v3 v3.0.0 github.com/testcontainers/testcontainers-go v0.43.0 github.com/tikv/client-go/v2 v2.0.7 github.com/xeipuuv/gojsonschema v1.2.0 diff --git a/go.sum b/go.sum index 8959088a3..74ce4d78e 100644 --- a/go.sum +++ b/go.sum @@ -1905,8 +1905,10 @@ github.com/t3rm1n4l/go-mega v0.0.0-20251120131202-6845944c051c/go.mod h1:BF/l2jN github.com/tailscale/depaware v0.0.0-20210622194025-720c4b409502/go.mod h1:p9lPsd+cx33L3H9nNoecRRxPssFKUwwI50I3pZ0yT+8= github.com/tarantool/go-iproto v1.1.0 h1:HULVOIHsiehI+FnHfM7wMDntuzUddO09DKqu2WnFQ5A= github.com/tarantool/go-iproto v1.1.0/go.mod h1:LNCtdyZxojUed8SbOiYHoc3v9NvaZTB7p96hUySMlIo= -github.com/tarantool/go-tarantool/v2 v2.4.2 h1:rkzYtFhLJLA9RDIhjzN93MJBN5PBxHW4+soq+RB90gE= -github.com/tarantool/go-tarantool/v2 v2.4.2/go.mod h1:MTbhdjFc3Jl63Lgi/UJr5D+QbT+QegqOzsNJGmaw7VM= +github.com/tarantool/go-option v1.1.0 h1:ShoOhNsdL41sRpm4hXCRDjV8H0WzPkd4UnKhLKbW//w= +github.com/tarantool/go-option v1.1.0/go.mod h1:hMr9z2JXOWlgdCBpCPSL2nwp8718GKYvNBJ+ZuzJbCo= +github.com/tarantool/go-tarantool/v3 v3.0.0 h1:zsIXS4nvSSXqZWtN/f1Bfuq973j6J85O5U/8RYQCRcE= +github.com/tarantool/go-tarantool/v3 v3.0.0/go.mod h1:TXxLWhUCgdxXFfelnTSkq+goKRTRj660zxq4/WXPe8k= github.com/testcontainers/testcontainers-go v0.43.0 h1:oEQx5MW2DGd9z3AeEQfB2lPM0eLs7ztyaGRu75bFo5A= github.com/testcontainers/testcontainers-go v0.43.0/go.mod h1:+VxkT2NQnKOZPKi6praMuMKYHYyOGXr0XSBSlSMCzFo= github.com/testcontainers/testcontainers-go/modules/compose v0.42.0 h1:+t1ZN31TD36cwxmeLqGwe7wIdvblBm0Z+vlj4SX8Mv0= diff --git a/weed/filer/tarantool/tarantool_slog.go b/weed/filer/tarantool/tarantool_slog.go new file mode 100644 index 000000000..e53741c76 --- /dev/null +++ b/weed/filer/tarantool/tarantool_slog.go @@ -0,0 +1,75 @@ +//go:build tarantool + +package tarantool + +import ( + "context" + "fmt" + "log/slog" + "strings" + + "github.com/seaweedfs/seaweedfs/weed/glog" +) + +// glogSlogHandler forwards go-tarantool/v3's structured logs into weed/glog, +// which is the logging facility used everywhere else in SeaweedFS. +type glogSlogHandler struct { + preformatted string +} + +func newGlogLogger() *slog.Logger { + return slog.New(&glogSlogHandler{}) +} + +func (h *glogSlogHandler) Enabled(context.Context, slog.Level) bool { + return true +} + +func (h *glogSlogHandler) Handle(ctx context.Context, r slog.Record) error { + msg := r.Message + var attrs []string + if h.preformatted != "" { + attrs = append(attrs, h.preformatted) + } + r.Attrs(func(a slog.Attr) bool { + attrs = append(attrs, fmt.Sprintf("%s=%v", a.Key, a.Value.Any())) + return true + }) + if len(attrs) > 0 { + msg = msg + " " + strings.Join(attrs, " ") + } + + switch { + case r.Level >= slog.LevelError: + glog.ErrorfCtx(ctx, "tarantool: %s", msg) + case r.Level >= slog.LevelWarn: + glog.WarningfCtx(ctx, "tarantool: %s", msg) + case r.Level >= slog.LevelInfo: + glog.InfofCtx(ctx, "tarantool: %s", msg) + default: + glog.V(1).InfofCtx(ctx, "tarantool: %s", msg) + } + return nil +} + +func (h *glogSlogHandler) WithAttrs(attrs []slog.Attr) slog.Handler { + if len(attrs) == 0 { + return h + } + var sb strings.Builder + if h.preformatted != "" { + sb.WriteString(h.preformatted) + sb.WriteString(" ") + } + for i, a := range attrs { + if i > 0 { + sb.WriteString(" ") + } + sb.WriteString(fmt.Sprintf("%s=%v", a.Key, a.Value.Any())) + } + return &glogSlogHandler{preformatted: sb.String()} +} + +func (h *glogSlogHandler) WithGroup(name string) slog.Handler { + return h +} diff --git a/weed/filer/tarantool/tarantool_store.go b/weed/filer/tarantool/tarantool_store.go index beba41ed4..5094c8e24 100644 --- a/weed/filer/tarantool/tarantool_store.go +++ b/weed/filer/tarantool/tarantool_store.go @@ -14,9 +14,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" "github.com/seaweedfs/seaweedfs/weed/util" weed_util "github.com/seaweedfs/seaweedfs/weed/util" - "github.com/tarantool/go-tarantool/v2" - "github.com/tarantool/go-tarantool/v2/crud" - "github.com/tarantool/go-tarantool/v2/pool" + "github.com/tarantool/go-option" + "github.com/tarantool/go-tarantool/v3" + "github.com/tarantool/go-tarantool/v3/crud" + "github.com/tarantool/go-tarantool/v3/pool" ) const ( @@ -28,7 +29,7 @@ func init() { } type TarantoolStore struct { - pool *pool.ConnectionPool + pool *pool.Pool } func (store *TarantoolStore) GetName() string { @@ -53,36 +54,34 @@ func (store *TarantoolStore) Initialize(configuration weed_util.Configuration, p return fmt.Errorf("parse tarantool store timeout: %w", err) } - maxReconnects := configuration.GetInt(prefix + "maxReconnects") - if maxReconnects < 0 { - return fmt.Errorf("maxReconnects is negative") + if maxReconnects := configuration.GetInt(prefix + "maxReconnects"); maxReconnects != 1000 { + glog.Warningf("tarantool store: \"maxReconnects\" is no longer supported with go-tarantool v3 (the connection pool manages reconnection internally); ignoring configured value %d", maxReconnects) } addresses := strings.Split(address, ",") - return store.initialize(addresses, user, password, timeout, uint(maxReconnects)) + return store.initialize(addresses, user, password, timeout) } -func (store *TarantoolStore) initialize(addresses []string, user string, password string, timeout time.Duration, maxReconnects uint) error { +func (store *TarantoolStore) initialize(addresses []string, user string, password string, timeout time.Duration) error { opts := tarantool.Opts{ - Timeout: timeout, - Reconnect: time.Second, - MaxReconnects: maxReconnects, + Timeout: timeout, } poolInstances := makePoolInstances(addresses, user, password, opts) poolOpts := pool.Opts{ CheckTimeout: time.Second, + Logger: newGlogLogger(), } ctx := context.Background() - p, err := pool.ConnectWithOpts(ctx, poolInstances, poolOpts) + p, err := pool.NewWithOpts(ctx, poolInstances, poolOpts) if err != nil { return fmt.Errorf("Can't create connection pool: %w", err) } - _, err = p.Do(tarantool.NewPingRequest(), pool.ANY).Get() + _, err = p.Do(tarantool.NewPingRequest(), pool.ModeAny).Get() if err != nil { return err } @@ -150,13 +149,13 @@ func (store *TarantoolStore) InsertEntry(ctx context.Context, entry *filer.Entry }, } - req := crud.MakeUpsertRequest(tarantoolSpaceName). + req := crud.NewUpsertRequest(tarantoolSpaceName). Tuple([]interface{}{dir, nil, name, ttl, string(meta)}). Operations(operations) ret := crud.Result{} - if err := store.pool.Do(req, pool.RW).GetTyped(&ret); err != nil { + if err := store.pool.Do(req, pool.ModeRW).GetTyped(&ret); err != nil { return fmt.Errorf("insert %s: %s", entry.FullPath, err) } @@ -171,19 +170,19 @@ func (store *TarantoolStore) FindEntry(ctx context.Context, fullpath weed_util.F dir, name := fullpath.DirAndName() findEntryGetOpts := crud.GetOpts{ - Fields: crud.MakeOptTuple([]interface{}{"data"}), - Mode: crud.MakeOptString("read"), - PreferReplica: crud.MakeOptBool(true), - Balance: crud.MakeOptBool(true), + Fields: option.SomeAny([]interface{}{"data"}), + Mode: option.SomeString("read"), + PreferReplica: option.SomeBool(true), + Balance: option.SomeBool(true), } - req := crud.MakeGetRequest(tarantoolSpaceName). - Key(crud.Tuple([]interface{}{dir, name})). + req := crud.NewGetRequest(tarantoolSpaceName). + Key([]interface{}{dir, name}). Opts(findEntryGetOpts) resp := crud.Result{} - err = store.pool.Do(req, pool.PreferRO).GetTyped(&resp) + err = store.pool.Do(req, pool.ModePreferRO).GetTyped(&resp) if err != nil { return nil, err } @@ -219,14 +218,14 @@ func (store *TarantoolStore) DeleteEntry(ctx context.Context, fullpath weed_util dir, name := fullpath.DirAndName() delOpts := crud.DeleteOpts{ - Noreturn: crud.MakeOptBool(true), + Noreturn: option.SomeBool(true), } - req := crud.MakeDeleteRequest(tarantoolSpaceName). - Key(crud.Tuple([]interface{}{dir, name})). + req := crud.NewDeleteRequest(tarantoolSpaceName). + Key([]interface{}{dir, name}). Opts(delOpts) - if _, err := store.pool.Do(req, pool.RW).Get(); err != nil { + if _, err := store.pool.Do(req, pool.ModeRW).Get(); err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) } @@ -237,7 +236,7 @@ func (store *TarantoolStore) DeleteFolderChildren(ctx context.Context, fullpath req := tarantool.NewCallRequest("filer_metadata.delete_by_directory_idx"). Args([]interface{}{fullpath}) - if _, err := store.pool.Do(req, pool.RW).Get(); err != nil { + if _, err := store.pool.Do(req, pool.ModeRW).Get(); err != nil { return fmt.Errorf("delete %s : %v", fullpath, err) } @@ -253,7 +252,7 @@ func (store *TarantoolStore) ListDirectoryEntries(ctx context.Context, dirPath w req := tarantool.NewCallRequest("filer_metadata.find_by_directory_idx_and_name"). Args([]interface{}{string(dirPath), startFileName, includeStartFile, limit}) - results, err := store.pool.Do(req, pool.PreferRO).Get() + results, err := store.pool.Do(req, pool.ModePreferRO).Get() if err != nil { return } diff --git a/weed/filer/tarantool/tarantool_store_kv.go b/weed/filer/tarantool/tarantool_store_kv.go index 0b4811c90..4123cb2ab 100644 --- a/weed/filer/tarantool/tarantool_store_kv.go +++ b/weed/filer/tarantool/tarantool_store_kv.go @@ -8,8 +8,9 @@ import ( "reflect" "github.com/seaweedfs/seaweedfs/weed/filer" - "github.com/tarantool/go-tarantool/v2/crud" - "github.com/tarantool/go-tarantool/v2/pool" + "github.com/tarantool/go-option" + "github.com/tarantool/go-tarantool/v3/crud" + "github.com/tarantool/go-tarantool/v3/pool" ) const ( @@ -26,12 +27,12 @@ func (store *TarantoolStore) KvPut(ctx context.Context, key []byte, value []byte }, } - req := crud.MakeUpsertRequest(tarantoolKVSpaceName). + req := crud.NewUpsertRequest(tarantoolKVSpaceName). Tuple([]interface{}{string(key), nil, string(value)}). Operations(operations) ret := crud.Result{} - if err := store.pool.Do(req, pool.RW).GetTyped(&ret); err != nil { + if err := store.pool.Do(req, pool.ModeRW).GetTyped(&ret); err != nil { return fmt.Errorf("kv put: %w", err) } @@ -41,19 +42,19 @@ func (store *TarantoolStore) KvPut(ctx context.Context, key []byte, value []byte func (store *TarantoolStore) KvGet(ctx context.Context, key []byte) (value []byte, err error) { getOpts := crud.GetOpts{ - Fields: crud.MakeOptTuple([]interface{}{"value"}), - Mode: crud.MakeOptString("read"), - PreferReplica: crud.MakeOptBool(true), - Balance: crud.MakeOptBool(true), + Fields: option.SomeAny([]interface{}{"value"}), + Mode: option.SomeString("read"), + PreferReplica: option.SomeBool(true), + Balance: option.SomeBool(true), } - req := crud.MakeGetRequest(tarantoolKVSpaceName). - Key(crud.Tuple([]interface{}{string(key)})). + req := crud.NewGetRequest(tarantoolKVSpaceName). + Key([]interface{}{string(key)}). Opts(getOpts) resp := crud.Result{} - err = store.pool.Do(req, pool.PreferRO).GetTyped(&resp) + err = store.pool.Do(req, pool.ModePreferRO).GetTyped(&resp) if err != nil { return nil, err } @@ -79,14 +80,14 @@ func (store *TarantoolStore) KvGet(ctx context.Context, key []byte) (value []byt func (store *TarantoolStore) KvDelete(ctx context.Context, key []byte) (err error) { delOpts := crud.DeleteOpts{ - Noreturn: crud.MakeOptBool(true), + Noreturn: option.SomeBool(true), } - req := crud.MakeDeleteRequest(tarantoolKVSpaceName). - Key(crud.Tuple([]interface{}{string(key)})). + req := crud.NewDeleteRequest(tarantoolKVSpaceName). + Key([]interface{}{string(key)}). Opts(delOpts) - if _, err := store.pool.Do(req, pool.RW).Get(); err != nil { + if _, err := store.pool.Do(req, pool.ModeRW).Get(); err != nil { return fmt.Errorf("kv delete: %w", err) } diff --git a/weed/filer/tarantool/tarantool_store_test.go b/weed/filer/tarantool/tarantool_store_test.go index 3fb6f7ebd..67b39e7eb 100644 --- a/weed/filer/tarantool/tarantool_store_test.go +++ b/weed/filer/tarantool/tarantool_store_test.go @@ -18,6 +18,6 @@ func TestStore(t *testing.T) { } store := &TarantoolStore{} addresses := []string{"127.0.1:3303"} - store.initialize(addresses, "client", "client", 5*time.Second, 1000) + store.initialize(addresses, "client", "client", 5*time.Second) store_test.TestFilerStore(t, store) }