Continued refactoring

memmetrics/roundtrip.go

@@ -7,12 +7,12 @@ import (
 	"github.com/mailgun/timetools"
 )
 
-// RoundTripMetrics provides aggregated performance metrics for HTTP requests processing
+// RTMetrics provides aggregated performance metrics for HTTP requests processing
 // such as round trip latency, response codes counters network error and total requests.
 // all counters are collected as rolling window counters with defined precision, histograms
 // are a rolling window histograms with defined precision as well.
-// See RoundTripOptions for more detail on parameters.
-type RoundTripMetrics struct {
+// See RTOptions for more detail on parameters.
+type RTMetrics struct {
 	total       *RollingCounter
 	netErrors   *RollingCounter
 	statusCodes map[int]*RollingCounter
@@ -23,35 +23,36 @@ type RoundTripMetrics struct {
 	clock      timetools.TimeProvider
 }
 
-type rrOptSetter func(r *RoundTripMetrics) error
+type rrOptSetter func(r *RTMetrics) error
 
+type NewRTMetricsFn func() (*RTMetrics, error)
 type NewCounterFn func() (*RollingCounter, error)
 type NewRollingHistogramFn func() (*RollingHDRHistogram, error)
 
-func RoundTripCounter(new NewCounterFn) rrOptSetter {
-	return func(r *RoundTripMetrics) error {
+func RTCounter(new NewCounterFn) rrOptSetter {
+	return func(r *RTMetrics) error {
 		r.newCounter = new
 		return nil
 	}
 }
 
-func RoundTripHistogram(new NewRollingHistogramFn) rrOptSetter {
-	return func(r *RoundTripMetrics) error {
+func RTHistogram(new NewRollingHistogramFn) rrOptSetter {
+	return func(r *RTMetrics) error {
 		r.newHist = new
 		return nil
 	}
 }
 
-func RoundTripClock(clock timetools.TimeProvider) rrOptSetter {
-	return func(r *RoundTripMetrics) error {
+func RTClock(clock timetools.TimeProvider) rrOptSetter {
+	return func(r *RTMetrics) error {
 		r.clock = clock
 		return nil
 	}
 }
 
-// NewRoundTripMetrics returns new instance of metrics collector.
-func NewRoundTripMetrics(settings ...rrOptSetter) (*RoundTripMetrics, error) {
-	m := &RoundTripMetrics{
+// NewRTMetrics returns new instance of metrics collector.
+func NewRTMetrics(settings ...rrOptSetter) (*RTMetrics, error) {
+	m := &RTMetrics{
 		statusCodes: make(map[int]*RollingCounter),
 	}
 	for _, s := range settings {
@@ -99,7 +100,7 @@ func NewRoundTripMetrics(settings ...rrOptSetter) (*RoundTripMetrics, error) {
 
 // GetNetworkErrorRatio calculates the amont of network errors such as time outs and dropped connection
 // that occured in the given time window compared to the total requests count.
-func (m *RoundTripMetrics) NetworkErrorRatio() float64 {
+func (m *RTMetrics) NetworkErrorRatio() float64 {
 	if m.total.Count() == 0 {
 		return 0
 	}
@@ -107,7 +108,7 @@ func (m *RoundTripMetrics) NetworkErrorRatio() float64 {
 }
 
 // GetResponseCodeRatio calculates ratio of count(startA to endA) / count(startB to endB)
-func (m *RoundTripMetrics) ResponseCodeRatio(startA, endA, startB, endB int) float64 {
+func (m *RTMetrics) ResponseCodeRatio(startA, endA, startB, endB int) float64 {
 	a := int64(0)
 	b := int64(0)
 	for code, v := range m.statusCodes {
@@ -124,7 +125,7 @@ func (m *RoundTripMetrics) ResponseCodeRatio(startA, endA, startB, endB int) flo
 	return 0
 }
 
-func (m *RoundTripMetrics) Record(code int, err error, duration time.Duration) {
+func (m *RTMetrics) Record(code int, err error, duration time.Duration) {
 	m.total.Inc()
 	if _, ok := err.(net.Error); ok {
 		m.netErrors.Inc()
@@ -134,17 +135,17 @@ func (m *RoundTripMetrics) Record(code int, err error, duration time.Duration) {
 }
 
 // GetTotalCount returns total count of processed requests collected.
-func (m *RoundTripMetrics) TotalCount() int64 {
+func (m *RTMetrics) TotalCount() int64 {
 	return m.total.Count()
 }
 
 // GetNetworkErrorCount returns total count of processed requests observed
-func (m *RoundTripMetrics) NetworkErrorCount() int64 {
+func (m *RTMetrics) NetworkErrorCount() int64 {
 	return m.netErrors.Count()
 }
 
 // GetStatusCodesCounts returns map with counts of the response codes
-func (m *RoundTripMetrics) StatusCodesCounts() map[int]int64 {
+func (m *RTMetrics) StatusCodesCounts() map[int]int64 {
 	sc := make(map[int]int64)
 	for k, v := range m.statusCodes {
 		if v.Count() != 0 {
@@ -155,27 +156,27 @@ func (m *RoundTripMetrics) StatusCodesCounts() map[int]int64 {
 }
 
 // GetLatencyHistogram computes and returns resulting histogram with latencies observed.
-func (m *RoundTripMetrics) LatencyHistogram() (*HDRHistogram, error) {
+func (m *RTMetrics) LatencyHistogram() (*HDRHistogram, error) {
 	return m.histogram.Merged()
 }
 
-func (m *RoundTripMetrics) Reset() {
+func (m *RTMetrics) Reset() {
 	m.histogram.Reset()
 	m.total.Reset()
 	m.netErrors.Reset()
 	m.statusCodes = make(map[int]*RollingCounter)
 }
 
-func (m *RoundTripMetrics) recordNetError() error {
+func (m *RTMetrics) recordNetError() error {
 	m.netErrors.Inc()
 	return nil
 }
 
-func (m *RoundTripMetrics) recordLatency(d time.Duration) error {
+func (m *RTMetrics) recordLatency(d time.Duration) error {
 	return m.histogram.RecordLatencies(d, 1)
 }
 
-func (m *RoundTripMetrics) recordStatusCode(statusCode int) error {
+func (m *RTMetrics) recordStatusCode(statusCode int) error {
 	if c, ok := m.statusCodes[statusCode]; ok {
 		c.Inc()
 		return nil

memmetrics/roundtrip_test.go

@@ -21,7 +21,7 @@ func (s *RRSuite) SetUpSuite(c *C) {
 }
 
 func (s *RRSuite) TestDefaults(c *C) {
-	rr, err := NewRoundTripMetrics(RoundTripClock(s.tm))
+	rr, err := NewRTMetrics(RTClock(s.tm))
 	c.Assert(err, IsNil)
 	c.Assert(rr, NotNil)
 

netutils/netutils.go

@@ -0,0 +1,23 @@
+package netutils
+
+import (
+	"net/http"
+)
+
+type ProxyWriter struct {
+	W    http.ResponseWriter
+	Code int
+}
+
+func (p *ProxyWriter) Header() http.Header {
+	return p.W.Header()
+}
+
+func (p *ProxyWriter) Write(buf []byte) (int, error) {
+	return p.W.Write(buf)
+}
+
+func (p *ProxyWriter) WriteHeader(code int) {
+	p.Code = code
+	p.WriteHeader(code)
+}

retry/retry.go

@@ -2,6 +2,8 @@ package retry
 
 import (
 	"net/http"
+
+	"github.com/mailgun/oxy/netutils"
 )
 
 type Retry struct {
@@ -35,7 +37,7 @@ func New(next http.Handler, predicate string, settings ...optSetter) (*Retry, er
 func (r *Retry) ServeHTTP(w http.ResponseWriter, req *http.Request) {
 	c := &context{r: req}
 	for i := 0; i < r.maxAttempts; i++ {
-		pw := &proxyWriter{w: w}
+		pw := &netutils.ProxyWriter{W: w}
 		r.next.ServeHTTP(pw, req)
 		c.attempt = i + 1
 		if !r.predicate(c) {
@@ -63,21 +65,3 @@ func ErrorHandler(h http.Handler) optSetter {
 type optSetter func(r *Retry) error
 
 const DefaultMaxAttempts = 10
-
-type proxyWriter struct {
-	w    http.ResponseWriter
-	code int
-}
-
-func (p *proxyWriter) Header() http.Header {
-	return p.w.Header()
-}
-
-func (p *proxyWriter) Write(buf []byte) (int, error) {
-	return p.w.Write(buf)
-}
-
-func (p *proxyWriter) WriteHeader(code int) {
-	p.code = code
-	p.WriteHeader(code)
-}

roundrobin/rebalancer.go

@@ -0,0 +1,360 @@
+package roundrobin
+
+import (
+	"fmt"
+	"net/http"
+	"net/url"
+	"sync"
+	"time"
+
+	"github.com/mailgun/oxy/memmetrics"
+	"github.com/mailgun/oxy/netutils"
+	"github.com/mailgun/timetools"
+)
+
+type rbOptSetter func(*Rebalancer) error
+
+// Meter measures server peformance and returns it's relative value via rating
+type Meter interface {
+	Rating() float64
+	Record(int, time.Duration)
+	IsReady() bool
+}
+
+type NewMeterFn func() (Meter, error)
+
+// Rebalancer increases weights on servers that perform better than others. It also rolls back to original weights
+// if the servers have changed. It is designed as a wrapper on top of the roundrobin.
+type Rebalancer struct {
+	// mutex
+	mtx *sync.Mutex
+	// As usual, control time in tests
+	clock timetools.TimeProvider
+	// Time that freezes state machine to accumulate stats after updating the weights
+	backoffDuration time.Duration
+	// Timer is set to give probing some time to take place
+	timer time.Time
+	// server records that remember original weights
+	servers []*rbServer
+	// next is  internal load balancer next in chain
+	next balancerHandler
+
+	ratings []float64
+
+	// creates new meters
+	newMeter NewMeterFn
+}
+
+func RebalancerClock(clock timetools.TimeProvider) rbOptSetter {
+	return func(r *Rebalancer) error {
+		r.clock = clock
+		return nil
+	}
+}
+
+func RebalancerBackoff(d time.Duration) rbOptSetter {
+	return func(r *Rebalancer) error {
+		r.backoffDuration = d
+		return nil
+	}
+}
+
+func RebalancerMeter(newMeter NewMeterFn) rbOptSetter {
+	return func(r *Rebalancer) error {
+		r.newMeter = newMeter
+		return nil
+	}
+}
+
+func NewRebalancer(handler balancerHandler, opts ...rbOptSetter) (*Rebalancer, error) {
+	rb := &Rebalancer{
+		mtx: &sync.Mutex{},
+	}
+	for _, o := range opts {
+		if err := o(rb); err != nil {
+			return nil, err
+		}
+	}
+	if rb.clock == nil {
+		rb.clock = &timetools.RealTime{}
+	}
+	if rb.backoffDuration == 0 {
+		rb.backoffDuration = 10 * time.Second
+	}
+	if rb.newMeter == nil {
+		rb.newMeter = func() (Meter, error) {
+			rc, err := memmetrics.NewRatioCounter(10, time.Second, memmetrics.RatioClock(rb.clock))
+			if err != nil {
+				return nil, err
+			}
+			return &codeMeter{
+				r:     rc,
+				codeS: http.StatusInternalServerError,
+				codeE: http.StatusGatewayTimeout,
+			}, nil
+		}
+	}
+	return rb, nil
+}
+
+func (rb *Rebalancer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	pw := &netutils.ProxyWriter{W: w}
+	start := rb.clock.UtcNow()
+	rb.next.ServeHTTP(pw, req)
+	rb.recordMetrics(req.URL, pw.Code, rb.clock.UtcNow().Sub(start))
+}
+
+func (rb *Rebalancer) recordMetrics(u *url.URL, code int, latency time.Duration) {
+	rb.mtx.Lock()
+	defer rb.mtx.Unlock()
+	if srv, i := rb.findServer(u); i != -1 {
+		srv.meter.Record(code, latency)
+	}
+}
+
+func (rb *Rebalancer) reset() {
+	for _, s := range rb.servers {
+		s.curWeight = s.origWeight
+		rb.next.UpsertServer(s.url, Weight(s.origWeight))
+	}
+	rb.timer = rb.clock.UtcNow().Add(-1 * time.Second)
+	rb.ratings = make([]float64, len(rb.servers))
+}
+
+func (rb *Rebalancer) Wrap(next balancerHandler) error {
+	if rb.next != nil {
+		return fmt.Errorf("already bound to %T", rb.next)
+	}
+	rb.next = next
+	return nil
+}
+
+func (rb *Rebalancer) UpsertServer(u *url.URL, options ...serverSetter) error {
+	rb.mtx.Lock()
+	defer rb.mtx.Unlock()
+
+	if err := rb.next.UpsertServer(u, options...); err != nil {
+		return err
+	}
+	weight, _ := rb.next.ServerWeight(u)
+	rb.upsertServer(u, weight)
+	rb.reset()
+	return nil
+}
+
+func (rb *Rebalancer) RemoveServer(u *url.URL) error {
+	rb.mtx.Lock()
+	defer rb.mtx.Unlock()
+
+	return rb.removeServer(u)
+}
+
+func (rb *Rebalancer) removeServer(u *url.URL) error {
+	_, i := rb.findServer(u)
+	if i == -1 {
+		return fmt.Errorf("%v not found", u)
+	}
+	rb.servers = append(rb.servers[:i], rb.servers[i+1:]...)
+	rb.reset()
+	return nil
+}
+
+func (rb *Rebalancer) upsertServer(u *url.URL, weight int) {
+	if s, i := rb.findServer(u); i != -1 {
+		s.origWeight = weight
+	}
+	rb.servers = append(rb.servers, &rbServer{url: copyURL(u), origWeight: weight, curWeight: weight})
+}
+
+func (r *Rebalancer) findServer(u *url.URL) (*rbServer, int) {
+	if len(r.servers) == 0 {
+		return nil, -1
+	}
+	for i, s := range r.servers {
+		if sameURL(u, s.url) {
+			return s, i
+		}
+	}
+	return nil, -1
+}
+
+// Called on every load balancer ServeHTTP call, returns the suggested weights
+// on every call, can adjust weights if needed.
+func (rb *Rebalancer) adjustWeights() {
+	rb.mtx.Lock()
+	defer rb.mtx.Unlock()
+
+	// In this case adjusting weights would have no effect, so do nothing
+	if len(rb.servers) < 2 {
+		return
+	}
+	// Metrics are not ready
+	if !rb.metricsReady() {
+		return
+	}
+	if !rb.timerExpired() {
+		return
+	}
+	if rb.markServers() {
+		if rb.setMarkedWeights() {
+			rb.setTimer()
+		}
+	} else { // No servers that are different by their quality, so converge weights
+		if rb.convergeWeights() {
+			rb.setTimer()
+		}
+	}
+}
+
+func (rb *Rebalancer) applyWeights() {
+	for _, srv := range rb.servers {
+		rb.next.UpsertServer(srv.url, Weight(srv.curWeight))
+	}
+}
+
+func (rb *Rebalancer) setMarkedWeights() bool {
+	changed := false
+	// Increase weights on servers marked as good
+	for _, srv := range rb.servers {
+		if srv.good {
+			weight := increase(srv.curWeight)
+			if weight <= FSMMaxWeight {
+				srv.curWeight = weight
+				changed = true
+			}
+		}
+	}
+	if changed {
+		rb.normalizeWeights()
+		rb.applyWeights()
+		return true
+	}
+	return false
+}
+
+func (rb *Rebalancer) setTimer() {
+	rb.timer = rb.clock.UtcNow().Add(rb.backoffDuration)
+}
+
+func (rb *Rebalancer) timerExpired() bool {
+	return rb.timer.Before(rb.clock.UtcNow())
+}
+
+func (rb *Rebalancer) metricsReady() bool {
+	for _, s := range rb.servers {
+		if !s.meter.IsReady() {
+			return false
+		}
+	}
+	return true
+}
+
+// markServers splits servers into two groups of servers with bad and good failure rate.
+// It does compare relative performances of the servers though, so if all servers have approximately the same error rate
+// this function returns the result as if all servers are equally good.
+func (rb *Rebalancer) markServers() bool {
+	for i, srv := range rb.servers {
+		rb.ratings[i] = srv.meter.Rating()
+	}
+	g, b := memmetrics.SplitFloat64(splitThreshold, 0, rb.ratings)
+	for i, srv := range rb.servers {
+		if g[rb.ratings[i]] {
+			srv.good = true
+		} else {
+			srv.good = false
+		}
+	}
+	return len(g) != 0 && len(b) != 0
+}
+
+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
+		}
+		s.curWeight = decrease(s.origWeight, s.curWeight)
+	}
+	if !changed {
+		return false
+	}
+	rb.normalizeWeights()
+	return true
+}
+
+func (rb *Rebalancer) weightsGcd() int {
+	divisor := -1
+	for _, w := range rb.servers {
+		if divisor == -1 {
+			divisor = w.curWeight
+		} else {
+			divisor = gcd(divisor, w.curWeight)
+		}
+	}
+	return divisor
+}
+
+func (rb *Rebalancer) normalizeWeights() {
+	gcd := rb.weightsGcd()
+	if gcd <= 1 {
+		return
+	}
+	for _, s := range rb.servers {
+		s.curWeight = s.curWeight / gcd
+	}
+}
+
+func increase(weight int) int {
+	return weight * FSMGrowFactor
+}
+
+func decrease(target, current int) int {
+	adjusted := current / FSMGrowFactor
+	if adjusted < target {
+		return target
+	} else {
+		return adjusted
+	}
+}
+
+// rebalancer server record that keeps track of the original weight supplied by user
+type rbServer struct {
+	url        *url.URL
+	origWeight int // original weight supplied by user
+	curWeight  int // current weight
+	good       bool
+	meter      Meter
+}
+
+const (
+	// This is the maximum weight that handler will set for the server
+	FSMMaxWeight = 4096
+	// Multiplier for the server weight
+	FSMGrowFactor = 16
+)
+
+type codeMeter struct {
+	r     *memmetrics.RatioCounter
+	codeS int
+	codeE int
+}
+
+func (n *codeMeter) Rating() float64 {
+	return n.r.Ratio()
+}
+
+func (n *codeMeter) Record(code int, d time.Duration) {
+	if code >= n.codeS && code < n.codeE {
+		n.r.IncA()
+	} else {
+		n.r.IncB()
+	}
+}
+
+func (n *codeMeter) IsReady() bool {
+	return n.r.IsReady()
+}
+
+// splitThreshold tells how far the value should go from the median + median absolute deviation before it is considered an outlier
+const splitThreshold = 1.5

roundrobin/rr.go

@@ -113,6 +113,16 @@ func (r *RoundRobin) RemoveServer(u *url.URL) error {
 	return nil
 }
 
+func (rr *RoundRobin) ServerWeight(u *url.URL) (int, bool) {
+	rr.mutex.Lock()
+	defer rr.mutex.Unlock()
+
+	if s, _ := rr.findServerByURL(u); s != nil {
+		return s.weight, true
+	}
+	return -1, false
+}
+
 // In case if server is already present in the load balancer, returns error
 func (rr *RoundRobin) UpsertServer(u *url.URL, options ...serverSetter) error {
 	rr.mutex.Lock()
@@ -156,7 +166,7 @@ func (r *RoundRobin) findServerByURL(u *url.URL) (*server, int) {
 		return nil, -1
 	}
 	for i, s := range r.servers {
-		if u.Path == s.url.Path && u.Host == s.url.Host && u.Scheme == s.url.Scheme {
+		if sameURL(u, s.url) {
 			return s, i
 		}
 	}
@@ -212,3 +222,14 @@ func copyURL(i *url.URL) *url.URL {
 	}
 	return &out
 }
+
+func sameURL(a, b *url.URL) bool {
+	return a.Path == b.Path && a.Host == b.Host && a.Scheme == b.Scheme
+}
+
+type balancerHandler interface {
+	ServeHTTP(w http.ResponseWriter, req *http.Request)
+	ServerWeight(u *url.URL) (int, bool)
+	RemoveServer(u *url.URL) error
+	UpsertServer(u *url.URL, options ...serverSetter) error
+}