Merge branch 'afex-contrib' of github.com:vermapratyush/hystrix-go in…

…to afex-contrib

hystrix/hystrix_test.go

@@ -8,7 +8,6 @@ import (
 	"sync/atomic"
 
 	. "github.com/smartystreets/goconvey/convey"
-	"testing/quick"
 )
 
 func TestSuccess(t *testing.T) {
@@ -342,30 +341,6 @@ func TestFallbackAfterRejected(t *testing.T) {
 	})
 }
 
-func TestReturnTicket_QuickCheck(t *testing.T) {
-	compareTicket := func() bool {
-		defer Flush()
-		ConfigureCommand("", CommandConfig{Timeout: 2})
-		errChan := Go("", func() error {
-			c := make(chan struct{})
-			<-c // should block
-			return nil
-		}, nil)
-		err := <-errChan
-		So(err, ShouldResemble, ErrTimeout)
-		cb, _, err := GetCircuit("")
-		So(err, ShouldBeNil)
-		return cb.executorPool.ActiveCount() == 0
-	}
-
-	Convey("with a run command that doesn't return", t, func() {
-		Convey("checking many times that after Go(), the ticket returns to the pool after the timeout", func() {
-			err := quick.Check(compareTicket, nil)
-			So(err, ShouldBeNil)
-		})
-	})
-}
-
 func TestReturnTicket(t *testing.T) {
 	Convey("with a run command that doesn't return", t, func() {
 		defer Flush()

hystrix/pool_test.go

@@ -85,6 +85,7 @@ func TestWaitingCount(t *testing.T) {
 
 		Convey("WaitingTicket should be max-10", func() {
 			<-checkpoint
+			time.Sleep(500 * time.Millisecond)
 			So(pool.WaitingCount(), ShouldEqual, pool.QueueSizeRejectionThreshold-10)
 		})
 

plugins/graphite_aggregator.go

@@ -22,6 +22,7 @@ var makeCounterFunc = func() interface{} { return metrics.NewCounter() }
 // on how metrics are aggregated and expressed in graphite.
 type GraphiteCollector struct {
 	attemptsPrefix          string
+	queueSizePrefix         string
 	errorsPrefix            string
 	successesPrefix         string
 	failuresPrefix          string
@@ -60,6 +61,7 @@ func NewGraphiteCollector(name string) metricCollector.MetricCollector {
 	return &GraphiteCollector{
 		attemptsPrefix:          name + ".attempts",
 		errorsPrefix:            name + ".errors",
+		queueSizePrefix:         name + ".queueLength",
 		successesPrefix:         name + ".successes",
 		failuresPrefix:          name + ".failures",
 		rejectsPrefix:           name + ".rejects",
@@ -94,6 +96,12 @@ func (g *GraphiteCollector) IncrementAttempts() {
 	g.incrementCounterMetric(g.attemptsPrefix)
 }
 
+// IncrementQueueSize increments the number of elements in the queue.
+// Request that would have otherwise been rejected, but was queued before executing/rejection
+func (g *GraphiteCollector) IncrementQueueSize() {
+	g.incrementCounterMetric(g.queueSizePrefix)
+}
+
 // IncrementErrors increments the number of unsuccessful attempts.
 // Attempts minus Errors will equal successes within a time range.
 // Errors are any result from an attempt that is not a success.

plugins/statsd_collector.go

@@ -18,6 +18,7 @@ type StatsdCollector struct {
 	client                  statsd.Statter
 	circuitOpenPrefix       string
 	attemptsPrefix          string
+	queueSizePrefix         string
 	errorsPrefix            string
 	successesPrefix         string
 	failuresPrefix          string
@@ -96,6 +97,7 @@ func (s *StatsdCollectorClient) NewStatsdCollector(name string) metricCollector.
 		circuitOpenPrefix:       name + ".circuitOpen",
 		attemptsPrefix:          name + ".attempts",
 		errorsPrefix:            name + ".errors",
+		queueSizePrefix:         name + ".queueLength",
 		successesPrefix:         name + ".successes",
 		failuresPrefix:          name + ".failures",
 		rejectsPrefix:           name + ".rejects",
@@ -136,6 +138,12 @@ func (g *StatsdCollector) IncrementAttempts() {
 	g.incrementCounterMetric(g.attemptsPrefix)
 }
 
+// IncrementQueueSize increments the number of elements in the queue.
+// Request that would have otherwise been rejected, but was queued before executing/rejection
+func (g *StatsdCollector) IncrementQueueSize() {
+	g.incrementCounterMetric(g.queueSizePrefix)
+}
+
 // IncrementErrors increments the number of unsuccessful attempts.
 // Attempts minus Errors will equal successes within a time range.
 // Errors are any result from an attempt that is not a success.