TiUP 源码初探

想必tidber都使用过tiup，tiup给部署带来很大的便利性，在日常开发中也会涉及到分布式的部署，自己难免也会写一些工具，本文抛砖引玉，分享一下自己看tiup源码的一些步骤，希望您在看源码或写工具时多一个思考方向。环境部署可参考：启航TiDB：调试环境搭建（vscode+wsl+pd）

Cobra

Cobra 是一个 Go 语言开发的命令行（CLI）框架，是由 Go 团队成员 spf13 为 Hugo 项目创建的，并已被许多流行的 Go 项目所采用，如 Kubernetes、Helm、Docker (distribution)、Etcd 等。而Tiup也是以cobra为基础，进行开发的。网上有大量的介绍文章，这里就不赘述了，直接放一个简单的demo。

demo
├── cmd
│   └── root.go
|   └── version.go
├── go.mod
├── go.sum
└── main.go

// root.go
var (
	rootCmd *cobra.Command
)

func init() {
	rootCmd = &cobra.Command{
		Use:   "demo",
		Short: "Demo is a Cobra application",
		Long:  `This is a demo application to illustrate the use of Cobra library.`,
		Run: func(cmd *cobra.Command, args []string) {
			fmt.Println("Hello, Cobra!")
		},
	}
	rootCmd.AddCommand(newVersionCmd())
}
func Execute() {
	if err := rootCmd.Execute(); err != nil {
		fmt.Println(err)
		os.Exit(1)
	}
}

// version.go
func newVersionCmd() *cobra.Command {
	cmd := &cobra.Command{
		Use:   "version",
		Short: "Print the version number of Demo",
		Long:  `All software has versions. This is Demo's`,
		Run: func(cmd *cobra.Command, args []string) {
			fmt.Println("Demo v0.1 -- HEAD")
		},
	}
	return cmd
}

func init() {
	rootCmd.AddCommand(versionCmd)
}
// main.go
func main() {
	cmd.Execute()
}

编译后运行

> .\\demo.exe -h
This is a demo application to illustrate the use of Cobra library.

Usage:
  demo [flags]
  demo [command]

Available Commands:
  completion  Generate the autocompletion script for the specified shell
  help        Help about any command
  version     Print the version number of Demo

Flags:
  -h, --help   help for demo

Use "demo [command] --help" for more information about a command.

同样，Tiup的组织结构也很简单明确，在cmd文件夹下放着各种命令，root.go为注册命令的地方，目录结构如下:

├── cmd
│   ├── env.go
│   ├── list.go
│   ├── mirror.go
│   ├── root.go   # 命令注册
├── components  # 组件
│   ├── bench
│   ├── cluster   
|   |   |── command
|   |	│   ├── clean.go
|   |	│   ├── deploy.go
|   |	│   ├── root.go # cluster 命令注册
|   |   |—— main.go
│   └── playground
├── pkg
│   ├── cluster
│   │   ├── ansible
│   │   ├── api
│   │   ├── audit
│   │   ├── clusterutil
│   │   ├── ctxt
│   │   ├── executor   # 执行器，easyssh和nativessh
│   │   ├── manager    # 任务的生成，步骤的生成所有的管理都在这里
│   │   ├── module
│   │   ├── operation
│   │   ├── spec       # 拓扑结构--安装的说明书
│   │   ├── task       # 各种任务
│   │   └── template   # 
├── main.go

和demo不同的是，Tiup的二级命令是通过调用对应的执行文件来实现，这些执行文件在$TIUPHOME/components/cluster/vx.x.x/下，在执行的时候如果没有对应的子命令就会下载（只用tiup-cluster,就不会有tiup-dm），另一个好处是每一个子命令都是一个独立的命令行工具—可插拔组件。

// root.go
func init() {
	cobra.EnableCommandSorting = false
	_ = os.Setenv(localdata.EnvNameTelemetryEventUUID, eventUUID)

	rootCmd = &cobra.Command{
		Use: `tiup [flags]  [args...]`
	  // ...
		},
		PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
			// InitEnv
			return nil
		},
		RunE: func(cmd *cobra.Command, args []string) error {
			if len(args) == 0 {
				return cmd.Help()
			}
			env := environment.GlobalEnv()
			// ...
			return tiupexec.RunComponent(env, tag, componentSpec, binPath, args)
		},
	}
}

而这些命令最小的执行单位是任务。

任务

任务会有执行和回滚两个操作，在操作的时候有顺序和并行的两种任务，builder来构建不同的任务

type (
	// Task represents a operation while TiUP execution
	Task interface {
		fmt.Stringer
		Execute(ctx context.Context) error
		Rollback(ctx context.Context) error
	}

	// Serial will execute a bundle of task in serialized way
	Serial struct {
		ignoreError       bool
		hideDetailDisplay bool
		inner             []Task
	}

	// Parallel will execute a bundle of task in parallelism way
	Parallel struct {
		ignoreError       bool
		hideDetailDisplay bool
		inner             []Task
	}
)

任务也可以作为一种方法为其它任务调用比如：CopyComponent，任务最后都会调用执行器来结束任务，其实就是linux上的命令。

func (m *Mkdir) Execute(ctx context.Context) error {
 exec, found := ctxt.GetInner(ctx).GetExecutor(m.host)
	// ...
	_, _, err := exec.Execute(ctx, cmd, m.sudo) // use root to create the dir
	if err != nil {
		return errors.Trace(err)
	}
	// ...
	return nil
}

func (c *CopyComponent) Execute(ctx context.Context) error {
	// ...
	install := &InstallPackage{
		srcPath: srcPath,
		host:    c.host,
		dstDir:  c.dstDir,
	}
	return install.Execute(ctx)
}

func (c *InstallPackage) Execute(ctx context.Context) error {
	// Install package to remote server
	exec, found := ctxt.GetInner(ctx).GetExecutor(c.host)
	if !found {
		return ErrNoExecutor
	}
	dstDir := filepath.Join(c.dstDir, "bin")
	dstPath := filepath.Join(dstDir, path.Base(c.srcPath))

	err := exec.Transfer(ctx, c.srcPath, dstPath, false, 0, false)
	if err != nil {
		return errors.Annotatef(err, "failed to scp %s to %s:%s", c.srcPath, c.host, dstPath)
	}
	cmd := fmt.Sprintf(`tar --no-same-owner -zxf %s -C %s && rm %s`, dstPath, dstDir, dstPath)

	_, stderr, err := exec.Execute(ctx, cmd, false)
	if err != nil {
		return errors.Annotatef(err, "stderr: %s", string(stderr))
	}
	return nil
}

执行器

执行器放在上下文上的，它包含两个方法，执行器的设置是在需要SSH的时候，比如下面的RootSSH

Executor interface {
		Execute(ctx context.Context, cmd string, sudo bool, timeout ...time.Duration) (stdout []byte, stderr []byte, err error)
		Transfer(ctx context.Context, src, dst string, download bool, limit int, compress bool) error
	}

// Execute implements the Task interface
func (s *RootSSH) Execute(ctx context.Context) error {
	// ...
	e, err := executor.New(s.sshType, s.sudo, sc)
	if err != nil {
		return err
	}

	ctxt.GetInner(ctx).SetExecutor(s.host, e)
	return nil
}

执行器有easyssh和nativessh两种

// Deploy a cluster.
func (m *Manager) Deploy(
	name string,
	clusterVersion string,
	topoFile string,
	opt DeployOptions,
	afterDeploy func(b *task.Builder, newPart spec.Topology, gOpt operator.Options),
	skipConfirm bool,
	gOpt operator.Options,
) error

操作

可以看到tiup的基础就是任务，任务结束于执行器—ssh，多个任务就合并成一个步骤，多个步骤组合成一个操作：task.NewBuilder(m.logger).Atask().Btask.AS，下面我们看一个实际的操作Deploy

每一个操作都放在manager文件夹下，这里以Deploy为例，可以看到任务组一般都是以ssh任务开始，显示任务结束。

// command/root.go
func init() {
	rootCmd.AddCommand(
		newCheckCmd(),
		newDeploy(),
		// ...
		)
}
// command/deploy.go
func newDeploy() *cobra.Command {
	opt := manager.DeployOptions{
		IdentityFile: path.Join(utils.UserHome(), ".ssh", "id_rsa"),
	}
	cmd := &cobra.Command{
		Use:          "deploy   ",
		Short:        "Deploy a cluster for production",
		Long:         "Deploy a cluster for production. SSH connection will be used to deploy files, as well as creating system users for running the service.",
		SilenceUsage: true,
		RunE: func(cmd *cobra.Command, args []string) error {
		  // ...
			return cm.Deploy(clusterName, version, topoFile, opt, postDeployHook, skipConfirm, gOpt)
		},
	}
	// ...
	cmd.Flags().BoolVarP(&opt.NoLabels, "no-labels", "", false, "Don't check TiKV labels")
	return cmd
}
// manager/deploy.go
func (m *Manager) Deploy(
	name string,
	clusterVersion string,
	topoFile string,
	opt DeployOptions,
	afterDeploy func(b *task.Builder, newPart spec.Topology, gOpt operator.Options),
	skipConfirm bool,
	gOpt operator.Options,
) error {
// ...
var (
		envInitTasks      []*task.StepDisplay // tasks which are used to initialize environment
		downloadCompTasks []*task.StepDisplay // tasks which are used to download components
		deployCompTasks   []*task.StepDisplay // tasks which are used to copy components to remote host
	)
// ... 
		t := task.NewBuilder(m.logger).
			RootSSH(
				host,
				hostInfo.ssh,
				opt.User,
				sshConnProps.Password,
				// ...,
			).
			EnvInit(host, globalOptions.User, globalOptions.Group, opt.SkipCreateUser || globalOptions.User == opt.User, sudo).
			Mkdir(globalOptions.User, host, sudo, dirs...).
			BuildAsStep(fmt.Sprintf("  - Prepare %s:%d", host, hostInfo.ssh))
		envInitTasks = append(envInitTasks, t)
	// ...
	builder := task.NewBuilder(m.logger).
		Step("+ Generate SSH keys",
			task.NewBuilder(m.logger).
				SSHKeyGen(m.specManager.Path(name, "ssh", "id_rsa")).
				Build(),
			m.logger).
		ParallelStep("+ Download TiDB components", false, downloadCompTasks...).
		ParallelStep("+ Initialize target host environments", false, envInitTasks...).
		ParallelStep("+ Deploy TiDB instance", false, deployCompTasks...).
		ParallelStep("+ Copy certificate to remote host", gOpt.Force, certificateTasks...).
		ParallelStep("+ Init instance configs", gOpt.Force, refreshConfigTasks...).
		ParallelStep("+ Init monitor configs", gOpt.Force, monitorConfigTasks...)
		// ...
		m.logger.Infof("Cluster `%s` deployed successfully, you can start it with command: `%s`", name, hint)
	return nil
}