From d072fdd5919d8ae1b9e312762ff257123420ffa0 Mon Sep 17 00:00:00 2001 From: nolouch Date: Thu, 4 Jul 2024 15:42:32 +0800 Subject: [PATCH] fix Signed-off-by: nolouch --- client/resource_group/controller/config.go | 22 +++---------- .../resource_group/controller/controller.go | 31 ++++++++++--------- pkg/mcs/resource_manager/server/config.go | 9 +----- .../resource_manager/server/config_test.go | 5 --- .../resource_manager/resource_manager_test.go | 11 +++---- 5 files changed, 26 insertions(+), 52 deletions(-) diff --git a/client/resource_group/controller/config.go b/client/resource_group/controller/config.go index 887f6c718b7..8bcebf09485 100644 --- a/client/resource_group/controller/config.go +++ b/client/resource_group/controller/config.go @@ -53,15 +53,12 @@ const ( defaultTargetPeriod = 5 * time.Second // defaultMaxWaitDuration is the max duration to wait for the token before throwing error. defaultMaxWaitDuration = 30 * time.Second -<<<<<<< HEAD -======= // defaultLTBTokenRPCMaxDelay is the upper bound of backoff delay for local token bucket RPC. defaultLTBTokenRPCMaxDelay = 1 * time.Second // defaultWaitRetryTimes is the times to retry when waiting for the token. defaultWaitRetryTimes = 20 // defaultWaitRetryInterval is the interval to retry when waiting for the token. defaultWaitRetryInterval = 50 * time.Millisecond ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)) ) const ( @@ -76,12 +73,7 @@ const ( // Because the resource manager has not been deployed in microservice mode, // do not enable this function. -<<<<<<< HEAD - defaultDegradedModeWaitDuration = 0 -======= defaultDegradedModeWaitDuration = time.Duration(0) - defaultAvgBatchProportion = 0.7 ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)) ) // TokenRPCParams is the parameters for local bucket RPC. @@ -107,12 +99,9 @@ type BaseConfig struct { // LTBMaxWaitDuration is the max wait time duration for local token bucket. LTBMaxWaitDuration Duration `toml:"ltb-max-wait-duration" json:"ltb-max-wait-duration"` -<<<<<<< HEAD -======= // LTBTokenRPCMaxDelay is the upper bound of backoff delay for local token bucket RPC. LTBTokenRPCMaxDelay Duration `toml:"ltb-token-rpc-max-delay" json:"ltb-token-rpc-max-delay"` ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)) // RequestUnit is the configuration determines the coefficients of the RRU and WRU cost. // This configuration should be modified carefully. RequestUnit RequestUnitConfig `toml:"request-unit" json:"request-unit"` @@ -145,12 +134,6 @@ func (c *Config) Adjust() { // DefaultConfig returns the default resource manager controller configuration. func DefaultConfig() *Config { return &Config{ -<<<<<<< HEAD - DegradedModeWaitDuration: NewDuration(defaultDegradedModeWaitDuration), - LTBMaxWaitDuration: NewDuration(defaultMaxWaitDuration), - RequestUnit: DefaultRequestUnitConfig(), - EnableControllerTraceLog: false, -======= BaseConfig: BaseConfig{ DegradedModeWaitDuration: NewDuration(defaultDegradedModeWaitDuration), RequestUnit: DefaultRequestUnitConfig(), @@ -164,7 +147,6 @@ func DefaultConfig() *Config { WaitRetryTimes: defaultWaitRetryTimes, }, }, ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)) } } @@ -212,6 +194,8 @@ type RUConfig struct { // some config for client LTBMaxWaitDuration time.Duration + WaitRetryInterval time.Duration + WaitRetryTimes int DegradedModeWaitDuration time.Duration } @@ -231,6 +215,8 @@ func GenerateRUConfig(config *Config) *RUConfig { WriteBytesCost: RequestUnit(config.RequestUnit.WriteCostPerByte), CPUMsCost: RequestUnit(config.RequestUnit.CPUMsCost), LTBMaxWaitDuration: config.LTBMaxWaitDuration.Duration, + WaitRetryInterval: config.WaitRetryInterval.Duration, + WaitRetryTimes: config.WaitRetryTimes, DegradedModeWaitDuration: config.DegradedModeWaitDuration.Duration, } } diff --git a/client/resource_group/controller/controller.go b/client/resource_group/controller/controller.go index 1b39a9f338a..1542a33ff1d 100755 --- a/client/resource_group/controller/controller.go +++ b/client/resource_group/controller/controller.go @@ -37,8 +37,6 @@ import ( const ( controllerConfigPath = "resource_group/controller" - maxRetry = 10 - retryInterval = 50 * time.Millisecond maxNotificationChanLen = 200 needTokensAmplification = 1.1 trickleReserveDuration = 1250 * time.Millisecond @@ -100,6 +98,20 @@ func WithMaxWaitDuration(d time.Duration) ResourceControlCreateOption { } } +// WithWaitRetryInterval is the option to set the retry interval when waiting for the token. +func WithWaitRetryInterval(d time.Duration) ResourceControlCreateOption { + return func(controller *ResourceGroupsController) { + controller.ruConfig.WaitRetryInterval = d + } +} + +// WithWaitRetryTimes is the option to set the times to retry when waiting for the token. +func WithWaitRetryTimes(times int) ResourceControlCreateOption { + return func(controller *ResourceGroupsController) { + controller.ruConfig.WaitRetryTimes = times + } +} + var _ ResourceGroupKVInterceptor = (*ResourceGroupsController)(nil) // ResourceGroupsController implements ResourceGroupKVInterceptor. @@ -177,23 +189,14 @@ func loadServerConfig(ctx context.Context, provider ResourceGroupProvider) (*Con if err != nil { return nil, err } -<<<<<<< HEAD - if len(resp.Kvs) == 0 { -======= config := DefaultConfig() defer config.Adjust() kvs := resp.GetKvs() if len(kvs) == 0 { ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)) log.Warn("[resource group controller] server does not save config, load config failed") return config, nil } -<<<<<<< HEAD - config := &Config{} - err = json.Unmarshal(resp.Kvs[0].GetValue(), config) -======= err = json.Unmarshal(kvs[0].GetValue(), config) ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)) if err != nil { return nil, err } @@ -372,7 +375,7 @@ func (c *ResourceGroupsController) Start(ctx context.Context) { } for _, item := range resp { cfgRevision = item.Kv.ModRevision - config := &Config{} + config := DefaultConfig() if err := json.Unmarshal(item.Kv.Value, config); err != nil { continue } @@ -1195,7 +1198,7 @@ func (gc *groupCostController) onRequestWait( var i int var d time.Duration retryLoop: - for i = 0; i < maxRetry; i++ { + for i = 0; i < gc.mainCfg.WaitRetryTimes; i++ { switch gc.mode { case rmpb.GroupMode_RawMode: res := make([]*Reservation, 0, len(requestResourceLimitTypeList)) @@ -1219,7 +1222,7 @@ func (gc *groupCostController) onRequestWait( } } gc.metrics.requestRetryCounter.Inc() - time.Sleep(retryInterval) + time.Sleep(gc.mainCfg.WaitRetryInterval) } if err != nil { if errs.ErrClientResourceGroupThrottled.Equal(err) { diff --git a/pkg/mcs/resource_manager/server/config.go b/pkg/mcs/resource_manager/server/config.go index 49922ba48a2..7104b004c93 100644 --- a/pkg/mcs/resource_manager/server/config.go +++ b/pkg/mcs/resource_manager/server/config.go @@ -108,12 +108,6 @@ func (rmc *ControllerConfig) Adjust(meta *configutil.ConfigMetaData) { if rmc == nil { return } -<<<<<<< HEAD:pkg/mcs/resource_manager/server/config.go - rmc.RequestUnit.Adjust() - - configutil.AdjustDuration(&rmc.DegradedModeWaitDuration, defaultDegradedModeWaitDuration) - configutil.AdjustDuration(&rmc.LTBMaxWaitDuration, defaultMaxWaitDuration) -======= rmc.RequestUnit.Adjust(meta.Child("request-unit")) if !meta.IsDefined("degraded-mode-wait-duration") { configutil.AdjustDuration(&rmc.DegradedModeWaitDuration, defaultDegradedModeWaitDuration) @@ -124,7 +118,6 @@ func (rmc *ControllerConfig) Adjust(meta *configutil.ConfigMetaData) { if !meta.IsDefined("ltb-token-rpc-max-delay") { configutil.AdjustDuration(&rmc.LTBTokenRPCMaxDelay, defaultLTBTokenRPCMaxDelay) } ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)):pkg/mcs/resourcemanager/server/config.go failpoint.Inject("enableDegradedMode", func() { configutil.AdjustDuration(&rmc.DegradedModeWaitDuration, time.Second) }) @@ -149,7 +142,7 @@ type RequestUnitConfig struct { } // Adjust adjusts the configuration and initializes it with the default value if necessary. -func (ruc *RequestUnitConfig) Adjust() { +func (ruc *RequestUnitConfig) Adjust(_ *configutil.ConfigMetaData) { if ruc == nil { return } diff --git a/pkg/mcs/resource_manager/server/config_test.go b/pkg/mcs/resource_manager/server/config_test.go index d6761d1e628..47d1ed36515 100644 --- a/pkg/mcs/resource_manager/server/config_test.go +++ b/pkg/mcs/resource_manager/server/config_test.go @@ -43,14 +43,9 @@ read-cpu-ms-cost = 5.0 err = cfg.Adjust(&meta, false) re.NoError(err) -<<<<<<< HEAD:pkg/mcs/resource_manager/server/config_test.go - re.Equal(cfg.Controller.DegradedModeWaitDuration.Duration, time.Second*2) - re.Equal(cfg.Controller.LTBMaxWaitDuration.Duration, time.Second*60) -======= re.Equal(2*time.Second, cfg.Controller.DegradedModeWaitDuration.Duration) re.Equal(60*time.Second, cfg.Controller.LTBMaxWaitDuration.Duration) re.Equal(500*time.Millisecond, cfg.Controller.LTBTokenRPCMaxDelay.Duration) ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)):pkg/mcs/resourcemanager/server/config_test.go re.LessOrEqual(math.Abs(cfg.Controller.RequestUnit.CPUMsCost-5), 1e-7) re.LessOrEqual(math.Abs(cfg.Controller.RequestUnit.WriteCostPerByte-4), 1e-7) re.LessOrEqual(math.Abs(cfg.Controller.RequestUnit.WriteBaseCost-3), 1e-7) diff --git a/tests/integrations/mcs/resource_manager/resource_manager_test.go b/tests/integrations/mcs/resource_manager/resource_manager_test.go index 502259462f7..e34d00a1ea5 100644 --- a/tests/integrations/mcs/resource_manager/resource_manager_test.go +++ b/tests/integrations/mcs/resource_manager/resource_manager_test.go @@ -33,6 +33,7 @@ import ( "github.com/tikv/pd/client/resource_group/controller" "github.com/tikv/pd/pkg/mcs/resource_manager/server" "github.com/tikv/pd/pkg/utils/testutil" + "github.com/tikv/pd/pkg/utils/typeutil" "github.com/tikv/pd/tests" "go.uber.org/goleak" @@ -1238,10 +1239,6 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupControllerConfigCh tokenRPCMaxDelay := 2 * time.Second readBaseCost := 1.5 defaultCfg := controller.DefaultConfig() -<<<<<<< HEAD:tests/integrations/mcs/resource_manager/resource_manager_test.go - // failpoint enableDegradedMode will setup and set it be 1s. - defaultCfg.DegradedModeWaitDuration.Duration = time.Second -======= expectCfg := server.ControllerConfig{ // failpoint enableDegradedMode will setup and set it be 1s. DegradedModeWaitDuration: typeutil.NewDuration(time.Second), @@ -1250,13 +1247,13 @@ func (suite *resourceManagerClientTestSuite) TestResourceGroupControllerConfigCh RequestUnit: server.RequestUnitConfig(defaultCfg.RequestUnit), EnableControllerTraceLog: defaultCfg.EnableControllerTraceLog, } ->>>>>>> 6b25787af (resource_control: allow configuration of the maximum retry time for the local bucket (#8352)):tests/integrations/mcs/resourcemanager/resource_manager_test.go expectRUCfg := controller.GenerateRUConfig(defaultCfg) + expectRUCfg.DegradedModeWaitDuration = time.Second // initial config verification respString := sendRequest("GET", getAddr()+configURL, nil) - defaultString, err := json.Marshal(defaultCfg) + expectStr, err := json.Marshal(expectCfg) re.NoError(err) - re.JSONEq(string(respString), string(defaultString)) + re.JSONEq(string(respString), string(expectStr)) re.EqualValues(expectRUCfg, c1.GetConfig()) testCases := []struct {