Merge pull request #77 from marcoferrer/bugfix-queue-size-metrics

Fix bugs in metric reporting

Author: platinummonkey

core/metric_registry.go

@@ -25,7 +25,15 @@ func NewIntMetricSupplierWrapper(s func() int) MetricSupplier {
 	})
 }
 
-// NewFloat64MetricSupplierWrapper will wrap a int-return value func to a supplier func
+// NewUint64MetricSupplierWrapper will wrap a uint64-return value func to a supplier func
+func NewUint64MetricSupplierWrapper(s func() uint64) MetricSupplier {
+	return MetricSupplier(func() (float64, bool) {
+		val := s()
+		return float64(val), true
+	})
+}
+
+// NewFloat64MetricSupplierWrapper will wrap a float64-return value func to a supplier func
 func NewFloat64MetricSupplierWrapper(s func() float64) MetricSupplier {
 	return MetricSupplier(func() (float64, bool) {
 		val := s()
@@ -117,7 +125,7 @@ func NewCommonMetricSampler(registry MetricRegistry, limit Limit, name string, t
 // Sample will sample the current sample for metric reporting.
 func (s *CommonMetricSampler) Sample(rtt int64, inFlight int, didDrop bool) {
 	if didDrop {
-		s.InFlightListener.AddSample(1.0)
+		s.DropCounterListener.AddSample(1.0)
 	}
 	s.RTTListener.AddSample(float64(rtt))
 	s.InFlightListener.AddSample(float64(inFlight))

limiter/lifo_blocking.go

@@ -177,7 +177,7 @@ type LifoBlockingLimiter struct {
 	maxBacklogSize    uint64
 	maxBacklogTimeout time.Duration
 
-	backlog lifoQueue
+	backlog *lifoQueue
 	mu      sync.RWMutex
 }
 
@@ -203,16 +203,14 @@ func NewLifoBlockingLimiter(
 		delegate:          delegate,
 		maxBacklogSize:    uint64(maxBacklogSize),
 		maxBacklogTimeout: maxBacklogTimeout,
-		backlog:           lifoQueue{},
+		backlog:           &lifoQueue{},
 	}
 
-	registry.RegisterGauge(core.MetricLifoQueueLimit, func() (value float64, ok bool) {
-		return float64(maxBacklogSize), true
-	}, tags...)
+	registry.RegisterGauge(core.MetricLifoQueueLimit, core.NewIntMetricSupplierWrapper(func() int {
+		return maxBacklogSize
+	}), tags...)
 
-	registry.RegisterGauge(core.MetricLifoQueueSize, func() (value float64, ok bool) {
-		return float64(l.backlog.len()), true
-	}, tags...)
+	registry.RegisterGauge(core.MetricLifoQueueSize, core.NewUint64MetricSupplierWrapper(l.backlog.len), tags...)
 
 	return l
 }
@@ -261,7 +259,7 @@ func (l *LifoBlockingLimiter) tryAcquire(ctx context.Context) core.Listener {
 // If acquired the caller must call one of the Listener methods when the operation has been completed to release
 // the count.
 //
-// context Context for the request. The context is used by advanced strategies such as LookupPartitionStrategy.
+// ctx Context for the request. The context is used by advanced strategies such as LookupPartitionStrategy.
 func (l *LifoBlockingLimiter) Acquire(ctx context.Context) (core.Listener, bool) {
 	delegateListener := l.tryAcquire(ctx)
 	if delegateListener == nil {