cluster: add tt cluster command

* `tt cluster show` to show a configuration.
* `tt cluster public` to public a configuration.

Closes #506

Author: oleg-jukovec

.github/actions/prepare-ce-test-env/action.yml

@@ -14,7 +14,7 @@ inputs:
     description: Whether to skip etcd installation
     type: boolean
     required: false
-    default: true
+    default: false
 
 runs:
   using: "composite"

.github/actions/prepare-ee-test-env/action.yml

@@ -12,7 +12,7 @@ inputs:
     description: Whether to skip etcd installation
     type: boolean
     required: false
-    default: true
+    default: false
 
 runs:
   using: "composite"

.github/workflows/full-ci.yml

@@ -38,7 +38,6 @@ jobs:
         uses: ./.github/actions/prepare-ce-test-env
         with:
           tarantool-version: '${{ matrix.tarantool-version }}'
-          skip-etcd-install: false
 
       - name: Static code check
         uses: ./.github/actions/static-code-check
@@ -93,7 +92,6 @@ jobs:
         with:
           sdk-version: '${{ matrix.sdk-version }}'
           sdk-download-token: '${{ secrets.SDK_DOWNLOAD_TOKEN }}'
-          skip-etcd-install: false
 
       - name: Static code check
         uses: ./.github/actions/static-code-check

.github/workflows/tests.yml

@@ -127,6 +127,9 @@ jobs:
           brew install --overwrite go mage
           pip3 install -r test/requirements.txt
 
+      - name: Install etcd
+        uses: ./.github/actions/setup-etcd
+
       - name: Build tt
         env:
           TT_CLI_BUILD_SSL: 'static'

CHANGELOG.md

@@ -26,6 +26,8 @@ with the cartridge-cli. It is supported only by `tgz` type packing.
   modules into the pack bundle.
 - `tt connect`: added command `\shortcuts` listing all available 
   shortcuts and hotkeys in go-prompt.
+- Module ``tt cluster``, to show or publish a cluster or an instance
+  configuration.
 
 ### Fixed
 

cli/cluster/cluster.go

@@ -191,7 +191,14 @@ func MakeClusterConfig(config *Config) (ClusterConfig, error) {
 	}
 
 	err := yaml.Unmarshal([]byte(config.String()), &cconfig)
-	return cconfig, err
+	if err != nil {
+		err = fmt.Errorf(
+			"failed to parse a configuration data as a cluster config: %w",
+			err)
+		return cconfig, err
+
+	}
+	return cconfig, nil
 }
 
 // mergeExclude merges a high priority configuration with a low priority
@@ -372,3 +379,30 @@ func GetInstanceConfig(cluster ClusterConfig, instance string) (InstanceConfig,
 
 	return MakeInstanceConfig(iconfig)
 }
+
+// ReplaceInstanceConfig replaces an instance configuration.
+func ReplaceInstanceConfig(cconfig ClusterConfig,
+	instance string, iconfig *Config) (ClusterConfig, error) {
+	for gname, group := range cconfig.Groups {
+		for rname, replicaset := range group.Replicasets {
+			for iname, _ := range replicaset.Instances {
+				if instance == iname {
+					path := []string{groupsLabel, gname,
+						replicasetsLabel, rname,
+						instancesLabel, iname,
+					}
+					newConfig := NewConfig()
+					newConfig.Merge(cconfig.RawConfig)
+					if err := newConfig.Set(path, iconfig); err != nil {
+						err = fmt.Errorf("failed to set configuration: %w", err)
+						return cconfig, err
+					}
+					return MakeClusterConfig(newConfig)
+				}
+			}
+		}
+	}
+
+	return cconfig,
+		fmt.Errorf("cluster configuration have not an instance %q", instance)
+}

cli/cluster/cluster_test.go

@@ -472,3 +472,35 @@ func TestGetInstanceConfig_noinstance(t *testing.T) {
 
 	assert.EqualError(t, err, expected)
 }
+
+func TestReplcaseInstanceConfig_not_found(t *testing.T) {
+	config := cluster.NewConfig()
+	cconfig, err := cluster.MakeClusterConfig(config)
+	require.NoError(t, err)
+
+	cconfig, err = cluster.ReplaceInstanceConfig(cconfig, "any", cluster.NewConfig())
+	assert.EqualError(t, err, "cluster configuration have not an instance \"any\"")
+}
+
+func TestReplcaseInstanceConfig_replcase(t *testing.T) {
+	path := []string{"groups", "a", "replicasets", "b", "instances", "c", "foo"}
+	config := cluster.NewConfig()
+	err := config.Set(path, 1)
+	require.NoError(t, err)
+	old, err := cluster.MakeClusterConfig(config)
+	require.NoError(t, err)
+
+	replace := cluster.NewConfig()
+	err = replace.Set([]string{"foo"}, 2)
+	require.NoError(t, err)
+
+	newConfig, err := cluster.ReplaceInstanceConfig(old, "c", replace)
+	require.NoError(t, err)
+	oldValue, err := old.RawConfig.Get(path)
+	require.NoError(t, err)
+	newValue, err := newConfig.RawConfig.Get(path)
+	require.NoError(t, err)
+
+	assert.Equal(t, 1, oldValue)
+	assert.Equal(t, 2, newValue)
+}

cli/cluster/config.go

@@ -87,7 +87,12 @@ func (config *Config) Set(path []string, value any) error {
 	if m, err := config.createMaps(path[0:last]); err != nil {
 		return err
 	} else {
-		m[path[last]] = value
+		key := path[last]
+		if cfg, ok := value.(*Config); value != nil && ok {
+			m[key] = cfg.paths
+		} else {
+			m[key] = value
+		}
 	}
 
 	return nil

cli/cluster/config_test.go

@@ -42,6 +42,27 @@ func TestConfig_Set(t *testing.T) {
 	}
 }
 
+func TestConfig_Set_config(t *testing.T) {
+	base := cluster.NewConfig()
+	err := base.Set([]string{"foo", "bar", "zoo", "foo"}, 1)
+	require.NoError(t, err)
+
+	add := cluster.NewConfig()
+	err = add.Set([]string{"foo", "bar"}, 2)
+	require.NoError(t, err)
+
+	err = base.Set([]string{"foo", "bar", "zoo"}, add)
+	require.NoError(t, err)
+
+	valueAdd, err := add.Get([]string{"foo", "bar"})
+	require.NoError(t, err)
+	assert.Equal(t, 2, valueAdd)
+
+	valueAdded, err := base.Get([]string{"foo", "bar", "zoo", "foo", "bar"})
+	require.NoError(t, err)
+	assert.Equal(t, 2, valueAdded)
+}
+
 func TestConfig_Set_intersection(t *testing.T) {
 	base := [][]string{
 		nil,

cli/cluster/etcd.go

@@ -92,13 +92,8 @@ func MakeEtcdOpts(config *Config) (EtcdOpts, error) {
 		opts.SkipHostVerify = true
 	}
 	if parsed.Http.Request.Timeout != 0 {
-		var err error
-		timeout := fmt.Sprintf("%fs", parsed.Http.Request.Timeout)
-		opts.Timeout, err = time.ParseDuration(timeout)
-		if err != nil {
-			fmtErr := "unable to parse a etcd request timeout: %w"
-			return EtcdOpts{}, fmt.Errorf(fmtErr, err)
-		}
+		timeoutFloat := parsed.Http.Request.Timeout * float64(time.Second)
+		opts.Timeout = time.Duration(timeoutFloat)
 	}
 
 	return opts, nil
@@ -169,21 +164,25 @@ func NewEtcdCollector(getter EtcdGetter, prefix string, timeout time.Duration) E
 // Collect collects a configuration from the specified path with the specified
 // timeout.
 func (collector EtcdCollector) Collect() (*Config, error) {
-	prefix := strings.TrimRight(collector.prefix, "/")
-	key := fmt.Sprintf("%s/%s/", prefix, "config")
+	prefix := getConfigPrefix(collector.prefix)
 	ctx := context.Background()
 	if collector.timeout != 0 {
 		var cancel context.CancelFunc
 		ctx, cancel = context.WithTimeout(ctx, collector.timeout)
 		defer cancel()
 	}
 
-	resp, err := collector.getter.Get(ctx, key, clientv3.WithPrefix())
+	resp, err := collector.getter.Get(ctx, prefix, clientv3.WithPrefix())
 	if err != nil {
 		return nil, fmt.Errorf("failed to fetch data from etcd: %w", err)
 	}
 
 	cconfig := NewConfig()
+	if len(resp.Kvs) == 0 {
+		return nil, fmt.Errorf("a configuration data not found in prefix %q",
+			prefix)
+	}
+
 	for _, kv := range resp.Kvs {
 		if config, err := NewYamlCollector(kv.Value).Collect(); err != nil {
 			fmtErr := "failed to decode etcd config for key %q: %w"
@@ -196,6 +195,99 @@ func (collector EtcdCollector) Collect() (*Config, error) {
 	return cconfig, nil
 }
 
+// EtcdDataPublisher publishes a data into etcd.
+type EtcdDataPublisher struct {
+	getter  EtcdTxnGetter
+	prefix  string
+	timeout time.Duration
+}
+
+// EtcdUpdater is the interface that adds Txn method to EtcdGetter.
+type EtcdTxnGetter interface {
+	EtcdGetter
+	// Txn creates a transaction.
+	Txn(ctx context.Context) clientv3.Txn
+}
+
+// NewEtcdDataPublisher creates a new EtcdDataPublisher object to publish
+// a data to etcd with the prefix during the timeout.
+func NewEtcdDataPublisher(getter EtcdTxnGetter,
+	prefix string, timeout time.Duration) EtcdDataPublisher {
+	return EtcdDataPublisher{
+		getter:  getter,
+		prefix:  prefix,
+		timeout: timeout,
+	}
+}
+
+// Publish publishes the configuration into etcd to the given prefix.
+func (publisher EtcdDataPublisher) Publish(data []byte) error {
+	if data == nil {
+		return fmt.Errorf("failed to publish data into etcd: data does not exist")
+	}
+
+	prefix := getConfigPrefix(publisher.prefix)
+	key := prefix + "all"
+	ctx := context.Background()
+	if publisher.timeout != 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, publisher.timeout)
+		defer cancel()
+	}
+
+	for true {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		resp, err := publisher.getter.Get(ctx, prefix, clientv3.WithPrefix())
+		if err != nil {
+			return fmt.Errorf("failed to fetch data from etcd: %w", err)
+		}
+
+		var (
+			keys      []string
+			revisions []int64
+		)
+		for _, kv := range resp.Kvs {
+			// We need to skip the target key since etcd transactions do not
+			// support delete + put for the same key on a single transaction.
+			if string(kv.Key) != key {
+				keys = append(keys, string(kv.Key))
+				revisions = append(revisions, kv.ModRevision)
+			}
+		}
+
+		var (
+			cmps []clientv3.Cmp
+			opts []clientv3.Op
+		)
+		for i, key := range keys {
+			cmp := clientv3.Compare(clientv3.ModRevision(key), "=", revisions[i])
+			cmps = append(cmps, cmp)
+			opts = append(opts, clientv3.OpDelete(key))
+		}
+
+		opts = append(opts, clientv3.OpPut(key, string(data)))
+		txn := publisher.getter.Txn(ctx)
+		if len(cmps) > 0 {
+			txn = txn.If(cmps...)
+		}
+		tresp, err := txn.Then(opts...).Commit()
+
+		if err != nil {
+			return fmt.Errorf("failed to put data into etcd: %w", err)
+		}
+		if tresp != nil && tresp.Succeeded {
+			return nil
+		}
+	}
+	// Unreachable.
+	return nil
+}
+
 // loadRootCA and the code below is a copy-paste from Go sources. We need an
 // ability to load ceritificates from a directory, but there is no such public
 // function in `crypto`.
@@ -253,3 +345,9 @@ func isSameDirSymlink(f fs.DirEntry, dir string) bool {
 	target, err := os.Readlink(filepath.Join(dir, f.Name()))
 	return err == nil && !strings.Contains(target, "/")
 }
+
+// getConfigPrefix returns a full configuration prefix.
+func getConfigPrefix(basePrefix string) string {
+	prefix := strings.TrimRight(basePrefix, "/")
+	return fmt.Sprintf("%s/%s/", prefix, "config")
+}