[POC] Client side aggregation.

.travis.yml

@@ -1,7 +1,6 @@
 language: go
 
 go:
-  - 1.7
   - 1.8
   - 1.9
   - 1.10.x

statsd/aggregator.go

@@ -0,0 +1,158 @@
+package statsd
+
+import (
+	"strings"
+	"sync"
+	"time"
+)
+
+type (
+	countsMap map[string]*countMetric
+	gaugesMap map[string]*gaugeMetric
+	setsMap   map[string]*setMetric
+)
+
+type aggregator struct {
+	client *Client
+
+	counts  countsMap
+	countsM sync.RWMutex
+
+	gauges  gaugesMap
+	gaugesM sync.RWMutex
+
+	sets  setsMap
+	setsM sync.RWMutex
+
+	closed chan struct{}
+	exited chan struct{}
+}
+
+func newAggregator(c *Client) *aggregator {
+	return &aggregator{
+		client: c,
+		counts: countsMap{},
+		gauges: gaugesMap{},
+		sets:   setsMap{},
+		closed: make(chan struct{}),
+		exited: make(chan struct{}),
+	}
+}
+
+func (a *aggregator) start(flushInterval time.Duration) {
+	ticker := time.NewTicker(flushInterval)
+
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				a.sendMetrics()
+			case <-a.closed:
+				close(a.exited)
+				return
+			}
+		}
+	}()
+}
+
+func (a *aggregator) sendMetrics() {
+	for _, m := range a.flushMetrics() {
+		a.client.send(m)
+	}
+}
+
+func (a *aggregator) stop() {
+	close(a.closed)
+	<-a.exited
+	a.sendMetrics()
+}
+
+func (a *aggregator) flushMetrics() []metric {
+	metrics := []metric{}
+
+	// We reset the values to avoid sending 'zero' values for metrics not
+	// sampled during this flush interval
+
+	a.setsM.Lock()
+	sets := a.sets
+	a.sets = setsMap{}
+	a.setsM.Unlock()
+
+	for _, s := range sets {
+		metrics = append(metrics, s.flushUnsafe()...)
+	}
+
+	a.gaugesM.Lock()
+	gauges := a.gauges
+	a.gauges = gaugesMap{}
+	a.gaugesM.Unlock()
+
+	for _, g := range gauges {
+		metrics = append(metrics, g.flushUnsafe())
+	}
+
+	a.countsM.RLock()
+	counts := a.counts
+	a.counts = countsMap{}
+	a.countsM.RUnlock()
+
+	for _, c := range counts {
+		metrics = append(metrics, c.flushUnsafe())
+	}
+
+	return metrics
+}
+
+func getContext(name string, tags []string) string {
+	return name + ":" + strings.Join(tags, ",")
+}
+
+func (a *aggregator) count(name string, value int64, tags []string, rate float64) error {
+	context := getContext(name, tags)
+	a.countsM.RLock()
+	if count, found := a.counts[context]; found {
+		count.sample(value)
+		a.countsM.RUnlock()
+		return nil
+	}
+	a.countsM.RUnlock()
+
+	a.countsM.Lock()
+	a.counts[context] = newCountMetric(name, value, tags, rate)
+	a.countsM.Unlock()
+	return nil
+}
+
+func (a *aggregator) gauge(name string, value float64, tags []string, rate float64) error {
+	context := getContext(name, tags)
+	a.gaugesM.RLock()
+	if gauge, found := a.gauges[context]; found {
+		gauge.sample(value)
+		a.gaugesM.RUnlock()
+		return nil
+	}
+	a.gaugesM.RUnlock()
+
+	gauge := newGaugeMetric(name, value, tags, rate)
+
+	a.gaugesM.Lock()
+	a.gauges[context] = gauge
+	a.gaugesM.Unlock()
+	return nil
+}
+
+func (a *aggregator) set(name string, value string, tags []string, rate float64) error {
+	context := getContext(name, tags)
+	a.setsM.RLock()
+	if set, found := a.sets[context]; found {
+		set.sample(value)
+		a.setsM.RUnlock()
+		return nil
+	}
+	a.setsM.RUnlock()
+
+	a.setsM.Lock()
+	a.sets[context] = newSetMetric(name, value, tags, rate)
+	a.setsM.Unlock()
+	return nil
+}

statsd/aggregator_test.go

@@ -0,0 +1,131 @@
+package statsd
+
+import (
+	"sort"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAggregatorSample(t *testing.T) {
+	a := newAggregator(nil)
+
+	tags := []string{"tag1", "tag2"}
+
+	a.gauge("gaugeTest", 21, tags, 1)
+	assert.Len(t, a.gauges, 1)
+	assert.Contains(t, a.gauges, "gaugeTest:tag1,tag2")
+
+	a.count("countTest", 21, tags, 1)
+	assert.Len(t, a.counts, 1)
+	assert.Contains(t, a.counts, "countTest:tag1,tag2")
+
+	a.set("setTest", "value1", tags, 1)
+	assert.Len(t, a.sets, 1)
+	assert.Contains(t, a.sets, "setTest:tag1,tag2")
+
+	a.gauge("gaugeTest", 123, tags, 1)
+	assert.Len(t, a.gauges, 1)
+	assert.Contains(t, a.gauges, "gaugeTest:tag1,tag2")
+
+	a.count("countTest", 10, tags, 1)
+	assert.Len(t, a.counts, 1)
+	assert.Contains(t, a.counts, "countTest:tag1,tag2")
+
+	a.set("setTest", "value1", tags, 1)
+	assert.Len(t, a.sets, 1)
+	assert.Contains(t, a.sets, "setTest:tag1,tag2")
+}
+
+func TestAggregatorFlush(t *testing.T) {
+	a := newAggregator(nil)
+
+	tags := []string{"tag1", "tag2"}
+
+	a.gauge("gaugeTest1", 21, tags, 1)
+	a.gauge("gaugeTest1", 10, tags, 1)
+	a.gauge("gaugeTest2", 15, tags, 1)
+
+	a.count("countTest1", 21, tags, 1)
+	a.count("countTest1", 10, tags, 1)
+	a.count("countTest2", 1, tags, 1)
+
+	a.set("setTest1", "value1", tags, 1)
+	a.set("setTest1", "value1", tags, 1)
+	a.set("setTest1", "value2", tags, 1)
+	a.set("setTest2", "value1", tags, 1)
+
+	metrics := a.flushMetrics()
+
+	assert.Len(t, a.gauges, 0)
+	assert.Len(t, a.counts, 0)
+	assert.Len(t, a.sets, 0)
+
+	assert.Len(t, metrics, 7)
+
+	sort.Slice(metrics, func(i, j int) bool {
+		if metrics[i].metricType == metrics[j].metricType {
+			res := strings.Compare(metrics[i].name, metrics[j].name)
+			// this happens fo set
+			if res == 0 {
+				return strings.Compare(metrics[i].svalue, metrics[j].svalue) != 1
+			}
+			return res != 1
+		}
+		return metrics[i].metricType < metrics[j].metricType
+	})
+
+	assert.Equal(t, metrics, []metric{
+		metric{
+			metricType: gauge,
+			name:       "gaugeTest1",
+			tags:       tags,
+			rate:       1,
+			fvalue:     float64(10),
+		},
+		metric{
+			metricType: gauge,
+			name:       "gaugeTest2",
+			tags:       tags,
+			rate:       1,
+			fvalue:     float64(15),
+		},
+		metric{
+			metricType: count,
+			name:       "countTest1",
+			tags:       tags,
+			rate:       1,
+			ivalue:     int64(31),
+		},
+		metric{
+			metricType: count,
+			name:       "countTest2",
+			tags:       tags,
+			rate:       1,
+			ivalue:     int64(1),
+		},
+		metric{
+			metricType: set,
+			name:       "setTest1",
+			tags:       tags,
+			rate:       1,
+			svalue:     "value1",
+		},
+		metric{
+			metricType: set,
+			name:       "setTest1",
+			tags:       tags,
+			rate:       1,
+			svalue:     "value2",
+		},
+		metric{
+			metricType: set,
+			name:       "setTest2",
+			tags:       tags,
+			rate:       1,
+			svalue:     "value1",
+		},
+	})
+
+}