package steps

import (
	"context"
	"fmt"
	"net/url"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/host"
	"github.com/pingcap/test-infra/caselib/pkg/model/workload"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

var tiupTiDBCluster = &host.TiUPCluster{}

type initWorkload struct {
	workload *apiresource.ResourceHandle
}

func NewInitWorkload(workload *apiresource.ResourceHandle) Step {
	return &initWorkload{
		workload: workload,
	}
}

func (s *initWorkload) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.workload)
	if err != nil {
		return err
	}

	return GetInner(ctx).Board().SetWorkload(resolved.(resource.WorkloadNode))
}

func (s *initWorkload) String() string {
	return fmt.Sprintf("initWorkload: workload=%+v", s.workload)
}

type initTiup struct {
	tiup       *apiresource.ResourceHandle
	downStream *apiresource.ResourceHandle
}

func NewInitTiup(tiup, sink *apiresource.ResourceHandle) Step {
	return &initTiup{
		tiup:       tiup,
		downStream: sink,
	}
}

func (s *initTiup) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.tiup)
	if err != nil {
		return err
	}

	GetInner(ctx).Board().SetTiup(resolved.(resource.WorkloadNode))

	resolved, err = core.ResolveResourceHandle(s.downStream)
	if err != nil {
		return errors.Trace(err)
	}
	downstream := resolved.(resource.CDCSink)

	if err := GetInner(ctx).Board().SetSink(downstream); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (s *initTiup) String() string {
	return fmt.Sprintf("initTiup: workload=%+v", s.tiup)
}

type runWorkload struct {
	option workload.Option
}

func NewRunWorkload(option workload.Option) Step {
	return &runWorkload{
		option: option,
	}
}

// Execute runWorkload run the specified workload in a background goroutine.
func (s *runWorkload) Execute(ctx context.Context) (err error) {
	tc, err := GetInner(ctx).Board().TiDBCluster()
	if err != nil {
		return err
	}

	dbURL, err := tc.ServiceURL(resource.DBAddr)
	if err != nil {
		return errors.Trace(err)
	}

	command := s.option.Parse(dbURL)
	w, err := GetInner(ctx).Board().Workload()
	if err != nil {
		return errors.Trace(err)
	}

	workloadHost := host.NewWorkloadHost(w)

	start := time.Now()

	if err := workloadHost.Run(command); err != nil {
		if err != host.TimeoutErr {
			log.Error("run workload failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
			return errors.Trace(err)
		}
	}

	log.Info("run workload finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runWorkload) String() string {
	return fmt.Sprintf("runWorkload: option=%+v", s.option)
}

type runWorkloadTiup struct {
	workloadOption workload.Option
}

func NewRunWorkloadTiup(workloadOption workload.Option) Step {
	return &runWorkloadTiup{
		workloadOption: workloadOption,
	}
}

func (s *runWorkloadTiup) Execute(ctx context.Context) (err error) {

	dbURL, err := tiupTiDBCluster.ServiceURL(resource.DBAddr)
	if err != nil {
		return errors.Trace(err)
	}

	w, err := GetInner(ctx).Board().Workload()
	if err != nil {
		return errors.Trace(err)
	}

	command := s.workloadOption.Parse(dbURL)

	workloadHost := host.NewWorkloadHost(w)

	start := time.Now()

	if err := workloadHost.Run(command); err != nil {
		log.Error("run workload for tiup cluster failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run workload for tiup cluster finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runWorkloadTiup) String() string {
	return fmt.Sprintf("runWorkloadTiup: option=%+v", s.workloadOption)
}

type runTiupMirrorSet struct {
	tiupMirrorOption workload.TiupMirrorOption
}

func NewRunTiupMirrorSet(tiupMirrorOption workload.TiupMirrorOption) Step {
	return &runTiupMirrorSet{
		tiupMirrorOption: tiupMirrorOption,
	}
}

func (s *runTiupMirrorSet) Execute(ctx context.Context) (err error) {
	t, err := GetInner(ctx).Board().Tiup()
	if err != nil {
		return errors.Trace(err)
	}

	var u *url.URL
	command := s.tiupMirrorOption.Parse(u)

	workloadHost := host.NewWorkloadHost(t)

	start := time.Now()

	if err := workloadHost.Run(command); err != nil {
		log.Error("run tiup mirror setting failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run tiup mirror setting finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runTiupMirrorSet) String() string {
	return fmt.Sprintf("runTiupMirrorTiup: option=%+v", *s.tiupMirrorOption.MirrorAdress)
}

type deployTiupClusterPDDown struct {
	option workload.TiupClusterOption
}

func NewDeployTiupClusterPDDown(option workload.TiupClusterOption) Step {
	return &deployTiupClusterPDDown{
		option: option,
	}
}

func (s *deployTiupClusterPDDown) Execute(ctx context.Context) (err error) {
	w, err := GetInner(ctx).Board().Tiup()
	if err != nil {
		return errors.Trace(err)
	}
	tiupCluster := host.NewTiUPCluster(w, s.option.ControlNode, s.option.TopoFile, s.option.ClusterName)

	start := time.Now()
	// tiupCluster.Destroy()

	if err := tiupCluster.Deploy(*s.option.Version); err != nil {
		log.Error("run tiup cluster deploy failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	tiupCluster.Start()

	pdURL, err := tiupCluster.ServiceURL(resource.PDAddr)
	if err != nil {
		return errors.Trace(err)
	}

	tiupCluster.ScaleOut("pd_servers:\n  - host: tiup-peer\n    client_port: 2479\n    peer_port: 2480\n    deploy_dir: \"/home/tidb/pd-2479\"\n  - host: tiup-peer\n    client_port: 2579\n    peer_port: 2580\n    deploy_dir: \"/home/tidb/pd-2579\"\n")

	log.Info("Waiting for 10m to update member list...")
	time.Sleep(10 * time.Minute)

	tiupCluster.Stop(fmt.Sprintf("-N %s:%s", strings.Split(pdURL.Hostname(), ".")[0], pdURL.Port()))

	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))
	tiupTiDBCluster = tiupCluster

	return nil
}

func (s *deployTiupClusterPDDown) String() string {
	return fmt.Sprintf("deployTiupClusterPDDown: option=%+v, version=%+v", s.option, *s.option.Version)
}

type deployTiupCluster struct {
	option workload.TiupClusterOption
}

func NewDeployTiupCluster(option workload.TiupClusterOption) Step {
	return &deployTiupCluster{
		option: option,
	}
}

func (s *deployTiupCluster) Execute(ctx context.Context) (err error) {
	w, err := GetInner(ctx).Board().Tiup()
	if err != nil {
		return errors.Trace(err)
	}

	tiupCluster := host.NewTiUPCluster(w, s.option.ControlNode, s.option.TopoFile, s.option.ClusterName)

	start := time.Now()

	if err := tiupCluster.Deploy(*s.option.Version); err != nil {
		log.Error("run tiup cluster deploy failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	tiupCluster.Start()

	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))
	tiupTiDBCluster = tiupCluster

	return nil
}

func (s *deployTiupCluster) String() string {
	return fmt.Sprintf("deployTiupCluster: option=%+v", s.option)
}

type destroyTiupCluster struct {
	option workload.TiupClusterOption
}

func NewDestroyTiupCluster(option workload.TiupClusterOption) Step {
	return &destroyTiupCluster{
		option: option,
	}
}

func (s *destroyTiupCluster) Execute(ctx context.Context) (err error) {
	err = tiupTiDBCluster.Destroy()
	if err != nil {
		log.Error("destroy tiup cluster failed")
		return errors.Trace(err)
	}

	log.Info("destroy tiup finished")

	return nil
}

func (s *destroyTiupCluster) String() string {
	return fmt.Sprintf("destroyTiupCluster: option=%+v", s.option)
}

// tiup deploy binlog cluster
type deployBinlogTiupCluster struct {
	option   workload.TiupClusterOption
	workload *apiresource.ResourceHandle
}

func NewDeployBinlogTiupCluster(option workload.TiupClusterOption, workload *apiresource.ResourceHandle) Step {
	return &deployBinlogTiupCluster{
		option:   option,
		workload: workload,
	}
}

func (s *deployBinlogTiupCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPCluster(wl, s.option.ControlNode, s.option.TopoFile, s.option.ClusterName)

	start := time.Now()

	if err := tiupCluster.Deploy(*s.option.Version); err != nil {
		log.Error("run tiup cluster deploy failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	tiupCluster.Start()

	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))
	tiupTiDBCluster = tiupCluster

	return nil
}

func (s *deployBinlogTiupCluster) String() string {
	return fmt.Sprintf("deployBinlogTiupCluster: option=%+v", s.option)
}

type upgradeTiupCluster struct {
	version *string
}

func NewUpgradeTiupCluster(version *string) Step {
	return &upgradeTiupCluster{
		version: version,
	}
}

func (s *upgradeTiupCluster) Execute(ctx context.Context) (err error) {
	err = tiupTiDBCluster.Upgrade(*s.version)
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (s *upgradeTiupCluster) String() string {
	return fmt.Sprintf("upgradeTiupCluster: version=%+v", *s.version)
}

// Tiup scale tidb cluster
type scaleOutTiupBinlogCluster struct {
	workload *apiresource.ResourceHandle
	option   workload.TiupClusterOption
}

func NewScaleOutTiupBinlogCluster(workload *apiresource.ResourceHandle, option workload.TiupClusterOption) Step {
	return &scaleOutTiupBinlogCluster{
		workload: workload,
		option:   option,
	}
}

func (s *scaleOutTiupBinlogCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPCluster(wl, s.option.ControlNode, s.option.TopoFile, s.option.ClusterName)

	start := time.Now()

	if err := tiupCluster.ScaleOut(s.option.TopoFile); err != nil {
		log.Error("run tiup cluster scale-out failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	err = tiupCluster.Start()
	if err != nil {
		return err
	}
	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *scaleOutTiupBinlogCluster) String() string {
	return fmt.Sprintf("scaleOutTiupBinlogCluster: option=%+v", s.option)
}

// Tiup scale tidb cluster
type scaleInTiupBinlogCluster struct {
	workload *apiresource.ResourceHandle
	option   workload.TiupClusterOption
	node     string
}

func NewScaleInTiupBinlogCluster(workload *apiresource.ResourceHandle, option workload.TiupClusterOption, node string) Step {
	return &scaleInTiupBinlogCluster{
		workload: workload,
		option:   option,
		node:     node,
	}
}

func (s *scaleInTiupBinlogCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPCluster(wl, s.option.ControlNode, s.option.TopoFile, s.option.ClusterName)

	start := time.Now()

	if err := tiupCluster.ScaleIn(s.node); err != nil {
		log.Error("run tiup cluster scale-out failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	err = tiupCluster.Start()
	if err != nil {
		return err
	}
	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *scaleInTiupBinlogCluster) String() string {
	return fmt.Sprintf("scaleInTiupBinlogCluster: option=%+v", s.option)
}

type tiupCreateChangefeed struct {
	option workload.TiupChangefeedOption
}

func NewTiupCreateChangefeed(option workload.TiupChangefeedOption) Step {
	return &tiupCreateChangefeed{
		option: option,
	}
}

func (s *tiupCreateChangefeed) Execute(ctx context.Context) error {

	w, err := GetInner(ctx).Board().Tiup()
	if err != nil {
		return errors.Trace(err)
	}

	workloadHost := host.NewWorkloadHost(w)

	start := time.Now()

	// To get upstream pd address for changefeed creation, pdURL, err := tiupTiDBCluster.ServiceURL(resource.PDAddr)

	sink, err := GetInner(ctx).Board().Sink()
	if err != nil {
		return errors.Trace(err)
	}

	dbURL, err := sink.SinkEndpoint()
	if err != nil {
		return errors.Trace(err)
	}

	command := s.option.Parse(dbURL)

	if err := workloadHost.Run(command); err != nil {
		log.Error("run tiup changefeed task failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}
	log.Info("run tiup changeFeed task finished", zap.Error(err))
	return err
}

func (s *tiupCreateChangefeed) String() string {
	return fmt.Sprintf("tiup create changefeed: option=%+v", s.option)
}

type waitWorkloadFinish struct {
}

func NewWaitWorkloadFinish() Step {
	return &waitWorkloadFinish{}
}

func (s *waitWorkloadFinish) Execute(ctx context.Context) (err error) {
	start := time.Now()
	defer func() {
		log.Info("waitWorkloadFinish", zap.Duration("elapsed", time.Since(start)), zap.Error(err))
	}()

	<-GetInner(ctx).Board().workloadDone
	return nil
}

func (s *waitWorkloadFinish) String() string {
	return "waitWorkloadFinish"
}

type signalWorkloadFinish struct {
}

func NewSignalWorkloadFinish() Step {
	return &signalWorkloadFinish{}
}

func (s *signalWorkloadFinish) Execute(ctx context.Context) error {
	GetInner(ctx).Board().workloadDone <- struct{}{}
	return nil
}

func (s *signalWorkloadFinish) String() string {
	return "signalWorkloadFinish"
}

type runDMWorkload struct {
	upStream *apiresource.ResourceHandle
	dm       *apiresource.ResourceHandle
	workload *apiresource.ResourceHandle
	option   workload.Option
}

func NewRunDMWorkload(upStream, workload *apiresource.ResourceHandle, option workload.Option) Step {
	return &runDMWorkload{
		upStream: upStream,
		workload: workload,
		option:   option,
	}
}

// Execute runDMWorkload run the specified workload in a background goroutine.
func (r *runDMWorkload) Execute(ctx context.Context) error {
	up := utils.Try(core.ResolveResourceHandle(r.upStream)).(resource.MySQL)
	dbURL := utils.Try(up.ServiceURL(resource.DBAddr)).(*url.URL)
	command := utils.Try(r.option.Parse(dbURL)).(*workload.CMD)

	wl := utils.Try(core.ResolveResourceHandle(r.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)
	start := time.Now()
	if err := workloadHost.Run(command); err != nil {
		log.Error("run workload failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run workload finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (r *runDMWorkload) String() string {
	return fmt.Sprintf("runWorkload: option=%+v", r.option)
}

func (b *Builder) DmRunWorkload(upStream, dm, workload *apiresource.ResourceHandle, option workload.Option) *Builder {
	b.steps = append(b.steps, &runDMWorkload{
		upStream: upStream,
		dm:       dm,
		workload: workload,
		option:   option,
	})
	return b
}

type runBinlogWorkload struct {
	tidb     *apiresource.ResourceHandle
	workload *apiresource.ResourceHandle
	option   workload.Option
}

func NewRunBinlogWorkload(tidbcluster, workload *apiresource.ResourceHandle, option workload.Option) Step {
	return &runBinlogWorkload{
		tidb:     tidbcluster,
		workload: workload,
		option:   option,
	}
}

// Execute runWorkload run the specified workload in a background goroutine.
func (r *runBinlogWorkload) Execute(ctx context.Context) error {
	up := utils.Try(core.ResolveResourceHandle(r.tidb)).(resource.TiDBCluster)
	dbURL := utils.Try(up.ServiceURL(resource.DBAddr)).(*url.URL)
	command := utils.Try(r.option.Parse(dbURL)).(*workload.CMD)

	wl := utils.Try(core.ResolveResourceHandle(r.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)
	start := time.Now()
	if err := workloadHost.Run(command); err != nil {
		log.Error("run workload failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run workload finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runBinlogWorkload) String() string {
	return fmt.Sprintf("binlogRunWorkload: option=%+v", s.option)
}

// Used for binlog workerload
func (b *Builder) RunBinlogWorkload(handle, workload *apiresource.ResourceHandle, option workload.Option) *Builder {
	b.steps = append(b.steps, NewRunBinlogWorkload(handle, workload, option))
	return b
}

// Deploy dm cluster
type deployTiupDmCluster struct {
	// dm       *apiresource.ResourceHandle
	workload *apiresource.ResourceHandle
	option   workload.TiupDmOption
}

func NewDeployTiupDmCluster(workload *apiresource.ResourceHandle, option workload.TiupDmOption) Step {
	return &deployTiupDmCluster{
		workload: workload,
		option:   option,
	}
}

func (s *deployTiupDmCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPDmCluster(wl, s.option.TopoFile, s.option.DmName)

	start := time.Now()

	if err := tiupCluster.Deploy(*s.option.Version); err != nil {
		log.Error("run tiup dm deploy failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	err = tiupCluster.Start()
	if err != nil {
		return err
	}
	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *deployTiupDmCluster) String() string {
	return fmt.Sprintf("deployTiupDmCluster: option=%+v", s.option)
}

type runTiupDmMirrorSet struct {
	workload           *apiresource.ResourceHandle
	tiupDmMirrorOption workload.TiupDmMirrorOption
}

func NewRunTiupDmMirrorSet(tiupDmMirrorOption workload.TiupDmMirrorOption, workload *apiresource.ResourceHandle) Step {
	return &runTiupDmMirrorSet{
		tiupDmMirrorOption: tiupDmMirrorOption,
		workload:           workload,
	}
}

func (s *runTiupDmMirrorSet) Execute(ctx context.Context) (err error) {
	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	var u *url.URL
	command := s.tiupDmMirrorOption.Parse(u)
	workloadHost := host.NewWorkloadHost(wl)

	start := time.Now()

	if err := workloadHost.Run(command); err != nil {
		log.Error("run tiup mirror setting failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run tiup mirror setting finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runTiupDmMirrorSet) String() string {
	return fmt.Sprintf("runTiupDmMirrorTiup: option=%+v", *s.tiupDmMirrorOption.MirrorAdress)
}

type stopWorkload struct {
	KeyWord  *string
	workload *apiresource.ResourceHandle
}

func (s stopWorkload) String() string {
	return fmt.Sprintf("Kill running workloads with keyword %s", *s.KeyWord)
}

func (s stopWorkload) Execute(ctx context.Context) error {
	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)
	command := &workload.CMD{Command: fmt.Sprintf("pkill %s", *s.KeyWord), Timeout: 10 * time.Second}
	if err := workloadHost.Run(command); err != nil {
		log.Error("Kill workload failed", zap.Error(err), zap.Any("command", command))
		return errors.Trace(err)
	}
	return nil
}

func NewStopWorkload(keyword *string, workload *apiresource.ResourceHandle) Step {
	return &stopWorkload{keyword, workload}
}

type checkWorkloadRunning struct {
	KeyWord  *string
	workload *apiresource.ResourceHandle
}

func (c checkWorkloadRunning) String() string {
	return fmt.Sprintf("Check workload running with keyword: %s", *c.KeyWord)
}

func (c checkWorkloadRunning) Execute(ctx context.Context) error {
	wl := utils.Try(core.ResolveResourceHandle(c.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)
	command := &workload.CMD{Command: fmt.Sprintf("pgrep %s", *c.KeyWord), Timeout: 10 * time.Second}
	if err := workloadHost.Run(command); err != nil {
		log.Error("Check workload running failed", zap.Error(err), zap.Any("command", command))
		return errors.Trace(err)
	}
	return nil
}

func (b *Builder) CheckWorkloadRunning(keyword *string, workload *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, NewCheckWorkloadRunning(keyword, workload))
	return b
}

func NewCheckWorkloadRunning(keyword *string, workload *apiresource.ResourceHandle) Step {
	return &checkWorkloadRunning{keyword, workload}
}

type runWorkloadAsync struct {
	tidb                *apiresource.ResourceHandle
	workload            *apiresource.ResourceHandle
	option              workload.Option
	async               bool
	killRunningWorkload bool // Check whether there is running workload, and kill it and restart workload if killRunningWorkload = true
}

func NewRunWorkloadAsync(tidbcluster, workload *apiresource.ResourceHandle, option workload.Option, async, killRunningWorkload bool) Step {
	return &runWorkloadAsync{
		tidb:                tidbcluster,
		workload:            workload,
		option:              option,
		async:               async,
		killRunningWorkload: killRunningWorkload,
	}
}

// Execute runWorkload run the specified workload in a background goroutine.
func (r *runWorkloadAsync) Execute(ctx context.Context) error {
	up := utils.Try(core.ResolveResourceHandle(r.tidb)).(resource.TiDBCluster)
	dbURL := utils.Try(up.ServiceURL(resource.DBAddr)).(*url.URL)
	command := utils.Try(r.option.Parse(dbURL)).(*workload.CMD)

	wl := utils.Try(core.ResolveResourceHandle(r.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)

	// check workload are running by pgrep xx (e.g. pgrep go-ycsb)
	cmd := strings.Split(command.Command, " ")[0]
	keyword := strings.Split(cmd, "/")[len(strings.Split(cmd, "/"))-1]
	pgrepCmd := &workload.CMD{Command: fmt.Sprintf("pgrep %s", keyword), Timeout: 10 * time.Second}
	// check whether need to kill the workload and restart it.
	if err := workloadHost.Run(pgrepCmd); err == nil {
		if r.killRunningWorkload == false {
			log.Info("Workload already running, and killRunningWorkload=false, nothing to do.")
			return nil
		} else {
			log.Info("Workload already running, and killRunningWorkload=true, will kill the workload and restart it.")
			killCmd := &workload.CMD{Command: fmt.Sprintf("pkill %s", keyword), Timeout: 10 * time.Second}
			if err := workloadHost.Run(killCmd); err != nil {
				log.Error("Kill workload failed", zap.Error(err), zap.Any("command", killCmd))
				return errors.Trace(err)
			}
		}
	} else {
		log.Info("Workload not running, will start it now.")
	}

	if r.async == true {
		err := workloadHost.ExecAsync(command, 10*time.Second)
		if err != nil {
			return errors.Trace(err)
		} else {
			// sleep 10s  and check workload process command are running
			time.Sleep(10 * time.Second)
			// check workload are running by pgrep xx (e.g. pgrep go-ycsb)
			cmd := strings.Split(command.Command, " ")[0]
			keyword := strings.Split(cmd, "/")[len(strings.Split(cmd, "/"))-1]
			command := &workload.CMD{Command: fmt.Sprintf("pgrep %s", keyword), Timeout: 10 * time.Second}
			if err := workloadHost.Run(command); err != nil {
				log.Error("Failed to pgrep the process, workload may not running.", zap.Error(err), zap.Any("command", command))
				return errors.Trace(err)
			}
			return nil
		}
	} else {
		start := time.Now()
		if err := workloadHost.Run(command); err != nil {
			log.Error("run workload failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
			return errors.Trace(err)
		}
		log.Info("run workload finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return nil
	}
	return nil
}

func (r *runWorkloadAsync) String() string {
	return fmt.Sprintf("runWorkloadAsync: option=%+v", r.option)
}

// RunWorkloadAsync Run workload async, and return immediately
func (b *Builder) RunWorkloadAsync(handle, workload *apiresource.ResourceHandle, option workload.Option, async, killRunningWorkload bool) *Builder {
	b.steps = append(b.steps, NewRunWorkloadAsync(handle, workload, option, async, killRunningWorkload))
	return b
}

// Scale dm cluster
type scaleOutTiupDmCluster struct {
	// dm       *apiresource.ResourceHandle
	workload *apiresource.ResourceHandle
	option   workload.TiupDmOption
}

func NewScaleOutTiupDmCluster(workload *apiresource.ResourceHandle, option workload.TiupDmOption) Step {
	return &scaleOutTiupDmCluster{
		workload: workload,
		option:   option,
	}
}

func (s *scaleOutTiupDmCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPDmCluster(wl, s.option.TopoFile, s.option.DmName)

	start := time.Now()

	if err := tiupCluster.ScaleOut(); err != nil {
		log.Error("run tiup dm scale-out failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	err = tiupCluster.Start()
	if err != nil {
		return err
	}
	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *scaleOutTiupDmCluster) String() string {
	return fmt.Sprintf("scaleOutTiupDmCluster: option=%+v", s.option)
}

// Scale in dm cluster
type scaleInTiupDmCluster struct {
	workload *apiresource.ResourceHandle
	option   workload.TiupDmOption
}

func NewScaleInTiupDmCluster(workload *apiresource.ResourceHandle, option workload.TiupDmOption) Step {
	return &scaleInTiupDmCluster{
		workload: workload,
		option:   option,
	}
}

func (s *scaleInTiupDmCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPDmCluster(wl, s.option.TopoFile, s.option.DmName)

	start := time.Now()

	if err := tiupCluster.ScaleIn(); err != nil {
		log.Error("run tiup dm scale-in failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	err = tiupCluster.Start()
	if err != nil {
		return err
	}
	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *scaleInTiupDmCluster) String() string {
	return fmt.Sprintf("scaleInTiupDmCluster: option=%+v", s.option)
}

// Upgrade out dm cluster
type upgradeTiupDmCluster struct {
	workload *apiresource.ResourceHandle
	option   workload.TiupDmOption
}

func NEWUpgradeTiupDmCluster(workload *apiresource.ResourceHandle, option workload.TiupDmOption) Step {
	return &upgradeTiupDmCluster{
		workload: workload,
		option:   option,
	}
}

func (s *upgradeTiupDmCluster) Execute(ctx context.Context) (err error) {

	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	tiupCluster := host.NewTiUPDmCluster(wl, s.option.TopoFile, s.option.DmName)

	start := time.Now()

	if err := tiupCluster.Upgrade(*s.option.Version); err != nil {
		log.Error("run tiup dm upgrade failed", zap.Error(err), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run tiup finished", zap.Duration("elapsed", time.Since(start)))

	return nil
}

func (s *upgradeTiupDmCluster) String() string {
	return fmt.Sprintf("upgradeTiupDmCluster: option=%+v", s.option)
}

// run binlog tiup workload
type runBinlogWorkloadTiup struct {
	workloadOption workload.Option
	workload       *apiresource.ResourceHandle
}

func NewRunBinlogWorkloadTiup(workloadOption workload.Option, workload *apiresource.ResourceHandle) Step {
	return &runBinlogWorkloadTiup{
		workloadOption: workloadOption,
		workload:       workload,
	}
}

func (s *runBinlogWorkloadTiup) Execute(ctx context.Context) (err error) {

	dbURL, err := tiupTiDBCluster.ServiceURL(resource.DBAddr)
	if err != nil {
		return errors.Trace(err)
	}
	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)
	start := time.Now()

	command := s.workloadOption.Parse(dbURL)

	if err := workloadHost.Run(command); err != nil {
		log.Error("run workload for tiup cluster failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run workload for tiup cluster finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runBinlogWorkloadTiup) String() string {
	return fmt.Sprintf("runBinlogWorkloadTiup: option=%+v", s.workloadOption)
}

type runBinlogTiupMirrorSet struct {
	tiupMirrorOption workload.TiupMirrorOption
	workload         *apiresource.ResourceHandle
}

func NewRunBinlogTiupMirrorSet(tiupMirrorOption workload.TiupMirrorOption, workload *apiresource.ResourceHandle) Step {
	return &runBinlogTiupMirrorSet{
		tiupMirrorOption: tiupMirrorOption,
		workload:         workload,
	}
}

func (s *runBinlogTiupMirrorSet) Execute(ctx context.Context) (err error) {
	var u *url.URL
	command := s.tiupMirrorOption.Parse(u)
	wl := utils.Try(core.ResolveResourceHandle(s.workload)).(resource.WorkloadNode)
	workloadHost := host.NewWorkloadHost(wl)

	// workloadHost := host.NewWorkloadHost(t)

	start := time.Now()

	if err := workloadHost.Run(command); err != nil {
		log.Error("run tiup mirror setting failed", zap.Error(err), zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
		return errors.Trace(err)
	}

	log.Info("run tiup mirror setting finished", zap.Any("command", command), zap.Duration("elapsed", time.Since(start)))
	return nil
}

func (s *runBinlogTiupMirrorSet) String() string {
	return fmt.Sprintf("runBinlogTiupMirrorSet: option=%+v", *s.tiupMirrorOption.MirrorAdress)
}

// Scale cdc cluster
type scaleOutTiCDCClusterTiup struct {
	option workload.TiupClusterOption
}

func NewScaleOutTiCDCClusterTiup(option workload.TiupClusterOption) Step {
	return &scaleOutTiCDCClusterTiup{
		option:   option,
	}
}

func (s *scaleOutTiCDCClusterTiup) Execute(ctx context.Context) (err error) {
	if err := tiupTiDBCluster.ScaleOut("cdc_servers:\n  - host: tiup-peer\n    port: 8501\n  - host: tiup-peer\n    port: 8502\n");
		err != nil {
		log.Error("tiup cluster cdc scale out failed", zap.Error(err))
		return errors.Trace(err)
	}
	time.Sleep(1 * time.Minute)
	if err := tiupTiDBCluster.ScaleIn("tiup-peer:8502");
		err != nil {
		log.Error("tiup cluster cdc scale in failed", zap.Error(err))
		return errors.Trace(err)
	}

	log.Info("tiup cluster cdc scale success", zap.Any("option", s))
	return nil
}

func (s *scaleOutTiCDCClusterTiup) String() string {
	return fmt.Sprintf("scaleOutTiCDCClusterTiup: option=%+v", s.option)
}

// Restart cdc cluster
type restartTiCDCTiup struct {
	option workload.TiupClusterOption
}

func NewRestartTiCDCTiup(option workload.TiupClusterOption) Step {
	return &restartTiCDCTiup{
		option:   option,
	}
}

func (s *restartTiCDCTiup) Execute(ctx context.Context) (err error) {
	if err := tiupTiDBCluster.Restart("-R cdc");
		err != nil {
		log.Error("tiup cluster cdc restart failed", zap.Error(err))
		return errors.Trace(err)
	}
	log.Info("tiup cluster cdc restart success", zap.Any("option", s))

	return nil
}

func (s *restartTiCDCTiup) String() string {
	return fmt.Sprintf("restartTiCDCTiup: option=%+v", s.option)
}


type recordTime struct {
	time *time.Time
}

func NewRecordTime(time *time.Time) Step {
	return &recordTime{
		time: time,
	}
}

func (s *recordTime) Execute(ctx context.Context) (err error) {
	*s.time = time.Now()
	return nil
}

func (s *recordTime) String() string {
	return fmt.Sprintf("recordTime: time=%+v", *s.time)
}