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 <karimov_m@inbox.ru>

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}'
+

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

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=

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
+}

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
 	}

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)
 	}
 

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)
 }