From aced6d8587dffcda5bf4741d5f002837eb4448de Mon Sep 17 00:00:00 2001 From: JmPotato Date: Fri, 24 Nov 2023 17:45:39 +0800 Subject: [PATCH] Refine the test Signed-off-by: JmPotato --- client/http/client.go | 2 +- pkg/utils/tsoutil/tsoutil.go | 5 +++ server/cluster/cluster.go | 6 ++- tests/integrations/client/http_client_test.go | 42 +++++++++++-------- 4 files changed, 34 insertions(+), 21 deletions(-) diff --git a/client/http/client.go b/client/http/client.go index 437dd7fee37..ae0f7fee347 100644 --- a/client/http/client.go +++ b/client/http/client.go @@ -437,7 +437,7 @@ func (c *client) SetPlacementRule(ctx context.Context, rule *Rule) error { http.MethodPost, bytes.NewBuffer(ruleJSON), nil) } -// SetPlacementRules sets the placement rules in batch. +// SetPlacementRuleInBatch sets the placement rules in batch. func (c *client) SetPlacementRuleInBatch(ctx context.Context, ruleOps []*RuleOp) error { ruleOpsJSON, err := json.Marshal(ruleOps) if err != nil { diff --git a/pkg/utils/tsoutil/tsoutil.go b/pkg/utils/tsoutil/tsoutil.go index 796012ae031..43d8b09aa49 100644 --- a/pkg/utils/tsoutil/tsoutil.go +++ b/pkg/utils/tsoutil/tsoutil.go @@ -25,6 +25,11 @@ const ( logicalBits = (1 << physicalShiftBits) - 1 ) +// TimeToTS converts a `time.Time` to an `uint64` TS. +func TimeToTS(t time.Time) uint64 { + return ComposeTS(t.UnixNano()/int64(time.Millisecond), 0) +} + // ParseTS parses the ts to (physical,logical). func ParseTS(ts uint64) (time.Time, uint64) { physical, logical := ParseTSUint64(ts) diff --git a/server/cluster/cluster.go b/server/cluster/cluster.go index 1c3d8a03a98..ec8ca3a0d65 100644 --- a/server/cluster/cluster.go +++ b/server/cluster/cluster.go @@ -2241,7 +2241,9 @@ func (c *RaftCluster) SetMinResolvedTS(storeID, minResolvedTS uint64) error { return nil } -func (c *RaftCluster) checkAndUpdateMinResolvedTS() (uint64, bool) { +// CheckAndUpdateMinResolvedTS checks and updates the min resolved ts of the cluster. +// This is exported for testing purpose. +func (c *RaftCluster) CheckAndUpdateMinResolvedTS() (uint64, bool) { c.Lock() defer c.Unlock() @@ -2284,7 +2286,7 @@ func (c *RaftCluster) runMinResolvedTSJob() { case <-ticker.C: interval = c.opt.GetMinResolvedTSPersistenceInterval() if interval != 0 { - if current, needPersist := c.checkAndUpdateMinResolvedTS(); needPersist { + if current, needPersist := c.CheckAndUpdateMinResolvedTS(); needPersist { c.storage.SaveMinResolvedTS(current) } } else { diff --git a/tests/integrations/client/http_client_test.go b/tests/integrations/client/http_client_test.go index 2bdd1d5b5ee..25625ac1033 100644 --- a/tests/integrations/client/http_client_test.go +++ b/tests/integrations/client/http_client_test.go @@ -20,6 +20,7 @@ import ( "net/http" "sort" "testing" + "time" "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" @@ -28,6 +29,7 @@ import ( "github.com/tikv/pd/pkg/schedule/labeler" "github.com/tikv/pd/pkg/schedule/placement" "github.com/tikv/pd/pkg/utils/testutil" + "github.com/tikv/pd/pkg/utils/tsoutil" "github.com/tikv/pd/tests" ) @@ -73,27 +75,30 @@ func (suite *httpClientTestSuite) TearDownSuite() { func (suite *httpClientTestSuite) TestGetMinResolvedTSByStoresIDs() { re := suite.Require() - var ( - minResolvedTS uint64 - storeMinResolvedTSMap map[uint64]uint64 - err error - ) - // Wait for the cluster-level min resolved TS to be initialized. + testMinResolvedTS := tsoutil.TimeToTS(time.Now()) + raftCluster := suite.cluster.GetLeaderServer().GetRaftCluster() + err := raftCluster.SetMinResolvedTS(1, testMinResolvedTS) + re.NoError(err) + // Make sure the min resolved TS is updated. testutil.Eventually(re, func() bool { - minResolvedTS, storeMinResolvedTSMap, err = suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, nil) - re.NoError(err) - return minResolvedTS > 0 && len(storeMinResolvedTSMap) == 0 + minResolvedTS, _ := raftCluster.CheckAndUpdateMinResolvedTS() + return minResolvedTS == testMinResolvedTS }) + // Wait for the cluster-level min resolved TS to be initialized. + minResolvedTS, storeMinResolvedTSMap, err := suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, nil) + re.NoError(err) + re.Equal(testMinResolvedTS, minResolvedTS) + re.Empty(storeMinResolvedTSMap) // Get the store-level min resolved TS. minResolvedTS, storeMinResolvedTSMap, err = suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, []uint64{1}) re.NoError(err) - re.Greater(minResolvedTS, uint64(0)) + re.Equal(testMinResolvedTS, minResolvedTS) re.Len(storeMinResolvedTSMap, 1) re.Equal(minResolvedTS, storeMinResolvedTSMap[1]) // Get the store-level min resolved TS with an invalid store ID. minResolvedTS, storeMinResolvedTSMap, err = suite.client.GetMinResolvedTSByStoresIDs(suite.ctx, []uint64{1, 2}) re.NoError(err) - re.Greater(minResolvedTS, uint64(0)) + re.Equal(testMinResolvedTS, minResolvedTS) re.Len(storeMinResolvedTSMap, 2) re.Equal(minResolvedTS, storeMinResolvedTSMap[1]) re.Equal(uint64(math.MaxUint64), storeMinResolvedTSMap[2]) @@ -265,16 +270,17 @@ func (suite *httpClientTestSuite) TestRegionLabel() { func (suite *httpClientTestSuite) TestAccelerateSchedule() { re := suite.Require() - r1 := core.NewTestRegionInfo(10, 1, []byte("a1"), []byte("a2")) - r2 := core.NewTestRegionInfo(11, 1, []byte("a2"), []byte("a3")) raftCluster := suite.cluster.GetLeaderServer().GetRaftCluster() - err := raftCluster.HandleRegionHeartbeat(r1) - re.NoError(err) - err = raftCluster.HandleRegionHeartbeat(r2) - re.NoError(err) + for _, region := range []*core.RegionInfo{ + core.NewTestRegionInfo(10, 1, []byte("a1"), []byte("a2")), + core.NewTestRegionInfo(11, 1, []byte("a2"), []byte("a3")), + } { + err := raftCluster.HandleRegionHeartbeat(region) + re.NoError(err) + } suspectRegions := raftCluster.GetSuspectRegions() re.Len(suspectRegions, 0) - err = suite.client.AccelerateSchedule(suite.ctx, []byte("a1"), []byte("a2")) + err := suite.client.AccelerateSchedule(suite.ctx, []byte("a1"), []byte("a2")) re.NoError(err) suspectRegions = raftCluster.GetSuspectRegions() re.Len(suspectRegions, 1)