address

Signed-off-by: nolouch <[email protected]>

Author: nolouch

client/resource_group/controller/controller.go

@@ -304,110 +304,112 @@ func (c *ResourceGroupsController) Start(ctx context.Context) {
 				c.run.inDegradedMode = true
 				c.executeOnAllGroups((*groupCostController).applyDegradedMode)
 				log.Warn("[resource group controller] enter degraded mode")
-
-			/* tickers */
-			case <-cleanupTicker.C:
-				c.cleanUpResourceGroup()
-			case <-stateUpdateTicker.C:
-				c.executeOnAllGroups((*groupCostController).updateRunState)
-				c.executeOnAllGroups((*groupCostController).updateAvgRequestResourcePerSec)
-				if len(c.run.currentRequests) == 0 {
-					c.collectTokenBucketRequests(c.loopCtx, FromPeriodReport, periodicReport /* select resource groups which should be reported periodically */)
-				}
-			case <-watchRetryTimer.C:
-				if !c.ruConfig.isSingleGroupByKeyspace && watchMetaChannel == nil {
-					// Use WithPrevKV() to get the previous key-value pair when get Delete Event.
-					watchMetaChannel, err = c.provider.Watch(ctx, pd.GroupSettingsPathPrefixBytes, pd.WithRev(metaRevision), pd.WithPrefix(), pd.WithPrevKV())
-					if err != nil {
-						log.Warn("watch resource group meta failed", zap.Error(err))
+			default:
+				select {
+				/* tickers */
+				case <-cleanupTicker.C:
+					c.cleanUpResourceGroup()
+				case <-stateUpdateTicker.C:
+					c.executeOnAllGroups((*groupCostController).updateRunState)
+					c.executeOnAllGroups((*groupCostController).updateAvgRequestResourcePerSec)
+					if len(c.run.currentRequests) == 0 {
+						c.collectTokenBucketRequests(c.loopCtx, FromPeriodReport, periodicReport /* select resource groups which should be reported periodically */)
+					}
+				case <-watchRetryTimer.C:
+					if !c.ruConfig.isSingleGroupByKeyspace && watchMetaChannel == nil {
+						// Use WithPrevKV() to get the previous key-value pair when get Delete Event.
+						watchMetaChannel, err = c.provider.Watch(ctx, pd.GroupSettingsPathPrefixBytes, pd.WithRev(metaRevision), pd.WithPrefix(), pd.WithPrevKV())
+						if err != nil {
+							log.Warn("watch resource group meta failed", zap.Error(err))
+							watchRetryTimer.Reset(watchRetryInterval)
+							failpoint.Inject("watchStreamError", func() {
+								watchRetryTimer.Reset(20 * time.Millisecond)
+							})
+						}
+					}
+					if watchConfigChannel == nil {
+						watchConfigChannel, err = c.provider.Watch(ctx, pd.ControllerConfigPathPrefixBytes, pd.WithRev(cfgRevision), pd.WithPrefix())
+						if err != nil {
+							log.Warn("watch resource group config failed", zap.Error(err))
+							watchRetryTimer.Reset(watchRetryInterval)
+						}
+					}
+				case <-emergencyTokenAcquisitionTicker.C:
+					c.executeOnAllGroups((*groupCostController).resetEmergencyTokenAcquisition)
+				/* channels */
+				case <-c.loopCtx.Done():
+					resourceGroupStatusGauge.Reset()
+					return
+				case resp, ok := <-watchMetaChannel:
+					failpoint.Inject("disableWatch", func() {
+						if c.ruConfig.isSingleGroupByKeyspace {
+							panic("disableWatch")
+						}
+					})
+					if !ok {
+						watchMetaChannel = nil
 						watchRetryTimer.Reset(watchRetryInterval)
 						failpoint.Inject("watchStreamError", func() {
 							watchRetryTimer.Reset(20 * time.Millisecond)
 						})
+						continue
 					}
-				}
-				if watchConfigChannel == nil {
-					watchConfigChannel, err = c.provider.Watch(ctx, pd.ControllerConfigPathPrefixBytes, pd.WithRev(cfgRevision), pd.WithPrefix())
-					if err != nil {
-						log.Warn("watch resource group config failed", zap.Error(err))
-						watchRetryTimer.Reset(watchRetryInterval)
-					}
-				}
-			case <-emergencyTokenAcquisitionTicker.C:
-				c.executeOnAllGroups((*groupCostController).resetEmergencyTokenAcquisition)
-			/* channels */
-			case <-c.loopCtx.Done():
-				resourceGroupStatusGauge.Reset()
-				return
-			case resp, ok := <-watchMetaChannel:
-				failpoint.Inject("disableWatch", func() {
-					if c.ruConfig.isSingleGroupByKeyspace {
-						panic("disableWatch")
-					}
-				})
-				if !ok {
-					watchMetaChannel = nil
-					watchRetryTimer.Reset(watchRetryInterval)
-					failpoint.Inject("watchStreamError", func() {
-						watchRetryTimer.Reset(20 * time.Millisecond)
-					})
-					continue
-				}
-				for _, item := range resp {
-					metaRevision = item.Kv.ModRevision
-					group := &rmpb.ResourceGroup{}
-					switch item.Type {
-					case meta_storagepb.Event_PUT:
-						if err = proto.Unmarshal(item.Kv.Value, group); err != nil {
-							continue
-						}
-						if item, ok := c.groupsController.Load(group.Name); ok {
-							gc := item.(*groupCostController)
-							gc.modifyMeta(group)
-						}
-					case meta_storagepb.Event_DELETE:
-						if item.PrevKv != nil {
-							if err = proto.Unmarshal(item.PrevKv.Value, group); err != nil {
+					for _, item := range resp {
+						metaRevision = item.Kv.ModRevision
+						group := &rmpb.ResourceGroup{}
+						switch item.Type {
+						case meta_storagepb.Event_PUT:
+							if err = proto.Unmarshal(item.Kv.Value, group); err != nil {
 								continue
 							}
-							if _, ok := c.groupsController.LoadAndDelete(group.Name); ok {
-								resourceGroupStatusGauge.DeleteLabelValues(group.Name, group.Name)
+							if item, ok := c.groupsController.Load(group.Name); ok {
+								gc := item.(*groupCostController)
+								gc.modifyMeta(group)
+							}
+						case meta_storagepb.Event_DELETE:
+							if item.PrevKv != nil {
+								if err = proto.Unmarshal(item.PrevKv.Value, group); err != nil {
+									continue
+								}
+								if _, ok := c.groupsController.LoadAndDelete(group.Name); ok {
+									resourceGroupStatusGauge.DeleteLabelValues(group.Name, group.Name)
+								}
+							} else {
+								// Prev-kv is compacted means there must have been a delete event before this event,
+								// which means that this is just a duplicated event, so we can just ignore it.
+								log.Info("previous key-value pair has been compacted", zap.String("required-key", string(item.Kv.Key)), zap.String("value", string(item.Kv.Value)))
 							}
-						} else {
-							// Prev-kv is compacted means there must have been a delete event before this event,
-							// which means that this is just a duplicated event, so we can just ignore it.
-							log.Info("previous key-value pair has been compacted", zap.String("required-key", string(item.Kv.Key)), zap.String("value", string(item.Kv.Value)))
 						}
 					}
-				}
-			case resp, ok := <-watchConfigChannel:
-				if !ok {
-					watchConfigChannel = nil
-					watchRetryTimer.Reset(watchRetryInterval)
-					failpoint.Inject("watchStreamError", func() {
-						watchRetryTimer.Reset(20 * time.Millisecond)
-					})
-					continue
-				}
-				for _, item := range resp {
-					cfgRevision = item.Kv.ModRevision
-					config := DefaultConfig()
-					if err := json.Unmarshal(item.Kv.Value, config); err != nil {
+				case resp, ok := <-watchConfigChannel:
+					if !ok {
+						watchConfigChannel = nil
+						watchRetryTimer.Reset(watchRetryInterval)
+						failpoint.Inject("watchStreamError", func() {
+							watchRetryTimer.Reset(20 * time.Millisecond)
+						})
 						continue
 					}
-					config.Adjust()
-					c.ruConfig = GenerateRUConfig(config)
+					for _, item := range resp {
+						cfgRevision = item.Kv.ModRevision
+						config := DefaultConfig()
+						if err := json.Unmarshal(item.Kv.Value, config); err != nil {
+							continue
+						}
+						config.Adjust()
+						c.ruConfig = GenerateRUConfig(config)
 
-					// Stay compatible with serverless
-					for _, opt := range c.opts {
-						opt(c)
-					}
-					copyCfg := *c.ruConfig
-					c.safeRuConfig.Store(&copyCfg)
-					if enableControllerTraceLog.Load() != config.EnableControllerTraceLog {
-						enableControllerTraceLog.Store(config.EnableControllerTraceLog)
+						// Stay compatible with serverless
+						for _, opt := range c.opts {
+							opt(c)
+						}
+						copyCfg := *c.ruConfig
+						c.safeRuConfig.Store(&copyCfg)
+						if enableControllerTraceLog.Load() != config.EnableControllerTraceLog {
+							enableControllerTraceLog.Store(config.EnableControllerTraceLog)
+						}
+						log.Info("load resource controller config after config changed", zap.Reflect("config", config), zap.Reflect("ruConfig", c.ruConfig))
 					}
-					log.Info("load resource controller config after config changed", zap.Reflect("config", config), zap.Reflect("ruConfig", c.ruConfig))
 				}
 			}
 		}