diff --git a/client/resource_manager_client.go b/client/resource_manager_client.go index 9a394c7a503..90cfc977acf 100644 --- a/client/resource_manager_client.go +++ b/client/resource_manager_client.go @@ -36,6 +36,8 @@ const ( groupSettingsPathPrefix = "resource_group/settings" // errNotPrimary is returned when the requested server is not primary. errNotPrimary = "not primary" + // errNotLeader is returned when the requested server is not pd leader. + errNotLeader = "not leader" ) // ResourceManagerClient manages resource group info and token request. @@ -60,7 +62,7 @@ func (c *client) resourceManagerClient() (rmpb.ResourceManagerClient, error) { // gRPCErrorHandler is used to handle the gRPC error returned by the resource manager service. func (c *client) gRPCErrorHandler(err error) { - if strings.Contains(err.Error(), errNotPrimary) { + if strings.Contains(err.Error(), errNotPrimary) || strings.Contains(err.Error(), errNotLeader) { c.pdSvcDiscovery.ScheduleCheckMemberChanged() } } diff --git a/pkg/mcs/resource_manager/server/manager.go b/pkg/mcs/resource_manager/server/manager.go index b9c9b37c7d9..a884b064292 100644 --- a/pkg/mcs/resource_manager/server/manager.go +++ b/pkg/mcs/resource_manager/server/manager.go @@ -126,23 +126,26 @@ func (m *Manager) Init(ctx context.Context) { } m.storage.LoadResourceGroupStates(tokenHandler) - // Add default group - defaultGroup := &ResourceGroup{ - Name: reservedDefaultGroupName, - Mode: rmpb.GroupMode_RUMode, - RUSettings: &RequestUnitSettings{ - RU: &GroupTokenBucket{ - Settings: &rmpb.TokenLimitSettings{ - FillRate: math.MaxInt32, - BurstLimit: -1, + // Add default group if it's not inited. + if _, ok := m.groups[reservedDefaultGroupName]; !ok { + defaultGroup := &ResourceGroup{ + Name: reservedDefaultGroupName, + Mode: rmpb.GroupMode_RUMode, + RUSettings: &RequestUnitSettings{ + RU: &GroupTokenBucket{ + Settings: &rmpb.TokenLimitSettings{ + FillRate: math.MaxInt32, + BurstLimit: -1, + }, }, }, - }, - Priority: middlePriority, - } - if err := m.AddResourceGroup(defaultGroup.IntoProtoResourceGroup()); err != nil { - log.Warn("init default group failed", zap.Error(err)) + Priority: middlePriority, + } + if err := m.AddResourceGroup(defaultGroup.IntoProtoResourceGroup()); err != nil { + log.Warn("init default group failed", zap.Error(err)) + } } + // Start the background metrics flusher. go m.backgroundMetricsFlush(ctx) go func() { diff --git a/tests/integrations/mcs/resource_manager/resource_manager_test.go b/tests/integrations/mcs/resource_manager/resource_manager_test.go index 634c919e437..6dcc5dd6430 100644 --- a/tests/integrations/mcs/resource_manager/resource_manager_test.go +++ b/tests/integrations/mcs/resource_manager/resource_manager_test.go @@ -699,6 +699,19 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() { } }, }, + {"default", rmpb.GroupMode_RUMode, false, true, + `{"name":"default","mode":1,"r_u_settings":{"r_u":{"settings":{"fill_rate":10000,"burst_limit":-1},"state":{"initialized":false}}},"priority":0}`, + func(gs *rmpb.ResourceGroup) { + gs.RUSettings = &rmpb.GroupRequestUnitSettings{ + RU: &rmpb.TokenBucket{ + Settings: &rmpb.TokenLimitSettings{ + FillRate: 10000, + BurstLimit: -1, + }, + }, + } + }, + }, } checkErr := func(err error, success bool) { @@ -865,6 +878,25 @@ func (suite *resourceManagerClientTestSuite) TestBasicResourceGroupCURD() { re.Equal(1, len(groups1)) } } + + // test restart cluster + groups, err := cli.ListResourceGroups(suite.ctx) + re.NoError(err) + servers := suite.cluster.GetServers() + re.NoError(suite.cluster.StopAll()) + serverList := make([]*tests.TestServer, 0, len(servers)) + for _, s := range servers { + serverList = append(serverList, s) + } + re.NoError(suite.cluster.RunServers(serverList)) + suite.cluster.WaitLeader() + var newGroups []*rmpb.ResourceGroup + testutil.Eventually(suite.Require(), func() bool { + var err error + newGroups, err = cli.ListResourceGroups(suite.ctx) + return err == nil + }, testutil.WithWaitFor(time.Second)) + re.Equal(groups, newGroups) } func (suite *resourceManagerClientTestSuite) TestResourceManagerClientFailover() {