stat: Make start time of buckets align with bucketLength in AtomicBucketWrapArray (#197)

louyuting · web-flow · commit 07a5dba85dbd · 2020-08-05T22:21:21.000+08:00
- Fix error param bug when creating LeapArray
- Polish related code and error handling logic
diff --git a/core/circuitbreaker/circuit_breaker.go b/core/circuitbreaker/circuit_breaker.go
@@ -197,12 +197,16 @@ func newSlowRtCircuitBreakerWithStat(r *slowRtRule, stat *slowRequestLeapArray)
 	}
 }
 
-func newSlowRtCircuitBreaker(r *slowRtRule) *slowRtCircuitBreaker {
+func newSlowRtCircuitBreaker(r *slowRtRule) (*slowRtCircuitBreaker, error) {
 	interval := r.StatIntervalMs
 	stat := &slowRequestLeapArray{}
-	stat.data = sbase.NewLeapArray(1, interval, stat)
+	leapArray, err := sbase.NewLeapArray(1, interval, stat)
+	if err != nil {
+		return nil, err
+	}
+	stat.data = leapArray
 
-	return newSlowRtCircuitBreakerWithStat(r, stat)
+	return newSlowRtCircuitBreakerWithStat(r, stat), nil
 }
 
 func (b *slowRtCircuitBreaker) BoundStat() interface{} {
@@ -379,12 +383,15 @@ func newErrorRatioCircuitBreakerWithStat(r *errorRatioRule, stat *errorCounterLe
 	}
 }
 
-func newErrorRatioCircuitBreaker(r *errorRatioRule) *errorRatioCircuitBreaker {
+func newErrorRatioCircuitBreaker(r *errorRatioRule) (*errorRatioCircuitBreaker, error) {
 	interval := r.StatIntervalMs
 	stat := &errorCounterLeapArray{}
-	stat.data = sbase.NewLeapArray(1, interval, stat)
-
-	return newErrorRatioCircuitBreakerWithStat(r, stat)
+	leapArray, err := sbase.NewLeapArray(1, interval, stat)
+	if err != nil {
+		return nil, err
+	}
+	stat.data = leapArray
+	return newErrorRatioCircuitBreakerWithStat(r, stat), nil
 }
 
 func (b *errorRatioCircuitBreaker) BoundStat() interface{} {
@@ -556,12 +563,15 @@ func newErrorCountCircuitBreakerWithStat(r *errorCountRule, stat *errorCounterLe
 	}
 }
 
-func newErrorCountCircuitBreaker(r *errorCountRule) *errorCountCircuitBreaker {
+func newErrorCountCircuitBreaker(r *errorCountRule) (*errorCountCircuitBreaker, error) {
 	interval := r.StatIntervalMs
 	stat := &errorCounterLeapArray{}
-	stat.data = sbase.NewLeapArray(1, interval, stat)
-
-	return newErrorCountCircuitBreakerWithStat(r, stat)
+	leapArray, err := sbase.NewLeapArray(1, interval, stat)
+	if err != nil {
+		return nil, err
+	}
+	stat.data = leapArray
+	return newErrorCountCircuitBreakerWithStat(r, stat), nil
 }
 
 func (b *errorCountCircuitBreaker) BoundStat() interface{} {
diff --git a/core/circuitbreaker/rule_manager.go b/core/circuitbreaker/rule_manager.go
@@ -10,7 +10,7 @@ import (
 	"github.com/pkg/errors"
 )
 
-type CircuitBreakerGenFunc func(r Rule, reuseStat interface{}) CircuitBreaker
+type CircuitBreakerGenFunc func(r Rule, reuseStat interface{}) (CircuitBreaker, error)
 
 var (
 	logger       = logging.GetDefaultLogger()
@@ -24,10 +24,10 @@ var (
 )
 
 func init() {
-	cbGenFuncMap[SlowRequestRatio] = func(r Rule, reuseStat interface{}) CircuitBreaker {
+	cbGenFuncMap[SlowRequestRatio] = func(r Rule, reuseStat interface{}) (CircuitBreaker, error) {
 		rtRule, ok := r.(*slowRtRule)
 		if !ok || rtRule == nil {
-			return nil
+			return nil, errors.Errorf("rule don't match the SlowRequestRatio strategy, rule: %s", r.String())
 		}
 		if reuseStat == nil {
 			return newSlowRtCircuitBreaker(rtRule)
@@ -37,13 +37,13 @@ func init() {
 			logger.Warnf("Expect to generate circuit breaker with reuse statistic, but fail to do type assertion, expect:*slowRequestLeapArray, in fact: %+v", stat)
 			return newSlowRtCircuitBreaker(rtRule)
 		}
-		return newSlowRtCircuitBreakerWithStat(rtRule, stat)
+		return newSlowRtCircuitBreakerWithStat(rtRule, stat), nil
 	}
 
-	cbGenFuncMap[ErrorRatio] = func(r Rule, reuseStat interface{}) CircuitBreaker {
+	cbGenFuncMap[ErrorRatio] = func(r Rule, reuseStat interface{}) (CircuitBreaker, error) {
 		errRatioRule, ok := r.(*errorRatioRule)
 		if !ok || errRatioRule == nil {
-			return nil
+			return nil, errors.Errorf("rule don't match the ErrorRatio strategy, rule: %s", r.String())
 		}
 		if reuseStat == nil {
 			return newErrorRatioCircuitBreaker(errRatioRule)
@@ -53,13 +53,13 @@ func init() {
 			logger.Warnf("Expect to generate circuit breaker with reuse statistic, but fail to do type assertion, expect:*errorCounterLeapArray, in fact: %+v", stat)
 			return newErrorRatioCircuitBreaker(errRatioRule)
 		}
-		return newErrorRatioCircuitBreakerWithStat(errRatioRule, stat)
+		return newErrorRatioCircuitBreakerWithStat(errRatioRule, stat), nil
 	}
 
-	cbGenFuncMap[ErrorCount] = func(r Rule, reuseStat interface{}) CircuitBreaker {
+	cbGenFuncMap[ErrorCount] = func(r Rule, reuseStat interface{}) (CircuitBreaker, error) {
 		errCountRule, ok := r.(*errorCountRule)
 		if !ok || errCountRule == nil {
-			return nil
+			return nil, errors.Errorf("rule don't match the ErrorCount strategy, rule: %s", r.String())
 		}
 		if reuseStat == nil {
 			return newErrorCountCircuitBreaker(errCountRule)
@@ -69,7 +69,7 @@ func init() {
 			logger.Warnf("Expect to generate circuit breaker with reuse statistic, but fail to do type assertion, expect:*errorCounterLeapArray, in fact: %+v", stat)
 			return newErrorCountCircuitBreaker(errCountRule)
 		}
-		return newErrorCountCircuitBreakerWithStat(errCountRule, stat)
+		return newErrorCountCircuitBreakerWithStat(errCountRule, stat), nil
 	}
 }
 
@@ -223,13 +223,14 @@ func onRuleUpdate(rules []Rule) (err error) {
 			}
 
 			var cb CircuitBreaker
+			var e error
 			if reuseStatIdx >= 0 {
-				cb = generator(r, oldResCbs[reuseStatIdx].BoundStat())
+				cb, e = generator(r, oldResCbs[reuseStatIdx].BoundStat())
 			} else {
-				cb = generator(r, nil)
+				cb, e = generator(r, nil)
 			}
-			if cb == nil {
-				logger.Warnf("Ignoring the rule due to bad generated circuit breaker: %v", r)
+			if cb == nil || e != nil {
+				logger.Warnf("Ignoring the rule due to bad generated circuit breaker, r: %s, err: %+v", r.String(), e)
 				continue
 			}
 
diff --git a/core/stat/base/atomic_window_wrap_array_test.go b/core/stat/base/atomic_window_wrap_array_test.go
@@ -9,52 +9,9 @@ import (
 	"time"
 
 	"github.com/alibaba/sentinel-golang/util"
+	"github.com/stretchr/testify/assert"
 )
 
-func Test_newAtomicBucketWrapArray_normal(t *testing.T) {
-	type args struct {
-		len              int
-		bucketLengthInMs uint32
-		bg               BucketGenerator
-	}
-	tests := []struct {
-		name string
-		args args
-		want *AtomicBucketWrapArray
-	}{
-		{
-			name: "Test_newAtomicBucketWrapArray_normal",
-			args: args{
-				len:              int(SampleCount),
-				bucketLengthInMs: BucketLengthInMs,
-				bg:               &leapArrayMock{},
-			},
-			want: nil,
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			ret := NewAtomicBucketWrapArray(tt.args.len, tt.args.bucketLengthInMs, tt.args.bg)
-			if ret == nil || uintptr(ret.base) == uintptr(0) || ret.length != tt.args.len || ret.data == nil || len(ret.data) == 0 {
-				t.Errorf("NewAtomicBucketWrapArray() %+v is illegal.\n", ret)
-				return
-			}
-			dataNil := false
-			for _, v := range ret.data {
-				if v == nil {
-					dataNil = true
-					break
-				}
-			}
-			if dataNil {
-				t.Error("NewAtomicBucketWrapArray exists nil BucketWrap.")
-			}
-
-		})
-	}
-}
-
 func Test_atomicBucketWrapArray_elementOffset(t *testing.T) {
 	type args struct {
 		len              int
@@ -80,7 +37,8 @@ func Test_atomicBucketWrapArray_elementOffset(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			aa := NewAtomicBucketWrapArray(tt.args.len, tt.args.bucketLengthInMs, tt.args.bg)
+			now := uint64(1596199310000)
+			aa := NewAtomicBucketWrapArrayWithTime(tt.args.len, tt.args.bucketLengthInMs, now, tt.args.bg)
 			if got := uintptr(aa.elementOffset(tt.args.idx)) - uintptr(aa.base); got != tt.want {
 				t.Errorf("AtomicBucketWrapArray.elementOffset() = %v, want %v \n", got, tt.want)
 			}
@@ -113,7 +71,8 @@ func Test_atomicBucketWrapArray_get(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			aa := NewAtomicBucketWrapArray(tt.args.len, tt.args.bucketLengthInMs, tt.args.bg)
+			now := uint64(1596199310000)
+			aa := NewAtomicBucketWrapArrayWithTime(tt.args.len, tt.args.bucketLengthInMs, now, tt.args.bg)
 			tt.want = aa.data[9]
 			if got := aa.get(tt.args.idx); !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("AtomicBucketWrapArray.get() = %v, want %v", got, tt.want)
@@ -147,7 +106,8 @@ func Test_atomicBucketWrapArray_compareAndSet(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			aa := NewAtomicBucketWrapArray(tt.args.len, tt.args.bucketLengthInMs, tt.args.bg)
+			now := uint64(1596199310000)
+			aa := NewAtomicBucketWrapArrayWithTime(tt.args.len, tt.args.bucketLengthInMs, now, tt.args.bg)
 			update := &BucketWrap{
 				BucketStart: 8888888888888,
 				Value:       atomic.Value{},
@@ -183,7 +143,8 @@ func taskGet(wg *sync.WaitGroup, at *AtomicBucketWrapArray, t *testing.T) {
 }
 
 func Test_atomicBucketWrapArray_Concurrency_Get(t *testing.T) {
-	ret := NewAtomicBucketWrapArray(int(SampleCount), BucketLengthInMs, &leapArrayMock{})
+	now := uint64(1596199310000)
+	ret := NewAtomicBucketWrapArrayWithTime(int(SampleCount), BucketLengthInMs, now, &leapArrayMock{})
 	for _, ww := range ret.data {
 		c := new(int64)
 		*c = 0
@@ -230,7 +191,8 @@ func taskSet(wg *sync.WaitGroup, at *AtomicBucketWrapArray, t *testing.T) {
 }
 
 func Test_atomicBucketWrapArray_Concurrency_Set(t *testing.T) {
-	ret := NewAtomicBucketWrapArray(int(SampleCount), BucketLengthInMs, &leapArrayMock{})
+	now := uint64(1596199310000)
+	ret := NewAtomicBucketWrapArrayWithTime(int(SampleCount), BucketLengthInMs, now, &leapArrayMock{})
 	for _, ww := range ret.data {
 		c := new(int64)
 		*c = 0
@@ -253,3 +215,43 @@ func Test_atomicBucketWrapArray_Concurrency_Set(t *testing.T) {
 	}
 	t.Log("all done")
 }
+
+func TestNewAtomicBucketWrapArrayWithTime(t *testing.T) {
+	t.Run("TestNewAtomicBucketWrapArrayWithTime_normal", func(t *testing.T) {
+		now := uint64(1596199317001)
+		arrayStartTime := uint64(1596199316000)
+		idx := int((now - arrayStartTime) / 200)
+		a := NewAtomicBucketWrapArrayWithTime(10, 200, now, &leapArrayMock{})
+
+		targetTime := arrayStartTime + uint64(idx*200)
+		for i := idx; i < 10; i++ {
+			b := a.get(i)
+			assert.True(t, b.BucketStart == targetTime, "Check start failed")
+			targetTime += 200
+		}
+		for i := 0; i < idx; i++ {
+			b := a.get(i)
+			assert.True(t, b.BucketStart == targetTime, "Check start failed")
+			targetTime += 200
+		}
+	})
+
+	t.Run("TestNewAtomicBucketWrapArrayWithTime_edge1", func(t *testing.T) {
+		now := uint64(1596199316000)
+		arrayStartTime := uint64(1596199316000)
+		idx := int((now - arrayStartTime) / 200)
+		a := NewAtomicBucketWrapArrayWithTime(10, 200, now, &leapArrayMock{})
+
+		targetTime := arrayStartTime + uint64(idx*200)
+		for i := idx; i < 10; i++ {
+			b := a.get(i)
+			assert.True(t, b.BucketStart == targetTime, "Check start failed")
+			targetTime += 200
+		}
+		for i := 0; i < idx; i++ {
+			b := a.get(i)
+			assert.True(t, b.BucketStart == targetTime, "Check start failed")
+			targetTime += 200
+		}
+	})
+}
diff --git a/core/stat/base/bucket_leap_array_test.go b/core/stat/base/bucket_leap_array_test.go
@@ -95,7 +95,7 @@ func coroutineTask(wg *sync.WaitGroup, slidingWindow *BucketLeapArray, now uint6
 
 func TestBucketLeapArray_resetBucketTo(t *testing.T) {
 	bla := NewBucketLeapArray(SampleCount, IntervalInMs)
-	idx := 6
+	idx := 19
 	oldBucketWrap := bla.data.array.get(idx)
 	oldBucket := oldBucketWrap.Value.Load()
 	if oldBucket == nil {
diff --git a/core/stat/base/leap_array.go b/core/stat/base/leap_array.go
diff --git a/core/stat/base/leap_array_test.go b/core/stat/base/leap_array_test.go

Original file line number	Diff line number	Diff line change
`@@ -10,7 +10,7 @@ import (`
`10`	`10`	`"github.com/pkg/errors"`
`11`	`11`	`)`
`12`	`12`
`13`		`-type CircuitBreakerGenFunc func(r Rule, reuseStat interface{}) CircuitBreaker`
	`13`	`+type CircuitBreakerGenFunc func(r Rule, reuseStat interface{}) (CircuitBreaker, error)`
`14`	`14`
`15`	`15`	`var (`
`16`	`16`	`logger = logging.GetDefaultLogger()`
`@@ -24,10 +24,10 @@ var (`
`24`	`24`	`)`
`25`	`25`
`26`	`26`	`func init() {`
`27`		`- cbGenFuncMap[SlowRequestRatio] = func(r Rule, reuseStat interface{}) CircuitBreaker {`
	`27`	`+ cbGenFuncMap[SlowRequestRatio] = func(r Rule, reuseStat interface{}) (CircuitBreaker, error) {`
`28`	`28`	`rtRule, ok := r.(*slowRtRule)`
`29`	`29`	`if !ok \|\| rtRule == nil {`
`30`		`- return nil`
	`30`	`+ return nil, errors.Errorf("rule don't match the SlowRequestRatio strategy, rule: %s", r.String())`
`31`	`31`	`}`
`32`	`32`	`if reuseStat == nil {`
`33`	`33`	`return newSlowRtCircuitBreaker(rtRule)`
`@@ -37,13 +37,13 @@ func init() {`
`37`	`37`	`logger.Warnf("Expect to generate circuit breaker with reuse statistic, but fail to do type assertion, expect:*slowRequestLeapArray, in fact: %+v", stat)`
`38`	`38`	`return newSlowRtCircuitBreaker(rtRule)`
`39`	`39`	`}`
`40`		`- return newSlowRtCircuitBreakerWithStat(rtRule, stat)`
	`40`	`+ return newSlowRtCircuitBreakerWithStat(rtRule, stat), nil`
`41`	`41`	`}`
`42`	`42`
`43`		`- cbGenFuncMap[ErrorRatio] = func(r Rule, reuseStat interface{}) CircuitBreaker {`
	`43`	`+ cbGenFuncMap[ErrorRatio] = func(r Rule, reuseStat interface{}) (CircuitBreaker, error) {`
`44`	`44`	`errRatioRule, ok := r.(*errorRatioRule)`
`45`	`45`	`if !ok \|\| errRatioRule == nil {`
`46`		`- return nil`
	`46`	`+ return nil, errors.Errorf("rule don't match the ErrorRatio strategy, rule: %s", r.String())`
`47`	`47`	`}`
`48`	`48`	`if reuseStat == nil {`
`49`	`49`	`return newErrorRatioCircuitBreaker(errRatioRule)`
`@@ -53,13 +53,13 @@ func init() {`
`53`	`53`	`logger.Warnf("Expect to generate circuit breaker with reuse statistic, but fail to do type assertion, expect:*errorCounterLeapArray, in fact: %+v", stat)`
`54`	`54`	`return newErrorRatioCircuitBreaker(errRatioRule)`
`55`	`55`	`}`
`56`		`- return newErrorRatioCircuitBreakerWithStat(errRatioRule, stat)`
	`56`	`+ return newErrorRatioCircuitBreakerWithStat(errRatioRule, stat), nil`
`57`	`57`	`}`
`58`	`58`
`59`		`- cbGenFuncMap[ErrorCount] = func(r Rule, reuseStat interface{}) CircuitBreaker {`
	`59`	`+ cbGenFuncMap[ErrorCount] = func(r Rule, reuseStat interface{}) (CircuitBreaker, error) {`
`60`	`60`	`errCountRule, ok := r.(*errorCountRule)`
`61`	`61`	`if !ok \|\| errCountRule == nil {`
`62`		`- return nil`
	`62`	`+ return nil, errors.Errorf("rule don't match the ErrorCount strategy, rule: %s", r.String())`
`63`	`63`	`}`
`64`	`64`	`if reuseStat == nil {`
`65`	`65`	`return newErrorCountCircuitBreaker(errCountRule)`
`@@ -69,7 +69,7 @@ func init() {`
`69`	`69`	`logger.Warnf("Expect to generate circuit breaker with reuse statistic, but fail to do type assertion, expect:*errorCounterLeapArray, in fact: %+v", stat)`
`70`	`70`	`return newErrorCountCircuitBreaker(errCountRule)`
`71`	`71`	`}`
`72`		`- return newErrorCountCircuitBreakerWithStat(errCountRule, stat)`
	`72`	`+ return newErrorCountCircuitBreakerWithStat(errCountRule, stat), nil`
`73`	`73`	`}`
`74`	`74`	`}`
`75`	`75`
`@@ -223,13 +223,14 @@ func onRuleUpdate(rules []Rule) (err error) {`
`223`	`223`	`}`
`224`	`224`
`225`	`225`	`var cb CircuitBreaker`
	`226`	`+ var e error`
`226`	`227`	`if reuseStatIdx >= 0 {`
`227`		`- cb = generator(r, oldResCbs[reuseStatIdx].BoundStat())`
	`228`	`+ cb, e = generator(r, oldResCbs[reuseStatIdx].BoundStat())`
`228`	`229`	`} else {`
`229`		`- cb = generator(r, nil)`
	`230`	`+ cb, e = generator(r, nil)`
`230`	`231`	`}`
`231`		`- if cb == nil {`
`232`		`- logger.Warnf("Ignoring the rule due to bad generated circuit breaker: %v", r)`
	`232`	`+ if cb == nil \|\| e != nil {`
	`233`	`+ logger.Warnf("Ignoring the rule due to bad generated circuit breaker, r: %s, err: %+v", r.String(), e)`
`233`	`234`	`continue`
`234`	`235`	`}`
`235`	`236`