Repository navigation
Expand file tree
/
Copy pathgobreaker.go
More file actions
392 lines (340 loc) · 10.1 KB
/
Copy pathgobreaker.go
File metadata and controls
392 lines (340 loc) · 10.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
// Package gobreaker implements the Circuit Breaker pattern.
// See https://msdn.microsoft.com/en-us/library/dn589784.aspx.
package gobreaker
import (
"errors"
"fmt"
"sync"
"time"
)
// State is a type that represents a state of CircuitBreaker.
type State int
// These constants are states of CircuitBreaker.
const (
StateClosed State = iota
StateHalfOpen
StateOpen
)
var (
// ErrTooManyRequests is returned when the CB state is half open and the requests count is over the cb maxRequests
ErrTooManyRequests = errors.New("too many requests")
// ErrOpenState is returned when the CB state is open
ErrOpenState = errors.New("circuit breaker is open")
)
// String implements stringer interface.
func (s State) String() string {
switch s {
case StateClosed:
return "closed"
case StateHalfOpen:
return "half-open"
case StateOpen:
return "open"
default:
return fmt.Sprintf("unknown state: %d", s)
}
}
// Settings configures CircuitBreaker:
//
// Name is the name of the CircuitBreaker.
//
// MaxRequests is the maximum number of requests allowed to pass through
// when the CircuitBreaker is half-open.
// If MaxRequests is 0, the CircuitBreaker allows only 1 request.
//
// Interval is the cyclic period of the closed state
// for the CircuitBreaker to clear the internal Counts.
// If Interval is less than or equal to 0, the CircuitBreaker does not clear internal Counts during the closed state.
//
// BucketPeriod defines the time duration for each bucket in the rolling window strategy.
// The internal Counts will be updated and reset gradually for each bucket.
// Interval will be automatically adjusted to be a multiple of BucketPeriod.
// If BucketPeriod is less than or equal to 0, the CircuitBreaker will use a fixed window strategy instead.
//
// Timeout is the period of the open state,
// after which the state of the CircuitBreaker becomes half-open.
// If Timeout is less than or equal to 0, the timeout value of the CircuitBreaker is set to 60 seconds.
//
// ReadyToTrip is called with a copy of Counts whenever a request fails in the closed state.
// If ReadyToTrip returns true, the CircuitBreaker will be placed into the open state.
// If ReadyToTrip is nil, default ReadyToTrip is used.
// Default ReadyToTrip returns true when the number of consecutive failures is more than 5.
//
// OnStateChange is called whenever the state of the CircuitBreaker changes.
8000
//
// IsSuccessful is called with the error returned from a request.
// If IsSuccessful returns true, the error is counted as a success.
// Otherwise the error is counted as a failure.
// If IsSuccessful is nil, default IsSuccessful is used, which returns false for all non-nil errors.
//
// IsExcluded determines whether a request error should be ignored
// for the purposes of updating the circuit breaker metrics.
// If IsExcluded returns true for a given error,
// the request is neither counted as a success nor as a failure.
// This can be used, for example, to ignore context cancellations or
// other errors that should not affect the circuit breaker state.
// If IsExcluded is nil, no requests are excluded.
type Settings struct {
Name string
MaxRequests uint32
Interval time.Duration
BucketPeriod time.Duration
Timeout time.Duration
// MaxTimeout is the maximum timeout used for open-state recovery delay.
// If MaxTimeout is 0, open-state timeout is fixed at Timeout.
// If MaxTimeout is greater than 0, each failed half-open probe doubles Timeout
// up to MaxTimeout.
MaxTimeout time.Duration
ReadyToTrip func(counts Counts) bool
OnStateChange func(name string, from State, to State)
IsSuccessful func(err error) bool
IsExcluded func(err error) bool
}
// CircuitBreaker is a state machine to prevent sending requests that are likely to fail.
type CircuitBreaker[T any] struct {
name string
maxRequests uint32
interval time.Duration
bucketPeriod time.Duration
timeout time.Duration
baseTimeout time.Duration
maxTimeout time.Duration
readyToTrip func(counts Counts) bool
isSuccessful func(err error) bool
isExcluded func(err error) bool
onStateChange func(name string, from State, to State)
mutex sync.Mutex
state State
generation uint64
counts *rollingCounts
start time.Time
expiry time.Time
}
// NewCircuitBreaker returns a new CircuitBreaker configured with the given Settings.
func NewCircuitBreaker[T any](st Settings) *CircuitBreaker[T] {
cb := new(CircuitBreaker[T])
cb.name = st.Name
cb.>
if st.MaxRequests == 0 {
cb.maxRequests = 1
} else {
cb.maxRequests = st.MaxRequests
}
var numBuckets int64
if st.Interval <= 0 {
cb.interval = defaultInterval
cb.bucketPeriod = cb.interval
numBuckets = 1
} else if st.BucketPeriod <= 0 {
cb.interval = st.Interval
cb.bucketPeriod = cb.interval
numBuckets = 1
} else {
cb.interval = (st.Interval + st.BucketPeriod - 1) / st.BucketPeriod * st.BucketPeriod
cb.bucketPeriod = st.BucketPeriod
numBuckets = int64(cb.interval / cb.bucketPeriod)
}
cb.counts = newRollingCounts(numBuckets)
if st.Timeout <= 0 {
cb.timeout = defaultTimeout
} else {
cb.timeout = st.Timeout
}
cb.baseTimeout = cb.timeout
cb.maxTimeout = 0
if st.MaxTimeout > 0 {
if st.MaxTimeout < cb.baseTimeout {
cb.maxTimeout = cb.baseTimeout
} else {
cb.maxTimeout = st.MaxTimeout
}
}
if st.ReadyToTrip == nil {
cb.readyToTrip = defaultReadyToTrip
} else {
cb.readyToTrip = st.ReadyToTrip
}
if st.IsSuccessful == nil {
cb.isSuccessful = defaultIsSuccessful
} else {
cb.isSuccessful = st.IsSuccessful
}
if st.IsExcluded == nil {
cb.isExcluded = defaultIsExcluded
} else {
cb.isExcluded = st.IsExcluded
}
cb.toNewGeneration(time.Now())
return cb
}
const defaultInterval = time.Duration(0) * time.Second
const defaultTimeout = time.Duration(60) * time.Second
func defaultReadyToTrip(counts Counts) bool {
return counts.ConsecutiveFailures > 5
}
func defaultIsSuccessful(err error) bool {
return err == nil
}
func defaultIsExcluded(_ error) bool {
return false
}
// Name returns the name of the CircuitBreaker.
func (cb *CircuitBreaker[T]) Name() string {
return cb.name
}
// State returns the current state of the CircuitBreaker.
func (cb *CircuitBreaker[T]) State() State {
cb.mutex.Lock()
defer cb.mutex.Unlock()
now := time.Now()
state, _, _ := cb.currentState(now)
return state
}
// Counts returns internal counters
func (cb *CircuitBreaker[T]) Counts() Counts {
cb.mutex.Lock()
defer cb.mutex.Unlock()
return cb.counts.Counts
}
// Execute runs the given request if the CircuitBreaker accepts it.
// Execute returns an error instantly if the CircuitBreaker rejects the request.
// Otherwise, Execute returns the result of the request.
// If a panic occurs in the request, the CircuitBreaker handles it as an error
// and causes the same panic again.
func (cb *CircuitBreaker[T]) Execute(req func() (T, error)) (T, error) {
generation, age, err := cb.beforeRequest()
if err != nil {
var defaultValue T
return defaultValue, err
}
defer func() {
e := recover()
if e != nil {
cb.afterRequest(generation, age, fmt.Errorf("%v", e))
panic(e)
}
}()
result, err := req()
cb.afterRequest(generation, age, err)
return result, err
}
func (cb *CircuitBreaker[T]) beforeRequest() (uint64, uint64, error) {
cb.mutex.Lock()
defer cb.mutex.Unlock()
now := time.Now()
state, generation, age := cb.currentState(now)
if state == StateOpen {
return generation, age, ErrOpenState
} else if state == StateHalfOpen && cb.counts.validRequests() >= cb.maxRequests {
return generation, age, ErrTooManyRequests
}
cb.counts.onRequest()
return generation, age, nil
}
func (cb *CircuitBreaker[T]) afterRequest(previous uint64, age uint64, err error) {
cb.mutex.Lock()
defer cb.mutex.Unlock()
now := time.Now()
state, generation, _ := cb.currentState(now)
if generation != previous {
return
}
if cb.isExcluded(err) {
cb.onExclusion(age)
return
}
if cb.isSuccessful(err) {
cb.onSuccess(state, age, now)
} else {
cb.onFailure(state, age, now)
}
}
func (cb *CircuitBreaker[T]) onSuccess(state State, age uint64, now time.Time) {
switch state {
case StateClosed:
cb.counts.onSuccess(age)
case StateHalfOpen:
cb.counts.onSuccess(age)
if cb.counts.ConsecutiveSuccesses >= cb.maxRequests {
cb.setState(StateClosed, now)
}
}
}
func (cb *CircuitBreaker[T]) onFailure(state State, age uint64, now time.Time) {
switch state {
case StateClosed:
cb.counts.onFailure(age)
if cb.readyToTrip(cb.counts.Counts) {
cb.setState(StateOpen, now)
}
case StateHalfOpen:
cb.setState(StateOpen, now)
}
}
func (cb *CircuitBreaker[T]) onExclusion(age uint64) {
cb.counts.onExclusion(age)
}
func (cb *CircuitBreaker[T]) currentState(now time.Time) (State, uint64, uint64) {
switch cb.state {
case StateClosed:
if !cb.expiry.IsZero() && cb.expiry.Before(now) {
cb.toNewGeneration(now)
} else if len(cb.counts.buckets) >= 2 {
cb.counts.grow(cb.age(now))
}
case StateOpen:
if cb.expiry.Before(now) {
cb.setState(StateHalfOpen, now)
}
}
return cb.state, cb.generation, cb.counts.age
}
func (cb *CircuitBreaker[T]) age(now time.Time) uint64 {
if cb.bucketPeriod == 0 {
return 0
}
elapsed := now.Sub(cb.start)
age := int64(elapsed / cb.bucketPeriod)
if age < 0 {
return 0
}
return uint64(age)
}
func (cb *CircuitBreaker[T]) setState(state State, now time.Time) {
if cb.state == state {
return
}
prev := cb.state
if cb.maxTimeout > 0 && prev == StateHalfOpen && state == StateOpen {
cb.timeout *= 2
if cb.timeout > cb.maxTimeout {
cb.timeout = cb.maxTimeout
}
}
if cb.maxTimeout > 0 && prev == StateHalfOpen && state == StateClosed {
cb.timeout = cb.baseTimeout
}
cb.state = state
cb.toNewGeneration(now)
if cb.onStateChange != nil {
cb.onStateChange(cb.name, prev, state)
}
}
func (cb *CircuitBreaker[T]) toNewGeneration(now time.Time) {
cb.generation++
cb.start = now
cb.counts.clear()
var zero time.Time
switch cb.state {
case StateClosed:
if cb.interval == 0 || len(cb.counts.buckets) >= 2 {
cb.expiry = zero
} else {
cb.expiry = now.Add(cb.interval)
}
case StateOpen:
cb.expiry = now.Add(cb.timeout)
default: // StateHalfOpen
cb.expiry = zero
}
}