resource_manager: add metrics for avaiable RU (#6523)

ref #4399

Signed-off-by: Cabinfever_B <cabinfeveroier@gmail.com>

Co-authored-by: ti-chi-bot[bot] <108142056+ti-chi-bot[bot]@users.noreply.github.com>

pkg/mcs/resourcemanager/server/apis/v1/api.go

@@ -151,7 +151,7 @@ func (s *Service) putResourceGroup(c *gin.Context) {
 //
 //	@Tags		ResourceManager
 //	@Summary	Get resource group by name.
-//	@Success	200		{string}	json	format	of	rmpb.ResourceGroup
+//	@Success	200		{string}	json	format	of	rmserver.ResourceGroup
 //	@Failure	404		{string}	error
 //	@Param		name	path		string	true	"groupName"
 //	@Router		/config/group/{name} [GET]
@@ -167,7 +167,7 @@ func (s *Service) getResourceGroup(c *gin.Context) {
 //
 //	@Tags		ResourceManager
 //	@Summary	get all resource group with a list.
-//	@Success	200	{string}	json	format	of	[]rmpb.ResourceGroup
+//	@Success	200	{string}	json	format	of	[]rmserver.ResourceGroup
 //	@Failure	404	{string}	error
 //	@Router		/config/groups [GET]
 func (s *Service) getResourceGroupList(c *gin.Context) {

pkg/mcs/resourcemanager/server/manager.go

@@ -38,6 +38,7 @@ const (
 	defaultConsumptionChanSize = 1024
 	metricsCleanupInterval     = time.Minute
 	metricsCleanupTimeout      = 20 * time.Minute
+	metricsAvailableRUInterval = 30 * time.Second
 
 	reservedDefaultGroupName = "default"
 	middlePriority           = 8
@@ -287,8 +288,10 @@ func (m *Manager) persistResourceGroupRunningState() {
 // Receive the consumption and flush it to the metrics.
 func (m *Manager) backgroundMetricsFlush(ctx context.Context) {
 	defer logutil.LogPanic()
-	ticker := time.NewTicker(metricsCleanupInterval)
-	defer ticker.Stop()
+	cleanUpTicker := time.NewTicker(metricsCleanupInterval)
+	defer cleanUpTicker.Stop()
+	availableRUTicker := time.NewTicker(metricsAvailableRUInterval)
+	defer availableRUTicker.Stop()
 	for {
 		select {
 		case <-ctx.Done():
@@ -342,7 +345,7 @@ func (m *Manager) backgroundMetricsFlush(ctx context.Context) {
 
 			m.consumptionRecord[name] = time.Now()
 
-		case <-ticker.C:
+		case <-cleanUpTicker.C:
 			// Clean up the metrics that have not been updated for a long time.
 			for name, lastTime := range m.consumptionRecord {
 				if time.Since(lastTime) > metricsCleanupTimeout {
@@ -355,9 +358,23 @@ func (m *Manager) backgroundMetricsFlush(ctx context.Context) {
 					sqlCPUCost.DeleteLabelValues(name)
 					requestCount.DeleteLabelValues(name, readTypeLabel)
 					requestCount.DeleteLabelValues(name, writeTypeLabel)
+					availableRUCounter.DeleteLabelValues(name)
 					delete(m.consumptionRecord, name)
 				}
 			}
+		case <-availableRUTicker.C:
+			m.RLock()
+			for name, group := range m.groups {
+				if name == reservedDefaultGroupName {
+					continue
+				}
+				ru := group.getRUToken()
+				if ru < 0 {
+					ru = 0
+				}
+				availableRUCounter.WithLabelValues(name).Set(ru)
+			}
+			m.RUnlock()
 		}
 	}
 }

pkg/mcs/resourcemanager/server/metrics.go

@@ -96,6 +96,14 @@ var (
 			Name:      "request_count",
 			Help:      "The number of read/write requests for all resource groups.",
 		}, []string{resourceGroupNameLabel, typeLabel})
+
+	availableRUCounter = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: namespace,
+			Subsystem: ruSubsystem,
+			Name:      "available_ru",
+			Help:      "Counter of the available RU for all resource groups.",
+		}, []string{resourceGroupNameLabel})
 )
 
 func init() {
@@ -108,4 +116,5 @@ func init() {
 	prometheus.MustRegister(kvCPUCost)
 	prometheus.MustRegister(sqlCPUCost)
 	prometheus.MustRegister(requestCount)
+	prometheus.MustRegister(availableRUCounter)
 }

pkg/mcs/resourcemanager/server/resource_group.go

@@ -76,6 +76,12 @@ func (rg *ResourceGroup) Copy() *ResourceGroup {
 	return &newRG
 }
 
+func (rg *ResourceGroup) getRUToken() float64 {
+	rg.Lock()
+	defer rg.Unlock()
+	return rg.RUSettings.RU.Tokens
+}
+
 // PatchSettings patches the resource group settings.
 // Only used to patch the resource group when updating.
 // Note: the tokens is the delta value to patch.