add reservoir timers

logging_sink.go

@@ -92,6 +92,10 @@ func (s *loggingSink) FlushGauge(name string, value uint64) { s.log(name, "gauge
 
 func (s *loggingSink) FlushTimer(name string, value float64) { s.log(name, "timer", value) }
 
+func (s *loggingSink) FlushSampledTimer(name string, value, _ float64) {
+	s.FlushTimer(name, value)
+}
+
 func (s *loggingSink) Flush() { s.log("", "all stats", 0) }
 
 // Logger

mock/sink.go

@@ -103,6 +103,12 @@ func (s *Sink) FlushTimer(name string, val float64) {
 	atomic.AddInt64(&p.count, 1)
 }
 
+// FlushSampledTimer implements the stats.Sink.FlushSampledTimer method and adds val to
+// stat name.
+func (s *Sink) FlushSampledTimer(name string, val, _ float64) {
+	s.FlushTimer(name, val)
+}
+
 // LoadCounter returns the value for stat name and if it was found.
 func (s *Sink) LoadCounter(name string) (uint64, bool) {
 	v, ok := s.counters().Load(name)

net_sink.go

@@ -272,6 +272,36 @@ func (s *netSink) FlushTimer(name string, value float64) {
 	}
 }
 
+func (s *netSink) FlushSampledTimer(name string, value, sampleRate float64) {
+	timerSuffix := "|ms"
+	sampleSuffix := "|@"
+	metricSuffix := "\n"
+
+	// todo: see if we can dedup code here without the expense of efficiency
+	var writeValue func(*buffer)
+	if 0 <= value && value < math.MaxUint64 && math.Trunc(value) == value {
+		writeValue = func(b *buffer) { b.WriteUnit64(uint64(value)) }
+	} else {
+		writeValue = func(b *buffer) { b.WriteFloat64(value) }
+	}
+
+	b := pbFree.Get().(*buffer)
+
+	b.WriteString(name)
+	b.WriteChar(':')
+	writeValue(b)
+	b.WriteString(timerSuffix)
+
+	b.WriteString(sampleSuffix)
+	b.writeFloat64WithPrecision(sampleRate, 2) // todo: deteremine how many decimal places we need
+	b.WriteString(metricSuffix)
+
+	s.writeBuffer(b)
+
+	b.Reset()
+	pbFree.Put(b)
+}
+
 func (s *netSink) run() {
 	addr := net.JoinHostPort(s.conf.StatsdHost, strconv.Itoa(s.conf.StatsdPort))
 
@@ -417,5 +447,9 @@ func (b *buffer) WriteUnit64(val uint64) {
 }
 
 func (b *buffer) WriteFloat64(val float64) {
-	*b = strconv.AppendFloat(*b, val, 'f', 6, 64)
+	b.writeFloat64WithPrecision(val, 6)
+}
+
+func (b *buffer) writeFloat64WithPrecision(val float64, precision int) {
+	*b = strconv.AppendFloat(*b, val, 'f', precision, 64)
 }

net_sink_test.go

@@ -45,6 +45,12 @@ func (s *testStatSink) FlushTimer(name string, value float64) {
 	s.Unlock()
 }
 
+func (s *testStatSink) FlushSampledTimer(name string, value, sampleRate float64) {
+	s.Lock()
+	s.record += fmt.Sprintf("%s:%f|ms|@%f\n", name, value, sampleRate)
+	s.Unlock()
+}
+
 func TestCreateTimer(t *testing.T) {
 	sink := &testStatSink{}
 	store := NewStore(sink, true)

net_util_test.go

@@ -272,6 +272,14 @@ func (s *netTestSink) String() string {
 	return str
 }
 
+func (s *netTestSink) Pull() string {
+	s.mu.Lock()
+	str := s.buf.String()
+	s.buf.Reset()
+	s.mu.Unlock()
+	return str
+}
+
 func (s *netTestSink) Host(t testing.TB) string {
 	t.Helper()
 	host, _, err := net.SplitHostPort(s.conn.Address().String())

null_sink.go

@@ -13,4 +13,6 @@ func (s nullSink) FlushGauge(name string, value uint64) {} //nolint:revive
 
 func (s nullSink) FlushTimer(name string, value float64) {} //nolint:revive
 
+func (s nullSink) FlushSampledTimer(name string, value, sampleRate float64) {} //nolint:revive
+
 func (s nullSink) Flush() {}

settings.go

@@ -20,6 +20,15 @@ const (
 	DefaultFlushIntervalS = 5
 	// DefaultLoggingSinkDisabled is the default behavior of logging sink suppression, default is false.
 	DefaultLoggingSinkDisabled = false
+	// DefaultUseReservoirTimer defines if all timers should be reservoir timers by default.
+	DefaultUseReservoirTimer = false
+	// FixedTimerReservoirSize is the max capacity of the reservoir for reservoir timers.
+	// note: needs to be rounded to a power of two e.g. 1 << bits.Len(uint(100)) = 128
+	// todo: see if it's worth an efficiency trade off to reduce tech debt of this magic number and allowing any number.
+	//       we could determine the difference between the defined size and next power of two
+	//       and use that to offset the counter when ANDing it against the mask,
+	//       once the result is 0 we just increment offset by "original offset"
+	FixedTimerReservoirSize = 128
 )
 
 // The Settings type is used to configure gostats. gostats uses environment
@@ -38,6 +47,8 @@ type Settings struct {
 	// Disable the LoggingSink when USE_STATSD is false and use the NullSink instead.
 	// This will cause all stats to be silently dropped.
 	LoggingSinkDisabled bool `envconfig:"GOSTATS_LOGGING_SINK_DISABLED" default:"false"`
+	// Make all timers reservoir timers with implied sampling under flush interval of FlushIntervalS
+	UseReservoirTimer bool `envconfig:"GOSTATS_USE_RESERVOIR_TIMER" default:"false"`
 }
 
 // An envError is an error that occurred parsing an environment variable
@@ -84,6 +95,7 @@ func envBool(key string, def bool) (bool, error) {
 }
 
 // GetSettings returns the Settings gostats will run with.
+// todo: can we optimize this by storing the result for subsequent calls
 func GetSettings() Settings {
 	useStatsd, err := envBool("USE_STATSD", DefaultUseStatsd)
 	if err != nil {
@@ -101,13 +113,18 @@ func GetSettings() Settings {
 	if err != nil {
 		panic(err)
 	}
+	useReservoirTimer, err := envBool("GOSTATS_USE_RESERVOIR_TIMER", DefaultUseReservoirTimer)
+	if err != nil {
+		panic(err)
+	}
 	return Settings{
 		UseStatsd:           useStatsd,
 		StatsdHost:          envOr("STATSD_HOST", DefaultStatsdHost),
 		StatsdProtocol:      envOr("STATSD_PROTOCOL", DefaultStatsdProtocol),
 		StatsdPort:          statsdPort,
 		FlushIntervalS:      flushIntervalS,
 		LoggingSinkDisabled: loggingSinkDisabled,
+		UseReservoirTimer:   useReservoirTimer,
 	}
 }
 

sink.go

@@ -6,6 +6,7 @@ type Sink interface {
 	FlushCounter(name string, value uint64)
 	FlushGauge(name string, value uint64)
 	FlushTimer(name string, value float64)
+	FlushSampledTimer(name string, value, sampleRate float64)
 }
 
 // FlushableSink is an extension of Sink that provides a Flush() function that

stats.go

@@ -214,22 +214,34 @@ type StatGenerator interface {
 // NewStore returns an Empty store that flushes to Sink passed as an argument.
 // Note: the export argument is unused.
 func NewStore(sink Sink, _ bool) Store {
-	return &statStore{sink: sink}
+	return &statStore{
+		sink:      sink,
+		timerType: standard,
+	}
+}
+
+func newStatStore(sink Sink, export bool, conf Settings) *statStore {
+	store := NewStore(sink, export).(*statStore)
+	if conf.UseReservoirTimer {
+		store.timerType = reservoir
+	}
+	return store
 }
 
 // NewDefaultStore returns a Store with a TCP statsd sink, and a running flush timer.
+// note: this is the only way to use reservoir timers as they rely on the store flush loop
 func NewDefaultStore() Store {
 	var newStore Store
 	settings := GetSettings()
 	if !settings.UseStatsd {
 		if settings.LoggingSinkDisabled {
-			newStore = NewStore(NewNullSink(), false)
+			newStore = newStatStore(NewNullSink(), false, settings)
 		} else {
-			newStore = NewStore(NewLoggingSink(), false)
+			newStore = newStatStore(NewLoggingSink(), false, settings)
 		}
 		go newStore.Start(time.NewTicker(10 * time.Second))
 	} else {
-		newStore = NewStore(NewTCPStatsdSink(), false)
+		newStore = newStatStore(NewTCPStatsdSink(), false, settings)
 		go newStore.Start(time.NewTicker(time.Duration(settings.FlushIntervalS) * time.Second))
 	}
 	return newStore
@@ -298,30 +310,104 @@ func (c *gauge) Value() uint64 {
 	return atomic.LoadUint64(&c.value)
 }
 
-type timer struct {
+type timerType int
+
+const (
+	standard timerType = iota
+	reservoir
+)
+
+type timer interface {
+	time(time.Duration)
+	AddDuration(time.Duration)
+	AddValue(float64)
+	AllocateSpan() Timespan
+	Empty() ([]float64, uint64)
+}
+
+type standardTimer struct {
 	base time.Duration
 	name string
 	sink Sink
 }
 
-func (t *timer) time(dur time.Duration) {
+func (t *standardTimer) time(dur time.Duration) {
 	t.AddDuration(dur)
 }
 
-func (t *timer) AddDuration(dur time.Duration) {
+func (t *standardTimer) AddDuration(dur time.Duration) {
 	t.AddValue(float64(dur / t.base))
 }
 
-func (t *timer) AddValue(value float64) {
+func (t *standardTimer) AddValue(value float64) {
 	t.sink.FlushTimer(t.name, value)
 }
 
-func (t *timer) AllocateSpan() Timespan {
+func (t *standardTimer) AllocateSpan() Timespan {
 	return &timespan{timer: t, start: time.Now()}
 }
 
+// values are not collected for this timer
+func (t *standardTimer) Empty() ([]float64, uint64) {
+	return nil, 0
+}
+
+type reservoirTimer struct {
+	mu       sync.Mutex
+	base     time.Duration
+	name     string
+	ringSize uint64 // immutable
+	ringMask uint64 // immutable
+	values   []float64
+	count    uint64
+}
+
+func (t *reservoirTimer) time(dur time.Duration) {
+	t.AddDuration(dur)
+}
+
+func (t *reservoirTimer) AddDuration(dur time.Duration) {
+	t.AddValue(float64(dur / t.base))
+}
+
+func (t *reservoirTimer) AddValue(value float64) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	t.values[t.count&t.ringMask] = value
+	t.count++
+}
+
+func (t *reservoirTimer) AllocateSpan() Timespan {
+	return &timespan{timer: t, start: time.Now()}
+}
+
+// resets the reservoir returning all valeus collected and a count of the total inflow,
+// including what exceeded the collection size and was dropped
+func (t *reservoirTimer) Empty() ([]float64, uint64) {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	count := t.count
+
+	var accumulation uint64
+	if count > t.ringSize {
+		accumulation = t.ringSize
+	} else {
+		accumulation = count
+	}
+
+	// make a copy to avoid data races
+	values := make([]float64, accumulation)
+	copy(values, t.values[:accumulation]) // since the slice memory is reused only copy what we accumulated in the current processing itteration
+
+	t.count = 0 // new values can start being written to the slice
+
+	return values, count
+}
+
 type timespan struct {
-	timer *timer
+	timer timer
 	start time.Time
 }
 
@@ -336,9 +422,11 @@ func (ts *timespan) CompleteWithDuration(value time.Duration) {
 }
 
 type statStore struct {
-	counters sync.Map
-	gauges   sync.Map
-	timers   sync.Map
+	// these maps may grow indefinitely however slots in this maps are reused as stats names are stable over the lifetime of the process
+	counters  sync.Map
+	gauges    sync.Map
+	timers    sync.Map
+	timerType timerType
 
 	mu             sync.RWMutex
 	statGenerators []StatGenerator
@@ -393,6 +481,26 @@ func (s *statStore) Flush() {
 		return true
 	})
 
+	s.timers.Range(func(key, v interface{}) bool {
+		if timer, ok := v.(*reservoirTimer); ok {
+			values, count := timer.Empty()
+			reservoirSize := timer.ringSize // assuming this is immutable
+
+			var sampleRate float64
+			if count <= reservoirSize {
+				sampleRate = 1.0
+			} else {
+				sampleRate = float64(reservoirSize) / float64(count)
+			}
+
+			for _, value := range values {
+				s.sink.FlushSampledTimer(key.(string), value, sampleRate)
+			}
+		}
+
+		return true
+	})
+
 	flushableSink, ok := s.sink.(FlushableSink)
 	if ok {
 		flushableSink.Flush()
@@ -490,14 +598,35 @@ func (s *statStore) NewPerInstanceGauge(name string, tags map[string]string) Gau
 	return s.newGaugeWithTagSet(name, tagspkg.TagSet(nil).MergePerInstanceTags(tags))
 }
 
-func (s *statStore) newTimer(serializedName string, base time.Duration) *timer {
+func (s *statStore) newTimer(serializedName string, base time.Duration) timer {
 	if v, ok := s.timers.Load(serializedName); ok {
-		return v.(*timer)
+		return v.(timer)
 	}
-	t := &timer{name: serializedName, sink: s.sink, base: base}
+
+	var t timer
+	switch s.timerType {
+	case reservoir:
+		t = &reservoirTimer{
+			name:     serializedName,
+			base:     base,
+			ringSize: FixedTimerReservoirSize,
+			ringMask: FixedTimerReservoirSize - 1,
+			values:   make([]float64, FixedTimerReservoirSize),
+		}
+	case standard: // this should allow backward compatible a backwards compatible fallback as standard is the zero value of s.timerType
+		fallthrough
+	default:
+		t = &standardTimer{
+			name: serializedName,
+			sink: s.sink,
+			base: base,
+		}
+	}
+
 	if v, loaded := s.timers.LoadOrStore(serializedName, t); loaded {
-		return v.(*timer)
+		return v.(timer)
 	}
+
 	return t
 }