package drautosync

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"io"
	"io/ioutil"
	"math/rand"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	. "github.com/onsi/gomega"
	"github.com/pingcap/endless/pkg/client/tidb"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/workloadnode"
	"github.com/pingcap/endless/pkg/workloadnode/tiup"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
	"github.com/pingcap/endless/testcase/oltp/tiuptest/tiuppkg"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	"github.com/tidwall/gjson"
	"go.uber.org/zap"
)

type DrClusterInfo struct {
	Cluster         *tiup.TiUPCluster
	PrimaryTikvs    []string
	DrTikvs         []string
	DrPD            string
	PrimaryPds      []string
	DbURL           *url.URL
	PdURL           *url.URL
	TiupNode        resource.WorkloadNode
	ClusterName     string
	Version         string
	PrimaryStoreIds []string
	DrStoreIds      []string
	ChaosIDs        []map[string]resource.ChaosIdentifier
	NodeWorkloadMap map[string]resource.WorkloadNode
	BankDbName      string
	RunTime         oltppkg.RunTime
	PromURL         *url.URL
	TiupMirror      string
}

func CompareTwoMap(map1 map[string]int64, map2 map[string]int64) int {
	if len(map1) != len(map2) {
		log.Info(fmt.Sprintf("map count is not equal %v %v", map1, map2),
			zap.Int("len(map1)", len(map1)),
			zap.Int("len(map2)", len(map2)))
		return -1
	}
	for k, v := range map1 {
		if map2[k] == v {
			continue
		}
		if map2[k] < v {
			log.Info("table count is more than original", zap.String("tableName", k), zap.Int64("original count", v), zap.Int64("last count", map2[k]))
			return 1
		}
		if map2[k] > v {
			log.Info("table count is less than original ", zap.String("tableName", k), zap.Int64("original count", v), zap.Int64("last count", map2[k]))
			return -1
		}
	}
	//log.Info(fmt.Sprintf("compare succeeded,original:%v last:%v", map1, map2))
	return 0
}
func (c *DrClusterInfo) GetTableSchemaVersion(ctx context.Context, dbName string, tableName string) error {
	url := fmt.Sprintf("http://%s.%s:10080/schema/%s/%s", "tidb2-peer", c.Cluster.TestBed, dbName, tableName)
	resp, err := http.Get(url)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	body, err := io.ReadAll(resp.Body)
	if err != nil {
		return err
	}
	log.Info("", zap.ByteString("", body))
	// res := gjson.GetBytes(body, "id")
	// log.Info(res.Str)
	return nil
}
func (c *DrClusterInfo) GetAllTablesCounts(ctx context.Context, dbName string) map[string]int64 {
	log.Info("Get all tables")
	var tableName string
	var tables []string
	db := Try(c.Cluster.GetDB(dbName)).(*sql.DB)
	defer db.Close()

	rows, err := db.Query("show tables;")
	Expect(err).Should(Succeed())
	for rows.Next() {
		rows.Scan(&tableName)
		tables = append(tables, tableName)
	}
	defer rows.Close()
	log.Info("tables info", zap.Strings("tables", tables), zap.Int("tableNum", len(tables)))

	tableCount := map[string]int64{}
	for _, tableName := range tables {
		c.GetTableSchemaVersion(ctx, dbName, tableName)
		var count int64
		log.Info(fmt.Sprintf("select count(*) from %s;", tableName))
		ignoreTables := oltppkg.StrSlice{"CLUSTER_LOG", "METRICS_SUMMARY",
			"TABLE_STORAGE_STATS", "INSPECTION_SUMMARY", "METRICS_SUMMARY_BY_LABEL", "TIDB_HOT_REGIONS_HISTORY"}
		if ignoreTables.Has(tableName) {
			log.Info("skip table", zap.String("tableName", tableName))
			continue
		}
	Loop:
		for i := 0; i < 1; i++ {
			rows, err = db.Query(fmt.Sprintf("select count(*) from %s;", tableName))
			if err == nil {
				for rows.Next() {
					rows.Scan(&count)
					tableCount[tableName] = count
					break Loop
				}
			} else {
				log.Info("query failed,retry", zap.String("tableName", tableName), zap.String("error mesg", err.Error()))
				oltppkg.WaitFor(ctx, "waiting for ready", 10*time.Second)
				continue
			}
		}
		Expect(err).Should(Succeed())
	}
	//oltppkg.WaitFor(ctx, "waiting for scan next table", 5*time.Second)
	log.Info(fmt.Sprintf("tableCount %v", tableCount))
	return tableCount
}

//
func (c *DrClusterInfo) InjectNetworkLatency(node string) (resource.ChaosIdentifier, error) {
	experiment := &resource.NetworkLatencySpec{Latency: "300ms", Correlation: "50", Jitter: "100ms"}
	selector, err := c.NodeWorkloadMap[node].ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})
	Expect(err).Should(BeNil())
	chaoId, err := c.NodeWorkloadMap[node].ChaosClient().Create(
		resource.InjectChaosOptions{
			Selector:       selector,
			ExperimentSpec: experiment,
		},
	)
	Expect(err).Should(BeNil())
	return chaoId, err
}

//
func WaitForReplicateStatus(tc resource.MySQLCompatible, timeOut time.Duration, interval time.Duration, status string) {
	Eventually(func() string {
		curStatus := oltppkg.GetRepliactionStatus(tc)
		return curStatus
	}, timeOut, interval).Should(Equal(status))
}
func (c *DrClusterInfo) PrintDrStat(ctx context.Context, nodeList []string) map[string]string {
	nodeStatus := map[string]string{}
	for _, node := range nodeList {
		workloadNode := c.NodeWorkloadMap[node]
		stdout, _, _, err := workloadNode.Exec(resource.WorkloadNodeExecOptions{
			Command: "cat /tiup/data/pd-2379/DR_STATE",
			Peer:    0})
		Expect(err).Should(BeNil())
		log.Info(stdout)
		state := gjson.GetBytes([]byte(stdout), "state").String()
		nodeStatus[node] = state
		stdout, _, _, err = workloadNode.Exec(resource.WorkloadNodeExecOptions{
			Command: "ls -lrt /tiup/data/pd-2379/DR_STATE",
			Peer:    0})
		Expect(err).Should(BeNil())
		log.Info(stdout)
	}
	return nodeStatus

}
func (c *DrClusterInfo) InitStoreId() {
	storeIdMap := oltppkg.GetStoreIdAddrMap(c.PdURL, "id", "address")
	for id, address := range storeIdMap {
		for _, pTikv := range c.PrimaryTikvs {
			if strings.HasPrefix(address, pTikv+"-peer") {
				c.PrimaryStoreIds = append(c.PrimaryStoreIds, id)
			}
		}
		for _, pTikv := range c.DrTikvs {
			if strings.HasPrefix(address, pTikv) {
				c.DrStoreIds = append(c.DrStoreIds, id)
			}
		}
	}
	log.Info(fmt.Sprintf("storeIdMap %v %v %v", storeIdMap, c.PrimaryStoreIds, c.DrStoreIds))
}

func (c *DrClusterInfo) RunSysbench(ctx context.Context) {
	log.Info("run sysbench write only")
	dbName := "test"
	dbURL := Try(c.Cluster.ServiceURL(resource.DBAddr)).(*url.URL)
	c.DbURL = dbURL
	sysbench := oltppkg.NewSysbench("oltp_insert", "run")
	sysbenchcmd := sysbench.Db(dbName).Host(c.DbURL.Hostname()).Port(c.DbURL.Port()).Threads(32).
		Tables(32).TableSize(10000000).IgnoreError("2013,1213,1105,1205,8022,8027,8028,9004,9007")
	runcmd := sysbenchcmd.Build()
	oltppkg.ExecWorkloadCmd(ctx, c.NodeWorkloadMap["benchtoolset"], runcmd, false, nil)
}
func (c *DrClusterInfo) StartCluster(ctx context.Context) {
	//_, err := c.TiupInstallCluster(ctx)
	err := c.Cluster.Start()
	Expect(err).Should(BeNil())
}
func (c *DrClusterInfo) RestoreBackDc(ctx context.Context) {
	log.Info("Remove fail stores")
	c.TiupNode.WriteFile(resource.WorkloadNodeWriteFileOptions{
		Content:  strings.NewReader(DrMetaConfig),
		FilePath: fmt.Sprintf("/root/.tiup/storage/cluster/clusters/%s/meta.yaml", c.ClusterName),
	})
	cmd := fmt.Sprintf("sed -i '/pd-server/ s#pd-server#pd-server --force-new-cluster#' %s/scripts/run_pd.sh ",
		"/tiup/deploy/pd-2379")
	err := workloadnode.CommandContext(ctx, cmd).Run(c.NodeWorkloadMap["pd3"])
	Expect(err).Should(BeNil())
	cmd = fmt.Sprintf("tiup cluster stop %s -N %s-peer:2379 -y", c.ClusterName, "pd3")
	err = workloadnode.CommandContext(ctx, cmd).Run(c.TiupNode)
	Expect(err).Should(BeNil())
	cmd = fmt.Sprintf("tiup cluster start %s -N %s-peer:2379 -y", c.ClusterName, "pd3")
	err = workloadnode.CommandContext(ctx, cmd).Run(c.TiupNode)
	Expect(err).Should(BeNil())
	oltppkg.WaitFor(ctx, "waiting for leader", 2*time.Minute)
	c.RemoveFailStoresOnline(ctx)
	cmd = fmt.Sprintf(`/bin/pd-recover --from-old-member --endpoints=http://%s-peer:2379`, "pd3")
	err = workloadnode.CommandContext(ctx, cmd).Run(c.NodeWorkloadMap["benchtoolset"])
	Expect(err).Should(BeNil())
}
func (c *DrClusterInfo) RestartBackDc(ctx context.Context) {
	cmd := fmt.Sprintf("tiup cluster start %s -y", c.ClusterName)
	err := workloadnode.CommandContext(ctx, cmd).Run(c.TiupNode)
	Expect(err).Should(BeNil())
	c.Cluster.Display()
	Try(c.Cluster.Start())

	oltppkg.WaitFor(ctx, "waiting for cluster ready", 2*time.Minute)
	oltppkg.CheckClusterHealth(c.Cluster)

	c.PdURL = Try(c.Cluster.ServiceURL(resource.PDAddr)).(*url.URL)
	SetPlacementRule(ctx, c.Cluster, c.TiupNode, DrPlacementRuleConfig)
	c.TiupExecPdCtl(ctx, "config set replication-mode majority")
	c.TiupExecPdCtl(ctx, "config placement-rules show")
	cmd = fmt.Sprintf("tiup cluster restart %s -y", c.ClusterName)
	for i := 0; i < 3; i++ {
		err = workloadnode.CommandContext(ctx, cmd).Run(c.TiupNode)
		if err == nil {
			break
		}
	}
	Expect(err).Should(BeNil())
}
func DestroyPrimaryCluster(ctx context.Context, clusterName string, tiupNode resource.WorkloadNode) {
	cmd := fmt.Sprintf("tiup cluster destroy %s -y --force", clusterName)
	tiupNode.WriteFile(resource.WorkloadNodeWriteFileOptions{
		Content:  strings.NewReader(MetaConfig),
		FilePath: fmt.Sprintf("/root/.tiup/storage/cluster/clusters/%s/meta.yaml", clusterName),
	})
	workloadnode.CommandContext(ctx, cmd).Run(tiupNode)
}

func (c *DrClusterInfo) RemoveFailStores(ctx context.Context) {
	//primaryTikvPeer := []string{"tikv-1", "tikv-2", "tikv-3"}
	log.Info("Begin to remove fail stores")
	defer func() {
		log.Info("Finish to remove fail stores.")
	}()
	backupTikvPeer := c.DrTikvs
	removeStoreIDs := c.PrimaryStoreIds
	tikvCtlPath := "/root/.tiup/components/ctl"
	dataDir := "/tiup/data/tikv-20160"

	Expect(len(removeStoreIDs)).Should(Equal(3))

	for _, peer := range backupTikvPeer {
		cmd := fmt.Sprintf(
			`scp -o "UserKnownHostsFile=/dev/null" -o "StrictHostKeyChecking=no" \
		%s/*%s*/tikv-ctl \
		%s-peer:/root/`, tikvCtlPath, c.Version, peer)
		err := workloadnode.CommandContext(ctx, cmd).Run(c.TiupNode)
		Expect(err).Should(BeNil())
		stopCmd := fmt.Sprintf("tiup cluster stop dbcluster -N %s-peer -y", peer)
		err = workloadnode.CommandContext(ctx, stopCmd).Run(c.TiupNode)
		Expect(err).Should(BeNil())
	}
	//ssh ${i%,*} ./tikv-ctl --data-dir ${i#*,} unsafe-recover remove-fail-stores -s ${STORE_ID} --all-regions
	for _, peer := range backupTikvPeer {
		for _, storeid := range removeStoreIDs {
			cmd := fmt.Sprintf(
				`./tikv-ctl --data-dir %s unsafe-recover remove-fail-stores -s %s --all-regions`, dataDir, storeid)
			err := workloadnode.CommandContext(ctx, cmd).Run(c.NodeWorkloadMap[peer])
			Expect(err).Should(BeNil())
			// run twice here to w/a issue https://github.com/tikv/tikv/issues/12307
			err = workloadnode.CommandContext(ctx, cmd).Run(c.NodeWorkloadMap[peer])
			Expect(err).Should(BeNil())
		}
	}
}

func (c *DrClusterInfo) RemoveFailStoresOnline(ctx context.Context) {
	log.Info("Begin to remove fail stores")
	defer func() {
		log.Info("Finish to remove fail stores.")
	}()
	removeStoreIDs := c.PrimaryStoreIds
	//Expect(len(removeStoreIDs)).Should(Equal(3))

	storeIDs := strings.Join(removeStoreIDs, ",")
	pdCtlCmd := fmt.Sprintf("unsafe remove-failed-stores %s", storeIDs)
	pdUrl := fmt.Sprintf("http://%s-peer.%s:2379", "pd3", c.Cluster.TestBed)
	cmd := fmt.Sprintf("tiup ctl:%s pd -u %s %s", c.Version, pdUrl, pdCtlCmd)
	log.Info("will run cmd", zap.String("cmd:", cmd))
	stdout, _, _, err := c.TiupNode.Exec(resource.WorkloadNodeExecOptions{Command: cmd})
	log.Info("cmd output", zap.String("cmd output:", stdout))
	Expect(err).Should(BeNil())
	Expect(strings.Contains(stdout, "Success")).Should(BeTrue())
	Try(c.GetOnlineRecoveryResult(ctx, 1*time.Minute, 1*time.Hour))
}

func (c *DrClusterInfo) DownPrimaryDc(ctx context.Context) {
	chaosIds := map[string]resource.ChaosIdentifier{}
	for _, tikv := range append(c.PrimaryPds, c.PrimaryTikvs...) {
		log.Info("Down node %s", zap.String("downNode", tikv))
		id, _ := NetworkDownWorkloadNode(ctx, c.NodeWorkloadMap[tikv])
		chaosIds[tikv] = id
	}
	c.ChaosIDs = append(c.ChaosIDs, chaosIds)
}
func (c *DrClusterInfo) DownBackupDc(ctx context.Context) (map[string]resource.ChaosIdentifier, error) {
	log.Info("Down backup dc now")
	chaosIds := map[string]resource.ChaosIdentifier{}
	for _, tikv := range append(c.DrTikvs, c.DrPD) {
		log.Info("Down node %s", zap.String("downNode", tikv))
		id, err := NetworkDownWorkloadNode(ctx, c.NodeWorkloadMap[tikv])
		if err != nil {
			return nil, err
		}
		chaosIds[tikv] = id
	}
	c.ChaosIDs = append(c.ChaosIDs, chaosIds)
	return chaosIds, nil
}
func NetworkDownWorkloadNode(ctx context.Context, controlerNode resource.WorkloadNode) (resource.ChaosIdentifier, error) {
	log.Info("Inject chaos to workloadNode", zap.String("workloadName", controlerNode.Handle().Name))
	selectors, _ := controlerNode.ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})
	experiment := &resource.NetworkLossSpec{Loss: "100", Correlation: "100"}
	id, err := controlerNode.ChaosClient().Create(
		resource.InjectChaosOptions{
			Selector:       selectors,
			ExperimentSpec: experiment,
		},
	)
	return id, err
}
func (c *DrClusterInfo) InjectNetworkDelayBetweenDC(networkDelay string) (resource.ChaosIdentifier, error) {
	soureNodes := strings.Join(c.PrimaryTikvs, ",") + "," + strings.Join(c.PrimaryPds, ",") + "," + "tidb1"
	var sourceSelectors []resource.ChaosSelector
	for _, n := range strings.Split(soureNodes, ",") {
		w := c.NodeWorkloadMap[n]
		sourceSelectors = append(sourceSelectors,
			Try(w.ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})).([]resource.ChaosSelector)...,
		)
	}
	targetNodes := strings.Join(c.DrTikvs, ",") + "," + c.DrPD + "," + "tidb2"
	var targetSelectors []resource.ChaosSelector
	for _, n := range strings.Split(targetNodes, ",") {
		w := c.NodeWorkloadMap[n]
		targetSelectors = append(targetSelectors,
			Try(w.ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})).([]resource.ChaosSelector)...,
		)
	}

	workloadNode := c.NodeWorkloadMap[strings.Split(soureNodes, ",")[0]]
	log.Info("Run chaos",
		zap.String("name", "network-delay"),
		zap.String("soureNodes", soureNodes),
		zap.String("targetNodes", targetNodes),
		zap.String("networkDelay", networkDelay))
	id, err := workloadNode.ChaosClient().Create(
		resource.InjectChaosOptions{
			Selector: sourceSelectors,
			Target:   &targetSelectors,
			ExperimentSpec: &resource.NetworkLatencySpec{
				Latency:     networkDelay,
				Correlation: "80",
				Jitter:      "0ms",
			},
		},
	)
	chaosIds := map[string]resource.ChaosIdentifier{strings.Split(soureNodes, ",")[0]: id}
	c.ChaosIDs = append(c.ChaosIDs, chaosIds)
	return id, err
}
func (c *DrClusterInfo) InjectPartitionNetwork(soureNodes string, targetNodes string) (resource.ChaosIdentifier, error) {
	var sourceSelectors []resource.ChaosSelector
	for _, n := range strings.Split(soureNodes, ",") {
		w := c.NodeWorkloadMap[n]
		sourceSelectors = append(sourceSelectors,
			Try(w.ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})).([]resource.ChaosSelector)...,
		)
	}
	var targetSelectors []resource.ChaosSelector
	for _, n := range strings.Split(targetNodes, ",") {
		w := c.NodeWorkloadMap[n]
		targetSelectors = append(targetSelectors,
			Try(w.ChaosSelect(resource.WorkloadNodeSelectorOpts{All: true})).([]resource.ChaosSelector)...,
		)
	}

	workloadNode := c.NodeWorkloadMap[strings.Split(soureNodes, ",")[0]]
	id, err := workloadNode.ChaosClient().Create(
		resource.InjectChaosOptions{
			Selector:       sourceSelectors,
			Target:         &targetSelectors,
			ExperimentSpec: &resource.NetworkPartitionSpec{},
		},
	)
	return id, err
}

func StaleRead(ctx context.Context, db *sql.DB, tableName string) {
	//db, _, err := ConnectDB(ctx, cluster, "tidb-2", "test", 14000)
	//Expect(err).Should(BeNil())
	_, err := db.ExecContext(ctx, `set @@tidb_replica_read='closest-replicas'`)
	Expect(err).Should(BeNil())
	rows, err := db.QueryContext(ctx, `show VARIABLES like "tidb_replica_read"`)
	if err != nil {
		log.Info(err.Error())
	} else {
		r, err := oltppkg.DumpToByteRows(rows)
		Expect(err).Should(BeNil())
		result := r.ConvertToString("")
		log.Info(result)
	}
	defer db.Close()
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		staleSubfix := "AS OF TIMESTAMP TIDB_BOUNDED_STALENESS(NOW() - INTERVAL 2000 SECOND, NOW())"
		rows, err := db.QueryContext(ctx, fmt.Sprintf("select count(*) from %s %s", tableName, staleSubfix))
		if err != nil {
			log.Info(err.Error())
		} else {
			r, err := oltppkg.DumpToByteRows(rows)
			Expect(err).Should(BeNil())
			result := r.ConvertToString("")
			log.Info(result)
		}

	}
}
func ParallelStaleRead(ctx context.Context, cluster *tiup.TiUPCluster, dbName string) {
	concurrency := 100
	var wg sync.WaitGroup
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			//tname := tableNameList[GetRandInt(0, len(tableNameList))]
			db := Try(cluster.GetDB("")).(*sql.DB)
			StaleRead(ctx, db, "sbtest1")
		}()
	}
	wg.Wait()
}
func WriteLocalFileToRemote(
	ctx context.Context,
	controlerNode resource.WorkloadNode,
	localFile string,
	remoteFile string,
) error {
	str, _ := os.Getwd()
	path := fmt.Sprintf("%s/%s", str, localFile)
	rule, err := ioutil.ReadFile(path)
	if err != nil {
		return err
	}
	// write rule to remote
	err = controlerNode.WriteFile(resource.WorkloadNodeWriteFileOptions{
		Content:  bytes.NewReader(rule),
		FilePath: remoteFile,
	})
	if err != nil {
		return err
	}
	return nil
}
func SetPlacementRule(ctx context.Context, tc resource.MySQLCompatible, tiupNode resource.WorkloadNode, ruleContent string) {
	filePath := fmt.Sprintf("/tmp/%s.rule.json", time.Now().Format(time.RFC3339))
	Try(tiupNode.WriteFile(resource.WorkloadNodeWriteFileOptions{
		Content:  strings.NewReader(ruleContent),
		FilePath: filePath,
	}))
	oltppkg.BasicClusterExecPdCtl(tc, fmt.Sprintf("config placement-rules rule-bundle save --in=%s", filePath))
	oltppkg.BasicClusterExecPdCtl(tc, "config placement-rules show")
}
func (c *DrClusterInfo) ImportData(ctx context.Context) error {
	pdURL := c.PdURL
	db := Try(c.ConnectDB("")).(*sql.DB)
	s3Parms := oltppkg.S3Parms{S3Endpoint: "http://minio.pingcap.net:9000", S3AccessKeyID: "minioadmin", S3SecretAccessKey: "minioadmin"}
	tidbClient := tidb.NewClient(db)
	brcmd := oltppkg.NewBr("/bin", "restore full", s3Parms.S3AccessKeyID, s3Parms.S3SecretAccessKey)
	brTpccCmd := brcmd.Pd(pdURL.String()).Storage("s3://benchmark/sysbench-32-10000000").S3endpoint(s3Parms.S3Endpoint).CheckReq(false).Build()

	exist := Try(tidbClient.IsTableExist(ctx, "test", "sbtest1")).(bool)
	if exist {
		_, err := db.ExecContext(ctx, fmt.Sprintf("drop database if exists %s", "test"))
		if err != nil {
			return err
		}
	}
	oltppkg.ExecWorkloadCmd(ctx, c.NodeWorkloadMap["benchtoolset"], brTpccCmd, true, nil)
	return nil
}

func (c *DrClusterInfo) ConnectDB(dbName string) (*sql.DB, error) {
	dsn := resource.BuildDSN(c.DbURL, dbName)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		return nil, err
	}

	if err = db.Ping(); err != nil {
		return nil, err
	}
	return db, nil
}

func (c *DrClusterInfo) TiupExecPdCtl(
	ctx context.Context,
	pdCtlCmd string,
) {
	cmd := fmt.Sprintf("tiup ctl:%s pd -u %s %s", c.Version, c.PdURL.String(), pdCtlCmd)
	Try(workloadnode.CommandContext(ctx, cmd).Run(c.TiupNode))
}
func (c *DrClusterInfo) BuildCluster(topoFile string) error {
	str, _ := os.Getwd()
	path := fmt.Sprintf("%s/%s", str, topoFile)
	t, err := ioutil.ReadFile(path)
	if err != nil {
		return err
	}
	tiupTopo := string(t)
	cluster := tiup.NewTiUPCluster(c.TiupNode, "tiup-peer", tiupTopo, c.ClusterName).WithCMDTimeout(60 * time.Minute)
	c.Cluster = cluster
	return nil
}

// func (c *DrClusterInfo) TiupInstallCluster(
// 	ctx context.Context, topo string,
// ) (*tiup.TiUPClusterInfo, error) {
// 	c.Cluster = tiup.NewTiUPCluster(c.TiupNode, "tiup-peer", topo, c.ClusterName).WithCMDTimeout(60 * time.Minute)
// 	info, err := c.Cluster.Display()
// 	if err != nil {
// 		Try(c.Cluster.Deploy(c.Version))
// 		Try(c.Cluster.Start())
// 		info, err = c.Cluster.Display()
// 		Expect(err).Should(BeNil())
// 	}
// 	log.Info("Get cluster version", zap.String("current version:", info.ClusterMeta.ClusterVersion))
// 	return &info, nil
// }
func ParallelRunInsert(ctx context.Context, tc resource.TiDBCluster) {
	concurrency := 5
	var wg sync.WaitGroup
	tableNameList := oltppkg.GetAllTablesNames(ctx, tc, "test")
	dbName := "test"
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			tname := tableNameList[oltppkg.GetRandInt(0, len(tableNameList))]
			InsertToDB(ctx, tc, dbName, tname)
		}()
	}
	wg.Wait()
}

func RandStringBytes(bytes []byte, rnd *rand.Rand) {
	letterBytes := "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
	for i := range bytes {
		bytes[i] = letterBytes[rnd.Intn(len(letterBytes))]
	}
}
func InsertToDB(ctx context.Context, tc resource.TiDBCluster, dbName string, tname string) {
	db := Try(tc.GetDB(dbName)).(*sql.DB)
	defer db.Close()
	// err = prepare_insert(ctx, db, tname)
	// if err != nil {
	// 	return
	// }
	//ctx, cancel := context.WithTimeout(ctxr, 30*time.Second)
	for {
		select {
		case <-ctx.Done():
			return
		default:
			// insert data into table from
			//start_time := time.Now()
			tx, err := db.BeginTx(ctx, nil)
			if err != nil {
				//ri <- false
				break
			}

			sqlStmt := fmt.Sprintf("INSERT INTO %s(k,c) values(%d,'%s')",
				tname, rand.Intn(1000), RandStringRunes(50))
			_, err = tx.ExecContext(ctx, sqlStmt)
			if err != nil {
				if rollbackErr := tx.Rollback(); rollbackErr != nil {
					log.Warn(fmt.Sprintf("Insert into failed: %v, unable to rollback: %v\n", err, rollbackErr))
				}
				//ri <- false
				break
			}
			err = tx.Commit()
			if err != nil {
				//ri <- false
				break
			} else {
				log.Info("Commit succeed:", zap.String("sqlStmt", sqlStmt))
			}
		}
	}
}

func RandStringRunes(n int) string {
	rand.Seed(time.Now().UnixNano())
	var letterRunes = []rune("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")
	b := make([]rune, n)
	for i := range b {
		b[i] = letterRunes[rand.Intn(len(letterRunes))]
	}
	return string(b)
}

func InitDrCluster(caseParams *oltppkg.CaseParams) *DrClusterInfo {
	ctx := caseParams.Ctx
	tiupConfig := caseParams.CaseCmdParms.TiupClusterConfig
	tc := caseParams.Tc.(*tiup.TiUPCluster)
	version := tiupConfig.Version
	tiupMirror := tiupConfig.TiupMirror
	clusterName := tiupConfig.ClusterName
	runTime := caseParams.CaseCmdParms.RunTime
	log.Info("Prepare TiDB Cluster")
	podMaps := make(map[string]resource.WorkloadNode)
	podMaps["benchtoolset"] = ctx.Resource("benchtoolset").(resource.WorkloadNode)
	podMaps["br"] = ctx.Resource("benchtoolset").(resource.WorkloadNode)
	controlerNode := ctx.Resource("tiup").(resource.WorkloadNode)

	// clusterName := "dbcluster"
	clusterInfo := DrClusterInfo{
		Cluster:         tc,
		TiupNode:        controlerNode,
		PrimaryTikvs:    []string{"tikv1", "tikv2", "tikv3"},
		DrTikvs:         []string{"tikv4", "tikv5", "tikv6"},
		PrimaryPds:      []string{"pd1", "pd2"},
		DrPD:            "pd3",
		ClusterName:     clusterName,
		Version:         version,
		NodeWorkloadMap: podMaps,
		BankDbName:      "bank",
		RunTime:         runTime,
		TiupMirror:      tiupMirror,
	}

	// is use big cluster
	if tiupConfig.TiupDrAutoSyncConfig.PrimayBackupTikvNum == "6:6" {
		clusterInfo.PrimaryTikvs = append(clusterInfo.PrimaryTikvs, []string{"tikv7", "tikv8", "tikv9"}...)
		clusterInfo.DrTikvs = append(clusterInfo.DrTikvs, []string{"tikv10", "tikv11", "tikv12"}...)
	}
	for _, podName := range []string{"tidb1", "tidb2"} {
		podMaps[podName] = ctx.Resource(podName).(resource.WorkloadNode)
	}
	for _, podName := range append(clusterInfo.PrimaryTikvs, clusterInfo.DrTikvs...) {
		podMaps[podName] = ctx.Resource(podName).(resource.WorkloadNode)
	}
	for _, podName := range append(clusterInfo.PrimaryPds, clusterInfo.DrPD) {
		podMaps[podName] = ctx.Resource(podName).(resource.WorkloadNode)
	}
	clusterInfo.NodeWorkloadMap = podMaps
	for k, v := range clusterInfo.NodeWorkloadMap {
		log.Info("NodeWorkloadMap ", zap.String(k, v.Handle().Name))
	}

	Try(os.Setenv("ENDLESS_CMD_STDOUT", "stdout"))
	oltppkg.LogAndBy("Step: Init and start cluster")
	tiuppkg.SetTiupMirror(ctx, clusterInfo.TiupNode, tiupConfig.TiupMirror)
	installCtl := fmt.Sprintf("tiup install ctl:%s", tiupConfig.Version)
	Try(workloadnode.CommandContext(ctx, installCtl).Run(clusterInfo.TiupNode))

	clusterInfo.PromURL = Try(tc.ServiceURL(resource.Prometheus)).(*url.URL)
	clusterInfo.DbURL = Try(tc.ServiceURL(resource.DBAddr)).(*url.URL)
	clusterInfo.PdURL = Try(tc.ServiceURL(resource.PDAddr)).(*url.URL)

	log.Info("set some parameters")
	clusterInfo.InitStoreId()

	clusterInfo.CreatDatabase(ctx, BankConfig.DbName)

	return &clusterInfo
}
func SetDrAutoSyncPlacementRule(ctx context.Context, tiupNode resource.WorkloadNode, caseParams *oltppkg.CaseParams) {
	backStoreRole := caseParams.CaseCmdParms.TiupClusterConfig.TiupDrAutoSyncConfig.BackupDcRole
	if backStoreRole == "" {
		backStoreRole = "voter"
	}
	var PlacementRuleConfig string
	replicaNum := caseParams.CaseCmdParms.TiupClusterConfig.TiupDrAutoSyncConfig.ReplicaNum
	if replicaNum == 0 || replicaNum == 5 {
		PlacementRuleConfig = fmt.Sprintf(PlacementRuleConfigTmp, 3, backStoreRole, 2, "logic4", "logic5")
	} else {
		PlacementRuleConfig = fmt.Sprintf(PlacementRuleConfigTmp, 2, backStoreRole, 1, "logic4", "logic5")
	}

	SetPlacementRule(ctx, caseParams.Tc, tiupNode, PlacementRuleConfig)

	WaitForReplicateStatus(caseParams.Tc, 30*time.Minute, 60*time.Second, "sync")
}
func (c *DrClusterInfo) CreatDatabase(ctx context.Context, dbName string) {
	db := Try(c.ConnectDB("")).(*sql.DB)
	defer db.Close()
	tidbClient := tidb.NewClient(db)
	Try(tidbClient.ShowClusterVersionInformation())
	Try(db.ExecContext(ctx, fmt.Sprintf("create database IF NOT EXISTS %s ", dbName)))
}

func (dr *DrClusterInfo) CleanChaos() {
	clusterInfo := dr
	log.Info(fmt.Sprintf("clusterInfo %v", clusterInfo))
	if clusterInfo.ChaosIDs != nil {
		log.Info(fmt.Sprintf("Delete chaos %v", clusterInfo.ChaosIDs))
		for _, chaosMap := range clusterInfo.ChaosIDs {
			for node, chaoId := range chaosMap {
				log.Info(fmt.Sprintf("Delete chaos %v %v", node, chaoId))
				if chaoId != nil {
					err := clusterInfo.NodeWorkloadMap[node].ChaosClient().Delete(chaoId)
					if err != nil {
						log.Warn("delete chaos failed,error info", zap.Error(err))
					}
				}
			}
		}
	}
}

func ImportData(ctx context.Context, clusterInfo *DrClusterInfo) {
	tiupCluster := clusterInfo.Cluster
	pdURL := Try(tiupCluster.ServiceURL(resource.PDAddr)).(*url.URL)
	db := Try(tiupCluster.GetDB("")).(*sql.DB)
	defer db.Close()
	tidbClient := tidb.NewClient(db)

	s3Parms := oltppkg.S3Parms{S3Endpoint: "http://minio.pingcap.net:9000",
		S3AccessKeyID: "minioadmin", S3SecretAccessKey: "minioadmin"}
	brcmd := oltppkg.NewBr("/bin", "restore full", s3Parms.S3AccessKeyID, s3Parms.S3SecretAccessKey)
	brTpccCmd := brcmd.Pd(pdURL.String()).Storage("s3://benchmark/sysbench-32-10000000").
		S3endpoint(s3Parms.S3Endpoint).CheckReq(false).Build()
	exist := Try(tidbClient.IsTableExist(ctx, "test", "sbtest1")).(bool)
	if !exist {
		Try(oltppkg.ExecWorkloadCmd(ctx, clusterInfo.NodeWorkloadMap["br"], brTpccCmd, true, nil))
	}
}

func (c *DrClusterInfo) GetOnlineRecoveryResult(ctx context.Context, interval time.Duration, timeout time.Duration) error {
	failedInfo := "unsafe recovery failed"
	successInfo := "unsafe recovery finished"
	timeOut := time.After(timeout)
	for {
		select {
		case <-ctx.Done():
			log.Info("ctx done.")
			return nil
		case <-timeOut:
			return errors.New("online recovery timeout.")
		default:
		}
		oltppkg.WaitFor(ctx, "wait for getting online recovery result", interval)
		pdCtlCmd := "unsafe remove-failed-stores show"
		pdUrl := fmt.Sprintf("http://%s-peer.%s:2379", "pd3", c.Cluster.TestBed)
		cmd := fmt.Sprintf("tiup ctl:%s pd -u %s %s", c.Version, pdUrl, pdCtlCmd)
		log.Info("will run cmd", zap.String("cmd:", cmd))
		stdout, _, _, err := c.TiupNode.Exec(resource.WorkloadNodeExecOptions{Command: cmd})
		log.Info("cmd output", zap.String("cmd output:", stdout))
		Expect(err).Should(BeNil())
		bytes := []byte(stdout)
		recoveryInfo := gjson.GetBytes(bytes, "#.info").Value()
		var infoStr string
		for _, info := range recoveryInfo.([]interface{}) {
			infoStr = strings.ToLower(info.(string))
			if strings.Contains(infoStr, successInfo) || strings.Contains(infoStr, failedInfo) {
				break
			}
		}
		if strings.Contains(infoStr, successInfo) {
			return nil
		} else if strings.Contains(infoStr, failedInfo) {
			return errors.New(infoStr)
		}
	}
}

func (c *DrClusterInfo) ResetToMinResolvedTs(ctx context.Context) {
	oltppkg.LogAndBy("Step: Reset to MinResolvedTs")
	pdUrl := fmt.Sprintf("http://%s-peer.%s:2379", "pd3", c.Cluster.TestBed)
	cmd := fmt.Sprintf("tiup ctl:%s pd -u %s %s", c.Version, pdUrl, "min-resolved-ts")
	log.Info("will run cmd", zap.String("cmd:", cmd))
	stdout, _, _, err := c.TiupNode.Exec(resource.WorkloadNodeExecOptions{Command: cmd})
	log.Info("cmd output", zap.String("cmd output:", stdout))
	Expect(err).Should(BeNil())
	bytes := []byte(stdout)
	minResolvedTs := gjson.GetBytes(bytes, "min_resolved_ts").Value()
	minResolvedTsInt := minResolvedTs.(float64)
	log.Info("minResolvedTs", zap.Float64("minResolvedTsInt:", minResolvedTsInt))

	for _, peer := range c.DrTikvs {
		cmd := fmt.Sprintf("tiup ctl:%s tikv --host %s-peer:20160 reset-to-version -v %d", c.Version, peer, int64(minResolvedTsInt))
		log.Info("Will run cmd", zap.String("command", cmd))
		stdout, _, _, err := c.TiupNode.Exec(resource.WorkloadNodeExecOptions{Command: cmd})
		log.Info("cmd output", zap.String("cmd output:", stdout))
		Expect(err).Should(BeNil())
	}
}