1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
package limithandler
import (
"context"
"strings"
"time"
grpc_logrus "github.com/grpc-ecosystem/go-grpc-middleware/logging/logrus"
"github.com/prometheus/client_golang/prometheus"
)
const acquireDurationLogThreshold = 10 * time.Millisecond
var (
histogramEnabled = false
histogramVec *prometheus.HistogramVec
inprogressGaugeVec = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "gitaly",
Subsystem: "rate_limiting",
Name: "in_progress",
Help: "Gauge of number of number of concurrent invocations currently in progress for this endpoint",
},
[]string{"system", "grpc_service", "grpc_method"},
)
queuedGaugeVec = prometheus.NewGaugeVec(
prometheus.GaugeOpts{
Namespace: "gitaly",
Subsystem: "rate_limiting",
Name: "queued",
Help: "Gauge of number of number of invocations currently queued for this endpoint",
},
[]string{"system", "grpc_service", "grpc_method"},
)
)
type promMonitor struct {
queuedGauge prometheus.Gauge
inprogressGauge prometheus.Gauge
histogram prometheus.Observer
}
func init() {
prometheus.MustRegister(inprogressGaugeVec, queuedGaugeVec)
}
func splitMethodName(fullMethodName string) (string, string) {
fullMethodName = strings.TrimPrefix(fullMethodName, "/") // remove leading slash
if i := strings.Index(fullMethodName, "/"); i >= 0 {
return fullMethodName[:i], fullMethodName[i+1:]
}
return "unknown", "unknown"
}
// EnableAcquireTimeHistogram enables histograms for acquisition times
func EnableAcquireTimeHistogram(buckets []float64) {
histogramEnabled = true
histogramOpts := prometheus.HistogramOpts{
Namespace: "gitaly",
Subsystem: "rate_limiting",
Name: "acquiring_seconds",
Help: "Histogram of lock acquisition latency (seconds) for endpoint rate limiting",
Buckets: buckets,
}
histogramVec = prometheus.NewHistogramVec(
histogramOpts,
[]string{"system", "grpc_service", "grpc_method"},
)
prometheus.Register(histogramVec)
}
func (c *promMonitor) Queued(ctx context.Context) {
c.queuedGauge.Inc()
}
func (c *promMonitor) Dequeued(ctx context.Context) {
c.queuedGauge.Dec()
}
func (c *promMonitor) Enter(ctx context.Context, acquireTime time.Duration) {
c.inprogressGauge.Inc()
if acquireTime > acquireDurationLogThreshold {
logger := grpc_logrus.Extract(ctx)
logger.WithField("acquire_ms", acquireTime.Seconds()*1000).Info("Rate limit acquire wait")
}
if c.histogram != nil {
c.histogram.Observe(acquireTime.Seconds())
}
}
func (c *promMonitor) Exit(ctx context.Context) {
c.inprogressGauge.Dec()
}
// NewPromMonitor creates a new ConcurrencyMonitor that tracks limiter
// activity in Prometheus.
func NewPromMonitor(system string, fullMethod string) ConcurrencyMonitor {
serviceName, methodName := splitMethodName(fullMethod)
queuedGauge := queuedGaugeVec.WithLabelValues(serviceName, methodName, system)
inprogressGauge := inprogressGaugeVec.WithLabelValues(serviceName, methodName, system)
var histogram prometheus.Observer
if histogramVec != nil {
histogram = histogramVec.WithLabelValues(system, serviceName, methodName)
}
return &promMonitor{queuedGauge, inprogressGauge, histogram}
}
|
