feat: qpsTotal, requestsTotal, requestsSucceedTotal

- improve QpsMetric

metrics/api.go

@@ -17,6 +17,19 @@
 
 package metrics
 
+import (
+	"encoding/json"
+	"sync"
+)
+
+import (
+	"github.com/dubbogo/gost/log/logger"
+)
+
+import (
+	"dubbo.apache.org/dubbo-go/v3/metrics/util/aggregate"
+)
+
 var (
 	registries = make(map[string]func(*ReporterConfig) MetricRegistry)
 	collectors = make([]CollectorFunc, 0)
@@ -110,6 +123,10 @@ func NewMetricId(key *MetricKey, level MetricLevel) *MetricId {
 	return &MetricId{Name: key.Name, Desc: key.Desc, Tags: level.Tags()}
 }
 
+func NewMetricIdByLabels(key *MetricKey, labels map[string]string) *MetricId {
+	return &MetricId{Name: key.Name, Desc: key.Desc, Tags: labels}
+}
+
 // MetricSample a metric sample，This is the final data presentation,
 // not an intermediate result(like summary，histogram they will export to a set of MetricSample)
 type MetricSample struct {
@@ -149,3 +166,54 @@ func (c *BaseCollector) StateCount(total, succ, fail *MetricKey, level MetricLev
 		c.R.Counter(NewMetricId(fail, level)).Inc()
 	}
 }
+
+func labelsToString(labels map[string]string) string {
+	labelsJson, err := json.Marshal(labels)
+	if err != nil {
+		logger.Errorf("json.Marshal(labels) = error:%v", err)
+		return ""
+	}
+	return string(labelsJson)
+}
+
+type QpsMetric interface {
+	Record(labels map[string]string)
+}
+
+func NewQpsMetric(metricKey *MetricKey, metricRegistry MetricRegistry) QpsMetric {
+	return &DefaultQpsMetric{
+		metricRegistry: metricRegistry,
+		metricKey:      metricKey,
+		mux:            sync.RWMutex{},
+		cache:          make(map[string]*aggregate.TimeWindowCounter),
+	}
+}
+
+type DefaultQpsMetric struct {
+	metricRegistry MetricRegistry
+	metricKey      *MetricKey
+	mux            sync.RWMutex
+	cache          map[string]*aggregate.TimeWindowCounter // key: metrics labels, value: TimeWindowCounter
+}
+
+func (d *DefaultQpsMetric) Record(labels map[string]string) {
+	key := labelsToString(labels)
+	if key == "" {
+		return
+	}
+	d.mux.RLock()
+	twc, ok := d.cache[key]
+	d.mux.RUnlock()
+	if !ok {
+		d.mux.Lock()
+		twc, ok = d.cache[key]
+		if !ok {
+			twc = aggregate.NewTimeWindowCounter(defaultBucketNum, defaultTimeWindowSeconds)
+			d.cache[key] = twc
+		}
+		d.mux.Unlock()
+	}
+	twc.Inc()
+	d.metricRegistry.Gauge(NewMetricIdByLabels(d.metricKey, labels)).Set(twc.Count() / float64(twc.LivedSeconds()))
+
+}

metrics/rpc/collector.go

@@ -28,10 +28,10 @@ var (
 
 func init() {
 	var collectorFunc metrics.CollectorFunc
-	collectorFunc = func(exporter metrics.MetricRegistry, c *metrics.ReporterConfig) {
+	collectorFunc = func(registry metrics.MetricRegistry, c *metrics.ReporterConfig) {
 		rc := &rpcCollector{
-			exporter:  exporter,
-			metricSet: buildMetricSet(),
+			registry:  registry,
+			metricSet: buildMetricSet(registry),
 		}
 		go rc.start()
 	}
@@ -40,7 +40,7 @@ func init() {
 }
 
 type rpcCollector struct {
-	exporter  metrics.MetricRegistry
+	registry  metrics.MetricRegistry
 	metricSet *metricSet
 }
 
@@ -67,6 +67,7 @@ func (c *rpcCollector) beforeInvokeHandler(event *MetricsEvent) {
 		return
 	}
 	labels := buildLabels(url)
+	c.recordQps(role, labels)
 	c.incRequestsProcessingTotal(role, labels)
 }
 
@@ -78,27 +79,60 @@ func (c *rpcCollector) afterInvokeHandler(event *MetricsEvent) {
 		return
 	}
 	labels := buildLabels(url)
+	c.incRequestsTotal(role, labels)
 	c.decRequestsProcessingTotal(role, labels)
+	if event.result != nil {
+		if event.result.Error() == nil {
+			c.incRequestsSucceedTotal(role, labels)
+		}
+	}
 }
 
 func newMetricId(key *metrics.MetricKey, labels map[string]string) *metrics.MetricId {
 	return &metrics.MetricId{Name: key.Name, Desc: key.Desc, Tags: labels}
 }
 
+func (c *rpcCollector) recordQps(role string, labels map[string]string) {
+	switch role {
+	case providerField:
+		c.metricSet.provider.qpsTotal.Record(labels)
+	case consumerField:
+		c.metricSet.consumer.qpsTotal.Record(labels)
+	}
+}
+
+func (c *rpcCollector) incRequestsTotal(role string, labels map[string]string) {
+	switch role {
+	case providerField:
+		c.registry.Counter(newMetricId(c.metricSet.provider.requestsTotal, labels)).Inc()
+	case consumerField:
+		c.registry.Counter(newMetricId(c.metricSet.consumer.requestsTotal, labels)).Inc()
+	}
+}
+
 func (c *rpcCollector) incRequestsProcessingTotal(role string, labels map[string]string) {
 	switch role {
 	case providerField:
-		c.exporter.Gauge(newMetricId(c.metricSet.provider.requestsProcessingTotal, labels)).Inc()
+		c.registry.Gauge(newMetricId(c.metricSet.provider.requestsProcessingTotal, labels)).Inc()
 	case consumerField:
-		c.exporter.Gauge(newMetricId(c.metricSet.consumer.requestsProcessingTotal, labels)).Inc()
+		c.registry.Gauge(newMetricId(c.metricSet.consumer.requestsProcessingTotal, labels)).Inc()
 	}
 }
 
 func (c *rpcCollector) decRequestsProcessingTotal(role string, labels map[string]string) {
 	switch role {
 	case providerField:
-		c.exporter.Gauge(newMetricId(c.metricSet.provider.requestsProcessingTotal, labels)).Dec()
+		c.registry.Gauge(newMetricId(c.metricSet.provider.requestsProcessingTotal, labels)).Dec()
+	case consumerField:
+		c.registry.Gauge(newMetricId(c.metricSet.consumer.requestsProcessingTotal, labels)).Dec()
+	}
+}
+
+func (c *rpcCollector) incRequestsSucceedTotal(role string, labels map[string]string) {
+	switch role {
+	case providerField:
+		c.registry.Counter(newMetricId(c.metricSet.provider.requestsSucceedTotal, labels)).Inc()
 	case consumerField:
-		c.exporter.Gauge(newMetricId(c.metricSet.consumer.requestsProcessingTotal, labels)).Dec()
+		c.registry.Counter(newMetricId(c.metricSet.consumer.requestsSucceedTotal, labels)).Inc()
 	}
 }

metrics/rpc/constant.go

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package rpc
+
+import "dubbo.apache.org/dubbo-go/v3/common/constant"
+
+const (
+	applicationNameKey = constant.ApplicationNameKey
+	groupKey           = constant.GroupKey
+	hostnameKey        = constant.HostnameKey
+	interfaceKey       = constant.InterfaceKey
+	ipKey              = constant.IpKey
+	methodKey          = constant.MethodKey
+	versionKey         = constant.VersionKey
+)
+
+const (
+	providerField = "provider"
+	consumerField = "consumer"
+
+	qpsField      = "qps"
+	requestsField = "requests"
+	rtField       = "rt"
+
+	milliSecondsField = "milliseconds"
+
+	minField  = "min"
+	maxField  = "max"
+	sumField  = "sum"
+	avgField  = "avg"
+	lastField = "last"
+
+	totalField      = "total"
+	aggregateField  = "aggregate"
+	processingField = "processing"
+	succeedField    = "succeed"
+)

metrics/rpc/metric_set.go

@@ -35,11 +35,11 @@ type consumerMetrics struct {
 }
 
 type rpcCommonMetrics struct {
-	//qpsTotal                      *qpsGaugeVec
-	//requestsTotal                 *prometheus.CounterVec
+	qpsTotal      metrics.QpsMetric
+	requestsTotal *metrics.MetricKey
 	//requestsTotalAggregate        *aggregateCounterGaugeVec
 	requestsProcessingTotal *metrics.MetricKey
-	//requestsSucceedTotal          *prometheus.CounterVec
+	requestsSucceedTotal    *metrics.MetricKey
 	//requestsSucceedTotalAggregate *aggregateCounterGaugeVec
 	//rtMillisecondsMin             *GaugeVecWithSyncMap
 	//rtMillisecondsMax             *GaugeVecWithSyncMap
@@ -50,20 +50,26 @@ type rpcCommonMetrics struct {
 	//rtMillisecondsAggregate       *aggregateFunctionsGaugeVec
 }
 
-func buildMetricSet() *metricSet {
+func buildMetricSet(registry metrics.MetricRegistry) *metricSet {
 	ms := &metricSet{
 		provider: &providerMetrics{},
 		consumer: &consumerMetrics{},
 	}
-	ms.provider.init()
-	ms.consumer.init()
+	ms.provider.init(registry)
+	ms.consumer.init(registry)
 	return ms
 }
 
-func (pm *providerMetrics) init() {
+func (pm *providerMetrics) init(registry metrics.MetricRegistry) {
+	pm.qpsTotal = metrics.NewQpsMetric(metrics.NewMetricKey("dubbo_provider_qps_total", "The number of requests received by the provider per second"), registry)
+	pm.requestsTotal = metrics.NewMetricKey("dubbo_provider_requests_total", "The total number of received requests by the provider")
 	pm.requestsProcessingTotal = metrics.NewMetricKey("dubbo_provider_requests_processing_total", "The number of received requests being processed by the provider")
+	pm.requestsSucceedTotal = metrics.NewMetricKey("dubbo_provider_requests_succeed_total", "The number of requests successfully received by the provider")
 }
 
-func (cm *consumerMetrics) init() {
+func (cm *consumerMetrics) init(registry metrics.MetricRegistry) {
+	cm.qpsTotal = metrics.NewQpsMetric(metrics.NewMetricKey("dubbo_consumer_qps_total", "The number of requests sent by consumers per second"), registry)
+	cm.requestsTotal = metrics.NewMetricKey("dubbo_consumer_requests_total", "The total number of sent requests by consumers")
 	cm.requestsProcessingTotal = metrics.NewMetricKey("dubbo_consumer_requests_processing_total", "The number of received requests being processed by the consumer")
+	cm.requestsSucceedTotal = metrics.NewMetricKey("dubbo_consumer_requests_succeed_total", "The number of successful requests sent by consumers")
 }

metrics/rpc/util.go

@@ -29,16 +29,6 @@ import (
 	"dubbo.apache.org/dubbo-go/v3/common/constant"
 )
 
-const (
-	applicationNameKey = constant.ApplicationNameKey
-	groupKey           = constant.GroupKey
-	hostnameKey        = constant.HostnameKey
-	interfaceKey       = constant.InterfaceKey
-	ipKey              = constant.IpKey
-	methodKey          = constant.MethodKey
-	versionKey         = constant.VersionKey
-)
-
 func buildLabels(url *common.URL) map[string]string {
 	return map[string]string{
 		applicationNameKey: url.GetParam(constant.ApplicationKey, ""),
@@ -51,11 +41,6 @@ func buildLabels(url *common.URL) map[string]string {
 	}
 }
 
-const (
-	providerField = "provider"
-	consumerField = "consumer"
-)
-
 func getRole(url *common.URL) (role string) {
 	if isProvider(url) {
 		role = providerField