sample.go

Documentation: github.com/rcrowley/go-metrics

     1  package metrics
     2  
     3  import (
     4  	"math"
     5  	"math/rand"
     6  	"sort"
     7  	"sync"
     8  	"time"
     9  )
    10  
    11  const rescaleThreshold = time.Hour
    12  
    13  // Samples maintain a statistically-significant selection of values from
    14  // a stream.
    15  type Sample interface {
    16  	Clear()
    17  	Count() int64
    18  	Max() int64
    19  	Mean() float64
    20  	Min() int64
    21  	Percentile(float64) float64
    22  	Percentiles([]float64) []float64
    23  	Size() int
    24  	Snapshot() Sample
    25  	StdDev() float64
    26  	Sum() int64
    27  	Update(int64)
    28  	Values() []int64
    29  	Variance() float64
    30  }
    31  
    32  // ExpDecaySample is an exponentially-decaying sample using a forward-decaying
    33  // priority reservoir.  See Cormode et al's "Forward Decay: A Practical Time
    34  // Decay Model for Streaming Systems".
    35  //
    36  // <http://dimacs.rutgers.edu/~graham/pubs/papers/fwddecay.pdf>
    37  type ExpDecaySample struct {
    38  	alpha         float64
    39  	count         int64
    40  	mutex         sync.Mutex
    41  	reservoirSize int
    42  	t0, t1        time.Time
    43  	values        *expDecaySampleHeap
    44  }
    45  
    46  // NewExpDecaySample constructs a new exponentially-decaying sample with the
    47  // given reservoir size and alpha.
    48  func NewExpDecaySample(reservoirSize int, alpha float64) Sample {
    49  	if UseNilMetrics {
    50  		return NilSample{}
    51  	}
    52  	s := &ExpDecaySample{
    53  		alpha:         alpha,
    54  		reservoirSize: reservoirSize,
    55  		t0:            time.Now(),
    56  		values:        newExpDecaySampleHeap(reservoirSize),
    57  	}
    58  	s.t1 = s.t0.Add(rescaleThreshold)
    59  	return s
    60  }
    61  
    62  // Clear clears all samples.
    63  func (s *ExpDecaySample) Clear() {
    64  	s.mutex.Lock()
    65  	defer s.mutex.Unlock()
    66  	s.count = 0
    67  	s.t0 = time.Now()
    68  	s.t1 = s.t0.Add(rescaleThreshold)
    69  	s.values.Clear()
    70  }
    71  
    72  // Count returns the number of samples recorded, which may exceed the
    73  // reservoir size.
    74  func (s *ExpDecaySample) Count() int64 {
    75  	s.mutex.Lock()
    76  	defer s.mutex.Unlock()
    77  	return s.count
    78  }
    79  
    80  // Max returns the maximum value in the sample, which may not be the maximum
    81  // value ever to be part of the sample.
    82  func (s *ExpDecaySample) Max() int64 {
    83  	return SampleMax(s.Values())
    84  }
    85  
    86  // Mean returns the mean of the values in the sample.
    87  func (s *ExpDecaySample) Mean() float64 {
    88  	return SampleMean(s.Values())
    89  }
    90  
    91  // Min returns the minimum value in the sample, which may not be the minimum
    92  // value ever to be part of the sample.
    93  func (s *ExpDecaySample) Min() int64 {
    94  	return SampleMin(s.Values())
    95  }
    96  
    97  // Percentile returns an arbitrary percentile of values in the sample.
    98  func (s *ExpDecaySample) Percentile(p float64) float64 {
    99  	return SamplePercentile(s.Values(), p)
   100  }
   101  
   102  // Percentiles returns a slice of arbitrary percentiles of values in the
   103  // sample.
   104  func (s *ExpDecaySample) Percentiles(ps []float64) []float64 {
   105  	return SamplePercentiles(s.Values(), ps)
   106  }
   107  
   108  // Size returns the size of the sample, which is at most the reservoir size.
   109  func (s *ExpDecaySample) Size() int {
   110  	s.mutex.Lock()
   111  	defer s.mutex.Unlock()
   112  	return s.values.Size()
   113  }
   114  
   115  // Snapshot returns a read-only copy of the sample.
   116  func (s *ExpDecaySample) Snapshot() Sample {
   117  	s.mutex.Lock()
   118  	defer s.mutex.Unlock()
   119  	vals := s.values.Values()
   120  	values := make([]int64, len(vals))
   121  	for i, v := range vals {
   122  		values[i] = v.v
   123  	}
   124  	return &SampleSnapshot{
   125  		count:  s.count,
   126  		values: values,
   127  	}
   128  }
   129  
   130  // StdDev returns the standard deviation of the values in the sample.
   131  func (s *ExpDecaySample) StdDev() float64 {
   132  	return SampleStdDev(s.Values())
   133  }
   134  
   135  // Sum returns the sum of the values in the sample.
   136  func (s *ExpDecaySample) Sum() int64 {
   137  	return SampleSum(s.Values())
   138  }
   139  
   140  // Update samples a new value.
   141  func (s *ExpDecaySample) Update(v int64) {
   142  	s.update(time.Now(), v)
   143  }
   144  
   145  // Values returns a copy of the values in the sample.
   146  func (s *ExpDecaySample) Values() []int64 {
   147  	s.mutex.Lock()
   148  	defer s.mutex.Unlock()
   149  	vals := s.values.Values()
   150  	values := make([]int64, len(vals))
   151  	for i, v := range vals {
   152  		values[i] = v.v
   153  	}
   154  	return values
   155  }
   156  
   157  // Variance returns the variance of the values in the sample.
   158  func (s *ExpDecaySample) Variance() float64 {
   159  	return SampleVariance(s.Values())
   160  }
   161  
   162  // update samples a new value at a particular timestamp.  This is a method all
   163  // its own to facilitate testing.
   164  func (s *ExpDecaySample) update(t time.Time, v int64) {
   165  	s.mutex.Lock()
   166  	defer s.mutex.Unlock()
   167  	s.count++
   168  	if s.values.Size() == s.reservoirSize {
   169  		s.values.Pop()
   170  	}
   171  	s.values.Push(expDecaySample{
   172  		k: math.Exp(t.Sub(s.t0).Seconds()*s.alpha) / rand.Float64(),
   173  		v: v,
   174  	})
   175  	if t.After(s.t1) {
   176  		values := s.values.Values()
   177  		t0 := s.t0
   178  		s.values.Clear()
   179  		s.t0 = t
   180  		s.t1 = s.t0.Add(rescaleThreshold)
   181  		for _, v := range values {
   182  			v.k = v.k * math.Exp(-s.alpha*s.t0.Sub(t0).Seconds())
   183  			s.values.Push(v)
   184  		}
   185  	}
   186  }
   187  
   188  // NilSample is a no-op Sample.
   189  type NilSample struct{}
   190  
   191  // Clear is a no-op.
   192  func (NilSample) Clear() {}
   193  
   194  // Count is a no-op.
   195  func (NilSample) Count() int64 { return 0 }
   196  
   197  // Max is a no-op.
   198  func (NilSample) Max() int64 { return 0 }
   199  
   200  // Mean is a no-op.
   201  func (NilSample) Mean() float64 { return 0.0 }
   202  
   203  // Min is a no-op.
   204  func (NilSample) Min() int64 { return 0 }
   205  
   206  // Percentile is a no-op.
   207  func (NilSample) Percentile(p float64) float64 { return 0.0 }
   208  
   209  // Percentiles is a no-op.
   210  func (NilSample) Percentiles(ps []float64) []float64 {
   211  	return make([]float64, len(ps))
   212  }
   213  
   214  // Size is a no-op.
   215  func (NilSample) Size() int { return 0 }
   216  
   217  // Sample is a no-op.
   218  func (NilSample) Snapshot() Sample { return NilSample{} }
   219  
   220  // StdDev is a no-op.
   221  func (NilSample) StdDev() float64 { return 0.0 }
   222  
   223  // Sum is a no-op.
   224  func (NilSample) Sum() int64 { return 0 }
   225  
   226  // Update is a no-op.
   227  func (NilSample) Update(v int64) {}
   228  
   229  // Values is a no-op.
   230  func (NilSample) Values() []int64 { return []int64{} }
   231  
   232  // Variance is a no-op.
   233  func (NilSample) Variance() float64 { return 0.0 }
   234  
   235  // SampleMax returns the maximum value of the slice of int64.
   236  func SampleMax(values []int64) int64 {
   237  	if 0 == len(values) {
   238  		return 0
   239  	}
   240  	var max int64 = math.MinInt64
   241  	for _, v := range values {
   242  		if max < v {
   243  			max = v
   244  		}
   245  	}
   246  	return max
   247  }
   248  
   249  // SampleMean returns the mean value of the slice of int64.
   250  func SampleMean(values []int64) float64 {
   251  	if 0 == len(values) {
   252  		return 0.0
   253  	}
   254  	return float64(SampleSum(values)) / float64(len(values))
   255  }
   256  
   257  // SampleMin returns the minimum value of the slice of int64.
   258  func SampleMin(values []int64) int64 {
   259  	if 0 == len(values) {
   260  		return 0
   261  	}
   262  	var min int64 = math.MaxInt64
   263  	for _, v := range values {
   264  		if min > v {
   265  			min = v
   266  		}
   267  	}
   268  	return min
   269  }
   270  
   271  // SamplePercentiles returns an arbitrary percentile of the slice of int64.
   272  func SamplePercentile(values int64Slice, p float64) float64 {
   273  	return SamplePercentiles(values, []float64{p})[0]
   274  }
   275  
   276  // SamplePercentiles returns a slice of arbitrary percentiles of the slice of
   277  // int64.
   278  func SamplePercentiles(values int64Slice, ps []float64) []float64 {
   279  	scores := make([]float64, len(ps))
   280  	size := len(values)
   281  	if size > 0 {
   282  		sort.Sort(values)
   283  		for i, p := range ps {
   284  			pos := p * float64(size+1)
   285  			if pos < 1.0 {
   286  				scores[i] = float64(values[0])
   287  			} else if pos >= float64(size) {
   288  				scores[i] = float64(values[size-1])
   289  			} else {
   290  				lower := float64(values[int(pos)-1])
   291  				upper := float64(values[int(pos)])
   292  				scores[i] = lower + (pos-math.Floor(pos))*(upper-lower)
   293  			}
   294  		}
   295  	}
   296  	return scores
   297  }
   298  
   299  // SampleSnapshot is a read-only copy of another Sample.
   300  type SampleSnapshot struct {
   301  	count  int64
   302  	values []int64
   303  }
   304  
   305  func NewSampleSnapshot(count int64, values []int64) *SampleSnapshot {
   306  	return &SampleSnapshot{
   307  		count:  count,
   308  		values: values,
   309  	}
   310  }
   311  
   312  // Clear panics.
   313  func (*SampleSnapshot) Clear() {
   314  	panic("Clear called on a SampleSnapshot")
   315  }
   316  
   317  // Count returns the count of inputs at the time the snapshot was taken.
   318  func (s *SampleSnapshot) Count() int64 { return s.count }
   319  
   320  // Max returns the maximal value at the time the snapshot was taken.
   321  func (s *SampleSnapshot) Max() int64 { return SampleMax(s.values) }
   322  
   323  // Mean returns the mean value at the time the snapshot was taken.
   324  func (s *SampleSnapshot) Mean() float64 { return SampleMean(s.values) }
   325  
   326  // Min returns the minimal value at the time the snapshot was taken.
   327  func (s *SampleSnapshot) Min() int64 { return SampleMin(s.values) }
   328  
   329  // Percentile returns an arbitrary percentile of values at the time the
   330  // snapshot was taken.
   331  func (s *SampleSnapshot) Percentile(p float64) float64 {
   332  	return SamplePercentile(s.values, p)
   333  }
   334  
   335  // Percentiles returns a slice of arbitrary percentiles of values at the time
   336  // the snapshot was taken.
   337  func (s *SampleSnapshot) Percentiles(ps []float64) []float64 {
   338  	return SamplePercentiles(s.values, ps)
   339  }
   340  
   341  // Size returns the size of the sample at the time the snapshot was taken.
   342  func (s *SampleSnapshot) Size() int { return len(s.values) }
   343  
   344  // Snapshot returns the snapshot.
   345  func (s *SampleSnapshot) Snapshot() Sample { return s }
   346  
   347  // StdDev returns the standard deviation of values at the time the snapshot was
   348  // taken.
   349  func (s *SampleSnapshot) StdDev() float64 { return SampleStdDev(s.values) }
   350  
   351  // Sum returns the sum of values at the time the snapshot was taken.
   352  func (s *SampleSnapshot) Sum() int64 { return SampleSum(s.values) }
   353  
   354  // Update panics.
   355  func (*SampleSnapshot) Update(int64) {
   356  	panic("Update called on a SampleSnapshot")
   357  }
   358  
   359  // Values returns a copy of the values in the sample.
   360  func (s *SampleSnapshot) Values() []int64 {
   361  	values := make([]int64, len(s.values))
   362  	copy(values, s.values)
   363  	return values
   364  }
   365  
   366  // Variance returns the variance of values at the time the snapshot was taken.
   367  func (s *SampleSnapshot) Variance() float64 { return SampleVariance(s.values) }
   368  
   369  // SampleStdDev returns the standard deviation of the slice of int64.
   370  func SampleStdDev(values []int64) float64 {
   371  	return math.Sqrt(SampleVariance(values))
   372  }
   373  
   374  // SampleSum returns the sum of the slice of int64.
   375  func SampleSum(values []int64) int64 {
   376  	var sum int64
   377  	for _, v := range values {
   378  		sum += v
   379  	}
   380  	return sum
   381  }
   382  
   383  // SampleVariance returns the variance of the slice of int64.
   384  func SampleVariance(values []int64) float64 {
   385  	if 0 == len(values) {
   386  		return 0.0
   387  	}
   388  	m := SampleMean(values)
   389  	var sum float64
   390  	for _, v := range values {
   391  		d := float64(v) - m
   392  		sum += d * d
   393  	}
   394  	return sum / float64(len(values))
   395  }
   396  
   397  // A uniform sample using Vitter's Algorithm R.
   398  //
   399  // <http://www.cs.umd.edu/~samir/498/vitter.pdf>
   400  type UniformSample struct {
   401  	count         int64
   402  	mutex         sync.Mutex
   403  	reservoirSize int
   404  	values        []int64
   405  }
   406  
   407  // NewUniformSample constructs a new uniform sample with the given reservoir
   408  // size.
   409  func NewUniformSample(reservoirSize int) Sample {
   410  	if UseNilMetrics {
   411  		return NilSample{}
   412  	}
   413  	return &UniformSample{
   414  		reservoirSize: reservoirSize,
   415  		values:        make([]int64, 0, reservoirSize),
   416  	}
   417  }
   418  
   419  // Clear clears all samples.
   420  func (s *UniformSample) Clear() {
   421  	s.mutex.Lock()
   422  	defer s.mutex.Unlock()
   423  	s.count = 0
   424  	s.values = make([]int64, 0, s.reservoirSize)
   425  }
   426  
   427  // Count returns the number of samples recorded, which may exceed the
   428  // reservoir size.
   429  func (s *UniformSample) Count() int64 {
   430  	s.mutex.Lock()
   431  	defer s.mutex.Unlock()
   432  	return s.count
   433  }
   434  
   435  // Max returns the maximum value in the sample, which may not be the maximum
   436  // value ever to be part of the sample.
   437  func (s *UniformSample) Max() int64 {
   438  	s.mutex.Lock()
   439  	defer s.mutex.Unlock()
   440  	return SampleMax(s.values)
   441  }
   442  
   443  // Mean returns the mean of the values in the sample.
   444  func (s *UniformSample) Mean() float64 {
   445  	s.mutex.Lock()
   446  	defer s.mutex.Unlock()
   447  	return SampleMean(s.values)
   448  }
   449  
   450  // Min returns the minimum value in the sample, which may not be the minimum
   451  // value ever to be part of the sample.
   452  func (s *UniformSample) Min() int64 {
   453  	s.mutex.Lock()
   454  	defer s.mutex.Unlock()
   455  	return SampleMin(s.values)
   456  }
   457  
   458  // Percentile returns an arbitrary percentile of values in the sample.
   459  func (s *UniformSample) Percentile(p float64) float64 {
   460  	s.mutex.Lock()
   461  	defer s.mutex.Unlock()
   462  	return SamplePercentile(s.values, p)
   463  }
   464  
   465  // Percentiles returns a slice of arbitrary percentiles of values in the
   466  // sample.
   467  func (s *UniformSample) Percentiles(ps []float64) []float64 {
   468  	s.mutex.Lock()
   469  	defer s.mutex.Unlock()
   470  	return SamplePercentiles(s.values, ps)
   471  }
   472  
   473  // Size returns the size of the sample, which is at most the reservoir size.
   474  func (s *UniformSample) Size() int {
   475  	s.mutex.Lock()
   476  	defer s.mutex.Unlock()
   477  	return len(s.values)
   478  }
   479  
   480  // Snapshot returns a read-only copy of the sample.
   481  func (s *UniformSample) Snapshot() Sample {
   482  	s.mutex.Lock()
   483  	defer s.mutex.Unlock()
   484  	values := make([]int64, len(s.values))
   485  	copy(values, s.values)
   486  	return &SampleSnapshot{
   487  		count:  s.count,
   488  		values: values,
   489  	}
   490  }
   491  
   492  // StdDev returns the standard deviation of the values in the sample.
   493  func (s *UniformSample) StdDev() float64 {
   494  	s.mutex.Lock()
   495  	defer s.mutex.Unlock()
   496  	return SampleStdDev(s.values)
   497  }
   498  
   499  // Sum returns the sum of the values in the sample.
   500  func (s *UniformSample) Sum() int64 {
   501  	s.mutex.Lock()
   502  	defer s.mutex.Unlock()
   503  	return SampleSum(s.values)
   504  }
   505  
   506  // Update samples a new value.
   507  func (s *UniformSample) Update(v int64) {
   508  	s.mutex.Lock()
   509  	defer s.mutex.Unlock()
   510  	s.count++
   511  	if len(s.values) < s.reservoirSize {
   512  		s.values = append(s.values, v)
   513  	} else {
   514  		r := rand.Int63n(s.count)
   515  		if r < int64(len(s.values)) {
   516  			s.values[int(r)] = v
   517  		}
   518  	}
   519  }
   520  
   521  // Values returns a copy of the values in the sample.
   522  func (s *UniformSample) Values() []int64 {
   523  	s.mutex.Lock()
   524  	defer s.mutex.Unlock()
   525  	values := make([]int64, len(s.values))
   526  	copy(values, s.values)
   527  	return values
   528  }
   529  
   530  // Variance returns the variance of the values in the sample.
   531  func (s *UniformSample) Variance() float64 {
   532  	s.mutex.Lock()
   533  	defer s.mutex.Unlock()
   534  	return SampleVariance(s.values)
   535  }
   536  
   537  // expDecaySample represents an individual sample in a heap.
   538  type expDecaySample struct {
   539  	k float64
   540  	v int64
   541  }
   542  
   543  func newExpDecaySampleHeap(reservoirSize int) *expDecaySampleHeap {
   544  	return &expDecaySampleHeap{make([]expDecaySample, 0, reservoirSize)}
   545  }
   546  
   547  // expDecaySampleHeap is a min-heap of expDecaySamples.
   548  // The internal implementation is copied from the standard library's container/heap
   549  type expDecaySampleHeap struct {
   550  	s []expDecaySample
   551  }
   552  
   553  func (h *expDecaySampleHeap) Clear() {
   554  	h.s = h.s[:0]
   555  }
   556  
   557  func (h *expDecaySampleHeap) Push(s expDecaySample) {
   558  	n := len(h.s)
   559  	h.s = h.s[0 : n+1]
   560  	h.s[n] = s
   561  	h.up(n)
   562  }
   563  
   564  func (h *expDecaySampleHeap) Pop() expDecaySample {
   565  	n := len(h.s) - 1
   566  	h.s[0], h.s[n] = h.s[n], h.s[0]
   567  	h.down(0, n)
   568  
   569  	n = len(h.s)
   570  	s := h.s[n-1]
   571  	h.s = h.s[0 : n-1]
   572  	return s
   573  }
   574  
   575  func (h *expDecaySampleHeap) Size() int {
   576  	return len(h.s)
   577  }
   578  
   579  func (h *expDecaySampleHeap) Values() []expDecaySample {
   580  	return h.s
   581  }
   582  
   583  func (h *expDecaySampleHeap) up(j int) {
   584  	for {
   585  		i := (j - 1) / 2 // parent
   586  		if i == j || !(h.s[j].k < h.s[i].k) {
   587  			break
   588  		}
   589  		h.s[i], h.s[j] = h.s[j], h.s[i]
   590  		j = i
   591  	}
   592  }
   593  
   594  func (h *expDecaySampleHeap) down(i, n int) {
   595  	for {
   596  		j1 := 2*i + 1
   597  		if j1 >= n || j1 < 0 { // j1 < 0 after int overflow
   598  			break
   599  		}
   600  		j := j1 // left child
   601  		if j2 := j1 + 1; j2 < n && !(h.s[j1].k < h.s[j2].k) {
   602  			j = j2 // = 2*i + 2  // right child
   603  		}
   604  		if !(h.s[j].k < h.s[i].k) {
   605  			break
   606  		}
   607  		h.s[i], h.s[j] = h.s[j], h.s[i]
   608  		i = j
   609  	}
   610  }
   611  
   612  type int64Slice []int64
   613  
   614  func (p int64Slice) Len() int           { return len(p) }
   615  func (p int64Slice) Less(i, j int) bool { return p[i] < p[j] }
   616  func (p int64Slice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
   617
View as plain text