failover: support TcS as config storage

Closes #1193

@TarantoolBot document
Title: `tt cluster failover` commands support TcS as config storage

Author: elhimov

CHANGELOG.md

@@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 - `tt create`: add template for Tarantool Config Storage.
 - `tt create`: add template for non-vshard Cluster.
+- `tt cluster failover`: support Tarantool Config Storage.
 
 ### Changed
 

cli/cluster/cmd/failover.go

@@ -9,8 +9,6 @@ import (
 	"github.com/google/uuid"
 	libcluster "github.com/tarantool/tt/lib/cluster"
 	"github.com/tarantool/tt/lib/connect"
-	"go.etcd.io/etcd/api/v3/mvccpb"
-	clientv3 "go.etcd.io/etcd/client/v3"
 	"gopkg.in/yaml.v2"
 )
 
@@ -63,36 +61,22 @@ type SwitchStatusCtx struct {
 	TaskID string
 }
 
-func makeEtcdOpts(uriOpts connect.UriOpts) libcluster.EtcdOpts {
-	opts := libcluster.EtcdOpts{
-		Endpoints:      []string{uriOpts.Endpoint},
-		Username:       uriOpts.Username,
-		Password:       uriOpts.Password,
-		KeyFile:        uriOpts.KeyFile,
-		CertFile:       uriOpts.CertFile,
-		CaPath:         uriOpts.CaPath,
-		CaFile:         uriOpts.CaFile,
-		SkipHostVerify: uriOpts.SkipHostVerify,
-		Timeout:        uriOpts.Timeout,
-	}
-
-	return opts
-}
-
 // Switch master instance.
 func Switch(url string, switchCtx SwitchCtx) error {
 	uriOpts, err := connect.CreateUriOpts(url)
 	if err != nil {
 		return fmt.Errorf("invalid URL %q: %w", url, err)
 	}
+	connOpts := libcluster.ConnectOpts{
+		Username: switchCtx.Username,
+		Password: switchCtx.Password,
+	}
 
-	opts := makeEtcdOpts(uriOpts)
-
-	etcd, err := libcluster.ConnectEtcd(opts)
+	conn, err := libcluster.ConnectCStorage(uriOpts, connOpts)
 	if err != nil {
-		return fmt.Errorf("unable to connect to etcd: %w", err)
+		return fmt.Errorf("unable to connect to config storage: %w", err)
 	}
-	defer etcd.Close()
+	defer conn.Close()
 
 	cmd := switchCmd{
 		Command:   "switch",
@@ -109,54 +93,40 @@ func Switch(url string, switchCtx SwitchCtx) error {
 	key := uriOpts.Prefix + failoverPath + uuid
 
 	if switchCtx.Wait {
-		ctx, cancel_watch := context.WithTimeout(context.Background(),
+		ctxWatch, cancelWatch := context.WithTimeout(context.Background(),
 			time.Duration(switchCtx.Timeout)*time.Second+cmdAdditionalWait)
-		outputChan := make(chan *clientv3.Event)
-		defer cancel_watch()
-
-		go func() {
-			waitChan := etcd.Watch(ctx, key)
-			defer close(outputChan)
-
-			for resp := range waitChan {
-				for _, ev := range resp.Events {
-					switch ev.Type {
-					case mvccpb.PUT:
-						outputChan <- ev
-					}
-				}
-			}
-		}()
+		defer cancelWatch()
+		watchChan := conn.Watch(ctxWatch, key)
 
-		ctx_put, cancel := context.WithTimeout(context.Background(), defaultEtcdTimeout)
-		_, err = etcd.Put(ctx_put, key, string(yamlCmd))
+		ctx, cancel := context.WithTimeout(context.Background(), defaultEtcdTimeout)
+		err = conn.Put(ctx, key, string(yamlCmd))
 		cancel()
 
 		if err != nil {
 			return err
 		}
 
-		for ev := range outputChan {
-			result := switchCmdResult{}
-			err = yaml.Unmarshal(ev.Kv.Value, &result)
+		for ev := range watchChan {
+			var result switchCmdResult
+			err = yaml.Unmarshal(ev.Data, &result)
 			if err != nil {
 				return err
 			}
-			fmt.Printf("%s", ev.Kv.Value)
+			fmt.Printf("%s", ev)
 			if result.Status == "success" || result.Status == "failed" {
 				return nil
 			}
 		}
-		if ctx.Err() == context.DeadlineExceeded {
+		if ctxWatch.Err() == context.DeadlineExceeded {
 			log.Info("Timeout for command execution reached.")
 			return nil
 		}
 
-		return ctx.Err()
+		return fmt.Errorf("unexpected problem with watch context: %w", ctxWatch.Err())
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultEtcdTimeout)
-	_, err = etcd.Put(ctx, key, string(yamlCmd))
+	err = conn.Put(ctx, key, string(yamlCmd))
 	cancel()
 
 	if err != nil {
@@ -177,30 +147,28 @@ func SwitchStatus(url string, switchCtx SwitchStatusCtx) error {
 	if err != nil {
 		return fmt.Errorf("invalid URL %q: %w", url, err)
 	}
-
-	opts := makeEtcdOpts(uriOpts)
-
-	etcd, err := libcluster.ConnectEtcd(opts)
+	var connOpts libcluster.ConnectOpts
+	conn, err := libcluster.ConnectCStorage(uriOpts, connOpts)
 	if err != nil {
-		return fmt.Errorf("unable to connect to etcd: %w", err)
+		return fmt.Errorf("unable to connect to config storage: %w", err)
 	}
-	defer etcd.Close()
+	defer conn.Close()
 
 	key := uriOpts.Prefix + failoverPath + switchCtx.TaskID
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultEtcdTimeout)
-	result, err := etcd.Get(ctx, key, clientv3.WithLimit(1))
+	result, err := conn.Get(ctx, key)
 	cancel()
 
 	if err != nil {
 		return err
 	}
 
-	if len(result.Kvs) != 1 {
+	if len(result) != 1 {
 		return fmt.Errorf("task with id `%s` is not found", switchCtx.TaskID)
 	}
 
-	fmt.Print(string(result.Kvs[0].Value))
+	fmt.Print(string(result[0].Value))
 
 	return nil
 }

lib/cluster/configstorage.go

@@ -0,0 +1,37 @@
+package cluster
+
+import (
+	"context"
+	"fmt"
+
+	libconnect "github.com/tarantool/tt/lib/connect"
+)
+
+type CSWatchEvent struct {
+	Data []byte
+}
+
+type CSConnection interface {
+	Close() error
+	Get(ctx context.Context, key string) ([]Data, error)
+	Put(ctx context.Context, key, value string) error
+	Watch(ctx context.Context, key string) <-chan CSWatchEvent
+}
+
+func ConnectCStorage(
+	uriOpts libconnect.UriOpts,
+	connOpts ConnectOpts,
+) (CSConnection, error) {
+	sc, errEtcd := connectEtcdCS(uriOpts, connOpts)
+	if errEtcd == nil {
+		return sc, nil
+	}
+
+	sc, errTarantool := connectTarantoolCS(uriOpts, connOpts)
+	if errTarantool == nil {
+		return sc, nil
+	}
+
+	return nil, fmt.Errorf("failed to establish a connection to tarantool or etcd: %w, %w",
+		errTarantool, errEtcd)
+}

lib/cluster/etcd.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/tarantool/go-tarantool/v2"
 	libconnect "github.com/tarantool/tt/lib/connect"
+	"go.etcd.io/etcd/api/v3/mvccpb"
 	"go.etcd.io/etcd/client/pkg/v3/transport"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
@@ -882,3 +883,71 @@ func MakeEtcdOptsFromUriOpts(src libconnect.UriOpts) EtcdOpts {
 		Timeout:        src.Timeout,
 	}
 }
+
+type etcdCSConnection struct {
+	cli *clientv3.Client
+}
+
+func connectEtcdCS(uriOpts libconnect.UriOpts, connOpts ConnectOpts) (CSConnection, error) {
+	cli, err := ConnectEtcdUriOpts(uriOpts, connOpts)
+	if err != nil {
+		return nil, err
+	}
+	return &etcdCSConnection{
+		cli: cli,
+	}, nil
+}
+
+func (c *etcdCSConnection) Close() error {
+	return c.cli.Close()
+}
+
+func (c *etcdCSConnection) Get(ctx context.Context, key string) ([]Data, error) {
+	resp, err := c.cli.Get(ctx, key)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch data from etcd: %w", err)
+	}
+
+	switch {
+	case len(resp.Kvs) == 0:
+		// It should not happen, but we need to be sure to avoid a null pointer
+		// dereference.
+		return nil, fmt.Errorf("a configuration data not found in etcd for key %q",
+			key)
+	case len(resp.Kvs) > 1:
+		return nil, fmt.Errorf("too many responses (%v) from etcd for key %q",
+			resp.Kvs, key)
+	}
+
+	collected := []Data{
+		{
+			Source:   string(resp.Kvs[0].Key),
+			Value:    resp.Kvs[0].Value,
+			Revision: resp.Kvs[0].ModRevision,
+		},
+	}
+	return collected, nil
+}
+
+func (c *etcdCSConnection) Put(ctx context.Context, key, value string) error {
+	_, err := c.cli.Put(ctx, key, value)
+	return err
+}
+
+func (c *etcdCSConnection) Watch(ctx context.Context, key string) <-chan CSWatchEvent {
+	ch := make(chan CSWatchEvent)
+	innerCh := c.cli.Watch(ctx, key)
+	go func() {
+		defer close(ch)
+
+		for resp := range innerCh {
+			for _, ev := range resp.Events {
+				switch ev.Type {
+				case mvccpb.PUT:
+					ch <- CSWatchEvent{Data: ev.Kv.Value}
+				}
+			}
+		}
+	}()
+	return ch
+}

lib/cluster/tarantool.go

@@ -618,3 +618,82 @@ func ConnectTarantool(uriOpts libconnect.UriOpts,
 	}
 	return conn, nil
 }
+
+type tarantoolCSConnection struct {
+	conn tarantool.Connector
+}
+
+func connectTarantoolCS(uriOpts libconnect.UriOpts, connOpts ConnectOpts) (CSConnection, error) {
+	conn, err := ConnectTarantool(uriOpts, connOpts)
+	if err != nil {
+		return nil, err
+	}
+	return &tarantoolCSConnection{
+		conn: conn,
+	}, nil
+}
+
+func (c *tarantoolCSConnection) Close() error {
+	return c.conn.Close()
+}
+
+func (c *tarantoolCSConnection) Get(ctx context.Context, key string) ([]Data, error) {
+	data, err := c.call(ctx, "config.storage.get", []any{key})
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch data from tarantool: %w", err)
+	}
+	if len(data) != 1 {
+		return nil, fmt.Errorf("unexpected response from tarantool: %q", data)
+	}
+
+	var resp tarantoolGetResponse
+	if err := mapstructure.Decode(data[0], &resp); err != nil {
+		return nil, fmt.Errorf("failed to map response from tarantool: %q", data[0])
+	}
+
+	switch {
+	case len(resp.Data) == 0:
+		return nil, fmt.Errorf("a configuration data not found in tarantool for key %q",
+			key)
+	case len(resp.Data) > 1:
+		// It should not happen, but we need to be sure to avoid a null pointer
+		// dereference.
+		return nil, fmt.Errorf("too many responses (%v) from tarantool for key %q",
+			resp, key)
+	}
+
+	return []Data{
+		{
+			Source:   key,
+			Value:    []byte(resp.Data[0].Value),
+			Revision: resp.Data[0].ModRevision,
+		},
+	}, err
+}
+
+func (c *tarantoolCSConnection) Put(ctx context.Context, key, value string) error {
+	_, err := c.call(ctx, "config.storage.put", []any{key, value})
+	if err != nil {
+		return fmt.Errorf("failed to put data into tarantool: %w", err)
+	}
+	return nil
+}
+
+func (c *tarantoolCSConnection) Watch(ctx context.Context, key string) <-chan CSWatchEvent {
+	return nil
+}
+
+func (c *tarantoolCSConnection) call(
+	ctx context.Context,
+	fun string,
+	args []any,
+) ([]any, error) {
+	req := tarantool.NewCallRequest(fun).Args(args)
+
+	var result []any
+	if err := c.conn.Do(req.Context(ctx)).GetTyped(&result); err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}