package bench

import (
	"context"
	"flag"
	"fmt"
	"os"
	"strings"
	"time"

	_ "github.com/go-sql-driver/mysql"
	. "github.com/onsi/ginkgo"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"

	"github.com/pingcap/endless/pkg/workloadnode"

	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/pkg/util"
)

type arrayFlags []string

func (i *arrayFlags) String() string {
	return "my string representation"
}

func (i *arrayFlags) Set(value string) error {
	*i = append(*i, value)
	return nil
}

type row struct {
	serverType string
	version    string
	gitHash    string
}

const (
	warmupDuration = time.Minute * 10
	cli            = `cli/argo_parse.py process --file %s.output %s --benchmark-id %s --dsn '%s' --bench-sub-type '%s' --thread %s --start-time %d --end-time %d --email '%s' --bench-env '%s'`

	installCmdTpl  = "curl https://binaries.cockroachdb.com/cockroach-%s.linux-%s.tgz | tar -xz && sudo cp -f cockroach-%s.linux-%s/cockroach /usr/local/bin"
	startCmdTpl    = "cockroach start  --insecure --store=/crdb/cocknode --listen-addr=crdb%d-peer:60000 --http-addr=0.0.0.0:50000 --join=%s --background"
	initCmdTpl     = "cockroach init --insecure --host=crdb1-peer:60000"
	connTestCmdTpl = "cockroach sql  --insecure --host=lb.%s:60000 -e 'select * from pg_catalog.pg_constraint'"
	restoreCmdTpl  = "cockroach sql --insecure --execute=\"RESTORE database %s FROM %s \" --user=root --host=lb:60000 --database=%s"
	//restoreCmdTpl  = "cockroach sql --insecure --execute=\"RESTORE database test FROM '2022/02/24-103022.14' IN 's3://benchmark/sysbench-crdb-test?AWS_ACCESS_KEY_ID=minioadmin&AWS_SECRET_ACCESS_KEY=minioadmin&AWS_ENDPOINT=http://minio.pingcap.net:9000'\" --user=root --host=lb:60000 --database=test"
	sysbenchCmdTpl  = "sysbench %s run --pgsql-host=%s --pgsql-port=%s --pgsql-user=root --db-driver=pgsql --tables=%s --table-size=%s --pgsql-db=%s --threads=%s --time=%d --auto-inc=false --report-interval=10 %s >sysbench.output"
	//destroyCmdTpl   = "ps -ef | grep cockroach| grep -v grep | awk '{print $2}'|xargs kill -9 |rm -rf /crdb/cocknode"
	destroyCmdTpl = "cockroach sql --insecure --host=crdb1-peer:60000 -e 'drop database test CASCADE;'"
	profileCmdTpl   = "go tool pprof --text http://crdb1-peer:50000/debug/pprof/profile?seconds=10"
	cpprofileCmdTpl = "cd /bin;./mc alias set qa-minio http://minio.pingcap.net:9000 minioadmin minioadmin --api S3v4;./mc cp /root/pprof/pprof.* qa-minio/crdb-profile/%s/%s_%s.pb.gz;rm -rf /root/pprof"
)

var _ = util.ParameterizedDescribe("bench", func(flagSet *flag.FlagSet) {
	ctx := suiteTestCtx
	var (
		email           string
		dsn             string
		arch            string
		benchType       string
		replicas        int
		threads         string
		benchSubTypes   string
		ignoreErrors    string
		versions        string
		benchmarkID     string
		storageURL      string
		dbName          string
		duration        time.Duration
		warehouses      int
		jdbc            string
		warmUP          bool
		reserveEnv      time.Duration
		tables          string
		tablesize       string
		sysbenchOptions string
	)
	flagSet.StringVar(&arch, "arch", "amd64", "arch type, by default is amd64, specify arm64 for arm machine")
	flagSet.StringVar(&benchType, "bench-type", "tpcc", "bench type, ycsb|sysbench|tpcc|benchmarksql|tpch|tpcds|job|bank|shenma|ch|apdogfood|corebanking|hzbank")
	flagSet.IntVar(&replicas, "replicas", 3, "crdb replicas in the cluster")
	flagSet.StringVar(&threads, "threads", "100", "mutiple threads supported, e.g. 100,200,400")
	flagSet.StringVar(&benchSubTypes, "bench-sub-types", "benchmark", "bench sub type for sysbench and ycsb usage, sysbench: oltp_read_write,oltp_point_select,oltp_update_non_index,oltp_update_index,oltp_insert,select_random_points,select_random_ranges,oltp_read_only,oltp_delete,bulk_insert, ycsb: workloada,workloadb,workloadc,workloadd,workloade")
	flagSet.StringVar(&ignoreErrors, "mysql-ignore-errors", "", "mysql ignore errors for sysbench, like, -mysql-ignore-errors 2013")
	flagSet.StringVar(&versions, "versions", "nightly", "versions to run the workload, e.g. v5.2.2,v5.2.1")
	flagSet.StringVar(&benchmarkID, "benchmark-id", "300001", "benchmark id to save the result the workload result")
	flagSet.StringVar(&storageURL, "storage-url", "'2022/03/14-144252.25' IN 's3://benchmark/sysbench-crdb-32-10000000-auto?AWS_ACCESS_KEY_ID=minioadmin&AWS_SECRET_ACCESS_KEY=minioadmin&AWS_ENDPOINT=http://minio.pingcap.net:9000'", "s3 storage url for the database")
	flagSet.StringVar(&dbName, "db-name", "test", "db name pgsql connect to")
	flagSet.DurationVar(&duration, "duration", 10*time.Minute, "duration to run the workload")
	flagSet.IntVar(&warehouses, "warehouses", 1000, "warehouses number for tpc-c workload")
	flagSet.StringVar(&jdbc, "jdbc", "?serverTimezone=UTC&characterEncoding=utf-8&useServerPrepStmts=true&cachePrepStmts=true&useConfig=maxPerformance&prepStmtCacheSize=1000&prepStmtCacheSqlLimit=2048&useSSL=false&rewriteBatchedStatements=true&zeroDateTimeBehavior=convertToNull", "jdbc connection settings")
	flagSet.BoolVar(&warmUP, "warmup", true, "warmup is needed or not")
	flagSet.DurationVar(&reserveEnv, "reserve-env", 0*time.Minute, "how long to reserve a env, if reserve-env > 0, no benchmark will be executed")
	flagSet.StringVar(&dsn, "dsn", "root@tcp(127.0.0.1:3306)", "target dsn to save the db")
	flagSet.StringVar(&email, "email", "", "the email address for the case submitter")
	flagSet.StringVar(&sysbenchOptions, "sysbench-options", "", "extra options for sysbench, like, -auto-inc=false")
	flagSet.StringVar(&tables, "tables", "32", "count of tables default 32")
	flagSet.StringVar(&tablesize, "table-size", "10000000", "table size default 10000000")
	Context("bench", func() {
		It("bench", func() {
			// Initialize parameters
			// S3 parameters
			//s3Endpoint := ctx.Param("s3_endpoint", "http://minio.pingcap.net:9000")
			//s3AccessKeyID := ctx.Param("s3_access_key_id", "minioadmin")
			//s3SecretAccessKey := ctx.Param("s3_secret_access_key", "minioadmin")
			// Initialize variables
			// tc := ctx.Resource("tc").(resource.TiDBCluster) // NOTE: deploy cluster, use tiup Implementation
			versionList := strings.Split(versions, ",")
			benchSubTypeList := strings.Split(benchSubTypes, ",")
			threadList := strings.Split(threads, ",")

			workload := ctx.Resource("toolset").(resource.WorkloadNode)
			testbedName := workload.Handle().Testbed

			crdbList := [](resource.WorkloadNode){}
			for i := 1; i <= replicas; i++ {
				crdbList = append(crdbList, ctx.Resource(fmt.Sprintf("crdb%d", i)).(resource.WorkloadNode))
			}
			joinList := []string{}
			for i := 1; i <= replicas; i++ {
				joinList = append(joinList, fmt.Sprintf("crdb%d-peer:60000", i))
			}
			joinStr := strings.Join(joinList, ",")
			//stabilityGap := 3 * time.Minute
			//pre-process default values
			//if len(storageURL) == 0 || len(dbName) == 0 {
			//	storageURL, dbName = defaultDataSet(benchType)
			//}

			if len(ignoreErrors) > 0 {
				ignoreErrors = fmt.Sprintf("--mysql-ignore-errors=%s", ignoreErrors)
			}
			loadBalancer := ctx.Resource("lb").(resource.LoadBalancer)
			dbHost, port := loadBalancer.Endpoint()
			dbPort := fmt.Sprint(port)
			//jdbc = jdbcEscape(benchType, jdbc)

			//startTime := fmt.Sprintf("%d", time.Now().Unix())
			for _, version := range versionList {
				for _, benchSubType := range benchSubTypeList {
					for _, thread := range threadList {
						benchEnv := fmt.Sprintf(`dbname: %s, version: %s, benchSubType: %s, thread: %s,benchmarkID: %s,testbedName: %s`,
							dbName, version, benchSubType, thread, benchmarkID, testbedName)
						log.Info(benchEnv)
						ctx, _ := context.WithTimeout(ctx, 7*24*time.Hour)

						// crdb installation and start backend
						for i, crdb := range crdbList {
							s1 := fmt.Sprintf(installCmdTpl, version, arch, version, arch)
							installCmd := workloadnode.CommandContext(ctx, s1)
							Try(installCmd.Run(crdb))
							s2 := fmt.Sprintf(startCmdTpl, i+1, joinStr)

							startCmd := workloadnode.CommandContext(ctx, s2)
							startCmd.Run(crdb)
						}

						// crdb init
						s := fmt.Sprintf(initCmdTpl)
						initCmd := workloadnode.CommandContext(ctx, s)
						initCmd.Run(crdbList[0])

						// connect to crdb
						s = fmt.Sprintf(connTestCmdTpl, testbedName)
						startCmd := workloadnode.CommandContext(ctx, s)
						Try(startCmd.Run(crdbList[0]))

						// restore database from minio
						s = fmt.Sprintf(restoreCmdTpl, dbName, storageURL, dbName)
						restoreCmd := workloadnode.CommandContext(ctx, s)
						Try(restoreCmd.Run(crdbList[0]))

						//profiling
						go Profile(workload, ctx)

						// run bench
						beginAt := time.Now().Unix()
						s = getBenchCmd(benchType, benchSubType, dbHost, dbPort, tables, tablesize, dbName, thread, duration, sysbenchOptions, storageURL)
						benchCmd := workloadnode.CommandContext(ctx, s)
						Try(benchCmd.Run(workload))

						// parse the bench result into dsn by cli script
						endAt := time.Now().Unix()
						s3 := ""
						if benchType == "sysbench" || benchType == "tpcc" {
							explainStr := ""
							s3 = fmt.Sprintf(cli, benchType, explainStr, benchmarkID, dsn, benchSubType, thread, beginAt, endAt, email, benchEnv)
						} else {
							explainStr := "--explain-file explain.txt"
							s3 = fmt.Sprintf(cli, benchType, explainStr, benchmarkID, dsn, benchSubType, thread, beginAt, endAt, email, benchEnv)

						}
						log.Info(s3)
						//tail the result log
						tailLog(benchType, workload, ctx)
						//parse the result
						cliCmd := workloadnode.CommandContext(ctx, s3)
						Try(cliCmd.Run(workload))

						//copy the profile to minio
						s = fmt.Sprintf(cpprofileCmdTpl, testbedName, benchSubType, thread)
						cpprofileCmd := workloadnode.CommandContext(ctx, s)
						Try(cpprofileCmd.Run(workload))

						//destroy the cluster
						//for i := 0; i < replicas; i++ {
							s = fmt.Sprintf(destroyCmdTpl)
							destroyCmd := workloadnode.CommandContext(ctx, s)
							Try(destroyCmd.Run(crdbList[0]))
							
						

					}
				}
			}
		})
	})
})

func getBenchCmd(benchType, benchSubType, dbHost, dbPort, tables, tablesize, dbName, thread string, duration time.Duration, sysbenchOptions, storageURL string) string {
	var cmd string
	switch {
	case benchType == "sysbench":
		cmd = fmt.Sprintf(sysbenchCmdTpl, benchSubType, dbHost, dbPort, tables, tablesize, dbName, thread, int(duration.Seconds()), sysbenchOptions)
	}
	return cmd
}
func tailLog(benchType string, workload resource.WorkloadNode, ctx context.Context) {
	tailCmd := workloadnode.CommandContext(ctx, fmt.Sprintf("tail -100 %s.output", benchType))
	tailCmd.Stdout = os.Stdout
	tailCmd.Run(workload)
}
func Profile(workload resource.WorkloadNode, ctx context.Context) {
	time.Sleep(time.Duration(5) * time.Minute)
	proCmd := workloadnode.CommandContext(ctx, fmt.Sprintf(profileCmdTpl))
	proCmd.Run(workload)
}