cluster,dm: auto detect cpu arch (#1423)

.github/workflows/integrate-cluster-cmd.yaml

@@ -33,9 +33,9 @@ jobs:
       fail-fast: true
       matrix:
         cases:
-          - "test_cmd"
+          - 'test_cmd'
           # - "test_cmd_tls_native_ssh"
-          - "test_upgrade"
+          - 'test_upgrade'
           # - "test_upgrade_tls"
     env:
       working-directory: ${{ github.workspace }}/go/src/github.com/${{ github.repository }}
@@ -73,6 +73,8 @@ jobs:
         id: test
         working-directory: ${{ env.working-directory }}
         run: |
+          # ensuere docker ssh is ready
+          sleep 3
           # should not use -it
           # ref: https://stackoverflow.com/questions/43099116/error-the-input-device-is-not-a-tty
           docker exec tiup-cluster-control bash /tiup-cluster/tests/tiup-cluster/run.sh ${{ matrix.cases }}

.github/workflows/integrate-cluster-scale.yaml

@@ -33,10 +33,10 @@ jobs:
       fail-fast: true
       matrix:
         cases:
-          - "test_scale_core"
-          - "test_scale_tools"
-          - "test_scale_core_tls"
-          - "test_scale_tools_tls"
+          - 'test_scale_core'
+          - 'test_scale_tools'
+          - 'test_scale_core_tls'
+          - 'test_scale_tools_tls'
     env:
       working-directory: ${{ github.workspace }}/go/src/github.com/${{ github.repository }}
     steps:
@@ -73,6 +73,8 @@ jobs:
         id: test
         working-directory: ${{ env.working-directory }}
         run: |
+          # ensuere docker ssh is ready
+          sleep 5
           # should not use -it
           # ref: https://stackoverflow.com/questions/43099116/error-the-input-device-is-not-a-tty
           docker exec tiup-cluster-control bash /tiup-cluster/tests/tiup-cluster/run.sh ${{ matrix.cases }}

.github/workflows/integrate-dm.yaml

@@ -33,10 +33,10 @@ jobs:
       fail-fast: true
       matrix:
         cases:
-          - "--do-cases test_cmd"
-          - "--do-cases test_upgrade"
-          - "--native-ssh --do-cases test_cmd"
-          - "--native-ssh --do-cases test_upgrade"
+          - '--do-cases test_cmd'
+          - '--do-cases test_upgrade'
+          - '--native-ssh --do-cases test_cmd'
+          - '--native-ssh --do-cases test_upgrade'
     env:
       working-directory: ${{ github.workspace }}/go/src/github.com/${{ github.repository }}
     steps:
@@ -77,6 +77,8 @@ jobs:
         id: test
         working-directory: ${{ env.working-directory }}
         run: |
+          # ensuere docker ssh is ready
+          sleep 4
           # should not use -it
           # ref: https://stackoverflow.com/questions/43099116/error-the-input-device-is-not-a-tty
           docker exec tiup-cluster-control bash /tiup-cluster/tests/tiup-dm/run.sh ${{ matrix.cases }}

components/dm/spec/topology_dm.go

@@ -673,6 +673,11 @@ func (s *Specification) GetMasterList() []string {
 	return masterList
 }
 
+// FillHostArch fills the topology with the given host->arch
+func (s *Specification) FillHostArch(hostArch map[string]string) error {
+	return spec.FillHostArch(s, hostArch)
+}
+
 // Merge returns a new Topology which sum old ones
 func (s *Specification) Merge(that spec.Topology) spec.Topology {
 	spec := that.(*Specification)

components/dm/spec/topology_dm_test.go

@@ -181,6 +181,7 @@ master_servers:
     arch: "aarch64"
 worker_servers:
   - host: 172.16.5.138
+    arch: "amd64"
 `), &topo)
 	assert.NotNil(t, err)
 	assert.Equal(t, "platform mismatch for '172.16.5.138' between 'master_servers:linux/arm64' and 'worker_servers:linux/amd64'", err.Error())

pkg/cluster/ansible/import.go

@@ -61,7 +61,9 @@ func parseInventoryFile(invFile io.Reader) (string, *spec.ClusterMeta, *aini.Inv
 
 	clsMeta := &spec.ClusterMeta{
 		Topology: &spec.Specification{
-			GlobalOptions:    spec.GlobalOptions{},
+			GlobalOptions: spec.GlobalOptions{
+				Arch: "amd64",
+			},
 			MonitoredOptions: spec.MonitoredOptions{},
 			TiDBServers:      make([]*spec.TiDBSpec, 0),
 			TiKVServers:      make([]*spec.TiKVSpec, 0),

pkg/cluster/ansible/import_test.go

@@ -56,7 +56,7 @@ process_supervision = systemd
 172.16.10.1
 172.16.10.2
 172.16.10.3
-	
+
 [all:vars]
 deploy_dir = /data1/deploy
 process_supervision = systemd
@@ -71,7 +71,7 @@ process_supervision = systemd
 172.16.10.1 deploy_dir=/data/deploy
 172.16.10.2 deploy_dir=/data/deploy
 172.16.10.3 deploy_dir=/data/deploy
-	
+
 [all:vars]
 deploy_dir = /data1/deploy
 process_supervision = systemd
@@ -86,7 +86,7 @@ process_supervision = systemd
 172.16.10.1 deploy_dir=/data/deploy1
 172.16.10.2 deploy_dir=/data/deploy2
 172.16.10.3 deploy_dir=/data/deploy3
-	
+
 [all:vars]
 deploy_dir = /data1/deploy
 process_supervision = systemd
@@ -113,6 +113,7 @@ func (s *ansSuite) TestParseInventoryFile(c *C) {
 	expected := []byte(`global:
     user: tiops
     deploy_dir: /home/tiopsimport/ansible-deploy
+    arch: amd64
 monitored:
     deploy_dir: /home/tiopsimport/ansible-deploy
     data_dir: /home/tiopsimport/ansible-deploy/data

pkg/cluster/ansible/inventory.go

@@ -176,6 +176,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			if port, ok := grpVars["tidb_port"]; ok {
@@ -218,6 +219,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			if port, ok := grpVars["tikv_port"]; ok {
@@ -263,6 +265,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 			if tmpIns.Host != srv.Name {
 				tmpIns.Name = srv.Name // use alias as the name of PD
@@ -311,6 +314,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			if tcpPort, ok := grpVars["tcp_port"]; ok {
@@ -388,6 +392,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			if port, ok := grpVars["prometheus_port"]; ok {
@@ -433,6 +438,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			if port, ok := grpVars["alertmanager_port"]; ok {
@@ -472,6 +478,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			if port, ok := grpVars["grafana_port"]; ok {
@@ -516,6 +523,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			// nothing in pump_servers.yml
@@ -557,6 +565,7 @@ func parseGroupVars(dir, ansCfgFile string, clsMeta *spec.ClusterMeta, inv *aini
 				Host:     host,
 				SSHPort:  getHostPort(srv, ansCfg),
 				Imported: true,
+				Arch:     "amd64",
 			}
 
 			// nothing in drainer_servers.yml

pkg/cluster/manager/check.go

@@ -97,6 +97,10 @@ func (m *Manager) CheckCluster(clusterOrTopoName string, opt CheckOptions, gOpt
 		}
 	}
 
+	if err := m.fillHostArch(sshConnProps, &topo, &gOpt, opt.User); err != nil {
+		return err
+	}
+
 	if err := checkSystemInfo(sshConnProps, &topo, &gOpt, &opt); err != nil {
 		return err
 	}

pkg/cluster/manager/deploy.go

@@ -137,12 +137,6 @@ func (m *Manager) Deploy(
 		return err
 	}
 
-	if !skipConfirm {
-		if err := m.confirmTopology(name, clusterVersion, topo, set.NewStringSet()); err != nil {
-			return err
-		}
-	}
-
 	var sshConnProps *tui.SSHConnectionProps = &tui.SSHConnectionProps{}
 	if gOpt.SSHType != executor.SSHTypeNone {
 		var err error
@@ -151,6 +145,16 @@ func (m *Manager) Deploy(
 		}
 	}
 
+	if err := m.fillHostArch(sshConnProps, topo, &gOpt, opt.User); err != nil {
+		return err
+	}
+
+	if !skipConfirm {
+		if err := m.confirmTopology(name, clusterVersion, topo, set.NewStringSet()); err != nil {
+			return err
+		}
+	}
+
 	if err := os.MkdirAll(m.specManager.Path(name), 0755); err != nil {
 		return errorx.InitializationFailed.
 			Wrap(err, "Failed to create cluster metadata directory '%s'", m.specManager.Path(name)).

pkg/cluster/manager/manager.go

@@ -14,12 +14,14 @@
 package manager
 
 import (
+	"context"
 	"fmt"
 	"strings"
 
 	"github.com/fatih/color"
 	"github.com/joomcode/errorx"
 	perrs "github.com/pingcap/errors"
+	"github.com/pingcap/tiup/pkg/cluster/ctxt"
 	operator "github.com/pingcap/tiup/pkg/cluster/operation"
 	"github.com/pingcap/tiup/pkg/cluster/spec"
 	"github.com/pingcap/tiup/pkg/cluster/task"
@@ -133,3 +135,54 @@ func (m *Manager) sshTaskBuilder(name string, topo spec.Topology, user string, o
 		).
 		ClusterSSH(topo, user, opts.SSHTimeout, opts.SSHType, topo.BaseTopo().GlobalOptions.SSHType)
 }
+
+func (m *Manager) fillHostArch(s *tui.SSHConnectionProps, topo spec.Topology, gOpt *operator.Options, user string) error {
+	hostArch := map[string]string{}
+	var detectTasks []*task.StepDisplay
+	topo.IterInstance(func(inst spec.Instance) {
+		if _, ok := hostArch[inst.GetHost()]; ok {
+			return
+		}
+		hostArch[inst.GetHost()] = ""
+		if inst.Arch() != "" {
+			return
+		}
+
+		tf := task.NewBuilder().
+			RootSSH(
+				inst.GetHost(),
+				inst.GetSSHPort(),
+				user,
+				s.Password,
+				s.IdentityFile,
+				s.IdentityFilePassphrase,
+				gOpt.SSHTimeout,
+				gOpt.SSHType,
+				topo.BaseTopo().GlobalOptions.SSHType,
+			).
+			Shell(inst.GetHost(), "uname -m", "", false).
+			BuildAsStep(fmt.Sprintf("  - Detecting node %s", inst.GetHost()))
+		detectTasks = append(detectTasks, tf)
+	})
+	if len(detectTasks) == 0 {
+		return nil
+	}
+
+	ctx := ctxt.New(context.Background(), gOpt.Concurrency)
+	t := task.NewBuilder().
+		ParallelStep("+ Detect CPU Arch", false, detectTasks...).
+		Build()
+
+	if err := t.Execute(ctx); err != nil {
+		return perrs.Annotate(err, "failed to fetch cpu arch")
+	}
+
+	for host := range hostArch {
+		stdout, _, ok := ctxt.GetInner(ctx).GetOutputs(host)
+		if !ok {
+			return fmt.Errorf("no check results found for %s", host)
+		}
+		hostArch[host] = strings.Trim(string(stdout), "\n")
+	}
+	return topo.FillHostArch(hostArch)
+}

pkg/cluster/manager/scale_out.go

@@ -84,6 +84,18 @@ func (m *Manager) ScaleOut(
 		return err
 	}
 
+	var sshConnProps *tui.SSHConnectionProps = &tui.SSHConnectionProps{}
+	if gOpt.SSHType != executor.SSHTypeNone {
+		var err error
+		if sshConnProps, err = tui.ReadIdentityFileOrPassword(opt.IdentityFile, opt.UsePassword); err != nil {
+			return err
+		}
+	}
+
+	if err := m.fillHostArch(sshConnProps, newPart, &gOpt, opt.User); err != nil {
+		return err
+	}
+
 	// Abort scale out operation if the merged topology is invalid
 	mergedTopo := topo.MergeTopo(newPart)
 	if err := mergedTopo.Validate(); err != nil {
@@ -138,14 +150,6 @@ func (m *Manager) ScaleOut(
 		}
 	}
 
-	var sshConnProps *tui.SSHConnectionProps = &tui.SSHConnectionProps{}
-	if gOpt.SSHType != executor.SSHTypeNone {
-		var err error
-		if sshConnProps, err = tui.ReadIdentityFileOrPassword(opt.IdentityFile, opt.UsePassword); err != nil {
-			return err
-		}
-	}
-
 	// Build the scale out tasks
 	t, err := buildScaleOutTask(
 		m, name, metadata, mergedTopo, opt, sshConnProps, newPart,