*: add API concurrency metrics

pkg/ratelimit/concurrency_limiter.go

@@ -20,6 +20,9 @@ type concurrencyLimiter struct {
 	mu      syncutil.RWMutex
 	current uint64
 	limit   uint64
+
+	// statistic
+	maxLimit uint64
 }
 
 func newConcurrencyLimiter(limit uint64) *concurrencyLimiter {
@@ -34,6 +37,9 @@ func (l *concurrencyLimiter) allow() bool {
 
 	if l.limit == unlimit || l.current+1 <= l.limit {
 		l.current++
+		if l.current > l.maxLimit {
+			l.maxLimit = l.current
+		}
 		return true
 	}
 	return false
@@ -68,3 +74,13 @@ func (l *concurrencyLimiter) getCurrent() uint64 {
 
 	return l.current
 }
+
+func (l *concurrencyLimiter) getMaxConcurrency() uint64 {
+	l.mu.RLock()
+	defer func() {
+		l.maxLimit = 0
+		l.mu.RUnlock()
+	}()
+
+	return l.maxLimit
+}

pkg/ratelimit/concurrency_limiter_test.go

@@ -31,9 +31,18 @@ func TestConcurrencyLimiter(t *testing.T) {
 	cl.release()
 	re.True(cl.allow())
 	re.Equal(uint64(10), cl.getLimit())
+	re.Equal(uint64(10), cl.getMaxConcurrency())
+	re.Equal(uint64(0), cl.getMaxConcurrency())
 	cl.setLimit(5)
 	re.Equal(uint64(5), cl.getLimit())
 	re.Equal(uint64(10), cl.getCurrent())
 	cl.release()
 	re.Equal(uint64(9), cl.getCurrent())
+	for i := 0; i < 9; i++ {
+		cl.release()
+	}
+	for i := 0; i < 5; i++ {
+		re.True(cl.allow())
+	}
+	re.Equal(uint64(5), cl.getMaxConcurrency())
 }

pkg/ratelimit/controller.go

@@ -70,7 +70,7 @@ func (l *Controller) collectMetrics() {
 				label := key.(string)
 				// Due to not in hot path, no need to save sub Gauge.
 				if con := limiter.getConcurrencyLimiter(); con != nil {
-					l.concurrencyGauge.WithLabelValues(l.apiType, label).Set(float64(con.getCurrent()))
+					l.concurrencyGauge.WithLabelValues(l.apiType, label).Set(float64(con.getMaxConcurrency()))
 				}
 				return true
 			})