feat: add report before invocation with RequestProcessingGauge metric…

… implement

filter/metrics/filter.go

@@ -44,6 +44,11 @@ type Filter struct {
 
 // Invoke collect the duration of invocation and then report the duration by using goroutine
 func (p *Filter) Invoke(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) protocol.Result {
+	go func() {
+		for _, reporter := range p.reporters {
+			reporter.ReportBeforeInvocation(ctx, invoker, invocation)
+		}
+	}()
 	start := time.Now()
 	res := invoker.Invoke(ctx, invocation)
 	end := time.Now()

metrics/prometheus/after_invocation.go

@@ -23,61 +23,22 @@ import (
 )
 
 import (
-	"dubbo.apache.org/dubbo-go/v3/common"
-	"dubbo.apache.org/dubbo-go/v3/common/constant"
 	"dubbo.apache.org/dubbo-go/v3/protocol"
 )
 
-import (
-	"github.com/dubbogo/gost/log/logger"
-	"github.com/prometheus/client_golang/prometheus"
-)
-
 func (reporter *PrometheusReporter) ReportAfterInvocation(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation, cost time.Duration, res protocol.Result) {
 	if !reporter.reporterConfig.Enable {
 		return
 	}
-
 	url := invoker.GetURL()
 
-	var role string // provider or consumer
-	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())
+	role := getRole(url)
+	if role == "" {
 		return
 	}
-	labels := prometheus.Labels{
-		applicationNameKey: url.GetParam(constant.ApplicationKey, ""),
-		groupKey:           url.Group(),
-		hostnameKey:        "",
-		interfaceKey:       url.Service(),
-		ipKey:              common.GetLocalIp(),
-		versionKey:         url.GetParam(constant.AppVersionKey, ""),
-		methodKey:          invocation.MethodName(),
-	}
+	labels := buildLabels(url)
 
 	reporter.reportRTSummaryVec(role, &labels, cost.Milliseconds())
 	reporter.reportRequestTotalCounterVec(role, &labels)
-}
-
-func (r *PrometheusReporter) reportRTSummaryVec(role string, labels *prometheus.Labels, costMs int64) {
-	switch role {
-	case providerField:
-		r.providerRTSummaryVec.With(*labels).Observe(float64(costMs))
-	case consumerField:
-		r.consumerRTSummaryVec.With(*labels).Observe(float64(costMs))
-	}
-}
-
-func (r *PrometheusReporter) reportRequestTotalCounterVec(role string, labels *prometheus.Labels) {
-	switch role {
-	case providerField:
-		r.providerRequestTotalCounterVec.With(*labels).Inc()
-	case consumerField:
-		r.consumerRequestTotalCounterVec.With(*labels).Inc()
-	}
+	reporter.decRequestProcessingGaugeVec(role, &labels)
 }

metrics/prometheus/before_invocation.go

@@ -16,3 +16,23 @@
  */
 
 package prometheus
+
+import (
+	"context"
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+)
+
+func (reporter *PrometheusReporter) ReportBeforeInvocation(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation) {
+	if !reporter.reporterConfig.Enable {
+		return
+	}
+	url := invoker.GetURL()
+
+	role := getRole(url)
+	if role == "" {
+		return
+	}
+	labels := buildLabels(url)
+
+	reporter.incRequestProcessingGaugeVec(role, &labels)
+}

metrics/prometheus/common.go

@@ -18,6 +18,7 @@
 package prometheus
 
 import (
+	"github.com/dubbogo/gost/log/logger"
 	"strconv"
 	"strings"
 	"time"
@@ -33,6 +34,31 @@ import (
 	"dubbo.apache.org/dubbo-go/v3/common/constant"
 )
 
+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, "")

metrics/prometheus/constant.go

@@ -41,5 +41,6 @@ const (
 	counterField = "counter"
 	summaryField = "summary"
 
-	totalField = "total"
+	totalField      = "total"
+	processingField = "processing"
 )

metrics/prometheus/metric_set.go

@@ -35,24 +35,37 @@ type metricSet struct {
 	consumerRTSummaryVec *prometheus.SummaryVec
 	// report the provider-side's rt gauge data
 	providerRTSummaryVec *prometheus.SummaryVec
+
 	// report the provider-side's request total counter data
 	providerRequestTotalCounterVec *prometheus.CounterVec
+	// report the provider-side's processing request counter data
+	providerRequestProcessingGaugeVec *prometheus.GaugeVec
+
 	// report the consumer-side's request total counter data
 	consumerRequestTotalCounterVec *prometheus.CounterVec
-	// report the provider-side's processing request counter data
-	// providerRequestProcessingGaugeVec *prometheus.GaugeVec
 	// report the consumer-side's processing request counter data
-	// consumerRequestProcessingGaugeVec *prometheus.GaugeVec
+	consumerRequestProcessingGaugeVec *prometheus.GaugeVec
 }
 
 var labelNames = []string{applicationNameKey, groupKey, hostnameKey, interfaceKey, ipKey, methodKey, versionKey}
 
 // 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.consumerRequestTotalCounterVec = newAutoCounterVec(buildMetricsName(consumerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
-	ms.providerRequestTotalCounterVec = newAutoCounterVec(buildMetricsName(providerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
+	ms.consumerRTSummaryVec = newSummaryVec(buildMetricsName(consumerField, rtField, milliSecondsField, summaryField), reporterConfig.Namespace, labelNames, reporterConfig.SummaryMaxAge)
+	ms.providerRTSummaryVec = newSummaryVec(buildMetricsName(providerField, rtField, milliSecondsField, summaryField), reporterConfig.Namespace, labelNames, reporterConfig.SummaryMaxAge)
+	ms.consumerRequestTotalCounterVec = newCounterVec(buildMetricsName(consumerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
+	ms.providerRequestTotalCounterVec = newCounterVec(buildMetricsName(providerField, requestsField, totalField), reporterConfig.Namespace, labelNames)
+	ms.consumerRequestProcessingGaugeVec = newGaugeVec(buildMetricsName(consumerField, requestsField, processingField), reporterConfig.Namespace, labelNames)
+	ms.providerRequestProcessingGaugeVec = newGaugeVec(buildMetricsName(providerField, requestsField, processingField), reporterConfig.Namespace, labelNames)
+
+	prometheus.DefaultRegisterer.MustRegister(
+		ms.consumerRTSummaryVec,
+		ms.providerRTSummaryVec,
+		ms.consumerRequestTotalCounterVec,
+		ms.providerRequestTotalCounterVec,
+		ms.consumerRequestProcessingGaugeVec,
+		ms.providerRequestProcessingGaugeVec,
+	)
 }
 
 func buildMetricsName(args ...string) string {

metrics/prometheus/reporter.go

@@ -102,3 +102,39 @@ func (reporter *PrometheusReporter) shutdownServer() {
 		}
 	}
 }
+
+func (reporter *PrometheusReporter) reportRTSummaryVec(role string, labels *prometheus.Labels, costMs int64) {
+	switch role {
+	case providerField:
+		reporter.providerRTSummaryVec.With(*labels).Observe(float64(costMs))
+	case consumerField:
+		reporter.consumerRTSummaryVec.With(*labels).Observe(float64(costMs))
+	}
+}
+
+func (reporter *PrometheusReporter) reportRequestTotalCounterVec(role string, labels *prometheus.Labels) {
+	switch role {
+	case providerField:
+		reporter.providerRequestTotalCounterVec.With(*labels).Inc()
+	case consumerField:
+		reporter.consumerRequestTotalCounterVec.With(*labels).Inc()
+	}
+}
+
+func (reporter *PrometheusReporter) incRequestProcessingGaugeVec(role string, labels *prometheus.Labels) {
+	switch role {
+	case providerField:
+		reporter.providerRequestProcessingGaugeVec.With(*labels).Inc()
+	case consumerField:
+		reporter.consumerRequestProcessingGaugeVec.With(*labels).Inc()
+	}
+}
+
+func (reporter *PrometheusReporter) decRequestProcessingGaugeVec(role string, labels *prometheus.Labels) {
+	switch role {
+	case providerField:
+		reporter.providerRequestProcessingGaugeVec.With(*labels).Dec()
+	case consumerField:
+		reporter.consumerRequestProcessingGaugeVec.With(*labels).Dec()
+	}
+}

metrics/reporter.go

@@ -63,4 +63,5 @@ func NewReporterConfig() *ReporterConfig {
 type Reporter interface {
 	ReportAfterInvocation(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation,
 		cost time.Duration, res protocol.Result)
+	ReportBeforeInvocation(ctx context.Context, invoker protocol.Invoker, invocation protocol.Invocation)
 }