Balancing

balancing/balance_breaker.go

@@ -117,7 +117,7 @@ func (meter *CallMeter) Calls() float64 {
 
 // CallsInLastPeriod returns number of calls in last duration
 func (meter *CallMeter) CallsInLastPeriod(period time.Duration) float64 {
-	lastPeriodSeries := meter.histogram.pickLastSeries(period)
+	lastPeriodSeries := meter.histogram.PickLastSeries(period)
 	sum := float64(0)
 	now := meter.now()
 	for _, series := range lastPeriodSeries {
@@ -146,7 +146,7 @@ func (meter *CallMeter) SetActive(active bool) {
 
 // TimeSpent returns float64 repesentation of time spent in execution
 func (meter *CallMeter) TimeSpent() float64 {
-	allSeries := meter.histogram.pickLastSeries(meter.resolution)
+	allSeries := meter.histogram.PickLastSeries(meter.resolution)
 	sum := float64(0)
 	now := meter.now()
 
@@ -170,14 +170,12 @@ func (series *dataSeries) Add(value float64, dateTime time.Time) {
 	series.data = append(series.data, &timeValue{dateTime, value})
 }
 
-func (series *dataSeries) ValueRangeFun(timeStart, timeEnd time.Time, fun func(*timeValue)) []float64 {
-	dataRange := []float64{}
+func (series *dataSeries) ValueRangeFun(timeStart, timeEnd time.Time, fun func(*timeValue)) {
 	for _, timeVal := range series.data {
 		if (timeStart == timeVal.date || timeStart.Before(timeVal.date)) && timeEnd.After(timeVal.date) {
 			fun(timeVal)
 		}
 	}
-	return dataRange
 }
 
 func (series *dataSeries) ValueRange(timeStart, timeEnd time.Time) []float64 {
@@ -226,7 +224,8 @@ func (h *histogram) pickSeries(at time.Time) *dataSeries {
 	return h.data[idx]
 }
 
-func (h *histogram) pickLastSeries(period time.Duration) []*dataSeries {
+// PickLastSeries returns slice of dataSeries tracking at least given period of time
+func (h *histogram) PickLastSeries(period time.Duration) []*dataSeries {
 	h.mx.Lock()
 	defer h.mx.Unlock()
 	if period > h.retention {
@@ -294,7 +293,7 @@ func newBreaker(retention int, callTimeLimit time.Duration,
 	closeDelay, maxDelay time.Duration) Breaker {
 	return &NodeBreaker{
 		timeData:            newLenLimitCounter(retention),
-		successData:         newLenLimitCounter(retention),
+		failures:            newLenLimitCounter(retention),
 		rate:                errorRate,
 		callTimeLimit:       callTimeLimit,
 		timeLimitPercentile: timeLimitPercentile,
@@ -310,25 +309,25 @@ type NodeBreaker struct {
 	callTimeLimit       time.Duration
 	timeLimitPercentile float64
 	timeData            *lengthDelimitedCounter
-	successData         *lengthDelimitedCounter
+	failures            *lengthDelimitedCounter
 	now                 func() time.Time
 	closeDelay          time.Duration
 	maxDelay            time.Duration
 	state               *openStateTracker
 }
 
-// Record collects call data and returns bool if breaker should be open
+// Record collects call data and returns bool if breaker should be opened
 func (breaker *NodeBreaker) Record(duration time.Duration, success bool) bool {
 	breaker.timeData.Add(float64(duration))
-	successValue := float64(1)
+	failValue := float64(1)
 	if success {
-		successValue = float64(0)
+		failValue = float64(0)
 	}
-	breaker.successData.Add(successValue)
+	breaker.failures.Add(failValue)
 	return breaker.ShouldOpen()
 }
 
-// ShouldOpen checks if breaker should be open
+// ShouldOpen checks if breaker should be opened
 func (breaker *NodeBreaker) ShouldOpen() bool {
 	exceeded := breaker.limitsExceeded()
 	if breaker.state != nil {
@@ -368,7 +367,7 @@ func (breaker *NodeBreaker) limitsExceeded() bool {
 	percentile := breaker.timeData.Percentile(breaker.timeLimitPercentile)
 	if percentile > float64(breaker.callTimeLimit) {
 		breaker.openBreaker()
-		log.Debugf("Breaker: time percentile exceeded %f", percentile)
+		log.Debugf("Breaker: time percentile exceeded %f / %f", percentile, float64(breaker.callTimeLimit))
 		return true
 	}
 	return false
@@ -384,12 +383,12 @@ func (breaker *NodeBreaker) openBreaker() {
 
 func (breaker *NodeBreaker) reset() {
 	breaker.timeData.Reset()
-	breaker.successData.Reset()
+	breaker.failures.Reset()
 }
 
 func (breaker *NodeBreaker) errorRate() float64 {
-	sum := breaker.successData.Sum()
-	count := float64(len(breaker.successData.values))
+	sum := breaker.failures.Sum()
+	count := float64(len(breaker.failures.values))
 	return sum / count
 }
 
@@ -522,24 +521,28 @@ func (ms *MeasuredStorage) RoundTrip(req *http.Request) (*http.Response, error)
 	log.Debugf("MeasuredStorage %s: Got request id %s\n", ms.Name, reqID)
 	resp, err := ms.RoundTripper.RoundTrip(req)
 	duration := time.Since(start)
-	success := backend.IsSuccessful(resp, err)
+	success := backendSuccess(resp, err)
 	open := ms.Breaker.Record(duration, success)
-	log.Debugf("MeasuredStorage %s: Request %s took %s was successful: %t, opened breaker %t\n", ms.Name, reqID, duration, success, open)
+	log.Debugf("s %s: Request %s took %s was successful: %t, opened breaker %t\n", ms.Name, reqID, duration, success, open)
 
 	ms.Node.UpdateTimeSpent(duration)
 	ms.Node.SetActive(!open)
-	raportMetrics(ms.RoundTripper, start, open)
+	reportMetrics(ms.RoundTripper, start, open)
 	return resp, err
 }
 
+func backendSuccess(response *http.Response, err error) bool {
+	return err == nil && response != nil && response.StatusCode < 500
+}
+
 // IsActive checks Breaker status propagates it to Node compound
 func (ms *MeasuredStorage) IsActive() bool {
 	isActive := !ms.Breaker.ShouldOpen()
 	ms.Node.SetActive(isActive)
 	return ms.Node.IsActive()
 }
 
-func raportMetrics(rt http.RoundTripper, since time.Time, open bool) {
+func reportMetrics(rt http.RoundTripper, since time.Time, open bool) {
 	if b, ok := rt.(*backend.Backend); ok {
 		prefix := fmt.Sprintf("reqs.backend.%s.balancer", b.Name)
 		metrics.UpdateSince(prefix+".duration", since)

balancing/balance_breaker_test.go

@@ -107,27 +107,34 @@ func TestCallMeterConcurrency(t *testing.T) {
 	sampleDuration := time.Millisecond
 	waitGroup := &sync.WaitGroup{}
 	waitGroup.Add(numberOfSamples)
-	callMeter := newCallMeter(5*time.Second, 5*time.Second)
+	clockAdvance := time.Microsecond
+	timer := &mockTimer{
+		baseTime:   time.Now(),
+		advanceDur: clockAdvance}
+	retention := 5 * time.Second
+	resolution := 5 * time.Second
+	callMeter := newCallMeterWithTimer(retention, resolution, timer.now)
+
 	for i := 0; i < numberOfSamples; i++ {
 		go func() {
 			callMeter.UpdateTimeSpent(sampleDuration)
+			timer.advance()
 			waitGroup.Done()
 		}()
 	}
 	waitGroup.Wait()
-	require.Equal(t, float64(numberOfSamples*int(sampleDuration)), callMeter.TimeSpent())
-	require.Equal(t, float64(numberOfSamples), callMeter.Calls())
+	require.Equal(t, float64(numberOfSamples*int(sampleDuration)), callMeter.TimeSpent(), "Time spent missmatch")
+	require.Equal(t, float64(numberOfSamples), callMeter.Calls(), "Number of calls missmatch")
 }
 
 func TestCallMeterRetention(t *testing.T) {
 	numberOfSamples := 100
 	timer := &mockTimer{
 		baseTime:   time.Now(),
 		advanceDur: 100 * time.Millisecond}
-
-	callMeter := newCallMeter(5*time.Second, 1*time.Second)
-	callMeter.now = timer.now
-	callMeter.histogram.now = timer.now
+	retention := 5 * time.Second
+	resolution := time.Second
+	callMeter := newCallMeterWithTimer(retention, resolution, timer.now)
 
 	for i := 0; i < numberOfSamples; i++ {
 		callMeter.UpdateTimeSpent(time.Millisecond)
@@ -196,9 +203,7 @@ func TestCallMeterNoActivity(t *testing.T) {
 	require.Equal(t, expectedCalls, callMeter.Calls())
 	require.Equal(t, expectedTime, callMeter.TimeSpent())
 	timer.baseTime = timer.baseTime.Add(6*resolution + time.Second)
-	fmt.Println("Time shifted", timer.baseTime)
 	require.Equal(t, float64(0), callMeter.TimeSpent())
-	fmt.Println("Update time spent after delay")
 	callMeter.UpdateTimeSpent(timeSpent)
 	timer.baseTime = timer.baseTime.Add(time.Millisecond)
 	require.Equal(t, float64(timeSpent), callMeter.TimeSpent())

config/validator.go

@@ -67,7 +67,6 @@ func (c *YamlConfig) validateRegionCluster(policyName string, policies confregio
 	}
 
 	for _, policy := range policies.Shards {
-		println("dupa")
 		fmt.Printf("sharding policies %v\n", c.Shards)
 		_, exists := c.Shards[policy.ShardName]
 		if !exists {

glide.yaml

@@ -15,7 +15,6 @@ import:
 - package: github.com/serialx/hashring
 - package: github.com/sirupsen/logrus
   version: ^1.0.3
-
 - package: golang.org/x/sync
   subpackages:
   - syncmap
@@ -24,13 +23,15 @@ import:
 - package: gopkg.in/tylerb/graceful.v1
   version: ^1.2.15
 - package: gopkg.in/yaml.v2
-- package: github.com/bnogas/minio-go/
+- package: github.com/bnogas/minio-go
   version: verify_signature
   subpackages:
   - s3signer
 - package: github.com/hashicorp/consul
   subpackages:
   - api
+- package: github.com/jinzhu/gorm
+  version: ^1.9.2
 testImport:
 - package: github.com/DATA-DOG/go-sqlmock
   version: ^1.2.0

storages/shardclient.go

@@ -50,7 +50,7 @@ func (c *ShardClient) balancerRoundTrip(req *http.Request) (resp *http.Response,
 	for node := c.balancer.GetMostAvailable(notFoundNodes...); node != nil; node = c.balancer.GetMostAvailable(notFoundNodes...) {
 		log.Printf("Balancer roundTrip node loop %s %s", node.Name, reqID)
 		if node == nil {
-			return nil, fmt.Errorf("no avialable node")
+			return nil, fmt.Errorf("no available node")
 		}
 		resp, err = node.RoundTrip(req)
 		if (resp == nil && err != balancing.ErrNoActiveNodes) || resp.StatusCode == http.StatusNotFound {