package bench import ( "database/sql" "flag" "fmt" "net/url" "strings" "time" . "github.com/onsi/ginkgo" "go.uber.org/zap" . "github.com/pingcap/endless/pkg/matcher" . "github.com/pingcap/endless/pkg/util" "github.com/pingcap/endless/pkg/workloadnode" "github.com/pingcap/log" "github.com/pingcap/test-infra/sdk/resource" ) //1 参数化 bench_type， bench_sub_type, threads, -tidb-globals, -ddl command //2. 结果入库 //3. delete testbed //4. 监控 url： private-tidb.5ea21abf.b1f931b7.us-west-2.shared.aws.tidbcloud.com 对应的监控 //5. https://grafana.oauth.staging.shared.aws.tidbcloud.com/d/eDbRZpnWa2/performance-overview?var-cluster_id=1373933076658062026&&var-tenant_id=1369847559691367379&from=1656577720103&to=1656579010032 //s, err :=tc.GetStatus() // s.ClusterID // tc.GetSpec().OrgID // tc.GetSpec().ProjectID // 6 支持一个 TiFlash 的集群 TPCH/TPC-DS 50G // 可以用 db.benchbot.pingcap.net 这个域名连 benchbot 的 DB 了 var _ = ParameterizedDescribe("tidbcloud", func(flagSet *flag.FlagSet) { ctx := suiteTestCtx var ( email string arch string tiflashReplicas int benchType string dsn string threads string apThreads string benchSubTypes string sysbenchOptions string ycsbOptions string versions string benchmarkID string tidbGlobals arrayFlags storageURL string dbName string duration time.Duration warehouses int queries string jdbc string warmUP bool commands string commandDelay time.Duration resetEnv bool ) // misc flagSet.StringVar(&email, "email", "", "the email address for the case submitter") flagSet.StringVar(&benchmarkID, "benchmark-id", "9000001", "benchmark id to save the result the workload result") // 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.StringVar(&benchType, "bench-type", "", "bench type, ycsb|sysbench|tpcc|benchmarksql|tpch|tpcds|job|bank|shenma|ch|apdogfood|corebanking|hzbank|booking") 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]", "options for sysbench, like, --tables, --table-size, -mysql-ignore-errors") flagSet.StringVar(&ycsbOptions, "ycsb-options", "-p recordcount=100000000 -p operationcount=50000 -p workload=core -p fieldcount=10 -p fieldlength=100 -p requestdistribution=uniform", "options for ycsb, list of -p key=value") // cluster.global 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(&resetEnv, "reset-env", true, "if it's reset-env is true, the data will be refreshed 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") // cluster.tidb flagSet.Var(&tidbGlobals, "tidb-globals", "set global variables before play the workload") // cluster.tikv // cluster.tiflash flagSet.IntVar(&tiflashReplicas, "tiflash-replicas", 0, "tiflash replicas in the topo") // cluster.pd Context("tidbcloud", func() { It("tidbcloud", func() { // 0. initialize varibables tc := ctx.Resource("tc").(resource.TiDBCloudCluster) workloadNode := ctx.Resource("toolset").(resource.WorkloadNode) // NOTE: s3 path = s3:/// backupBucketName := "qa-workload-datasets" benchSubTypeList := strings.Split(benchSubTypes, ",") threadList := strings.Split(threads, ",") apThreadList := strings.Split(apThreads, ",") if len(tidbGlobals) == 0 { tidbGlobals = []string{""} } //pre-process default values if len(storageURL) == 0 || len(dbName) == 0 { storageURL, dbName = defaultDataSetCloud(benchType) } if len(queries) == 0 { queries = defaultQueries(benchType) } jdbc = jdbcEscape(benchType, jdbc) startTime := time.Now().Unix() stabilityGap := 2 * time.Minute dbURL := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL) dbHost, dbPort := dbURL.Hostname(), dbURL.Port() password, _ := dbURL.User.Password() firstTimeFlag := true for _, benchSubType := range benchSubTypeList { for _, thread := range threadList { for _, apThread := range apThreadList { for _, tidbGlobal := range tidbGlobals { benchEnv := fmt.Sprintf(`storage: %s, dbname: %s, version: %s, benchSubType: %s, thread: %s, apThread: %s, tidbGlobal: %s`, storageURL, dbName, versions, benchSubType, thread, apThread, tidbGlobal) log.Info(benchEnv) if resetEnv || firstTimeFlag { // 1. restore data Try(tc.InplaceRestore(ctx, resource.TiDBCloudInplaceRestoreOptions{ BackupType: resource.TiDBCloudInplaceRestoreBackupTypeFull, DBName: &dbName, BucketName: backupBucketName, ObjectPrefix: storageURL, })) time.Sleep(midwait) log.Info("data is ready.") firstTimeFlag = false } // 2. skip: warm up // for bank workload, need warmup before gather stats to avoid out-of-range predicate if benchType == "bank" { s2 := fmt.Sprintf(mysqlCmdTplCloud, password, dbHost, dbPort, dbName, bankResetCreateTimeSQL) mysqlCmd := workloadnode.CommandContext(ctx, s2) Try(mysqlCmd.Run(workloadNode)) warmupStr := fmt.Sprintf(bankCmdTplCloud, password, dbHost, dbPort, dbName, thread, 120) runCmd := workloadnode.CommandContext(ctx, warmupStr) runCmd.Run(workloadNode) } // for hzbank workload, need to create 3 sequences before workload if benchType == "hzbank" { s2 := fmt.Sprintf(mysqlCmdTplCloud, password, dbHost, dbPort, dbName, hzbankPrepQuery) mysqlCmd := workloadnode.CommandContext(ctx, s2) Try(mysqlCmd.Run(workloadNode)) } // 3. prepared the global setting if len(tidbGlobal) > 0 { s1 := fmt.Sprintf(mysqlCmdTplCloud, password, dbHost, dbPort, dbName, tidbGlobal) mysqlCmd := workloadnode.CommandContext(ctx, s1) mysqlCmd.Run(workloadNode) } // 4. gather statstics tableList := getAnalyzeList(benchType) if len(tableList) > 0 { s2 := fmt.Sprintf(mysqlCmdTpl2Cloud, password, dbHost, dbPort, dbName, tableList) mysqlCmd := workloadnode.CommandContext(ctx, s2) Try(mysqlCmd.Run(workloadNode)) } // 5. 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(truncateSysbenchTablesCloud, password, dbHost, dbPort, dbName, password, dbHost, dbPort, dbName) mysqlCmd := workloadnode.CommandContext(ctx, s2) Try(mysqlCmd.Run(workloadNode)) warmUP = false } // need password so I added a sysbench options. But I don't think this is a good approach. Maybe tune getBenchCmd a little? // Get the replace testbed name with hostname testbedCmdStr := fmt.Sprintf(testbedCmpTpl, dbHost, benchType) // log.Info(testbedCmdStr) testbedCmd := workloadnode.CommandContext(ctx, testbedCmdStr) Try(testbedCmd.Run(workloadNode)) // Get the commit info commitCmdStr := fmt.Sprintf(commitCmpTplCloud, password, dbHost, dbPort, benchType) // log.Info(commitCmdStr) commitCmd := workloadnode.CommandContext(ctx, commitCmdStr) Try(commitCmd.Run(workloadNode)) // 6. Run the workload. beginAt := time.Now().Unix() cmd := getBenchCmdCloud(benchType, benchSubType, dbHost, dbPort, dbName, thread, apThread, duration, warehouses, sysbenchOptions, ycsbOptions, queries, tiflashReplicas, jdbc, password) benchmarkCmd := workloadnode.CommandContext(ctx, cmd) go runBackendCommandsCloud(password, dbHost, dbPort, dbName, commands, commandDelay, workloadNode, ctx) Try(benchmarkCmd.Run(workloadNode)) tailLog(benchType, workloadNode, ctx) endAt := time.Now().Unix() 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 } // collect from prometheus promAPI, err := tc.PrometheusAPI() Try(err) // 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, workloadNode, ctx) //parse the result cliCmd := workloadnode.CommandContext(ctx, s3) Try(cliCmd.Run(workloadNode)) } //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 := getBenchResultWithPromAPI(promAPI, seconds, endTime) // gather version and git commit hash db2, err := sql.Open("mysql", fmt.Sprintf(defaultDSNCloud, password, dbHost, dbPort, dbName)) 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, dbHost, 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, dbHost, 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)) } } if resetEnv { Try(tc.ResetCluster()) } } } } } } s, err := tc.GetStatus() Try(err) clusterID := s.ClusterID projectID := tc.GetSpec().ProjectID url := fmt.Sprintf(cloudMonitorURL, clusterID, projectID, startTime*1000, time.Now().Unix()*1000) typeList := getType(benchType) log.Info(url) for _, t := range typeList { updateStmt := fmt.Sprintf(updateMonitorUrl, t, url, arch, dbHost) log.Info(updateStmt) db, err := sql.Open("mysql", dsn) if err != nil { panic(err) } defer db.Close() _, err = db.Exec(updateStmt) if err != nil { log.Error("Save cloud monitor url error: ", zap.Error(err)) } } }) }) })