package common import ( //"mime/multipart" "database/sql" "errors" "net/http" //"bytes" "fmt" //"io" "io/ioutil" "os" "strconv" "strings" "testing" "time" l4g "github.com/alecthomas/log4go" "github.com/dailymotion/allure-go" "github.com/gavv/httpexpect/v2" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "github.com/tidbcloud/qa-PlatformAPIAuto/models" ) const ( FeaturePiTR = "featurePiTR" FeatureDualRegion = "dualRegionBR" ) type DataFormat string const ( csvFormat DataFormat = "csv" sqlFormat DataFormat = "sql" parquetFormat DataFormat = "parquet" snappyFormat DataFormat = "snappy" auroraFormat DataFormat = "aurora" ) func CheckDataSyncWithMysql(t assert.TestingT, orgId, projectId, clusterId, mysqlHost, mysqlPassword string, mysqlPost int, sqlUrl, downSqlUrl, dmUser string, expctResult, downExpctResult bool, maxCheckTimes int) { //获取cluster信息 pathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, pathParam, nil, nil) tidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) //添加 filter AddTrafficFilter(t, orgId, projectId, clusterId, false) //连接数据库创建表 db, err := ConnectTidb(mysqlHost, int64(mysqlPost), mysqlPassword, database, dmUser) result, terr := CreateTable(db, sqlUrl) value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("ConnectTidb and Create cdc Table, result is:%s, connectTidb failed reason:%s, CreateS3Table failed reason:%s, tidbAddress:%s", result, err, terr, tidbAddress)), allure.Action(func() { value.Equal(expctResult) })) l4g.Info("create table in upstream tidb.") //连接下游数据库 time.Sleep(time.Minute * 5) db, err = ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") if err == nil { terr = Retry(maxCheckTimes, time.Second, func() error { result, terr = SelectData(db, downSqlUrl, false, 1, "autotest") if terr != nil { l4g.Error("select downstream data failed, sql: %s, err: %v", downSqlUrl, terr) return errors.New("select downstream data failed") } return nil }) } value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s", result, err, terr)), allure.Action(func() { value.Equal(downExpctResult) })) } func CheckStandardBackupOperation(t assert.TestingT, projectId, clusterId, provider string, isDevTier bool, tidb_count, tikv_count, tifish_count int) { name := "Standard-backup-" + GenTempName() CheckBackupOperation(t, projectId, clusterId, provider, isDevTier, tidb_count, tikv_count, tifish_count, name, RestoreModeSnapshot, int64(0)) //自动备份 check CheckAutoBackupOperation(t, provider, projectId, clusterId, isDevTier) } func CheckEBSBackupOperation(t assert.TestingT, projectId, clusterId, provider string, isDevTier bool, tidb_count, tikv_count, tifish_count int) { name := "ebs-backup-" + GenTempName() var region string dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if provider == "aws" { region = dbassRequestInfo.CommonEnv.AwsRegion } else { region = dbassRequestInfo.CommonEnv.GcpRegion } clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} backupSettingBody := NewBackupSetting(provider, clusterId, region, "", false, false, true, "18:00", 1) backupSettingResp := dbassRequestInfo.DbassRequest(t, "backupSetting", backupSettingBody, clusterPathParam, nil, nil) l4g.Info("backupSettingResp is: %s, body is:%v, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam) allure.Step( allure.Description(fmt.Sprintf("backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam)), allure.Action(func() { backupSettingResp.Status(http.StatusOK) })) CheckBackupOperation(t, projectId, clusterId, provider, isDevTier, tidb_count, tikv_count, tifish_count, name, RestoreModeSnapshot, int64(0)) } func DoSQLInCluster(t assert.TestingT, projectId, clusterId, sql string, checkDataFlag bool, idValue int, nameValue string) (bool, error) { l4g.Info("begin DoSQLInCluster") dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId //获取cluster信息 clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) tidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) db, _ := ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") var result bool var err error if strings.Contains(sql, "CREATE") { result, err = CreateTable(db, sql) } else if strings.Contains(sql, "INSERT") { result, err = InsertDataToTest(db, sql) } else if strings.Contains(sql, "select") { result, err = SelectData(db, sql, checkDataFlag, idValue, nameValue) } else if strings.Contains(sql, "DROP") { result, err = DropTestDatabases(db, sql) } else if strings.Contains(sql, "DELETE") { result, err = DeleteData(db, sql) } l4g.Info("do sql:%s, result:%s, err:%s", sql, result, err) return result, err } func CheckBackupOperation(t assert.TestingT, projectId, clusterId, provider string, isDevTier bool, tidb_count, tikv_count, tifish_count int, cluster_name string, restore_mode RestoreMode, restored_ts int64) { if clusterId == "" { return } //主动备份集群 dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if dbassRequestInfo.CommonEnv.TiflashCpu == "" { tifish_count = 0 } //获取cluster信息 clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) tidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) tidbVersion := getClusterInfoResp.JSON().Path("$.cluster.version").String().Raw() db, err := ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") port := dbassRequestInfo.CommonEnv.Port result, ierr := CreateTable(db, createTable) if result { result, ierr = InsertDataToTest(db, insert) } //value := httpexpect.NewBoolean(t, result) /*allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err, ierr, tidbAddress)), allure.Action(func() { value.Equal(true) }))*/ l4g.Info("begin to maual backup!") backupName := GenTempName() manualBackup := NewManualBackup(backupName) manualBackupResp := dbassRequestInfo.DbassRequest(t, "manualBackup", manualBackup, clusterPathParam, nil, nil) l4g.Info("do maual backup, manualBackup is %+v, clusterPathParam is %+v, manualBackupResp is %s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do manaul backup, body:%+v, clusterPathParam:%+v, manualBackupResp:%s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw())), allure.Action(func() { manualBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("id") })) if manualBackupResp.Raw().StatusCode != StatusOK { return } //查询backup列表 getBackupListResp := dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backup list, getBackupListResp is: %s", getBackupListResp.JSON().Raw()) CheckBackupInfo(getBackupListResp, backupName, "name", backupName) backupId := GetBackupId(getBackupListResp, backupName, "id") for round := 0; round <= 15; round++ { time.Sleep(time.Second * 60) l4g.Info("get backup list, round is %d", round) getBackupListResp = dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backup list, getBackupListResp is: %s", getBackupListResp.JSON().Raw()) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) == "SUCCESS" { l4g.Info("manual backup is success! ", round) break } } allure.Step(allure.Description(fmt.Sprintf("check manaul backup status is success, EXpectStatus:%s, ActualStatus:%s, "+ "EXpectBackupSize:%d, ActualBackupSize:%d, backupId:%s", "SUCCESS", GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus), 290000, CheckBackupSize(getBackupListResp, backupName), backupId)), allure.Action(func() { CheckBackupInfo(getBackupListResp, backupName, "status", "SUCCESS") backsize := CheckBackupSize(getBackupListResp, backupName) assert.Greater(t, backsize, 290000) })) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) != "SUCCESS" { return } //恢复备份 l4g.Info("begin to restore the backup !") var tikvStorage, tiflashStorage int var tidbProfileId, tikvProfileId, tiflashProfileId string var region string if provider == "aws" { /*tikvStorage = AwsTikvStorage tiflashStorage = AwsTiflashStorage tidbProfileId = AwsTidbProfileId tikvProfileId = AwsTikvProfileId tiflashProfileId = AwsTiflashProfileId*/ region = dbassRequestInfo.CommonEnv.AwsRegion } else { /*tikvStorage = GcpTikvStorage tiflashStorage = GcpTiflashStorage tidbProfileId = GcpTidbProfileId tikvProfileId = GcpTikvProfileId tiflashProfileId = GcpTiflashProfileId*/ region = dbassRequestInfo.CommonEnv.GcpRegion } tidbProfileId, tikvProfileId, tiflashProfileId, tikvStorage, tiflashStorage = GetRestoreProfileId(t, orgId, projectId, clusterId, backupId.(string)) if tidbProfileId == "" || tikvProfileId == "" || tikvStorage == 0 { return } restoreBackup := NewRestoreBackup(isDevTier, cluster_name, port, restore_mode, restored_ts, region, tidb_count, tikv_count, tifish_count, tikvStorage, tiflashStorage, tidbProfileId, tikvProfileId, tiflashProfileId, region) restorePathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId.(string)} restoreBackupResp := dbassRequestInfo.DbassRequest(t, "restore", restoreBackup, restorePathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("restore the backup, body:%v, pathParam:%s, restoreBackupResp:%s", restoreBackup, restorePathParam, restoreBackupResp.JSON().Raw())), allure.Action(func() { restoreBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("cluster_id") })) l4g.Info("do restore, the body is: %s, pathParam is: %s, restoreBackupResp is:%s", restoreBackup, restorePathParam, restoreBackupResp.JSON().Raw()) backupClusterId := restoreBackupResp.JSON().Path("$.cluster_id").String().Raw() l4g.Info("get the backupClusterId, backupClusterId is %s", backupClusterId) if backupClusterId == "" { return } backupClusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": backupClusterId} for round := 0; round <= 30; round += 1 { l4g.Info("get cluster info, round is %d", round) time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, backupClusterPathParam, nil, nil) l4g.Info("get cluster info, getClusterInfoResp is: %s", getClusterInfoResp.Raw()) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("restore is success! ", round) break } } allure.Step(allure.Description(fmt.Sprintf("check restore the backup success, ExpectStatus:%s, ActualStatus:%s, pathParam:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw(), backupClusterPathParam)), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") //getClusterInfoResp.JSON().Path("$.region").String().Equal(dbassRequestInfo.CommonEnv.Region) getClusterInfoResp.JSON().Path("$.cluster.cluster_components.tikv.replicas").Equal(tikv_count) getClusterInfoResp.JSON().Path("$.cluster.cluster_components.tidb.replicas").Equal(tidb_count) if tifish_count != 0 { getClusterInfoResp.JSON().Path("$.cluster.cluster_components.tiflash.replicas").Equal(tifish_count) } getClusterInfoResp.JSON().Path("$.cluster.version").IsEqual(tidbVersion) })) time.Sleep(time.Second * 2 * 60) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { //更新密码 UpdatePassword(t, projectId, backupClusterId, dbassRequestInfo.CommonEnv.DbPassword) //创建filter AddTrafficFilter(t, orgId, projectId, backupClusterId, false) //获取cluster信息 getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, backupClusterPathParam, nil, nil) tidbAddress = getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort = int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) //check备份的数据是否正确 db, err = ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, ierr = SelectData(db, testUserSelect, true, 1, "autotest") value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check restore data correctness，result is:%v, "+ "connectTidb failed reason:%s, selectData failed reason:%s, tidbAddress:%s", result, err, ierr, tidbAddress)), allure.Action(func() { value.Equal(true) })) //删除备份集群 delClusterResp := dbassRequestInfo.DbassRequest(t, "delCluster", nil, backupClusterPathParam, nil, nil) l4g.Info("delete the backup cluster, delClusterResp is: %s", delClusterResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete cluster, pathParam:%s, delClusterResp:%s", backupClusterPathParam, delClusterResp.JSON().Raw())), allure.Action(func() { delClusterResp.Status(StatusOK) })) if delClusterResp.Raw().StatusCode == 200 { DeleteIdInMap(clusterId) } //删除备份 deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, restorePathParam, nil, nil) l4g.Info("delete the backup, deleteBackupResp is: %s", deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete backup, pathParam:%s, deleteBackupResp:%s", restorePathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) })) } l4g.Info("begin to maual backup!") backupName = GenTempName() manualBackup = NewManualBackup(backupName) manualBackupResp = dbassRequestInfo.DbassRequest(t, "manualBackup", manualBackup, clusterPathParam, nil, nil) l4g.Info("do maual backup, manualBackup is %+v, clusterPathParam is %+v, manualBackupResp is %s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do manaul backup, body:%+v, clusterPathParam:%+v, manualBackupResp:%s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw())), allure.Action(func() { manualBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("id") })) for round := 0; round <= 15*60; round++ { time.Sleep(time.Second * 1) l4g.Info("get backup list, round is %d", round) getBackupListResp = dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backup list, getBackupListResp is: %s", getBackupListResp.JSON().Raw()) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) == "RUNNING" || GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) == "SUCCESS" { l4g.Info("manual backup is running! ", round) backupId = GetBackupId(getBackupListResp, backupName, "id") break } } deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, restorePathParam, nil, nil) l4g.Info("delete the backup, deleteBackupResp is: %s", deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete backup, pathParam:%s, deleteBackupResp:%s", restorePathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) })) } func ServerlessBackupAndRestore(t require.TestingT, projectId, clusterId string) { dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } //获取cluster信息 clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) tidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) token := getClusterInfoResp.JSON().Path("$.cluster.endpoint.token").String().Raw() db, err := ConnectServerlessUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, token+".root") //port := dbassRequestInfo.CommonEnv.Port result, ierr := CreateTable(db, createTable) value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err, ierr, tidbAddress)), allure.Action(func() { value.Equal(true) })) var hour, min, backupName, backupId string var getBackupListResp *httpexpect.Response accountInfo := dbassRequestInfo.DbassRequest(t, "orgs", nil, nil, nil, nil) zone := GetOrgTimeZone(accountInfo, orgId) offset := strconv.Itoa(int(zone)) + "h" dur, _ := time.ParseDuration(offset) current := time.Now().UTC().Add(dur) if current.Hour() < 10 { hour = "0" + strconv.Itoa(current.Hour()) } else { hour = strconv.Itoa(current.Hour()) } if current.Minute()+2 < 10 { min = "0" + strconv.Itoa(current.Minute()+2) } else if current.Minute()+2 < 60 { min = strconv.Itoa(current.Minute() + 2) } else { min = strconv.Itoa(current.Minute() + 2 - 60) hour1, _ := strconv.Atoi(hour) hour = strconv.Itoa(hour1 + 1) } if min == "0" { min = "00" } data := fmt.Sprintf("%s:%s", hour, min) maxBackup := 7 autoBackup := NewAutoBackup(data, maxBackup) autoBackupResp := dbassRequestInfo.DbassRequest(t, "autoBackup", autoBackup, clusterPathParam, nil, nil) l4g.Info("do the auto backup, body is: %s, pathParam is: %s, autoBackupResp is: %s", autoBackup, clusterPathParam, autoBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("auto backup, body:%v, pathParam:%s, autoBackupBody:%s, autoBackupResp:%s", autoBackup, clusterPathParam, autoBackup, autoBackupResp.JSON().Raw())), allure.Action(func() { autoBackupResp.Status(StatusOK) })) //查询backup列表 for i := 0; i < 60; i++ { getBackupListResp = dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backup list, getBackupListResp is: %s", getBackupListResp.JSON().Raw()) backupId, backupName = GetLastBackupId(getBackupListResp) if backupId != "000000" { break } time.Sleep(time.Minute * 1) } if backupId == "000000" { allure.Step(allure.Description("can not find backup in this serverless tier"), allure.Action(func() { require.NotEqual(t, backupId, "000000") })) return } backupPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId} for round := 0; round <= 15; round++ { time.Sleep(time.Second * 60) l4g.Info("get backup list, round is %d", round) getBackupListResp = dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backup list, getBackupListResp is: %s", getBackupListResp.JSON().Raw()) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) == "SUCCESS" { l4g.Info("manual backup is success! ", round) break } } allure.Step(allure.Description(fmt.Sprintf("check manaul backup status is success, EXpectStatus:%s, ActualStatus:%s, backupId:%s", "SUCCESS", GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus), backupId)), allure.Action(func() { CheckBackupInfo(getBackupListResp, backupName, "status", "SUCCESS") })) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) != "SUCCESS" { return } if result { result, ierr = InsertDataToTest(db, insert) } time.Sleep(time.Second * 120) pitrRestoreTime := time.Now().Unix() //恢复备份 l4g.Info("begin to restore the backup !") restoreBackup := ServerlessRestoreBackup{"RESTORE_MODE_SNAPSHOT", ""} restoreBackupResp := dbassRequestInfo.DbassRequest(t, "restore", restoreBackup, backupPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("restore the backup, body:%v, pathParam:%s, restoreBackupResp:%s", restoreBackup, backupPathParam, restoreBackupResp.JSON().Raw())), allure.Action(func() { restoreBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("cluster_id") })) l4g.Info("do restore, the body is: %s, pathParam is: %s, restoreBackupResp is:%s", restoreBackup, backupPathParam, restoreBackupResp.JSON().Raw()) restoreId := restoreBackupResp.JSON().Path("$.id").String().Raw() l4g.Info("get the restoreId, restoreId is %s", restoreId) if restoreId == "" { return } for round := 0; round <= 30; round += 1 { l4g.Info("get restore progress, round is %d", round) time.Sleep(time.Second * 60) getRestoreProgressResp := dbassRequestInfo.DbassRequest(t, "getRestoreProgress", nil, clusterPathParam, nil, nil) l4g.Info("get restore progress, getRestoreProgressResp is: %s", getRestoreProgressResp.Raw()) if getRestoreProgressResp.JSON().Path("$.restore_status").String().Raw() == "RESTORE_STATUS_SUCCESS" { l4g.Info("restore is success! ", round) break } } for round := 0; round <= 3; round += 1 { l4g.Info("get backup list, round is %d", round) time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) l4g.Info("get cluster info, getClusterInfoResp is: %s", getClusterInfoResp.Raw()) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("restore is success! ", round) break } } allure.Step(allure.Description(fmt.Sprintf("check restore the backup success, ExpectStatus:%s, ActualStatus:%s, pathParam:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw(), clusterPathParam)), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") })) db, err = ConnectServerlessUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, token+".root") result, ierr = SelectData(db, testUserSelect, true, 1, "autotest") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check restore data correctness，result is:%v, "+ "connectTidb failed reason:%s, selectData failed reason:%s, tidbAddress:%s", result, err, ierr, tidbAddress)), allure.Action(func() { value.Equal(false) })) l4g.Info("begin to restore the backup !") getPitrRangeResp := dbassRequestInfo.DbassRequest(t, "getServerlessPitrRange", nil, clusterPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("get pitr range, body:%v, pathParam:%s, restoreBackupResp:%s", nil, clusterPathParam, getPitrRangeResp.JSON().Raw())), allure.Action(func() { getPitrRangeResp.Status(StatusOK) })) l4g.Info("getPitrRangeResp:%v", getPitrRangeResp.JSON().Raw()) restoreBackup = ServerlessRestoreBackup{"RESTORE_MODE_PITR", fmt.Sprintf("%d", pitrRestoreTime)} backupPathParam = map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": "0"} restoreBackupResp = dbassRequestInfo.DbassRequest(t, "restore", restoreBackup, backupPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("restore the backup, body:%v, pathParam:%s, restoreBackupResp:%s", restoreBackup, backupPathParam, restoreBackupResp.JSON().Raw())), allure.Action(func() { restoreBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("id") })) l4g.Info("do restore, the body is: %s, pathParam is: %s, restoreBackupResp is:%s", restoreBackup, backupPathParam, restoreBackupResp.JSON().Raw()) restoreId = restoreBackupResp.JSON().Path("$.id").String().Raw() l4g.Info("get the restoreId, restoreId is %s", restoreId) if restoreId == "" { return } for round := 0; round <= 30; round += 1 { l4g.Info("get restore progress, round is %d", round) time.Sleep(time.Second * 60) getRestoreProgressResp := dbassRequestInfo.DbassRequest(t, "getRestoreProgress", nil, clusterPathParam, nil, nil) l4g.Info("get restore progress, getRestoreProgressResp is: %s", getRestoreProgressResp.Raw()) if getRestoreProgressResp.JSON().Path("$.restore_status").String().Raw() == "RESTORE_STATUS_SUCCESS" { l4g.Info("restore is success! ", round) break } } for round := 0; round <= 3; round += 1 { l4g.Info("get backup list, round is %d", round) time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) l4g.Info("get cluster info, getClusterInfoResp is: %s", getClusterInfoResp.Raw()) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("restore is success! ", round) break } } allure.Step(allure.Description(fmt.Sprintf("check restore the backup success, ExpectStatus:%s, ActualStatus:%s, pathParam:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw(), clusterPathParam)), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") })) db, err = ConnectServerlessUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, token+".root") result, ierr = SelectData(db, testUserSelect, true, 1, "autotest") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check restore data correctness，result is:%v, "+ "connectTidb failed reason:%s, selectData failed reason:%s, tidbAddress:%s", result, err, ierr, tidbAddress)), allure.Action(func() { value.Equal(true) })) restoreHistory := dbassRequestInfo.DbassRequest(t, "getRestoreHistory", nil, clusterPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("get restore history, body:%v, pathParam:%s, restoreHistory:%s", restoreBackup, backupPathParam, restoreHistory.JSON().Raw())), allure.Action(func() { restoreHistory.Status(StatusOK).JSON().Path("$.total").Number().Equal(2) })) } func CheckAutoBackupOperation(t assert.TestingT, provider, projectId, clusterId string, isDevTier bool) { if clusterId == "" { return } var region string dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if provider == "aws" { region = dbassRequestInfo.CommonEnv.AwsRegion } else { region = dbassRequestInfo.CommonEnv.GcpRegion } clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} //自动备份 var hour, min string accountInfo := dbassRequestInfo.DbassRequest(t, "orgs", nil, nil, nil, nil) zone := GetOrgTimeZone(accountInfo, orgId) offset := strconv.Itoa(int(zone)) + "h" dur, _ := time.ParseDuration(offset) current := time.Now().UTC().Add(dur) if current.Hour() < 10 { hour = "0" + strconv.Itoa(current.Hour()) } else { hour = strconv.Itoa(current.Hour()) } if current.Minute()+2 < 10 { min = "0" + strconv.Itoa(current.Minute()+2) } else if current.Minute()+2 < 60 { min = strconv.Itoa(current.Minute() + 2) } else { min = strconv.Itoa(current.Minute() + 2 - 60) hour1, _ := strconv.Atoi(hour) hour = strconv.Itoa(hour1 + 1) } if min == "0" { min = "00" } data := fmt.Sprintf("%s:%s", hour, min) maxBackup := 7 if isDevTier { maxBackup = 1 } //autoBackup := NewAutoBackup(data, maxBackup) autoBackup := NewUpgradeBackupSetting(region, "", false, false, fmt.Sprintf("%d", maxBackup), "", data) autoBackupResp := dbassRequestInfo.DbassRequest(t, "backupSettingv2", autoBackup, clusterPathParam, nil, nil) l4g.Info("do the auto backup, body is: %s, pathParam is: %s, autoBackupResp is: %s", autoBackup, clusterPathParam, autoBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("auto backup, body:%v, pathParam:%s, autoBackupBody:%s, autoBackupResp:%s", autoBackup, clusterPathParam, autoBackup, autoBackupResp.JSON().Raw())), allure.Action(func() { autoBackupResp.Status(StatusOK) })) time.Sleep(time.Second * 180) //查询backup列表 getBackupListResp := dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backup list, getBackupListResp is: %s", getBackupListResp.JSON().Raw()) backupId, _ := GetLastBackupId(getBackupListResp) backupPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId} //删除自动备份 deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, backupPathParam, nil, nil) l4g.Info("delete auto backup, body is: %s, pathParam is:%s, deleteBackupResp is: %s", nil, backupPathParam, deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete auto backup, pathParam:%s, deleteBackupResp:%s", backupPathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) })) } func CheckLogBackup(t *testing.T, projectId, clusterId, provider string, isDevTier bool, tidb_count, tikv_count, tifish_count int) { r := require.New(t) var region string // success := false dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if provider == "aws" { region = dbassRequestInfo.CommonEnv.AwsRegion } else { region = dbassRequestInfo.CommonEnv.GcpRegion } if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if isDevTier { l4g.Info("%s/%s is dev tier, skip log backup test", projectId, clusterId) return } // check whether cluster is in white list /*isEnablePiTR := checkEnableDPFeature(t, FeaturePiTR, dbassRequestInfo, orgId, projectId, clusterId) if !isEnablePiTR { l4g.Info("%s/%s is not it in feature %s white list, skip dual region log backup test", projectId, clusterId, FeaturePiTR) return } l4g.Info("check %s/%s is in feature %s white list success", projectId, clusterId, FeaturePiTR)*/ // check log backup is in progress as normal result, err := DoSQLInCluster(t, projectId, clusterId, createTable, false, 0, "") /*value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) }))*/ result, err = DoSQLInCluster(t, projectId, clusterId, insert, false, 0, "") /*value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) }))*/ clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getBackupSettingResp := dbassRequestInfo.DbassRequest(t, "getBackupSettingV2", nil, clusterPathParam, nil, nil) l4g.Info("getBackupSettingResp is: %s, body is:%s, param is:%s", getBackupSettingResp.JSON().Raw(), nil, clusterPathParam) allure.Step( allure.Description(fmt.Sprintf("getBackupSettingResp is: %s, body is:%s, param is:%s", getBackupSettingResp.JSON().Raw(), nil, clusterPathParam)), allure.Action(func() { getBackupSettingResp.Status(http.StatusOK) })) backupSettingBody := NewUpgradeBackupSetting(region, "", false, true, "1", "", "18:00") backupSettingResp := dbassRequestInfo.DbassRequest(t, "backupSettingv2", backupSettingBody, clusterPathParam, nil, nil) l4g.Info("backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam) allure.Step( allure.Description(fmt.Sprintf("backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam)), allure.Action(func() { backupSettingResp.Status(http.StatusOK) })) //checkLogBackupProgressNormal(t, dbassRequestInfo, orgId, projectId, clusterId) // check run snapshot backup at the same time backupId := checkManmaualBackup(t, dbassRequestInfo, orgId, projectId, clusterId) // get pitr range var ( restoredTS int64 pitrRangeSuccess bool ) result, err = DoSQLInCluster(t, projectId, clusterId, insertforLog, false, 0, "") value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) })) insertTime := time.Now().Unix() //time.Sleep(time.Minute * 10) for round := 0; round <= 15; round++ { pitrRangeResp := dbassRequestInfo.DbassRequest(t, "listPiTRTimeRanges", nil, clusterPathParam, nil, nil, UseRequireReporter(t)) l4g.Info("%s/%s list pitr time ranges, round is %d, pitrRangeResp is: %s", projectId, clusterId, round, pitrRangeResp.JSON().Raw()) allure.Step( allure.Description(fmt.Sprintf("check %s/%s listPiTRTimeRanges, round is %d, response is: %s", projectId, clusterId, round, pitrRangeResp.JSON().Raw())), allure.Action(func() { pitrRangeResp.Status(http.StatusOK) })) timeRange, err := GetLatestPiTRTimeRange(pitrRangeResp) if err != nil { l4g.Error("get %s/%s pitr time range failed, round is %d, err: %s", projectId, clusterId, round, err.Error()) } else if time.Time(timeRange.EndTs).Unix() < insertTime { l4g.Info("get %s/%s pitr ranges time < insert data time, response is : %s", projectId, clusterId, round, pitrRangeResp.JSON().Raw()) time.Sleep(time.Second * 60) continue } else { pitrRangeSuccess = true restoredTS = time.Time(timeRange.EndTs).Unix() break } time.Sleep(time.Second * 60) } allure.Step( allure.Description(fmt.Sprintf("%s/%s check list pitr time ranges whether is success", projectId, clusterId)), allure.Action(func() { r.True(pitrRangeSuccess) })) if time.Now().Unix()-restoredTS > 5*60 { allure.Step( allure.Description(fmt.Sprintf("%s/%s check time point is running, result:false", projectId, clusterId)), allure.Action(func() { r.True(false) })) } // execute pitr restoreClusterId := doRestore(t, dbassRequestInfo, orgId, projectId, clusterId, provider, restoredTS, region, isDevTier, tidb_count, tikv_count, tifish_count, region, backupId) checkPitrRestore(t, dbassRequestInfo, orgId, projectId, clusterId, restoreClusterId, restoredTS, region) //delete manual backup deletePathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId} deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, deletePathParam, nil, nil, UseRequireReporter(t)) l4g.Info("delete the backup, deleteBackupResp is: %s", deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete backup, pathParam:%s, deleteBackupResp:%s", deletePathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) })) l4g.Info("check %s/%s log backup operation success", projectId, clusterId) result, err = DoSQLInCluster(t, projectId, clusterId, deleteforLog, false, 0, "") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, delete data failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) })) } func CheckDualRegionBackup(t *testing.T, projectId, clusterId, provider string, isDevTier bool, tidb_count, tikv_count, tifish_count int) { l4g.Info("check %s/%s dual region backup", projectId, clusterId) var region string r := require.New(t) dbassRequestInfo = GetDbassRequestInfo(t) if provider == "aws" { region = dbassRequestInfo.CommonEnv.AwsRegion } else { region = dbassRequestInfo.CommonEnv.GcpRegion } orgId := dbassRequestInfo.CommonEnv.OrgId var err error if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if isDevTier { l4g.Info("%s/%s is dev tier, skip dual region backup test", projectId, clusterId) return } // check whether cluster is in white list /*isEnableDualRegion := checkEnableDPFeature(t, FeatureDualRegion, dbassRequestInfo, orgId, projectId, clusterId) if !isEnableDualRegion { l4g.Info("%s/%s is not it in feature %s white list, skip dual region log backup test", projectId, clusterId, FeatureDualRegion) return } l4g.Info("check %s/%s is in feature %s white list success", projectId, clusterId, FeatureDualRegion)*/ // check log backup is in progress as normal // check whether gcp cluster /*l4g.Info("check %s/%s is gcp cluster", projectId, clusterId) clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) l4g.Info("cluster resp is %s", getClusterInfoResp.JSON().Raw()) providerRes := getClusterInfoResp.JSON().Path("$.cluster.provider").String().Raw() if providerRes != "gcp" { l4g.Info("%s/%s is not gcp cluster, skip dual region log backup test", projectId, clusterId) return } l4g.Info("check %s/%s is gcp cluster success", projectId, clusterId)*/ // get optional dual region clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} l4g.Info("check %s/%s get optional dual region", projectId, clusterId) backupOptionsRes := dbassRequestInfo.DbassRequest(t, "getBackupOptions", nil, clusterPathParam, nil, nil) checkDesc := fmt.Sprintf("%s/%s getBackupOptions check, response is:%s", projectId, clusterId, backupOptionsRes.JSON().Raw()) allure.Step( allure.Description(checkDesc), allure.Action(func() { require.Equal(t, StatusOK, backupOptionsRes.Raw().StatusCode, checkDesc) })) /*if strings.EqualFold(env, "dev") { l4g.Info("it is in %s env, just test basic interface getBackupOptions, skip dual region log backup test", projectId, clusterId, env) return }*/ clusterRegion, backupOptinal := GetOptionalDualRegion(backupOptionsRes) checkDesc = fmt.Sprintf("%s/%s getOptionalDualRegion check", projectId, clusterId) allure.Step( allure.Description(checkDesc), allure.Action(func() { require.NotNil(t, backupOptinal, checkDesc) })) l4g.Info("%s/%s get optional dual region success, clusterRegion is %s, backupOptinalRegion is %s, continent is %s", projectId, clusterId, clusterRegion, backupOptinal.RegionOptions[0].Region.Name, backupOptinal.RegionOptions[0].Continent) result, err := DoSQLInCluster(t, projectId, clusterId, createTable, false, 0, "") l4g.Info("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err) /*value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) }))*/ result, err = DoSQLInCluster(t, projectId, clusterId, insert, false, 0, "") l4g.Info("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err) /*value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) }))*/ var prePitrFlag bool var getBackupSettingResp *httpexpect.Response for i := 0; i <= 10; i++ { getBackupSettingResp = dbassRequestInfo.DbassRequest(t, "getBackupSettingV2", nil, clusterPathParam, nil, nil) if getBackupSettingResp.JSON().Path("$.status").String().Raw() == "Normal" { break } time.Sleep(time.Second * 60) } l4g.Info("getBackupSettingResp is: %s, body is:%s, param is:%s", getBackupSettingResp.JSON().Raw(), nil, clusterPathParam) allure.Step( allure.Description(fmt.Sprintf("getBackupSettingResp is: %s, body is:%s, param is:%s", getBackupSettingResp.JSON().Raw(), nil, clusterPathParam)), allure.Action(func() { getBackupSettingResp.Status(http.StatusOK) prePitrFlag = getBackupSettingResp.JSON().Path("$.continuous_backup_schedule.enable").Boolean().Raw() })) backupSettingBody := NewUpgradeBackupSetting(region, backupOptinal.RegionOptions[0].Region.Name, false, prePitrFlag, "1", "", "18:00") backupSettingResp := dbassRequestInfo.DbassRequest(t, "backupSettingv2", backupSettingBody, clusterPathParam, nil, nil) l4g.Info("backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam) allure.Step( allure.Description(fmt.Sprintf("open dual region, backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam)), allure.Action(func() { backupSettingResp.Status(http.StatusOK) })) time.Sleep(time.Minute * 10) backupSettingBody = NewUpgradeBackupSetting(region, backupOptinal.RegionOptions[0].Region.Name, false, true, "1", "", "18:00") backupSettingResp = dbassRequestInfo.DbassRequest(t, "backupSettingv2", backupSettingBody, clusterPathParam, nil, nil) l4g.Info("open pitr, backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam) allure.Step( allure.Description(fmt.Sprintf("backupSettingResp is: %s, body is:%s, param is:%s", backupSettingResp.JSON().Raw(), backupSettingBody, clusterPathParam)), allure.Action(func() { backupSettingResp.Status(http.StatusOK) })) // check log backup progress is normal //checkLogBackupProgressNormal(t, dbassRequestInfo, orgId, projectId, clusterId) // run snapshot backup at the same time //backupId0 := checkManmaualBackup(t, dbassRequestInfo, orgId, projectId, clusterId) // set dual region /*l4g.Info("%s/%s set dual region, regions is %s and %v", projectId, clusterId, clusterRegion, backupOptinal) dualRegionBackup := NewDualRegionBackup([]string{clusterRegion, backupOptinal.RegionOptions[0].Region.Name}, "") modifyBackupSettingRes := dbassRequestInfo.DbassRequest(t, "modifyBackupSetting", dualRegionBackup, clusterPathParam, nil, nil) checkDesc = fmt.Sprintf("%s/%s modifyBackupSetting check, response is:%s", projectId, clusterId, modifyBackupSettingRes.JSON().Raw()) allure.Step( allure.Description(checkDesc), allure.Action(func() { require.Equal(t, StatusOK, modifyBackupSettingRes.Raw().StatusCode, checkDesc) })) backupSetting := GetDualRegionSeting(modifyBackupSettingRes) checkDesc = fmt.Sprintf("%s/%s backupSetting check", projectId, clusterId) allure.Step( allure.Description(checkDesc), allure.Action(func() { require.NotNil(t, backupSetting, checkDesc) require.GreaterOrEqual(t, len(backupSetting.Regions), 2) })) l4g.Info("%s/%s set dual region success, regions is %v", projectId, clusterId, backupSetting.Regions)*/ // run snapshot backup at the same time backupId1 := checkManmaualBackup(t, dbassRequestInfo, orgId, projectId, clusterId) result, err = DoSQLInCluster(t, projectId, clusterId, insertforLog, false, 0, "") value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, insertTidb failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) })) insertTime := time.Now().Unix() // check log backup progress is normal //checkLogBackupProgressNormal(t, dbassRequestInfo, orgId, projectId, clusterId) // get restore region and times l4g.Info("%s/%s list pitr region and time ranges", projectId, clusterId) var ( ranges *models.DataflowPiTRTimeRangeListResp pitrRangeSuccess bool ) for round := 0; round <= 15; round++ { pitrRangeResp := dbassRequestInfo.DbassRequest(t, "listPiTRTimeRanges", nil, clusterPathParam, nil, nil, UseRequireReporter(t)) checkDesc = fmt.Sprintf("check %s/%s listPiTRTimeRanges, round is %d, response is: %s", projectId, clusterId, round, pitrRangeResp.JSON().Raw()) allure.Step( allure.Description(checkDesc), allure.Action(func() { pitrRangeResp.Status(http.StatusOK) })) ranges, err = GetPiTRRanges(pitrRangeResp) if err != nil { l4g.Error("get %s/%s pitr time range failed, round is %d, err: %s", projectId, clusterId, round, err.Error()) time.Sleep(time.Second * 60) continue } if len(ranges.Items) < 2 { l4g.Info("get %s/%s pitr ranges size < 2, response is : %s", projectId, clusterId, round, pitrRangeResp.JSON().Raw()) time.Sleep(time.Second * 60) continue } if time.Time(ranges.Items[0].TimeRanges[0].EndTs).Unix() < insertTime || time.Time(ranges.Items[1].TimeRanges[0].EndTs).Unix() < insertTime { l4g.Info("get %s/%s pitr ranges time < insert data time, response is : %s", projectId, clusterId, round, pitrRangeResp.JSON().Raw()) time.Sleep(time.Second * 60) continue } pitrRangeSuccess = true break } checkDesc = fmt.Sprintf("%s/%s check list pitr region and time ranges whether is success", projectId, clusterId) allure.Step( allure.Description(checkDesc), allure.Action(func() { require.True(t, pitrRangeSuccess, checkDesc) })) l4g.Info("%s/%s list pitr region and time ranges success, ranges is %v", projectId, clusterId, ranges) l4g.Info("%s/%s check dual region pitr restore to ranges %v", projectId, clusterId, ranges) region1 := ranges.Items[0].Region.Name region2 := ranges.Items[1].Region.Name //region1StartRestoredTS := time.Time(ranges.Items[0].TimeRanges[0].StartTs).Unix() region1EndRestoredTS := time.Time(ranges.Items[0].TimeRanges[0].EndTs).Unix() //region2StartRestoredTS := time.Time(ranges.Items[1].TimeRanges[0].StartTs).Unix() region2EndRestoredTS := time.Time(ranges.Items[1].TimeRanges[0].EndTs).Unix() if time.Now().Unix()-region2EndRestoredTS > 5*60 { allure.Step( allure.Description(fmt.Sprintf("%s/%s check time point is running, result:false", projectId, clusterId)), allure.Action(func() { r.True(false) })) } // restore to the time range after switch bucket restoreClusterId1 := doRestore(t, dbassRequestInfo, orgId, projectId, clusterId, provider, region1EndRestoredTS, region1, isDevTier, tidb_count, tikv_count, tifish_count, region1, backupId1) restoreClusterId2 := doRestore(t, dbassRequestInfo, orgId, projectId, clusterId, provider, region2EndRestoredTS, region2, isDevTier, tidb_count, tikv_count, tifish_count, region2, backupId1) checkPitrRestore(t, dbassRequestInfo, orgId, projectId, clusterId, restoreClusterId1, region1EndRestoredTS, region1) checkPitrRestore(t, dbassRequestInfo, orgId, projectId, clusterId, restoreClusterId2, region2EndRestoredTS, region2) var restoreClusterId3 string if provider == "aws" { restoreClusterId3 = doRestore(t, dbassRequestInfo, orgId, projectId, clusterId, provider, region2EndRestoredTS, region1, isDevTier, tidb_count, tikv_count, tifish_count, region2, backupId1) checkPitrRestore(t, dbassRequestInfo, orgId, projectId, clusterId, restoreClusterId3, region2EndRestoredTS, region1) } // restore to the time range before swiStartTstch bucket //checkPitrRestore(t, dbassRequestInfo, orgId, projectId, clusterId, provider, region1StartRestoredTS, region1, isDevTier, tidb_count, tikv_count, tifish_count) //checkPitrRestore(t, dbassRequestInfo, orgId, projectId, clusterId, provider, region2StartRestoredTS, region2, isDevTier, tidb_count, tikv_count, tifish_count) //l4g.Info("%s/%s check dual region pitr restore to ranges %v success", projectId, clusterId, ranges) //delete manual backup /*deletePathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId0} deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, deletePathParam, nil, nil, UseRequireReporter(t)) l4g.Info("delete the backup, deleteBackupResp is: %s", deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete backup, pathParam:%s, deleteBackupResp:%s", deletePathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) }))*/ //delete manual backup deletePathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId1} deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, deletePathParam, nil, nil, UseRequireReporter(t)) l4g.Info("delete the backup, deleteBackupResp is: %s", deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete backup, pathParam:%s, deleteBackupResp:%s", deletePathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) })) l4g.Info("check %s/%s dual region backup success", projectId, clusterId) result, err = DoSQLInCluster(t, projectId, clusterId, deleteforLog, false, 0, "") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, ConnectTidb failed reason:%s, delete data failed reason:%s, tidbAddress:%s", result, err)), allure.Action(func() { value.Equal(true) })) } func checkEnableDPFeature(t *testing.T, featureName string, dbassRequestInfo *DbassRequestInfo, orgId, projectId, clusterId string) bool { l4g.Info("check %s/%s is it in feature %s white list", projectId, clusterId, featureName) clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} enableFeatureResp := dbassRequestInfo.DbassRequest(t, "getDPEnableFeautres", nil, clusterPathParam, nil, nil) checkDesc := fmt.Sprintf("%s/%s getDPEnableFeautres check, response is:%s", projectId, clusterId, enableFeatureResp.JSON().Raw()) allure.Step( allure.Description(checkDesc), allure.Action(func() { require.Equal(t, StatusOK, enableFeatureResp.Raw().StatusCode, checkDesc) })) return IsEnableDPFeature(enableFeatureResp, featureName) } func checkLogBackupProgressNormal(t *testing.T, dbassRequestInfo *DbassRequestInfo, orgId, projectId, clusterId string) { r := require.New(t) success := false clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} // check log backup is in progress as normal l4g.Info("check %s/%s log backup is in progress as normal", projectId, clusterId) var logBackup0 *models.DataflowLogBackup var logBackup1 *models.DataflowLogBackup for round := 0; round <= 30; round += 1 { logBackupListResp := dbassRequestInfo.DbassRequest(t, "getLogBackupList", nil, clusterPathParam, nil, nil, UseRequireReporter(t)) allure.Step( allure.Description(fmt.Sprintf("check %s/%s getLogBackupList, round is %d, response is:%s", projectId, clusterId, round, logBackupListResp.JSON().Raw())), allure.Action(func() { logBackupListResp.Status(http.StatusOK) })) logBackup1 = GetRunningLogBackup(logBackupListResp) if logBackup1 == nil { success = false l4g.Info("%s/%s get running log backup failed, round is %d", projectId, clusterId, round) time.Sleep(time.Second * 30) continue } if logBackup0 == nil { logBackup0 = logBackup1 l4g.Info("%s/%s get running log backup0 success, round is %d", projectId, clusterId, round) time.Sleep(time.Second * 30) continue } if !logBackup0.CheckPointTs.Equal(logBackup1.CheckPointTs) { success = true l4g.Info("%s/%s log backup checkpoint ts updated, checkpointTs0 is %s, checkpoinTs1 is %s, round is %d", projectId, clusterId, logBackup0.CheckPointTs.String(), logBackup1.CheckPointTs.String(), round) break } else { l4g.Info("%s/%s wait log backup checkpoint ts update, checkpointTs is %s, round is %d", projectId, clusterId, logBackup0.CheckPointTs.String(), round) time.Sleep(time.Second * 30) } } allure.Step( allure.Description(fmt.Sprintf("check %s/%s log backup is in progress", projectId, clusterId)), allure.Action(func() { r.True(success) })) l4g.Info("check %s/%s log backup is in progress success", projectId, clusterId) } func checkManmaualBackup(t *testing.T, dbassRequestInfo *DbassRequestInfo, orgId, projectId, clusterId string) string { r := require.New(t) success := false clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} // check run snapshot backup at the same time l4g.Info("check %s/%s can run snapshot backup", projectId, clusterId) l4g.Info("begin to maual backup!") backupName := GenTempName() manualBackup := NewManualBackup(backupName) manualBackupResp := dbassRequestInfo.DbassRequest(t, "manualBackup", manualBackup, clusterPathParam, nil, nil, UseRequireReporter(t)) l4g.Info("do maual backup, manualBackup is %+v, clusterPathParam is %+v, manualBackupResp is %s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do manaul backup, body:%+v, clusterPathParam:%+v, manualBackupResp:%s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw())), allure.Action(func() { manualBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("id") })) //查询backup列表 var backupId string for round := 0; round <= 15; round++ { l4g.Info("%s/%s get backup list, round is %d", projectId, clusterId, round) getBackupListResp := dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("%s/%s get backup list, getBackupListResp is: %s", projectId, clusterId, getBackupListResp.JSON().Raw()) status, ok := GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) if ok && status == "SUCCESS" { backupId = GetBackupId(getBackupListResp, backupName, "id").(string) l4g.Info("manual backup %s is success! backupId is %s, round is %d", backupName, backupId, round) success = true break } l4g.Info("wait manual backup %s complete, backupId is %s, round is %d", backupName, backupId, round) time.Sleep(time.Second * 60) } allure.Step( allure.Description(fmt.Sprintf("%s/%s check maual snapshot backup is success", projectId, clusterId)), allure.Action(func() { r.True(success) })) l4g.Info("check %s/%s can run snapshot backup at the same time success", projectId, clusterId) return backupId } func doRestore(t *testing.T, dbassRequestInfo *DbassRequestInfo, orgId, projectId, clusterId, provider string, restoredTS int64, region string, isDevTier bool, tidb_count, tikv_count, tifish_count int, backupRegion, backupId string) string { l4g.Info("%s/%s check pitr restore, restoredTS is %d, region is %s", projectId, clusterId, restoredTS, region) l4g.Info("begin to execute pitr!") port := dbassRequestInfo.CommonEnv.Port var tikvStorage, tiflashStorage int var tidbProfileId, tikvProfileId, tiflashProfileId string if provider == "aws" { //tikvStorage = AwsTikvStorage //tiflashStorage = AwsTiflashStorage //tidbProfileId = AwsTidbProfileId //tikvProfileId = AwsTikvProfileId //tiflashProfileId = AwsTiflashProfileId } else { //tikvStorage = GcpTikvStorage //tiflashStorage = GcpTiflashStorage //tidbProfileId = GcpTidbProfileId //tikvProfileId = GcpTikvProfileId //tiflashProfileId = GcpTiflashProfileId } tidbProfileId, tikvProfileId, tiflashProfileId, tikvStorage, tiflashStorage = GetRestoreProfileId(t, orgId, projectId, clusterId, backupId) if tidbProfileId == "" || tikvProfileId == "" || tikvStorage == 0 { return "" } restoreBackup := NewRestoreBackup(isDevTier, "ScentAutoPiTR-"+GenTempName(), port, RestoreModePiTR, restoredTS, region, tidb_count, tikv_count, tifish_count, tikvStorage, tiflashStorage, tidbProfileId, tikvProfileId, tiflashProfileId, backupRegion) restorePathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": GenTempName()} restoreBackupResp := dbassRequestInfo.DbassRequest(t, "restore", restoreBackup, restorePathParam, nil, nil, UseRequireReporter(t)) allure.Step( allure.Description(fmt.Sprintf("restore the backup, body:%v, pathParam:%s, restoreBackupResp:%s", restoreBackup, restorePathParam, restoreBackupResp.JSON().Raw())), allure.Action(func() { restoreBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("cluster_id") })) l4g.Info("do restore, the body is: %s, pathParam is: %s, restoreBackupResp is:%s", restoreBackup, restorePathParam, restoreBackupResp.JSON().Raw()) restoreClusterID := restoreBackupResp.JSON().Path("$.cluster_id").String().Raw() return restoreClusterID } func checkPitrRestore(t *testing.T, dbassRequestInfo *DbassRequestInfo, orgId, projectId, clusterId, restoreClusterID string, restoredTS int64, region string) { r := require.New(t) l4g.Info("%s/%s check pitr restore, restoredTS is %d, region is %s", projectId, clusterId, restoredTS, region) l4g.Info("begin to execute pitr!") DeleteClusterIdList = append(DeleteClusterIdList, map[string]string{"projectId": projectId, "clusterId": restoreClusterID, "delete": "0"}) l4g.Info("get the restoreClusterID: %s", restoreClusterID) l4g.Info("begin to wait restored cluster recovering!") restoreClusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": restoreClusterID} restoreSuccess, restoreStarted := false, false for round := 0; round <= 60; round += 1 { l4g.Info("get restore cluster info, round is %d", round) time.Sleep(time.Second * 30) getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, restoreClusterPathParam, nil, nil, UseRequireReporter(t)) l4g.Info("get restore cluster info, getClusterInfoResp is: %s", getClusterInfoResp.JSON().Raw()) allure.Step( allure.Description(fmt.Sprintf("get restore cluster info, pathParam:%s, getClusterInfoResp:%s", restoreClusterPathParam, getClusterInfoResp.JSON().Raw())), allure.Action(func() { getClusterInfoResp.Status(StatusOK) })) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "RECOVERING" { l4g.Info("restore started!", round) restoreStarted = true break } // if restore process is too short, we may not watch the RECOVERING status if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("restore is success!", round) restoreStarted = true restoreSuccess = true break } } allure.Step( allure.Description(fmt.Sprintf("%s/%s check pitr has started", projectId, clusterId)), allure.Action(func() { r.True(restoreStarted) })) if !restoreSuccess { /*l4g.Info("begin to wait restore progress to 100") restoreFinished := false for round := 0; round <= 20; round += 1 { l4g.Info("get restore progress, round is %d", round) time.Sleep(time.Second * 30) restoreProgressResp := dbassRequestInfo.DbassRequest(t, "getRestoreProgress", nil, restoreClusterPathParam, nil, nil, UseRequireReporter(t)) l4g.Info("get restore progress, restoreProgressResp is: %s", restoreProgressResp.JSON().Raw()) allure.Step( allure.Description(fmt.Sprintf("get restore progress, pathParam:%s, restoreProgressResp:%s", restoreClusterPathParam, restoreProgressResp.JSON().Raw())), allure.Action(func() { restoreProgressResp.Status(StatusOK) })) if restoreProgressResp.JSON().Path("$.progress").Number().Raw() == 100.0 { l4g.Info("restore finished", round) restoreFinished = true break } } allure.Step( allure.Description(fmt.Sprintf("%s/%s check restore has finished", projectId, clusterId)), allure.Action(func() { r.True(restoreFinished) }))*/ l4g.Info("begin to wait restore cluster normal") for round := 0; round <= 30; round += 1 { l4g.Info("get restore cluster info, round is %d", round) time.Sleep(time.Second * 30) getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, restoreClusterPathParam, nil, nil, UseRequireReporter(t)) l4g.Info("get restore cluster info, getClusterInfoResp is: %s", getClusterInfoResp.JSON().Raw()) allure.Step( allure.Description(fmt.Sprintf("get restore cluster info, pathParam:%s, getClusterInfoResp:%s", restoreClusterPathParam, getClusterInfoResp.JSON().Raw())), allure.Action(func() { getClusterInfoResp.Status(StatusOK) })) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("restore is success!", round) restoreSuccess = true break } } } allure.Step( allure.Description(fmt.Sprintf("%s/%s check execute pitr whether is success", projectId, clusterId)), allure.Action(func() { r.True(restoreSuccess) })) UpdatePassword(t, projectId, restoreClusterID, dbassRequestInfo.CommonEnv.DbPassword) AddTrafficFilter(t, orgId, projectId, restoreClusterID, false) result1, err1 := DoSQLInCluster(t, projectId, restoreClusterID, testUserSelect, true, 2, "logtest") result2, err2 := DoSQLInCluster(t, projectId, restoreClusterID, testUserSelect, true, 1, "autotest") value := httpexpect.NewBoolean(t, result1 && result2) allure.Step(allure.Description(fmt.Sprintf("standard connect tidb result:%s, select1 failed reason:%s, select1 failed reason:%s", result1 && result2, err1, err2)), allure.Action(func() { value.Equal(true) })) // delete cluster CheckDeleteCluster(t, restoreClusterID, projectId, false) l4g.Info("execute pitr success!") } func GetRestoreProfileId(t assert.TestingT, orgId, projectId, clusterId, backupId string) (string, string, string, int, int) { var tidbProfileId, tikvProfileId, tiflashProfileId string var tikvStorage, tiflashStorage int clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getBackupsResp := dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) l4g.Info("get backups list, body is: %+v, pathParam is: %s, getBackupsResp is %s", nil, clusterPathParam, getBackupsResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("get backups list, pathParam: %s, getBackupsResp:%s", clusterPathParam, getBackupsResp.JSON().Raw())), allure.Action(func() { getBackupsResp.Status(StatusOK) })) allure.Step(allure.Description(fmt.Sprintf("get backups profile id")), allure.Action(func() { tidbProfileId, tikvProfileId, tiflashProfileId, tikvStorage, tiflashStorage = GetRecycleBinProfile(getBackupsResp, backupId) value := httpexpect.NewString(t, tidbProfileId) value.NotEmpty() value = httpexpect.NewString(t, tikvProfileId) value.NotEmpty() num := httpexpect.NewNumber(t, float64(tikvStorage)) num.NotEqual(0) })) return tidbProfileId, tikvProfileId, tiflashProfileId, tikvStorage, tiflashStorage } // 主动备份与RecycleBin func CheckRecycleBin(t assert.TestingT, projectId, clusterId, provider string, isDevTier bool, tidb_count, tikv_count, tifish_count int) string { if clusterId == "" { return "" } dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } pathparam := map[string]string{"org_id": orgId, "project_id": projectId} port := dbassRequestInfo.CommonEnv.Port //查看集群的详细信息 clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) clusterName := getClusterInfoResp.JSON().Path("$.cluster.name").String().Raw() //主动备份集群 backupName := GenTempName() manualBackup := NewManualBackup(backupName) manualBackupResp := dbassRequestInfo.DbassRequest(t, "manualBackup", manualBackup, clusterPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("manual backup, body:%v, pathParam:%s, manualBackupResp:%s", manualBackup, clusterPathParam, manualBackupResp.JSON().Raw())), allure.Action(func() { manualBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("id") })) if manualBackupResp.Raw().StatusCode != StatusOK { return "" } //查询backup列表 getBackupListResp := dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) CheckBackupInfo(getBackupListResp, backupName, "name", backupName) backupId := GetBackupId(getBackupListResp, backupName, "id") for round := 0; round <= 10; round++ { time.Sleep(time.Second * 600) getBackupListResp = dbassRequestInfo.DbassRequest(t, "getBackupList", nil, clusterPathParam, nil, nil) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) == "SUCCESS" { break } } allure.Step(allure.Description(fmt.Sprintf("check manual backup status, expectVaule is:%s, actualValue is:%s, backupId:%s", "SUCCESS", GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus), backupId)), allure.Action(func() { CheckBackupInfo(getBackupListResp, backupName, "status", "SUCCESS") })) if GetBackupId(getBackupListResp, backupName, "status").(models.CentralBackupStatus) != "SUCCESS" { return "" } //删除集群 delClusterResp := dbassRequestInfo.DbassRequest(t, "delCluster", nil, clusterPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("delete cluster, pathParam:%s, delClusterResp:%s", clusterPathParam, delClusterResp.JSON().Raw())), allure.Action(func() { delClusterResp.Status(StatusOK) })) if delClusterResp.Raw().StatusCode == 200 { DeleteIdInMap(clusterId) } //查看recycle bin列表 param := map[string]string{"provider": provider, "page": "1", "per_page": "100"} allure.Step(allure.Description(fmt.Sprintf("check recycle bin list")), allure.Action(func() { CheckRecycleBinList(t, pathparam, param, clusterName) })) //restore /* //restore - add offline scale up only in staging l4g.Info("begin to restore the backup and scale up!") //取region var region string if provider == "aws" { region = dbassRequestInfo.CommonEnv.AwsRegion } else { region = dbassRequestInfo.CommonEnv.GcpRegion } tidbCpu := dbassRequestInfo.CommonEnv.TidbCpu tikvCpu := dbassRequestInfo.CommonEnv.TikvCpu tiflashCpu := dbassRequestInfo.CommonEnv.TiflashCpu //由于recycle bin的测试是最后一步测试，生成的集群会立刻被删掉，目前写死成比较大的配置数，避免这边 api 请求的测试时间拉长。 tidbCount_int := 1 tikvCount_int := 6 tiflashCount_int := 2 */ var restoreBackup *RestoreBackup // if Envn != "staging" || isDevTier { // //仅在 staging 做 offline scale up 测试，以减少生产环境花销 // restoreBackup = NewRestoreBackup(isDevTier, dbassRequestInfo.CommonEnv.DbPassword, GenTempName(), []ComponentData{}, port) // } else { // //取更大的 cpuNum 配置的 profileId // profileBody := NewNodeProfilesRequest(isDevTier, orgId, provider, region) // getProfilesResp := dbassRequestInfo.DbassRequest(t, "getProfiles", profileBody, nil, nil, nil) // tikvStorage, tiflashStorage, tidbProfileId, tikvProfileId, tiflashProfileId := GetScaleUpProfileIdAndStorage(getProfilesResp, tidbCpu, tikvCpu, tiflashCpu, isDevTier) // l4g.Info("tikvStorage:%d, tiflashStorage:%d, tidbProfileId:%s, tikvProfileId:%s, tiflashProfileId:%s", tikvStorage, // tiflashStorage, tidbProfileId, tikvProfileId, tiflashProfileId) // //构建[]componentdata // tidbComponentType := NewComponentData(tidbType, tidbProfileId, tidbCount_int, 0) // tikvComponentType := NewComponentData(tikvType, tikvProfileId, tikvCount_int, tikvStorage) // tiflashComponentType := NewComponentData(tiflashType, tiflashProfileId, tiflashCount_int, tiflashStorage) // components := []ComponentData{tidbComponentType, tikvComponentType, tiflashComponentType} // //构建restore backup // restoreBackup = NewRestoreBackup(isDevTier, dbassRequestInfo.CommonEnv.DbPassword, GenTempName(), components, port) // } var tikvStorage, tiflashStorage int var tidbProfileId, tikvProfileId, tiflashProfileId, region string if provider == "aws" { //tikvStorage = AwsTikvStorage //tiflashStorage = AwsTiflashStorage //tidbProfileId = AwsTidbProfileId //tikvProfileId = AwsTikvProfileId //tiflashProfileId = AwsTiflashProfileId region = dbassRequestInfo.CommonEnv.AwsRegion } else { //tikvStorage = GcpTikvStorage //tiflashStorage = GcpTiflashStorage //tidbProfileId = GcpTidbProfileId //tikvProfileId = GcpTikvProfileId //tiflashProfileId = GcpTiflashProfileId region = dbassRequestInfo.CommonEnv.GcpRegion } tidbProfileId, tikvProfileId, tiflashProfileId, tikvStorage, tiflashStorage = GetRestoreProfileId(t, orgId, projectId, clusterId, backupId.(string)) if tidbProfileId == "" || tikvProfileId == "" || tikvStorage == 0 { return "" } restoreBackup = NewRestoreBackup(isDevTier, "ScentAuto-"+GenTempName(), port, RestoreModeSnapshot, 0, region, tidb_count, tikv_count, tifish_count, tikvStorage, tiflashStorage, tidbProfileId, tikvProfileId, tiflashProfileId, region) restorePathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": backupId.(string)} restoreBackupResp := dbassRequestInfo.DbassRequest(t, "restore", restoreBackup, restorePathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("restore from recycle bin, body:%v, pathParam:%s, restoreBackupResp:%s", restoreBackup, restorePathParam, restoreBackupResp.JSON().Raw())), allure.Action(func() { restoreBackupResp.Status(StatusOK).JSON().Object().Keys().Contains("cluster_id") })) l4g.Info("do restore, the body is: %s, pathParam is: %s, restoreBackupResp is:%s", restoreBackup, restorePathParam, restoreBackupResp.JSON().Raw()) backupClusterId := restoreBackupResp.JSON().Path("$.cluster_id").String().Raw() l4g.Info("get the backupClusterId, backupClusterId is %s", backupClusterId) backupClusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": backupClusterId} for round := 0; round <= 30; round += 1 { l4g.Info("get restore cluster, round is %d", round) time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, backupClusterPathParam, nil, nil) l4g.Info("get cluster info, getClusterInfoResp is: %s", getClusterInfoResp.JSON().Raw()) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("restore is success! ", round) break } } allure.Step(allure.Description(fmt.Sprintf("check restore cluster status, expectStatus is:%s, actualStatus is:%s, pathParam:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw(), backupClusterPathParam)), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") })) //校验恢复的集群的计费 //BillingCompute(t, clusterId, price) //删除备份集群 if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { /*delClusterResp = dbassRequestInfo.DbassRequest(t, "delCluster", nil, backupClusterPathParam, nil, nil) l4g.Info("delete the backup cluster, delClusterResp is: %s", delClusterResp.Raw()) allure.Step(allure.Description(fmt.Sprintf("delete the backup cluster, pathParam:%s, delClusterResp:%s", backupClusterPathParam, delClusterResp.JSON().Raw())), allure.Action(func() { delClusterResp.Status(StatusOK) })) if delClusterResp.Raw().StatusCode == 200{ DeleteIdInMap(clusterId) }*/ UpdatePassword(t, projectId, backupClusterId, dbassRequestInfo.CommonEnv.DbPassword) //添加 filter AddTrafficFilter(t, orgId, projectId, backupClusterId, false) if !isDevTier { //删除前先确认backup数量 getRecycleBinResp := dbassRequestInfo.DbassRequest(t, "getRecycleBin", nil, pathparam, param, nil) total := getRecycleBinResp.JSON().Path("$.total").Raw().(float64) deleteBackupResp := dbassRequestInfo.DbassRequest(t, "deleteBackup", nil, restorePathParam, nil, nil) l4g.Info("delete the backup, deleteBackupResp is: %s", deleteBackupResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete backup, pathParam:%s, deleteBackupResp:%s", restorePathParam, deleteBackupResp.JSON().Raw())), allure.Action(func() { deleteBackupResp.Status(StatusOK) })) //删除recycle bin DeleteRecycleBin(t, projectId, clusterId) //查看recycle bin列表 getRecycleBinResp = dbassRequestInfo.DbassRequest(t, "getRecycleBin", nil, pathparam, param, nil) allure.Step(allure.Description(fmt.Sprintf("get recycle bin list,check recycle bin is be deleted, ExpectNum is:%v, "+ "ActualNum is:%v", total-1, getRecycleBinResp.JSON().Path("$.total").Raw())), allure.Action(func() { getRecycleBinResp.JSON().Path("$.total").Equal(total - 1) })) } DeleteClusterIdList = append(DeleteClusterIdList, map[string]string{"projectId": projectId, "clusterId": backupClusterId, "delete": "0"}) return backupClusterId } return "" } func DeleteRecycleBin(t assert.TestingT, projectId, clusterId string) { if clusterId == "" { return } dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} delRecycleBinResp := dbassRequestInfo.DbassRequest(t, "delRecycleBin", nil, clusterPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("delete recycle bin, pathParam:%s, delRecycleBinResp:%s", clusterPathParam, delRecycleBinResp.JSON().Raw())), allure.Action(func() { delRecycleBinResp.Status(StatusOK) })) } func PutAwsImportPolicy(t assert.TestingT, clusterId string) string { dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId projectId := dbassRequestInfo.CommonEnv.ProjectId clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} //查询cluster的role_arn getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getDataIngestionProfile", nil, clusterPathParam, nil, nil) accountId := getClusterInfoResp.JSON().Path("$.aws_account_id").String().Raw() externalId := getClusterInfoResp.JSON().Path("$.aws_external_id").String().Raw() l4g.Info("getDataIngestionProfile:%v", getClusterInfoResp.JSON().Raw()) //bucket := strings.Split(dbassRequestInfo.CommonEnv.sourceUrl, "//") //put policy PutAwsBucketPolicy(accountId, externalId, dbassRequestInfo.CommonEnv.Ec2Host, dbassRequestInfo.CommonEnv.CerPath, dbassRequestInfo.CommonEnv.LoginUser, dbassRequestInfo.CommonEnv.IamUser) return externalId } func PutGcpImportPolicy(t assert.TestingT, clusterId string, dataFormat DataFormat) { dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId projectId := dbassRequestInfo.CommonEnv.ProjectId clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} //查询cluster的role_arn getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getDataIngestionProfile", nil, clusterPathParam, nil, nil) gcp_account_id := getClusterInfoResp.JSON().Path("$.gcp_account_id").String().Raw() l4g.Info("getDataIngestionProfile:%v", getClusterInfoResp.JSON().Raw()) var bucketList []string switch dataFormat { case sqlFormat: bucketList = strings.Split(dbassRequestInfo.CommonEnv.GcpSourceUrl, "/") case parquetFormat: bucketList = strings.Split(dbassRequestInfo.CommonEnv.GcpParquetS3, "/") case csvFormat: bucketList = strings.Split(dbassRequestInfo.CommonEnv.GcpCsvS3, "/") case snappyFormat: bucketList = strings.Split(dbassRequestInfo.CommonEnv.GcpSnappyParquetS3, "/") default: bucketList = strings.Split(dbassRequestInfo.CommonEnv.GcpAuroraS3, "/") } //bucket := strings.Split(dbassRequestInfo.CommonEnv.sourceUrl, "//") //put policy PutGcpBucketPolicy(fmt.Sprintf("gs://%s", bucketList[2]), gcp_account_id, dbassRequestInfo.CommonEnv.Ec2Host, dbassRequestInfo.CommonEnv.CerPath, dbassRequestInfo.CommonEnv.LoginUser) } func CheckSqlImportOperation(t assert.TestingT, clusterId, projectId, provider, region string, isServerless bool) { if clusterId == "" { return } l4g.Info("begin to import data!") dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if region == "" { region = dbassRequestInfo.CommonEnv.importRegion } clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} var sourceUrl, roleArn string var dataSourceType int if provider == "aws" { sourceUrl = dbassRequestInfo.CommonEnv.AwsSourceUrl dataSourceType = 1 externalId := PutAwsImportPolicy(t, clusterId) roleArn = fmt.Sprintf("%s%s", dbassRequestInfo.CommonEnv.AwsRoleArn, externalId) time.Sleep(time.Second * 60) } else { sourceUrl = dbassRequestInfo.CommonEnv.GcpSourceUrl dataSourceType = 2 roleArn = "" PutGcpImportPolicy(t, clusterId, sqlFormat) } //PutImportPolicy(t, clusterId, "sql") CheckImportOperation(t, clusterId, projectId, sourceUrl, roleArn, region, 1, dataSourceType, true, nil) //再次导入 var importId string importInfo := NewImportData(sourceUrl, dbassRequestInfo.CommonEnv.DbPassword, dbassRequestInfo.CommonEnv.importRegion, 1, dataSourceType, roleArn, "", "", true, nil) importResp := dbassRequestInfo.DbassRequest(t, "import", importInfo, clusterPathParam, nil, nil) l4g.Info("do import again, the body is %s, pathParam is %s, importResp is %s", importInfo, clusterPathParam, importResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do import again, body:%s, pathParam:%s, importResp:%s", importInfo, clusterPathParam, importResp.JSON().Raw())), allure.Action(func() { importId = importResp.JSON().Path("$.id").String().Raw() })) //取消导入 // wait for 20 seconds , and this can be reverted after DP team fix 'import API object hasn't been created' //查询import状态 var getImportInfoResp *httpexpect.Response getImportPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": importId} for round := 0; round <= 5; round += 1 { time.Sleep(time.Second * 1 * 5) getImportInfoResp = dbassRequestInfo.DbassRequest(t, "getImportInfo", nil, getImportPathParam, nil, nil) l4g.Info("get import status, the body is %s, pathParam is %s, getImportInfoResp is %s", nil, getImportPathParam, getImportInfoResp.JSON().Raw()) if getImportInfoResp.JSON().Path("$.status").String().Raw() == "IMPORTING" { l4g.Info("second import success, import status is IMPORTING") break } } if getImportInfoResp.JSON().Path("$.status").String().Raw() == "IMPORTING" || isServerless { cancleImportResp := dbassRequestInfo.DbassRequest(t, "cancleImport", nil, getImportPathParam, nil, nil) l4g.Info("cancle the import, the body is %s, pathParam is %s, cancleImportResp is %s", nil, getImportPathParam, cancleImportResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("cancle the import, pathParam:%s, cancleImportResp:%s", getImportPathParam, cancleImportResp.JSON().Raw())), allure.Action(func() { cancleImportResp.Status(StatusOK) })) //查询import状态 if !isServerless { for round := 0; round <= 5; round += 1 { time.Sleep(time.Second * 1 * 60) getImportInfoResp = dbassRequestInfo.DbassRequest(t, "getImportInfo", nil, getImportPathParam, nil, nil) l4g.Info("get import status, the body is %s, pathParam is %s, getImportInfoResp is %s", nil, getImportPathParam, getImportInfoResp.JSON().Raw()) if getImportInfoResp.JSON().Path("$.status").String().Raw() == "CANCELED" { l4g.Info("import cancle success, import status is CANCELED") break } } allure.Step(allure.Description(fmt.Sprintf("get import status, expectStatus:%s, actualStatus:%s", "CANCELED", getImportInfoResp.JSON().Path("$.status").String().Raw())), allure.Action(func() { getImportInfoResp.JSON().Path("$.status").String().Equal("CANCELED") })) } } else { l4g.Warn("the second import status is: %s", getImportInfoResp.JSON().Path("$.status").String().Raw()) } //退出导入 getImportPathParam = map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": importId} importExitResp := dbassRequestInfo.DbassRequest(t, "importExit", nil, getImportPathParam, nil, nil) l4g.Info("do import exit, the body is %s, pathParam is %s, importExitResp is %v", nil, getImportPathParam, importExitResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do import exit, importExitResp:%s", importExitResp.JSON().Raw())), allure.Action(func() { importExitResp.Status(StatusOK) })) //查询集群状态正常 getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) l4g.Info("get cluster info, the body is %s, pathParam is %s, getClusterInfoResp is %v", nil, clusterPathParam, getClusterInfoResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("get cluster status, expectStatus:%s, actualStatus:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw())), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") })) //删除测试数据库 var db *sql.DB _, terr := DropTestDatabases(db, dropDataBasetest) if terr != nil { l4g.Info("drop database failed") } _, aberr := DropTestDatabases(db, dropDataBaseab) if aberr != nil { l4g.Info("drop database failed") } _, dberr := DropTestDatabases(db, dropDataBasedbaas) if dberr != nil { l4g.Info("drop database failed") } _, imerr := DropTestDatabases(db, dropDataBaseimport) if imerr != nil { l4g.Info("drop database failed") } } // Aurora Backup Snapshot func CheckImportAuroraS3Operation(t assert.TestingT, clusterId, projectId, provider, region string, isDevTier bool) { if clusterId == "" { return } l4g.Info("begin to import data!") dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if region == "" { region = dbassRequestInfo.CommonEnv.importRegion } clusterPathParam := map[string]string{"org_id": orgId, "project_id": dbassRequestInfo.CommonEnv.ProjectId, "cluster_id": clusterId} var auroraS3, roleArn string var dataSourceType int if provider == "aws" { auroraS3 = dbassRequestInfo.CommonEnv.AwsAuroraS3 dataSourceType = 1 externalId := PutAwsImportPolicy(t, clusterId) roleArn = fmt.Sprintf("%s%s", dbassRequestInfo.CommonEnv.AwsRoleArn, externalId) } else { auroraS3 = dbassRequestInfo.CommonEnv.GcpAuroraS3 dataSourceType = 2 roleArn = "" PutGcpImportPolicy(t, clusterId, auroraFormat) } //获取cluster信息 getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) tidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) var user string if isDevTier { token := getClusterInfoResp.JSON().Path("$.cluster.endpoint.token").String().Raw() user = token + ".root" } else { user = "root" } //PutImportPolicy(t, clusterId, "parquet") //连接数据库创建表 var db *sql.DB var err error if isDevTier { db, err = ConnectServerlessUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, user) } else { db, err = ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, user) } result, terr := CreateTable(db, parquetCreateTable) value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("ConnectTidb and CreateS3Table, result is:%s, connectTidb failed reason:%s, CreateS3Table failed reason:%s, tidbAddress:%s", result, err, terr, tidbAddress)), allure.Action(func() { value.Equal(true) })) CheckImportOperation(t, clusterId, projectId, auroraS3, roleArn, region, 2, dataSourceType, true, nil) //删除测试数据库 // var db *sql.DB _, cerr := DropTestDatabases(db, dropDataBasetest) if cerr != nil { l4g.Info("drop database failed") } _, aberr := DropTestDatabases(db, dropDataBaseab) if aberr != nil { l4g.Info("drop database failed") } _, dberr := DropTestDatabases(db, dropDataBasedbaas) if dberr != nil { l4g.Info("drop database failed") } _, imerr := DropTestDatabases(db, dropDataBaseimport) if imerr != nil { l4g.Info("drop database failed") } } func CheckImportParquetOperation(t assert.TestingT, clusterId, projectId, provider, region, parquetType string) { if clusterId == "" { return } l4g.Info("begin to import data!") dbassRequestInfo = GetDbassRequestInfo(t) //orgId := dbassRequestInfo.CommonEnv.OrgId //clusterPathParam := map[string]string{"org_id": orgId, "project_id": dbassRequestInfo.CommonEnv.ProjectId, "cluster_id": clusterId} var parquetS3, roleArn string var dataSourceType int if provider == "aws" { if parquetType == "snappy" { parquetS3 = dbassRequestInfo.CommonEnv.AwsSnappyParquetS3 } else { parquetS3 = dbassRequestInfo.CommonEnv.AwsParquetS3 } dataSourceType = 1 externalId := PutAwsImportPolicy(t, clusterId) roleArn = fmt.Sprintf("%s%s", dbassRequestInfo.CommonEnv.AwsRoleArn, externalId) } else { if parquetType == "snappy" { parquetS3 = dbassRequestInfo.CommonEnv.GcpSnappyParquetS3 PutGcpImportPolicy(t, clusterId, snappyFormat) } else { parquetS3 = dbassRequestInfo.CommonEnv.GcpParquetS3 PutGcpImportPolicy(t, clusterId, parquetFormat) } dataSourceType = 2 roleArn = "" } //获取cluster信息 //getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) //tidbAddress := getClusterInfoResp.JSON().Path("$.endpoint.address").String().Raw() //tidbPort := int64(getClusterInfoResp.JSON().Path("$.endpoint.port").Raw().(float64)) //PutImportPolicy(t, clusterId, "parquet") CheckImportOperation(t, clusterId, projectId, parquetS3, roleArn, region, 4, dataSourceType, true, nil) //删除测试数据库 var db *sql.DB _, terr := DropTestDatabases(db, dropDataBasetest) if terr != nil { l4g.Info("drop database failed") } _, aberr := DropTestDatabases(db, dropDataBaseab) if aberr != nil { l4g.Info("drop database failed") } _, dberr := DropTestDatabases(db, dropDataBasedbaas) if dberr != nil { l4g.Info("drop database failed") } _, imerr := DropTestDatabases(db, dropDataBaseimport) if imerr != nil { l4g.Info("drop database failed") } } // csv func CheckImportCsvOperation(t assert.TestingT, clusterId, projectId, provider, region string) { if clusterId == "" { return } l4g.Info("begin to import data!") dbassRequestInfo = GetDbassRequestInfo(t) if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if region == "" { region = dbassRequestInfo.CommonEnv.importRegion } if provider == "aws" { externalId := PutAwsImportPolicy(t, clusterId) roleArn := fmt.Sprintf("%s%s", dbassRequestInfo.CommonEnv.AwsRoleArn, externalId) CheckImportOperation(t, clusterId, projectId, dbassRequestInfo.CommonEnv.AwsCsvS3, roleArn, region, 3, 1, true, nil) //删除测试数据库 var db *sql.DB _, terr := DropTestDatabases(db, dropDataBasetest) if terr != nil { l4g.Info("drop database failed") } _, aberr := DropTestDatabases(db, dropDataBaseab) if aberr != nil { l4g.Info("drop database failed") } _, dberr := DropTestDatabases(db, dropDataBasedbaas) if dberr != nil { l4g.Info("drop database failed") } _, imerr := DropTestDatabases(db, dropDataBaseimport) if imerr != nil { l4g.Info("drop database failed") } } else { PutGcpImportPolicy(t, clusterId, csvFormat) CheckImportOperation(t, clusterId, projectId, dbassRequestInfo.CommonEnv.GcpCsvS3, "", region, 3, 2, true, nil) } } func CheckImportOperation(t assert.TestingT, clusterId string, projectId, S3Url, roleArn, region string, dataFormat, dataSourceType int, header bool, localImportCfg *LocalImportCfg) { if clusterId == "" { return } l4g.Info("begin to import data!") dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId } if region == "" { region = dbassRequestInfo.CommonEnv.importRegion } clusterPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} //import var importInfo *ImportData if localImportCfg != nil { importInfo = NewImportData("", "", region, dataFormat, dataSourceType, "", "", "", header, localImportCfg) } else if dataFormat == 3 { importInfo = NewImportData(S3Url, dbassRequestInfo.CommonEnv.DbPassword, region, dataFormat, dataSourceType, roleArn, dbassRequestInfo.CommonEnv.AwsCsvObjectNamePattern, dbassRequestInfo.CommonEnv.AwsCsvTargetTablePattern, header, nil) } else { importInfo = NewImportData(S3Url, dbassRequestInfo.CommonEnv.DbPassword, region, dataFormat, dataSourceType, roleArn, "", "", true, nil) } importResp := dbassRequestInfo.DbassRequest(t, "import", importInfo, clusterPathParam, nil, nil) l4g.Info("do import, the body is %s, pathParam is %s, importResp is %s", importInfo, clusterPathParam, importResp.JSON().Raw()) var importId string allure.Step(allure.Description(fmt.Sprintf("do import, body:%s, pathParam:%s, importResp:%s", importInfo, clusterPathParam, importResp.JSON().Raw())), allure.Action(func() { importId = importResp.JSON().Path("$.id").String().Raw() })) if importResp.Raw().StatusCode != StatusOK { return } //查询import详情 getImportPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": importId} getImportInfoResp := dbassRequestInfo.DbassRequest(t, "getImportInfo", nil, getImportPathParam, nil, nil) l4g.Info("get import info, the body is %s, pathParam is %s, getImportInfoResp is %s", nil, getImportPathParam, getImportInfoResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("check import status, expect NOT Failed, actualValue:%s, pathParam:%s", getImportInfoResp.JSON().Path("$.status").String().Raw(), getImportPathParam)), allure.Action(func() { getImportInfoResp.JSON().Path("$.status").String().NotEqual("FAILED") })) //查询导入状态 // if isDevTier != true { // var isDevTier bool getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) l4g.Info("get cluster info, the body is %s, pathParam is %s, getClusterInfoResp is %s", nil, clusterPathParam, getClusterInfoResp.JSON().Raw()) //如果是 serverless 集群，不检查集群状态 l4g.Info("get cluster type, this cluster is %s cluster", getClusterInfoResp.JSON().Path("$.cluster.is_dev_tier").Boolean().Raw()) value := getClusterInfoResp.JSON().Path("$.cluster.is_dev_tier").Boolean().Raw() if !value { allure.Step(allure.Description(fmt.Sprintf("get cluster status, expectValue:%s, actualValue:%s", "IMPORTING", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw())), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("IMPORTING") })) } //查询import状态 for round := 0; round <= 20; round += 1 { getImportInfoResp = dbassRequestInfo.DbassRequest(t, "getImportInfo", nil, getImportPathParam, nil, nil) l4g.Info("get import status, the body is %s, pathParam is %s, getImportInfoResp is %s", nil, getImportPathParam, getImportInfoResp.JSON().Raw()) if getImportInfoResp.JSON().Path("$.status").String().Raw() == "COMPLETED" { l4g.Info("import success, import status is COMPLETED") break } else if getImportInfoResp.JSON().Path("$.status").String().Raw() == "FAILED" { l4g.Info("import failed, import status is FAILED") break } time.Sleep(time.Second * 60) } allure.Step(allure.Description(fmt.Sprintf("get import status, expectValue:%s, actualValue:%s, getImportInfoResp:%v", "COMPLETED", getImportInfoResp.JSON().Path("$.status").String().Raw(), getImportInfoResp.JSON().Raw())), allure.Action(func() { getImportInfoResp.JSON().Path("$.status").String().Equal("COMPLETED") getImportInfoResp.JSON().Path("$.total_size").String().NotEqual("0") //getImportInfoResp.JSON().Path("$.total_files").NotEqual(0) })) //退出导入 getImportPathParam = map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId, "id": importId} importExitResp := dbassRequestInfo.DbassRequest(t, "importExit", nil, getImportPathParam, nil, nil) l4g.Info("do import exit, the body is %s, pathParam is %s, importExitResp is %s", nil, getImportPathParam, importExitResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do import exit, importExitResp:%s", importExitResp.JSON().Raw())), allure.Action(func() { importExitResp.Status(StatusOK) })) //查询集群状态正常 for round := 0; round <= 5; round += 1 { getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, clusterPathParam, nil, nil) l4g.Info("get cluster info, the body is %s, pathParam is %s, getClusterInfoResp is %v", nil, clusterPathParam, getClusterInfoResp.JSON().Raw()) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { break } time.Sleep(time.Second * 30) } allure.Step(allure.Description(fmt.Sprintf("get cluster status, expectStatus:%s, actualStatus:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw())), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") })) //删除测试数据库 var db *sql.DB _, terr := DropTestDatabases(db, dropDataBasetest) if terr != nil { l4g.Info("drop database failed") } _, aberr := DropTestDatabases(db, dropDataBaseab) if aberr != nil { l4g.Info("drop database failed") } _, dberr := DropTestDatabases(db, dropDataBasedbaas) if dberr != nil { l4g.Info("drop database failed") } _, imerr := DropTestDatabases(db, dropDataBaseimport) if imerr != nil { l4g.Info("drop database failed") } } func DataToDownstream(t assert.TestingT, provider, clusterId, downstreamType string) { if clusterId == "" { return } dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId projectId := dbassRequestInfo.CommonEnv.ProjectId pathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} var ( protocol *models.DataflowProtocol //dispatchType models.DataflowDispatchType separator = "-" topicPrefix = "cdc" topicSuffix = "cloud" ) // protocol = models.NewDataflowProtocol(models.DataflowProtocolCANALJSON) //testConnector var testConnector []models.DataflowConnectorCreateReq var url, user, password, topic, kafkaHost, kafkaCerPath, loginUser, iamUser, ec2Host, kafkaInstanceId, kafkaLoginUser, cerPath string if downstreamType == "mysql" { url = dbassRequestInfo.CommonEnv.MysqlUrl user = dbassRequestInfo.CommonEnv.MysqlUser password = dbassRequestInfo.CommonEnv.MysqlPassword testConnector1 := NewTestConector(orgId, projectId, clusterId, downstreamType, url, user, password, protocol, "", "", "", "", "", "") // testConnector1 := NewTestConector(orgId, projectId, clusterId, downstreamType, url, user, password, "", "", "", "", "", "", "") testConnector = append(testConnector, testConnector1) } else { ec2Host = dbassRequestInfo.CommonEnv.Ec2Host cerPath = dbassRequestInfo.CommonEnv.CerPath url = dbassRequestInfo.CommonEnv.KafkaUrl topic = dbassRequestInfo.CommonEnv.KafkaTopic kafkaHost = dbassRequestInfo.CommonEnv.KafkaEc2Host kafkaInstanceId = dbassRequestInfo.CommonEnv.KafkaInstanceId kafkaCerPath = dbassRequestInfo.CommonEnv.KafkaCerPath loginUser = dbassRequestInfo.CommonEnv.LoginUser kafkaLoginUser = dbassRequestInfo.CommonEnv.KafkaLoginUser iamUser = dbassRequestInfo.CommonEnv.IamUser Chmod600(kafkaCerPath) StartEC2Instance(ec2Host, cerPath, loginUser, kafkaInstanceId, iamUser) if GetEC2InstanceStatus(ec2Host, cerPath, loginUser, kafkaInstanceId, iamUser) == false { l4g.Info("start kafka instance failed!") return } //publicIp, kafkaHost = GetKafkaInstancePublicIp(ec2Host, cerPath, loginUser, kafkaInstanceId, iamUser) //url = fmt.Sprintf("%s:9092", publicIp) l4g.Info("url:%s", url) testConnector1 := NewTestConector(orgId, projectId, clusterId, downstreamType, url, "", "", protocol, models.DataflowDispatchTypeDISPATCHTYPEONETOPIC, topic, topicPrefix, separator, topicSuffix, "") l4g.Info("kafka testConnector1 is: +%v", testConnector1) testConnector = append(testConnector, testConnector1) testConnector2 := NewTestConector(orgId, projectId, clusterId, downstreamType, url, "", "", protocol, models.DataflowDispatchTypeDISPATCHTYPEBYTABLE, topic, topicPrefix, separator, topicSuffix, "") l4g.Info("kafka testConnector2 is: +%v", testConnector2) testConnector = append(testConnector, testConnector2) testConnector3 := NewTestConector(orgId, projectId, clusterId, downstreamType, url, "", "", protocol, models.DataflowDispatchTypeDISPATCHTYPEBYDATABASE, topic, topicPrefix, separator, topicSuffix, "") l4g.Info("kafka testConnector3 is: +%v", testConnector3) testConnector = append(testConnector, testConnector3) //RunZookeeper(kafkaHost, kafkaCerPath, loginUser) //time.Sleep(time.Second * 60) RunKafka(kafkaHost, kafkaCerPath, kafkaLoginUser) } time.Sleep(time.Second * 60) l4g.Info("testConnector:%s", topic) for k, testConnectorId := range testConnector { fmt.Printf("k(topic dispatch type: 0->one topic, 1->by table, 2->by database) is: %d\n", k) testConnectorResp := dbassRequestInfo.DbassRequest(t, "connectorTest", testConnectorId, pathParam, nil, nil) l4g.Info("testConnector:%v", testConnector) l4g.Info("testConnectorId:%v", testConnectorId) l4g.Info("testConnectorId.clusterid:%v", *testConnectorId.ClusterID) l4g.Info("testConnectorId.ConnectorType:%v", *testConnectorId.ConnectorType) l4g.Info("testConnectorId.OrgID:%v", *testConnectorId.OrgID) l4g.Info("testConnectorId.ProjectID:%v", *testConnectorId.ProjectID) l4g.Info("testConnectorId.URL:%v", *testConnectorId.URL) l4g.Info("testConnectorId.MysqlInfo:%+v, ", *testConnectorId.MysqlInfo) l4g.Info("testConnectorResp:", testConnectorResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("test %s Connector, body:%v, pathParam:%s, testConnectorResp:%s", downstreamType, testConnectorId, pathParam, testConnectorResp.JSON().Raw())), allure.Action(func() { testConnectorResp.Status(StatusOK) })) if testConnectorResp.Raw().StatusCode != StatusOK { if downstreamType == "kafka" { StopEC2Instance(ec2Host, cerPath, loginUser, kafkaInstanceId, iamUser) } return } //confirm confirmConnectorResp := dbassRequestInfo.DbassRequest(t, "connectorConfirm", testConnectorId, pathParam, nil, nil) connectorId := confirmConnectorResp.JSON().Path("$.connector_id").String().Raw() fmt.Println("confirmConnectorResp:", confirmConnectorResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("confirm %s Connector, body:%v, pathParam:%s, confirmConnectorResp:%s", downstreamType, testConnectorId, pathParam, confirmConnectorResp.JSON().Raw())), allure.Action(func() { confirmConnectorResp.Status(StatusOK) })) //查询connector状态 var getConnectorInfoResp *httpexpect.Response clusterPathParam := map[string]string{"org_id": orgId, "cluster_id": clusterId, "project_id": projectId, "connector_id": connectorId} for round := 0; round <= 20; round += 1 { time.Sleep(time.Second * 60) getConnectorInfoResp = dbassRequestInfo.DbassRequest(t, "getConnectorInfo", nil, clusterPathParam, nil, nil) l4g.Info("getConnectorInfoResp is: %s, round is: %d", getConnectorInfoResp.JSON().Raw(), round) if getConnectorInfoResp.JSON().Path("$.status").String().Raw() == "RUNNING" { break } } allure.Step(allure.Description(fmt.Sprintf("get connector status, expectValue:%s, actualValue:%s", "RUNNING", getConnectorInfoResp.JSON().Path("$.status").String().Raw())), allure.Action(func() { getConnectorInfoResp.JSON().Path("$.status").String().Equal("RUNNING") })) if getConnectorInfoResp.JSON().Path("$.status").String().Raw() != "RUNNING" { return } getConnectorsResp := dbassRequestInfo.DbassRequest(t, "getConnectors", nil, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("check connector status in connector list, getConnectorsResp:%v", getConnectorsResp.JSON().Raw())), allure.Action(func() { CheckConectorInConectors(t, getConnectorsResp, connectorId, true) })) //查询使用ticdc cpu core 数量 getTcuResp := dbassRequestInfo.DbassRequest(t, "getTCU", nil, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("check tiCDC cpu Cores, getTcuResp:%v", getTcuResp.JSON().Raw())), allure.Action(func() { getTcuResp.Status(StatusOK) cdcCores, _ := strconv.Atoi(getTcuResp.JSON().Path("$.tcu_number").String().Raw()) assert.Greater(t, cdcCores, 0) })) //获取cluster信息 getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, pathParam, nil, nil) tidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() tidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) //添加 filter AddTrafficFilter(t, orgId, projectId, clusterId, false) if getConnectorInfoResp.JSON().Path("$.status").String().Raw() == "RUNNING" { //check 数据同步 if downstreamType == "kafka" { err := InitConsumer(url) if err != nil { panic(err) } l4g.Info("kafka initConsumer") } //连接数据库创建表 db, err := ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, terr := CreateTable(db, cdcCreateTable) value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("ConnectTidb and Create cdc Table, result is:%s, connectTidb failed reason:%s, CreateS3Table failed reason:%s, tidbAddress:%s", result, err, terr, tidbAddress)), allure.Action(func() { value.Equal(true) })) l4g.Info("create table in upstream tidb.") //连接下游数据库 if downstreamType == "mysql" { slice := strings.Split(url, ":") port, _ := strconv.ParseInt(slice[1], 10, 64) db, err = ConnectTidb(slice[0], port, password, database, "root") result, terr = SelectData(db, testCdcSelect, false, 1, "autotest") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s, tidbAddress:%s", result, err, terr, url)), allure.Action(func() { value.Equal(true) })) } else if downstreamType == "kafka" { l4g.Info("begin consume kafka data.") topic = dbassRequestInfo.CommonEnv.KafkaTopic //topicName := topicPrefix + separator + "test" + separator + tableName + separator + topicSuffix var topicName string switch k { case 0: topicName = topic case 1: topicName = topicPrefix + "test" + separator + tableName + topicSuffix case 2: topicName = topicPrefix + "test" + topicSuffix default: topicName = topic } l4g.Info("topicName is %s \n", topicName) result, terr = LoopConsumer(topicName, cdcCreateTable) value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s, tidbAddress:%s", result, err, terr, url)), allure.Action(func() { value.Equal(true) })) } } //有connector的时候不能导入 var roleArn string if provider == "aws" { roleArn = fmt.Sprintf("%s%s", dbassRequestInfo.CommonEnv.AwsRoleArn, "zmtest") } else { roleArn = "" } importInfo := NewImportData(dbassRequestInfo.CommonEnv.AwsParquetS3, dbassRequestInfo.CommonEnv.DbPassword, dbassRequestInfo.CommonEnv.importRegion, 1, 1, roleArn, "", "", true, nil) importResp := dbassRequestInfo.DbassRequest(t, "import", importInfo, pathParam, nil, nil) l4g.Info("error code is %f", importResp.JSON().Path("$.code").Number().Raw()) allure.Step(allure.Description(fmt.Sprintf("do import, body:%s, pathParam:%s, importResp:%s", importInfo, clusterPathParam, importResp.JSON().Raw())), allure.Action(func() { importResp.JSON().Path("$.code").Number().Equal(float64(403)) })) //删除connector time.Sleep(time.Second * 60) l4g.Info("begin to delete connector.") deleteConnectorResp := dbassRequestInfo.DbassRequest(t, "deleteConnector", nil, clusterPathParam, nil, nil) fmt.Println("deleteConnectorResp:", deleteConnectorResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("delete Connector %s, body:%v, pathParam:%s, deleteConnectorResp:%s", connectorId, nil, clusterPathParam, deleteConnectorResp.JSON().Raw())), allure.Action(func() { deleteConnectorResp.Status(StatusOK) })) time.Sleep(time.Second * 60) //校验数据 l4g.Info("Begin to drop cdc table \n") db, err := ConnectTidbUseTLS(tidbAddress, tidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, terr := CreateTable(db, dropCdcTable) value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("ConnectTidb and drop cdc Table, result is:%s, connectTidb failed reason:%s, drop cdc Table failed reason:%s, tidbAddress:%s", result, err, terr, tidbAddress)), allure.Action(func() { value.Equal(true) })) //连接下游数据库 if downstreamType == "mysql" { slice := strings.Split(url, ":") port, _ := strconv.ParseInt(slice[1], 10, 64) db, err = ConnectTidb(slice[0], port, password, database, "root") result, terr = SelectData(db, testCdcSelect, false, 1, "autotest") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s, tidbAddress:%s", result, err, terr, url)), allure.Action(func() { value.Equal(true) })) } else if downstreamType == "kafka" { //l4g.Info("Begin to consumer topic\n") //result, terr = LoopConsumer(topic, dropCdcTable) //value = httpexpect.NewBoolean(t, result) //allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s, tidbAddress:%s", result, err, terr, url)), // allure.Action(func() { // value.Equal(false) // })) } l4g.Info("Begin to get connector\n") getConnectorsResp = dbassRequestInfo.DbassRequest(t, "getConnectors", nil, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("check connector status in connector list, getConnectorsResp:%v", getConnectorsResp.JSON().Raw())), allure.Action(func() { CheckConectorInConectors(t, getConnectorsResp, connectorId, false) })) if getConnectorsResp.JSON().Path("$.total").Raw().(float64) != 0 { dbassRequestInfo.DbassRequest(t, "deleteConnector", nil, clusterPathParam, nil, nil) getConnectorsResp = dbassRequestInfo.DbassRequest(t, "getConnectors", nil, pathParam, nil, nil) } fmt.Printf("############################################ Waiting for deleting connector and cdc pod completely in background ######################################\n") time.Sleep(time.Minute * 2) } if downstreamType == "kafka" { StopEC2Instance(ec2Host, cerPath, loginUser, kafkaInstanceId, iamUser) } } func CrossRegionReplication(t assert.TestingT, provider, primaryClusterId string, isDevTier bool, tidb_count, tikv_count, tifish_count int, detachType string, isEks bool) { if primaryClusterId == "" { return } dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId projectId := dbassRequestInfo.CommonEnv.ProjectId var primaryRegion string if provider == "aws" { primaryRegion = dbassRequestInfo.CommonEnv.AwsRegion } else { primaryRegion = dbassRequestInfo.CommonEnv.GcpRegion } if dbassRequestInfo.CommonEnv.TiflashCpu == "" { tifish_count = 0 } pathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": primaryClusterId} getSecondaryClusterProfilesResp := dbassRequestInfo.DbassRequest(t, "getSecondaryClusterProfiles", nil, pathParam, nil, nil) var secondaryRegion string allure.Step(allure.Description(fmt.Sprintf("GetSecondaryClusterRegion, body:%v, pathParam:%s, GetSecondaryClusterRegionResp:%s", nil, pathParam, getSecondaryClusterProfilesResp.JSON().Raw())), allure.Action(func() { secondaryRegion = GetSecondaryClusterRegion(getSecondaryClusterProfilesResp, primaryRegion) })) l4g.Info("secondaryRegion is:%s", secondaryRegion) if secondaryRegion == "" { return } CheckCreateCluster(t, provider, isDevTier, tidb_count, tikv_count, tifish_count, projectId, primaryClusterId, secondaryRegion, "") var secondaryClusterId, replicationId string for round := 0; round <= 60; round += 1 { time.Sleep(time.Second * 60) getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, pathParam, nil, nil) secondaryClusterId, replicationId = GetReplicationId(getClusterInfoResp) l4g.Info("getClusterInfoResp is: %s, round is: %d", getClusterInfoResp.JSON().Raw(), round) if replicationId != "" { break } } CheckClusterStatus(t, projectId, secondaryClusterId, "$.cluster.status", "NORMAL", 1) l4g.Info("secondaryClusterId is:%s, replicationId is:%s", secondaryClusterId, replicationId) if secondaryClusterId == "" { return } secPathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": secondaryClusterId} UpdatePassword(t, projectId, secondaryClusterId, dbassRequestInfo.CommonEnv.DbPassword) AddTrafficFilter(t, orgId, projectId, primaryClusterId, false) AddTrafficFilter(t, orgId, projectId, secondaryClusterId, false) //查询使用ticdc cpu core 数量 getFirstCluterTcuResp := dbassRequestInfo.DbassRequest(t, "getTCU", nil, pathParam, nil, nil) getSecCluterTcuResp := dbassRequestInfo.DbassRequest(t, "getTCU", nil, secPathParam, nil, nil) allure.Step(allure.Description("check TiCDC CPU Cores"), allure.Action(func() { getFirstCluterTcuResp.Status(StatusOK) getSecCluterTcuResp.Status(StatusOK) //下游部署 getFirstCluterTcuResp.JSON().Path("$.tcu_number").Equal("0") secCores, _ := strconv.Atoi(getSecCluterTcuResp.JSON().Path("$.tcu_number").String().Raw()) assert.Greater(t, secCores, 0) })) //同步数据 getClusterInfoResp := dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, pathParam, nil, nil) priTidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() priTidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) //连接数据库创建表 db, err := ConnectTidb(priTidbAddress, priTidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, terr := CreateTable(db, cdcCreateTable) value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("ConnectTidb and Create cdc Table, result is:%s, connectTidb failed reason:%s, CreateS3Table failed reason:%s, tidbAddress:%s", result, err, terr, priTidbAddress)), allure.Action(func() { value.Equal(true) })) l4g.Info("create table in upstream tidb.") getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, secPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("Check sencondary cluster status is read-only? getClusterInfoResp:%s", getClusterInfoResp.JSON().Raw())), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.write_status").String().Equal("READONLY") })) secTidbAddress := getClusterInfoResp.JSON().Path("$.cluster.endpoint.address").String().Raw() secTidbPort := int64(getClusterInfoResp.JSON().Path("$.cluster.endpoint.port").Raw().(float64)) l4g.Info("secTidbAddress:%s, secTidbPort:%s", secTidbAddress, secTidbPort) //replicationId := GetReplicationId(getClusterInfoResp) time.Sleep(time.Second * 15) db, err = ConnectTidb(secTidbAddress, secTidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, terr = SelectData(db, testCdcSelect, false, 1, "autotest") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s, tidbAddress:%s", result, err, terr, secTidbAddress)), allure.Action(func() { value.Equal(true) })) //scale //CheckScaleCluster(t, provider, projectId, primaryClusterId, tidb_count, tikv_count, tifish_count, isDevTier) _, tidbProfileId, tikvProfileId, tiflashProfileId, _ := ScaleCluster(t, provider, projectId, primaryClusterId, tidb_count+1, tikv_count, tifish_count, isDevTier, isEks) CheckClusterStatus(t, projectId, primaryClusterId, "$.cluster.status", "SCALING", 10) CheckClusterStatus(t, projectId, primaryClusterId, "$.cluster.status", "NORMAL", 30) CheckClusterStatus(t, projectId, secondaryClusterId, "$.cluster.status", "SCALING", 10) CheckClusterStatus(t, projectId, secondaryClusterId, "$.cluster.status", "NORMAL", 30) if provider == "aws" { CheckClusterInfoByDetail(t, secondaryClusterId, tifish_count, AwsTikvStorage+10, AwsTiflashStorage+10, tidbProfileId, tikvProfileId, tiflashProfileId, projectId) } else { CheckClusterInfoByDetail(t, secondaryClusterId, tifish_count, GcpTikvStorage+10, GcpTiflashStorage+10, tidbProfileId, tikvProfileId, tiflashProfileId, projectId) } /*for round := 0; round <= 15; round += 1 { time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, secPathParam, nil, nil) l4g.Info("get cluster info, body is: %+v, pathParam is: %s, getClusterInfoResp is %s", nil, secPathParam, getClusterInfoResp.JSON().Raw()) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" { l4g.Info("scale status is normal!") break } } allure.Step(allure.Description(fmt.Sprintf("check scale result status, expectStatus:%s, actualStatus:%s, pathParam:%s", "NORMAL", getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw(), secPathParam)), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.status").String().Equal("NORMAL") }))*/ //detach detachPathParam := map[string]string{"org_id": orgId, "cluster_id": primaryClusterId, "replication_id": replicationId, "detach_type": detachType} forceDetachResp := dbassRequestInfo.DbassRequest(t, "detach", nil, detachPathParam, nil, nil) l4g.Info("detachResp is:%s", forceDetachResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("detachResp:%s", forceDetachResp.JSON().Raw())), allure.Action(func() { forceDetachResp.Status(StatusOK) })) //已有cluster查询 getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, secPathParam, nil, nil) getClusterInfoResp.JSON().Path("$.cluster.replications[0].status").String().Equal("detaching") for round := 0; round <= 36; round += 1 { time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, secPathParam, nil, nil) l4g.Info("getClusterInfoResp is: %s, round is: %d", getClusterInfoResp.JSON().Raw(), round) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" && getClusterInfoResp.JSON().Path("$.cluster.write_status").String().Raw() == "WRITABLE" { break } } allure.Step(allure.Description(fmt.Sprintf("Check sencondary cluster status is write-able? getClusterInfoResp:%s", getClusterInfoResp.JSON().Raw())), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.write_status").String().Equal("WRITABLE") })) for round := 0; round <= 10; round += 1 { time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, pathParam, nil, nil) l4g.Info("getClusterInfoResp is: %s, round is: %d", getClusterInfoResp.JSON().Raw(), round) if getClusterInfoResp.JSON().Path("$.cluster.status").String().Raw() == "NORMAL" && getClusterInfoResp.JSON().Path("$.cluster.write_status").String().Raw() == "READONLY" { break } } allure.Step(allure.Description(fmt.Sprintf("Check primary cluster status is read-only? getClusterInfoResp:%s", getClusterInfoResp.JSON().Raw())), allure.Action(func() { getClusterInfoResp.JSON().Path("$.cluster.write_status").String().Equal("READONLY") })) //主集群置为可写 makeWriteableResp := dbassRequestInfo.DbassRequest(t, "makeWriteable", nil, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("set primary cluster is writeable. makeWriteableResp:%s", makeWriteableResp.JSON().Raw())), allure.Action(func() { makeWriteableResp.Status(StatusOK) })) for round := 0; round <= 10; round += 1 { time.Sleep(time.Second * 60) getClusterInfoResp = dbassRequestInfo.DbassRequest(t, "getClusterInfo", nil, pathParam, nil, nil) l4g.Info("get cluster info, body is: %+v, pathParam is: %s, getClusterInfoResp is %s", nil, secPathParam, getClusterInfoResp.JSON().Raw()) if getClusterInfoResp.JSON().Path("$.cluster.write_status").String().Raw() == "WRITABLE" { l4g.Info("primary write status is writable!") break } } //同步数据 db, err = ConnectTidb(priTidbAddress, priTidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, terr = CreateTable(db, dropCdcTable) value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("ConnectTidb and drop cdc Table, result is:%s, connectTidb failed reason:%s, drop cdc Table failed reason:%s, tidbAddress:%s", result, err, terr, priTidbAddress)), allure.Action(func() { value.Equal(true) })) //连接下游数据库 db, err = ConnectTidb(secTidbAddress, secTidbPort, dbassRequestInfo.CommonEnv.DbPassword, database, "root") result, terr = SelectData(db, testCdcSelect, false, 1, "autotest") value = httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("check downstream mysql data, result is:%s, connectTidb failed reason:%s, SelectData failed reason:%s, tidbAddress:%s", result, err, terr, secTidbAddress)), allure.Action(func() { value.Equal(true) })) //scale /*var scale *Scale if tifish_count != 0 { tifish_count += 1 awsTiflashStorage += 10 gcpTiflashStorage += 10 } if dbassRequestInfo.CommonEnv.TidbCpu != "2" && dbassRequestInfo.CommonEnv.TidbCpu != "4"{ tikv_count += 3 } if provider == "aws" { scale = NewScale(tidb_count+1, tikv_count, tifish_count, awsTikvStorage + 10, awsTiflashStorage, awsTidbProfileId, awsTikvProfileId, awsTiflashProfileId) } else { scale = NewScale(tidb_count+1, tikv_count, tifish_count, gcpTikvStorage + 10, gcpTiflashStorage, gcpTidbProfileId, gcpTikvProfileId, gcpTiflashProfileId) } scaleResp := dbassRequestInfo.DbassRequest(t, "scale", scale, secPathParam, nil, nil) l4g.Info("do the scale, body is: %+v, pathParam is: %s, scaleResp is %v", scale, secPathParam, scaleResp.JSON().Raw()) allure.Step(allure.Description(fmt.Sprintf("do scale, body is: %+v, pathParam is: %s, scaleResp:%v", scale, secPathParam, scaleResp.JSON().Raw())), allure.Action(func() { scaleResp.JSON().Path("$.base_resp.err_code").String().Equal("0") }))*/ _, tidbProfileId, tikvProfileId, tiflashProfileId, _ = ScaleCluster(t, provider, projectId, secondaryClusterId, tidb_count, tikv_count, tifish_count, isDevTier, isEks) CheckClusterStatus(t, projectId, secondaryClusterId, "$.cluster.status", "SCALING", 1) CheckClusterStatus(t, projectId, primaryClusterId, "$.cluster.status", "NORMAL", 1) if provider == "aws" { CheckClusterInfoByDetail(t, secondaryClusterId, tifish_count, AwsTikvStorage+10, AwsTiflashStorage+10, tidbProfileId, tikvProfileId, tiflashProfileId, projectId) } else if provider == "gcp" { CheckClusterInfoByDetail(t, secondaryClusterId, tifish_count, GcpTikvStorage+10, GcpTiflashStorage+10, tidbProfileId, tikvProfileId, tiflashProfileId, projectId) } //删除集群 CheckDeleteCluster(t, primaryClusterId, projectId, true) CheckDeleteCluster(t, secondaryClusterId, projectId, true) } func CheckJobStatus(t assert.TestingT, projectId, clusterId, jobId, urlPath, resultPath, expectStatus string, maxRound int, importMode string) string { dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId //var auto_project = 1 if projectId == "" { projectId = dbassRequestInfo.CommonEnv.ProjectId //auto_project = 0 } var getJobInfoResp *httpexpect.Response var status, respImportMode string var result bool jobPathParam := map[string]string{"org_id": orgId, "cluster_id": clusterId, "project_id": projectId, "job_id": jobId} for round := 0; round <= 2*maxRound; round += 1 { getJobInfoResp = dbassRequestInfo.DbassRequest(t, urlPath, nil, jobPathParam, nil, nil) l4g.Info("getJobInfoResp is: %s, round is: %d", getJobInfoResp.JSON().Raw(), round) data := getJobInfoResp.JSON().Path("$.job.sub_tasks").Raw() respImportMode = getJobInfoResp.JSON().Path("$.job.import_mode").String().Raw() var dataList []string var newdata, stepdata []string dataList = strings.Split(fmt.Sprintf("%v", data), " ") for _, value := range dataList { if strings.Contains(value, "stage") { newdata1 := strings.Split(value, ":") newdata = strings.Split(newdata1[1], "]") } else if strings.Contains(value, "current_step") { stepdata = strings.Split(value, ":") } } l4g.Info("newdata:%s, expectStatus, expectStep:%s, expectStep", newdata[0], expectStatus, stepdata[2], "Sync") status = newdata[0] if newdata[0] == expectStatus && (respImportMode == importMode || importMode == "") && stepdata[2] == "Sync" { result = true break } if newdata[0] == "Failed" { result = false break } time.Sleep(time.Second * 30) } //status = getJobInfoResp.JSON().Path(resultPath).String().Raw() allure.Step( allure.Story("CheckJobStatus"), allure.Description(fmt.Sprintf("get job status, check the status is normal. "+ "Expected value：%s, Actual value: %s, jobId is:%s, respImportMode:%s, expect import mode:%s", expectStatus, status, jobId, respImportMode, importMode)), allure.Action(func() { //getJobInfoResp.JSON().Path(resultPath).String().Equal(expectStatus) value := httpexpect.NewBoolean(t, result) value.Equal(true) })) return status } func DataMigration(t assert.TestingT, provider, clusterId, model, importMode string) { if clusterId == "" { return } var region, dmHost, connMethod string dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId projectId := dbassRequestInfo.CommonEnv.ProjectId if provider == "aws" { region = dbassRequestInfo.CommonEnv.AwsRegion } else { region = dbassRequestInfo.CommonEnv.GcpRegion } if model == "vpc-peering" { dmHost = dbassRequestInfo.CommonEnv.DMPrivateHost connMethod = "vpc-peering" } else { dmHost = dbassRequestInfo.CommonEnv.DMHost connMethod = "direct" } dmPublicHost := dbassRequestInfo.CommonEnv.DMHost dmPort := dbassRequestInfo.CommonEnv.DMPort dmUser := dbassRequestInfo.CommonEnv.DMUser dmPassword := dbassRequestInfo.CommonEnv.DMPassword dmType := dbassRequestInfo.CommonEnv.DMType tidbUser := "root" tidbPassword := dbassRequestInfo.CommonEnv.DbPassword mode := "all" name := "dm-test" dmPortData, _ := strconv.Atoi(dmPort) specId := "1" connProfileData := NewConnProfileData(connMethod, dmHost, dmPassword, region, dmType, dmUser, "", dmPortData) precheckData := NewDmData(tidbPassword, tidbUser, mode, name, "", "", "", true, connProfileData, importMode) pathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} precheckResp := dbassRequestInfo.DbassRequest(t, "dmPrecheck", precheckData, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("do precheck, body:%+v, pathParam:%s, precheckResp:%v", precheckData, pathParam, precheckResp.JSON().Raw())), allure.Action(func() { precheckResp.JSON().Path("$.id").NotNull() })) precheckId := precheckResp.JSON().Path("$.id").String().Raw() l4g.Info("precheckResp is:%s, precheckId:%s", precheckResp.JSON().Raw(), precheckId) connProfileData1 := NewConnProfileData(connMethod, dmHost, dmPassword, region, dmType, dmUser, name, dmPortData) connProfilesResp := dbassRequestInfo.DbassRequest(t, "dmConnProfiles", connProfileData1, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("do connProfiles, body:%v, pathParam:%s, connProfilesResp:%s", connProfileData1, pathParam, connProfilesResp.JSON().Raw())), allure.Action(func() { connProfilesResp.JSON().Path("$.id").NotNull() })) connProfilesId := connProfilesResp.JSON().Path("$.id").String().Raw() l4g.Info("connProfilesResp is:%s, connProfilesId:%s", connProfilesResp.JSON().Raw(), connProfilesId) addDMJobData := NewDmData(tidbPassword, tidbUser, mode, name, precheckId, specId, connProfilesId, true, connProfileData, importMode) addDMJobResp := dbassRequestInfo.DbassRequest(t, "dmCreateJob", addDMJobData, pathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("create DM job, body:%+v, pathParam:%s, addDMJobResp:%s", addDMJobData.GoString(), pathParam, addDMJobResp.JSON().Raw())), allure.Action(func() { addDMJobResp.JSON().Path("$.id").NotNull() })) jobId := addDMJobResp.JSON().Path("$.id").String().Raw() l4g.Info("addDMJobResp is:%s, jobId:%s", addDMJobResp.JSON().Raw(), jobId) if jobId == "" { return } CheckJobStatus(t, projectId, clusterId, jobId, "queryDmJobInfo", "$.job.sub_tasks."+connProfilesId+".stage", "Running", 30, importMode) time.Sleep(time.Minute * 1) status := CheckJobStatus(t, projectId, clusterId, jobId, "queryDmJobInfo", "$.job.sub_tasks."+connProfilesId+".stage", "Running", 30, importMode) if status == "Running" { maxCheckTimes := 1 if importMode == "physical" { maxCheckTimes = 12 } CheckDataSyncWithMysql(t, orgId, projectId, clusterId, dmPublicHost, dmPassword, dmPortData, cdcCreateTable, testCdcSelect, dmUser, true, true, maxCheckTimes) } else { return } jobPathParam := map[string]string{"org_id": orgId, "cluster_id": clusterId, "project_id": projectId, "job_id": jobId} //pause pauseDMJobResp := dbassRequestInfo.DbassRequest(t, "dmPause", map[string]string{"op": "Pause"}, jobPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("pause DM job, body:%v, pathParam:%s, pauseDMJobResp:%s", nil, jobPathParam, pauseDMJobResp.JSON().Raw())), allure.Action(func() { pauseDMJobResp.Status(StatusOK) })) l4g.Info("pauseDMJobResp is:%s, jobId:%s", pauseDMJobResp.JSON().Raw(), jobId) l4g.Info("Begin to drop cdc table \n") CheckDataSyncWithMysql(t, orgId, projectId, clusterId, dmPublicHost, dmPassword, dmPortData, dropCdcTable, testCdcSelect, dmUser, true, true, 1) //resume resumeDMJobResp := dbassRequestInfo.DbassRequest(t, "dmPause", map[string]string{"op": "Resume"}, jobPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("resume DM job, body:%v, pathParam:%s, resumeDMJobResp:%s", nil, jobPathParam, resumeDMJobResp.JSON().Raw())), allure.Action(func() { pauseDMJobResp.Status(StatusOK) })) l4g.Info("resumeDMJobResp is:%s, jobId:%s", resumeDMJobResp.JSON().Raw(), jobId) //连接数据库创建表 CheckDataSyncWithMysql(t, orgId, projectId, clusterId, dmPublicHost, dmPassword, dmPortData, cdcCreateTable, testCdcSelect, dmUser, true, true, 1) CheckDataSyncWithMysql(t, orgId, projectId, clusterId, dmPublicHost, dmPassword, dmPortData, dropCdcTable, testCdcSelect, dmUser, true, false, 1) scaleDMJobResp := dbassRequestInfo.DbassRequest(t, "scaleupDm", map[string]string{"to_spec_id": "2"}, jobPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("scale DM job, body:%v, pathParam:%s, scaleDMJobResp:%s", nil, jobPathParam, scaleDMJobResp.JSON().Raw())), allure.Action(func() { scaleDMJobResp.Status(StatusOK) })) CheckJobStatus(t, projectId, clusterId, jobId, "queryDmJobInfo", "$.job.sub_tasks."+connProfilesId+".stage", "Scaling", 2, "") CheckJobStatus(t, projectId, clusterId, jobId, "queryDmJobInfo", "$.job.sub_tasks."+connProfilesId+".stage", "Running", 15, "") deleteDMJobResp := dbassRequestInfo.DbassRequest(t, "deleteDmJob", nil, jobPathParam, nil, nil) allure.Step(allure.Description(fmt.Sprintf("delete DM job, body:%v, pathParam:%s, deleteDMJobResp:%s", nil, jobPathParam, deleteDMJobResp.JSON().Raw())), allure.Action(func() { deleteDMJobResp.Status(StatusOK) })) l4g.Info("deleteDMJobResp is:%s, jobId:%s", deleteDMJobResp.JSON().Raw(), jobId) } func LocalImport(t assert.TestingT, clusterId string, isDevTier bool) { if clusterId == "" { return } var region string dbassRequestInfo = GetDbassRequestInfo(t) orgId := dbassRequestInfo.CommonEnv.OrgId projectId := dbassRequestInfo.CommonEnv.ProjectId fileName := dbassRequestInfo.CommonEnv.LocalImportFileName filePath := dbassRequestInfo.CommonEnv.LocalImportFilePath contentLength := dbassRequestInfo.CommonEnv.LocalImportFileSize if isDevTier == true { region = dbassRequestInfo.CommonEnv.DevTierRegion } csvHeader := false pathParam := map[string]string{"org_id": orgId, "project_id": projectId, "cluster_id": clusterId} upLoadData := NewUploadData(contentLength, fileName, region) localImportUploadResp := dbassRequestInfo.DbassRequest(t, "localImportUpload", upLoadData, pathParam, nil, nil) l4g.Info("localImportUploadResp:%s, upLoadData:%s", localImportUploadResp.JSON().Raw(), upLoadData) allure.Step(allure.Description(fmt.Sprintf("do localImportUpload, body:%v, pathParam:%s, localImportUploadResp:%s", upLoadData, nil, localImportUploadResp.JSON().Raw())), allure.Action(func() { localImportUploadResp.JSON().Path("$.url").String().NotEmpty() })) url := localImportUploadResp.JSON().Path("$.url").String().Raw() newFileName := localImportUploadResp.JSON().Path("$.file_name").String().Raw() if url == "" { return } l4g.Info("url:%s", url) data := strings.Split(url, "?") urlPathData := strings.Split(data[0], "/") urlData := fmt.Sprintf("https://%s", urlPathData[2]) path := fmt.Sprintf("/%s/%s", urlPathData[3], urlPathData[4]) path = strings.ReplaceAll(path, "%3A", ":") paramData := strings.Split(data[1], "&") //credentialList := strings.Split(strings.Split(paramData[1], "=")[1], "%2F") credentialData := strings.ReplaceAll(strings.Split(paramData[1], "=")[1], "%2F", "/") //token :=strings.Split(paramData[4], "=")[1] token := strings.ReplaceAll(strings.Split(paramData[4], "=")[1], "%2F", "/") token = strings.ReplaceAll(token, "%2B", "+") token = strings.ReplaceAll(token, "%3D", "=") header := strings.ReplaceAll(strings.Split(paramData[5], "=")[1], "%3B", ";") param := map[string]string{"X-Amz-Algorithm": strings.Split(paramData[0], "=")[1], "X-Amz-Credential": credentialData, "X-Amz-Date": strings.Split(paramData[2], "=")[1], "X-Amz-Expires": strings.Split(paramData[3], "=")[1], "X-Amz-Security-Token": token, "X-Amz-SignedHeaders": header, "X-Amz-Signature": strings.Split(paramData[6], "=")[1]} //var buf bytes.Buffer file, _ := os.Open(filePath) defer file.Close() fileByte, _ := ioutil.ReadAll(file) upLoadFileToS3Resp := Put1(t, []string{urlData, path}, map[string]string{"Content-Type": "text/csv"}, fileByte, nil, param) l4g.Info("upLoadFileToS3Resp:%s", upLoadFileToS3Resp) allure.Step(allure.Description(fmt.Sprintf("do upLoadFileToS3, body:%v, pathParam:%s, upLoadFileToS3Resp:%s", nil, nil, upLoadFileToS3Resp)), allure.Action(func() { upLoadFileToS3Resp.Status(StatusOK) })) //preview localCsvFormatData := NewLocalCsvFormat(newFileName, region, csvHeader) previewResp := dbassRequestInfo.DbassRequest(t, "localImportPreview", localCsvFormatData, pathParam, nil, nil) l4g.Info("previewResp:%s, localCsvFormatData:%s", previewResp.JSON().Raw(), localCsvFormatData) allure.Step(allure.Description(fmt.Sprintf("do preview, body:%v, pathParam:%s, previewResp:%s", localCsvFormatData, nil, previewResp.JSON().Raw())), allure.Action(func() { previewResp.Status(StatusOK) previewResp.JSON().Path("$.table.row_data_types[0]").String().NotEmpty() previewResp.JSON().Path("$.table.row_values[0].column_values[0]").String().NotEmpty() })) //解析 rowDataType, _ := GetRowDataTypeAndHeader(previewResp) var result bool if dbassRequestInfo.CommonEnv.LocalImportExceptColumnsCount != "" { expectColumnsCount, _ := strconv.Atoi(dbassRequestInfo.CommonEnv.LocalImportExceptColumnsCount) if expectColumnsCount == len(rowDataType) { result = true } value := httpexpect.NewBoolean(t, result) allure.Step(allure.Description(fmt.Sprintf("preview result:%s, expectColumnsCount:%d, actualColumnsCount:%d", result, expectColumnsCount, len(rowDataType))), allure.Action(func() { value.Equal(true) })) } var columns []*ColumeInfo for index, column := range rowDataType { var columnData *ColumeInfo if index == 0 { columnData = NewColumeInfo("rank", column) } else { columnData = NewColumeInfo(fmt.Sprintf("column_%d", index+1), column) } columns = append(columns, columnData) } pkColumns := []string{"rank"} contentSize, _ := strconv.Atoi(contentLength) localImportCfg := NewLocalImportCfg(newFileName, fileName, "test_Local1", "test_Local1_table", columns, contentSize, pkColumns) CheckImportOperation(t, clusterId, projectId, "", "", region, 3, 3, false, localImportCfg) }