added metrics

Change-Id: I7727bd43911339f48e2d096f8bb43729eb65b3ad

vendor/github.com/beorn7/perks/LICENSE

@@ -0,0 +1,20 @@
+Copyright (C) 2013 Blake Mizerany
+
+Permission is hereby granted, free of charge, to any person obtaining
+a copy of this software and associated documentation files (the
+"Software"), to deal in the Software without restriction, including
+without limitation the rights to use, copy, modify, merge, publish,
+distribute, sublicense, and/or sell copies of the Software, and to
+permit persons to whom the Software is furnished to do so, subject to
+the following conditions:
+
+The above copyright notice and this permission notice shall be
+included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

vendor/github.com/beorn7/perks/quantile/bench_test.go

@@ -0,0 +1,63 @@
+package quantile
+
+import (
+	"testing"
+)
+
+func BenchmarkInsertTargeted(b *testing.B) {
+	b.ReportAllocs()
+
+	s := NewTargeted(Targets)
+	b.ResetTimer()
+	for i := float64(0); i < float64(b.N); i++ {
+		s.Insert(i)
+	}
+}
+
+func BenchmarkInsertTargetedSmallEpsilon(b *testing.B) {
+	s := NewTargeted(TargetsSmallEpsilon)
+	b.ResetTimer()
+	for i := float64(0); i < float64(b.N); i++ {
+		s.Insert(i)
+	}
+}
+
+func BenchmarkInsertBiased(b *testing.B) {
+	s := NewLowBiased(0.01)
+	b.ResetTimer()
+	for i := float64(0); i < float64(b.N); i++ {
+		s.Insert(i)
+	}
+}
+
+func BenchmarkInsertBiasedSmallEpsilon(b *testing.B) {
+	s := NewLowBiased(0.0001)
+	b.ResetTimer()
+	for i := float64(0); i < float64(b.N); i++ {
+		s.Insert(i)
+	}
+}
+
+func BenchmarkQuery(b *testing.B) {
+	s := NewTargeted(Targets)
+	for i := float64(0); i < 1e6; i++ {
+		s.Insert(i)
+	}
+	b.ResetTimer()
+	n := float64(b.N)
+	for i := float64(0); i < n; i++ {
+		s.Query(i / n)
+	}
+}
+
+func BenchmarkQuerySmallEpsilon(b *testing.B) {
+	s := NewTargeted(TargetsSmallEpsilon)
+	for i := float64(0); i < 1e6; i++ {
+		s.Insert(i)
+	}
+	b.ResetTimer()
+	n := float64(b.N)
+	for i := float64(0); i < n; i++ {
+		s.Query(i / n)
+	}
+}

vendor/github.com/beorn7/perks/quantile/example_test.go

@@ -0,0 +1,121 @@
+// +build go1.1
+
+package quantile_test
+
+import (
+	"bufio"
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+	"time"
+
+	"github.com/beorn7/perks/quantile"
+)
+
+func Example_simple() {
+	ch := make(chan float64)
+	go sendFloats(ch)
+
+	// Compute the 50th, 90th, and 99th percentile.
+	q := quantile.NewTargeted(map[float64]float64{
+		0.50: 0.005,
+		0.90: 0.001,
+		0.99: 0.0001,
+	})
+	for v := range ch {
+		q.Insert(v)
+	}
+
+	fmt.Println("perc50:", q.Query(0.50))
+	fmt.Println("perc90:", q.Query(0.90))
+	fmt.Println("perc99:", q.Query(0.99))
+	fmt.Println("count:", q.Count())
+	// Output:
+	// perc50: 5
+	// perc90: 16
+	// perc99: 223
+	// count: 2388
+}
+
+func Example_mergeMultipleStreams() {
+	// Scenario:
+	// We have multiple database shards. On each shard, there is a process
+	// collecting query response times from the database logs and inserting
+	// them into a Stream (created via NewTargeted(0.90)), much like the
+	// Simple example. These processes expose a network interface for us to
+	// ask them to serialize and send us the results of their
+	// Stream.Samples so we may Merge and Query them.
+	//
+	// NOTES:
+	// * These sample sets are small, allowing us to get them
+	// across the network much faster than sending the entire list of data
+	// points.
+	//
+	// * For this to work correctly, we must supply the same quantiles
+	// a priori the process collecting the samples supplied to NewTargeted,
+	// even if we do not plan to query them all here.
+	ch := make(chan quantile.Samples)
+	getDBQuerySamples(ch)
+	q := quantile.NewTargeted(map[float64]float64{0.90: 0.001})
+	for samples := range ch {
+		q.Merge(samples)
+	}
+	fmt.Println("perc90:", q.Query(0.90))
+}
+
+func Example_window() {
+	// Scenario: We want the 90th, 95th, and 99th percentiles for each
+	// minute.
+
+	ch := make(chan float64)
+	go sendStreamValues(ch)
+
+	tick := time.NewTicker(1 * time.Minute)
+	q := quantile.NewTargeted(map[float64]float64{
+		0.90: 0.001,
+		0.95: 0.0005,
+		0.99: 0.0001,
+	})
+	for {
+		select {
+		case t := <-tick.C:
+			flushToDB(t, q.Samples())
+			q.Reset()
+		case v := <-ch:
+			q.Insert(v)
+		}
+	}
+}
+
+func sendStreamValues(ch chan float64) {
+	// Use your imagination
+}
+
+func flushToDB(t time.Time, samples quantile.Samples) {
+	// Use your imagination
+}
+
+// This is a stub for the above example. In reality this would hit the remote
+// servers via http or something like it.
+func getDBQuerySamples(ch chan quantile.Samples) {}
+
+func sendFloats(ch chan<- float64) {
+	f, err := os.Open("exampledata.txt")
+	if err != nil {
+		log.Fatal(err)
+	}
+	sc := bufio.NewScanner(f)
+	for sc.Scan() {
+		b := sc.Bytes()
+		v, err := strconv.ParseFloat(string(b), 64)
+		if err != nil {
+			log.Fatal(err)
+		}
+		ch <- v
+	}
+	if sc.Err() != nil {
+		log.Fatal(sc.Err())
+	}
+	close(ch)
+}

vendor/github.com/beorn7/perks/quantile/stream.go

@@ -0,0 +1,292 @@
+// Package quantile computes approximate quantiles over an unbounded data
+// stream within low memory and CPU bounds.
+//
+// A small amount of accuracy is traded to achieve the above properties.
+//
+// Multiple streams can be merged before calling Query to generate a single set
+// of results. This is meaningful when the streams represent the same type of
+// data. See Merge and Samples.
+//
+// For more detailed information about the algorithm used, see:
+//
+// Effective Computation of Biased Quantiles over Data Streams
+//
+// http://www.cs.rutgers.edu/~muthu/bquant.pdf
+package quantile
+
+import (
+	"math"
+	"sort"
+)
+
+// Sample holds an observed value and meta information for compression. JSON
+// tags have been added for convenience.
+type Sample struct {
+	Value float64 `json:",string"`
+	Width float64 `json:",string"`
+	Delta float64 `json:",string"`
+}
+
+// Samples represents a slice of samples. It implements sort.Interface.
+type Samples []Sample
+
+func (a Samples) Len() int           { return len(a) }
+func (a Samples) Less(i, j int) bool { return a[i].Value < a[j].Value }
+func (a Samples) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
+
+type invariant func(s *stream, r float64) float64
+
+// NewLowBiased returns an initialized Stream for low-biased quantiles
+// (e.g. 0.01, 0.1, 0.5) where the needed quantiles are not known a priori, but
+// error guarantees can still be given even for the lower ranks of the data
+// distribution.
+//
+// The provided epsilon is a relative error, i.e. the true quantile of a value
+// returned by a query is guaranteed to be within (1±Epsilon)*Quantile.
+//
+// See http://www.cs.rutgers.edu/~muthu/bquant.pdf for time, space, and error
+// properties.
+func NewLowBiased(epsilon float64) *Stream {
+	ƒ := func(s *stream, r float64) float64 {
+		return 2 * epsilon * r
+	}
+	return newStream(ƒ)
+}
+
+// NewHighBiased returns an initialized Stream for high-biased quantiles
+// (e.g. 0.01, 0.1, 0.5) where the needed quantiles are not known a priori, but
+// error guarantees can still be given even for the higher ranks of the data
+// distribution.
+//
+// The provided epsilon is a relative error, i.e. the true quantile of a value
+// returned by a query is guaranteed to be within 1-(1±Epsilon)*(1-Quantile).
+//
+// See http://www.cs.rutgers.edu/~muthu/bquant.pdf for time, space, and error
+// properties.
+func NewHighBiased(epsilon float64) *Stream {
+	ƒ := func(s *stream, r float64) float64 {
+		return 2 * epsilon * (s.n - r)
+	}
+	return newStream(ƒ)
+}
+
+// NewTargeted returns an initialized Stream concerned with a particular set of
+// quantile values that are supplied a priori. Knowing these a priori reduces
+// space and computation time. The targets map maps the desired quantiles to
+// their absolute errors, i.e. the true quantile of a value returned by a query
+// is guaranteed to be within (Quantile±Epsilon).
+//
+// See http://www.cs.rutgers.edu/~muthu/bquant.pdf for time, space, and error properties.
+func NewTargeted(targets map[float64]float64) *Stream {
+	ƒ := func(s *stream, r float64) float64 {
+		var m = math.MaxFloat64
+		var f float64
+		for quantile, epsilon := range targets {
+			if quantile*s.n <= r {
+				f = (2 * epsilon * r) / quantile
+			} else {
+				f = (2 * epsilon * (s.n - r)) / (1 - quantile)
+			}
+			if f < m {
+				m = f
+			}
+		}
+		return m
+	}
+	return newStream(ƒ)
+}
+
+// Stream computes quantiles for a stream of float64s. It is not thread-safe by
+// design. Take care when using across multiple goroutines.
+type Stream struct {
+	*stream
+	b      Samples
+	sorted bool
+}
+
+func newStream(ƒ invariant) *Stream {
+	x := &stream{ƒ: ƒ}
+	return &Stream{x, make(Samples, 0, 500), true}
+}
+
+// Insert inserts v into the stream.
+func (s *Stream) Insert(v float64) {
+	s.insert(Sample{Value: v, Width: 1})
+}
+
+func (s *Stream) insert(sample Sample) {
+	s.b = append(s.b, sample)
+	s.sorted = false
+	if len(s.b) == cap(s.b) {
+		s.flush()
+	}
+}
+
+// Query returns the computed qth percentiles value. If s was created with
+// NewTargeted, and q is not in the set of quantiles provided a priori, Query
+// will return an unspecified result.
+func (s *Stream) Query(q float64) float64 {
+	if !s.flushed() {
+		// Fast path when there hasn't been enough data for a flush;
+		// this also yields better accuracy for small sets of data.
+		l := len(s.b)
+		if l == 0 {
+			return 0
+		}
+		i := int(float64(l) * q)
+		if i > 0 {
+			i -= 1
+		}
+		s.maybeSort()
+		return s.b[i].Value
+	}
+	s.flush()
+	return s.stream.query(q)
+}
+
+// Merge merges samples into the underlying streams samples. This is handy when
+// merging multiple streams from separate threads, database shards, etc.
+//
+// ATTENTION: This method is broken and does not yield correct results. The
+// underlying algorithm is not capable of merging streams correctly.
+func (s *Stream) Merge(samples Samples) {
+	sort.Sort(samples)
+	s.stream.merge(samples)
+}
+
+// Reset reinitializes and clears the list reusing the samples buffer memory.
+func (s *Stream) Reset() {
+	s.stream.reset()
+	s.b = s.b[:0]
+}
+
+// Samples returns stream samples held by s.
+func (s *Stream) Samples() Samples {
+	if !s.flushed() {
+		return s.b
+	}
+	s.flush()
+	return s.stream.samples()
+}
+
+// Count returns the total number of samples observed in the stream
+// since initialization.
+func (s *Stream) Count() int {
+	return len(s.b) + s.stream.count()
+}
+
+func (s *Stream) flush() {
+	s.maybeSort()
+	s.stream.merge(s.b)
+	s.b = s.b[:0]
+}
+
+func (s *Stream) maybeSort() {
+	if !s.sorted {
+		s.sorted = true
+		sort.Sort(s.b)
+	}
+}
+
+func (s *Stream) flushed() bool {
+	return len(s.stream.l) > 0
+}
+
+type stream struct {
+	n float64
+	l []Sample
+	ƒ invariant
+}
+
+func (s *stream) reset() {
+	s.l = s.l[:0]
+	s.n = 0
+}
+
+func (s *stream) insert(v float64) {
+	s.merge(Samples{{v, 1, 0}})
+}
+
+func (s *stream) merge(samples Samples) {
+	// TODO(beorn7): This tries to merge not only individual samples, but
+	// whole summaries. The paper doesn't mention merging summaries at
+	// all. Unittests show that the merging is inaccurate. Find out how to
+	// do merges properly.
+	var r float64
+	i := 0
+	for _, sample := range samples {
+		for ; i < len(s.l); i++ {
+			c := s.l[i]
+			if c.Value > sample.Value {
+				// Insert at position i.
+				s.l = append(s.l, Sample{})
+				copy(s.l[i+1:], s.l[i:])
+				s.l[i] = Sample{
+					sample.Value,
+					sample.Width,
+					math.Max(sample.Delta, math.Floor(s.ƒ(s, r))-1),
+					// TODO(beorn7): How to calculate delta correctly?
+				}
+				i++
+				goto inserted
+			}
+			r += c.Width
+		}
+		s.l = append(s.l, Sample{sample.Value, sample.Width, 0})
+		i++
+	inserted:
+		s.n += sample.Width
+		r += sample.Width
+	}
+	s.compress()
+}
+
+func (s *stream) count() int {
+	return int(s.n)
+}
+
+func (s *stream) query(q float64) float64 {
+	t := math.Ceil(q * s.n)
+	t += math.Ceil(s.ƒ(s, t) / 2)
+	p := s.l[0]
+	var r float64
+	for _, c := range s.l[1:] {
+		r += p.Width
+		if r+c.Width+c.Delta > t {
+			return p.Value
+		}
+		p = c
+	}
+	return p.Value
+}
+
+func (s *stream) compress() {
+	if len(s.l) < 2 {
+		return
+	}
+	x := s.l[len(s.l)-1]
+	xi := len(s.l) - 1
+	r := s.n - 1 - x.Width
+
+	for i := len(s.l) - 2; i >= 0; i-- {
+		c := s.l[i]
+		if c.Width+x.Width+x.Delta <= s.ƒ(s, r) {
+			x.Width += c.Width
+			s.l[xi] = x
+			// Remove element at i.
+			copy(s.l[i:], s.l[i+1:])
+			s.l = s.l[:len(s.l)-1]
+			xi -= 1
+		} else {
+			x = c
+			xi = i
+		}
+		r -= c.Width
+	}
+}
+
+func (s *stream) samples() Samples {
+	samples := make(Samples, len(s.l))
+	copy(samples, s.l)
+	return samples
+}

vendor/github.com/beorn7/perks/quantile/stream_test.go

@@ -0,0 +1,188 @@
+package quantile
+
+import (
+	"math"
+	"math/rand"
+	"sort"
+	"testing"
+)
+
+var (
+	Targets = map[float64]float64{
+		0.01: 0.001,
+		0.10: 0.01,
+		0.50: 0.05,
+		0.90: 0.01,
+		0.99: 0.001,
+	}
+	TargetsSmallEpsilon = map[float64]float64{
+		0.01: 0.0001,
+		0.10: 0.001,
+		0.50: 0.005,
+		0.90: 0.001,
+		0.99: 0.0001,
+	}
+	LowQuantiles  = []float64{0.01, 0.1, 0.5}
+	HighQuantiles = []float64{0.99, 0.9, 0.5}
+)
+
+const RelativeEpsilon = 0.01
+
+func verifyPercsWithAbsoluteEpsilon(t *testing.T, a []float64, s *Stream) {
+	sort.Float64s(a)
+	for quantile, epsilon := range Targets {
+		n := float64(len(a))
+		k := int(quantile * n)
+		lower := int((quantile - epsilon) * n)
+		if lower < 1 {
+			lower = 1
+		}
+		upper := int(math.Ceil((quantile + epsilon) * n))
+		if upper > len(a) {
+			upper = len(a)
+		}
+		w, min, max := a[k-1], a[lower-1], a[upper-1]
+		if g := s.Query(quantile); g < min || g > max {
+			t.Errorf("q=%f: want %v [%f,%f], got %v", quantile, w, min, max, g)
+		}
+	}
+}
+
+func verifyLowPercsWithRelativeEpsilon(t *testing.T, a []float64, s *Stream) {
+	sort.Float64s(a)
+	for _, qu := range LowQuantiles {
+		n := float64(len(a))
+		k := int(qu * n)
+
+		lowerRank := int((1 - RelativeEpsilon) * qu * n)
+		upperRank := int(math.Ceil((1 + RelativeEpsilon) * qu * n))
+		w, min, max := a[k-1], a[lowerRank-1], a[upperRank-1]
+		if g := s.Query(qu); g < min || g > max {
+			t.Errorf("q=%f: want %v [%f,%f], got %v", qu, w, min, max, g)
+		}
+	}
+}
+
+func verifyHighPercsWithRelativeEpsilon(t *testing.T, a []float64, s *Stream) {
+	sort.Float64s(a)
+	for _, qu := range HighQuantiles {
+		n := float64(len(a))
+		k := int(qu * n)
+
+		lowerRank := int((1 - (1+RelativeEpsilon)*(1-qu)) * n)
+		upperRank := int(math.Ceil((1 - (1-RelativeEpsilon)*(1-qu)) * n))
+		w, min, max := a[k-1], a[lowerRank-1], a[upperRank-1]
+		if g := s.Query(qu); g < min || g > max {
+			t.Errorf("q=%f: want %v [%f,%f], got %v", qu, w, min, max, g)
+		}
+	}
+}
+
+func populateStream(s *Stream) []float64 {
+	a := make([]float64, 0, 1e5+100)
+	for i := 0; i < cap(a); i++ {
+		v := rand.NormFloat64()
+		// Add 5% asymmetric outliers.
+		if i%20 == 0 {
+			v = v*v + 1
+		}
+		s.Insert(v)
+		a = append(a, v)
+	}
+	return a
+}
+
+func TestTargetedQuery(t *testing.T) {
+	rand.Seed(42)
+	s := NewTargeted(Targets)
+	a := populateStream(s)
+	verifyPercsWithAbsoluteEpsilon(t, a, s)
+}
+
+func TestLowBiasedQuery(t *testing.T) {
+	rand.Seed(42)
+	s := NewLowBiased(RelativeEpsilon)
+	a := populateStream(s)
+	verifyLowPercsWithRelativeEpsilon(t, a, s)
+}
+
+func TestHighBiasedQuery(t *testing.T) {
+	rand.Seed(42)
+	s := NewHighBiased(RelativeEpsilon)
+	a := populateStream(s)
+	verifyHighPercsWithRelativeEpsilon(t, a, s)
+}
+
+// BrokenTestTargetedMerge is broken, see Merge doc comment.
+func BrokenTestTargetedMerge(t *testing.T) {
+	rand.Seed(42)
+	s1 := NewTargeted(Targets)
+	s2 := NewTargeted(Targets)
+	a := populateStream(s1)
+	a = append(a, populateStream(s2)...)
+	s1.Merge(s2.Samples())
+	verifyPercsWithAbsoluteEpsilon(t, a, s1)
+}
+
+// BrokenTestLowBiasedMerge is broken, see Merge doc comment.
+func BrokenTestLowBiasedMerge(t *testing.T) {
+	rand.Seed(42)
+	s1 := NewLowBiased(RelativeEpsilon)
+	s2 := NewLowBiased(RelativeEpsilon)
+	a := populateStream(s1)
+	a = append(a, populateStream(s2)...)
+	s1.Merge(s2.Samples())
+	verifyLowPercsWithRelativeEpsilon(t, a, s2)
+}
+
+// BrokenTestHighBiasedMerge is broken, see Merge doc comment.
+func BrokenTestHighBiasedMerge(t *testing.T) {
+	rand.Seed(42)
+	s1 := NewHighBiased(RelativeEpsilon)
+	s2 := NewHighBiased(RelativeEpsilon)
+	a := populateStream(s1)
+	a = append(a, populateStream(s2)...)
+	s1.Merge(s2.Samples())
+	verifyHighPercsWithRelativeEpsilon(t, a, s2)
+}
+
+func TestUncompressed(t *testing.T) {
+	q := NewTargeted(Targets)
+	for i := 100; i > 0; i-- {
+		q.Insert(float64(i))
+	}
+	if g := q.Count(); g != 100 {
+		t.Errorf("want count 100, got %d", g)
+	}
+	// Before compression, Query should have 100% accuracy.
+	for quantile := range Targets {
+		w := quantile * 100
+		if g := q.Query(quantile); g != w {
+			t.Errorf("want %f, got %f", w, g)
+		}
+	}
+}
+
+func TestUncompressedSamples(t *testing.T) {
+	q := NewTargeted(map[float64]float64{0.99: 0.001})
+	for i := 1; i <= 100; i++ {
+		q.Insert(float64(i))
+	}
+	if g := q.Samples().Len(); g != 100 {
+		t.Errorf("want count 100, got %d", g)
+	}
+}
+
+func TestUncompressedOne(t *testing.T) {
+	q := NewTargeted(map[float64]float64{0.99: 0.01})
+	q.Insert(3.14)
+	if g := q.Query(0.90); g != 3.14 {
+		t.Error("want PI, got", g)
+	}
+}
+
+func TestDefaults(t *testing.T) {
+	if g := NewTargeted(map[float64]float64{0.99: 0.001}).Query(0.99); g != 0 {
+		t.Errorf("want 0, got %f", g)
+	}
+}