package host

import (
	"bytes"
	"fmt"
	"net/url"
	"strconv"
	"strings"
	"time"

	"github.com/BurntSushi/toml"
	"github.com/gogo/protobuf/jsonpb"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	dmconfig "github.com/pingcap/tiflow/dm/dm/config"
	"github.com/pingcap/tiflow/dm/dm/pb"
	"go.uber.org/zap"
	"gopkg.in/yaml.v2"

	"github.com/pingcap/test-infra/caselib/pkg/command"
	dmmodel "github.com/pingcap/test-infra/caselib/pkg/model/dm"
	"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"
)

const (
	SOURCE           = "source-"
	TIMEOUT          = 30 * time.Second
	SYNCDIFF_TIMEOUT = 10 * time.Minute
)

type DMHost struct {
	name        string // resource name of DM cluster
	testBedName string
	dm          *resource.DMCluster
	upStreams   []*resource.MySQL
	downStream  *resource.TiDBCluster
	syncDiff    *resource.WorkloadNode
	// map: source-id -> host:port
	upUrls  map[string]*url.URL
	downUrl *url.URL
}

func (d *DMHost) Init() {
	d.upUrls = make(map[string]*url.URL)
	for i, up := range d.upStreams {
		uri := utils.Try((*up).ServiceURL(resource.DBAddr)).(*url.URL)
		// sourceID: source-[i]
		d.upUrls[SOURCE+strconv.Itoa(i)] = uri
	}
	d.downUrl = utils.Try((*d.downStream).ServiceURL(resource.DBAddr)).(*url.URL)
}

func (d *DMHost) GetMasterHost() string {
	return fmt.Sprintf("http://dm-dm-master.%s:8261", d.testBedName)
}

func (d *DMHost) GetSourceHost(sourceName string) string {
	return d.upUrls[sourceName].Hostname()
}

func (d *DMHost) GetSourcePort(sourceName string) int {
	return utils.Try(strconv.Atoi(d.upUrls[sourceName].Port())).(int)
}

func (d *DMHost) GetTargetHost() string {
	return d.downUrl.Hostname()
}

func (d *DMHost) GetTargetPort() int {
	return utils.Try(strconv.Atoi(d.downUrl.Port())).(int)
}

func (d *DMHost) Commands() *command.DmCmd {
	return &command.DmCmd{ExecCmd: command.ExecCmd{Command: "", Options: nil}}
}

func (d *DMHost) KillWorker(name string) (*command.ExecOutput, error) {
	stdout, stderr, exitCode, err := d.execOnWorker(name, "kill", "1")
	return &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}, nil
}

func (d *DMHost) DeleteWorkerRelayDir(name string) error {
	_, _, _, err := d.execOnWorker(name, "rm", "-rf", "/relay-dir")
	if err != nil {
		return err
	}
	return nil
}

func (d *DMHost) KillMaster(name string) (*command.ExecOutput, error) {
	stdout, stderr, exitCode, err := d.execOnMaster(name, "kill", "1")
	return &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}, nil
}

func (d *DMHost) ExecDmCmd(cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	done := make(chan *command.ExecOutput, 1)
	log.Info("Execute command on dm", zap.String("command", cmd), zap.Duration("timeout", timeout))
	go func() {
		podName := fmt.Sprintf("%s-dm-master-0", d.name)
		stdout, stderr, exitCode, err := d.execOnMaster(podName, "/bin/sh", "-c", cmd)
		done <- &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}
	}()

	select {
	case <-time.After(timeout):
		return nil, TimeoutErr
	case output := <-done:
		return output, errors.Trace(output.Err)
	}
}

func (d *DMHost) ExecDmctlCmd(cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	fullCmd := strings.Join([]string{command.Dmctl, cmd}, " ")
	return d.ExecDmCmd(fullCmd, timeout)
}

func (d *DMHost) ExecTiupDmctlCmd(cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	fullCmd := strings.Join([]string{command.TiupDmctl, cmd}, " ")
	return d.ExecDmCmd(fullCmd, timeout)
}

func (d *DMHost) execOnMaster(podName string, cmd ...string) (string, string, int, error) {
	return (*d.dm).Exec(podName, resource.DMMasterContainerName, cmd...)
}

func (d *DMHost) execOnWorker(podName string, cmd ...string) (string, string, int, error) {
	return (*d.dm).Exec(podName, resource.DMWorkerContainerName, cmd...)
}

func (d *DMHost) ExecUpCmd(cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	done := make(chan *command.ExecOutput, 1)
	log.Info("Execute command on up stream", zap.String("command", cmd), zap.Duration("timeout", timeout))
	go func() {
		stdout, stderr, exitCode, err := (*d.upStreams[0]).Exec(resource.WorkloadNodeExecOptions{Command: cmd})
		done <- &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}
	}()

	select {
	case <-time.After(timeout):
		return nil, TimeoutErr
	case output := <-done:
		return output, errors.Trace(output.Err)
	}
}

func (d *DMHost) ExecSyncDiffCmd(cmd string, timeout time.Duration) (*command.ExecOutput, error) {
	done := make(chan *command.ExecOutput, 1)
	log.Info("Execute command on up sync diff", zap.String("command", cmd), zap.Duration("timeout", timeout))
	go func() {
		stdout, stderr, exitCode, err := (*d.syncDiff).Exec(resource.WorkloadNodeExecOptions{Command: cmd})
		done <- &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}
	}()

	select {
	case <-time.After(timeout):
		return nil, TimeoutErr
	case output := <-done:
		return output, nil
	}
}

func (d *DMHost) OperateSource(sourceOpt *dmmodel.SourceOption) error {
	operation := "operate-source " + sourceOpt.String()
	res, err := d.ExecDmctlCmd(operation, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() || !utils.CheckResultTrue(res.Stdout, 2) {
		return errors.Errorf("DM operate source failed", zap.Stringer("operation", sourceOpt))
	}
	return nil
}

func (d *DMHost) StartRelay(sourceName string, workerNames ...string) error {
	if sourceName == "" && len(workerNames) == 0 {
		return errors.Errorf("task name and source id is empty")
	}
	cmd := []string{"start-relay"}
	if sourceName != "" {
		cmd = append(cmd, "-s", sourceName)
	}
	for _, s := range workerNames {
		cmd = append(cmd, s)
	}
	cmdStr := strings.Join(cmd, " ")
	res, err := d.ExecDmctlCmd(cmdStr, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() {
		return errors.Errorf("DM start relay failed", zap.String("cmd", cmdStr), zap.Reflect("res", res))
	}
	return nil
}

func (d *DMHost) TiupStartTask(taskPath string) error {
	operation := "start-task " + taskPath
	res, err := d.ExecTiupDmctlCmd(operation, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() || !utils.CheckResultFalse(res.Stdout, 0) {
		return errors.Errorf("DM start task failed")
	}
	return nil
}

func (d *DMHost) TiupCreateSource(taskPath string) error {
	operation := "operate-source create " + taskPath
	res, err := d.ExecTiupDmctlCmd(operation, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() || !utils.CheckResultFalse(res.Stdout, 0) {
		return errors.Errorf("DM start task failed")
	}
	return nil
}

func (d *DMHost) QueryStatus(taskName string, sources ...string) (string, *pb.QueryStatusListResponse, error) {
	// result of query-status has 2 format, we only support format below(must give at least one of source or task）
	// {
	//    "result": true,
	//    "msg": "",
	//    "sources": []
	//    ...
	// }
	if taskName == "" && len(sources) == 0 {
		return "", nil, errors.Errorf("task name and source id is empty")
	}

	cmd := []string{"query-status"}
	for _, s := range sources {
		cmd = append(cmd, "-s", s)
	}
	if taskName != "" {
		cmd = append(cmd, taskName)
	}
	cmdStr := strings.Join(cmd, " ")
	res, err := d.ExecDmctlCmd(cmdStr, TIMEOUT)
	if err != nil {
		return "", nil, errors.Trace(err)
	}
	if res.CheckErr() {
		return "", nil, errors.Errorf("DM operate source failed", zap.String("cmd", cmdStr), zap.Reflect("res", res))
	}
	var result pb.QueryStatusListResponse
	if err = jsonpb.Unmarshal(strings.NewReader(res.Stdout), &result); err != nil {
		// return "", nil, errors.Errorf("failed to get query-status result", zap.String("cmd", cmdStr), zap.Reflect("res", res))
		return "", nil, err
	}
	return res.Stdout, &result, nil
}

func (d *DMHost) ListMember(args ...string) (*pb.ListMemberResponse, error) {
	cmd := []string{"list-member"}
	cmd = append(cmd, args...)
	cmdStr := strings.Join(cmd, " ")
	res, err := d.ExecDmctlCmd(cmdStr, TIMEOUT)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if res.CheckErr() {
		return nil, errors.Errorf("list member failed", zap.String("cmd", cmdStr), zap.Reflect("res", res))
	}
	var result pb.ListMemberResponse
	if err = jsonpb.Unmarshal(strings.NewReader(res.Stdout), &result); err != nil {
		return nil, errors.Errorf("failed to get list-member result", zap.String("cmd", cmdStr), zap.Reflect("res", res))
	}
	return &result, nil
}

func (d *DMHost) OperateTask(taskOpt *dmmodel.TaskOption) error {
	operation := taskOpt.String()
	res, err := d.ExecDmctlCmd(operation, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() || !utils.CheckResultFalse(res.Stdout, 0) {
		return errors.Errorf("DM operate task failed", zap.Stringer("operation", taskOpt))
	}
	return nil
}

func (d *DMHost) AdjustSourceCfg(paths []string) error {
	for _, path := range paths {
		// 1. decode from file
		cfg := utils.Try(dmconfig.LoadFromFile(path)).(*dmconfig.SourceConfig)

		// 2. adjust
		if url, ok := d.upUrls[cfg.SourceID]; ok {
			cfg.From.Host = url.Hostname()
			cfg.From.Port = utils.Try(strconv.Atoi(url.Port())).(int)
		} else {
			return errors.New("Does not have this up stream.")
		}

		// 3. encode to file
		cfgString := string(utils.Try(yaml.Marshal(cfg)).([]byte))
		cfgString = strings.ReplaceAll(cfgString, "case-sensitive: false", "")
		cfgString = strings.ReplaceAll(cfgString, "filters: []", "")
		tmpPath := utils.TempPath(path)
		res, err := d.ExecDmCmd(d.Commands().WriteCfgToFile(cfgString, tmpPath).Compose(), TIMEOUT)
		if err != nil {
			return errors.Trace(err)
		}
		if res.CheckErr() {
			return errors.New("create source cfg in dm failed")
		}
		log.Info("create source cfg in dm success", zap.String("stdout", res.Stdout))
	}
	return nil
}

func (d *DMHost) AdjustTaskCfg(paths []string) error {
	for _, path := range paths {
		cfg := dmconfig.NewTaskConfig()

		// 1. decode from file
		utils.Try(cfg.DecodeFile(path))

		// 2. check
		for _, up := range cfg.MySQLInstances {
			if _, ok := d.upUrls[up.SourceID]; !ok {
				return errors.Errorf("Does not have this source: %s", up.SourceID)
			}
		}
		// 3. adjust
		cfg.TargetDB.Host = d.downUrl.Hostname()
		cfg.TargetDB.Port = utils.Try(strconv.Atoi(d.downUrl.Port())).(int)
		for _, m := range cfg.MySQLInstances {
			if len(m.MydumperConfigName) > 0 && m.Mydumper != nil {
				m.Mydumper = nil
			}
			if len(m.LoaderConfigName) > 0 && m.Loader != nil {
				m.Loader = nil
			}
			if len(m.SyncerConfigName) > 0 && m.Syncer != nil {
				m.Syncer = nil
			}
		}

		// 4. encode to file
		cfgString := string(utils.Try(yaml.Marshal(cfg)).([]byte))
		cfgString = strings.ReplaceAll(cfgString, "compact: false", "")
		cfgString = strings.ReplaceAll(cfgString, "multiple-rows: false", "")
		cfgString = strings.ReplaceAll(cfgString, "shadow-table-rules: []", "")
		cfgString = strings.ReplaceAll(cfgString, "trash-table-rules: []", "")
		cfgString = strings.ReplaceAll(cfgString, "expression-filter: {}", "")
		// annotation this scripts for dm > 2.0.3
		// cfgString = strings.ReplaceAll(cfgString, "online-ddl: false", "")
		// cfgString = strings.ReplaceAll(cfgString, "online-ddl: true", "")
		cfgString = strings.ReplaceAll(cfgString, "expression-filters: []", "")
		cfgString = strings.ReplaceAll(cfgString, "import-mode: \"\"", "")
		cfgString = strings.ReplaceAll(cfgString, "on-duplicate: \"\"", "")
		cfgString = strings.ReplaceAll(cfgString, "continuous-validator-config-name: \"\"", "")
		cfgString = strings.ReplaceAll(cfgString, "validators: {}", "")
		cfgString = strings.ReplaceAll(cfgString, "collation_compatible: loose", "")
		cfgString = strings.ReplaceAll(cfgString, "experimental:", "")
		cfgString = strings.ReplaceAll(cfgString, "async-checkpoint-flush: false", "")
		tmpPath := utils.TempPath(path)
		res, err := d.ExecDmCmd(d.Commands().WriteCfgToFile(cfgString, tmpPath).Compose(), TIMEOUT)
		if err != nil {
			return errors.Trace(err)
		}
		if res.CheckErr() {
			return errors.New("create task cfg in dm failed")
		}
		logger.Info("create task cfg in dm success")
	}
	return nil
}

func (d *DMHost) AdjustDiffCfg(paths []string) error {
	for _, path := range paths {
		cfg := &dmmodel.Config{}

		// 1. decode from file
		meta, err := toml.DecodeFile(path, cfg)
		if err != nil {
			return errors.Trace(err)
		}
		if len(meta.Undecoded()) > 0 {
			return errors.Errorf("unknown keys in config file %s: %v", path, meta.Undecoded())
		}

		// 2. adjust source and target
		for _, sourceName := range cfg.Task.Source {
			if url, ok := d.upUrls[sourceName]; ok {
				cfg.DataSources[sourceName].Host = url.Hostname()
				cfg.DataSources[sourceName].Port = utils.Try(strconv.Atoi(url.Port())).(int)
			} else {
				return errors.New("Does not have this up stream.")
			}
		}
		if cfg, ok := cfg.DataSources[cfg.Task.Target]; ok {
			cfg.Host = d.downUrl.Hostname()
			cfg.Port = utils.Try(strconv.Atoi(d.downUrl.Port())).(int)
		} else {
			return errors.New("Does not have this down stream.")
		}

		// 3. encode to file
		var b bytes.Buffer
		if err := toml.NewEncoder(&b).Encode(cfg); err != nil {
			return err
		}
		tmpPath := utils.TempPath(path)
		res, err := d.ExecSyncDiffCmd(d.Commands().WriteCfgToFile(b.String(), tmpPath).Compose(), TIMEOUT)
		if err != nil {
			return errors.Trace(err)
		}
		if res.CheckErr() {
			return errors.New("create diff cfg in sync-diff failed")
		}
		logger.Info("create diff cfg in dm success")
	}
	return nil
}

func (d *DMHost) CheckSyncDiff(path string, tryTimes int) error {
	cmd := "./sync_diff_inspector --config=" + utils.TempPath(path)
	var i int
	for i = 0; i < tryTimes; i++ {
		res, err := d.ExecSyncDiffCmd(cmd, SYNCDIFF_TIMEOUT)
		if err != nil || res.CheckErr() {
			log.Info("exec sync diff failed, will try again", zap.Int("try", i+1))
			if i == tryTimes-1 {
				fmt.Println(res.Stdout)
				fmt.Println(res.Stderr)
				log.Info("exec sync diff failed", zap.Reflect("res err", res.Err), zap.Reflect("err", err))
			}
			time.Sleep(time.Second)
			continue
		}
		break
	}
	if i >= tryTimes {
		return errors.Errorf("Sync diff failed %d times", tryTimes)
	}
	return nil
}

func (d *DMHost) SetDM(dm *apiresource.ResourceHandle) {
	dmRes := utils.Try(core.ResolveResourceHandle(dm)).(resource.DMCluster)
	d.name = dm.Name
	d.dm = &dmRes
	d.testBedName = dm.Testbed
}

func (d *DMHost) SetUpStreams(upStreams []*apiresource.ResourceHandle) {
	for _, up := range upStreams {
		upStream := utils.Try(core.ResolveResourceHandle(up)).(resource.MySQL)
		d.upStreams = append(d.upStreams, &upStream)
	}
}

func (d *DMHost) SetDownStream(downStream *apiresource.ResourceHandle) {
	downRes := utils.Try(core.ResolveResourceHandle(downStream)).(resource.TiDBCluster)
	d.downStream = &downRes
}

func (d *DMHost) SetSyncDiff(syncDiff *apiresource.ResourceHandle) {
	diffRes := utils.Try(core.ResolveResourceHandle(syncDiff)).(resource.WorkloadNode)
	d.syncDiff = &diffRes
}

// nolint: unused
func (d *DMHost) EnableGtid(u *url.URL) error {
	sql := "SET GLOBAL ENFORCE_GTID_CONSISTENCY = 'WARN';"
	sql += "SET GLOBAL ENFORCE_GTID_CONSISTENCY = 'ON';"
	sql += "SET GLOBAL GTID_MODE = 'OFF_PERMISSIVE';"
	sql += "SET GLOBAL GTID_MODE = 'ON_PERMISSIVE';"
	sql += "SET GLOBAL GTID_MODE = 'ON';"
	res, err := d.ExecUpCmd(command.NewMysqlCmd().Login(u).Sql(sql).Compose(), TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.ExitCode != 0 {
		return errors.New("enable gtid in dm failed")
	}
	return nil
}

func (d *DMHost) DisableGtid(u *url.URL) error {
	sql := "SET GLOBAL GTID_MODE = 'ON_PERMISSIVE';"
	sql += "SET GLOBAL GTID_MODE = 'OFF_PERMISSIVE';"
	sql += "SET GLOBAL GTID_MODE = 'OFF';"
	res, err := d.ExecUpCmd(command.NewMysqlCmd().Login(u).Sql(sql).Compose(), TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.ExitCode != 0 {
		return errors.New("disable gtid in dm failed")
	}
	return nil
}

func (d *DMHost) GetGtidStatus(handle *apiresource.ResourceHandle) (string, error) {
	resolved, err := core.ResolveResourceHandle(handle)
	if err != nil {
		return "", errors.Trace(err)
	}

	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return "", errors.Trace(err)
	}

	defer func() {
		if err = db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	sql := "SELECT @@GLOBAL.GTID_MODE;"
	rows, err := db.Query(sql)
	if err != nil {
		return "", errors.Trace(err)
	}
	defer func() {
		if err = rows.Close(); err != nil {
			log.Warn("close rows failed", zap.Error(err))
		}
	}()
	var res string
	for rows.Next() {
		rows.Scan(&res)
	}
	log.Info("get gtid status sucessfully", zap.String("gtid_mode", res))
	return res, nil
}

func (d *DMHost) OperateRelay(relayOpt *dmmodel.RelayOption) error {
	operation := relayOpt.String()
	res, err := d.ExecDmctlCmd(operation, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() || !utils.CheckResultFalse(res.Stdout, 0) {
		return errors.Errorf("DM operate relay failed", zap.Stringer("operation", relayOpt))
	}
	return nil
}

func (d *DMHost) TransferSource(sourceName, workerName string) error {
	cmd := []string{"transfer-source", sourceName, workerName}
	cmdStr := strings.Join(cmd, " ")
	res, err := d.ExecDmctlCmd(cmdStr, TIMEOUT)
	if err != nil {
		return errors.Trace(err)
	}
	if res.CheckErr() {
		return errors.Errorf("transfer-source failed", zap.String("cmd", cmdStr), zap.Reflect("res", res))
	}
	var result pb.TransferSourceResponse
	if err = jsonpb.Unmarshal(strings.NewReader(res.Stdout), &result); err != nil {
		return errors.Errorf("failed to get transfer-source result", zap.String("cmd", cmdStr), zap.Reflect("res", res))
	}
	if !result.Result {
		return errors.Errorf("failed to transfer-source", zap.String("cmd", cmdStr), zap.String("msg", result.Msg))
	}
	return nil
}

func (d *DMHost) ShowDDLLocks(sourceName, taskName string) (error, pb.ShowDDLLocksResponse) {
	var result pb.ShowDDLLocksResponse

	cmd := []string{"show-ddl-locks"}
	if sourceName != "" {
		cmd = append(cmd, fmt.Sprintf("--source=%s", sourceName))
	}
	if taskName != "" {
		cmd = append(cmd, taskName)
	}
	cmdStr := strings.Join(cmd, " ")

	resp, err := d.ExecDmctlCmd(cmdStr, TIMEOUT)
	if err != nil {
		return errors.Trace(err), result
	}
	if resp.CheckErr() {
		return errors.Errorf("show-ddl-locks failed", zap.String("cmd", cmdStr), zap.Reflect("resp", resp)), result
	}
	if err = jsonpb.Unmarshal(strings.NewReader(resp.Stdout), &result); err != nil {
		return errors.Errorf("failed to get show-ddl-locks result", zap.String("cmd", cmdStr), zap.Error(err), zap.Reflect("resp", resp)), result
	}
	if !result.Result {
		return errors.Errorf("failed to show-ddl-locks", zap.String("cmd", cmdStr), zap.String("msg", result.Msg)), result
	}

	return nil, result
}

func (d *DMHost) UnlockDDLLock(sourceName, taskName string, forceRemove bool) error {
	err, ddlLocks := d.ShowDDLLocks(sourceName, taskName)
	if err != nil {
		return err
	}

	ddlLock := ddlLocks.GetLocks()[0]
	lockID := ddlLock.ID
	owner := ddlLock.Owner
	cmd := []string{"unlock-ddl-lock"}
	if owner != "" {
		cmd = append(cmd, fmt.Sprintf("--owner=%s", owner))
	}
	if forceRemove {
		cmd = append(cmd, "--force-remove=true")
	}
	if lockID == "" {
		return errors.Errorf("unlock-ddl-lock must specify lock-id", zap.String("lock-id", lockID))
	}
	lockID = strings.ReplaceAll(lockID, "`", "\\`")
	cmd = append(cmd, lockID)

	cmdStr := strings.Join(cmd, " ")
	resp, err := d.ExecDmctlCmd(cmdStr, TIMEOUT*2)
	if err != nil {
		return errors.Trace(err)
	}
	if resp.CheckErr() {
		return errors.Errorf("unlock-ddl-lock failed", zap.String("cmd", cmdStr), zap.Reflect("resp", resp))
	}
	var result pb.UnlockDDLLockResponse
	if err = jsonpb.Unmarshal(strings.NewReader(resp.Stdout), &result); err != nil {
		return errors.Errorf("unlock-ddl-lock failed", zap.String("cmd", cmdStr), zap.Error(err), zap.Reflect("resp", resp))
	}
	if !result.Result {
		return errors.Errorf("unlock-ddl-lock failed", zap.String("cmd", cmdStr), zap.String("msg", result.Msg))
	}

	return nil
}

func (d *DMHost) Get1GBDMBinlogConsumeDuration(name string) (*command.ExecOutput, error) {
	_, _, _, err := d.execOnWorker(name, "/bin/sh", "-c", "/bin/grep 'event=rotate' /var/lib/dm-worker/log/dm-worker.log | awk '{print $1, $2}' | awk -F'202' -vOFS=\"202\" '{$1=\"\";$1=$1}1' | sed 's/\\//-/g' | awk '{sub(/.{4}$/,\"\")}1' > test")
	stdout, stderr, exitCode, err := d.execOnWorker(name, "/bin/sh", "-c", "res1=`head -n 1 /test`; res2=`tail -n 1 /test`; re1=$(date +%s -d \"$res1\"); re2=$(date +%s -d \"$res2\"); i=$(($re2-$re1)); echo $i")
	// stdout, stderr, exitCode, err := d.execOnWorker(name, cmd1)
	return &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}, nil
}

func (d *DMHost) DMWorkerVersion(name string) (*command.ExecOutput, error) {
	stdout, stderr, exitCode, err := d.execOnWorker(name, "/bin/sh", "-c", "/dm-worker -V")
	// stdout, stderr, exitCode, err := d.execOnWorker(name, cmd1)
	return &command.ExecOutput{Err: err, Stdout: stdout, Stderr: stderr, ExitCode: exitCode}, nil
}