Fix the bug of circuit breaker half-open state transformation when request is blocked by upcoming rules (#202)

louyuting · web-flow · commit d12f6116673c · 2020-08-24T22:24:06.000+08:00
* Refactor the workflow to fix the bug that circuit breaker may remain half-open state forever when the request is blocked by upcoming rules: revert the state change in exit handler (as a temporary workaround)
* Add exit handler in Entry as a per-invocation hook.
diff --git a/api/api.go b/api/api.go
@@ -140,6 +140,7 @@ func entry(resource string, options *EntryOptions) (*base.SentinelEntry, *base.B
 	options.Reset()
 	entryOptsPool.Put(options)
 	e := base.NewSentinelEntry(ctx, rw, sc)
+	ctx.SetEntry(e)
 	r := sc.Entry(ctx)
 	if r == nil {
 		// This indicates internal error in some slots, so just pass
diff --git a/core/base/context.go b/core/base/context.go
@@ -3,7 +3,8 @@ package base
 import "github.com/alibaba/sentinel-golang/util"
 
 type EntryContext struct {
-	// internal error when sentinel entry or
+	entry *SentinelEntry
+	// internal error when sentinel Entry or
 	// biz error of downstream
 	err error
 	// Use to calculate RT
@@ -21,6 +22,14 @@ type EntryContext struct {
 	Data map[interface{}]interface{}
 }
 
+func (ctx *EntryContext) SetEntry(entry *SentinelEntry) {
+	ctx.entry = entry
+}
+
+func (ctx *EntryContext) Entry() *SentinelEntry {
+	return ctx.entry
+}
+
 func (ctx *EntryContext) Err() error {
 	return ctx.err
 }
@@ -79,6 +88,7 @@ func (i *SentinelInput) reset() {
 // Reset init EntryContext,
 func (ctx *EntryContext) Reset() {
 	// reset all fields of ctx
+	ctx.entry = nil
 	ctx.err = nil
 	ctx.startTime = 0
 	ctx.rt = 0
diff --git a/core/base/entry.go b/core/base/entry.go
@@ -2,12 +2,18 @@ package base
 
 import (
 	"sync"
+
+	"github.com/alibaba/sentinel-golang/logging"
 )
 
+type ExitHandler func(entry *SentinelEntry, ctx *EntryContext) error
+
 type SentinelEntry struct {
 	res *ResourceWrapper
-	// one entry with one context
+	// one entry bounds with one context
 	ctx *EntryContext
+
+	exitHandlers []ExitHandler
 	// each entry holds a slot chain.
 	// it means this entry will go through the sc
 	sc *SlotChain
@@ -17,12 +23,17 @@ type SentinelEntry struct {
 
 func NewSentinelEntry(ctx *EntryContext, rw *ResourceWrapper, sc *SlotChain) *SentinelEntry {
 	return &SentinelEntry{
-		res: rw,
-		ctx: ctx,
-		sc:  sc,
+		res:          rw,
+		ctx:          ctx,
+		exitHandlers: make([]ExitHandler, 0),
+		sc:           sc,
 	}
 }
 
+func (e *SentinelEntry) WhenExit(exitHandler ExitHandler) {
+	e.exitHandlers = append(e.exitHandlers, exitHandler)
+}
+
 func (e *SentinelEntry) SetError(err error) {
 	if e.ctx != nil {
 		e.ctx.SetError(err)
@@ -56,13 +67,28 @@ func (e *SentinelEntry) Exit(exitOps ...ExitOption) {
 		opt(&options)
 	}
 	ctx := e.ctx
+	if ctx == nil {
+		return
+	}
 	if options.err != nil {
 		ctx.SetError(options.err)
 	}
 	e.exitCtl.Do(func() {
+		defer func() {
+			if err := recover(); err != nil {
+				logging.Panicf("Sentinel internal panic in entry exit func, err: %+v", err)
+			}
+			if e.sc != nil {
+				e.sc.RefurbishContext(ctx)
+			}
+		}()
+		for _, handler := range e.exitHandlers {
+			if err := handler(e, ctx); err != nil {
+				logging.Errorf("Fail to execute exitHandler for resource: %s, err: %+v", e.Resource().Name(), err)
+			}
+		}
 		if e.sc != nil {
 			e.sc.exit(ctx)
-			e.sc.RefurbishContext(ctx)
 		}
 	})
 }
diff --git a/core/base/entry_test.go b/core/base/entry_test.go
@@ -1 +1,27 @@
 package base
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var flag = 0
+
+func exitHandlerMock(entry *SentinelEntry, ctx *EntryContext) error {
+	flag += 1
+	return nil
+}
+
+func TestSentinelEntry_WhenExit(t *testing.T) {
+	flag = 0
+	sc := NewSlotChain()
+	ctx := sc.GetPooledContext()
+	entry := NewSentinelEntry(ctx, nil, sc)
+	entry.WhenExit(exitHandlerMock)
+	entry.Exit()
+	assert.True(t, flag == 1)
+
+	entry.Exit()
+	assert.True(t, flag == 1)
+}
diff --git a/core/base/slot_chain.go b/core/base/slot_chain.go
@@ -40,8 +40,9 @@ type StatSlot interface {
 	// StatSlots will do some statistic logic, such as QPS、log、etc
 	// blockError introduce the block detail
 	OnEntryBlocked(ctx *EntryContext, blockError *BlockError)
-	// onComplete function will be invoked when chain exits.
-	// The request may be executed successful or blocked or internal error
+	// OnCompleted function will be invoked when chain exits.
+	// The semantics of OnCompleted is the entry passed and completed
+	// Note: blocked entry will not call this function
 	OnCompleted(ctx *EntryContext)
 }
 
diff --git a/core/base/slot_chain_test.go b/core/base/slot_chain_test.go
@@ -220,6 +220,7 @@ func TestSlotChain_Entry_Pass_And_Exit(t *testing.T) {
 	ctx := sc.GetPooledContext()
 	rw := NewResourceWrapper("abc", ResTypeCommon, Inbound)
 	ctx.Resource = rw
+	ctx.SetEntry(NewSentinelEntry(ctx, rw, sc))
 	ctx.StatNode = &StatNodeMock{}
 	ctx.Input = &SentinelInput{
 		AcquireCount: 1,
@@ -261,6 +262,7 @@ func TestSlotChain_Entry_Block(t *testing.T) {
 	sc := NewSlotChain()
 	ctx := sc.GetPooledContext()
 	rw := NewResourceWrapper("abc", ResTypeCommon, Inbound)
+	ctx.SetEntry(NewSentinelEntry(ctx, rw, sc))
 	ctx.Resource = rw
 	ctx.StatNode = &StatNodeMock{}
 	ctx.Input = &SentinelInput{
diff --git a/core/circuitbreaker/circuit_breaker.go b/core/circuitbreaker/circuit_breaker.go
@@ -89,6 +89,7 @@ type CircuitBreaker interface {
 	CurrentState() State
 	// OnRequestComplete record a completed request with the given response time as well as error (if present),
 	// and handle state transformation of the circuit breaker.
+	// OnRequestComplete is called only when a passed invocation finished.
 	OnRequestComplete(rtt uint64, err error)
 }
 
@@ -137,11 +138,28 @@ func (b *circuitBreakerBase) fromClosedToOpen(snapshot interface{}) bool {
 
 // fromOpenToHalfOpen updates circuit breaker state machine from open to half-open.
 // Return true only if current goroutine successfully accomplished the transformation.
-func (b *circuitBreakerBase) fromOpenToHalfOpen() bool {
+func (b *circuitBreakerBase) fromOpenToHalfOpen(ctx *base.EntryContext) bool {
 	if b.state.casState(Open, HalfOpen) {
 		for _, listener := range stateChangeListeners {
 			listener.OnTransformToHalfOpen(Open, b.rule)
 		}
+
+		entry := ctx.Entry()
+		if entry == nil {
+			logging.Errorf("nil entry when probing, rule: %+v", b.rule)
+		} else {
+			// add hook for entry exit
+			// if the current circuit breaker performs the probe through this entry, but the entry was blocked,
+			// this hook will guarantee current circuit breaker state machine will rollback to Open from Half-Open
+			entry.WhenExit(func(entry *base.SentinelEntry, ctx *base.EntryContext) error {
+				if ctx.IsBlocked() && b.state.casState(HalfOpen, Open) {
+					for _, listener := range stateChangeListeners {
+						listener.OnTransformToOpen(HalfOpen, b.rule, 1.0)
+					}
+				}
+				return nil
+			})
+		}
 		return true
 	}
 	return false
@@ -215,13 +233,13 @@ func (b *slowRtCircuitBreaker) BoundStat() interface{} {
 }
 
 // TryPass checks circuit breaker based on state machine of circuit breaker.
-func (b *slowRtCircuitBreaker) TryPass(_ *base.EntryContext) bool {
+func (b *slowRtCircuitBreaker) TryPass(ctx *base.EntryContext) bool {
 	curStatus := b.CurrentState()
 	if curStatus == Closed {
 		return true
 	} else if curStatus == Open {
 		// switch state to half-open to probe if retry timeout
-		if b.retryTimeoutArrived() && b.fromOpenToHalfOpen() {
+		if b.retryTimeoutArrived() && b.fromOpenToHalfOpen(ctx) {
 			return true
 		}
 	}
@@ -399,13 +417,13 @@ func (b *errorRatioCircuitBreaker) BoundStat() interface{} {
 	return b.stat
 }
 
-func (b *errorRatioCircuitBreaker) TryPass(_ *base.EntryContext) bool {
+func (b *errorRatioCircuitBreaker) TryPass(ctx *base.EntryContext) bool {
 	curStatus := b.CurrentState()
 	if curStatus == Closed {
 		return true
 	} else if curStatus == Open {
 		// switch state to half-open to probe if retry timeout
-		if b.retryTimeoutArrived() && b.fromOpenToHalfOpen() {
+		if b.retryTimeoutArrived() && b.fromOpenToHalfOpen(ctx) {
 			return true
 		}
 	}
@@ -579,13 +597,13 @@ func (b *errorCountCircuitBreaker) BoundStat() interface{} {
 	return b.stat
 }
 
-func (b *errorCountCircuitBreaker) TryPass(_ *base.EntryContext) bool {
+func (b *errorCountCircuitBreaker) TryPass(ctx *base.EntryContext) bool {
 	curStatus := b.CurrentState()
 	if curStatus == Closed {
 		return true
 	} else if curStatus == Open {
 		// switch state to half-open to probe if retry timeout
-		if b.retryTimeoutArrived() && b.fromOpenToHalfOpen() {
+		if b.retryTimeoutArrived() && b.fromOpenToHalfOpen(ctx) {
 			return true
 		}
 	}
diff --git a/core/circuitbreaker/rule.go b/core/circuitbreaker/rule.go
@@ -79,7 +79,7 @@ func (b *RuleBase) IsApplicable() error {
 	if b.RetryTimeoutMs <= 0 {
 		return errors.New("invalid RetryTimeoutMs")
 	}
-	if b.MinRequestAmount <= 0 {
+	if b.MinRequestAmount < 0 {
 		return errors.New("invalid MinRequestAmount")
 	}
 	if b.StatIntervalMs <= 0 {
diff --git a/core/circuitbreaker/rule_manager.go b/core/circuitbreaker/rule_manager.go
@@ -274,16 +274,21 @@ func logRuleUpdate(rules map[string][]Rule) {
 	logging.Info(sb.String())
 }
 
+// Note: this function is not thread-safe.
 func RegisterStateChangeListeners(listeners ...StateChangeListener) {
 	if len(listeners) == 0 {
 		return
 	}
-	updateMux.Lock()
-	defer updateMux.Unlock()
 
 	stateChangeListeners = append(stateChangeListeners, listeners...)
 }
 
+// ClearStateChangeListeners will clear the all StateChangeListener
+// Note: this function is not thread-safe.
+func ClearStateChangeListeners() {
+	stateChangeListeners = make([]StateChangeListener, 0)
+}
+
 // SetCircuitBreakerGenerator sets the circuit breaker generator for the given strategy.
 // Note that modifying the generator of default strategies is not allowed.
 func SetCircuitBreakerGenerator(s Strategy, generator CircuitBreakerGenFunc) error {
diff --git a/tests/core/circuitbreaker/circuitbreaker_slot_integration_test.go b/tests/core/circuitbreaker/circuitbreaker_slot_integration_test.go

Original file line number	Diff line number	Diff line change
`@@ -79,7 +79,7 @@ func (b *RuleBase) IsApplicable() error {`
`79`	`79`	`if b.RetryTimeoutMs <= 0 {`
`80`	`80`	`return errors.New("invalid RetryTimeoutMs")`
`81`	`81`	`}`
`82`		`- if b.MinRequestAmount <= 0 {`
	`82`	`+ if b.MinRequestAmount < 0 {`
`83`	`83`	`return errors.New("invalid MinRequestAmount")`
`84`	`84`	`}`
`85`	`85`	`if b.StatIntervalMs <= 0 {`