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) FlushAggregatedTimer(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)
 }
 
+// FlushAggregatedTimer implements the stats.Sink.FlushAggregatedTimer method and adds val to
+// stat name.
+func (s *Sink) FlushAggregatedTimer(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

@@ -260,15 +260,23 @@ func (s *netSink) FlushGauge(name string, value uint64) {
 }
 
 func (s *netSink) FlushTimer(name string, value float64) {
-	// Since we mistakenly use floating point values to represent time
-	// durations this method is often passed an integer encoded as a
+	s.flushFloatOptimized(name, "|ms\n", value)
+}
+
+func (s *netSink) FlushAggregatedTimer(name string, value, sampleRate float64) {
+	suffix := fmt.Sprintf("|ms|@%.1f\n", sampleRate)
+	s.flushFloatOptimized(name, suffix, value)
+}
+
+func (s *netSink) flushFloatOptimized(name, suffix string, value float64) {
+	// Since we historitically used floating point values to represent time
+	// durations, metrics (particularly timers) are often recorded as an integer encoded as a
 	// float. Formatting integers is much faster (>2x) than formatting
-	// floats so use integer formatting whenever possible.
-	//
+	// floats, so we should convert to an integer whenever possible.
 	if 0 <= value && value < math.MaxUint64 && math.Trunc(value) == value {
-		s.flushUint64(name, "|ms\n", uint64(value))
+		s.flushUint64(name, suffix, uint64(value))
 	} else {
-		s.flushFloat64(name, "|ms\n", value)
+		s.flushFloat64(name, suffix, value)
 	}
 }
 

net_sink_test.go

@@ -45,6 +45,12 @@ func (s *testStatSink) FlushTimer(name string, value float64) {
 	s.Unlock()
 }
 
+func (s *testStatSink) FlushAggregatedTimer(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)

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) FlushAggregatedTimer(name string, value, sampleRate float64) {} //nolint:revive
+
 func (s nullSink) Flush() {}

settings.go

@@ -20,6 +20,7 @@ const (
 	DefaultFlushIntervalS = 5
 	// DefaultLoggingSinkDisabled is the default behavior of logging sink suppression, default is false.
 	DefaultLoggingSinkDisabled = false
+	DefaultTimerReservoirSize  = 0
 )
 
 // The Settings type is used to configure gostats. gostats uses environment
@@ -38,6 +39,7 @@ 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"`
+	TimerReservoirSize  int  `envconfig:"GOSTATS_TIMER_RESERVOIR_SIZE" default:"0"`
 }
 
 // An envError is an error that occurred parsing an environment variable
@@ -101,17 +103,26 @@ func GetSettings() Settings {
 	if err != nil {
 		panic(err)
 	}
+	timerReservoirSize, err := envInt("GOSTATS_TIMER_RESERVOIR_SIZE", DefaultTimerReservoirSize)
+	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,
+		TimerReservoirSize:  timerReservoirSize,
 	}
 }
 
 // FlushInterval returns the flush interval duration.
 func (s *Settings) FlushInterval() time.Duration {
 	return time.Duration(s.FlushIntervalS) * time.Second
 }
+
+func (s *Settings) isTimerReservoirEnabled() bool {
+	return s.TimerReservoirSize > 0
+}

sink.go

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

stats.go

@@ -214,7 +214,10 @@ 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,
+		conf: GetSettings(), // todo: right now the environment is being loaded in multiple places and can be made more efficient
+	}
 }
 
 // NewDefaultStore returns a Store with a TCP statsd sink, and a running flush timer.
@@ -298,30 +301,95 @@ func (c *gauge) Value() uint64 {
 	return atomic.LoadUint64(&c.value)
 }
 
-type timer struct {
+type timer interface {
+	time(time.Duration)
+	AddDuration(time.Duration)
+	AddValue(float64)
+	AllocateSpan() Timespan
+	CollectedValue() []float64
+	SampleRate() float64
+}
+
+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()}
 }
 
+func (t *standardTimer) CollectedValue() []float64 {
+	return nil // since we flush right away nothing will be collected
+}
+
+func (t *standardTimer) SampleRate() float64 {
+	return 1.0 // metrics which are not sampled have an implicit sample rate 1.0
+}
+
+type reservoirTimer struct {
+	base     time.Duration
+	name     string
+	capacity int
+	values   []float64
+	count    int
+	mu       sync.Mutex
+}
+
+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()
+
+	if t.count < t.capacity {
+		t.values = append(t.values, value)
+	} else {
+		t.values = append(t.values[1:], value) // discard the oldest value when the reservoir is full, this can probably be smarter
+	}
+
+	t.count++
+}
+
+func (t *reservoirTimer) AllocateSpan() Timespan {
+	return &timespan{timer: t, start: time.Now()}
+}
+
+func (t *reservoirTimer) CollectedValue() []float64 {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	// return a copy of the values slice to avoid data races
+	values := make([]float64, len(t.values))
+	copy(values, t.values)
+	return values
+}
+
+func (t *reservoirTimer) SampleRate() float64 {
+	return float64(len(t.values)) / float64(t.count)
+}
+
 type timespan struct {
-	timer *timer
+	timer timer
 	start time.Time
 }
 
@@ -336,6 +404,7 @@ func (ts *timespan) CompleteWithDuration(value time.Duration) {
 }
 
 type statStore struct {
+	// todo: no idea how memory is managed here, when are the map entries ever deleted?
 	counters sync.Map
 	gauges   sync.Map
 	timers   sync.Map
@@ -344,6 +413,8 @@ type statStore struct {
 	statGenerators []StatGenerator
 
 	sink Sink
+
+	conf Settings
 }
 
 var ReservedTagWords = map[string]bool{"asg": true, "az": true, "backend": true, "canary": true, "host": true, "period": true, "region": true, "shard": true, "window": true, "source": true, "project": true, "facet": true, "envoyservice": true}
@@ -380,6 +451,8 @@ func (s *statStore) Flush() {
 	}
 	s.mu.RUnlock()
 
+	// todo: if we're not deleting the data we flush from these maps, won't we just keep resending them?
+
 	s.counters.Range(func(key, v interface{}) bool {
 		// do not flush counters that are set to zero
 		if value := v.(*counter).latch(); value != 0 {
@@ -393,6 +466,18 @@ func (s *statStore) Flush() {
 		return true
 	})
 
+	s.timers.Range(func(key, v interface{}) bool {
+		if timer, ok := v.(*reservoirTimer); ok {
+			sampleRate := timer.SampleRate()
+			for _, value := range timer.CollectedValue() {
+				s.sink.FlushAggregatedTimer(key.(string), value, sampleRate)
+			}
+			s.timers.Delete(key) // delete it from the map so it's not flushed again
+		}
+
+		return true
+	})
+
 	flushableSink, ok := s.sink.(FlushableSink)
 	if ok {
 		flushableSink.Flush()
@@ -490,14 +575,32 @@ 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
+	if s.conf.isTimerReservoirEnabled() {
+		t = &reservoirTimer{
+			name:     serializedName,
+			base:     base,
+			capacity: s.conf.TimerReservoirSize,
+			values:   make([]float64, 0, s.conf.TimerReservoirSize),
+			count:    0,
+		}
+	} else {
+		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
 }