Merge pull request #73 from marcoferrer/add-lifo-queue-metrics

implement lifo queue gauge metrics

Author: platinummonkey

core/dep.go

@@ -19,8 +19,12 @@ const (
 	MetricMinRTT = "min_rtt"
 	// MetricWindowMinRTT is the name of the metric for the Window's Minimum Round Trip Time
 	MetricWindowMinRTT = "window.min_rtt"
-	// MetricWindowQueueSize represents the name of hte metric for the Window's Queue Size
+	// MetricWindowQueueSize represents the name of the metric for the Window's Queue Size
 	MetricWindowQueueSize = "window.queue_size"
+	// MetricLifoQueueSize represents the name of the metric for the size of a lifo queue
+	MetricLifoQueueSize = "lifo.queue_size"
+	// MetricLifoQueueLimit represents the name of the metric for the max size of a lifo queue
+	MetricLifoQueueLimit = "lifo.queue_limit"
 )
 
 // PrefixMetricWithName will prefix a given name with the metric name in the form "<name>.<metric>"

limiter/lifo_blocking.go

@@ -186,26 +186,46 @@ func NewLifoBlockingLimiter(
 	delegate core.Limiter,
 	maxBacklogSize int,
 	maxBacklogTimeout time.Duration,
+	registry core.MetricRegistry,
+	tags ...string,
 ) *LifoBlockingLimiter {
 	if maxBacklogSize <= 0 {
 		maxBacklogSize = 100
 	}
 	if maxBacklogTimeout == 0 {
 		maxBacklogTimeout = time.Millisecond * 1000
 	}
-	return &LifoBlockingLimiter{
+	if registry == nil {
+		registry = &core.EmptyMetricRegistry{}
+	}
+
+	l := &LifoBlockingLimiter{
 		delegate:          delegate,
 		maxBacklogSize:    uint64(maxBacklogSize),
 		maxBacklogTimeout: maxBacklogTimeout,
 		backlog:           lifoQueue{},
 	}
+
+	registry.RegisterGauge(core.MetricLifoQueueLimit, func() (value float64, ok bool) {
+		return float64(maxBacklogSize), true
+	}, tags...)
+
+	registry.RegisterGauge(core.MetricLifoQueueSize, func() (value float64, ok bool) {
+		return float64(l.backlog.len()), true
+	}, tags...)
+
+	return l
 }
 
 // NewLifoBlockingLimiterWithDefaults will create a new LifoBlockingLimiter with default values.
 func NewLifoBlockingLimiterWithDefaults(
 	delegate core.Limiter,
 ) *LifoBlockingLimiter {
-	return NewLifoBlockingLimiter(delegate, 100, time.Millisecond*1000)
+	return NewLifoBlockingLimiter(
+		delegate,
+		100, time.Millisecond*1000,
+		&core.EmptyMetricRegistry{},
+	)
 }
 
 func (l *LifoBlockingLimiter) tryAcquire(ctx context.Context) core.Listener {

limiter/lifo_blocking_test.go

@@ -168,7 +168,7 @@ func TestLifoBlockingLimiter(t *testing.T) {
 			limit.NoopLimitLogger{},
 			core.EmptyMetricRegistryInstance,
 		)
-		limiter := NewLifoBlockingLimiter(delegateLimiter, -1, 0)
+		limiter := NewLifoBlockingLimiter(delegateLimiter, -1, 0, nil)
 		asrt.NotNil(limiter)
 		asrt.True(strings.Contains(limiter.String(), "LifoBlockingLimiter{delegate=DefaultLimiter{"))
 	})
@@ -214,8 +214,8 @@ func TestLifoBlockingLimiter(t *testing.T) {
 			go func(j int) {
 				startupReady <- true
 				listener, ok := limiter.Acquire(context.Background())
-				asrt.True(ok)
-				asrt.NotNil(listener)
+				asrt.True(ok, "must be true for j %d", j)
+				asrt.NotNil(listener, "must be not be nil for j %d", j)
 				mu.Lock()
 				waitingListeners = append(waitingListeners, acquiredListenerLifo{id: j, listener: listener})
 				mu.Unlock()

patterns/pool/fixed_pool.go

@@ -86,7 +86,7 @@ func NewFixedPool(
 	case OrderingLIFO:
 		fp = FixedPool{
 			limit:    fixedLimit,
-			limiter:  limiter.NewLifoBlockingLimiter(defaultLimiter, maxBacklog, timeout),
+			limiter:  limiter.NewLifoBlockingLimiter(defaultLimiter, maxBacklog, timeout, metricRegistry),
 			ordering: ordering,
 		}
 	default:

patterns/pool/generic_pool.go

@@ -58,7 +58,7 @@ func NewPool(
 		}
 	case OrderingLIFO:
 		p = Pool{
-			limiter: limiter.NewLifoBlockingLimiter(delegateLimiter, maxBacklog, timeout),
+			limiter: limiter.NewLifoBlockingLimiter(delegateLimiter, maxBacklog, timeout, metricRegistry),
 		}
 	default:
 		p = Pool{