cluster: support switching master over TcS

Prior to this patch `failover switch` and `failover switch-status`
commands worked only with etcd config storage. Now they also work with
Tarantool 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.Value, &result)
 			if err != nil {
 				return err
 			}
-			fmt.Printf("%s", ev.Kv.Value)
+			fmt.Printf("%s", ev.Value)
 			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
 }

cli/cmd/cluster.go

@@ -82,8 +82,10 @@ environment variables < command flags < URL credentials.`,
 	})
 
 	failoverUriHelp = libconnect.MakeURLHelp(map[string]any{
-		"service":              "etcd",
-		"prefix":               "a base path to Tarantool configuration in etcd",
+		"service": "etcd or tarantool config storage",
+		"prefix": "a base path to Tarantool configuration in" +
+			" etcd or tarantool config storage",
+		"env_TT_CLI_auth":      "Tarantool",
 		"env_TT_CLI_ETCD_auth": "Etcd",
 		"footer": `The priority of credentials:
 environment variables < command flags < URL credentials.`,
@@ -235,9 +237,9 @@ func newClusterFailoverCmd() *cobra.Command {
 	}
 
 	switchCmd.Flags().StringVarP(&switchCtx.Username, "username", "u", "",
-		"username (used as etcd credentials)")
+		"username (used as etcd/tarantool config storage credentials)")
 	switchCmd.Flags().StringVarP(&switchCtx.Password, "password", "p", "",
-		"password (used as etcd credentials)")
+		"password (used as etcd/tarantool config storage credentials)")
 	switchCmd.Flags().Uint64VarP(&switchCtx.Timeout, "timeout", "t", defaultSwitchTimeout,
 		"timeout for command execution")
 	switchCmd.Flags().BoolVarP(&switchCtx.Wait, "wait", "w", false,

lib/cluster/configstorage.go

@@ -0,0 +1,44 @@
+package cluster
+
+import (
+	"context"
+	"fmt"
+
+	libconnect "github.com/tarantool/tt/lib/connect"
+)
+
+type CSWatchEvent struct {
+	Key   string
+	Value []byte
+}
+
+// CSConnection interface is to be used to implement access to config storage.
+type CSConnection interface {
+	// Close closes connection.
+	Close() error
+	// Get retrieves value for key.
+	Get(ctx context.Context, key string) ([]Data, error)
+	// Put puts a key-value pair into config storage.
+	Put(ctx context.Context, key, value string) error
+	// Watch watches on a key and return watched events through the returned channel.
+	Watch(ctx context.Context, key string) <-chan CSWatchEvent
+}
+
+// ConnectCStorage connects to config storage according to connection options.
+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,78 @@ 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
+}
+
+// Close implements ConnectCStorage interface.
+func (c *etcdCSConnection) Close() error {
+	return c.cli.Close()
+}
+
+// Get implements ConnectCStorage interface.
+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
+}
+
+// Put implements ConnectCStorage interface.
+func (c *etcdCSConnection) Put(ctx context.Context, key, value string) error {
+	_, err := c.cli.Put(ctx, key, value)
+	return err
+}
+
+// Watch implements ConnectCStorage interface.
+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{
+						Key:   string(ev.Kv.Value),
+						Value: ev.Kv.Value,
+					}
+				}
+			}
+		}
+	}()
+	return ch
+}