mini refactor playground to run component directly (#588)

Signed-off-by: lucklove <[email protected]>

cmd/root.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/fatih/color"
 	"github.com/pingcap/tiup/pkg/environment"
+	"github.com/pingcap/tiup/pkg/exec"
 	"github.com/pingcap/tiup/pkg/localdata"
 	"github.com/pingcap/tiup/pkg/repository"
 	"github.com/pingcap/tiup/pkg/version"
@@ -98,7 +99,7 @@ the latest stable version will be downloaded from the repository.`,
 						break
 					}
 				}
-				return runComponent(env, tag, componentSpec, binPath, transparentParams)
+				return exec.RunComponent(env, tag, componentSpec, binPath, transparentParams)
 			}
 			return cmd.Help()
 		},

cmd/telemetry.go

@@ -2,29 +2,12 @@ package cmd
 
 import (
 	"fmt"
-	"os"
-	"path/filepath"
 
 	"github.com/pingcap/tiup/pkg/environment"
-	"github.com/pingcap/tiup/pkg/localdata"
 	"github.com/pingcap/tiup/pkg/telemetry"
 	"github.com/spf13/cobra"
 )
 
-const telemetryFname = "meta.yaml"
-
-func getTelemetryMeta(env *environment.Environment) (meta *telemetry.Meta, fname string, err error) {
-	dir := env.Profile().Path(localdata.TelemetryDir)
-	err = os.MkdirAll(dir, 0755)
-	if err != nil {
-		return
-	}
-
-	fname = filepath.Join(dir, telemetryFname)
-	meta, err = telemetry.LoadFrom(fname)
-	return
-}
-
 func newTelemetryCmd() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "telemetry",
@@ -36,7 +19,7 @@ func newTelemetryCmd() *cobra.Command {
 		Short: "Reset the uuid used for telemetry",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			env := environment.GlobalEnv()
-			teleMeta, fname, err := getTelemetryMeta(env)
+			teleMeta, fname, err := telemetry.GetMeta(env)
 			if err != nil {
 				return err
 			}
@@ -57,7 +40,7 @@ func newTelemetryCmd() *cobra.Command {
 		Short: "Enable telemetry of tiup",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			env := environment.GlobalEnv()
-			teleMeta, fname, err := getTelemetryMeta(env)
+			teleMeta, fname, err := telemetry.GetMeta(env)
 			if err != nil {
 				return err
 			}
@@ -78,7 +61,7 @@ func newTelemetryCmd() *cobra.Command {
 		Short: "Disable telemetry of tiup",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			env := environment.GlobalEnv()
-			teleMeta, fname, err := getTelemetryMeta(env)
+			teleMeta, fname, err := telemetry.GetMeta(env)
 			if err != nil {
 				return err
 			}
@@ -99,7 +82,7 @@ func newTelemetryCmd() *cobra.Command {
 		Short: "Display the current status of tiup telemetry",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			env := environment.GlobalEnv()
-			teleMeta, _, err := getTelemetryMeta(env)
+			teleMeta, _, err := telemetry.GetMeta(env)
 			if err != nil {
 				return err
 			}

components/playground/grafana.go

@@ -19,13 +19,14 @@ import (
 	"io/ioutil"
 	"os"
 	"os/exec"
+	"path"
 	"path/filepath"
 	"regexp"
 	"strings"
 
 	"github.com/pingcap/errors"
-	"github.com/pingcap/tiup/components/playground/instance"
-	"github.com/pingcap/tiup/pkg/localdata"
+	"github.com/pingcap/tiup/pkg/environment"
+	tiupexec "github.com/pingcap/tiup/pkg/exec"
 	"github.com/pingcap/tiup/pkg/repository/v0manifest"
 	"github.com/pingcap/tiup/pkg/utils"
 )
@@ -188,24 +189,22 @@ http_port = %d
 	}
 
 	args := []string{
-		"tiup",
-		instance.CompVersion("grafana", v0manifest.Version(g.version)),
 		"--homepath",
 		dir,
 		"--config",
 		customeFName,
+		fmt.Sprintf("cfg:default.paths.logs=%s", path.Join(dir, "log")),
 	}
 
-	cmd := exec.CommandContext(ctx, args[0], args[1:]...)
-	cmd.Env = append(
-		os.Environ(),
-		fmt.Sprintf("%s=%s", localdata.EnvNameInstanceDataDir, dir),
-	)
+	env := environment.GlobalEnv()
+	cmd, err := tiupexec.PrepareCommand(ctx, "grafana", v0manifest.Version(g.version), "", "", dir, args, env)
+	if err != nil {
+		return errors.AddStack(err)
+	}
+	cmd.Stdout = nil
+	cmd.Stderr = nil
 
 	g.cmd = cmd
-	g.cmd.Stderr = os.Stderr
-	g.cmd.Stderr = os.Stdout
-	fmt.Println("start grafana: ", strings.Join(args, " "))
 
 	return g.cmd.Start()
 }

components/playground/instance/drainer.go

@@ -28,7 +28,7 @@ import (
 type Drainer struct {
 	instance
 	pds []*PDInstance
-	*Process
+	Process
 }
 
 var _ Instance = &Drainer{}
@@ -82,8 +82,6 @@ func (d *Drainer) Start(ctx context.Context, version v0manifest.Version) error {
 	}
 
 	args := []string{
-		"tiup", fmt.Sprintf("--binpath=%s", d.BinPath),
-		CompVersion("drainer", version),
 		fmt.Sprintf("--node-id=%s", d.NodeID()),
 		fmt.Sprintf("--addr=%s:%d", d.Host, d.Port),
 		fmt.Sprintf("--advertise-addr=%s:%d", advertiseHost(d.Host), d.Port),
@@ -94,8 +92,11 @@ func (d *Drainer) Start(ctx context.Context, version v0manifest.Version) error {
 		args = append(args, fmt.Sprintf("--config=%s", d.ConfigPath))
 	}
 
-	d.Process = NewProcess(ctx, d.Dir, args[0], args[1:]...)
-	logIfErr(d.Process.setOutputFile(d.LogFile()))
+	var err error
+	if d.Process, err = NewComponentProcess(ctx, d.Dir, d.BinPath, "drainer", version, args...); err != nil {
+		return err
+	}
+	logIfErr(d.Process.SetOutputFile(d.LogFile()))
 
 	return d.Process.Start()
 }

components/playground/instance/instance.go

@@ -52,7 +52,7 @@ type Instance interface {
 	// StatusAddrs return the address to pull metrics.
 	StatusAddrs() []string
 	// Wait Should only call this if the instance is started successfully.
-	Wait() error
+	Wait(ctx context.Context) error
 }
 
 func (inst *instance) StatusAddrs() (addrs []string) {

components/playground/instance/pd.go

@@ -30,7 +30,7 @@ type PDInstance struct {
 	instance
 	initEndpoints []*PDInstance
 	joinEndpoints []*PDInstance
-	*Process
+	Process
 }
 
 // NewPDInstance return a PDInstance
@@ -72,8 +72,6 @@ func (inst *PDInstance) Start(ctx context.Context, version v0manifest.Version) e
 	}
 	uid := inst.Name()
 	args := []string{
-		"tiup", fmt.Sprintf("--binpath=%s", inst.BinPath),
-		CompVersion("pd", version),
 		"--name=" + uid,
 		fmt.Sprintf("--data-dir=%s", filepath.Join(inst.Dir, "data")),
 		fmt.Sprintf("--peer-urls=http://%s:%d", inst.Host, inst.Port),
@@ -103,8 +101,11 @@ func (inst *PDInstance) Start(ctx context.Context, version v0manifest.Version) e
 		return errors.Errorf("must set the init or join instances.")
 	}
 
-	inst.Process = NewProcess(ctx, inst.Dir, args[0], args[1:]...)
-	logIfErr(inst.Process.setOutputFile(inst.LogFile()))
+	var err error
+	if inst.Process, err = NewComponentProcess(ctx, inst.Dir, inst.BinPath, "pd", version, args...); err != nil {
+		return err
+	}
+	logIfErr(inst.Process.SetOutputFile(inst.LogFile()))
 
 	return inst.Process.Start()
 }

components/playground/instance/process.go

@@ -2,41 +2,60 @@ package instance
 
 import (
 	"context"
-	"fmt"
 	"io"
 	"os"
 	"os/exec"
 	"time"
 
 	"github.com/pingcap/errors"
-	"github.com/pingcap/tiup/pkg/localdata"
+	"github.com/pingcap/tiup/pkg/environment"
+	tiupexec "github.com/pingcap/tiup/pkg/exec"
+	"github.com/pingcap/tiup/pkg/repository/v0manifest"
 )
 
-// Process represent process to be run by playground.
-type Process struct {
+// ErrorWaitTimeout is used to represent timeout of a command
+// Example:
+//		_ = syscall.Kill(cmd.Process.Pid, syscall.SIGKILL)
+//		if err := WaitContext(context.WithTimeout(context.Background(), 3), cmd); err == ErrorWaitTimeout {
+//			// Do something
+//		}
+var ErrorWaitTimeout = errors.New("wait command timeout")
+
+// Process represent process to be run by playground
+type Process interface {
+	Start() error
+	Wait(ctx context.Context) error
+	Pid() int
+	Uptime() string
+	SetOutputFile(fname string) error
+	Cmd() *exec.Cmd
+}
+
+// process implementes Process
+type process struct {
 	cmd       *exec.Cmd
 	startTime time.Time
 }
 
 // Start the process
-func (p *Process) Start() error {
+func (p *process) Start() error {
 	// fmt.Printf("Starting `%s`: %s", filepath.Base(p.cmd.Path), strings.Join(p.cmd.Args, " "))
 	p.startTime = time.Now()
 	return p.cmd.Start()
 }
 
 // Wait implements Instance interface.
-func (p *Process) Wait() error {
-	return p.cmd.Wait()
+func (p *process) Wait(ctx context.Context) error {
+	return WaitContext(ctx, p.cmd)
 }
 
 // Pid implements Instance interface.
-func (p *Process) Pid() int {
+func (p *process) Pid() int {
 	return p.cmd.Process.Pid
 }
 
 // Uptime implements Instance interface.
-func (p *Process) Uptime() string {
+func (p *process) Uptime() string {
 	s := p.cmd.ProcessState
 	if s != nil && s.Exited() {
 		return "exited"
@@ -46,7 +65,7 @@ func (p *Process) Uptime() string {
 	return duration.String()
 }
 
-func (p *Process) setOutputFile(fname string) error {
+func (p *process) SetOutputFile(fname string) error {
 	f, err := os.OpenFile(fname, os.O_RDWR|os.O_CREATE, 0666)
 	if err != nil {
 		return errors.AddStack(err)
@@ -55,27 +74,46 @@ func (p *Process) setOutputFile(fname string) error {
 	return nil
 }
 
-func (p *Process) setOutput(w io.Writer) {
+func (p *process) setOutput(w io.Writer) {
 	p.cmd.Stdout = w
 	p.cmd.Stderr = w
 }
 
-// NewProcess create a Process instance.
-func NewProcess(ctx context.Context, dir string, name string, arg ...string) *Process {
+func (p *process) Cmd() *exec.Cmd {
+	return p.cmd
+}
+
+// NewComponentProcess create a Process instance.
+func NewComponentProcess(ctx context.Context, dir, binPath, component string, version v0manifest.Version, arg ...string) (Process, error) {
 	if dir == "" {
 		panic("dir must be set")
 	}
 
-	cmd := exec.CommandContext(ctx, name, arg...)
-	cmd.Env = append(
-		os.Environ(),
-		fmt.Sprintf("%s=%s", localdata.EnvNameInstanceDataDir, dir),
-	)
+	env := environment.GlobalEnv()
+	cmd, err := tiupexec.PrepareCommand(ctx, component, version, binPath, "", dir, arg, env)
+	if err != nil {
+		return nil, err
+	}
 
-	cmd.Stdout = os.Stdout
-	cmd.Stderr = os.Stderr
+	return &process{cmd: cmd}, nil
+}
 
-	return &Process{
-		cmd: cmd,
+// WaitContext wrap cmd.Wait with context
+func WaitContext(ctx context.Context, cmd *exec.Cmd) error {
+	// We use cmd.Process.Wait instead of cmd.Wait because cmd.Wait is not reenterable
+	c := make(chan error, 1)
+	go func() {
+		if cmd == nil || cmd.Process == nil {
+			c <- nil
+		} else {
+			_, err := cmd.Process.Wait()
+			c <- err
+		}
+	}()
+	select {
+	case <-ctx.Done():
+		return ErrorWaitTimeout
+	case err := <-c:
+		return err
 	}
 }

components/playground/instance/pump.go

@@ -30,7 +30,7 @@ import (
 type Pump struct {
 	instance
 	pds []*PDInstance
-	*Process
+	Process
 }
 
 var _ Instance = &Pump{}
@@ -93,8 +93,6 @@ func (p *Pump) Start(ctx context.Context, version v0manifest.Version) error {
 	}
 
 	args := []string{
-		"tiup", fmt.Sprintf("--binpath=%s", p.BinPath),
-		CompVersion("pump", version),
 		fmt.Sprintf("--node-id=%s", p.NodeID()),
 		fmt.Sprintf("--addr=%s:%d", p.Host, p.Port),
 		fmt.Sprintf("--advertise-addr=%s:%d", advertiseHost(p.Host), p.Port),
@@ -105,8 +103,11 @@ func (p *Pump) Start(ctx context.Context, version v0manifest.Version) error {
 		args = append(args, fmt.Sprintf("--config=%s", p.ConfigPath))
 	}
 
-	p.Process = NewProcess(ctx, p.Dir, args[0], args[1:]...)
-	logIfErr(p.Process.setOutputFile(p.LogFile()))
+	var err error
+	if p.Process, err = NewComponentProcess(ctx, p.Dir, p.BinPath, "pump", version, args...); err != nil {
+		return err
+	}
+	logIfErr(p.Process.SetOutputFile(p.LogFile()))
 
 	return p.Process.Start()
 }