package bench import ( "context" "database/sql" "encoding/json" "flag" "fmt" "net/url" "strings" "time" _ "github.com/go-sql-driver/mysql" . "github.com/onsi/ginkgo" . "github.com/onsi/gomega" host "github.com/pingcap/endless/pkg/workloadnode/tiup" "github.com/pingcap/endless/testcase/oltp/oltppkg" "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/matcher" "github.com/pingcap/endless/pkg/util" "github.com/pingcap/endless/pkg/workloadnode" "go.uber.org/zap" ) 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 } var _ = util.ParameterizedDescribe("perftest", func(flagSet *flag.FlagSet) { ctx := suiteTestCtx logger := ctx.Logger() var ( email string arch string benchType string tidbReplicas int tikvReplicas int tiflashReplicas int dsn string threads string apThreads string benchSubTypes string sysbenchOptions string ycsbOptions string versions string benchmarkID string tidbGlobals arrayFlags tidbConfigs arrayFlags tikvConfigs arrayFlags pdConfigs arrayFlags pdCtls arrayFlags storageURL string dbName string duration time.Duration warehouses int queries string pdUrls string tidbUrls string tikvUrls string tiflashUrls string jdbc string pluginURLPrefix string warmUP bool staging bool reserveEnv time.Duration resetEnv bool ignoreConfigCheck bool restartWorkload bool destroyOnFailure bool uploadClinicAll bool continuousProfiling bool enableTLS bool commands string commandDelay time.Duration bisect bool ) // misc flagSet.StringVar(&email, "email", "", "the email address for the case submitter") 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(&benchmarkID, "benchmark-id", "6000001", "benchmark id to save the result the workload result") flagSet.BoolVar(&uploadClinicAll, "upload-clinic", false, "upload clinic before releasing the testbed") flagSet.BoolVar(&continuousProfiling, "continuous-profiling", true, "enable continuous-profiling or not") // benchbot playground flagSet.BoolVar(&restartWorkload, "restart-workload", false, "to restart workload or not, for benchbot playground") flagSet.BoolVar(&destroyOnFailure, "destroy-on-failure", true, "destroy cluster when the case failed, for benchbot playground, set to false") // workload.common flagSet.StringVar(&dbName, "db-name", "test", "db name mysql connect to") flagSet.BoolVar(&warmUP, "warmup", true, "warmup is needed or not") flagSet.StringVar(&storageURL, "storage-url", "", "s3 storage url for the database") flagSet.BoolVar(&resetEnv, "reset-env", true, "if it's reset-env is true, cluster will be destroyed after each round of test, otherwise, the data will be kept for next round run, typically for AP workload to save time for data restoration and tiflash data sync") flagSet.StringVar(&benchType, "bench-type", "", "bench type, ycsb|sysbench|tpcc|benchmarksql|tpch|tpcds|job|bank|shenma|ch|apdogfood|corebanking|hzbank|booking|taobench") flagSet.StringVar(&benchSubTypes, "bench-sub-types", "", "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.DurationVar(&duration, "duration", 30*time.Minute, "duration to run the workload") // workload.extra command flagSet.DurationVar(&commandDelay, "command-delay", 10*time.Minute, "duration delay to run the commands") flagSet.StringVar(&commands, "commands", "", "commands to run at backend, to support ddl") // workload.ap flagSet.IntVar(&warehouses, "warehouses", 1000, "warehouses number for tpc-c workload") flagSet.StringVar(&apThreads, "ap-threads", "", "mutiple ap threads for HTAP workload, e.g. 1,2,4") flagSet.StringVar(&queries, "queries", "", "queries list for ap workload, tpch|tpc-ds|job") // workload.tp flagSet.StringVar(&jdbc, "jdbc", "", "jdbc connection settings") flagSet.StringVar(&threads, "threads", "", "mutiple threads supported, e.g. 100,200,400") flagSet.StringVar(&sysbenchOptions, "sysbench-options", "--tables=32 --table-size=10000000 --mysql-ignore-errors=1062,2013,8028,9007 --auto-inc=false", "options for sysbench, like, --tables, --table-size, -mysql-ignore-errors") flagSet.StringVar(&ycsbOptions, "ycsb-options", "-p recordcount=100000000 -p operationcount=50000000 -p workload=core -p fieldcount=10 -p fieldlength=100 -p requestdistribution=zipfian", "options for ycsb, list of -p key=value") // cluster.global flagSet.BoolVar(&staging, "staging", false, "use staging mirror for pre-release versions") flagSet.StringVar(&versions, "versions", "", "versions to run the workload, e.g. v5.2.2,v5.2.1") flagSet.StringVar(&arch, "arch", "amd64", "arch type, by default is amd64, specify arm64 for arm machine") flagSet.StringVar(&dsn, "dsn", "root@tcp(127.0.0.1:3306)", "target dsn to save the db") flagSet.BoolVar(&enableTLS, "enable-tls", false, "enable global tls or not, by default it's false") flagSet.BoolVar(&ignoreConfigCheck, "ignore-config-check", false, "to skip the correctness check of the configuration of tidb/tikv/pd, by default, it's false") flagSet.BoolVar(&bisect, "bisect", false, "update the monitor url, by default is false, all the bench on the same testbed using the monitor_url, if true, only the latest bench is updated") // cluster.tidb flagSet.IntVar(&tidbReplicas, "tidb-replicas", 0, "tidb replicas in the topo") flagSet.Var(&tidbConfigs, "tidb-configs", "tib config in yaml format") flagSet.Var(&tidbGlobals, "tidb-globals", "set global variables before play the workload") flagSet.StringVar(&tidbUrls, "tidb-urls", "", "tidb urls to patch, e.g. http://fileserver.pingcap.net/download/builds/pingcap/tidb/${commit-hash}/centos7/tidb-server.tar.gz") flagSet.StringVar(&pluginURLPrefix, "plugin-url-prefix", "", "specify the http url prefix path for the plugins, currently support whitelist-1.so and audit-1.so, e.g. http://fileserver.pingcap.net/download/builds/pingcap/tidb-plugins/v4.0.14-20220303/centos7/") // cluster.tikv flagSet.IntVar(&tikvReplicas, "tikv-replicas", 0, "tikv replicas in the topo") flagSet.Var(&tikvConfigs, "tikv-configs", "tikv config in yaml format") flagSet.StringVar(&tikvUrls, "tikv-urls", "", "tikv urls to patch, e.g. http://fileserver.pingcap.net/download/builds/pingcap/tikv/${commit-hash}/centos7/tikv-server.tar.gz") // cluster.tiflash flagSet.IntVar(&tiflashReplicas, "tiflash-replicas", 0, "tiflash replicas in the topo") flagSet.StringVar(&tiflashUrls, "tiflash-urls", "", "tiflash urls to patch, e.g. http://fileserver.pingcap.net/download/builds/pingcap/tiflash/master/${commit-hash}/centos7/tiflash.tar.gz") // cluster.pd flagSet.Var(&pdConfigs, "pd-configs", "pd-config in yaml format") flagSet.StringVar(&pdUrls, "pd-urls", "", "pd url to patch, e.g. http://fileserver.pingcap.net/download/builds/pingcap/pd/${commit-hash}/centos7/pd-server.tar.gz") flagSet.Var(&pdCtls, "pd-ctls", "pd-ctls in json list") // 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") Context("perftest", func() { It("bench", func() { // Initialize parameters // Initialize variables controlNodeName := "tiup" // Node 的 hostname = $name-peer controlNode := ctx.Resource(controlNodeName).(resource.WorkloadNode) testbedName := controlNode.Handle().Testbed workload := ctx.Resource("toolset").(resource.WorkloadNode) if tidbReplicas == 0 { tidbReplicas = 3 } if tikvReplicas == 0 { tikvReplicas = 3 } if len(threads) == 0 { threads = "100" } if len(apThreads) == 0 { apThreads = "1" } if len(versions) == 0 { versions = "nightly" } if len(jdbc) == 0 { jdbc = "?serverTimezone=UTC&characterEncoding=utf-8&useServerPrepStmts=true&cachePrepStmts=true&useConfigs=maxPerformance&prepStmtCacheSize=1000&prepStmtCacheSqlLimit=2048&useSSL=false&rewriteBatchedStatements=true&zeroDateTimeBehavior=convertToNull" } versionList := strings.Split(versions, ",") benchSubTypeList := strings.Split(benchSubTypes, ",") threadList := strings.Split(threads, ",") apThreadList := strings.Split(apThreads, ",") var err error if len(tidbGlobals) == 0 { tidbGlobals = []string{""} } if len(tidbConfigs) == 0 { defaultConfig, _ := processConfig(emptyConfig, benchType, "tidb") tidbConfigs = []string{defaultConfig} } else { for i, e := range tidbConfigs { tidbConfigs[i], err = processConfig(e, benchType, "tidb") if err != nil { panic(err) } } } if len(tikvConfigs) == 0 { defaultConfig, _ := processConfig(emptyConfig, benchType, "tikv") tikvConfigs = []string{defaultConfig} } else { for i, e := range tikvConfigs { log.Info(e) tikvConfigs[i], err = processConfig(e, benchType, "tikv") if err != nil { panic(err) } } } if len(pdConfigs) == 0 { pdConfigs = []string{"{}"} } if len(pdCtls) == 0 { pdCtls = []string{""} } pdUrlList := strings.Split(pdUrls, ",") // the order of plugin version must be consistent with tidb url, otherwise, the cluster can not be started tidbUrlList := strings.Split(tidbUrls, ",") pluginURLPrefixList := strings.Split(pluginURLPrefix, ",") tikvUrlList := strings.Split(tikvUrls, ",") tiflashUrlList := strings.Split(tiflashUrls, ",") stabilityGap := 2 * time.Minute //pre-process default values if len(storageURL) == 0 || len(dbName) == 0 { storageURL, dbName = defaultDataSet(benchType) } if len(queries) == 0 { queries = defaultQueries(benchType) } jdbc = jdbcEscape(benchType, jdbc) firstTimeFlag := true startTime := fmt.Sprintf("%d", time.Now().Unix()) ng_port := "" tls := "false" promURL := &url.URL{Host: fmt.Sprintf("tiup-peer.%s:9090", testbedName), Scheme: "http"} for _, version := range versionList { for _, benchSubType := range benchSubTypeList { for _, thread := range threadList { for _, apThread := range apThreadList { for _, tidbGlobal := range tidbGlobals { for _, tidbConfig := range tidbConfigs { for _, tikvConfig := range tikvConfigs { for _, pdConfig := range pdConfigs { for _, pdCtl := range pdCtls { for _, pdUrl := range pdUrlList { for idx, tidbUrl := range tidbUrlList { for _, tikvUrl := range tikvUrlList { for _, tiflashUrl := range tiflashUrlList { benchEnv := fmt.Sprintf(`storage: %s, dbname: %s, version: %s, benchSubType: %s, thread: %s, apThread: %s, tidbGlobal: %s, tidbConfig: %s tikvConfig: %s, pdConfig: %s, pdCtl: %s, pdUrl: %s, tidbUrl: %s, tikvUrl: %s, tiflashUrl: %s`, storageURL, dbName, version, benchSubType, thread, apThread, tidbGlobal, tidbConfig, tikvConfig, pdConfig, pdCtl, pdUrl, tidbUrl, tikvUrl, tiflashUrl) log.Info(benchEnv) ctx2, cancel := context.WithTimeout(ctx, 7*24*time.Hour) defer cancel() setStr := "" if staging { setStr = fmt.Sprintf("tiup mirror set %s", stagingMirror) } else { setStr = fmt.Sprintf("tiup mirror set %s", productionMirror) } setCmd := workloadnode.CommandContext(ctx2, setStr) Try(setCmd.Run(controlNode)) log.Info("set mirror done") // disable telemetry to avoid dns error telemetryCmd := workloadnode.CommandContext(ctx2, "tiup telemetry disable") Try(telemetryCmd.Run(controlNode)) log.Info("telemetry disabled") topoStr := renderTopo(benchType, tidbReplicas, tikvReplicas, tiflashReplicas) // enable ng_port when version == "nightly" || version >= "v5.3.0" if version == "nightly" || version >= "v5.3.0" { ng_port = " ng_port: 12020" } if enableTLS { tls = "true" } deployTopo := fmt.Sprintf(tiupTopoTpl, arch, tls, pdConfig, tidbConfig, tikvConfig, topoStr, ng_port) log.Info(deployTopo) tc := host.NewTiUPCluster(controlNode, controlNodeName+"-peer", deployTopo, clusterName).WithCMDTimeout(1 * time.Hour) err1 := deployCluster(tc, version, ignoreConfigCheck) if err1 != nil { log.Error("Deploy BackOff stopped retrying with Error: ", zap.Error(err1)) continue } log.Info("Cluster deployed") // handle plugin ok, load, dir := checkPlugin(tidbConfig) if ok { log.Info(load) log.Info(dir) ok2, cmd := handlePlugins(load, dir, pluginURLPrefixList[idx], tidbReplicas) if ok2 { ctlCmd := workloadnode.CommandContext(ctx2, cmd) Try(ctlCmd.Run(controlNode)) log.Info("setup plugins") } } // if patch url is provided, if something wrong with the provided patch build, will skip this round defaultTimeout := 20 * time.Minute var err error = nil if len(pdUrl) > 0 { err = tc.Patch(pdUrl, defaultTimeout, "-R", "pd", "--offline", "--format", "json") } if len(tidbUrl) > 0 { err = tc.Patch(tidbUrl, defaultTimeout, "-R", "tidb", "--offline", "--format", "json") } if len(tikvUrl) > 0 { err = tc.Patch(tikvUrl, defaultTimeout, "-R", "tikv", "--offline", "--format", "json") } if len(tiflashUrl) > 0 { err = tc.Patch(tiflashUrl, defaultTimeout, "-R", "tiflash", "--offline", "--format", "json") } if err != nil { // destroy the cluster and retain the promtheus Try(tc.Destroy("--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force")) continue } log.Info("Cluster patched") Try(tc.Start("--format", "json", "-c", "2")) log.Info("Cluster started") // wait for all the tikv to be ready, to avoid missing peers oltppkg.WaitForTiKVUP(tc, 3*time.Minute) oltppkg.WaitForTikvCount(tc, tikvReplicas+tiflashReplicas, 3*time.Minute) // enable continuous-profiling, not available on arm64, version >= 5.3.0 required if (version == "nightly" || version >= "v5.3.0") && continuousProfiling && arch != "arm64" { ctlCmd := workloadnode.CommandContext(ctx2, enableProfilingCmdTpl) ctlCmd.Run(controlNode) log.Info("Continuous Profiling started") } loadBalancer := ctx.Resource("lb").(resource.LoadBalancer) dbHost, port := loadBalancer.Endpoint() dbPort := fmt.Sprint(port) pdURL := Try(tc.ServiceURL(resource.PDAddr)).(*url.URL) //br restore if resetEnv || firstTimeFlag { s := fmt.Sprintf(brCmdTpl, s3AccessKeyID, s3SecretAccessKey, version, dbName, pdURL.Host, storageURL, s3Endpoint) if benchType == "shenma" || benchType == "apdogfood" || benchType == "corebanking" { s = fmt.Sprintf(brFullCmdTpl, s3AccessKeyID, s3SecretAccessKey, version, pdURL.Host, storageURL, s3Endpoint) } // log.Info(s) brCmd := workloadnode.CommandContext(ctx2, s) Try(brCmd.Run(controlNode)) // wait for 10 Minute for the Region to be 100% health, only without TiFlash if tiflashReplicas == 0 { oltppkg.WaitForRegionHealth(tc, 10*time.Minute) } logger.Info("the cluster and data is ready.") firstTimeFlag = false } // truncate tables if it's oltp_insert && auto-inc=false, and disable warmup to avoid duplcatekeys errors if benchSubType == "oltp_insert" && strings.Contains(sysbenchOptions, "auto-inc=false") { s2 := fmt.Sprintf(truncateSysbenchTables, dbHost, dbPort, dbName, dbHost, dbPort, dbName) mysqlCmd := workloadnode.CommandContext(ctx2, s2) Try(mysqlCmd.Run(workload)) } // handle pd ctls if len(pdCtl) > 0 { var arr []string _ = json.Unmarshal([]byte(pdCtl), &arr) for _, ctl := range arr { ctlStr := fmt.Sprintf(pdCtlCmdTpl, version, pdURL.Host, ctl) ctlCmd := workloadnode.CommandContext(ctx2, ctlStr) Try(ctlCmd.Run(controlNode)) } } // wait for the TiFlash Replica to be synced if tiflashReplicas > 0 { // limit max tiflash replicas to 2, tips from Lilinghai tmpTiflashReplicas := tiflashReplicas if tmpTiflashReplicas > 2 { tmpTiflashReplicas = 2 } tableList := getTiFlashTableList(benchType, tmpTiflashReplicas) s2 := fmt.Sprintf(mysqlCmdTpl2, dbHost, dbPort, dbName, tableList) mysqlCmd := workloadnode.CommandContext(ctx2, s2) Try(mysqlCmd.Run(workload)) queryCount := fmt.Sprintf("SELECT count(*) FROM information_schema.tiflash_replica WHERE TABLE_SCHEMA = '%s' and (AVAILABLE!=1 or progress!=1)", dbName) db, err := tc.GetDB(dbName) Expect(err).Should(BeNil()) count := 1 for { row := db.QueryRow(queryCount) err = row.Scan(&count) Expect(err).Should(BeNil()) if count == 0 { break } log.Info("Waiting for TiFlash ready...sleep one minute") time.Sleep(time.Minute) } } // for bank workload, need warmup before gather stats to avoid out-of-range predicate if benchType == "bank" { s2 := fmt.Sprintf(mysqlCmdTpl, dbHost, dbPort, dbName, bankResetCreateTimeSQL) mysqlCmd := workloadnode.CommandContext(ctx2, s2) Try(mysqlCmd.Run(workload)) warmupStr := fmt.Sprintf(bankCmdTpl, dbHost, dbPort, dbName, thread, 120) runCmd := workloadnode.CommandContext(ctx2, warmupStr) err = runCmd.Run(workload) if err != nil { tailLog(benchType, workload, ctx2) // destroy the cluster and retain the promtheus if destroyOnFailure { Try(tc.Destroy("--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force")) continue } return } } // for hzbank workload, need to create 3 sequences before workload if benchType == "hzbank" { s2 := fmt.Sprintf(mysqlCmdTpl, dbHost, dbPort, dbName, hzbankPrepQuery) mysqlCmd := workloadnode.CommandContext(ctx2, s2) Try(mysqlCmd.Run(workload)) } // prepared the global setting if len(tidbGlobal) > 0 { s1 := fmt.Sprintf(mysqlCmdTpl, dbHost, dbPort, dbName, tidbGlobal) mysqlCmd := workloadnode.CommandContext(ctx2, s1) mysqlCmd.Run(workload) } // gather statstics tableList := getAnalyzeList(benchType) if len(tableList) > 0 { s2 := fmt.Sprintf(mysqlCmdTpl2, dbHost, dbPort, dbName, tableList) mysqlCmd := workloadnode.CommandContext(ctx2, s2) Try(mysqlCmd.Run(workload)) } //handle reserveEnv if reserveEnv > 0*time.Minute { cmd := getBenchCmd(benchType, benchSubType, dbHost, dbPort, dbName, thread, apThread, duration, warehouses, sysbenchOptions, ycsbOptions, queries, tiflashReplicas, jdbc) log.Info(fmt.Sprintf("login toolset pod and run bench: %s", cmd)) time.Sleep(reserveEnv) return } // handle warmup if warmUP { cmd := getWarmupCmd(benchType, benchSubType, dbHost, dbPort, dbName, thread, apThread, warmupDuration, warehouses, sysbenchOptions, ycsbOptions, jdbc) if len(cmd) > 0 { warmUPCmd := workloadnode.CommandContext(ctx2, cmd) err = warmUPCmd.Run(workload) if err != nil { tailLog(benchType, workload, ctx2) // destroy the cluster and retain the promtheus Try(tc.Destroy("--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force", "--format", "json")) continue } } } // truncate tables if it's oltp_insert && auto-inc=false, and disable warmup to avoid duplcatekeys errors if benchSubType == "oltp_insert" && strings.Contains(sysbenchOptions, "auto-inc=false") { s2 := fmt.Sprintf(truncateSysbenchTables, dbHost, dbPort, dbName, dbHost, dbPort, dbName) mysqlCmd := workloadnode.CommandContext(ctx2, s2) Try(mysqlCmd.Run(workload)) } // Get the testbed name testbedCmdStr := fmt.Sprintf(testbedCmpTpl, testbedName, benchType) // log.Info(testbedCmdStr) testbedCmd := workloadnode.CommandContext(ctx2, testbedCmdStr) Try(testbedCmd.Run(workload)) // Get the commit info commitCmdStr := fmt.Sprintf(commitCmpTpl, dbHost, dbPort, benchType) // log.Info(commitCmdStr) commitCmd := workloadnode.CommandContext(ctx2, commitCmdStr) Try(commitCmd.Run(workload)) // Run the workload. beginAt := time.Now().Unix() cmd := getBenchCmd(benchType, benchSubType, dbHost, dbPort, dbName, thread, apThread, duration, warehouses, sysbenchOptions, ycsbOptions, queries, tiflashReplicas, jdbc) benchmarkCmd := workloadnode.CommandContext(ctx2, cmd) go runBackendCommands(dbHost, dbPort, dbName, commands, commandDelay, workload, ctx2) err = benchmarkCmd.Run(workload) tailLog(benchType, workload, ctx2) if !destroyOnFailure { log.Info("Return after the first bench due to destroyOnFailure is false.") return } endAt := time.Now().Unix() // skip the result where there error of the benchmark client and the bench duration < duration - 1 minutes if err != nil && endAt-beginAt < int64(duration.Seconds())-60 { // destroy the cluster and retain the promtheus Try(tc.Destroy("--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force", "--format", "json")) continue } if endAt-beginAt > int64((1 * time.Hour).Seconds()) { stabilityGap = 2 * stabilityGap } endTime := time.Now().Add(-stabilityGap) // remove the prefix/postfix stabilityGap data seconds := endAt - beginAt - 2*int64(stabilityGap.Seconds()) if seconds <= 0 { endTime = time.Now() seconds = endAt - beginAt + 60 } // parse the bench result into dsn by cli script if benchType == "tpch" || benchType == "tpcds" || benchType == "job" { 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, ctx2) //parse the result cliCmd := workloadnode.CommandContext(ctx2, s3) Try(cliCmd.Run(workload)) } //gather metrics from promtheus and persist perf result into target db logic if !(benchType == "tpch" || benchType == "tpcds" || benchType == "job") { db, err := sql.Open("mysql", dsn) if err != nil { panic(err) } defer db.Close() //avg avgQps, avgTps, avgP90, avgP99, avgP999, avgLatency, maxLatency, minQps, minTps, minP90, minP99, minP999, maxQps, maxTps, maxP90, maxP99, maxP999, idle_80_not_in_txn, idle_80_in_txn, parse_80, compile_80, execute_80, tidb_prewrite_80, tidb_kv_get_80, tidb_coprocessor_80, tidb_cpu_avg, tidb_cpu_delta_avg, plan_cache_ops, pd_tso_80, tikv_cpu_avg, tikv_cpu_delta_avg, tikv_coprocessor_80, tikv_kv_get_80, tikv_prewrite_80, apply_80, append_80, commit_80, idle_99_not_in_txn, idle_99_in_txn, parse_99, compile_99, execute_99, tidb_prewrite_99, tidb_kv_get_99, tidb_coprocessor_99, pd_tso_99, tikv_coprocessor_99, tikv_kv_get_99, tikv_prewrite_99, apply_99, append_99, commit_99, p80_latency, qps_stddev, p80_latency_stddev, p99_latency_stddev, p999_latency_stddev, idle_80_not_in_txn_stddev, idle_80_in_txn_stddev, parse_80_stddev, compile_80_stddev, execute_80_stddev, tidb_prewrite_80_stddev, tidb_kv_get_80_stddev, tidb_coprocessor_80_stddev, pd_tso_80_stddev, tikv_coprocessor_80_stddev, tikv_kv_get_80_stddev, tikv_prewrite_80_stddev, apply_80_stddev, append_80_stddev, commit_80_stddev, idle_99_not_in_txn_stddev, idle_99_in_txn_stddev, parse_99_stddev, compile_99_stddev, execute_99_stddev, tidb_prewrite_99_stddev, tidb_kv_get_99_stddev, tidb_coprocessor_99_stddev, pd_tso_99_stddev, tikv_coprocessor_99_stddev, tikv_kv_get_99_stddev, tikv_prewrite_99_stddev, apply_99_stddev, append_99_stddev, commit_99_stddev, idle_avg_not_in_txn, idle_avg_in_txn, parse_avg, compile_avg, execute_avg, tidb_prewrite_avg, tidb_kv_get_avg, tidb_coprocessor_avg, pd_tso_avg, tikv_coprocessor_avg, tikv_kv_get_avg, tikv_prewrite_avg, apply_avg, append_avg, commit_avg, idle_avg_not_in_txn_stddev, idle_avg_in_txn_stddev, parse_avg_stddev, compile_avg_stddev, execute_avg_stddev, tidb_prewrite_avg_stddev, tidb_kv_get_avg_stddev, tidb_coprocessor_avg_stddev, pd_tso_avg_stddev, tikv_coprocessor_avg_stddev, tikv_kv_get_avg_stddev, tikv_prewrite_avg_stddev, apply_avg_stddev, append_avg_stddev, commit_avg_stddev := getBenchResult(promURL, seconds, endTime) // gather version and git commit hash db2, err := sql.Open("mysql", fmt.Sprintf(defaultDSN, dbHost, dbPort)) if err != nil { panic(err) } defer db2.Close() rows, err := db2.Query(commitQuery) if err != nil { panic(err) } versionMap := map[string]string{"tidb": "", "tikv": "", "pd": "", "tiflash": ""} gitHashMap := map[string]string{"tidb": "", "tikv": "", "pd": "", "tiflash": ""} for rows.Next() { var r row err = rows.Scan(&r.serverType, &r.version, &r.gitHash) if err != nil { panic(err) } versionMap[r.serverType] = r.version gitHashMap[r.serverType] = r.gitHash } // insert the oltp_summary_logs record if benchType != "apdogfood" { _, err = db.Exec(oltpInsert, benchEnv, email, benchType, benchSubType, thread, versionMap["tidb"], gitHashMap["tidb"], versionMap["tikv"], gitHashMap["tikv"], versionMap["pd"], gitHashMap["pd"], versionMap["tiflash"], gitHashMap["tiflash"], avgQps, avgTps, avgP90, avgP99, avgP999, avgLatency, maxLatency, minQps, minTps, minP90, minP99, minP999, maxQps, maxTps, maxP90, maxP99, maxP999, beginAt, endAt, endAt-beginAt, testbedName, idle_80_not_in_txn, idle_80_in_txn, parse_80, compile_80, execute_80, tidb_prewrite_80, tidb_kv_get_80, tidb_coprocessor_80, tidb_cpu_avg, tidb_cpu_delta_avg, plan_cache_ops, pd_tso_80, tikv_cpu_avg, tikv_cpu_delta_avg, tikv_coprocessor_80, tikv_kv_get_80, tikv_prewrite_80, apply_80, append_80, commit_80, idle_99_not_in_txn, idle_99_in_txn, parse_99, compile_99, execute_99, tidb_prewrite_99, tidb_kv_get_99, tidb_coprocessor_99, pd_tso_99, tikv_coprocessor_99, tikv_kv_get_99, tikv_prewrite_99, apply_99, append_99, commit_99, p80_latency, qps_stddev, p80_latency_stddev, p99_latency_stddev, p999_latency_stddev, idle_80_not_in_txn_stddev, idle_80_in_txn_stddev, parse_80_stddev, compile_80_stddev, execute_80_stddev, tidb_prewrite_80_stddev, tidb_kv_get_80_stddev, tidb_coprocessor_80_stddev, pd_tso_80_stddev, tikv_coprocessor_80_stddev, tikv_kv_get_80_stddev, tikv_prewrite_80_stddev, apply_80_stddev, append_80_stddev, commit_80_stddev, idle_99_not_in_txn_stddev, idle_99_in_txn_stddev, parse_99_stddev, compile_99_stddev, execute_99_stddev, tidb_prewrite_99_stddev, tidb_kv_get_99_stddev, tidb_coprocessor_99_stddev, pd_tso_99_stddev, tikv_coprocessor_99_stddev, tikv_kv_get_99_stddev, tikv_prewrite_99_stddev, apply_99_stddev, append_99_stddev, commit_99_stddev, idle_avg_not_in_txn, idle_avg_in_txn, parse_avg, compile_avg, execute_avg, tidb_prewrite_avg, tidb_kv_get_avg, tidb_coprocessor_avg, pd_tso_avg, tikv_coprocessor_avg, tikv_kv_get_avg, tikv_prewrite_avg, apply_avg, append_avg, commit_avg, idle_avg_not_in_txn_stddev, idle_avg_in_txn_stddev, parse_avg_stddev, compile_avg_stddev, execute_avg_stddev, tidb_prewrite_avg_stddev, tidb_kv_get_avg_stddev, tidb_coprocessor_avg_stddev, pd_tso_avg_stddev, tikv_coprocessor_avg_stddev, tikv_kv_get_avg_stddev, tikv_prewrite_avg_stddev, apply_avg_stddev, append_avg_stddev, commit_avg_stddev, benchmarkID) if err != nil { log.Error("Save Oltp Result Error: ", zap.Error(err)) panic(err) } } // insert the olap_summary_logs record if benchType == "ch" || benchType == "apdogfood" { err := saveOlapResult(db, db2, benchEnv, email, benchType, benchSubType, versionMap["tidb"], gitHashMap["tidb"], versionMap["tikv"], gitHashMap["tikv"], versionMap["pd"], gitHashMap["pd"], versionMap["tiflash"], gitHashMap["tiflash"], beginAt, endAt, endAt-beginAt+1, testbedName, idle_80_not_in_txn, idle_80_in_txn, parse_80, compile_80, execute_80, tidb_prewrite_80, tidb_kv_get_80, tidb_coprocessor_80, tidb_cpu_avg, tidb_cpu_delta_avg, plan_cache_ops, pd_tso_80, tikv_cpu_avg, tikv_cpu_delta_avg, tikv_coprocessor_80, tikv_kv_get_80, tikv_prewrite_80, apply_80, append_80, commit_80, idle_99_not_in_txn, idle_99_in_txn, parse_99, compile_99, execute_99, tidb_prewrite_99, tidb_kv_get_99, tidb_coprocessor_99, pd_tso_99, tikv_coprocessor_99, tikv_kv_get_99, tikv_prewrite_99, apply_99, append_99, commit_99, idle_avg_not_in_txn, idle_avg_in_txn, parse_avg, compile_avg, execute_avg, tidb_prewrite_avg, tidb_kv_get_avg, tidb_coprocessor_avg, pd_tso_avg, tikv_coprocessor_avg, tikv_kv_get_avg, tikv_prewrite_avg, apply_avg, append_avg, commit_avg, benchmarkID) if err != nil { log.Error("Save Olap Result Error: ", zap.Error(err)) } } } // upload logs and configs to clinic uploadClinicWrapper(tc, benchType, dsn, testbedName, startTime, arch, false, bisect) if version >= "v6.0.0" || version == "nightly" { uploadProfile(workload, ctx2, version, benchType, benchSubType, thread, dsn, testbedName, beginAt, endAt) } // destroy the cluster and retain the promtheus if resetEnv { Try(tc.Destroy("--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force", "--format", "json")) } else { Try(tc.Destroy("--retain-role-data", "tidb", "--retain-role-data", "tikv", "--retain-role-data", "pd", "--retain-role-data", "tiflash", "--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force", "--format", "json")) } } } } } } } } } } } } } } // need to redeploy and start the cluster, just to upload the metrics to clinic version := versionList[len(versionList)-1] tidbConfig, tikvConfig, pdConfig, setStr := "", "", "", "" ctx2, cancel := context.WithTimeout(ctx, 7*24*time.Hour) defer cancel() if staging { setStr = fmt.Sprintf("tiup mirror set %s", stagingMirror) } else { setStr = fmt.Sprintf("tiup mirror set %s", productionMirror) } setCmd := workloadnode.CommandContext(ctx2, setStr) Try(setCmd.Run(controlNode)) log.Info("set mirror done") // disable telemetry to avoid dns error telemetryCmd := workloadnode.CommandContext(ctx2, "tiup telemetry disable") Try(telemetryCmd.Run(controlNode)) log.Info("telemetry disabled") topoStr := renderTopo(benchType, tidbReplicas, tikvReplicas, tiflashReplicas) deployTopo := fmt.Sprintf(tiupTopoTpl, arch, tls, pdConfig, tidbConfig, tikvConfig, topoStr, ng_port) log.Info(deployTopo) tc := host.NewTiUPCluster(controlNode, controlNodeName+"-peer", deployTopo, clusterName) err1 := deployCluster(tc, version, ignoreConfigCheck) if err1 != nil { log.Error("Deploy BackOff stopped retrying with Error: ", zap.Error(err1)) return } log.Info("Cluster deployed") Try(tc.Start("--format", "json", "-c", "2")) log.Info("Cluster started") // make sure promtheus is ready promReadyCheck(promURL.String()) // upload logs and monitor to clinic uploadClinicWrapper(tc, benchType, dsn, testbedName, startTime, arch, true, bisect) Try(tc.Destroy("--retain-role-data", "prometheus", "--retain-role-data", "grafana", "--force", "--format", "json")) }) //injection It("injection", func() { // Initialize parameters // Initialize variables ctx2, _ := context.WithTimeout(ctx, 7*24*time.Hour) controlNodeName := "tiup" // Node 的 hostname = $name-peer controlNode := ctx.Resource(controlNodeName).(resource.WorkloadNode) //testbedName := controlNode.Handle().Testbed workload := ctx.Resource("toolset").(resource.WorkloadNode) tc := host.NewTiUPCluster(controlNode, controlNodeName+"-peer", "", clusterName) loadBalancer := ctx.Resource("lb").(resource.LoadBalancer) dbHost, port := loadBalancer.Endpoint() dbPort := fmt.Sprint(port) pdURL := Try(tc.ServiceURL(resource.PDAddr)).(*url.URL) // if patch url is provided, if something wrong with the provided patch build, will skip this round defaultTimeout := 20 * time.Minute if len(tidbGlobals) == 0 { tidbGlobals = []string{""} } if len(tidbConfigs) == 0 { tidbConfigs = []string{""} } if len(tikvConfigs) == 0 { tikvConfigs = []string{""} } if len(pdConfigs) == 0 { pdConfigs = []string{""} } if len(pdCtls) == 0 { pdCtls = []string{""} } //pre-process default values if len(threads) == 0 { threads = "100" } if len(apThreads) == 0 { apThreads = "1" } if len(jdbc) == 0 { jdbc = "?serverTimezone=UTC&characterEncoding=utf-8&useServerPrepStmts=true&cachePrepStmts=true&useConfigs=maxPerformance&prepStmtCacheSize=1000&prepStmtCacheSqlLimit=2048&useSSL=false&rewriteBatchedStatements=true&zeroDateTimeBehavior=convertToNull" } if len(storageURL) == 0 || len(dbName) == 0 { storageURL, dbName = defaultDataSet(benchType) } if len(queries) == 0 { queries = defaultQueries(benchType) } jdbc = jdbcEscape(benchType, jdbc) benchEnv := fmt.Sprintf(`storage: %s, dbname: %s, version: %s, benchtype: %s, benchSubType: %s, thread: %s, apThread: %s, tidbGlobal: %s, tidbConfig: %s tikvConfig: %s, pdConfig: %s, pdUrl: %s, tidbUrl: %s, tikvUrl: %s, tiflashUrl: %s, restart: %v, uploadClinic: %v`, storageURL, dbName, versions, benchType, benchSubTypes, threads, apThreads, tidbGlobals[0], tidbConfigs[0], tikvConfigs[0], pdConfigs[0], pdUrls, tidbUrls, tikvUrls, tiflashUrls, restartWorkload, uploadClinicAll) log.Info(benchEnv) // restart the workload if needed if restartWorkload { pkillCmd := workloadnode.CommandContext(ctx2, pkill) pkillCmd.Run(workload) } // patch the pd/tidb/tikv/tiflash, if the url is specified. if len(pdUrls) > 0 { _ = tc.Patch(pdUrls, defaultTimeout, "-R", "pd", "--format", "json") } if len(tidbUrls) > 0 { _ = tc.Patch(tidbUrls, defaultTimeout, "-R", "tidb", "--format", "json") } if len(tikvUrls) > 0 { _ = tc.Patch(tikvUrls, defaultTimeout, "-R", "tikv", "--format", "json") } if len(tiflashUrls) > 0 { _ = tc.Patch(tiflashUrls, defaultTimeout, "-R", "tiflash", "--format", "json") } // set mirror, prepare for the upgrade options setStr := "" if staging { setStr = fmt.Sprintf("tiup mirror set %s", stagingMirror) } else { setStr = fmt.Sprintf("tiup mirror set %s", productionMirror) } setCmd := workloadnode.CommandContext(ctx2, setStr) Try(setCmd.Run(controlNode)) log.Info("set mirror done") // upgrade to specified versions if len(versions) > 0 { Try(tc.Upgrade(versions)) } else { // set for pd ctls versions = "nightly" } // prepared the global setting if len(tidbGlobals) > 0 && len(tidbGlobals[0]) > 0 { s1 := fmt.Sprintf(mysqlCmdTpl, dbHost, dbPort, dbName, tidbGlobals[0]) mysqlCmd := workloadnode.CommandContext(ctx2, s1) Try(mysqlCmd.Run(workload)) } // handle pd/tidb/tikv config change var err error if len(pdConfigs) == 1 && len(pdConfigs[0]) > 0 { if config, e := YamlToToml(pdConfigs[0]); e == nil { err = tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{ Target: resource.PD, Config: config, Timeout: 5 * time.Minute, }) } } if len(tidbConfigs) == 1 && len(tidbConfigs[0]) > 0 { config, e := YamlToToml(tidbConfigs[0]) if e == nil { err = tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{ Target: resource.TiDB, Config: config, Timeout: 5 * time.Minute, }) } else { log.Error("transform tidb configs error: ", zap.Error(e)) } } if len(tikvConfigs) == 1 && len(tikvConfigs[0]) > 0 { config, e := YamlToToml(tikvConfigs[0]) if e == nil { err = tc.UpdateConfig(resource.TiDBClusterUpdateConfigOptions{ Target: resource.TiKV, Config: config, Timeout: 10 * time.Minute, }) // sleep 1 minute to wait for leader balance after tikv reloading time.Sleep(time.Minute) } else { log.Error("transform tikv configs error: ", zap.Error(e)) } } if err != nil { log.Error("update configs error: ", zap.Error(err)) return } // handle pd ctls if len(pdCtls) > 0 { var arr []string _ = json.Unmarshal([]byte(pdCtls[0]), &arr) for _, ctl := range arr { ctlStr := fmt.Sprintf(pdCtlCmdTpl, versions, pdURL.Host, ctl) ctlCmd := workloadnode.CommandContext(ctx2, ctlStr) Try(ctlCmd.Run(controlNode)) } } //TODO: support scale-out or scale-in // restart the workload if needed if restartWorkload { cmd := getBenchCmd(benchType, benchSubTypes, dbHost, dbPort, dbName, threads, apThreads, duration, warehouses, sysbenchOptions, ycsbOptions, queries, tiflashReplicas, jdbc) benchmarkCmd := workloadnode.CommandContext(ctx2, cmd) benchmarkCmd.Run(workload) } if uploadClinicAll { pkillCmd := workloadnode.CommandContext(ctx2, pkill) pkillCmd.Run(workload) db, err := sql.Open("mysql", fmt.Sprintf(defaultDSN, dbHost, dbPort)) if err != nil { panic(err) } defer db.Close() var createTime string row := db.QueryRow("select DATE_FORMAT(create_time, '%Y-%m-%d %T') from information_schema.tables order by create_time desc limit 1") err = row.Scan(&createTime) if err != nil { panic(err) } url := uploadClinic(tc, "-f", fmt.Sprintf(`"%s +0800"`, createTime)) log.Info(url) } }) }) })