From 45bccd7e9ffbda4b9bbe9c1e1da70b22430e8d44 Mon Sep 17 00:00:00 2001 From: Ryan Leung Date: Thu, 16 Jul 2026 10:23:22 +0800 Subject: [PATCH] test: stabilize resource manager client suite Wait for standalone resource manager metadata and RU statistics to become observable before asserting their values. Measure token bucket wait time using the duration returned by the controller so host scheduling delays do not make timing assertions flaky. Signed-off-by: Ryan Leung --- .../resourcemanager/resource_manager_test.go | 71 +++++++++++++++---- 1 file changed, 58 insertions(+), 13 deletions(-) diff --git a/tests/integrations/mcs/resourcemanager/resource_manager_test.go b/tests/integrations/mcs/resourcemanager/resource_manager_test.go index d30a5cb372..00a970dd7a 100644 --- a/tests/integrations/mcs/resourcemanager/resource_manager_test.go +++ b/tests/integrations/mcs/resourcemanager/resource_manager_test.go @@ -727,7 +727,14 @@ func (suite *resourceManagerClientTestSuite) TestWatchResourceGroup() { re.NoError(err) re.Contains(resp, "Success!") // Make sure the resource group active - meta, err = controller.GetResourceGroup(group.Name) + testutil.Eventually(re, func() bool { + meta, err = controller.GetResourceGroup(group.Name) + if err != nil || meta == nil { + return false + } + meta = controller.GetActiveResourceGroup(group.Name) + return meta != nil + }, testutil.WithTickInterval(50*time.Millisecond)) re.NotNil(meta) re.NoError(err) modifySettings(group, 30000) @@ -736,7 +743,7 @@ func (suite *resourceManagerClientTestSuite) TestWatchResourceGroup() { re.Contains(resp, "Success!") testutil.Eventually(re, func() bool { meta = controller.GetActiveResourceGroup(group.Name) - return meta.RUSettings.RU.Settings.FillRate == uint64(30000) + return meta != nil && meta.RUSettings.RU.Settings.FillRate == uint64(30000) }, testutil.WithTickInterval(100*time.Millisecond)) re.NoError(failpoint.Disable("github.com/tikv/pd/client/resource_group/controller/watchStreamError")) @@ -791,6 +798,12 @@ func (suite *resourceManagerClientTestSuite) TestWatchWithSingleGroupByKeyspace( resp, err := cli.AddResourceGroup(suite.ctx, group) re.NoError(err) re.Contains(resp, "Success!") + // Under standalone RM discovery, metadata writes are applied on PD first and + // become visible to the RM service after the metadata watcher catches up. + testutil.Eventually(re, func() bool { + _, getErr := cli.GetResourceGroup(suite.ctx, group.Name) + return getErr == nil + }, testutil.WithTickInterval(50*time.Millisecond)) tcs := tokenConsumptionPerSecond{rruTokensAtATime: 100} _, _, _, _, err = controller.OnRequestWait(suite.ctx, group.Name, tcs.makeReadRequest()) @@ -931,12 +944,12 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupController() { wreq := cas.tcs[i].makeWriteRequest() rres := cas.tcs[i].makeReadResponse() wres := cas.tcs[i].makeWriteResponse() - startTime := time.Now() - _, _, _, _, err := rgsController.OnRequestWait(suite.ctx, cas.resourceGroupName, rreq) + _, _, waitDuration, _, err := rgsController.OnRequestWait(suite.ctx, cas.resourceGroupName, rreq) re.NoError(err) - _, _, _, _, err = rgsController.OnRequestWait(suite.ctx, cas.resourceGroupName, wreq) + sum += waitDuration + _, _, waitDuration, _, err = rgsController.OnRequestWait(suite.ctx, cas.resourceGroupName, wreq) re.NoError(err) - sum += time.Since(startTime) + sum += waitDuration _, err = rgsController.OnResponse(cas.resourceGroupName, rreq, rres) re.NoError(err) _, err = rgsController.OnResponse(cas.resourceGroupName, wreq, wres) @@ -1622,10 +1635,28 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupRUConsumption() { }}, }, } + waitForResourceGroup := func( + cli pd.Client, + check func(*rmpb.ResourceGroup) bool, + opts ...pd.GetResourceGroupOption, + ) (*rmpb.ResourceGroup, error) { + var ( + actual *rmpb.ResourceGroup + getErr error + ) + testutil.Eventually(re, func() bool { + actual, getErr = cli.GetResourceGroup(suite.ctx, group.Name, opts...) + return getErr == nil && actual != nil && check(actual) + }, testutil.WithTickInterval(50*time.Millisecond)) + return actual, getErr + } + _, err := cli.AddResourceGroup(suite.ctx, group) re.NoError(err) - g, err := cli.GetResourceGroup(suite.ctx, group.Name) + g, err := waitForResourceGroup(cli, func(actual *rmpb.ResourceGroup) bool { + return reflect.DeepEqual(group, actual) + }) re.NoError(err) re.Equal(group, g) @@ -1654,8 +1685,9 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupRUConsumption() { ClientUniqueId: 1, }) re.NoError(err) - time.Sleep(10 * time.Millisecond) - g, err = cli.GetResourceGroup(suite.ctx, group.Name, pd.WithRUStats) + g, err = waitForResourceGroup(cli, func(actual *rmpb.ResourceGroup) bool { + return reflect.DeepEqual(actual.RUStats, testConsumption) + }, pd.WithRUStats) re.NoError(err) re.Equal(g.RUStats, testConsumption) @@ -1663,7 +1695,9 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupRUConsumption() { g.RUSettings.RU.Settings.FillRate = 12345 _, err = cli.ModifyResourceGroup(suite.ctx, g) re.NoError(err) - g1, err := cli.GetResourceGroup(suite.ctx, group.Name, pd.WithRUStats) + g1, err := waitForResourceGroup(cli, func(actual *rmpb.ResourceGroup) bool { + return reflect.DeepEqual(g, actual) + }, pd.WithRUStats) re.NoError(err) re.Equal(g1, g) @@ -1676,7 +1710,9 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupRUConsumption() { suite.client = suite.setupPDClient(re) cli = suite.client // check ru stats not loss after restart - g, err = cli.GetResourceGroup(suite.ctx, group.Name, pd.WithRUStats) + g, err = waitForResourceGroup(cli, func(actual *rmpb.ResourceGroup) bool { + return reflect.DeepEqual(actual.RUStats, testConsumption) + }, pd.WithRUStats) re.NoError(err) re.Equal(g.RUStats, testConsumption) } @@ -1695,7 +1731,11 @@ func (suite *resourceManagerClientTestSuite) TestResourceManagerClientFailover() addResp, err := cli.AddResourceGroup(suite.ctx, group) re.NoError(err) re.Contains(addResp, "Success!") - getResp, err := cli.GetResourceGroup(suite.ctx, group.GetName()) + var getResp *rmpb.ResourceGroup + testutil.Eventually(re, func() bool { + getResp, err = cli.GetResourceGroup(suite.ctx, group.GetName()) + return err == nil && getResp != nil && reflect.DeepEqual(*group, *getResp) + }, testutil.WithTickInterval(50*time.Millisecond)) re.NoError(err) re.NotNil(getResp) re.Equal(*group, *getResp) @@ -1707,7 +1747,12 @@ func (suite *resourceManagerClientTestSuite) TestResourceManagerClientFailover() re.NoError(err) re.Contains(modifyResp, "Success!") suite.resignAndWaitLeader(re) - getResp, err = cli.GetResourceGroup(suite.ctx, group.GetName()) + testutil.Eventually(re, func() bool { + getResp, err = cli.GetResourceGroup(suite.ctx, group.GetName()) + return err == nil && getResp != nil && + group.GetRUSettings().GetRU().GetSettings().GetFillRate() == + getResp.GetRUSettings().GetRU().GetSettings().GetFillRate() + }, testutil.WithTickInterval(50*time.Millisecond)) re.NoError(err) re.NotNil(getResp) re.Equal(group.RUSettings.RU.Settings.FillRate, getResp.RUSettings.RU.Settings.FillRate)