feat: add some metrics about RT (#2340)

* feat: add min RT metrics

* refactor: adapt to Go 1.17 and bundle gaugeVec with syncMap

* feat: add max and sum RT metrics

* feat: avg RT metrics

* refactor: RT sum metrics

* feat: last RT metrics

* refactor: file names and method names

metrics/prometheus/constant.go

@@ -34,12 +34,14 @@ const (
 
 	requestsField = "requests"
 	rtField       = "rt"
-	tpsField      = "tps"
 
 	milliSecondsField = "milliseconds"
 
-	counterField = "counter"
-	summaryField = "summary"
+	minField  = "min"
+	maxField  = "max"
+	sumField  = "sum"
+	avgField  = "avg"
+	lastField = "last"
 
 	totalField      = "total"
 	processingField = "processing"

metrics/prometheus/metric_set.go

@@ -29,40 +29,56 @@ import (
 	"dubbo.apache.org/dubbo-go/v3/metrics"
 )
 
-// metricSet is a set of metrics that are reported to prometheus in dubbo go
+// metricSet is a set of metrics that are reported to prometheus in dubbo-go
 type metricSet struct {
-	// report the consumer-side's rt gauge data
-	consumerRTSummaryVec *prometheus.SummaryVec
-	// report the provider-side's rt gauge data
-	providerRTSummaryVec *prometheus.SummaryVec
+	provider providerMetrics
+	consumer consumerMetrics
+}
+
+func (ms *metricSet) init(reporterConfig *metrics.ReporterConfig) {
+	ms.provider.init(reporterConfig)
+	ms.consumer.init(reporterConfig)
+}
 
-	// report the provider-side's request total counter data
-	providerRequestsTotalCounterVec *prometheus.CounterVec
-	// report the provider-side's processing request counter data
-	providerRequestsProcessingTotalGaugeVec *prometheus.GaugeVec
-	// The number of requests successfully received by the provider
-	providerRequestsSucceedTotalCounterVec *prometheus.CounterVec
+type rpcCommonMetrics struct {
+	requestsTotal           *prometheus.CounterVec
+	requestsProcessingTotal *prometheus.GaugeVec
+	requestsSucceedTotal    *prometheus.CounterVec
+	rtMillisecondsMin       *GaugeVecWithSyncMap
+	rtMillisecondsMax       *GaugeVecWithSyncMap
+	rtMillisecondsSum       *prometheus.CounterVec
+	rtMillisecondsAvg       *GaugeVecWithSyncMap
+	rtMillisecondsLast      *prometheus.GaugeVec
+}
 
-	// report the consumer-side's request total counter data
-	consumerRequestsTotalCounterVec *prometheus.CounterVec
-	// report the consumer-side's processing request counter data
-	consumerRequestsProcessingTotalGaugeVec *prometheus.GaugeVec
-	// The number of successful requests sent by consumers
-	consumerRequestsSucceedTotalCounterVec *prometheus.CounterVec
+type providerMetrics struct {
+	rpcCommonMetrics
 }
 
-var labelNames = []string{applicationNameKey, groupKey, hostnameKey, interfaceKey, ipKey, methodKey, versionKey}
+func (pm *providerMetrics) init(reporterConfig *metrics.ReporterConfig) {
+	pm.requestsTotal = newAutoCounterVec(buildMetricsName(providerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
+	pm.requestsProcessingTotal = newAutoGaugeVec(buildMetricsName(providerField, requestsField, processingField, totalField), reporterConfig.Namespace, labelNames)
+	pm.requestsSucceedTotal = newAutoCounterVec(buildMetricsName(providerField, requestsField, succeedField, totalField), reporterConfig.Namespace, labelNames)
+	pm.rtMillisecondsMin = newAutoGaugeVecWithSyncMap(buildMetricsName(providerField, rtField, milliSecondsField, minField), reporterConfig.Namespace, labelNames)
+	pm.rtMillisecondsMax = newAutoGaugeVecWithSyncMap(buildMetricsName(providerField, rtField, milliSecondsField, maxField), reporterConfig.Namespace, labelNames)
+	pm.rtMillisecondsSum = newAutoCounterVec(buildMetricsName(providerField, rtField, milliSecondsField, sumField), reporterConfig.Namespace, labelNames)
+	pm.rtMillisecondsAvg = newAutoGaugeVecWithSyncMap(buildMetricsName(providerField, rtField, milliSecondsField, avgField), reporterConfig.Namespace, labelNames)
+	pm.rtMillisecondsLast = newAutoGaugeVec(buildMetricsName(providerField, rtField, milliSecondsField, lastField), reporterConfig.Namespace, labelNames)
+}
+
+type consumerMetrics struct {
+	rpcCommonMetrics
+}
 
-// init metric set and register to prometheus
-func (ms *metricSet) initAndRegister(reporterConfig *metrics.ReporterConfig) {
-	ms.consumerRTSummaryVec = newAutoSummaryVec(buildMetricsName(consumerField, rtField, milliSecondsField, summaryField), reporterConfig.Namespace, labelNames, reporterConfig.SummaryMaxAge)
-	ms.providerRTSummaryVec = newAutoSummaryVec(buildMetricsName(providerField, rtField, milliSecondsField, summaryField), reporterConfig.Namespace, labelNames, reporterConfig.SummaryMaxAge)
-	ms.consumerRequestsTotalCounterVec = newAutoCounterVec(buildMetricsName(consumerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
-	ms.providerRequestsTotalCounterVec = newAutoCounterVec(buildMetricsName(providerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
-	ms.consumerRequestsProcessingTotalGaugeVec = newAutoGaugeVec(buildMetricsName(consumerField, requestsField, processingField, totalField), reporterConfig.Namespace, labelNames)
-	ms.providerRequestsProcessingTotalGaugeVec = newAutoGaugeVec(buildMetricsName(providerField, requestsField, processingField, totalField), reporterConfig.Namespace, labelNames)
-	ms.consumerRequestsSucceedTotalCounterVec = newAutoCounterVec(buildMetricsName(consumerField, requestsField, succeedField, totalField), reporterConfig.Namespace, labelNames)
-	ms.providerRequestsSucceedTotalCounterVec = newAutoCounterVec(buildMetricsName(providerField, requestsField, succeedField, totalField), reporterConfig.Namespace, labelNames)
+func (cm *consumerMetrics) init(reporterConfig *metrics.ReporterConfig) {
+	cm.requestsTotal = newAutoCounterVec(buildMetricsName(consumerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
+	cm.requestsProcessingTotal = newAutoGaugeVec(buildMetricsName(consumerField, requestsField, processingField, totalField), reporterConfig.Namespace, labelNames)
+	cm.requestsSucceedTotal = newAutoCounterVec(buildMetricsName(consumerField, requestsField, succeedField, totalField), reporterConfig.Namespace, labelNames)
+	cm.rtMillisecondsMin = newAutoGaugeVecWithSyncMap(buildMetricsName(consumerField, rtField, milliSecondsField, minField), reporterConfig.Namespace, labelNames)
+	cm.rtMillisecondsMax = newAutoGaugeVecWithSyncMap(buildMetricsName(consumerField, rtField, milliSecondsField, maxField), reporterConfig.Namespace, labelNames)
+	cm.rtMillisecondsSum = newAutoCounterVec(buildMetricsName(consumerField, rtField, milliSecondsField, sumField), reporterConfig.Namespace, labelNames)
+	cm.rtMillisecondsAvg = newAutoGaugeVecWithSyncMap(buildMetricsName(consumerField, rtField, milliSecondsField, avgField), reporterConfig.Namespace, labelNames)
+	cm.rtMillisecondsLast = newAutoGaugeVec(buildMetricsName(consumerField, rtField, milliSecondsField, lastField), reporterConfig.Namespace, labelNames)
 }
 
 func buildMetricsName(args ...string) string {

metrics/prometheus/common.go → metrics/prometheus/model.go

@@ -18,63 +18,17 @@
 package prometheus
 
 import (
-	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 )
 
 import (
-	"github.com/dubbogo/gost/log/logger"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 )
 
-import (
-	"dubbo.apache.org/dubbo-go/v3/common"
-	"dubbo.apache.org/dubbo-go/v3/common/constant"
-)
-
-var (
-	defaultHistogramBucket = []float64{10, 50, 100, 200, 500, 1000, 10000}
-)
-
-func buildLabels(url *common.URL) prometheus.Labels {
-	return prometheus.Labels{
-		applicationNameKey: url.GetParam(constant.ApplicationKey, ""),
-		groupKey:           url.Group(),
-		hostnameKey:        "not implemented yet",
-		interfaceKey:       url.Service(),
-		ipKey:              common.GetLocalIp(),
-		versionKey:         url.GetParam(constant.AppVersionKey, ""),
-		methodKey:          url.GetParam(constant.MethodKey, ""),
-	}
-}
-
-// return the role of the application, provider or consumer, if the url is not a valid one, return empty string
-func getRole(url *common.URL) (role string) {
-	if isProvider(url) {
-		role = providerField
-	} else if isConsumer(url) {
-		role = consumerField
-	} else {
-		logger.Warnf("The url belongs neither the consumer nor the provider, "+
-			"so the invocation will be ignored. url: %s", url.String())
-	}
-	return
-}
-
-// isProvider shows whether this url represents the application received the request as server
-func isProvider(url *common.URL) bool {
-	role := url.GetParam(constant.RegistryRoleKey, "")
-	return strings.EqualFold(role, strconv.Itoa(common.PROVIDER))
-}
-
-// isConsumer shows whether this url represents the application sent then request as client
-func isConsumer(url *common.URL) bool {
-	role := url.GetParam(constant.RegistryRoleKey, "")
-	return strings.EqualFold(role, strconv.Itoa(common.CONSUMER))
-}
-
 func newHistogramVec(name, namespace string, labels []string) *prometheus.HistogramVec {
 	return prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
@@ -194,3 +148,108 @@ func newAutoSummaryVec(name, namespace string, labels []string, maxAge int64) *p
 		labels,
 	)
 }
+
+type GaugeVecWithSyncMap struct {
+	GaugeVec *prometheus.GaugeVec
+	SyncMap  *sync.Map
+}
+
+func newAutoGaugeVecWithSyncMap(name, namespace string, labels []string) *GaugeVecWithSyncMap {
+	return &GaugeVecWithSyncMap{
+		GaugeVec: newAutoGaugeVec(name, namespace, labels),
+		SyncMap:  &sync.Map{},
+	}
+}
+
+func convertLabelsToMapKey(labels prometheus.Labels) string {
+	return strings.Join([]string{
+		labels[applicationNameKey],
+		labels[groupKey],
+		labels[hostnameKey],
+		labels[interfaceKey],
+		labels[ipKey],
+		labels[versionKey],
+		labels[methodKey],
+	}, "_")
+}
+
+func (gv *GaugeVecWithSyncMap) updateMin(labels *prometheus.Labels, curValue int64) {
+	key := convertLabelsToMapKey(*labels)
+	cur := &atomic.Value{} // for first store
+	cur.Store(curValue)
+	for {
+		if actual, loaded := gv.SyncMap.LoadOrStore(key, cur); loaded {
+			store := actual.(*atomic.Value)
+			storeValue := store.Load().(int64)
+			if curValue < storeValue {
+				if store.CompareAndSwap(storeValue, curValue) {
+					// value is not changed, should update
+					gv.GaugeVec.With(*labels).Set(float64(curValue))
+					break
+				}
+				// value has changed, continue for loop
+			} else {
+				// no need to update
+				break
+			}
+		} else {
+			// store current curValue as this labels' init value
+			gv.GaugeVec.With(*labels).Set(float64(curValue))
+			break
+		}
+	}
+}
+
+func (gv *GaugeVecWithSyncMap) updateMax(labels *prometheus.Labels, curValue int64) {
+	key := convertLabelsToMapKey(*labels)
+	cur := &atomic.Value{} // for first store
+	cur.Store(curValue)
+	for {
+		if actual, loaded := gv.SyncMap.LoadOrStore(key, cur); loaded {
+			store := actual.(*atomic.Value)
+			storeValue := store.Load().(int64)
+			if curValue > storeValue {
+				if store.CompareAndSwap(storeValue, curValue) {
+					// value is not changed, should update
+					gv.GaugeVec.With(*labels).Set(float64(curValue))
+					break
+				}
+				// value has changed, continue for loop
+			} else {
+				// no need to update
+				break
+			}
+		} else {
+			// store current curValue as this labels' init value
+			gv.GaugeVec.With(*labels).Set(float64(curValue))
+			break
+		}
+	}
+}
+
+func (gv *GaugeVecWithSyncMap) updateAvg(labels *prometheus.Labels, curValue int64) {
+	key := convertLabelsToMapKey(*labels)
+	cur := &atomic.Value{} // for first store
+	type avgPair struct {
+		Sum int64
+		N   int64
+	}
+	cur.Store(avgPair{Sum: curValue, N: 1})
+
+	for {
+		if actual, loaded := gv.SyncMap.LoadOrStore(key, cur); loaded {
+			store := actual.(*atomic.Value)
+			storeValue := store.Load().(avgPair)
+			newValue := avgPair{Sum: storeValue.Sum + curValue, N: storeValue.N + 1}
+			if store.CompareAndSwap(storeValue, newValue) {
+				// value is not changed, should update
+				gv.GaugeVec.With(*labels).Set(float64(newValue.Sum / newValue.N))
+				break
+			}
+		} else {
+			// store current curValue as this labels' init value
+			gv.GaugeVec.With(*labels).Set(float64(curValue))
+			break
+		}
+	}
+}