Repository: alibaba/sentinel-golang
Pull request #362 (2020-12-22 09:55:02)

sync special/metric-event branch changes to new branch
<!--  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


### 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


### Special notes for reviews


Patch:
diff --git a/core/base/metric_item.go b/core/base/metric_item.go
index 4413477..da471c8 100644
--- a/core/base/metric_item.go
+++ b/core/base/metric_item.go
@@ -27,17 +27,18 @@ const metricPartSeparator = "|"
 
 // MetricItem represents the data of metric log per line.
 type MetricItem struct {
-	Resource       string
-	Classification int32
-	Timestamp      uint64
-
-	PassQps         uint64
-	BlockQps        uint64
-	CompleteQps     uint64
-	ErrorQps        uint64
-	AvgRt           uint64
-	OccupiedPassQps uint64
-	Concurrency     uint32
+	Resource             string
+	Classification       int32
+	Timestamp            uint64
+	PassQps              uint64
+	BlockQps             uint64
+	MonitorBlockQps      uint64
+	CompleteQps          uint64
+	ErrorQps             uint64
+	AvgRt                uint64
+	OccupiedPassQps      uint64
+	Concurrency          uint32
+	SecondMaxConcurrency uint32
 }
 
 type MetricItemRetriever interface {
@@ -49,10 +50,10 @@ func (m *MetricItem) ToFatString() (string, error) {
 	timeStr := util.FormatTimeMillis(m.Timestamp)
 	// All "|" in the resource name will be replaced with "_"
 	finalName := strings.ReplaceAll(m.Resource, "|", "_")
-	_, err := fmt.Fprintf(&b, "%d|%s|%s|%d|%d|%d|%d|%d|%d|%d|%d",
+	_, err := fmt.Fprintf(&b, "%d|%s|%s|%d|%d|%d|%d|%d|%d|%d|%d|%d",
 		m.Timestamp, timeStr, finalName, m.PassQps,
 		m.BlockQps, m.CompleteQps, m.ErrorQps, m.AvgRt,
-		m.OccupiedPassQps, m.Concurrency, m.Classification)
+		m.OccupiedPassQps, m.Concurrency, m.Classification, m.MonitorBlockQps)
 	if err != nil {
 		return "", err
 	}
@@ -62,10 +63,10 @@ func (m *MetricItem) ToFatString() (string, error) {
 func (m *MetricItem) ToThinString() (string, error) {
 	b := strings.Builder{}
 	finalName := strings.ReplaceAll(m.Resource, "|", "_")
-	_, err := fmt.Fprintf(&b, "%d|%s|%d|%d|%d|%d|%d|%d|%d|%d",
+	_, err := fmt.Fprintf(&b, "%d|%s|%d|%d|%d|%d|%d|%d|%d|%d|%d",
 		m.Timestamp, finalName, m.PassQps,
 		m.BlockQps, m.CompleteQps, m.ErrorQps, m.AvgRt,
-		m.OccupiedPassQps, m.Concurrency, m.Classification)
+		m.OccupiedPassQps, m.Concurrency, m.Classification, m.MonitorBlockQps)
 	if err != nil {
 		return "", err
 	}
@@ -134,5 +135,12 @@ func MetricItemFromFatString(line string) (*MetricItem, error) {
 		}
 		item.Classification = int32(cl)
 	}
+	if len(arr) >= 12 {
+		mb, err := strconv.ParseUint(arr[11], 10, 64)
+		if err != nil {
+			return nil, err
+		}
+		item.MonitorBlockQps = mb
+	}
 	return item, nil
 }
diff --git a/core/base/stat.go b/core/base/stat.go
index c4b27fc..5559ab5 100644
--- a/core/base/stat.go
+++ b/core/base/stat.go
@@ -35,6 +35,8 @@ const (
 	MetricEventError
 	// request execute rt, unit is millisecond
 	MetricEventRt
+	// Specific for verifying mode
+	MetricEventMonitorBlock
 	// hack for the number of event
 	MetricEventTotal
 )
diff --git a/core/log/metric/aggregator.go b/core/log/metric/aggregator.go
index 43a3539..b117487 100644
--- a/core/log/metric/aggregator.go
+++ b/core/log/metric/aggregator.go
@@ -74,6 +74,11 @@ func InitTask() (err error) {
 	return err
 }
 
+// StopTask stops the task.
+func StopTask() {
+	stopChan <- struct{}{}
+}
+
 func writeTaskLoop() {
 	for {
 		select {
diff --git a/core/stat/base/bucket_leap_array.go b/core/stat/base/bucket_leap_array.go
index 5050d2c..66daa4a 100644
--- a/core/stat/base/bucket_leap_array.go
+++ b/core/stat/base/bucket_leap_array.go
@@ -111,6 +111,29 @@ func (bla *BucketLeapArray) addCountWithTime(now uint64, event base.MetricEvent,
 	b.Add(event, count)
 }
 
+func (bla *BucketLeapArray) UpdateMaxConcurrency(count int64) {
+	curBucket, err := bla.data.CurrentBucket(bla)
+	if err != nil {
+		logging.Error(err, "Failed to get current bucket")
+		return
+	}
+	if curBucket == nil {
+		logging.Error(errors.New("Failed to add count: current bucket is nil"), "Failed to add count: current bucket is nil")
+		return
+	}
+	mb := curBucket.Value.Load()
+	if mb == nil {
+		logging.Error(errors.New("Failed to add count: current bucket atomic value is nil"), "Failed to add count: current bucket atomic value is nil")
+		return
+	}
+	b, ok := mb.(*MetricBucket)
+	if !ok {
+		logging.Error(errors.New("Failed to add count: bucket data type error"), "Failed to add count: bucket data type error")
+		return
+	}
+	b.UpdateMaxConcurrency(count)
+}
+
 // Read method, need to adapt upper application
 // it might panic
 func (bla *BucketLeapArray) Count(event base.MetricEvent) int64 {
diff --git a/core/stat/base/metric_bucket.go b/core/stat/base/metric_bucket.go
index f7ed20f..b5d2c5c 100644
--- a/core/stat/base/metric_bucket.go
+++ b/core/stat/base/metric_bucket.go
@@ -26,8 +26,9 @@ import (
 // Note that all operations of the MetricBucket are required to be thread-safe.
 type MetricBucket struct {
 	// Value of statistic
-	counter [base.MetricEventTotal]int64
-	minRt   int64
+	counter        [base.MetricEventTotal]int64
+	minRt          int64
+	maxConcurrency int64
 }
 
 func NewMetricBucket() *MetricBucket {
@@ -80,3 +81,13 @@ func (mb *MetricBucket) AddRt(rt int64) {
 func (mb *MetricBucket) MinRt() int64 {
 	return atomic.LoadInt64(&mb.minRt)
 }
+
+func (mb *MetricBucket) UpdateMaxConcurrency(count int64) {
+	if count > atomic.LoadInt64(&mb.maxConcurrency) {
+		atomic.StoreInt64(&mb.maxConcurrency, count)
+	}
+}
+
+func (mb *MetricBucket) MaxConcurrency() int64 {
+	return atomic.LoadInt64(&mb.maxConcurrency)
+}
diff --git a/core/stat/base/sliding_window_metric.go b/core/stat/base/sliding_window_metric.go
index a51b5a5..4d779ac 100644
--- a/core/stat/base/sliding_window_metric.go
+++ b/core/stat/base/sliding_window_metric.go
@@ -15,6 +15,7 @@
 package base
 
 import (
+	"fmt"
 	"reflect"
 	"sync/atomic"
 
@@ -214,6 +215,7 @@ func (m *SlidingWindowMetric) metricItemFromBuckets(ts uint64, ws []*BucketWrap)
 		item.BlockQps += uint64(mb.Get(base.MetricEventBlock))
 		item.ErrorQps += uint64(mb.Get(base.MetricEventError))
 		item.CompleteQps += uint64(mb.Get(base.MetricEventComplete))
+		item.MonitorBlockQps += uint64(mb.Get(base.MetricEventMonitorBlock))
 		allRt += mb.Get(base.MetricEventRt)
 	}
 	if item.CompleteQps > 0 {
@@ -237,11 +239,12 @@ func (m *SlidingWindowMetric) metricItemFromBucket(w *BucketWrap) *base.MetricIt
 	}
 	completeQps := mb.Get(base.MetricEventComplete)
 	item := &base.MetricItem{
-		PassQps:     uint64(mb.Get(base.MetricEventPass)),
-		BlockQps:    uint64(mb.Get(base.MetricEventBlock)),
-		ErrorQps:    uint64(mb.Get(base.MetricEventError)),
-		CompleteQps: uint64(completeQps),
-		Timestamp:   w.BucketStart,
+		PassQps:         uint64(mb.Get(base.MetricEventPass)),
+		BlockQps:        uint64(mb.Get(base.MetricEventBlock)),
+		MonitorBlockQps: uint64(mb.Get(base.MetricEventMonitorBlock)),
+		ErrorQps:        uint64(mb.Get(base.MetricEventError)),
+		CompleteQps:     uint64(completeQps),
+		Timestamp:       w.BucketStart,
 	}
 	if completeQps > 0 {
 		item.AvgRt = uint64(mb.Get(base.MetricEventRt) / completeQps)
@@ -250,3 +253,59 @@ func (m *SlidingWindowMetric) metricItemFromBucket(w *BucketWrap) *base.MetricIt
 	}
 	return item
 }
+
+func (m *SlidingWindowMetric) MaxConcurrency() int64 {
+	now := util.CurrentTimeMillis()
+	start, end := m.getBucketStartRange(now)
+	satisfiedBuckets := m.real.ValuesConditional(now, func(ws uint64) bool {
+		return ws >= start && ws <= end
+	})
+	var maxConcurrency int64
+	for _, w := range satisfiedBuckets {
+		mb := w.Value.Load()
+		if mb == nil {
+			logging.Error(errors.New("Illegal state: current bucket value is nil when calculating max concurrency"),
+				"Illegal state: current bucket value is nil when calculating max concurrency")
+			continue
+		}
+		counter, ok := mb.(*MetricBucket)
+		if !ok {
+			logging.Error(errors.New("failed to cast to type MetricBucket"),
+				fmt.Sprintf("Failed to cast data value(%+v) to MetricBucket type", mb))
+			continue
+		}
+		v := counter.MaxConcurrency()
+		if v > maxConcurrency {
+			maxConcurrency = v
+		}
+	}
+	return maxConcurrency
+}
+
+func (m *SlidingWindowMetric) SecondMaxConcurrency() int64 {
+	end := util.CurrentTimeMillis()
+	start := end - 1000
+	satisfiedBuckets := m.real.ValuesConditional(end, func(ws uint64) bool {
+		return ws >= start && ws <= end
+	})
+	var maxConcurrency int64
+	for _, w := range satisfiedBuckets {
+		mb := w.Value.Load()
+		if mb == nil {
+			logging.Error(errors.New("Illegal state: current bucket value is nil when calculating max concurrency"),
+				"Illegal state: current bucket value is nil when calculating max concurrency")
+			continue
+		}
+		counter, ok := mb.(*MetricBucket)
+		if !ok {
+			logging.Error(errors.New("failed to cast to type MetricBucket"),
+				fmt.Sprintf("Failed to cast data value(%+v) to MetricBucket type", mb))
+			continue
+		}
+		v := counter.MaxConcurrency()
+		if v > maxConcurrency {
+			maxConcurrency = v
+		}
+	}
+	return maxConcurrency
+}
diff --git a/core/stat/base_node.go b/core/stat/base_node.go
index 255be66..b59cbda 100644
--- a/core/stat/base_node.go
+++ b/core/stat/base_node.go
@@ -80,12 +80,23 @@ func (n *BaseStatNode) MinRT() float64 {
 	return float64(n.metric.MinRT())
 }
 
+// MaxConcurrency returns the max concurrency count of whole sliding window.
+func (n *BaseStatNode) MaxConcurrency() int64 {
+	return n.metric.MaxConcurrency()
+}
+
+// SecondMaxConcurrency returns the max concurrency count of latest second.
+func (n *BaseStatNode) SecondMaxConcurrency() int64 {
+	return n.metric.SecondMaxConcurrency()
+}
+
 func (n *BaseStatNode) CurrentConcurrency() int32 {
 	return atomic.LoadInt32(&(n.concurrency))
 }
 
 func (n *BaseStatNode) IncreaseConcurrency() {
 	atomic.AddInt32(&(n.concurrency), 1)
+	n.arr.UpdateMaxConcurrency(int64(n.CurrentConcurrency()))
 }
 
 func (n *BaseStatNode) DecreaseConcurrency() {


Test patch:
diff --git a/core/base/metric_item_test.go b/core/base/metric_item_test.go
index 2547bb6..83acc33 100644
--- a/core/base/metric_item_test.go
+++ b/core/base/metric_item_test.go
@@ -21,7 +21,7 @@ import (
 )
 
 func TestMetricItemFromFatStringLegal(t *testing.T) {
-	line1 := "1564382218000|2019-07-29 14:36:58|/foo/*|4|9|3|0|25|0|2|1"
+	line1 := "1564382218000|2019-07-29 14:36:58|/foo/*|4|9|3|0|25|0|2|1|10"
 	item1, err := MetricItemFromFatString(line1)
 	assert.NoError(t, err)
 	assert.Equal(t, uint64(1564382218000), item1.Timestamp)
@@ -32,6 +32,7 @@ func TestMetricItemFromFatStringLegal(t *testing.T) {
 	assert.Equal(t, uint64(25), item1.AvgRt)
 	assert.Equal(t, "/foo/*", item1.Resource)
 	assert.Equal(t, int32(1), item1.Classification)
+	assert.Equal(t, uint64(10), item1.MonitorBlockQps)
 }
 
 func TestMetricItemFromFatStringIllegal(t *testing.T) {
diff --git a/core/stat/base/bucket_leap_array_test.go b/core/stat/base/bucket_leap_array_test.go
index 43ffa52..8f4a654 100644
--- a/core/stat/base/bucket_leap_array_test.go
+++ b/core/stat/base/bucket_leap_array_test.go
@@ -173,3 +173,13 @@ func TestDataType(t *testing.T) {
 	dataType := bla.DataType()
 	assert.True(t, dataType == "MetricBucket")
 }
+
+func TestBucketLeapArray_UpdateMaxConcurrency(t *testing.T) {
+	bla := NewBucketLeapArray(SampleCount, IntervalInMs)
+	bla.UpdateMaxConcurrency(5)
+	bla.UpdateMaxConcurrency(2)
+	bla.UpdateMaxConcurrency(4)
+	swm, err := NewSlidingWindowMetric(SampleCount, IntervalInMs, bla)
+	assert.Nil(t, err)
+	assert.Equal(t, int64(5), swm.MaxConcurrency())
+}
diff --git a/core/stat/base/metric_bucket_test.go b/core/stat/base/metric_bucket_test.go
index 945e31a..263a20d 100644
--- a/core/stat/base/metric_bucket_test.go
+++ b/core/stat/base/metric_bucket_test.go
@@ -27,7 +27,7 @@ func Test_metricBucket_MemSize(t *testing.T) {
 	mb := NewMetricBucket()
 	t.Log("mb:", mb)
 	size := unsafe.Sizeof(*mb)
-	if size != 48 {
+	if size != 64 {
 		t.Error("unexpect memory size of MetricBucket")
 	}
 }
@@ -135,3 +135,14 @@ func Test_Reset(t *testing.T) {
 	rt := mb.MinRt()
 	assert.True(t, rt == base.DefaultStatisticMaxRt)
 }
+
+func TestMetricBucket_MaxConcurrency(t *testing.T) {
+	mb := NewMetricBucket()
+	mb.UpdateMaxConcurrency(10)
+	assert.Equal(t, int64(10), mb.maxConcurrency)
+	mb.UpdateMaxConcurrency(9)
+	assert.Equal(t, int64(10), mb.maxConcurrency)
+	mb.UpdateMaxConcurrency(11)
+	assert.Equal(t, int64(11), mb.maxConcurrency)
+
+}
diff --git a/core/stat/base_node_test.go b/core/stat/base_node_test.go
new file mode 100644
index 0000000..ba7d711
--- /dev/null
+++ b/core/stat/base_node_test.go
@@ -0,0 +1,39 @@
+package stat
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+const (
+	// timespan of per slot
+	BucketLengthInMs uint32 = 500
+	// the number of slots
+	SampleCount uint32 = 20
+	// interval(ms) of sliding window, 10s
+	IntervalInMs uint32 = 10 * 1000
+)
+
+func TestBaseStatNodeGoroutineNum(t *testing.T) {
+	bsn := NewBaseStatNode(SampleCount, IntervalInMs)
+	bsn.IncreaseConcurrency()
+	assert.Equal(t, int64(1), bsn.MaxConcurrency())
+	assert.Equal(t, int64(1), bsn.SecondMaxConcurrency())
+	bsn.DecreaseConcurrency()
+	assert.Equal(t, int64(1), bsn.MaxConcurrency())
+	assert.Equal(t, int64(1), bsn.SecondMaxConcurrency())
+
+	bsn.IncreaseConcurrency()
+	bsn.IncreaseConcurrency()
+	assert.Equal(t, int64(2), bsn.MaxConcurrency())
+	assert.Equal(t, int64(2), bsn.SecondMaxConcurrency())
+	bsn.DecreaseConcurrency()
+	bsn.DecreaseConcurrency()
+
+	time.Sleep(time.Second * 1)
+	bsn.IncreaseConcurrency()
+	assert.Equal(t, int64(2), bsn.MaxConcurrency())
+	assert.Equal(t, int64(1), bsn.SecondMaxConcurrency())
+}