hanzoai/oxy
2
0
Fork 0
mirror of https://github.com/hanzoai/oxy.git synced 2026-09-30 23:42:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

Recover more tests

Browse Source
This commit is contained in:
Alexander Klizhentas
2014-12-11 21:17:54 -08:00
parent b1b2642a12
commit f393f65290
5 changed files with 163 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
circuitbreaker/cbreaker.go
+46 -27
View File
@@ -68,7 +68,8 @@ type CircuitBreaker struct {
// New creates a new CircuitBreaker middleware
func New(next http.Handler, expression string, options ...optSetter) (*CircuitBreaker, error) {
cb := &CircuitBreaker{
m: &sync.RWMutex{},
m: &sync.RWMutex{},
next: next,
}
for _, s := range options {
@@ -94,56 +95,53 @@ func New(next http.Handler, expression string, options ...optSetter) (*CircuitBr
return cb, nil
}
// String returns log-friendly representation of the circuit breaker state
func (c *CircuitBreaker) String() string {
switch c.state {
case stateTripped, stateRecovering:
return fmt.Sprintf("CircuitBreaker(state=%v, until=%v)", c.state, c.until)
default:
return fmt.Sprintf("CircuitBreaker(state=%v)", c.state)
func (c *CircuitBreaker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if c.activateFallback(w, req) {
c.fallback.ServeHTTP(w, req)
return
}
c.serve(w, req)
}
func (c *CircuitBreaker) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// standby mode simply records the metrics and periodically checks the tripping condition
// updateState updates internal state and returns true if fallback should be used and false otherwise
func (c *CircuitBreaker) activateFallback(w http.ResponseWriter, req *http.Request) bool {
// Quick check with read locks optimized for normal operation use-case
if c.isStandby() {
c.serveRequest(w, req)
return false
}
// Circuit breaker is in tripped or recovering state
c.m.Lock()
defer c.m.Unlock()
c.log.Infof("%v is in error handling state", c)
c.log.Infof("%v is in error state", c)
switch c.state {
case stateStandby:
// other goroutine has set it to standby state
return
// someone else has set it to standby just now
return false
case stateTripped:
if c.clock.UtcNow().Before(c.until) {
c.fallback.ServeHTTP(w, req)
return
return true
}
// We have been in active state enough, enter recovering state
c.setRecovering()
fallthrough
case stateRecovering:
// We have been in recovering state enough, enter standby
// We have been in recovering state enough, enter standby and allow request
if c.clock.UtcNow().After(c.until) {
c.serveRequest(w, req)
c.setState(stateStandby, c.clock.UtcNow())
return
return false
}
// ratio controller allows this request
if c.rc.allowRequest() {
c.serveRequest(w, req)
return
return false
}
c.fallback.ServeHTTP(w, req)
return true
}
return false
}
func (c *CircuitBreaker) serveRequest(w http.ResponseWriter, req *http.Request) {
func (c *CircuitBreaker) serve(w http.ResponseWriter, req *http.Request) {
start := c.clock.UtcNow()
p := &utils.ProxyWriter{W: w}
@@ -163,6 +161,16 @@ func (c *CircuitBreaker) isStandby() bool {
return c.state == stateStandby
}
// String returns log-friendly representation of the circuit breaker state
func (c *CircuitBreaker) String() string {
switch c.state {
case stateTripped, stateRecovering:
return fmt.Sprintf("CircuitBreaker(state=%v, until=%v)", c.state, c.until)
default:
return fmt.Sprintf("CircuitBreaker(state=%v)", c.state)
}
}
// exec executes side effect
func (c *CircuitBreaker) exec(s SideEffect) {
if s == nil {
@@ -208,12 +216,12 @@ func (c *CircuitBreaker) checkAndSet() {
}
c.lastCheck = c.clock.UtcNow().Add(c.checkPeriod)
if !c.condition(c) {
if c.state == stateTripped {
c.log.Infof("%v skip set tripped", c)
return
}
if c.state == stateTripped {
c.log.Infof("%v skip set tripped", c)
if !c.condition(c) {
return
}
@@ -246,6 +254,10 @@ func setDefaults(cb *CircuitBreaker) {
if cb.fallback == nil {
cb.fallback = defaultFallback
}
if cb.log == nil {
cb.log = utils.NullLogger
}
}
type optSetter func(*CircuitBreaker) error
@@ -299,6 +311,13 @@ func Fallback(h http.Handler) optSetter {
}
}
func Logger(l utils.Logger) optSetter {
return func(c *CircuitBreaker) error {
c.log = l
return nil
}
}
// cbState is the state of the circuit breaker
type cbState int
circuitbreaker/cbreaker_test.go
+98 -113
View File
@@ -1,126 +1,108 @@
package circuitbreaker
import (
"fmt"
"io/ioutil"
// "fmt"
"net/http"
"net/url"
"net/http/httptest"
"testing"
"time"
"github.com/mailgun/oxy/memmetrics"
"github.com/mailgun/oxy/testutils"
"github.com/mailgun/timetools"
"github.com/mailgun/vulcan/errors"
"github.com/mailgun/vulcan/metrics"
"github.com/mailgun/vulcan/middleware"
"github.com/mailgun/vulcan/netutils"
"github.com/mailgun/vulcan/request"
"github.com/mailgun/vulcan/testutils"
"github.com/mailgun/vulcan/threshold"
. "gopkg.in/check.v1"
)
func TestCircuitBreaker(t *testing.T) { TestingT(t) }
type CBSuite struct {
tm *timetools.FreezedTime
clock *timetools.FreezedTime
}
var _ = Suite(&CBSuite{
tm: &timetools.FreezedTime{
clock: &timetools.FreezedTime{
CurrentTime: time.Date(2012, 3, 4, 5, 6, 7, 0, time.UTC),
},
})
var triggerNetRatio threshold.Predicate
var fallbackResponse middleware.Middleware
var fallbackRedirect middleware.Middleware
const triggerNetRatio = `NetworkErrorRatio() > 0.5`
var fallbackResponse http.Handler
var fallbackRedirect http.Handler
func (s CBSuite) SetUpSuite(c *C) {
triggerNetRatio = MustParseExpression(`NetworkErrorRatio() > 0.5`)
f, err := NewResponseFallback(Response{StatusCode: 400, Body: []byte("Come back later")})
c.Assert(err, IsNil)
fallbackResponse = f
r, err := NewRedirectFallback(Redirect{URL: "http://localhost:5000"})
rdr, err := NewRedirectFallback(Redirect{URL: "http://localhost:5000"})
c.Assert(err, IsNil)
fallbackRedirect = r
fallbackRedirect = rdr
}
func (s *CBSuite) advanceTime(d time.Duration) {
s.tm.CurrentTime = s.tm.CurrentTime.Add(d)
}
func (s *CBSuite) new(c *C, condition threshold.Predicate, fallback middleware.Middleware, o Options) *CircuitBreaker {
o.TimeProvider = s.tm
cb, err := New(condition, fallback, o)
c.Assert(err, IsNil)
return cb
s.clock.CurrentTime = s.clock.CurrentTime.Add(d)
}
func (s *CBSuite) TestStandbyCycle(c *C) {
cb := s.new(c, triggerNetRatio, fallbackResponse, Options{})
handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.Write([]byte("hello"))
})
// Nothing happened
req := makeRequest(O{})
re, err := cb.ProcessRequest(req)
c.Assert(re, IsNil)
cb, err := New(handler, triggerNetRatio)
c.Assert(err, IsNil)
cb.ProcessResponse(req, req.Attempts[0])
c.Assert(cb.state, Equals, cbState(stateStandby))
srv := httptest.NewServer(cb)
defer srv.Close()
re, err = cb.ProcessRequest(req)
c.Assert(re, IsNil)
re, body, err := testutils.Get(srv.URL)
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusOK)
c.Assert(string(body), Equals, "hello")
}
func (s *CBSuite) TestFullCycle(c *C) {
cb := s.new(c, triggerNetRatio, fallbackResponse,
Options{
FallbackDuration: 10 * time.Second,
RecoveryDuration: 10 * time.Second,
})
handler := http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
w.Write([]byte("hello"))
})
req := makeRequest(O{})
re, err := cb.ProcessRequest(req)
c.Assert(re, IsNil)
cb, err := New(handler, triggerNetRatio, Clock(s.clock))
c.Assert(err, IsNil)
srv := httptest.NewServer(cb)
defer srv.Close()
re, _, err := testutils.Get(srv.URL)
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusOK)
cb.metrics = statsNetErrors(0.6)
cb.ProcessResponse(req, req.Attempts[0])
s.advanceTime(defaultCheckPeriod + time.Millisecond)
re, _, err = testutils.Get(srv.URL)
c.Assert(err, IsNil)
c.Assert(cb.state, Equals, cbState(stateTripped))
re, err = cb.ProcessRequest(req)
c.Assert(re, NotNil)
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusBadRequest)
// Some time has passed, but we are still in triggered state.
// Some time has passed, but we are still in trpped state.
s.advanceTime(9 * time.Second)
re, err = cb.ProcessRequest(req)
c.Assert(re, NotNil)
re, _, err = testutils.Get(srv.URL)
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusBadRequest)
c.Assert(re.StatusCode, Equals, http.StatusServiceUnavailable)
c.Assert(cb.state, Equals, cbState(stateTripped))
// We should be in recovering state by now
okReq := makeRequest(O{})
s.advanceTime(time.Second*1 + time.Millisecond)
re, err = cb.ProcessRequest(okReq)
c.Assert(re, NotNil)
re, _, err = testutils.Get(srv.URL)
c.Assert(err, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusBadRequest)
c.Assert(re.StatusCode, Equals, http.StatusServiceUnavailable)
c.Assert(cb.state, Equals, cbState(stateRecovering))
cb.ProcessResponse(okReq, okReq.Attempts[0])
// 5 seconds after we should be allowing some requests to pass
s.advanceTime(5 * time.Second)
allowed := 0
for i := 0; i < 100; i++ {
re, err = cb.ProcessRequest(okReq)
if re == nil && err == nil {
re, _, err = testutils.Get(srv.URL)
if re.StatusCode == http.StatusOK && err == nil {
allowed++
}
}
@@ -128,12 +110,13 @@ func (s *CBSuite) TestFullCycle(c *C) {
// After some time, all is good and we should be in stand by mode again
s.advanceTime(5*time.Second + time.Millisecond)
re, err = cb.ProcessRequest(okReq)
re, _, err = testutils.Get(srv.URL)
c.Assert(cb.state, Equals, cbState(stateStandby))
c.Assert(err, IsNil)
c.Assert(re, IsNil)
c.Assert(re.StatusCode, Equals, http.StatusOK)
}
/*
func (s *CBSuite) TestRedirect(c *C) {
cb := s.new(c,
triggerNetRatio,
@@ -328,56 +311,7 @@ func (s *CBSuite) TestInvalidParams(c *C) {
}
}
func statsOK() *metrics.RoundTripMetrics {
m, err := metrics.NewRoundTripMetrics(metrics.RoundTripOptions{})
if err != nil {
panic(err)
}
return m
}
func statsNetErrors(threshold float64) *metrics.RoundTripMetrics {
m, err := metrics.NewRoundTripMetrics(metrics.RoundTripOptions{})
if err != nil {
panic(err)
}
for i := 0; i < 100; i++ {
if i < int(threshold*100) {
m.RecordMetrics(&request.BaseAttempt{Error: fmt.Errorf("boo")})
} else {
m.RecordMetrics(&request.BaseAttempt{Response: &http.Response{StatusCode: 200}})
}
}
return m
}
func statsLatencyAtQuantile(quantile float64, value time.Duration) *metrics.RoundTripMetrics {
m, err := metrics.NewRoundTripMetrics(metrics.RoundTripOptions{})
if err != nil {
panic(err)
}
m.RecordMetrics(&request.BaseAttempt{Duration: value})
return m
}
func statsResponseCodes(codes ...statusCode) *metrics.RoundTripMetrics {
m, err := metrics.NewRoundTripMetrics(metrics.RoundTripOptions{})
if err != nil {
panic(err)
}
for _, c := range codes {
for i := int64(0); i < c.Count; i++ {
m.RecordMetrics(&request.BaseAttempt{Response: &http.Response{StatusCode: c.Code}})
}
}
return m
}
type statusCode struct {
Code int
Count int64
}
type O struct {
stats *metrics.RoundTripMetrics
id int64
@@ -403,3 +337,54 @@ func expr(v string) threshold.Predicate {
}
return e
}
*/
func statsOK() *memmetrics.RTMetrics {
m, err := memmetrics.NewRTMetrics()
if err != nil {
panic(err)
}
return m
}
func statsNetErrors(threshold float64) *memmetrics.RTMetrics {
m, err := memmetrics.NewRTMetrics()
if err != nil {
panic(err)
}
for i := 0; i < 100; i++ {
if i < int(threshold*100) {
m.Record(http.StatusGatewayTimeout, 0)
} else {
m.Record(http.StatusOK, 0)
}
}
return m
}
func statsLatencyAtQuantile(quantile float64, value time.Duration) *memmetrics.RTMetrics {
m, err := memmetrics.NewRTMetrics()
if err != nil {
panic(err)
}
m.Record(http.StatusOK, value)
return m
}
func statsResponseCodes(codes ...statusCode) *memmetrics.RTMetrics {
m, err := memmetrics.NewRTMetrics()
if err != nil {
panic(err)
}
for _, c := range codes {
for i := int64(0); i < c.Count; i++ {
m.Record(c.Code, 0)
}
}
return m
}
type statusCode struct {
Code int
Count int64
}
circuitbreaker/fallback.go
+11 -19
View File
@@ -4,10 +4,6 @@ import (
"fmt"
"net/http"
"net/url"
"github.com/mailgun/vulcan/errors"
"github.com/mailgun/vulcan/netutils"
"github.com/mailgun/vulcan/request"
)
type Response struct {
@@ -16,10 +12,6 @@ type Response struct {
Body []byte
}
func (re *Response) getHTTPResponse(r request.Request) *http.Response {
return netutils.NewHttpResponse(r.GetHttpRequest(), re.StatusCode, re.Body, re.ContentType)
}
type ResponseFallback struct {
r Response
}
@@ -31,11 +23,12 @@ func NewResponseFallback(r Response) (*ResponseFallback, error) {
return &ResponseFallback{r: r}, nil
}
func (f *ResponseFallback) ProcessRequest(r request.Request) (*http.Response, error) {
return f.r.getHTTPResponse(r), nil
}
func (f *ResponseFallback) ProcessResponse(r request.Request, a request.Attempt) {
func (f *ResponseFallback) ServeHTTP(w http.ResponseWriter, req *http.Request) {
if f.r.ContentType != "" {
w.Header().Set("Content-Type", f.r.ContentType)
}
w.WriteHeader(f.r.StatusCode)
w.Write(f.r.Body)
}
type Redirect struct {
@@ -47,16 +40,15 @@ type RedirectFallback struct {
}
func NewRedirectFallback(r Redirect) (*RedirectFallback, error) {
u, err := netutils.ParseUrl(r.URL)
u, err := url.ParseRequestURI(r.URL)
if err != nil {
return nil, err
}
return &RedirectFallback{u: u}, nil
}
func (f *RedirectFallback) ProcessRequest(r request.Request) (*http.Response, error) {
return nil, &errors.RedirectError{URL: netutils.CopyUrl(f.u)}
}
func (f *RedirectFallback) ProcessResponse(r request.Request, a request.Attempt) {
func (f *RedirectFallback) ServeHTTP(w http.ResponseWriter, req *http.Request) {
w.Header().Set("Location", f.u.String())
w.WriteHeader(http.StatusFound)
w.Write([]byte(http.StatusText(http.StatusFound)))
}
circuitbreaker/predicates.go
+6 -3
View File
@@ -58,6 +58,7 @@ func latencyAtQuantile(quantile float64) toInt {
func networkErrorRatio() toFloat64 {
return func(c *CircuitBreaker) float64 {
c.log.Infof("NetworkErrorRatio() == %v", c.metrics.NetworkErrorRatio())
return c.metrics.NetworkErrorRatio()
}
}
@@ -107,7 +108,7 @@ func eq(m interface{}, value interface{}) (hpredicate, error) {
case toFloat64:
return float64EQ(mapper, value)
}
return nil, fmt.Errorf("unsupported argument: %T", m)
return nil, fmt.Errorf("eq: unsupported argument: %T", m)
}
// neq returns predicate that tests for inequality of the value of the mapper and the constant
@@ -127,7 +128,7 @@ func lt(m interface{}, value interface{}) (hpredicate, error) {
case toFloat64:
return float64LT(mapper, value)
}
return nil, fmt.Errorf("unsupported argument: %T", m)
return nil, fmt.Errorf("lt: unsupported argument: %T", m)
}
// le returns predicate that tests that value of the mapper function is less or equal than the constant
@@ -150,8 +151,10 @@ func gt(m interface{}, value interface{}) (hpredicate, error) {
switch mapper := m.(type) {
case toInt:
return intGT(mapper, value)
case toFloat64:
return float64GT(mapper, value)
}
return nil, fmt.Errorf("unsupported argument: %T", m)
return nil, fmt.Errorf("gt: unsupported argument: %T", m)
}
// ge returns predicate that tests that value of the mapper function is less or equal than the constant
circuitbreaker/predicates_test.go
+2 -3
View File
@@ -1,9 +1,6 @@
package circuitbreaker
import (
"time"
"github.com/mailgun/vulcan/request"
. "gopkg.in/check.v1"
)
@@ -12,6 +9,7 @@ type PredicatesSuite struct {
var _ = Suite(&PredicatesSuite{})
/*
func (s *PredicatesSuite) TestTriggered(c *C) {
predicates := []struct {
Expression string
@@ -88,3 +86,4 @@ func (s *PredicatesSuite) TestErrors(c *C) {
c.Assert(p(t.Request), Equals, false)
}
}
*/