circuit breaker rolling window config

hystrix/eventstream.go

@@ -7,7 +7,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/afex/hystrix-go/hystrix/rolling"
+	"github.com/raghavTayal/hystrix-go/hystrix/rolling"
 )
 
 const (

hystrix/metric_collector/default_metric_collector.go

@@ -3,7 +3,7 @@ package metricCollector
 import (
 	"sync"
 
-	"github.com/afex/hystrix-go/hystrix/rolling"
+	"github.com/raghavTayal/hystrix-go/hystrix/rolling"
 )
 
 // DefaultMetricCollector holds information about the circuit state.

hystrix/metrics.go

@@ -4,8 +4,7 @@ import (
 	"sync"
 	"time"
 
-	"github.com/afex/hystrix-go/hystrix/metric_collector"
-	"github.com/afex/hystrix-go/hystrix/rolling"
+	"github.com/raghavTayal/hystrix-go/hystrix/rolling"
 )
 
 type commandExecution struct {

hystrix/pool_metrics.go

@@ -3,7 +3,7 @@ package hystrix
 import (
 	"sync"
 
-	"github.com/afex/hystrix-go/hystrix/rolling"
+	"github.com/raghavTayal/hystrix-go/hystrix/rolling"
 )
 
 type poolMetrics struct {

hystrix/rolling/rolling.go

@@ -1,15 +1,26 @@
 package rolling
 
 import (
+	"errors"
 	"sync"
 	"time"
 )
 
-// Number tracks a numberBucket over a bounded number of
-// time buckets. Currently the buckets are one second long and only the last 10 seconds are kept.
+const (
+	defaultWindow = 10
+)
+
+var (
+	ErrNegativeWindow = errors.New("window must be positive integer")
+)
+
+// Number tracks a numberBucket over a bounded Number of
+// time buckets. Currently the buckets are one second long and only the last {window} seconds are kept.
+
 type Number struct {
 	Buckets map[int64]*numberBucket
 	Mutex   *sync.RWMutex
+	window  int64
 }
 
 type numberBucket struct {
@@ -19,12 +30,25 @@ type numberBucket struct {
 // NewNumber initializes a RollingNumber struct.
 func NewNumber() *Number {
 	r := &Number{
-		Buckets: make(map[int64]*numberBucket),
+		Buckets: make(map[int64]*numberBucket, defaultWindow),
 		Mutex:   &sync.RWMutex{},
+		window:  defaultWindow,
 	}
 	return r
 }
 
+// NewNumberWithWindow initializes a RollingNumber with window given
+func NewNumberWithWindow(window int64) (*Number, error) {
+	if window <= 0 {
+		return nil, ErrNegativeWindow
+	}
+	return &Number{
+		Buckets: make(map[int64]*numberBucket, window),
+		Mutex:   &sync.RWMutex{},
+		window:  window,
+	}, nil
+}
+
 func (r *Number) getCurrentBucket() *numberBucket {
 	now := time.Now().Unix()
 	var bucket *numberBucket
@@ -39,10 +63,9 @@ func (r *Number) getCurrentBucket() *numberBucket {
 }
 
 func (r *Number) removeOldBuckets() {
-	now := time.Now().Unix() - 10
+	now := time.Now().Unix() - 20
 
 	for timestamp := range r.Buckets {
-		// TODO: configurable rolling window
 		if timestamp <= now {
 			delete(r.Buckets, timestamp)
 		}
@@ -75,33 +98,31 @@ func (r *Number) UpdateMax(n float64) {
 	r.removeOldBuckets()
 }
 
-// Sum sums the values over the buckets in the last 10 seconds.
+// Sum sums the values over the buckets in the last {window} seconds.
 func (r *Number) Sum(now time.Time) float64 {
 	sum := float64(0)
 
 	r.Mutex.RLock()
 	defer r.Mutex.RUnlock()
 
 	for timestamp, bucket := range r.Buckets {
-		// TODO: configurable rolling window
-		if timestamp >= now.Unix()-10 {
+		if timestamp >= now.Unix()-20 {
 			sum += bucket.Value
 		}
 	}
 
 	return sum
 }
 
-// Max returns the maximum value seen in the last 10 seconds.
+// Max returns the maximum value seen in the last window seconds.
 func (r *Number) Max(now time.Time) float64 {
 	var max float64
 
 	r.Mutex.RLock()
 	defer r.Mutex.RUnlock()
 
 	for timestamp, bucket := range r.Buckets {
-		// TODO: configurable rolling window
-		if timestamp >= now.Unix()-10 {
+		if timestamp >= now.Unix()-20 {
 			if bucket.Value > max {
 				max = bucket.Value
 			}
@@ -112,5 +133,8 @@ func (r *Number) Max(now time.Time) float64 {
 }
 
 func (r *Number) Avg(now time.Time) float64 {
-	return r.Sum(now) / 10
+	//if r.window == 0 { // unexpected 0
+	//	return r.Sum(now)
+	//}
+	return r.Sum(now) / float64(20)
 }

hystrix/rolling/rolling_test.go

@@ -3,10 +3,22 @@ package rolling
 import (
 	"testing"
 	"time"
-
-	. "github.com/smartystreets/goconvey/convey"
 )
 
+func TestWindowAvg(t *testing.T) {
+	Convey("when adding values to a rolling number", t, func() {
+		n, _ := NewNumberWithWindow(2)
+		for _, x := range []float64{0.5, 1.5, 2.5, 3.5, 4.5} {
+			n.Increment(x)
+			time.Sleep(1 * time.Second)
+		}
+
+		Convey("it should calculate the average over the number of configured buckets", func() {
+			So(n.Avg(time.Now()), ShouldEqual, 4) // (3.5+4.5)/2
+		})
+	})
+}
+
 func TestMax(t *testing.T) {
 
 	Convey("when adding values to a rolling number", t, func() {

loadtest/service/main.go

@@ -11,10 +11,9 @@ import (
 	"runtime"
 	"time"
 
-	"github.com/afex/hystrix-go/hystrix"
-	"github.com/afex/hystrix-go/hystrix/metric_collector"
-	"github.com/afex/hystrix-go/plugins"
 	"github.com/cactus/go-statsd-client/statsd"
+	"github.com/raghavTayal/hystrix-go/hystrix"
+	"github.com/raghavTayal/hystrix-go/plugins"
 )
 
 const (

plugins/datadog_collector.go

@@ -4,7 +4,6 @@ import (
 
 	// Developed on https://github.com/DataDog/datadog-go/tree/a27810dd518c69be741a7fd5d0e39f674f615be8
 	"github.com/DataDog/datadog-go/statsd"
-	"github.com/afex/hystrix-go/hystrix/metric_collector"
 )
 
 // These metrics are constants because we're leveraging the Datadog tagging

plugins/graphite_aggregator.go

@@ -6,9 +6,6 @@ import (
 	"net"
 	"strings"
 	"time"
-
-	"github.com/afex/hystrix-go/hystrix/metric_collector"
-	"github.com/rcrowley/go-metrics"
 )
 
 var makeTimerFunc = func() interface{} { return metrics.NewTimer() }

plugins/statsd_collector.go

@@ -5,7 +5,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/afex/hystrix-go/hystrix/metric_collector"
 	"github.com/cactus/go-statsd-client/statsd"
 )