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

Restored original rebalancer tests. Todo for tomorrow:

Browse Source 
* Cleanup the rebalancer tests, make them shorter
* Add realtime tests with real response code meter
This commit is contained in:
Alexander Klizhentas
2014-12-10 20:59:30 -08:00
parent 1daee77194
commit d626e22e51
4 changed files with 249 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
roundrobin/rebalancer.go
+20 -3
View File
@@ -39,12 +39,21 @@ type Rebalancer struct {
// next is internal load balancer next in chain
next balancerHandler
log utils.Logger
ratings []float64
// creates new meters
newMeter NewMeterFn
}
func RebalancerLogger(log utils.Logger) rbOptSetter {
return func(r *Rebalancer) error {
r.log = log
return nil
}
}
func RebalancerClock(clock timetools.TimeProvider) rbOptSetter {
return func(r *Rebalancer) error {
r.clock = clock
@@ -82,6 +91,9 @@ func NewRebalancer(handler balancerHandler, opts ...rbOptSetter) (*Rebalancer, e
if rb.backoffDuration == 0 {
rb.backoffDuration = 10 * time.Second
}
if rb.log == nil {
rb.log = &utils.NOPLogger{}
}
if rb.newMeter == nil {
rb.newMeter = func() (Meter, error) {
rc, err := memmetrics.NewRatioCounter(10, time.Second, memmetrics.RatioClock(rb.clock))
@@ -103,6 +115,7 @@ func (rb *Rebalancer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
start := rb.clock.UtcNow()
rb.next.ServeHTTP(w, req)
rb.recordMetrics(req.URL, pw.Code, rb.clock.UtcNow().Sub(start))
rb.adjustWeights()
}
func (rb *Rebalancer) recordMetrics(u *url.URL, code int, latency time.Duration) {
@@ -234,6 +247,7 @@ func (rb *Rebalancer) setMarkedWeights() bool {
if srv.good {
weight := increase(srv.curWeight)
if weight <= FSMMaxWeight {
rb.log.Infof("increasing weight of %v from %v to %v", srv.url, srv.curWeight, weight)
srv.curWeight = weight
changed = true
}
@@ -286,10 +300,13 @@ func (rb *Rebalancer) convergeWeights() bool {
// If we have previoulsy changed servers try to restore weights to the original state
changed := false
for _, s := range rb.servers {
if s.origWeight != s.curWeight {
changed = true
if s.origWeight == s.curWeight {
continue
}
s.curWeight = decrease(s.origWeight, s.curWeight)
changed = true
newWeight := decrease(s.origWeight, s.curWeight)
rb.log.Infof("decreasing weight of %v from %v to %v", s.url, s.curWeight, newWeight)
s.curWeight = newWeight
}
if !changed {
return false
roundrobin/rebalancer_test.go
+221 -6
View File
@@ -3,14 +3,33 @@ package roundrobin
import (
"net/http"
"net/http/httptest"
"os"
"time"
"github.com/mailgun/oxy/forward"
"github.com/mailgun/oxy/testutils"
"github.com/mailgun/oxy/utils"
"github.com/mailgun/timetools"
. "gopkg.in/check.v1"
)
func (s *RRSuite) TestRebalancerNormalOperation(c *C) {
type RBSuite struct {
clock *timetools.FreezedTime
log utils.Logger
}
var _ = Suite(&RBSuite{
clock: &timetools.FreezedTime{
CurrentTime: time.Date(2012, 3, 4, 5, 6, 7, 0, time.UTC),
},
})
func (s *RBSuite) SetUpSuite(c *C) {
s.log = utils.NewFileLogger(os.Stdout, utils.INFO)
}
func (s *RBSuite) TestRebalancerNormalOperation(c *C) {
a := testutils.NewResponder("a")
defer a.Close()
@@ -31,10 +50,10 @@ func (s *RRSuite) TestRebalancerNormalOperation(c *C) {
proxy := httptest.NewServer(rb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
c.Assert(seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
}
func (s *RRSuite) TestRebalancerNoServers(c *C) {
func (s *RBSuite) TestRebalancerNoServers(c *C) {
fwd, err := forward.New()
c.Assert(err, IsNil)
@@ -52,7 +71,8 @@ func (s *RRSuite) TestRebalancerNoServers(c *C) {
c.Assert(re.StatusCode, Equals, http.StatusInternalServerError)
}
func (s *RRSuite) TestRebalancerOneIsDown(c *C) {
// Test scenario when one server goes down after what it recovers
func (s *RBSuite) TestRebalancerRecovery(c *C) {
a := testutils.NewResponder("a")
defer a.Close()
@@ -65,13 +85,208 @@ func (s *RRSuite) TestRebalancerOneIsDown(c *C) {
lb, err := New(fwd)
c.Assert(err, IsNil)
rb, err := NewRebalancer(lb)
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
rb.UpsertServer(testutils.ParseURI(b.URL))
rb.servers[0].meter.(*testMeter).rating = 0.3
proxy := httptest.NewServer(rb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
for i := 0; i < 6; i += 1 {
testutils.Get(proxy.URL)
testutils.Get(proxy.URL)
s.clock.CurrentTime = s.clock.CurrentTime.Add(rb.backoffDuration + time.Second)
}
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, FSMMaxWeight)
// server a is now recovering, the weights should go back to the original state
rb.servers[0].meter.(*testMeter).rating = 0
for i := 0; i < 6; i += 1 {
testutils.Get(proxy.URL)
testutils.Get(proxy.URL)
s.clock.CurrentTime = s.clock.CurrentTime.Add(rb.backoffDuration + time.Second)
}
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, 1)
}
// Test scenario when increaing the weight on good endpoints made it worse
func (s *RBSuite) TestRebalancerCascading(c *C) {
a := testutils.NewResponder("a")
defer a.Close()
b := testutils.NewResponder("b")
defer b.Close()
d := testutils.NewResponder("d")
defer d.Close()
fwd, err := forward.New()
c.Assert(err, IsNil)
lb, err := New(fwd)
c.Assert(err, IsNil)
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
rb.UpsertServer(testutils.ParseURI(b.URL))
rb.UpsertServer(testutils.ParseURI(d.URL))
rb.servers[0].meter.(*testMeter).rating = 0.3
proxy := httptest.NewServer(rb)
defer proxy.Close()
for i := 0; i < 6; i += 1 {
testutils.Get(proxy.URL)
testutils.Get(proxy.URL)
s.clock.CurrentTime = s.clock.CurrentTime.Add(rb.backoffDuration + time.Second)
}
// We have increased the load, and the situation became worse as the other servers started failing
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, FSMMaxWeight)
c.Assert(rb.servers[2].curWeight, Equals, FSMMaxWeight)
// server a is now recovering, the weights should go back to the original state
rb.servers[0].meter.(*testMeter).rating = 0.3
rb.servers[1].meter.(*testMeter).rating = 0.2
rb.servers[2].meter.(*testMeter).rating = 0.2
for i := 0; i < 6; i += 1 {
testutils.Get(proxy.URL)
testutils.Get(proxy.URL)
s.clock.CurrentTime = s.clock.CurrentTime.Add(rb.backoffDuration + time.Second)
}
// the algo reverted it back
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, 1)
c.Assert(rb.servers[2].curWeight, Equals, 1)
}
// Test scenario when all servers started failing
func (s *RBSuite) TestRebalancerAllBad(c *C) {
a := testutils.NewResponder("a")
defer a.Close()
b := testutils.NewResponder("b")
defer b.Close()
d := testutils.NewResponder("d")
defer d.Close()
fwd, err := forward.New()
c.Assert(err, IsNil)
lb, err := New(fwd)
c.Assert(err, IsNil)
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
rb.UpsertServer(testutils.ParseURI(b.URL))
rb.UpsertServer(testutils.ParseURI(d.URL))
rb.servers[0].meter.(*testMeter).rating = 0.12
rb.servers[1].meter.(*testMeter).rating = 0.13
rb.servers[2].meter.(*testMeter).rating = 0.11
proxy := httptest.NewServer(rb)
defer proxy.Close()
for i := 0; i < 6; i += 1 {
testutils.Get(proxy.URL)
testutils.Get(proxy.URL)
s.clock.CurrentTime = s.clock.CurrentTime.Add(rb.backoffDuration + time.Second)
}
// load balancer does nothing
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, 1)
c.Assert(rb.servers[2].curWeight, Equals, 1)
}
// Removing the server resets the state
func (s *RBSuite) TestRebalancerReset(c *C) {
a := testutils.NewResponder("a")
defer a.Close()
b := testutils.NewResponder("b")
defer b.Close()
d := testutils.NewResponder("d")
defer d.Close()
fwd, err := forward.New()
c.Assert(err, IsNil)
lb, err := New(fwd)
c.Assert(err, IsNil)
newMeter := func() (Meter, error) {
return &testMeter{}, nil
}
rb, err := NewRebalancer(lb, RebalancerMeter(newMeter), RebalancerClock(s.clock), RebalancerLogger(s.log))
c.Assert(err, IsNil)
rb.UpsertServer(testutils.ParseURI(a.URL))
rb.UpsertServer(testutils.ParseURI(b.URL))
rb.UpsertServer(testutils.ParseURI(d.URL))
rb.servers[0].meter.(*testMeter).rating = 0.3
rb.servers[1].meter.(*testMeter).rating = 0
rb.servers[2].meter.(*testMeter).rating = 0
proxy := httptest.NewServer(rb)
defer proxy.Close()
for i := 0; i < 6; i += 1 {
testutils.Get(proxy.URL)
testutils.Get(proxy.URL)
s.clock.CurrentTime = s.clock.CurrentTime.Add(rb.backoffDuration + time.Second)
}
// load balancer changed weights
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, FSMMaxWeight)
c.Assert(rb.servers[2].curWeight, Equals, FSMMaxWeight)
// Removing servers has reset the state
rb.RemoveServer(testutils.ParseURI(d.URL))
c.Assert(rb.servers[0].curWeight, Equals, 1)
c.Assert(rb.servers[1].curWeight, Equals, 1)
}
type testMeter struct {
rating float64
notReady bool
}
func (tm *testMeter) Rating() float64 {
return tm.rating
}
func (tm *testMeter) Record(int, time.Duration) {
}
func (tm *testMeter) IsReady() bool {
return !tm.notReady
}
roundrobin/rr.go
+1 -2
View File
@@ -216,8 +216,7 @@ type server struct {
weight int
}
// This is really a convenience, so users can add servers with smaller weights if they did not set it initially
const defaultWeight = 10
const defaultWeight = 1
func sameURL(a, b *url.URL) bool {
return a.Path == b.Path && a.Host == b.Host && a.Scheme == b.Scheme
roundrobin/rr_test.go
+7 -7
View File
@@ -75,7 +75,7 @@ func (s *RRSuite) TestOneServer(c *C) {
proxy := httptest.NewServer(lb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
c.Assert(seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
}
func (s *RRSuite) TestSimple(c *C) {
@@ -97,7 +97,7 @@ func (s *RRSuite) TestSimple(c *C) {
proxy := httptest.NewServer(lb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"a", "b", "a"})
c.Assert(seq(c, proxy.URL, 3), DeepEquals, []string{"a", "b", "a"})
}
func (s *RRSuite) TestRemoveServer(c *C) {
@@ -119,11 +119,11 @@ func (s *RRSuite) TestRemoveServer(c *C) {
proxy := httptest.NewServer(lb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"a", "b", "a"})
c.Assert(seq(c, proxy.URL, 3), DeepEquals, []string{"a", "b", "a"})
lb.RemoveServer(testutils.ParseURI(a.URL))
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"b", "b", "b"})
c.Assert(seq(c, proxy.URL, 3), DeepEquals, []string{"b", "b", "b"})
}
func (s *RRSuite) TestUpsertSame(c *C) {
@@ -142,7 +142,7 @@ func (s *RRSuite) TestUpsertSame(c *C) {
proxy := httptest.NewServer(lb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
c.Assert(seq(c, proxy.URL, 3), DeepEquals, []string{"a", "a", "a"})
}
func (s *RRSuite) TestWeighted(c *C) {
@@ -164,7 +164,7 @@ func (s *RRSuite) TestWeighted(c *C) {
proxy := httptest.NewServer(lb)
defer proxy.Close()
c.Assert(s.seq(c, proxy.URL, 6), DeepEquals, []string{"a", "a", "b", "a", "b", "a"})
c.Assert(seq(c, proxy.URL, 6), DeepEquals, []string{"a", "a", "b", "a", "b", "a"})
w, ok := lb.ServerWeight(testutils.ParseURI(a.URL))
c.Assert(w, Equals, 3)
@@ -179,7 +179,7 @@ func (s *RRSuite) TestWeighted(c *C) {
c.Assert(ok, Equals, false)
}
func (s *RRSuite) seq(c *C, url string, repeat int) []string {
func seq(c *C, url string, repeat int) []string {
out := []string{}
for i := 0; i < repeat; i++ {
_, body, err := testutils.Get(url)