add support to batch metric and send on a strict interval

net_sink.go

@@ -118,8 +118,8 @@ func NewNetSink(opts ...SinkOption) FlushableSink {
 		bufSize = defaultBufferSizeTCP
 	}
 
-	s.outc = make(chan *bytes.Buffer, approxMaxMemBytes/bufSize)
-	s.retryc = make(chan *bytes.Buffer, 1) // It should be okay to limit this given we preferentially process from this over outc.
+	s.outc = make(chan *bytes.Buffer, approxMaxMemBytes/bufSize) // todo: need to understand why/how this number was chosen and probably elevate it
+	s.retryc = make(chan *bytes.Buffer, 1)                       // It should be okay to limit this given we preferentially process from this over outc.
 
 	writer := &sinkWriter{outc: s.outc}
 	s.bufWriter = bufio.NewWriterSize(writer, bufSize)
@@ -274,11 +274,16 @@ func (s *netSink) FlushTimer(name string, value float64) {
 
 func (s *netSink) run() {
 	addr := net.JoinHostPort(s.conf.StatsdHost, strconv.Itoa(s.conf.StatsdPort))
+	batch := GetSettings().ForcedBatching
+
+	batchc := make(chan *bytes.Buffer, cap(s.outc))
 
 	var reconnectFailed bool // true if last reconnect failed
 
 	t := time.NewTicker(flushInterval)
 	defer t.Stop()
+
+	// metric write loop (actively draining content of outc channel)
 	for {
 		if s.conn == nil {
 			if err := s.connect(addr); err != nil {
@@ -306,38 +311,52 @@ func (s *netSink) run() {
 				s.handleFlushErrorSize(err, buf.Len())
 				s.mu.Unlock()
 			}
-			putBuffer(buf)
+			putBuffer(buf) // todo understand: we write the stats buffer we have successfully sent to a pool, but anytime we access data from that pool, we clear it and reuse the allocation for the next stat
 			continue
 		default:
 			// Drop through in case retryc has nothing.
 		}
 
+		// if the channel is at capacity, it's blocking, let's signal for it to be flushed and unblock
+		if len(s.outc) == cap(s.outc) {
+			s.Flush()
+		}
+
 		select {
-		case <-t.C:
-			s.flush()
 		case done := <-s.doFlush:
-			// Only flush pending buffers, this prevents an issue where
-			// continuous writes prevent the flush loop from exiting.
-			//
-			// If there is an error writeToConn() will set the conn to
-			// nil thus breaking the loop.
-			//
-			n := len(s.outc)
+			n := len(batchc)
 			for i := 0; i < n && s.conn != nil; i++ {
-				buf := <-s.outc
+				buf := <-batchc
 				if err := s.writeToConn(buf); err != nil {
 					s.retryc <- buf
 					continue
 				}
-				putBuffer(buf)
+				putBuffer(buf) // todo understand: we write the stats buffer we have successfully sent to a pool, but anytime we access data from that pool, we clear it and reuse the allocation for the next stat
 			}
+
 			close(done)
+		case <-t.C:
+			s.flush()
 		case buf := <-s.outc:
+			// Normally we will send stats anytime outc has data
+			//
+			// Gauages and Counters are written to outc at a cadence of GOSTATS_FLUSH_INTERVAL_SECONDS
+			// Timers are written adhoc to outc
+			//
+			// With batch we will rely on doFlush which is also controlled by the GOSTATS_FLUSH_INTERVAL_SECONDS
+			//
+			// Side effects:
+			// * Implied Timer batching under the flush interval
+			// * ~1 second delay to Gauge and Counter writes by batching these writes to the next internval
+			if batch {
+				batchc <- buf
+				continue
+			}
 			if err := s.writeToConn(buf); err != nil {
 				s.retryc <- buf
 				continue
 			}
-			putBuffer(buf)
+			putBuffer(buf) // todo understand: we write the stats buffer we have successfully sent to a pool, but anytime we access data from that pool, we clear it and reuse the allocation for the next stat
 		}
 	}
 }

net_sink_test.go

@@ -847,6 +847,68 @@ func testNetSinkIntegration(t *testing.T, protocol string) {
 	})
 }
 
+func TestFlushTimerNoBatching(t *testing.T) {
+	err := os.Setenv("GOSTATS_FORCED_BATCHING", "false")
+	if err != nil {
+		t.Fatalf("Failed to set environment variable: %s", err)
+	}
+
+	expected := [...]string{
+		"timer_int:1|ms\n",
+		"timer_float:1.230000|ms\n",
+	}
+
+	ts, sink := setupTestNetSink(t, "tcp", false) // the protocol is arbitrary and unimportant for batching
+	defer ts.Close()
+
+	sink.FlushTimer("timer_int", 1)
+	sink.FlushTimer("timer_float", 1.23)
+
+	time.Sleep(2001 * time.Millisecond)
+
+	exp := strings.Join(expected[:], "")
+	buf := ts.String()
+	if buf != exp {
+		t.Errorf("Not all stats were written\ngot:\n%q\nwant:\n%q\n", buf, exp)
+	}
+
+	os.Unsetenv("GOSTATS_FORCED_BATCHING")
+}
+
+func TestFlushTimerBatching(t *testing.T) {
+	err := os.Setenv("GOSTATS_FORCED_BATCHING", "true")
+	if err != nil {
+		t.Fatalf("Failed to set environment variable: %s", err)
+	}
+
+	expected := [...]string{
+		"timer_int:1|ms\n",
+		"timer_float:1.230000|ms\n",
+	}
+
+	ts, sink := setupTestNetSink(t, "tcp", false) // the protocol is arbitrary and unimportant for batching
+	defer ts.Close()
+
+	sink.FlushTimer("timer_int", 1)
+	sink.FlushTimer("timer_float", 1.23)
+	time.Sleep(2001 * time.Millisecond)
+
+	if ts.String() != "" {
+		t.Errorf("Stats were written despite forced batching")
+	}
+
+	sink.Flush()
+	time.Sleep(2001 * time.Millisecond)
+
+	exp := strings.Join(expected[:], "")
+	buf := ts.String()
+	if buf != exp {
+		t.Errorf("Not all stats were written\ngot:\n%q\nwant:\n%q\n", buf, exp)
+	}
+
+	os.Unsetenv("GOSTATS_FORCED_BATCHING")
+}
+
 func TestNetSink_Integration_TCP(t *testing.T) {
 	testNetSinkIntegration(t, "tcp")
 }

settings.go

@@ -20,6 +20,8 @@ const (
 	DefaultFlushIntervalS = 5
 	// DefaultLoggingSinkDisabled is the default behavior of logging sink suppression, default is false.
 	DefaultLoggingSinkDisabled = false
+	// DefaultForcedBatching defines if we want to force batching by default
+	DefaultForcedBatching = false
 )
 
 // The Settings type is used to configure gostats. gostats uses environment
@@ -38,6 +40,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"`
+	// Force batching under FlushIntervalS for all metrics
+	ForcedBatching bool `envconfig:"GOSTATS_FORCED_BATCHING" default:"false"`
 }
 
 // An envError is an error that occurred parsing an environment variable
@@ -101,13 +105,18 @@ func GetSettings() Settings {
 	if err != nil {
 		panic(err)
 	}
+	forcedBatching, err := envBool("GOSTATS_FORCED_BATCHING", DefaultForcedBatching)
+	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,
+		ForcedBatching:      forcedBatching,
 	}
 }
 

stats.go

@@ -313,7 +313,7 @@ func (t *timer) AddDuration(dur time.Duration) {
 }
 
 func (t *timer) AddValue(value float64) {
-	t.sink.FlushTimer(t.name, value)
+	t.sink.FlushTimer(t.name, value) // writes the timer to a buffered channel which will send immediately unless GOSTATS_FORCED_BATCHING is set
 }
 
 func (t *timer) AllocateSpan() Timespan {
@@ -357,6 +357,7 @@ func (s *statStore) validateTags(tags map[string]string) {
 	}
 }
 
+// flush loop based on specified ticker
 func (s *statStore) StartContext(ctx context.Context, ticker *time.Ticker) {
 	for {
 		select {
@@ -383,19 +384,19 @@ func (s *statStore) Flush() {
 	s.counters.Range(func(key, v interface{}) bool {
 		// do not flush counters that are set to zero
 		if value := v.(*counter).latch(); value != 0 {
-			s.sink.FlushCounter(key.(string), value)
+			s.sink.FlushCounter(key.(string), value) // writes the counter to a buffered channel which will be sent either right away or at the end of the stack depending on GOSTATS_FORCED_BATCHING
 		}
 		return true
 	})
 
 	s.gauges.Range(func(key, v interface{}) bool {
-		s.sink.FlushGauge(key.(string), v.(*gauge).Value())
+		s.sink.FlushGauge(key.(string), v.(*gauge).Value()) // writes the gauage to a buffered channel which will be sent either right away or at the end of the stack depending on GOSTATS_FORCED_BATCHING
 		return true
 	})
 
 	flushableSink, ok := s.sink.(FlushableSink)
 	if ok {
-		flushableSink.Flush()
+		flushableSink.Flush() // signal the buffered channel (buffered counters, gauges and timers) to be drained and sent if there is anythig to send
 	}
 }
 

stats_test.go

@@ -78,9 +78,9 @@ func TestValidateTags(t *testing.T) {
 	store.Flush()
 
 	expected := "test:1|c"
-	counter := sink.record
-	if !strings.Contains(counter, expected) {
-		t.Error("wanted counter value of test:1|c, got", counter)
+	output := sink.record
+	if !strings.Contains(output, expected) && !strings.Contains(output, "reserved_tag") {
+		t.Errorf("Expected without reserved tags: '%s' Got: '%s'", expected, output)
 	}
 
 	// A reserved tag should trigger adding the reserved_tag counter
@@ -89,10 +89,11 @@ func TestValidateTags(t *testing.T) {
 	store.NewCounterWithTags("test", map[string]string{"host": "i"}).Inc()
 	store.Flush()
 
-	expected = "reserved_tag:1|c\ntest.__host=i:1|c"
-	counter = sink.record
-	if !strings.Contains(counter, expected) {
-		t.Error("wanted counter value of test.___f=i:1|c, got", counter)
+	expected = "test.__host=i:1|c"
+	expectedReservedTag := "reserved_tag:1|c"
+	output = sink.record
+	if !strings.Contains(output, expected) && !strings.Contains(output, expectedReservedTag) {
+		t.Errorf("Expected: '%s' and '%s', In: '%s'", expected, expectedReservedTag, output)
 	}
 }