Add system adaptive rule implementation (alibaba#33)

sczyh30 · web-flow · commit c5b08d4db534 · 2020-02-12T10:41:26.000+08:00
Signed-off-by: Eric Zhao &lt;sczyh16@gmail.com&gt;
diff --git a/core/system/rule.go b/core/system/rule.go
@@ -0,0 +1,78 @@
+package system
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type MetricType uint32
+
+const (
+	// Load represents system load1 in Linux/Unix.
+	Load MetricType = iota
+	// AvgRT represents the average response time of all inbound requests.
+	AvgRT
+	// Concurrency represents the concurrency of all inbound requests.
+	Concurrency
+	InboundQPS
+	CpuUsage
+	// MetricTypeSize indicates the enum size of MetricType.
+	MetricTypeSize
+)
+
+func (t MetricType) String() string {
+	switch t {
+	case Load:
+		return "load"
+	case AvgRT:
+		return "avgRT"
+	case Concurrency:
+		return "concurrency"
+	case InboundQPS:
+		return "inboundQPS"
+	case CpuUsage:
+		return "cpuUsage"
+	default:
+		return fmt.Sprintf("unknown(%d)", t)
+	}
+}
+
+type AdaptiveStrategy int32
+
+const (
+	NoAdaptive AdaptiveStrategy = -1
+	BBR        AdaptiveStrategy = iota
+)
+
+func (t AdaptiveStrategy) String() string {
+	switch t {
+	case NoAdaptive:
+		return "none"
+	case BBR:
+		return "bbr"
+	default:
+		return fmt.Sprintf("unknown(%d)", t)
+	}
+}
+
+type SystemRule struct {
+	ID uint64 `json:"id,omitempty"`
+
+	MetricType   MetricType       `json:"metricType"`
+	TriggerCount float64          `json:"count"`
+	Strategy     AdaptiveStrategy `json:"adaptiveStrategy"`
+}
+
+func (r *SystemRule) String() string {
+	b, err := json.Marshal(r)
+	if err != nil {
+		// Return the fallback string
+		return fmt.Sprintf("SystemRule{metricType=%s, triggerCount=%.2f, adaptiveStrategy=%s}",
+			r.MetricType, r.TriggerCount, r.Strategy)
+	}
+	return string(b)
+}
+
+func (r *SystemRule) ResourceName() string {
+	return r.MetricType.String()
+}
diff --git a/core/system/rule_manager.go b/core/system/rule_manager.go
@@ -0,0 +1,97 @@
+package system
+
+import (
+	"github.com/sentinel-group/sentinel-golang/logging"
+	"github.com/sentinel-group/sentinel-golang/util"
+	"sync"
+)
+
+type RuleMap map[MetricType][]*SystemRule
+
+// const
+var (
+	logger = logging.GetDefaultLogger()
+
+	ruleMap    = make(RuleMap, 0)
+	ruleMapMux = new(sync.RWMutex)
+
+	ruleChan     = make(chan []*SystemRule, 10)
+	propertyInit sync.Once
+)
+
+func init() {
+	propertyInit.Do(func() {
+		initRuleRecvTask()
+	})
+}
+
+func initRuleRecvTask() {
+	go util.RunWithRecover(func() {
+		for {
+			select {
+			case rules := <-ruleChan:
+				err := onRuleUpdate(rules)
+				if err != nil {
+					logger.Errorf("Failed to update system rules: %+v", err)
+				}
+			}
+		}
+	}, logger)
+}
+
+func GetRules() []*SystemRule {
+	ruleMapMux.RLock()
+	defer ruleMapMux.RUnlock()
+
+	rules := make([]*SystemRule, 0)
+	for _, rs := range ruleMap {
+		rules = append(rules, rs...)
+	}
+	return rules
+}
+
+// LoadRules loads given system rules to the rule manager, while all previous rules will be replaced.
+func LoadRules(rules []*SystemRule) (bool, error) {
+	ruleChan <- rules
+	return true, nil
+}
+
+func onRuleUpdate(rules []*SystemRule) error {
+	m := buildRuleMap(rules)
+
+	ruleMapMux.Lock()
+	defer ruleMapMux.Unlock()
+
+	ruleMap = m
+	return nil
+}
+
+func buildRuleMap(rules []*SystemRule) RuleMap {
+	if len(rules) == 0 {
+		return make(RuleMap, 0)
+	}
+	m := make(RuleMap, 0)
+	for _, rule := range rules {
+		if !IsValidSystemRule(rule) {
+			logger.Warnf("Ignoring invalid system rule: %v", rule)
+			continue
+		}
+		rulesOfRes, exists := m[rule.MetricType]
+		if !exists {
+			m[rule.MetricType] = []*SystemRule{rule}
+		} else {
+			m[rule.MetricType] = append(rulesOfRes, rule)
+		}
+	}
+	return m
+}
+
+func IsValidSystemRule(rule *SystemRule) bool {
+	if rule == nil || rule.TriggerCount < 0 || rule.MetricType >= MetricTypeSize {
+		return false
+	}
+	if rule.MetricType == CpuUsage && rule.TriggerCount > 1 {
+		return false
+	}
+	return true
+}
diff --git a/core/system/slot.go b/core/system/slot.go
@@ -0,0 +1,76 @@
+package system
+
+import (
+	"github.com/sentinel-group/sentinel-golang/core/base"
+	"github.com/sentinel-group/sentinel-golang/core/stat"
+)
+
+const SlotName = "SystemAdaptiveSlot"
+
+type SystemAdaptiveSlot struct {
+}
+
+func (s *SystemAdaptiveSlot) Check(ctx *base.EntryContext) *base.TokenResult {
+	if ctx == nil || ctx.Resource == nil || ctx.Resource.FlowType() != base.Inbound {
+		return base.NewTokenResultPass()
+	}
+	rules := GetRules()
+	for _, rule := range rules {
+		passed, m := s.doCheckRule(rule)
+		if passed {
+			continue
+		}
+		return base.NewTokenResultBlockedWithCause(base.BlockTypeSystemFlow, base.BlockTypeSystemFlow.String(), rule, m)
+	}
+	return base.NewTokenResultPass()
+}
+
+func (s *SystemAdaptiveSlot) doCheckRule(rule *SystemRule) (bool, float64) {
+	threshold := rule.TriggerCount
+	switch rule.MetricType {
+	case InboundQPS:
+		qps := stat.InboundNode().GetQPS(base.MetricEventPass)
+		res := qps < threshold
+		return res, qps
+	case Concurrency:
+		n := float64(stat.InboundNode().CurrentGoroutineNum())
+		res := n < threshold
+		return res, n
+	case AvgRT:
+		rt := stat.InboundNode().AvgRT()
+		res := rt < threshold
+		return res, rt
+	case Load:
+		l := CurrentLoad()
+		if l > threshold {
+			if rule.Strategy != BBR || !checkBbrSimple() {
+				return false, l
+			}
+		}
+		return true, l
+	case CpuUsage:
+		c := CurrentCpuUsage()
+		if c > threshold {
+			if rule.Strategy != BBR || !checkBbrSimple() {
+				return false, c
+			}
+		}
+		return true, c
+	default:
+		return true, 0
+	}
+}
+
+func checkBbrSimple() bool {
+	concurrency := stat.InboundNode().CurrentGoroutineNum()
+	minRt := stat.InboundNode().MinRT()
+	maxComplete := stat.InboundNode().GetMaxAvg(base.MetricEventComplete)
+	if concurrency > 1 && float64(concurrency) > maxComplete*minRt/1000 {
+		return false
+	}
+	return true
+}
+
+func (s *SystemAdaptiveSlot) String() string {
+	return SlotName
+}
diff --git a/core/system/sys_stat.go b/core/system/sys_stat.go
@@ -0,0 +1,74 @@
+package system
+
+import (
+	"github.com/sentinel-group/sentinel-golang/util"
+	"sync/atomic"
+	"time"
+
+	"github.com/shirou/gopsutil/cpu"
+	"github.com/shirou/gopsutil/load"
+)
+
+const (
+	notRetrievedValue float64 = -1
+)
+
+var (
+	currentLoad     atomic.Value
+	currentCpuUsage atomic.Value
+
+	ssStopChan = make(chan struct{})
+)
+
+func init() {
+	currentLoad.Store(notRetrievedValue)
+	currentCpuUsage.Store(notRetrievedValue)
+
+	ticker := time.NewTicker(1 * time.Second)
+	go util.RunWithRecover(func() {
+		for {
+			select {
+			case <-ticker.C:
+				retrieveAndUpdateSystemStat()
+			case <-ssStopChan:
+				ticker.Stop()
+				return
+			}
+		}
+	}, logger)
+}
+
+func retrieveAndUpdateSystemStat() {
+	cpuStats, err := cpu.Times(false)
+	if err != nil {
+		logger.Warnf("Failed to retrieve current CPU usage: %+v", err)
+	}
+	loadStat, err := load.Avg()
+	if err != nil {
+		logger.Warnf("Failed to retrieve current system load: %+v", err)
+	}
+	if len(cpuStats) > 0 {
+		// TODO: calculate the real CPU usage
+		// cpuStat := cpuStats[0]
+		// currentCpuUsage.Store(cpuStat.User)
+	}
+	if loadStat != nil {
+		currentLoad.Store(loadStat.Load1)
+	}
+}
+
+func CurrentLoad() float64 {
+	r, ok := currentLoad.Load().(float64)
+	if !ok {
+		return notRetrievedValue
+	}
+	return r
+}
+
+func CurrentCpuUsage() float64 {
+	r, ok := currentCpuUsage.Load().(float64)
+	if !ok {
+		return notRetrievedValue
+	}
+	return r
+}
diff --git a/go.mod b/go.mod
@@ -3,6 +3,11 @@ module github.com/sentinel-group/sentinel-golang
 go 1.13
 
 require (
+	github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d // indirect
+	github.com/go-ole/go-ole v1.2.4 // indirect
 	github.com/pkg/errors v0.8.1
+	github.com/shirou/gopsutil v2.19.12+incompatible
 	github.com/stretchr/testify v1.4.0
+	golang.org/x/sys v0.0.0-20200107162124-548cf772de50 // indirect
+	gopkg.in/yaml.v2 v2.2.2
 )
diff --git a/go.sum b/go.sum
@@ -1,13 +1,21 @@
+github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d h1:G0m3OIz70MZUWq3EgK3CesDbo8upS2Vm9/P3FtgI+Jk=
+github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
 github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI=
+github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/shirou/gopsutil v2.19.12+incompatible h1:WRstheAymn1WOPesh+24+bZKFkqrdCR8JOc77v4xV3Q=
+github.com/shirou/gopsutil v2.19.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0 h1:2E4SXV/wtOkTonXsotYi4li6zVWxYlZuYNCXe9XRJyk=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
+golang.org/x/sys v0.0.0-20200107162124-548cf772de50 h1:YvQ10rzcqWXLlJZ3XCUoO25savxmscf4+SC+ZqiCHhA=
+golang.org/x/sys v0.0.0-20200107162124-548cf772de50/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
