Buffer metrics (#104)

* add metric for full work buffer

* add metric to measure time waiting on work buffer

* add larger buckets for buffer wait metric

rules/key_processor.go

@@ -2,6 +2,7 @@ package rules
 
 import (
 	"go.uber.org/zap"
+	"time"
 )
 
 type keyProc interface {
@@ -69,7 +70,11 @@ func (v3kp *v3KeyProcessor) dispatchWork(index int, rule staticRule, logger *zap
 		metricsInfo:      newMetricsInfo(context, keyPattern),
 		lockKey:          FormatWithAttributes(keyPattern, rule.getAttributes()),
 	}
+
+	start := time.Now()
 	v3kp.channel <- work
+	// measures the amount of time work is blocked from being added to the buffer
+	workBufferWaitTime(work.metricsInfo.method, keyPattern, start)
 }
 
 func newV3KeyProcessor(channel chan v3RuleWork, rm *ruleManager) v3KeyProcessor {

rules/prometheus.go

@@ -32,13 +32,21 @@ var (
 		Help:      "etcd rules engine worker queue wait time in ms",
 		Buckets:   []float64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000},
 	}, []string{"method"})
+	rulesEngineWorkBufferWaitTime = prometheus.NewHistogramVec(prometheus.HistogramOpts{
+		Name:      "work_buffer_wait_ms",
+		Subsystem: "etcd",
+		Namespace: "rules",
+		Help:      "etcd rules engine work buffer wait time in ms",
+		Buckets:   []float64{1, 2, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000, 10000, 30000, 60000, 300000, 600000},
+	}, []string{"method", "pattern"})
 )
 
 func init() {
 	prometheus.MustRegister(rulesEngineLockCount)
 	prometheus.MustRegister(rulesEngineSatisfiedThenNot)
 	prometheus.MustRegister(rulesEngineEvaluations)
 	prometheus.MustRegister(rulesEngineWorkerQueueWait)
+	prometheus.MustRegister(rulesEngineWorkBufferWaitTime)
 }
 
 func incLockMetric(methodName string, pattern string, lockSucceeded bool) {
@@ -56,3 +64,7 @@ func timesEvaluated(methodName string, ruleID string, count int) {
 func workerQueueWaitTime(methodName string, startTime time.Time) {
 	rulesEngineWorkerQueueWait.WithLabelValues(methodName).Observe(float64(time.Since(startTime).Nanoseconds() / 1e6))
 }
+
+func workBufferWaitTime(methodName, pattern string, startTime time.Time) {
+	rulesEngineWorkBufferWaitTime.WithLabelValues(methodName, pattern).Observe(float64(time.Since(startTime).Nanoseconds() / 1e6))
+}

rules/prometheus_test.go

@@ -61,3 +61,8 @@ func TestWokerQueueWaitTime(t *testing.T) {
 	workerQueueWaitTime("getKey", time.Now())
 	checkMetrics(t, `rules_etcd_worker_queue_wait_ms_count{method="getKey"} 1`)
 }
+
+func TestWorkBufferWaitTime(t *testing.T) {
+	workBufferWaitTime("getKey", "/desired/key/pattern", time.Now())
+	checkMetrics(t, `rules_etcd_work_buffer_wait_ms_count{method="getKey",pattern="/desired/key/pattern"} 1`)
+}

rules/worker.go

@@ -94,7 +94,7 @@ func (bw *baseWorker) doWork(loggerPtr **zap.Logger,
 		return
 	}
 	if !sat {
-		incSatisfiedThenNot(metricsInfo.method, metricsInfo.keyPattern, "worker.doWorkBeforeLock")
+		incSatisfiedThenNot(metricsInfo.method, metricsInfo.keyPattern, "worker.doWorkAfterLock")
 		bw.metrics.IncSatisfiedThenNot(metricsInfo.method, metricsInfo.keyPattern, "worker.doWorkAfterLock")
 	}
 	workerQueueWaitTime(metricsInfo.method, metricsInfo.startTime)