Add initial work of the core pipeline (slot chain) (alibaba#3)

Author: louyuting

README.md

@@ -7,4 +7,4 @@
 
 ## Contributing
 
-Contributions are always welcomed! Please see [CONTRIBUTING](./CONTRIBUTING.md) for detailed guidelines.
+Contributions are always welcomed! Please see [CONTRIBUTING](./CONTRIBUTING.md) for detailed guidelines.

core/context.go

@@ -0,0 +1,55 @@
+package core
+
+type Context struct {
+	ResWrapper  *ResourceWrapper
+	Entry       *CtxEntry
+	Node        node
+	Count       uint64
+	Input       *SentinelInput
+	Output      *SentinelOutput
+	FeatureData map[interface{}]interface{}
+}
+
+func NewContext() *Context {
+	ctx := &Context{
+		Input:  newInput(),
+		Output: newOutput(),
+	}
+	ctx.Input.Context = ctx
+	ctx.Output.Context = ctx
+	return ctx
+}
+
+type SentinelInput struct {
+	Context *Context
+	// store some values in this context when calling context in slot.
+	data map[interface{}]interface{}
+}
+
+func newInput() *SentinelInput {
+	return &SentinelInput{}
+}
+
+type SentinelOutput struct {
+	Context     *Context
+	CheckResult *RuleCheckResult
+	msg         string
+	// store output data.
+	data map[interface{}]interface{}
+}
+
+func newOutput() *SentinelOutput {
+	return &SentinelOutput{}
+}
+
+// Reset init Context,
+func (ctx *Context) Reset() {
+	// reset all fields of ctx
+	ctx.ResWrapper = nil
+	ctx.Entry = nil
+	ctx.Node = nil
+	ctx.Count = 0
+	ctx.Input = nil
+	ctx.Output = nil
+	ctx.FeatureData = nil
+}

core/context_test.go

@@ -0,0 +1 @@
+package core

core/entry.go

@@ -0,0 +1,59 @@
+package core
+
+// TrafficType describe the traffic type: Inbound or OutBound
+type TrafficType int32
+
+const (
+	InBound TrafficType = iota
+	OutBound
+)
+
+type ResourceWrapper struct {
+	// global unique resource name
+	ResourceName string
+	// InBound or OutBound
+	FlowType TrafficType
+}
+
+// CtxEntry means Context entry,
+type CtxEntry struct {
+	createTime uint64
+	rs         *ResourceWrapper
+	// one entry with one context
+	ctx *Context
+	// each entry holds a slot chain.
+	// it means this entry will go through the sc
+	sc *SlotChain
+	// caller node
+	originNode node
+	// current resource node
+	currentNode node
+}
+
+func NewCtEntry(ctx *Context, rw *ResourceWrapper, sc *SlotChain, cn node) *CtxEntry {
+	return &CtxEntry{
+		createTime:  GetTimeMilli(),
+		rs:          rw,
+		ctx:         ctx,
+		sc:          sc,
+		currentNode: cn,
+	}
+}
+
+func (e *CtxEntry) Exit() {
+	e.ExitWithCnt(1)
+}
+
+func (e *CtxEntry) ExitWithCnt(count int32) {
+	e.exitForContext(e.ctx, count)
+}
+
+func (e *CtxEntry) exitForContext(ctx *Context, count int32) {
+	if e.sc != nil {
+		e.sc.exit(ctx)
+	}
+}
+
+func (e *CtxEntry) GetCurrentNode() node {
+	return e.currentNode
+}

core/entry_test.go

@@ -0,0 +1 @@
+package core

core/global_variable.go

@@ -0,0 +1,8 @@
+package core
+
+const (
+	TotalInBoundResourceName        = "__total_inbound_traffic__"
+	WindowLengthImMs         uint32 = 200
+	SampleCount              uint32 = 5
+	IntervalInMs             uint32 = 1000
+)

core/logging.go

@@ -0,0 +1,34 @@
+package core
+
+import (
+	"fmt"
+	"log"
+	"os"
+)
+
+func InitDefaultLoggerToConsole() {
+	fmt.Println("Init default log, output to console")
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
+	log.SetPrefix("[sentinel]")
+}
+
+// outputFile is the full path(absolute path)
+func NewFileLogger(outputFile, prefix string, flag int) *log.Logger {
+	//get file info
+	var logFile *os.File
+	_, err := os.Stat(outputFile)
+	if err == nil {
+		logFile, err = os.Open(outputFile)
+		if err != nil {
+			log.Fatal("open log file error, ", err)
+		}
+	} else if err != nil && os.IsNotExist(err) {
+		logFile, err = os.Create(outputFile)
+		if err != nil {
+			log.Fatal("create log file error, ", err)
+		}
+	} else {
+		log.Fatal("open log file error, ", err)
+	}
+	return log.New(logFile, prefix, flag)
+}

core/node.go

@@ -0,0 +1,28 @@
+package core
+
+type node interface {
+	// total  = pass + blocked
+	TotalCountInMinute() uint64
+	PassCountInMinute() uint64
+	BlockCountInMinute() uint64
+	CompleteCountInMinute() uint64
+	ExceptionCountInMinute() uint64
+
+	TotalQps() uint64
+	PassQps() uint64
+	BlockQps() uint64
+	CompleteQps() uint64
+	ExceptionQps() uint64
+
+	AvgRt() uint64
+	CurrentGoroutineNum() uint64
+
+	AddPassRequest(count uint64)
+	AddRtAndCompleteRequest(rt, count uint64)
+	AddBlockRequest(count uint64)
+	AddExceptionRequest(count uint64)
+	IncreaseGoroutineNum()
+	DecreaseGoroutineNum()
+
+	Reset()
+}

core/node_test.go

@@ -0,0 +1,88 @@
+package core
+
+import "github.com/stretchr/testify/mock"
+
+type NodeMock struct {
+	mock.Mock
+}
+
+func (m *NodeMock) TotalCountInMinute() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) PassCountInMinute() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) BlockCountInMinute() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) CompleteCountInMinute() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) ExceptionCountInMinute() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+
+func (m *NodeMock) TotalQps() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) PassQps() uint64 {
+	args := m.Called()
+	return uint64(*args.Get(0).(*int))
+}
+func (m *NodeMock) BlockQps() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) CompleteQps() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) ExceptionQps() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) AvgRt() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+func (m *NodeMock) CurrentGoroutineNum() uint64 {
+	args := m.Called()
+	return uint64(args.Int(0))
+}
+
+func (m *NodeMock) AddPassRequest(count uint64) {
+	m.Called(count)
+	return
+}
+func (m *NodeMock) AddRtAndCompleteRequest(rt, count uint64) {
+	m.Called(rt, count)
+	return
+}
+func (m *NodeMock) AddBlockRequest(count uint64) {
+	m.Called(count)
+	return
+}
+func (m *NodeMock) AddExceptionRequest(count uint64) {
+	m.Called(count)
+	return
+}
+
+func (m *NodeMock) IncreaseGoroutineNum() {
+	m.Called()
+	return
+}
+func (m *NodeMock) DecreaseGoroutineNum() {
+	m.Called()
+	return
+}
+
+func (m *NodeMock) Reset() {
+	m.Called()
+	return
+}

core/result.go

@@ -0,0 +1,50 @@
+package core
+
+import (
+	"fmt"
+)
+
+type RuleBasedCheckBlockedEvent int8
+
+const (
+	UnknownEvent RuleBasedCheckBlockedEvent = iota
+)
+
+type SlotResultStatus int8
+
+const (
+	ResultStatusPass SlotResultStatus = iota
+	ResultStatusBlocked
+)
+
+type RuleCheckResult struct {
+	Status       SlotResultStatus
+	BlockedEvent RuleBasedCheckBlockedEvent
+	BlockedMsg   string
+}
+
+func (r *RuleCheckResult) status() string {
+	if r.Status == ResultStatusPass {
+		return "ResultStatusPass"
+	} else if r.Status == ResultStatusBlocked {
+		return "ResultStatusBlocked"
+	} else {
+		return "Unknown"
+	}
+}
+
+func (r *RuleCheckResult) toString() string {
+	return fmt.Sprintf("check result:%s; BlockedEvent is:%v; BlockedMsg is:%s;", r.status(), r.BlockedEvent, r.BlockedMsg)
+}
+
+func NewSlotResultPass() *RuleCheckResult {
+	return &RuleCheckResult{Status: ResultStatusPass}
+}
+
+func NewSlotResultBlocked(blockEvent RuleBasedCheckBlockedEvent, blockReason string) *RuleCheckResult {
+	return &RuleCheckResult{
+		Status:       ResultStatusBlocked,
+		BlockedEvent: blockEvent,
+		BlockedMsg:   blockReason,
+	}
+}

core/result_test.go

@@ -0,0 +1,8 @@
+package core
+
+import "testing"
+
+func TestNewSlotResultBlock_normal(t *testing.T) {
+	r := NewSlotResultBlocked(UnknownEvent, "UnknownEvent")
+	t.Log(r.toString())
+}