Skip to content

Commit

Permalink
test: make TestOperatorTestSuite more stable
Browse files Browse the repository at this point in the history 
Signed-off-by: lhy1024 <[email protected]>
  • Loading branch information
@lhy1024
lhy1024 committed Oct 26, 2023
1 parent 89128f1 commit fb57bb7
Show file tree
Hide file tree
Showing 4 changed files with 107 additions and 7 deletions.
8 changes: 8 additions & 0 deletions tests/pdctl/operator/operator_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
"github.com/pingcap/kvproto/pkg/metapb"
"github.com/stretchr/testify/suite"
"github.com/tikv/pd/pkg/core"
"github.com/tikv/pd/pkg/utils/testutil"
"github.com/tikv/pd/server/config"
"github.com/tikv/pd/tests"
"github.com/tikv/pd/tests/pdctl"
Expand Down Expand Up @@ -221,6 +222,13 @@ func (suite *operatorTestSuite) checkOperator(cluster *tests.TestCluster) {

_, err = pdctl.ExecuteCommand(cmd, "config", "set", "enable-placement-rules", "true")
re.NoError(err)
if sche := cluster.GetSchedulingPrimaryServer(); sche != nil {
// wait for the scheduler server to update the config
testutil.Eventually(re, func() bool {
return sche.GetPersistConfig().IsPlacementRulesCacheEnabled()
})
}

output, err = pdctl.ExecuteCommand(cmd, "operator", "add", "transfer-region", "1", "2", "3")
re.NoError(err)
re.Contains(string(output), "not supported")
Expand Down
4 changes: 1 addition & 3 deletions tests/pdctl/scheduler/scheduler_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -472,10 +472,8 @@ func (suite *schedulerTestSuite) checkSchedulerDiagnostic(cluster *tests.TestClu
result := make(map[string]interface{})
testutil.Eventually(re, func() bool {
mightExec(re, cmd, []string{"-u", pdAddr, "scheduler", "describe", schedulerName}, &result)
return len(result) != 0
return len(result) != 0 && expectedStatus == result["status"] && expectedSummary == result["summary"]
}, testutil.WithTickInterval(50*time.Millisecond))
re.Equal(expectedStatus, result["status"])
re.Equal(expectedSummary, result["summary"])
}

stores := []*metapb.Store{
Expand Down
48 changes: 44 additions & 4 deletions tests/server/api/operator_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,18 @@ func (suite *operatorTestSuite) TestOperator() {

func (suite *operatorTestSuite) checkAddRemovePeer(cluster *tests.TestCluster) {
re := suite.Require()

// pause rule checker to avoid unexpected operator
checkerName := "rule"
addr := cluster.GetLeaderServer().GetAddr()
resp := make(map[string]interface{})
url := fmt.Sprintf("%s/pd/api/v1/checker/%s", addr, checkerName)
err := tu.CheckPostJSON(testDialClient, url, []byte(`{"delay":1000}`), tu.StatusOK(re))
re.NoError(err)
err = tu.ReadGetJSON(re, testDialClient, url, &resp)
re.NoError(err)
re.True(resp["paused"].(bool))

stores := []*metapb.Store{
{
Id: 1,
Expand Down Expand Up @@ -106,13 +118,15 @@ func (suite *operatorTestSuite) checkAddRemovePeer(cluster *tests.TestCluster) {
ConfVer: 1,
Version: 1,
},
StartKey: []byte("a"),
EndKey: []byte("b"),
}
regionInfo := core.NewRegionInfo(region, peer1)
tests.MustPutRegionInfo(re, cluster, regionInfo)

urlPrefix := fmt.Sprintf("%s/pd/api/v1", cluster.GetLeaderServer().GetAddr())
regionURL := fmt.Sprintf("%s/operators/%d", urlPrefix, region.GetId())
err := tu.CheckGetJSON(testDialClient, regionURL, nil,
err = tu.CheckGetJSON(testDialClient, regionURL, nil,
tu.StatusNotOK(re), tu.StringContain(re, "operator not found"))
suite.NoError(err)
recordURL := fmt.Sprintf("%s/operators/records?from=%s", urlPrefix, strconv.FormatInt(time.Now().Unix(), 10))
Expand Down Expand Up @@ -168,14 +182,26 @@ func (suite *operatorTestSuite) checkAddRemovePeer(cluster *tests.TestCluster) {

// Fail to get operator if from is latest.
time.Sleep(time.Second)
url := fmt.Sprintf("%s/operators/records?from=%s", urlPrefix, strconv.FormatInt(time.Now().Unix(), 10))
url = fmt.Sprintf("%s/operators/records?from=%s", urlPrefix, strconv.FormatInt(time.Now().Unix(), 10))
err = tu.CheckGetJSON(testDialClient, url, nil,
tu.StatusNotOK(re), tu.StringContain(re, "operator not found"))
suite.NoError(err)
}

func (suite *operatorTestSuite) checkMergeRegionOperator(cluster *tests.TestCluster) {
re := suite.Require()

// pause rule checker to avoid unexpected operator
checkerName := "rule"
addr := cluster.GetLeaderServer().GetAddr()
resp := make(map[string]interface{})
url := fmt.Sprintf("%s/pd/api/v1/checker/%s", addr, checkerName)
err := tu.CheckPostJSON(testDialClient, url, []byte(`{"delay":1000}`), tu.StatusOK(re))
re.NoError(err)
err = tu.ReadGetJSON(re, testDialClient, url, &resp)
re.NoError(err)
re.True(resp["paused"].(bool))

r1 := core.NewTestRegionInfo(10, 1, []byte(""), []byte("b"), core.SetWrittenBytes(1000), core.SetReadBytes(1000), core.SetRegionConfVer(1), core.SetRegionVersion(1))
tests.MustPutRegionInfo(re, cluster, r1)
r2 := core.NewTestRegionInfo(20, 1, []byte("b"), []byte("c"), core.SetWrittenBytes(2000), core.SetReadBytes(0), core.SetRegionConfVer(2), core.SetRegionVersion(3))
Expand All @@ -184,7 +210,7 @@ func (suite *operatorTestSuite) checkMergeRegionOperator(cluster *tests.TestClus
tests.MustPutRegionInfo(re, cluster, r3)

urlPrefix := fmt.Sprintf("%s/pd/api/v1", cluster.GetLeaderServer().GetAddr())
err := tu.CheckPostJSON(testDialClient, fmt.Sprintf("%s/operators", urlPrefix), []byte(`{"name":"merge-region", "source_region_id": 10, "target_region_id": 20}`), tu.StatusOK(re))
err = tu.CheckPostJSON(testDialClient, fmt.Sprintf("%s/operators", urlPrefix), []byte(`{"name":"merge-region", "source_region_id": 10, "target_region_id": 20}`), tu.StatusOK(re))
suite.NoError(err)

tu.CheckDelete(testDialClient, fmt.Sprintf("%s/operators/%d", urlPrefix, 10), tu.StatusOK(re))
Expand All @@ -201,6 +227,18 @@ func (suite *operatorTestSuite) checkMergeRegionOperator(cluster *tests.TestClus

func (suite *operatorTestSuite) checkTransferRegionWithPlacementRule(cluster *tests.TestCluster) {
re := suite.Require()

// pause rule checker to avoid unexpected operator
checkerName := "rule"
addr := cluster.GetLeaderServer().GetAddr()
resp := make(map[string]interface{})
url := fmt.Sprintf("%s/pd/api/v1/checker/%s", addr, checkerName)
err := tu.CheckPostJSON(testDialClient, url, []byte(`{"delay":1000}`), tu.StatusOK(re))
re.NoError(err)
err = tu.ReadGetJSON(re, testDialClient, url, &resp)
re.NoError(err)
re.True(resp["paused"].(bool))

stores := []*metapb.Store{
{
Id: 1,
Expand Down Expand Up @@ -239,12 +277,14 @@ func (suite *operatorTestSuite) checkTransferRegionWithPlacementRule(cluster *te
ConfVer: 1,
Version: 1,
},
StartKey: []byte("a"),
EndKey: []byte("b"),
}
tests.MustPutRegionInfo(re, cluster, core.NewRegionInfo(region, peer1))

urlPrefix := fmt.Sprintf("%s/pd/api/v1", cluster.GetLeaderServer().GetAddr())
regionURL := fmt.Sprintf("%s/operators/%d", urlPrefix, region.GetId())
err := tu.CheckGetJSON(testDialClient, regionURL, nil,
err = tu.CheckGetJSON(testDialClient, regionURL, nil,
tu.StatusNotOK(re), tu.StringContain(re, "operator not found"))
re.NoError(err)
convertStepsToStr := func(steps []string) string {
Expand Down
54 changes: 54 additions & 0 deletions tests/server/config/config_test.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,15 @@ import (
"bytes"
"context"
"encoding/json"
"fmt"
"net/http"
"testing"

"github.com/stretchr/testify/require"
"github.com/tikv/pd/pkg/ratelimit"
"github.com/tikv/pd/pkg/utils/testutil"
"github.com/tikv/pd/server"
"github.com/tikv/pd/server/config"
"github.com/tikv/pd/tests"
)

Expand Down Expand Up @@ -74,3 +77,54 @@ func TestRateLimitConfigReload(t *testing.T) {
re.True(leader.GetServer().GetServiceMiddlewarePersistOptions().IsRateLimitEnabled())
re.Len(leader.GetServer().GetServiceMiddlewarePersistOptions().GetRateLimitConfig().LimiterConfig, 1)
}

func TestSyncConfig(t *testing.T) {
// FIXME: this test is failed, skip it for now.
// the code cannot sync config to scheduler server.
t.Skip("skip failed test")
re := require.New(t)
testDialClient := &http.Client{
Transport: &http.Transport{
DisableKeepAlives: true,
},
}
checkSyncConfig := func(cluster *tests.TestCluster) {
svr := cluster.GetLeaderServer()
url := fmt.Sprintf("%s/pd/api/v1/config", svr.GetAddr())
for i := 0; i < 10; i++ {
expected := i%2 == 0
data := make(map[string]interface{})
if expected {
data["enable-placement-rules"] = "true"
} else {
data["enable-placement-rules"] = "false"
}
reqData, e := json.Marshal(data)
re.NoError(e)
err := testutil.CheckPostJSON(testDialClient, url, reqData, testutil.StatusOK(re))
re.NoError(err)
err = testutil.ReadGetJSON(re, testDialClient, url, &data)
re.NoError(err)
flag := data["replication"].(map[string]interface{})["enable-placement-rules"].(string)
if expected {
re.Equal("true", flag)
} else {
re.Equal("false", flag)
}
if sche := cluster.GetSchedulingPrimaryServer(); sche != nil {
// wait for the scheduler server to update the config
testutil.Eventually(re, func() bool {
return sche.GetCluster().GetCheckerConfig().IsPlacementRulesCacheEnabled() == expected
})
}
}
}

opts := []tests.ConfigOption{
func(conf *config.Config, serverName string) {
conf.Replication.MaxReplicas = 1
},
}
env := tests.NewSchedulingTestEnvironment(t, opts...)
env.RunTestInTwoModes(checkSyncConfig)
}

0 comments on commit fb57bb7

Please sign in to comment.