package oltppkg import ( "bufio" "context" "database/sql" "errors" "fmt" "io" "net/url" "os" "regexp" "strconv" "strings" "sync" "time" yaml "gopkg.in/yaml.v3" . "github.com/onsi/gomega" . "github.com/pingcap/endless/pkg/matcher" "github.com/pingcap/endless/pkg/workloadnode" "github.com/pingcap/endless/pkg/workloadnode/cluster" "github.com/pingcap/log" "github.com/pingcap/test-infra/sdk/core" "github.com/pingcap/test-infra/sdk/resource" "go.uber.org/zap" k8swait "k8s.io/apimachinery/pkg/util/wait" ) var CaseCmdParmsDefault string = ` { workloadType: { type: [ 'sysbench_oltp_write_only' ], randomWorkloadScope: [ 'tpcc', 'sysbench_oltp_read_write', 'sysbench_oltp_write_only', 'sysbench_oltp_read_only' ] }, tpccParms: { storage: 's3://benchmark/tpcc-1000-4.0', db: 'tpcc', warehouse: 1000, threads: 512, ignoreErrors: '2013,1213,1105,1205,8022,8027,8028,9004,9007' workloadCheckIgnoreErros: '2013,1213,1105,1205,8022,8027,8028,9004,9007'}, sysbenchParms: { storage: 's3://benchmark/sysbench-32-10000000', db: 'test', tableNum: 32, tableSize: 10000000, threads: 512, ignoreErrors: '2013,1213,1105,1205,8022,8027,8028,9004,9007', workloadCheckIgnoreErros: '2013,1213,1105,1205,8022,8027,8028,9004,9007' }, clusterSpecParms: { tikvCPU: 8 }, tidbVariables: { globalVariables: [ "tidb_partition_prune_mode = 'static'" ]}, skipRunWorkload: false, upgradeParms: { upgradeByVersion: false, targetVersion: 'master', targetPdImage: 'hub-new.pingcap.net/qa/pd:master', targetTikvImage: 'hub-new.pingcap.net/qa/tikv:master', targetTidbImage: 'hub-new.pingcap.net/qa/tidb:master', masterPdVersion: '5.4.0', masterTidbVersion: '5.4.0', masterTikvVersion: '5.4.0' } } ` type CaseCmdParms struct { WorkloadType struct { Type []string `yaml:"type"` RandomWorkloadScope []string `yaml:"randomWorkloadScope"` } `yaml:"workloadType"` TpccParms struct { Storage string `yaml:"storage"` Db string `yaml:"db"` Warehouse int `yaml:"warehouse"` Threads int `yaml:"threads"` IgnoreErrors string `yaml:"ignoreErrors"` WorkloadCheckIgnoreErrors string `yaml:"workloadCheckIgnoreErrors"` } `yaml:"tpccParms"` SysbenchParms struct { Storage string `yaml:"storage"` Db string `yaml:"db"` TableNum int `yaml:"tableNum"` TableSize int `yaml:"tableSize"` Threads int `yaml:"threads"` IgnoreErrors string `yaml:"ignoreErrors"` WorkloadCheckIgnoreErrors string `yaml:"workloadCheckIgnoreErrors"` } `yaml:"sysbenchParms"` YcsbParms YcsbParms `yaml:"ycsbParms"` ClusterSpecParms struct { TiKVCPUNum int `yaml:"tikvCPU"` } `yaml:"clusterSpecParms"` TidbVariables struct { GlobalVariables []string `yaml:"globalVariables"` } `yaml:"tidbVariables"` RunTime RunTime `yaml:"runTime"` UpgradeParms UpgradeParms `yaml:"upgradeParms"` SkipRunWorkload bool `yaml:"skipRunWorkload"` SkipImportData bool `yaml:"skipImportData"` TiupClusterConfig TiupClusterConfig `yaml:"tiupClusterConfig"` } type TiupClusterConfig struct { Version string `yaml:"version"` Topo string `yaml:"topo"` ClusterName string `yaml:"clusterName"` TiupMirror string `yaml:"tiupMirror"` EnableTls bool `yaml:"enableTls"` TiupDrAutoSyncConfig TiupDrAutoSyncConfig `yaml:"tiupDrAutoSyncConfig"` } type TiupDrAutoSyncConfig struct { BackupDcRole string `yaml:"backupDcRole"` ReplicaNum int `yaml:"replicaNum"` PrimayBackupTikvNum string `yaml:"primayBackupTikvNum"` } type RunTime struct { TotalRunTime time.Duration `yaml:"totalRunTime"` WarmUpTime time.Duration `yaml:"warmUpTime"` FaultDuration time.Duration `yaml:"faultDuration"` RecoveryTime time.Duration `yaml:"recoveryTime"` } type YcsbParms struct { Storage string `yaml:"storage"` Db string `yaml:"db"` SubTypes []string `yaml:"subTypes"` RecordCount int `yaml:"recordCount"` OperationCount int `yaml:"operationCount"` ThreadCount int `yaml:"threadCount"` IgnoreErrors string `yaml:"ignoreErrors"` WorkloadCheckIgnoreErrors string `yaml:"workloadCheckIgnoreErrors"` } type UpgradeParms struct { UpgradeByVersion bool `yaml:"upgradeByVersion"` TargetVersion string `yaml:"targetVersion"` TargetPdImage string `yaml:"targetPdImage"` TargetTikvImage string `yaml:"targetTikvImage"` TargetTidbImage string `yaml:"targetTidbImage"` MasterPdVersion string `yaml:"masterPdVersion"` MasterTidbVersion string `yaml:"masterTidbVersion"` MasterTikvVersion string `yaml:"masterTikvVersion"` } type S3Parms struct { S3Endpoint string S3AccessKeyID string S3SecretAccessKey string } type RunTimePara struct { TotalRunTime time.Duration WarmUpTime time.Duration FaultDuration time.Duration RecoveryTime time.Duration RunInterval time.Duration } type WorkloadRunParms struct { //Tc resource.TiDBCluster WorkloadNode resource.WorkloadNode CmdList []*workloadnode.Cmd PauseParams chan PauseParams Cmdpw *io.PipeWriter Cmdpr *io.PipeReader RandType string } type PauseParams struct { IfPause bool CheckInterval time.Duration PauseDurtaion time.Duration } type CaseParams struct { Ctx core.TestContext Tc cluster.BaseCluster Br resource.BR BenchToolset resource.WorkloadNode CaseCmdParms CaseCmdParms WorkloadRunParms WorkloadRunParms S3Parms S3Parms SkipAfterEachCheck chan bool } func (c *CaseCmdParms) GetRunDbName() string { workloadType := c.WorkloadType.Type[0] var dbName string if strings.HasPrefix(workloadType, "sysbench_") { dbName = c.SysbenchParms.Db } else if strings.HasPrefix(workloadType, "tpcc") { dbName = c.TpccParms.Db } else { panic("invalid workloadType") } return dbName } func BuildCaseCmdParms(caseCmdParmsStr string) CaseCmdParms { caseCmdParmsStruct := new(CaseCmdParms) fmt.Printf("\n caseCmdParmsStr: \n %v \n", caseCmdParmsStr) err := yaml.Unmarshal([]byte(caseCmdParmsStr), caseCmdParmsStruct) Expect(err).Should(Succeed()) fmt.Printf("\n caseCmdParmsStruct: \n %v \n", *caseCmdParmsStruct) return *caseCmdParmsStruct } func NewRunTimePara(totalRunTime time.Duration, warmUpTime time.Duration, faultDuration time.Duration, recoveryTime time.Duration) *RunTimePara { return &RunTimePara{ TotalRunTime: totalRunTime, WarmUpTime: warmUpTime, FaultDuration: faultDuration, RecoveryTime: recoveryTime, } } type SysbenchCmd struct { options []string testname string command string } //--point_selects=N Number of point SELECT queries per transaction [10] //--random_points=N Number of random points in the IN() clause in generated SELECTs [10] //--auto_inc[=on|off] Use AUTO_INCREMENT column as Primary Key (for MySQL), or its alternatives in other DBMS. When disabled, use client-generated IDs [on] //--create_secondary[=on|off] Create a secondary index in addition to the PRIMARY KEY [on] //--rand-spec-pct=N percentage of values to be treated as 'special' (for special distribution) [1] //--rand-spec-res=N percentage of 'special' values to use (for special distribution) [75] //--delta=N Size of BETWEEN ranges [5] func NewSysbench(testname string, command string) *SysbenchCmd { s := &SysbenchCmd{testname: testname, command: command} s.options = append(s.options, "--mysql-user=root") s.options = append(s.options, "--db-driver=mysql") s.options = append(s.options, "--time=0") s.options = append(s.options, "--report-interval=10") return s } func (s *SysbenchCmd) RandomRangeDelta(delta int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--delta=%d", delta)) return s } func (s *SysbenchCmd) NumberOfRanges(number int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--number_of_ranges=%d", number)) return s } func (s *SysbenchCmd) RandSpecPct(pct int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--rand-spec-pct=%d", pct)) return s } func (s *SysbenchCmd) RandSpecRes(res int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--rand-spec-res=%d", res)) return s } func (s *SysbenchCmd) CreateSecondary(secondary string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--create_secondary=%s", secondary)) return s } func (s *SysbenchCmd) AutoInc(autInc string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--auto_inc=%s", autInc)) return s } func (s *SysbenchCmd) PointSelects(pointSelects int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--point_selects=%d", pointSelects)) return s } func (s *SysbenchCmd) RandomPoints(randomPoints int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--random_points=%d", randomPoints)) return s } func (s *SysbenchCmd) Db(db string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--mysql-db=%s", db)) return s } func (s *SysbenchCmd) RandType(randType string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--rand-type=%s", randType)) return s } func (s *SysbenchCmd) Testname(testname string) *SysbenchCmd { s.testname = testname return s } func (s *SysbenchCmd) Command(cmd string) *SysbenchCmd { s.command = cmd return s } func (s *SysbenchCmd) Host(host string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--mysql-host=%s", host)) return s } func (s *SysbenchCmd) Port(port string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--mysql-port=%s", port)) return s } func (s *SysbenchCmd) User(user string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--mysql-user=%s", user)) return s } func (s *SysbenchCmd) Tables(tables int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--tables=%d", tables)) return s } func (s *SysbenchCmd) TableSize(tablesize int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--table-size=%d", tablesize)) return s } func (s *SysbenchCmd) Threads(thread int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--threads=%d", thread)) return s } func (s *SysbenchCmd) IgnoreError(ignores string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--mysql-ignore-errors=%s", ignores)) return s } func (s *SysbenchCmd) RangeSize(rangesize int) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--range_size=%d", rangesize)) return s } func (s *SysbenchCmd) MysqlDebug(mysqlDebug string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--mysql-debug=%s", mysqlDebug)) return s } func (s *SysbenchCmd) Debug(debug string) *SysbenchCmd { s.options = append(s.options, fmt.Sprintf("--debug=%s", debug)) return s } func (s *SysbenchCmd) Build() string { return fmt.Sprintf("sysbench %s %s %s", strings.Join(s.options[:], " "), s.testname, s.command) } type TpccCmd struct { options []string command string } func NewTpcc(command string) *TpccCmd { s := &TpccCmd{command: command} return s } func (t *TpccCmd) Host(host string) *TpccCmd { t.options = append(t.options, fmt.Sprintf("--host %s", host)) return t } func (t *TpccCmd) Port(port string) *TpccCmd { t.options = append(t.options, fmt.Sprintf("-P%s", port)) return t } func (t *TpccCmd) Warehouses(warehouses int) *TpccCmd { t.options = append(t.options, fmt.Sprintf("--warehouses %d", warehouses)) return t } func (t *TpccCmd) Db(db string) *TpccCmd { t.options = append(t.options, fmt.Sprintf("-D %s", db)) return t } func (t *TpccCmd) Threads(threads int) *TpccCmd { t.options = append(t.options, fmt.Sprintf("-T %d", threads)) return t } func (t *TpccCmd) Command(command string) *TpccCmd { t.command = command return t } func (t *TpccCmd) Time(runtime string) *TpccCmd { t.options = append(t.options, fmt.Sprintf("--time %s", runtime)) return t } func (t *TpccCmd) IgnoreError(ignores string) *TpccCmd { t.options = append(t.options, fmt.Sprintf("--ignore-error '%s'", ignores)) return t } func (s *TpccCmd) Build() string { return fmt.Sprintf("go-tpc tpcc %s %s", s.command, strings.Join(s.options[:], " ")) } func RunFullTableScan(ctx context.Context, tc resource.TiDBCluster, timeout time.Duration, dbname string, tableprefix string, scannumber int) { scandb := dbname timer := time.NewTimer(timeout) Loop: for { select { case <-ctx.Done(): break Loop case <-timer.C: break Loop default: } var wg sync.WaitGroup for i := 1; i < scannumber; i++ { wg.Add(1) go func(ctx context.Context, scandb string, tableprefix string) { defer wg.Done() query := fmt.Sprintf("select count(*) from %s.%s", scandb, tableprefix+strconv.Itoa(i)) ctx2, cancel := context.WithTimeout(ctx, 10*time.Minute) defer cancel() start := time.Now() db := Try(tc.GetDB("")).(*sql.DB) db.SetMaxOpenConns(8192) db.SetMaxIdleConns(8192) defer db.Close() _, err := db.ExecContext(ctx2, query) cost := time.Since(start) if cost > 1*time.Second { log.Info("query cost:", zap.Duration("query cost:", cost)) } if err != nil { log.Warn("Scan table failed.", zap.String("query", query)) log.Warn(err.Error()) return } }(ctx, scandb, tableprefix) } wg.Wait() time.Sleep(30 * time.Second) } } func ExecWorkloadCmd(ctx context.Context, wl resource.WorkloadNode, runcmd string, syncflag bool, pw io.Writer) (*workloadnode.Cmd, error) { cmd := workloadnode.CommandContext(ctx, runcmd) if pw != nil { cmd.Stdout = pw } else { cmd.Stdout = resource.NopCloser(os.Stdout) } if syncflag { err := cmd.Run(wl) if err != nil { return nil, err } } else { err := cmd.Start(wl) if err != nil { return nil, err } // go func(ctx context.Context, pr2 *io.PipeReader) { // OutPutCheck(ctx, pr2, []string{}, []string{"error", "fail", "panic"}) // }(ctx, pr) } return cmd, nil } func RecreateDatabase(ctx context.Context, tc resource.MySQLCompatible, dbname string) error { dbUrl := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL) db, err := sql.Open("mysql", resource.BuildDSN(dbUrl, "")) if err != nil { log.Warn("Fail to connect to db.") return err } log.Info(fmt.Sprintf("drop database if exists %s", dbname)) _, err = db.ExecContext(ctx, fmt.Sprintf("drop database if exists %s", dbname)) if err != nil { return err } log.Info(fmt.Sprintf("create database %s", dbname)) _, err = db.ExecContext(ctx, fmt.Sprintf("create database %s", dbname)) if err != nil { return err } return nil } func GetGcLeader(ctx context.Context, tc resource.TiDBCluster) (int, error) { log.Info("Begin to get gc leader") dbURL, _ := tc.ServiceURL(resource.DBAddr) dbDSN := fmt.Sprintf("root@tcp(%s)/", dbURL.Host) db := Try(sql.Open("mysql", dbDSN)).(*sql.DB) defer db.Close() log.Info("Execute sql to get gc leader") rows, err := db.QueryContext(ctx, "select VARIABLE_VALUE from mysql.tidb where variable_name='tikv_gc_leader_desc'") defer rows.Close() if err != nil { log.Warn(fmt.Sprintf("%v", err)) return 0, err } gcleaders := make([]string, 0) for rows.Next() { var gcleader string if err := rows.Scan(&gcleader); err != nil { log.Error(fmt.Sprintf("%s", err)) return 0, err } gcleaders = append(gcleaders, gcleader) } var tidb_name string if len(gcleaders) < 1 { return 0, fmt.Errorf("gc leader return less than 1") } else { cur_gc := gcleaders[0] tidb_name = strings.Split(strings.Split(cur_gc, ",")[0], ":")[1] } instance_id := strings.Split(tidb_name, "-")[len(strings.Split(tidb_name, "-"))-1] peer_id, err := strconv.Atoi(instance_id) if err != nil { return 0, err } log.Info("Get gc leader ", zap.Int("gc_leader", peer_id)) return peer_id, nil } func WorkLoadErrLogCheck(ctx context.Context, pr *io.PipeReader, checkTime time.Duration, isCheck bool, ignoreErr []string) error { log.Info("start check workLoad...") b := bufio.NewReader(pr) timeOut := time.After(checkTime) for { select { case <-ctx.Done(): log.Info("Log check ctx done.") return nil case <-timeOut: log.Info("Log check ctx timeout.") return nil default: line, err := b.ReadString('\n') if err != nil { if err == io.EOF { break } if strings.Contains(err.Error(), "io: read/write on closed pipe") { log.Info(err.Error()) break } log.Info(err.Error()) return err } else { log.Info(line) if isCheck { isIgnoredErr := false if strings.Contains(strings.ToLower(line), strings.ToLower("error")) || //strings.Contains(strings.ToLower(line), strings.ToLower("fail")) || strings.Contains(strings.ToLower(line), strings.ToLower("fatal")) { if len(ignoreErr) != 0 { for _, s := range ignoreErr { if strings.Contains(strings.ToLower(line), strings.ToLower(s)) { isIgnoredErr = true break } } } if !isIgnoredErr { err = errors.New(line) log.Info("workload run with errors.") return err } } } } } } } type WorkloadErrCheckChan struct { IfCheck bool IgnoreErr []string IfPrint bool } func WorkLoadErrLogCheckWithChan(ctx context.Context, pr *io.PipeReader, checkTime time.Duration, workloadErrCheckParams chan WorkloadErrCheckChan) error { log.Info("start check workLoad...") b := bufio.NewReader(pr) timeOut := time.After(checkTime) lastWorkloadParms := WorkloadErrCheckChan{IfCheck: false, IgnoreErr: nil, IfPrint: false} for { select { case <-ctx.Done(): log.Info("Log check ctx done.") return nil case <-timeOut: log.Info("Log check ctx timeout.") return nil default: line, err := b.ReadString('\n') if err != nil { if err == io.EOF { break } if strings.Contains(err.Error(), "io: read/write on closed pipe") { log.Info(err.Error()) break } log.Info(err.Error()) return err } else { select { case wp := <-workloadErrCheckParams: log.Info("Got workloadErrCheckParams", zap.Bool("ifcheck", wp.IfCheck), zap.Bool("IfPrint", wp.IfPrint), zap.Strings("IgnoreErr", wp.IgnoreErr)) lastWorkloadParms = wp default: } if lastWorkloadParms.IfPrint { log.Info(line) } if !lastWorkloadParms.IfCheck { continue } isIgnoredErr := false if strings.Contains(strings.ToLower(line), strings.ToLower("error")) || //strings.Contains(strings.ToLower(line), strings.ToLower("fail")) || strings.Contains(strings.ToLower(line), strings.ToLower("fatal")) { if len(lastWorkloadParms.IgnoreErr) != 0 { for _, s := range lastWorkloadParms.IgnoreErr { if strings.Contains(strings.ToLower(line), strings.ToLower(s)) { isIgnoredErr = true break } } } if !isIgnoredErr { err = errors.New(line) log.Info("workload run with errors.") return err } } } } } } func RunSysbenchBulkInsert(ctx context.Context, tc resource.MySQLCompatible, caseCmdParms CaseCmdParms, workloadRunPara WorkloadRunParms) (*workloadnode.Cmd, error) { dbURL := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL) timestamp := strconv.FormatInt(time.Now().UnixNano()/1e6, 10) dbName := fmt.Sprintf("%s_%s", "bulkinsert", timestamp) Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("bulk_insert", "prepare") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(caseCmdParms.SysbenchParms.TableNum). IgnoreError(caseCmdParms.SysbenchParms.IgnoreErrors).Debug("on") Try(ExecWorkloadCmd(ctx, workloadRunPara.WorkloadNode, sysbenchcmd.Build(), true, nil)) sysbench = NewSysbench("bulk_insert", "run") sysbenchcmd = sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(caseCmdParms.SysbenchParms.TableNum). IgnoreError(caseCmdParms.SysbenchParms.IgnoreErrors).Debug("on") return ExecWorkloadCmd(ctx, workloadRunPara.WorkloadNode, sysbenchcmd.Build(), false, resource.NopCloser(workloadRunPara.Cmdpw)) } func StopWorkload(node resource.WorkloadNode, nodeType string) { var getProcessCmd string if nodeType == "goTpc" { getProcessCmd = "pgrep go-tpc" } else if nodeType == "sysBench" { getProcessCmd = "pgrep sysbench" } else { getProcessCmd = fmt.Sprintf("pgrep %s", nodeType) } log.Info(getProcessCmd) stdout, _, _, _ := node.Exec(resource.WorkloadNodeExecOptions{Command: getProcessCmd + "|wc -l"}) log.Info(stdout) if stdout != "0" { killProcessCmd := fmt.Sprintf(`kill -9 $(%s) `, getProcessCmd) log.Info(killProcessCmd) node.Exec(resource.WorkloadNodeExecOptions{Command: killProcessCmd}) } } func CreateSysbenchTable(ctx context.Context, tc resource.MySQLCompatible, dbName string, tableNum int, usePartition string, useSecondIndex bool) { var sqlTmp string if useSecondIndex { sqlTmp = `CREATE TABLE %s.sbtest%d ( id bigint(20) NOT NULL AUTO_INCREMENT, k int(11) NOT NULL DEFAULT '0', c char(120) NOT NULL DEFAULT '', pad char(60) NOT NULL DEFAULT '', PRIMARY KEY (id), KEY k_1 (k) ) ` } else { sqlTmp = `CREATE TABLE %s.sbtest%d ( id bigint(20) NOT NULL AUTO_INCREMENT, k int(11) NOT NULL DEFAULT '0', c char(120) NOT NULL DEFAULT '', pad char(60) NOT NULL DEFAULT '', PRIMARY KEY (id) ) ` } if usePartition == "hash" { sqlTmp = sqlTmp + "partition by hash(id) PARTITIONS 8" } if usePartition == "range" { sqlTmp = sqlTmp + `partition by range(id) ( partition p0 values less than (2000000), partition p1 values less than (4000000), partition p2 values less than (6000000), partition p3 values less than (8000000), partition p4 values less than (10000000), PARTITION p5 VALUES LESS THAN MAXVALUE)` } dbUrl := Try(tc.ServiceURL(resource.DBAddr)).(*url.URL) db, err := sql.Open("mysql", resource.BuildDSN(dbUrl, dbName, "timeout=300m", "readTimeout=300m", "writeTimeout=300m")) Expect(err).Should(Succeed()) defer db.Close() Try(RecreateDatabase(ctx, tc, dbName)) for i := 1; i <= tableNum; i++ { sql := fmt.Sprintf(sqlTmp, dbName, i) log.Info(fmt.Sprintf("Run sql: %s", sql)) Try(db.ExecContext(ctx, sql)) } } func GetWorkloadMap() map[string][]string { workloadTypeMap := map[string][]string{ "w": {"tpcc", "tpcc_prepare", "sysbench_prepare", "sysbench_oltp_insert", "sysbench_oltp_write_only", "sysbench_oltp_delete", "sysbench_bulk_insert", "sysbench_oltp_insert_large", "sysbench_oltp_insert_hash_partition", "sysbench_oltp_insert_range_partition", "sysbench_oltp_insert_without_index", "sysbench_oltp_update_index", "sysbench_oltp_update_non_index", "ycsb"}, "wr": {"sysbench_oltp_read_write"}, "r": {"sysbench_oltp_read_only", "sysbench_oltp_point_select", "sysbench_select_random_points", "sysbench_oltp_read_only_hot", "sysbench_select_random_ranges", "sysbench_oltp_read_only_special", "sysbench_oltp_read_only_stale_read"}, } return workloadTypeMap } func GetWorkloadTypeList(ioType *string) []string { workloadTypeMap := GetWorkloadMap() if ioType != nil { return workloadTypeMap[*ioType] } else { var workloadList []string for _, v := range workloadTypeMap { workloadList = append(workloadList, v...) } return workloadList } } func GetRandomWorkloadType() string { workloadList := GetWorkloadTypeList(nil) rand := GetRandInt(0, len(workloadList)) return workloadList[rand] } func GetWorkloadIoType(workloadType string) (string, error) { workloadTypeMap := GetWorkloadMap() for t, l := range workloadTypeMap { s := StrSlice(l) if s.Has(workloadType) { return t, nil } } log.Warn("Invalid workloadType", zap.String("workloadType", workloadType)) return "", fmt.Errorf("invalid workloadtype") } func WaitForAllTablesReady(ctx context.Context, tc resource.MySQLCompatible, timeout time.Duration, workload string, caseCmdParms CaseCmdParms) { log.Info("waiting for all tables are ready.") var dbName string var targetTableNum int if strings.HasPrefix(workload, "sysbench") { dbName = caseCmdParms.SysbenchParms.Db targetTableNum = caseCmdParms.SysbenchParms.TableNum } else if strings.HasPrefix(workload, "tpcc") { dbName = caseCmdParms.TpccParms.Db targetTableNum = 9 } else { err := errors.New("invalid workloadType") Expect(err).Should(Succeed()) } Eventually(func() int { tables := GetAllTablesNames(ctx, tc, dbName) tableNum := len(tables) return tableNum }, timeout, 1*time.Minute).Should(BeNumerically("==", targetTableNum)) log.Info("all tables are ready.") } func BuildWorkloadRuncmd(ctx context.Context, tc resource.MySQLCompatible, workload string, benchToolset resource.WorkloadNode, caseCmdParms *CaseCmdParms, randType string) string { if !StrSlice(GetWorkloadTypeList(nil)).Has(workload) { PanicWithOutput("Invalid workloadType.") } timestamp := strconv.FormatInt(time.Now().UnixNano()/1e6, 10) // var runcmd string dbURL, _ := tc.ServiceURL(resource.DBAddr) db := Try(tc.GetDB("")).(*sql.DB) defer db.Close() var ( runcmd string dbName string threads int tableNum int tableSize int warehouse int ignoreErrors string ) if strings.HasPrefix(workload, "tpcc") { dbName = caseCmdParms.TpccParms.Db threads = caseCmdParms.TpccParms.Threads warehouse = caseCmdParms.TpccParms.Warehouse ignoreErrors = caseCmdParms.TpccParms.IgnoreErrors } else if strings.HasPrefix(workload, "sysbench") { dbName = caseCmdParms.SysbenchParms.Db threads = caseCmdParms.SysbenchParms.Threads tableNum = caseCmdParms.SysbenchParms.TableNum tableSize = caseCmdParms.SysbenchParms.TableSize ignoreErrors = caseCmdParms.SysbenchParms.IgnoreErrors } else if strings.HasPrefix(workload, "ycsb") { } else { err := errors.New("invalid workloadType") Expect(err).Should(Succeed()) } switch { case workload == "ycsb": runcmd = BuildWorkloadYcsb(ctx, tc, *caseCmdParms) case workload == "sysbench_prepare": caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "sysbenchprepare", timestamp) dbName = caseCmdParms.SysbenchParms.Db Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("oltp_insert", "prepare") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors) runcmd = sysbenchcmd.Build() case workload == "sysbench_bulk_insert": caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "bulkinsert", timestamp) dbName = caseCmdParms.SysbenchParms.Db Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("bulk_insert", "prepare") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). IgnoreError(ignoreErrors).Debug("on") Try(ExecWorkloadCmd(ctx, benchToolset, sysbenchcmd.Build(), true, nil)) sysbench = NewSysbench("bulk_insert", "run") sysbenchcmd = sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). IgnoreError(ignoreErrors).Debug("on") runcmd = sysbenchcmd.Build() case workload == "sysbench_oltp_insert_large": caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "largetablenumberinsert", timestamp) dbName = caseCmdParms.SysbenchParms.Db tableNum = caseCmdParms.SysbenchParms.TableNum Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("oltp_insert", "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors) CreateSysbenchTable(ctx, tc, dbName, tableNum, "", true) runcmd = sysbenchcmd.Build() case workload == "sysbench_oltp_insert_without_index": caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "insertwoindex", timestamp) dbName = caseCmdParms.SysbenchParms.Db Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("oltp_insert", "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors).RandType(randType) CreateSysbenchTable(ctx, tc, dbName, tableNum, "", false) runcmd = sysbenchcmd.Build() case workload == "sysbench_oltp_insert_hash_partition": caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "partitioninsert", timestamp) dbName = caseCmdParms.SysbenchParms.Db Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("oltp_insert", "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors).RandType(randType) CreateSysbenchTable(ctx, tc, dbName, tableNum, "hash", true) runcmd = sysbenchcmd.Build() case workload == "sysbench_oltp_insert_range_partition": caseCmdParms.SysbenchParms.Db = fmt.Sprintf("%s_%s", "partitioninsert", timestamp) dbName = caseCmdParms.SysbenchParms.Db Try(RecreateDatabase(ctx, tc, dbName)) sysbench := NewSysbench("oltp_insert", "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors).RandType(randType) CreateSysbenchTable(ctx, tc, dbName, tableNum, "range", true) runcmd = sysbenchcmd.Build() case workload == "sysbench_select_random_points": sysbench := NewSysbench(workload[9:], "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors). RandType(randType).RandomPoints(GetRandInt(10, 200)).PointSelects(GetRandInt(1, 100)) runcmd = sysbenchcmd.Build() case workload == "sysbench_select_random_ranges": sysbench := NewSysbench(workload[9:], "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors). RandType(randType).RandomRangeDelta(GetRandInt(10, 200)) runcmd = sysbenchcmd.Build() case workload == "sysbench_oltp_read_only_hot": sysbench := NewSysbench("oltp_read_only", "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors).RandType(randType) sysbenchcmd.RangeSize(GetRandInt(100, 10000)) runcmd = sysbenchcmd.Build() case workload == "sysbench_oltp_read_only_special": sysbench := NewSysbench("oltp_read_only", "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(1).Command("run").TableSize(tableSize).IgnoreError(ignoreErrors).RandType("special").RandSpecPct(10).RandSpecRes(80) runcmd = sysbenchcmd.Build() case strings.HasPrefix(workload, "sysbench_oltp_point_select"): sysbench := NewSysbench(workload[9:], "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(256). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors).PointSelects(10) runcmd = sysbenchcmd.Build() case strings.HasPrefix(workload, "sysbench_"): sysbench := NewSysbench(workload[9:], "run") sysbenchcmd := sysbench.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Threads(threads). Tables(tableNum).TableSize(tableSize).IgnoreError(ignoreErrors).RandType(randType) runcmd = sysbenchcmd.Build() case workload == "tpcc_prepare": caseCmdParms.TpccParms.Db = fmt.Sprintf("%s_%s", "tpccprepare", timestamp) dbName = caseCmdParms.TpccParms.Db Try(RecreateDatabase(ctx, tc, dbName)) tpcc := NewTpcc("prepare") tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(ignoreErrors) runcmd = tpcccmd.Build() case workload == "tpcc": tpcc := NewTpcc("run") tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(ignoreErrors) runcmd = tpcccmd.Build() default: tpcc := NewTpcc("run") tpcccmd := tpcc.Db(dbName).Host(dbURL.Hostname()).Port(dbURL.Port()).Warehouses(warehouse).Threads(threads).Time("36000m").IgnoreError(ignoreErrors) runcmd = tpcccmd.Build() } return runcmd } func (workloadRunParms *WorkloadRunParms) Stop() { for _, cmd := range workloadRunParms.CmdList { cmd.Stop() } } func (workloadRunParms *WorkloadRunParms) Pause(ctx context.Context, duration time.Duration) { log.Info("pause workload") pauseParams := PauseParams{IfPause: true, CheckInterval: duration, PauseDurtaion: duration} workloadRunParms.SendPauseParamsWithTimeout(pauseParams, 10*time.Minute) StopWorkload(workloadRunParms.WorkloadNode, "goTpc") StopWorkload(workloadRunParms.WorkloadNode, "sysBench") SleepWithCtx(ctx, duration) pauseParams = PauseParams{IfPause: false, PauseDurtaion: duration} workloadRunParms.SendPauseParamsWithTimeout(pauseParams, 10*time.Minute) log.Info("pause workload finished") } func (workloadRunParms *WorkloadRunParms) SendPauseParamsWithTimeout(pauseParams PauseParams, timeOut time.Duration) { select { case workloadRunParms.PauseParams <- pauseParams: log.Info("pause workload finished") case <-time.After(timeOut): panic("Waiting for pause workload timeout") } } func (workloadRunParms *WorkloadRunParms) PauseWithResume(ctx context.Context, checkInterval time.Duration, checkTimeOut time.Duration, resumeFlag chan bool) { log.Info("pause workload") workloadRunParms.PauseParams <- PauseParams{IfPause: true, CheckInterval: checkInterval, PauseDurtaion: checkTimeOut} StopWorkload(workloadRunParms.WorkloadNode, "goTpc") StopWorkload(workloadRunParms.WorkloadNode, "sysBench") select { case <-ctx.Done(): log.Info("resume workload for context down") pauseParams := PauseParams{IfPause: false} workloadRunParms.SendPauseParamsWithTimeout(pauseParams, 10*time.Minute) break case <-resumeFlag: log.Info("resume workload") pauseParams := PauseParams{IfPause: false} workloadRunParms.SendPauseParamsWithTimeout(pauseParams, 10*time.Minute) } log.Info("Pause Resume workload finished") } func (workloadRunParms *WorkloadRunParms) InitAndStart(ctx context.Context, tc resource.MySQLCompatible, workLoadNode resource.WorkloadNode, caseCmdParams CaseCmdParms, ifRestart bool) { workloadRunParms.Init(ctx, tc, workLoadNode) workloadRunParms.Start(ctx, tc, caseCmdParams, ifRestart) } func (workloadRunParms *WorkloadRunParms) Init(ctx context.Context, tc resource.MySQLCompatible, workLoadNode resource.WorkloadNode) { cmdpr, cmdpw := io.Pipe() randType := "" if workloadRunParms.RandType == "" { var err error RandTypeList, err := GetSysbenchRandType(workLoadNode) if err != nil { log.Warn("Get sysbench rand type failed.") RandTypeList = []string{"uniform", "gaussian"} } randType = RandTypeList[GetRandInt(0, len(RandTypeList))] } else { randType = workloadRunParms.RandType } log.Info(fmt.Sprintf("sysbench workload rand type is %s", randType)) *workloadRunParms = WorkloadRunParms{ //Tc: tc, WorkloadNode: workLoadNode, Cmdpw: cmdpw, Cmdpr: cmdpr, PauseParams: make(chan PauseParams, 2), RandType: randType, } } func (workloadRunParms *WorkloadRunParms) Start(ctx context.Context, tc resource.MySQLCompatible, caseCmdParams CaseCmdParms, ifRestart bool) { workloadList := caseCmdParams.WorkloadType.Type LogAndBy(fmt.Sprintf("Start to run workload %v with ignoreErrors:%s", workloadList, caseCmdParams.SysbenchParms.IgnoreErrors)) for _, workload := range workloadList { //RandTypeList := []string{"uniform", "gaussian", "special", "pareto"} //support in 1.0.20 // RandTypeList := []string{"uniform", "gaussian", "pareto", "zipfian"} //1.1.0-e56f46e if workloadRunParms.RandType == "" { RandTypeList, err := GetSysbenchRandType(workloadRunParms.WorkloadNode) if err != nil { log.Warn("Get sysbench rand type failed.") RandTypeList = []string{"uniform", "gaussian"} } workloadRunParms.RandType = RandTypeList[GetRandInt(0, len(RandTypeList))] } runcmd := BuildWorkloadRuncmd(ctx, tc, workload, workloadRunParms.WorkloadNode, &caseCmdParams, workloadRunParms.RandType) cmd := Try(ExecWorkloadCmd(ctx, workloadRunParms.WorkloadNode, runcmd, false, resource.NopCloser(workloadRunParms.Cmdpw))).(*workloadnode.Cmd) if ifRestart { workloadRunParms.Restart(ctx, tc, cmd, caseCmdParams) } if workload == "tpcc_prepare" || workload == "sysbench_prepare" { WaitForAllTablesReady(ctx, tc, 120*time.Minute, workload, caseCmdParams) } WaitFor(ctx, "waiting for workload starting", 20*time.Second) workloadRunParms.CmdList = append(workloadRunParms.CmdList, cmd) } log.Info("Start finished.") } func (workloadRunParms *WorkloadRunParms) Restart(ctx context.Context, tc resource.MySQLCompatible, cmd *workloadnode.Cmd, caseCmdParms CaseCmdParms) { go func() { log.Info("Restart workload start.") Loop: for { select { case <-ctx.Done(): log.Info("Restart workload context done") break Loop default: } if cmd == nil { log.Info("cmd is null") break Loop } err := cmd.Wait() if err == nil { log.Info("Workload exist without error, no need to restart.") } else { log.Info("Workload exist, check if need to restart it.", zap.Error(err)) select { case PauseParams := <-workloadRunParms.PauseParams: if PauseParams.IfPause { log.Info("Pause workload,waiting for resume.", zap.Duration("PauseParams.PauseDurtaion", PauseParams.PauseDurtaion)) k8swait.PollImmediateWithContext(ctx, PauseParams.CheckInterval, PauseParams.PauseDurtaion, func(ctx context.Context) (done bool, err error) { select { case ResumeParams := <-workloadRunParms.PauseParams: if !ResumeParams.IfPause { return true, nil } default: } return false, nil }) } default: var err error cmdStr := cmd.String() if strings.Contains(cmdStr, "bulk_insert") { log.Info("restart bulk insert") cmd = Try(RunSysbenchBulkInsert(ctx, tc, caseCmdParms, *workloadRunParms)).(*workloadnode.Cmd) } else { log.Info("restart workload") cmd, err = ExecWorkloadCmd(ctx, workloadRunParms.WorkloadNode, cmd.String(), false, resource.NopCloser(workloadRunParms.Cmdpw)) if err != nil { log.Warn("Restart failed this time.") } } } } } log.Info("Restart workload finished.") }() } func BuildWorkloadYcsb(ctx context.Context, tc resource.MySQLCompatible, caseCmdParms CaseCmdParms) string { dbURL, _ := tc.ServiceURL(resource.DBAddr) ycsbCmdTpl := `go-ycsb run mysql -P /ycsb/workloads/%s -p mysql.host=%s -p mysql.port=%s -p recordcount=%d -p operationcount=%d -p workload=core -p fieldcount=10 -p fieldlength=100 -p threadcount=%d -p requestdistribution=zipfian` ycsbSubTypes := caseCmdParms.YcsbParms.SubTypes SubType := ycsbSubTypes[GetRandInt(0, len(ycsbSubTypes))] cmd := fmt.Sprintf(ycsbCmdTpl, SubType, dbURL.Hostname(), dbURL.Port(), caseCmdParms.YcsbParms.RecordCount, caseCmdParms.YcsbParms.OperationCount, caseCmdParms.YcsbParms.ThreadCount, ) return cmd } func GetSysbenchRandType(worklodNode resource.WorkloadNode) ([]string, error) { cmd := "sysbench --help |grep rand-type" stdout, _, _, err := worklodNode.Exec(resource.WorkloadNodeExecOptions{Command: cmd}) if err != nil { return nil, err } //stdout should like: --rand-type=STRING random numbers distribution {uniform,gaussian,special,pareto} [special] reg := regexp.MustCompile(`{.*?}`) s := reg.FindAllStringSubmatch(stdout, -1) if len(s) != 1 { return nil, fmt.Errorf("not find or find more than 1") } log.Info(fmt.Sprintf("random numbers distribution:%v", s)) randTypeStr1 := s[0][0] randTypeStr2 := strings.TrimLeft(randTypeStr1, "{") randTypeStr3 := strings.TrimRight(randTypeStr2, "}") randTypeStr := strings.Replace(randTypeStr3, " ", "", -1) randTypeList := strings.Split(randTypeStr, ",") log.Info("randTypeList:", zap.Strings("randTypeList", randTypeList)) return randTypeList, nil }