Memmetrics ported

Makefile

@@ -0,0 +1,18 @@
+test: clean
+	go test -v ./... -cover
+
+clean:
+	find . -name flymake_* -delete
+
+test-package: clean
+	go test -v ./$(p)
+
+test-grep-package: clean
+	go test -v ./$(p) -check.f=$(e)
+
+cover-package: clean
+	go test -v ./$(p)  -coverprofile=/tmp/coverage.out
+	go tool cover -html=/tmp/coverage.out
+
+sloccount:
+	 find . -path ./Godeps -prune -o -name "*.go" -print0 | xargs -0 wc -l

memmetrics/counter.go

@@ -7,9 +7,18 @@ import (
 	"github.com/mailgun/timetools"
 )
 
+type rcOptSetter func(*RollingCounter) error
+
+func CounterClock(c timetools.TimeProvider) rcOptSetter {
+	return func(r *RollingCounter) error {
+		r.clock = c
+		return nil
+	}
+}
+
 // Calculates in memory failure rate of an endpoint using rolling window of a predefined size
 type RollingCounter struct {
-	timeProvider   timetools.TimeProvider
+	clock          timetools.TimeProvider
 	resolution     time.Duration
 	values         []int
 	countedBuckets int // how many samples in different buckets have we collected so far
@@ -17,9 +26,10 @@ type RollingCounter struct {
 	lastUpdated    time.Time
 }
 
-// NewRollingCounter creates a counter with fixed amount of buckets that are rotated every resolition period.
+// NewCounter creates a counter with fixed amount of buckets that are rotated every resolution period.
 // E.g. 10 buckets with 1 second means that every new second the bucket is refreshed, so it maintains 10 second rolling window.
-func NewCounter(buckets int, resolution time.Duration, timeProvider timetools.TimeProvider) (*RollingCounter, error) {
+// By default creates a bucket with 10 buckets and 1 second resolution
+func NewCounter(buckets int, resolution time.Duration, options ...rcOptSetter) (*RollingCounter, error) {
 	if buckets <= 0 {
 		return nil, fmt.Errorf("Buckets should be >= 0")
 	}
@@ -27,12 +37,24 @@ func NewCounter(buckets int, resolution time.Duration, timeProvider timetools.Ti
 		return nil, fmt.Errorf("Resolution should be larger than a second")
 	}
 
-	return &RollingCounter{
-		resolution:   resolution,
-		timeProvider: timeProvider,
-		values:       make([]int, buckets),
-		lastBucket:   -1,
-	}, nil
+	rc := &RollingCounter{
+		lastBucket: -1,
+		resolution: resolution,
+
+		values: make([]int, buckets),
+	}
+
+	for _, o := range options {
+		if err := o(rc); err != nil {
+			return nil, err
+		}
+	}
+
+	if rc.clock == nil {
+		rc.clock = &timetools.RealTime{}
+	}
+
+	return rc, nil
 }
 
 func (c *RollingCounter) Reset() {
@@ -71,7 +93,7 @@ func (c *RollingCounter) Inc() {
 }
 
 func (c *RollingCounter) incBucketValue() {
-	now := c.timeProvider.UtcNow()
+	now := c.clock.UtcNow()
 	bucket := c.getBucket(now)
 	c.values[bucket]++
 	c.lastUpdated = now
@@ -93,7 +115,7 @@ func (c *RollingCounter) getBucket(t time.Time) int {
 
 // Reset buckets that were not updated
 func (c *RollingCounter) cleanup() {
-	now := c.timeProvider.UtcNow()
+	now := c.clock.UtcNow()
 	for i := 0; i < len(c.values); i++ {
 		now = now.Add(time.Duration(-1*i) * c.resolution)
 		if now.Truncate(c.resolution).After(c.lastUpdated.Truncate(c.resolution)) {

memmetrics/histogram.go

@@ -26,15 +26,12 @@ func NewHDRHistogram(low, high int64, sigfigs int) (h *HDRHistogram, err error)
 			err = fmt.Errorf("%s", msg)
 		}
 	}()
-
-	hdr := hdrhistogram.New(low, high, sigfigs)
-	h = &HDRHistogram{
+	return &HDRHistogram{
 		low:     low,
 		high:    high,
 		sigfigs: sigfigs,
-		h:       hdr,
-	}
-	return h, err
+		h:       hdrhistogram.New(low, high, sigfigs),
+	}, nil
 }
 
 // Returns latency at quantile with microsecond precision
@@ -60,78 +57,56 @@ func (h *HDRHistogram) RecordValues(v, n int64) error {
 }
 
 func (h *HDRHistogram) Merge(other *HDRHistogram) error {
+	if other == nil {
+		return fmt.Errorf("other is nil")
+	}
 	h.h.Merge(other.h)
 	return nil
 }
 
 type rhOptSetter func(r *RollingHDRHistogram) error
 
-func Clock(clock timetools.TimeProvider) rhOptSetter {
+func RollingClock(clock timetools.TimeProvider) rhOptSetter {
 	return func(r *RollingHDRHistogram) error {
 		r.clock = clock
 		return nil
 	}
 }
 
-func MinValue(low int) rhOptSetter {
-	return func(r *RollingHDRHistogram) error {
-		r.low = low
-		return nil
-	}
-}
-
-func MaxValue(max int) rhOptSetter {
-	return func(r *RollingHDRHistogram) error {
-		r.high = high
-		return nil
-	}
-}
-
-func SigFigs(v int) rhOptSetter {
-	return func(r *RollingHDRHistogram) error {
-		r.sigfigs = sigfigs
-		return nil
-	}
-}
-
 // RollingHistogram holds multiple histograms and rotates every period.
 // It provides resulting histogram as a result of a call of 'Merged' function.
 type RollingHDRHistogram struct {
-	low      int
-	high     int
-	sigfigs  int
-	idx      int
-	lastRoll time.Time
-	period   time.Duration
-	buckets  []*HDRHistogram
-	clock    timetools.TimeProvider
+	idx         int
+	lastRoll    time.Time
+	period      time.Duration
+	bucketCount int
+	low         int64
+	high        int64
+	sigfigs     int
+	buckets     []*HDRHistogram
+	clock       timetools.TimeProvider
 }
 
-func NewRollingHDRHistogram(bucketCount int, period time.Duration, settings ...rhOptSetter) (*RollingHDRHistogram, error) {
-	rh := &rollingHistogram{
-		buckets: buckets,
-		period:  period,
+func NewRollingHDRHistogram(low, high int64, sigfigs int, period time.Duration, bucketCount int, options ...rhOptSetter) (*RollingHDRHistogram, error) {
+	rh := &RollingHDRHistogram{
+		bucketCount: bucketCount,
+		period:      period,
+		low:         low,
+		high:        high,
+		sigfigs:     sigfigs,
 	}
 
-	for _, s := range settings {
-		if err := s(rh); err != nil {
+	for _, o := range options {
+		if err := o(rh); err != nil {
 			return nil, err
 		}
 	}
 
-	if rh.low == 0 {
-		rh.low = histMin
+	if rh.clock == nil {
+		rh.clock = &timetools.RealTime{}
 	}
 
-	if rh.high == 0 {
-		rh.high = histMax
-	}
-
-	if rh.sigfigs == 0 {
-		rw.sigfigs = histSignificantFigures
-	}
-
-	buckets := make([]Histogram, bucketCount)
+	buckets := make([]*HDRHistogram, rh.bucketCount)
 	for i := range buckets {
 		h, err := NewHDRHistogram(low, high, sigfigs)
 		if err != nil {
@@ -139,13 +114,13 @@ func NewRollingHDRHistogram(bucketCount int, period time.Duration, settings ...r
 		}
 		buckets[i] = h
 	}
-
+	rh.buckets = buckets
 	return rh, nil
 }
 
 func (r *RollingHDRHistogram) Reset() {
 	r.idx = 0
-	r.lastRoll = r.timeProvider.UtcNow()
+	r.lastRoll = r.clock.UtcNow()
 	for _, b := range r.buckets {
 		b.Reset()
 	}
@@ -157,7 +132,7 @@ func (r *RollingHDRHistogram) rotate() {
 }
 
 func (r *RollingHDRHistogram) Merged() (*HDRHistogram, error) {
-	m, err := r.maker()
+	m, err := NewHDRHistogram(r.low, r.high, r.sigfigs)
 	if err != nil {
 		return m, err
 	}
@@ -170,9 +145,9 @@ func (r *RollingHDRHistogram) Merged() (*HDRHistogram, error) {
 }
 
 func (r *RollingHDRHistogram) getHist() *HDRHistogram {
-	if r.timeProvider.UtcNow().Sub(r.lastRoll) >= r.period {
+	if r.clock.UtcNow().Sub(r.lastRoll) >= r.period {
 		r.rotate()
-		r.lastRoll = r.timeProvider.UtcNow()
+		r.lastRoll = r.clock.UtcNow()
 	}
 	return r.buckets[r.idx]
 }

memmetrics/histogram_test.go

@@ -49,11 +49,13 @@ func (s *HistogramSuite) TestMergeNil(c *C) {
 }
 
 func (s *HistogramSuite) TestRotation(c *C) {
-	h, err := NewRollingHistogram(
-		NewHDRHistogramFn(1, 3600000, 3),
-		2,           // 2 histograms in a window
+	h, err := NewRollingHDRHistogram(
+		1,           // min value
+		3600000,     // max value
+		3,           // significant figurwes
 		time.Second, // 1 second is a rolling period
-		s.tm)
+		2,           // 2 histograms in a window
+		RollingClock(s.tm))
 
 	c.Assert(err, IsNil)
 	c.Assert(h, NotNil)
@@ -81,11 +83,13 @@ func (s *HistogramSuite) TestRotation(c *C) {
 }
 
 func (s *HistogramSuite) TestReset(c *C) {
-	h, err := NewRollingHistogram(
-		NewHDRHistogramFn(1, 3600000, 3),
-		2,           // 2 histograms in a window
+	h, err := NewRollingHDRHistogram(
+		1,           // min value
+		3600000,     // max value
+		3,           // significant figurwes
 		time.Second, // 1 second is a rolling period
-		s.tm)
+		2,           // 2 histograms in a window
+		RollingClock(s.tm))
 
 	c.Assert(err, IsNil)
 	c.Assert(h, NotNil)

memmetrics/ratio.go

@@ -6,27 +6,48 @@ import (
 	"github.com/mailgun/timetools"
 )
 
-// RatioCounter calculates a ratio of a/a+b over a rolling window of predefined buckets
-type RatioCounter struct {
-	a *RollingCounter
-	b *RollingCounter
+type ratioOptSetter func(r *RatioCounter) error
+
+func RatioClock(clock timetools.TimeProvider) ratioOptSetter {
+	return func(r *RatioCounter) error {
+		r.clock = clock
+		return nil
+	}
 }
 
-func NewRatioCounter(buckets int, resolution time.Duration, timeProvider timetools.TimeProvider) (*RatioCounter, error) {
-	a, err := NewCounter(buckets, resolution, timeProvider)
+// RatioCounter calculates a ratio of a/a+b over a rolling window of predefined buckets
+type RatioCounter struct {
+	clock timetools.TimeProvider
+	a     *RollingCounter
+	b     *RollingCounter
+}
+
+func NewRatioCounter(buckets int, resolution time.Duration, options ...ratioOptSetter) (*RatioCounter, error) {
+	rc := &RatioCounter{}
+
+	for _, o := range options {
+		if err := o(rc); err != nil {
+			return nil, err
+		}
+	}
+
+	if rc.clock == nil {
+		rc.clock = &timetools.RealTime{}
+	}
+
+	a, err := NewCounter(buckets, resolution, CounterClock(rc.clock))
 	if err != nil {
 		return nil, err
 	}
 
-	b, err := NewCounter(buckets, resolution, timeProvider)
+	b, err := NewCounter(buckets, resolution, CounterClock(rc.clock))
 	if err != nil {
 		return nil, err
 	}
 
-	return &RatioCounter{
-		a: a,
-		b: b,
-	}, nil
+	rc.a = a
+	rc.b = b
+	return rc, nil
 }
 
 func (r *RatioCounter) Reset() {

memmetrics/ratio_test.go

@@ -1,14 +1,10 @@
 package memmetrics
 
 import (
-	"fmt"
 	"testing"
 	"time"
 
 	"github.com/mailgun/timetools"
-	. "github.com/mailgun/vulcan/endpoint"
-	. "github.com/mailgun/vulcan/request"
-
 	. "gopkg.in/check.v1"
 )
 
@@ -27,222 +23,151 @@ func (s *FailRateSuite) SetUpSuite(c *C) {
 }
 
 func (s *FailRateSuite) TestInvalidParams(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	// Invalid endpoint
-	_, err := NewRollingMeter(nil, 10, time.Second, s.tm, nil)
-	c.Assert(err, Not(IsNil))
-
 	// Bad buckets count
-	_, err = NewRollingMeter(e, 0, time.Second, s.tm, nil)
+	_, err := NewRatioCounter(0, time.Second, RatioClock(s.tm))
 	c.Assert(err, Not(IsNil))
 
 	// Too precise resolution
-	_, err = NewRollingMeter(e, 10, time.Millisecond, s.tm, nil)
+	_, err = NewRatioCounter(10, time.Millisecond, RatioClock(s.tm))
 	c.Assert(err, Not(IsNil))
 }
 
 func (s *FailRateSuite) TestNotReady(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
 	// No data
-	fr, err := NewRollingMeter(e, 10, time.Second, s.tm, nil)
+	fr, err := NewRatioCounter(10, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
 	c.Assert(fr.IsReady(), Equals, false)
-	c.Assert(fr.GetRate(), Equals, 0.0)
+	c.Assert(fr.Ratio(), Equals, 0.0)
 
 	// Not enough data
-	fr, err = NewRollingMeter(e, 10, time.Second, s.tm, nil)
+	fr, err = NewRatioCounter(10, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.CountA()
 	c.Assert(fr.IsReady(), Equals, false)
 }
 
-// Make sure we don't count the stats from the endpoints we don't care or requests with no attempts
-func (s *FailRateSuite) TestIgnoreOtherEndpoints(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-	e2 := MustParseUrl("http://localhost:5001")
-
-	fr, err := NewRollingMeter(e, 1, time.Second, s.tm, nil)
+func (s *FailRateSuite) TestNoB(c *C) {
+	fr, err := NewRatioCounter(1, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
-	fr.ObserveResponse(makeFailRequest(e))
-	fr.ObserveResponse(makeFailRequest(e2))
-
+	fr.IncA()
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 1.0)
+	c.Assert(fr.Ratio(), Equals, 1.0)
 }
 
-func (s *FailRateSuite) TestIgnoreRequestsWithoutAttempts(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 1, time.Second, s.tm, nil)
+func (s *FailRateSuite) TestNoA(c *C) {
+	fr, err := NewRatioCounter(1, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
-	fr.ObserveResponse(makeFailRequest(e))
-	fr.ObserveResponse(&BaseRequest{}, nil)
-
+	fr.IncB()
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 1.0)
-}
-
-func (s *FailRateSuite) TestNoSuccesses(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 1, time.Second, s.tm, nil)
-	c.Assert(err, IsNil)
-	fr.ObserveResponse(makeFailRequest(e))
-
-	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 1.0)
-}
-
-func (s *FailRateSuite) TestNoFailures(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 1, time.Second, s.tm, nil)
-	c.Assert(err, IsNil)
-	fr.ObserveResponse(makeOkRequest(e))
-
-	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 0.0)
+	c.Assert(fr.Ratio(), Equals, 0.0)
 }
 
 // Make sure that data is properly calculated over several buckets
 func (s *FailRateSuite) TestMultipleBuckets(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 3, time.Second, s.tm, nil)
+	fr, err := NewRatioCounter(3, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
 
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncB()
+	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
+	fr.IncA()
 
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
-
-	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
 
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, float64(2)/float64(3))
+	c.Assert(fr.Ratio(), Equals, float64(2)/float64(3))
 }
 
 // Make sure that data is properly calculated over several buckets
 // When we overwrite old data when the window is rolling
 func (s *FailRateSuite) TestOverwriteBuckets(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 3, time.Second, s.tm, nil)
+	fr, err := NewRatioCounter(3, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
 
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncB()
 
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
 
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
 
 	// This time we should overwrite the old data points
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
-	fr.ObserveResponse(makeOkRequest(e))
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncA()
+	fr.IncB()
+	fr.IncB()
 
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, float64(3)/float64(5))
+	c.Assert(fr.Ratio(), Equals, float64(3)/float64(5))
 }
 
 // Make sure we cleanup the data after periods of inactivity
 // So it does not mess up the stats
 func (s *FailRateSuite) TestInactiveBuckets(c *C) {
-	e := MustParseUrl("http://localhost:5000")
 
-	fr, err := NewRollingMeter(e, 3, time.Second, s.tm, nil)
+	fr, err := NewRatioCounter(3, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
 
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncB()
 
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
 
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
 
 	// This time we should overwrite the old data points with new data
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
-	fr.ObserveResponse(makeOkRequest(e))
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncA()
+	fr.IncB()
+	fr.IncB()
 
 	// Jump to the last bucket and change the data
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second * 2)
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncB()
 
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, float64(1)/float64(4))
+	c.Assert(fr.Ratio(), Equals, float64(1)/float64(4))
 }
 
 func (s *FailRateSuite) TestLongPeriodsOfInactivity(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 2, time.Second, s.tm, nil)
+	fr, err := NewRatioCounter(2, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
 
-	fr.ObserveResponse(makeOkRequest(e))
+	fr.IncB()
 
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
 
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 0.5)
+	c.Assert(fr.Ratio(), Equals, 0.5)
 
 	// This time we should overwrite all data points
 	s.tm.CurrentTime = s.tm.CurrentTime.Add(100 * time.Second)
-	fr.ObserveResponse(makeFailRequest(e))
-	c.Assert(fr.GetRate(), Equals, 1.0)
+	fr.IncA()
+	c.Assert(fr.Ratio(), Equals, 1.0)
 }
 
 func (s *FailRateSuite) TestReset(c *C) {
-	e := MustParseUrl("http://localhost:5000")
-
-	fr, err := NewRollingMeter(e, 1, time.Second, s.tm, nil)
+	fr, err := NewRatioCounter(1, time.Second, RatioClock(s.tm))
 	c.Assert(err, IsNil)
 
-	fr.ObserveResponse(makeOkRequest(e))
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncB()
+	fr.IncA()
 
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 0.5)
+	c.Assert(fr.Ratio(), Equals, 0.5)
 
 	// Reset the counter
 	fr.Reset()
 	c.Assert(fr.IsReady(), Equals, false)
 
 	// Now add some stats
-	fr.ObserveResponse(makeFailRequest(e))
-	fr.ObserveResponse(makeFailRequest(e))
+	fr.IncA()
+	fr.IncA()
 
 	// We are game again!
 	c.Assert(fr.IsReady(), Equals, true)
-	c.Assert(fr.GetRate(), Equals, 1.0)
-}
-
-func makeRequest(endpoint Endpoint, err error) Request {
-	return &BaseRequest{
-		Attempts: []Attempt{
-			&BaseAttempt{
-				Error:    err,
-				Endpoint: endpoint,
-			},
-		},
-	}
-}
-
-func makeFailRequest(endpoint Endpoint) (Request, Attempt) {
-	r := makeRequest(endpoint, fmt.Errorf("Oops"))
-	return r, r.GetLastAttempt()
-}
-
-func makeOkRequest(endpoint Endpoint) (Request, Attempt) {
-	r := makeRequest(endpoint, nil)
-	return r, r.GetLastAttempt()
+	c.Assert(fr.Ratio(), Equals, 1.0)
 }

memmetrics/roundtrip.go

@@ -1,11 +1,10 @@
 package memmetrics
 
 import (
+	"net"
 	"time"
 
-	"github.com/mailgun/log"
 	"github.com/mailgun/timetools"
-	"github.com/mailgun/vulcan/request"
 )
 
 // RoundTripMetrics provides aggregated performance metrics for HTTP requests processing
@@ -14,54 +13,72 @@ import (
 // are a rolling window histograms with defined precision as well.
 // See RoundTripOptions for more detail on parameters.
 type RoundTripMetrics struct {
-	o           *RoundTripOptions
 	total       *RollingCounter
 	netErrors   *RollingCounter
 	statusCodes map[int]*RollingCounter
-	histogram   RollingHistogram
+	histogram   *RollingHDRHistogram
+
+	newCounter NewCounterFn
+	newHist    NewRollingHistogramFn
+	clock      timetools.TimeProvider
 }
 
-type RoundTripOptions struct {
-	// CounterBuckets - how many buckets to allocate for rolling counter. Defaults to 10 buckets.
-	CounterBuckets int
-	// CounterResolution specifies the resolution for a single bucket
-	// (e.g. time.Second means that bucket will be counted for a second).
-	// defaults to time.Second
-	CounterResolution time.Duration
-	// HistMin - minimum non 0 value for a histogram (default 1)
-	HistMin int64
-	// HistMax - maximum value that can be recorded for a histogram (default 3,600,000,000)
-	HistMax int64
-	// HistSignificantFigures - defines precision for a value. e.g. 3 - 0.1%X precision, default is 2 - 1% precision for X
-	HistSignificantFigures int
-	// HistBuckets - how many sub histogram to keep in a rolling histogram, default is 6
-	HistBuckets int
-	// HistPeriod - rotation period for a histogram, default is 10 seconds
-	HistPeriod time.Duration
-	// TimeProvider - to provide time provider in tests, default is RealTime
-	TimeProvider timetools.TimeProvider
+type rrOptSetter func(r *RoundTripMetrics) error
+
+type NewCounterFn func() (*RollingCounter, error)
+type NewRollingHistogramFn func() (*RollingHDRHistogram, error)
+
+func RoundTripCounter(new NewCounterFn) rrOptSetter {
+	return func(r *RoundTripMetrics) error {
+		r.newCounter = new
+		return nil
+	}
+}
+
+func RoundTripHistogram(new NewRollingHistogramFn) rrOptSetter {
+	return func(r *RoundTripMetrics) error {
+		r.newHist = new
+		return nil
+	}
+}
+
+func RoundTripClock(clock timetools.TimeProvider) rrOptSetter {
+	return func(r *RoundTripMetrics) error {
+		r.clock = clock
+		return nil
+	}
 }
 
 // NewRoundTripMetrics returns new instance of metrics collector.
-func NewRoundTripMetrics(o RoundTripOptions) (*RoundTripMetrics, error) {
-	o = setDefaults(o)
-
-	h, err := NewRollingHistogram(
-		// this will create subhistograms
-		NewHDRHistogramFn(o.HistMin, o.HistMax, o.HistSignificantFigures),
-		// number of buckets in a rolling histogram
-		o.HistBuckets,
-		// rolling period for a histogram
-		o.HistPeriod,
-		o.TimeProvider)
-	if err != nil {
-		return nil, err
-	}
-
+func NewRoundTripMetrics(settings ...rrOptSetter) (*RoundTripMetrics, error) {
 	m := &RoundTripMetrics{
 		statusCodes: make(map[int]*RollingCounter),
-		histogram:   h,
-		o:           &o,
+	}
+	for _, s := range settings {
+		if err := s(m); err != nil {
+			return nil, err
+		}
+	}
+
+	if m.clock == nil {
+		m.clock = &timetools.RealTime{}
+	}
+
+	if m.newCounter == nil {
+		m.newCounter = func() (*RollingCounter, error) {
+			return NewCounter(counterBuckets, counterResolution, CounterClock(m.clock))
+		}
+	}
+
+	if m.newHist == nil {
+		m.newHist = func() (*RollingHDRHistogram, error) {
+			return NewRollingHDRHistogram(histMin, histMax, histSignificantFigures, histPeriod, histBuckets, RollingClock(m.clock))
+		}
+	}
+
+	h, err := m.newHist()
+	if err != nil {
+		return nil, err
 	}
 
 	netErrors, err := m.newCounter()
@@ -74,19 +91,15 @@ func NewRoundTripMetrics(o RoundTripOptions) (*RoundTripMetrics, error) {
 		return nil, err
 	}
 
+	m.histogram = h
 	m.netErrors = netErrors
 	m.total = total
 	return m, nil
 }
 
-// GetOptions returns settings used for this instance
-func (m *RoundTripMetrics) GetOptions() *RoundTripOptions {
-	return m.o
-}
-
 // 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) GetNetworkErrorRatio() float64 {
+func (m *RoundTripMetrics) NetworkErrorRatio() float64 {
 	if m.total.Count() == 0 {
 		return 0
 	}
@@ -94,7 +107,7 @@ func (m *RoundTripMetrics) GetNetworkErrorRatio() float64 {
 }
 
 // GetResponseCodeRatio calculates ratio of count(startA to endA) / count(startB to endB)
-func (m *RoundTripMetrics) GetResponseCodeRatio(startA, endA, startB, endB int) float64 {
+func (m *RoundTripMetrics) ResponseCodeRatio(startA, endA, startB, endB int) float64 {
 	a := int64(0)
 	b := int64(0)
 	for code, v := range m.statusCodes {
@@ -111,26 +124,27 @@ func (m *RoundTripMetrics) GetResponseCodeRatio(startA, endA, startB, endB int)
 	return 0
 }
 
-// RecordMetrics updates internal metrics collection based on the data from passed request.
-func (m *RoundTripMetrics) RecordMetrics(a request.Attempt) {
+func (m *RoundTripMetrics) Record(code int, err error, duration time.Duration) {
 	m.total.Inc()
-	m.recordNetError(a)
-	m.recordLatency(a)
-	m.recordStatusCode(a)
+	if _, ok := err.(net.Error); ok {
+		m.netErrors.Inc()
+	}
+	m.recordStatusCode(code)
+	m.recordLatency(duration)
 }
 
 // GetTotalCount returns total count of processed requests collected.
-func (m *RoundTripMetrics) GetTotalCount() int64 {
+func (m *RoundTripMetrics) TotalCount() int64 {
 	return m.total.Count()
 }
 
 // GetNetworkErrorCount returns total count of processed requests observed
-func (m *RoundTripMetrics) GetNetworkErrorCount() int64 {
+func (m *RoundTripMetrics) NetworkErrorCount() int64 {
 	return m.netErrors.Count()
 }
 
 // GetStatusCodesCounts returns map with counts of the response codes
-func (m *RoundTripMetrics) GetStatusCodesCounts() map[int]int64 {
+func (m *RoundTripMetrics) StatusCodesCounts() map[int]int64 {
 	sc := make(map[int]int64)
 	for k, v := range m.statusCodes {
 		if v.Count() != 0 {
@@ -141,7 +155,7 @@ func (m *RoundTripMetrics) GetStatusCodesCounts() map[int]int64 {
 }
 
 // GetLatencyHistogram computes and returns resulting histogram with latencies observed.
-func (m *RoundTripMetrics) GetLatencyHistogram() (Histogram, error) {
+func (m *RoundTripMetrics) LatencyHistogram() (*HDRHistogram, error) {
 	return m.histogram.Merged()
 }
 
@@ -152,38 +166,27 @@ func (m *RoundTripMetrics) Reset() {
 	m.statusCodes = make(map[int]*RollingCounter)
 }
 
-func (m *RoundTripMetrics) newCounter() (*RollingCounter, error) {
-	return NewRollingCounter(m.o.CounterBuckets, m.o.CounterResolution, m.o.TimeProvider)
+func (m *RoundTripMetrics) recordNetError() error {
+	m.netErrors.Inc()
+	return nil
 }
 
-func (m *RoundTripMetrics) recordNetError(a request.Attempt) {
-	if IsNetworkError(a) {
-		m.netErrors.Inc()
-	}
+func (m *RoundTripMetrics) recordLatency(d time.Duration) error {
+	return m.histogram.RecordLatencies(d, 1)
 }
 
-func (m *RoundTripMetrics) recordLatency(a request.Attempt) {
-	if err := m.histogram.RecordLatencies(a.GetDuration(), 1); err != nil {
-		log.Errorf("Failed to record latency: %v", err)
-	}
-}
-
-func (m *RoundTripMetrics) recordStatusCode(a request.Attempt) {
-	if a.GetResponse() == nil {
-		return
-	}
-	statusCode := a.GetResponse().StatusCode
+func (m *RoundTripMetrics) recordStatusCode(statusCode int) error {
 	if c, ok := m.statusCodes[statusCode]; ok {
 		c.Inc()
-		return
+		return nil
 	}
 	c, err := m.newCounter()
 	if err != nil {
-		log.Errorf("failed to create a counter: %v", err)
-		return
+		return err
 	}
 	c.Inc()
 	m.statusCodes[statusCode] = c
+	return nil
 }
 
 const (
@@ -195,31 +198,3 @@ const (
 	histBuckets            = 6                // number of sub-histograms in a rolling histogram
 	histPeriod             = 10 * time.Second // roll time
 )
-
-func setDefaults(o RoundTripOptions) RoundTripOptions {
-	if o.CounterBuckets == 0 {
-		o.CounterBuckets = counterBuckets
-	}
-	if o.CounterResolution == 0 {
-		o.CounterResolution = time.Second
-	}
-	if o.HistMin == 0 {
-		o.HistMin = histMin
-	}
-	if o.HistMax == 0 {
-		o.HistMax = histMax
-	}
-	if o.HistBuckets == 0 {
-		o.HistBuckets = histBuckets
-	}
-	if o.HistSignificantFigures == 0 {
-		o.HistSignificantFigures = histSignificantFigures
-	}
-	if o.HistPeriod == 0 {
-		o.HistPeriod = histPeriod
-	}
-	if o.TimeProvider == nil {
-		o.TimeProvider = &timetools.RealTime{}
-	}
-	return o
-}

memmetrics/roundtrip_test.go

@@ -1,12 +1,10 @@
 package memmetrics
 
 import (
-	"fmt"
-	"net/http"
+	"net"
 	"time"
 
 	"github.com/mailgun/timetools"
-	"github.com/mailgun/vulcan/request"
 	. "gopkg.in/check.v1"
 )
 
@@ -23,50 +21,34 @@ func (s *RRSuite) SetUpSuite(c *C) {
 }
 
 func (s *RRSuite) TestDefaults(c *C) {
-	rr, err := NewRoundTripMetrics(RoundTripOptions{TimeProvider: s.tm})
+	rr, err := NewRoundTripMetrics(RoundTripClock(s.tm))
 	c.Assert(err, IsNil)
 	c.Assert(rr, NotNil)
 
-	rr.RecordMetrics(makeAttempt(O{err: fmt.Errorf("o"), duration: time.Second}))
-	rr.RecordMetrics(makeAttempt(O{statusCode: 500, duration: 2 * time.Second}))
-	rr.RecordMetrics(makeAttempt(O{statusCode: 200, duration: time.Second}))
-	rr.RecordMetrics(makeAttempt(O{statusCode: 200, duration: time.Second}))
+	rr.Record(200, &net.OpError{}, time.Second)
+	rr.Record(500, nil, 2*time.Second)
+	rr.Record(200, nil, time.Second)
+	rr.Record(200, nil, time.Second)
 
-	c.Assert(rr.GetNetworkErrorCount(), Equals, int64(1))
-	c.Assert(rr.GetTotalCount(), Equals, int64(4))
-	c.Assert(rr.GetStatusCodesCounts(), DeepEquals, map[int]int64{500: 1, 200: 2})
-	c.Assert(rr.GetNetworkErrorRatio(), Equals, float64(1)/float64(4))
-	c.Assert(rr.GetResponseCodeRatio(500, 501, 200, 300), Equals, 0.5)
+	c.Assert(rr.NetworkErrorCount(), Equals, int64(1))
+	c.Assert(rr.TotalCount(), Equals, int64(4))
+	c.Assert(rr.StatusCodesCounts(), DeepEquals, map[int]int64{500: 1, 200: 3})
+	c.Assert(rr.NetworkErrorRatio(), Equals, float64(1)/float64(4))
+	c.Assert(rr.ResponseCodeRatio(500, 501, 200, 300), Equals, 1.0/3.0)
 
-	h, err := rr.GetLatencyHistogram()
+	h, err := rr.LatencyHistogram()
 	c.Assert(err, IsNil)
 	c.Assert(int(h.LatencyAtQuantile(100)/time.Second), Equals, 2)
 
 	rr.Reset()
-	c.Assert(rr.GetNetworkErrorCount(), Equals, int64(0))
-	c.Assert(rr.GetTotalCount(), Equals, int64(0))
-	c.Assert(rr.GetStatusCodesCounts(), DeepEquals, map[int]int64{})
-	c.Assert(rr.GetNetworkErrorRatio(), Equals, float64(0))
-	c.Assert(rr.GetResponseCodeRatio(500, 501, 200, 300), Equals, float64(0))
+	c.Assert(rr.NetworkErrorCount(), Equals, int64(0))
+	c.Assert(rr.TotalCount(), Equals, int64(0))
+	c.Assert(rr.StatusCodesCounts(), DeepEquals, map[int]int64{})
+	c.Assert(rr.NetworkErrorRatio(), Equals, float64(0))
+	c.Assert(rr.ResponseCodeRatio(500, 501, 200, 300), Equals, float64(0))
 
-	h, err = rr.GetLatencyHistogram()
+	h, err = rr.LatencyHistogram()
 	c.Assert(err, IsNil)
 	c.Assert(h.LatencyAtQuantile(100), Equals, time.Duration(0))
-}
 
-func makeAttempt(o O) *request.BaseAttempt {
-	a := &request.BaseAttempt{
-		Error:    o.err,
-		Duration: o.duration,
-	}
-	if o.statusCode != 0 {
-		a.Response = &http.Response{StatusCode: o.statusCode}
-	}
-	return a
-}
-
-type O struct {
-	statusCode int
-	err        error
-	duration   time.Duration
 }