Repository: alibaba/sentinel-golang
Pull request #298 (2020-10-21 10:54:59)

Fix bug that StatIntervalInMs do not take effect in throttling checker
<!--  Thanks for submitting a pull request! Here are some tips for you:
1. Please make sure you have read and understood the contributing guidelines: https://github.com/alibaba/blob/master/CONTRIBUTING.md
2. Please make sure the PR has a corresponding issue.
-->

### Describe what this PR does / why we need it
Fix bug that StatIntervalInMs do not take effect in throttling checker.

As the doc [流量控制](https://github.com/alibaba/sentinel-golang/wiki/%E6%B5%81%E9%87%8F%E6%8E%A7%E5%88%B6) describes:

> 这里特别强调一下StatIntervalInMs和Threshold这两个字段，这两个字段决定了流量控制器的灵敏度。以 Direct + Reject 的流控策略为例，流量控制器的行为就是在StatIntervalInMs周期内，允许的最大请求数量是Threshold。比如，如果StatIntervalInMs是10000，Threshold是10000，那么流量控制器的行为就是10s内运行最多10000次访问。


### Does this pull request fix one issue?

<!--If that, add "Fixes #xxxx" below in the next line. For example, Fixes #15. Otherwise, add "NONE" -->

### Describe how you did it


### Describe how to verify it
By unit test and example.


### Special notes for reviews

codecov-io: # [Codecov](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=h1) Report
> Merging [#298](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=desc) into [master](https://codecov.io/gh/alibaba/sentinel-golang/commit/a9496edafe1a2a6d13085db85ad1cec403a2ed2d?el=desc) will **increase** coverage by `0.01%`.
> The diff coverage is `59.57%`.

[![Impacted file tree graph](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/graphs/tree.svg?width=650&height=150&src=pr&token=SrpzbhKA3E)](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #298      +/-   ##
==========================================
+ Coverage   51.44%   51.46%   +0.01%     
==========================================
  Files          80       81       +1     
  Lines        4010     4104      +94     
==========================================
+ Hits         2063     2112      +49     
- Misses       1661     1706      +45     
  Partials      286      286              
```


| [Impacted Files](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [api/api.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-YXBpL2FwaS5nbw==) | `19.44% <0.00%> (ø)` | |
| [api/slot\_chain.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-YXBpL3Nsb3RfY2hhaW4uZ28=) | `76.47% <0.00%> (-16.87%)` | :arrow_down: |
| [core/base/stat.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9iYXNlL3N0YXQuZ28=) | `52.38% <0.00%> (-26.20%)` | :arrow_down: |
| [core/stat/base/bucket\_leap\_array.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9zdGF0L2Jhc2UvYnVja2V0X2xlYXBfYXJyYXkuZ28=) | `59.45% <ø> (+1.56%)` | :arrow_up: |
| [core/stat/base/metric\_bucket.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9zdGF0L2Jhc2UvbWV0cmljX2J1Y2tldC5nbw==) | `71.42% <0.00%> (-7.52%)` | :arrow_down: |
| [logging/logging.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-bG9nZ2luZy9sb2dnaW5nLmdv) | `63.70% <0.00%> (-0.52%)` | :arrow_down: |
| [core/misc/resource\_slot\_chain.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9taXNjL3Jlc291cmNlX3Nsb3RfY2hhaW4uZ28=) | `37.73% <28.57%> (ø)` | |
| [core/stat/base/leap\_array.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9zdGF0L2Jhc2UvbGVhcF9hcnJheS5nbw==) | `71.87% <45.45%> (-3.41%)` | :arrow_down: |
| [core/flow/rule\_manager.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9mbG93L3J1bGVfbWFuYWdlci5nbw==) | `61.53% <52.94%> (+0.17%)` | :arrow_up: |
| [core/flow/tc\_throttling.go](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree#diff-Y29yZS9mbG93L3RjX3Rocm90dGxpbmcuZ28=) | `66.66% <78.57%> (+3.70%)` | :arrow_up: |
| ... and [8 more](https://codecov.io/gh/alibaba/sentinel-golang/pull/298/diff?src=pr&el=tree-more) | |

------

[Continue to review full report at Codecov](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=footer). Last update [a9496ed...3cc1e05](https://codecov.io/gh/alibaba/sentinel-golang/pull/298?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/core/flow/rule_manager.go b/core/flow/rule_manager.go
index caa4750..77b10a4 100644
--- a/core/flow/rule_manager.go
+++ b/core/flow/rule_manager.go
@@ -68,7 +68,7 @@ func init() {
 			return nil, err
 		}
 		tsc.flowCalculator = NewDirectTrafficShapingCalculator(tsc, rule.Threshold)
-		tsc.flowChecker = NewThrottlingChecker(tsc, rule.MaxQueueingTimeMs)
+		tsc.flowChecker = NewThrottlingChecker(tsc, rule.MaxQueueingTimeMs, rule.StatIntervalInMs)
 		return tsc, nil
 	}
 	tcGenFuncMap[trafficControllerGenKey{
@@ -106,7 +106,7 @@ func init() {
 			return nil, err
 		}
 		tsc.flowCalculator = NewWarmUpTrafficShapingCalculator(tsc, rule)
-		tsc.flowChecker = NewThrottlingChecker(tsc, rule.MaxQueueingTimeMs)
+		tsc.flowChecker = NewThrottlingChecker(tsc, rule.MaxQueueingTimeMs, rule.StatIntervalInMs)
 		return tsc, nil
 	}
 }
@@ -355,13 +355,13 @@ func getTrafficControllerListFor(name string) []*TrafficShapingController {
 	return tcMap[name]
 }
 
-func calculateReuseIndexFor(r *Rule, oldResCbs []*TrafficShapingController) (equalIdx, reuseStatIdx int) {
-	// the index of equivalent rule in old circuit breaker slice
+func calculateReuseIndexFor(r *Rule, oldResTcs []*TrafficShapingController) (equalIdx, reuseStatIdx int) {
+	// the index of equivalent rule in old traffic shaping controller slice
 	equalIdx = -1
-	// the index of statistic reusable rule in old circuit breaker slice
+	// the index of statistic reusable rule in old traffic shaping controller slice
 	reuseStatIdx = -1
 
-	for idx, oldTc := range oldResCbs {
+	for idx, oldTc := range oldResTcs {
 		oldRule := oldTc.BoundRule()
 		if oldRule.isEqualsTo(r) {
 			// break if there is equivalent rule
@@ -399,10 +399,10 @@ func buildRulesOfRes(res string, rulesOfRes []*Rule) []*TrafficShapingController
 
 		// First check equals scenario
 		if equalIdx >= 0 {
-			// reuse the old cb
+			// reuse the old tc
 			equalOldTc := oldResTcs[equalIdx]
 			newTcsOfRes = append(newTcsOfRes, equalOldTc)
-			// remove old cb from oldResCbs
+			// remove old tc from oldResTcs
 			tcMap[res] = append(oldResTcs[:equalIdx], oldResTcs[equalIdx+1:]...)
 			continue
 		}
@@ -428,7 +428,7 @@ func buildRulesOfRes(res string, rulesOfRes []*Rule) []*TrafficShapingController
 			continue
 		}
 		if reuseStatIdx >= 0 {
-			// remove old cb from oldResCbs
+			// remove old tc from oldResTcs
 			tcMap[res] = append(oldResTcs[:reuseStatIdx], oldResTcs[reuseStatIdx+1:]...)
 		}
 		newTcsOfRes = append(newTcsOfRes, tc)
@@ -442,36 +442,36 @@ func IsValidRule(rule *Rule) error {
 		return errors.New("nil Rule")
 	}
 	if rule.Resource == "" {
-		return errors.New("empty resource name")
+		return errors.New("empty Resource")
 	}
 	if rule.Threshold < 0 {
-		return errors.New("negative threshold")
+		return errors.New("negative Threshold")
 	}
 	if int32(rule.TokenCalculateStrategy) < 0 {
-		return errors.New("invalid token calculate strategy")
+		return errors.New("negative TokenCalculateStrategy")
 	}
 	if int32(rule.ControlBehavior) < 0 {
-		return errors.New("invalid control behavior")
+		return errors.New("negative ControlBehavior")
 	}
 	if !(rule.RelationStrategy >= CurrentResource && rule.RelationStrategy <= AssociatedResource) {
-		return errors.New("invalid relation strategy")
+		return errors.New("invalid RelationStrategy")
 	}
 	if rule.RelationStrategy == AssociatedResource && rule.RefResource == "" {
-		return errors.New("Bad flow rule: invalid relation strategy")
+		return errors.New("RefResource must be non empty when RelationStrategy is AssociatedResource")
 	}
 	if rule.TokenCalculateStrategy == WarmUp {
 		if rule.WarmUpPeriodSec <= 0 {
-			return errors.New("invalid WarmUpPeriodSec")
+			return errors.New("WarmUpPeriodSec must be great than 0")
 		}
 		if rule.WarmUpColdFactor == 1 {
 			return errors.New("WarmUpColdFactor must be great than 1")
 		}
 	}
 	if rule.ControlBehavior == Throttling && rule.MaxQueueingTimeMs == 0 {
-		return errors.New("invalid MaxQueueingTimeMs")
+		return errors.New("MaxQueueingTimeMs can't be 0 when control behavior is Throttling")
 	}
-	if rule.StatIntervalInMs > config.GlobalStatisticIntervalMsTotal()*60 {
-		return errors.New("StatIntervalInMs must be less than 10 minutes")
+	if rule.StatIntervalInMs > 10*60*1000 {
+		logging.Info("StatIntervalInMs is great than 10 minutes, less than 10 minutes is recommended.")
 	}
 	return nil
 }
diff --git a/core/flow/tc_throttling.go b/core/flow/tc_throttling.go
index 19eded2..43c6ce3 100644
--- a/core/flow/tc_throttling.go
+++ b/core/flow/tc_throttling.go
@@ -3,25 +3,35 @@ package flow
 import (
 	"math"
 	"sync/atomic"
-	"time"
 
 	"github.com/alibaba/sentinel-golang/core/base"
+	"github.com/alibaba/sentinel-golang/core/config"
 	"github.com/alibaba/sentinel-golang/util"
 )
 
-const nanoUnitOffset = time.Second / time.Nanosecond
-
 // ThrottlingChecker limits the time interval between two requests.
 type ThrottlingChecker struct {
 	owner             *TrafficShapingController
 	maxQueueingTimeNs uint64
+	statIntervalNs    uint64
 	lastPassedTime    uint64
 }
 
-func NewThrottlingChecker(owner *TrafficShapingController, timeoutMs uint32) *ThrottlingChecker {
+func NewThrottlingChecker(owner *TrafficShapingController, timeoutMs uint32, statIntervalMs uint32) *ThrottlingChecker {
+	var statIntervalNs uint64
+	if statIntervalMs == 0 {
+		defaultIntervalMs := config.MetricStatisticIntervalMs()
+		if defaultIntervalMs == 0 {
+			defaultIntervalMs = 1000
+		}
+		statIntervalNs = uint64(defaultIntervalMs) * util.UnixTimeUnitOffset
+	} else {
+		statIntervalNs = uint64(statIntervalMs) * util.UnixTimeUnitOffset
+	}
 	return &ThrottlingChecker{
 		owner:             owner,
 		maxQueueingTimeNs: uint64(timeoutMs) * util.UnixTimeUnitOffset,
+		statIntervalNs:    statIntervalNs,
 		lastPassedTime:    0,
 	}
 }
@@ -39,26 +49,28 @@ func (c *ThrottlingChecker) DoCheck(_ base.StatNode, batchCount uint32, threshol
 	}
 	// Here we use nanosecond so that we could control the queueing time more accurately.
 	curNano := util.CurrentTimeNano()
+
 	// The interval between two requests (in nanoseconds).
-	interval := uint64(math.Ceil(float64(batchCount) / threshold * float64(nanoUnitOffset)))
+	intervalNs := uint64(math.Ceil(float64(batchCount) / threshold * float64(c.statIntervalNs)))
 
 	// Expected pass time of this request.
-	expectedTime := atomic.LoadUint64(&c.lastPassedTime) + interval
+	expectedTime := atomic.LoadUint64(&c.lastPassedTime) + intervalNs
 	if expectedTime <= curNano {
 		// Contention may exist here, but it's okay.
 		atomic.StoreUint64(&c.lastPassedTime, curNano)
 		return nil
 	}
-	estimatedQueueingDuration := atomic.LoadUint64(&c.lastPassedTime) + interval - util.CurrentTimeNano()
+
+	estimatedQueueingDuration := atomic.LoadUint64(&c.lastPassedTime) + intervalNs - util.CurrentTimeNano()
 	if estimatedQueueingDuration > c.maxQueueingTimeNs {
 		return base.NewTokenResultBlocked(base.BlockTypeFlow)
 	}
 
-	oldTime := atomic.AddUint64(&c.lastPassedTime, interval)
+	oldTime := atomic.AddUint64(&c.lastPassedTime, intervalNs)
 	estimatedQueueingDuration = oldTime - util.CurrentTimeNano()
 	if estimatedQueueingDuration > c.maxQueueingTimeNs {
 		// Subtract the interval.
-		atomic.AddUint64(&c.lastPassedTime, ^(interval - 1))
+		atomic.AddUint64(&c.lastPassedTime, ^(intervalNs - 1))
 		return base.NewTokenResultBlocked(base.BlockTypeFlow)
 	}
 	if estimatedQueueingDuration > 0 {


Test patch:
diff --git a/core/flow/rule_manager_test.go b/core/flow/rule_manager_test.go
index 1910bda..9081498 100644
--- a/core/flow/rule_manager_test.go
+++ b/core/flow/rule_manager_test.go
@@ -54,7 +54,7 @@ func TestIsValidFlowRule(t *testing.T) {
 	badRule1 := &Rule{Threshold: 1, Resource: ""}
 	badRule2 := &Rule{Threshold: -1.9, Resource: "test"}
 	badRule3 := &Rule{Threshold: 5, Resource: "test", TokenCalculateStrategy: WarmUp, ControlBehavior: Reject}
-	goodRule1 := &Rule{Threshold: 10, Resource: "test", TokenCalculateStrategy: WarmUp, ControlBehavior: Throttling, WarmUpPeriodSec: 10, MaxQueueingTimeMs: 10}
+	goodRule1 := &Rule{Threshold: 10, Resource: "test", TokenCalculateStrategy: WarmUp, ControlBehavior: Throttling, WarmUpPeriodSec: 10, MaxQueueingTimeMs: 10, StatIntervalInMs: 1000}
 	badRule4 := &Rule{Threshold: 5, Resource: "test", TokenCalculateStrategy: WarmUp, ControlBehavior: Reject, StatIntervalInMs: 6000000}
 
 	assert.Error(t, IsValidRule(badRule1))
@@ -88,6 +88,7 @@ func TestGetRules(t *testing.T) {
 			RefResource:            "",
 			WarmUpPeriodSec:        0,
 			MaxQueueingTimeMs:      10,
+			StatIntervalInMs:       1000,
 		}
 		if _, err := LoadRules([]*Rule{r1, r2}); err != nil {
 			t.Fatal(err)
@@ -130,6 +131,7 @@ func TestGetRules(t *testing.T) {
 			RefResource:            "",
 			WarmUpPeriodSec:        0,
 			MaxQueueingTimeMs:      10,
+			StatIntervalInMs:       1000,
 		}
 		if _, err := LoadRules([]*Rule{r1, r2}); err != nil {
 			t.Fatal(err)
diff --git a/core/flow/tc_throttling_test.go b/core/flow/tc_throttling_test.go
index 929f350..0fb1e46 100644
--- a/core/flow/tc_throttling_test.go
+++ b/core/flow/tc_throttling_test.go
@@ -11,47 +11,61 @@ import (
 )
 
 func TestThrottlingChecker_DoCheckNoQueueingSingleThread(t *testing.T) {
-	tc := NewThrottlingChecker(nil, 0)
-	var qps float64 = 5.0
+	intervalMs := 10000
+	threshold := 50.0
+	timeoutMs := 0
+
+	tc := NewThrottlingChecker(nil, uint32(timeoutMs), uint32(intervalMs))
 
 	// The first request will pass.
-	ret := tc.DoCheck(nil, 1, qps)
+	ret := tc.DoCheck(nil, 1, threshold)
 	assert.True(t, ret == nil || ret.IsPass())
 
-	for i := 0; i < int(qps); i++ {
-		assert.True(t, tc.DoCheck(nil, 1, qps).IsBlocked())
+	reqCount := 10
+	for i := 0; i < reqCount; i++ {
+		assert.True(t, tc.DoCheck(nil, 1, threshold).IsBlocked())
 	}
-	time.Sleep(time.Duration(1000/int(qps)+10) * time.Millisecond)
+	time.Sleep(time.Duration(intervalMs/int(threshold)*reqCount+10) * time.Millisecond)
 
-	assert.True(t, tc.DoCheck(nil, 1, qps) == nil)
-	assert.True(t, tc.DoCheck(nil, 1, qps).IsBlocked())
+	assert.True(t, tc.DoCheck(nil, 1, threshold) == nil)
+	assert.True(t, tc.DoCheck(nil, 1, threshold).IsBlocked())
 }
 
 func TestThrottlingChecker_DoCheckSingleThread(t *testing.T) {
-	tc := NewThrottlingChecker(nil, 1000)
-	var qps float64 = 5.0
+	intervalMs := 10000
+	threshold := 50.0
+	timeoutMs := 2000
+
+	tc := NewThrottlingChecker(nil, uint32(timeoutMs), uint32(intervalMs))
+
 	resultList := make([]*base.TokenResult, 0)
-	for i := 0; i < 10; i++ {
-		res := tc.DoCheck(nil, 1, qps)
+	reqCount := 20
+	for i := 0; i < reqCount; i++ {
+		res := tc.DoCheck(nil, 1, threshold)
 		resultList = append(resultList, res)
 	}
 	assert.True(t, resultList[0] == nil)
 
-	for i := 1; i <= int(qps); i++ {
+	// waitCount is count of request that will wait and not be blocked
+	waitCount := int(float64(timeoutMs) / (float64(intervalMs) / threshold))
+	for i := 1; i <= waitCount; i++ {
 		assert.True(t, resultList[i].Status() == base.ResultStatusShouldWait)
 		wt := resultList[i].WaitMs()
-		assert.InEpsilon(t, i*1000/int(qps), wt, 10)
+		assert.InEpsilon(t, i*1000/int(waitCount), wt, 10)
 	}
-	for i := int(qps) + 1; i < 10; i++ {
+	for i := waitCount + 1; i < reqCount; i++ {
 		assert.True(t, resultList[i].IsBlocked())
 	}
 }
 
 func TestThrottlingChecker_DoCheckQueueingParallel(t *testing.T) {
-	tc := NewThrottlingChecker(nil, 1000)
-	var qps float64 = 5.0
+	intervalMs := 10000
+	threshold := 50.0
+	timeoutMs := 0
+
+	tc := NewThrottlingChecker(nil, uint32(timeoutMs), uint32(intervalMs))
 
-	assert.True(t, tc.DoCheck(nil, 1, qps) == nil)
+	assert.True(t, tc.DoCheck(nil, 1, threshold) == nil)
 
 	wg := &sync.WaitGroup{}
 	gc := 24
@@ -60,7 +74,7 @@ func TestThrottlingChecker_DoCheckQueueingParallel(t *testing.T) {
 	var waitCount, blockCount int32 = 0, 0
 	for i := 0; i < gc; i++ {
 		go func() {
-			res := tc.DoCheck(nil, 1, qps)
+			res := tc.DoCheck(nil, 1, threshold)
 			if res.IsBlocked() {
 				atomic.AddInt32(&blockCount, 1)
 			}
@@ -73,5 +87,5 @@ func TestThrottlingChecker_DoCheckQueueingParallel(t *testing.T) {
 	wg.Wait()
 	assert.Equal(t, int32(gc), waitCount+blockCount)
 	// Non-strict mode may not be strictly accurate, so here we tolerate a delta.
-	assert.InEpsilon(t, qps, waitCount, 1)
+	assert.InEpsilon(t, threshold/(float64(intervalMs)/1000.0), waitCount, 1)
 }