Repository: atlassian/gostatsd
Pull request #228 (2019-03-12 22:03:17)

Process metricmaps
This brings `*MetricMap`s and `[]*Metric`s all through the pipeline, including to the `Aggregator`.  The `CloudHandler` does not have first class support for `*MetricMap` handling due to:
1) requiring a big refactor to support it
2) a CH shouldn't be required in http mode, as anything that would be looked up should be statically added on a forwarder

It's probably best to review each commit in isolation.  Each commit should build and test clean.


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index dd351b3..f5f0649 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+14.0.0
+------
+- More internal pipeline refactoring, to optimize the http ingestion path.
+
 13.0.1
 ------
 - Adjust log level about forwarding failures and stop logging retries.
diff --git a/metric_consolidator.go b/metric_consolidator.go
index 8f563ba..792b494 100644
--- a/metric_consolidator.go
+++ b/metric_consolidator.go
@@ -86,7 +86,6 @@ func (mc *MetricConsolidator) Flush(ctx context.Context) {
 func (mc *MetricConsolidator) ReceiveMetrics(metrics []*Metric) {
 	mmTo := <-mc.maps
 	for _, m := range metrics {
-		m.TagsKey = m.FormatTagsKey()
 		mmTo.Receive(m)
 	}
 	mc.maps <- mmTo
diff --git a/metric_map.go b/metric_map.go
index 514fad5..4efbc52 100644
--- a/metric_map.go
+++ b/metric_map.go
@@ -28,7 +28,7 @@ func NewMetricMap() *MetricMap {
 
 // Receive adds a single Metric to the MetricMap, and releases the Metric.
 func (mm *MetricMap) Receive(m *Metric) {
-	tagsKey := m.TagsKey
+	tagsKey := m.FormatTagsKey()
 
 	switch m.Type {
 	case COUNTER:
@@ -129,6 +129,53 @@ func (mm *MetricMap) Merge(mmFrom *MetricMap) {
 	})
 }
 
+func (mm *MetricMap) IsEmpty() bool {
+	return len(mm.Counters) + len(mm.Timers) +  len(mm.Sets) + len(mm.Gauges) == 0
+}
+
+// Split will split a MetricMap up in to multiple MetricMaps, where each one contains metrics only for its buckets.
+func (mm *MetricMap) Split(count int) []*MetricMap {
+	maps := make([]*MetricMap, count)
+	for i := 0; i < count; i++ {
+		maps[i] = NewMetricMap()
+	}
+
+	mm.Counters.Each(func(metricName string, tagsKey string, c Counter) {
+		mmSplit := maps[Bucket(metricName, c.Hostname, count)]
+		if v, ok := mmSplit.Counters[metricName]; ok {
+			v[tagsKey] = c
+		} else {
+			mmSplit.Counters[metricName] = map[string]Counter{tagsKey: c}
+		}
+	})
+	mm.Gauges.Each(func(metricName string, tagsKey string, g Gauge) {
+		mmSplit := maps[Bucket(metricName, g.Hostname, count)]
+		if v, ok := mmSplit.Gauges[metricName]; ok {
+			v[tagsKey] = g
+		} else {
+			mmSplit.Gauges[metricName] = map[string]Gauge{tagsKey: g}
+		}
+	})
+	mm.Timers.Each(func(metricName string, tagsKey string, t Timer) {
+		mmSplit := maps[Bucket(metricName, t.Hostname, count)]
+		if v, ok := mmSplit.Timers[metricName]; ok {
+			v[tagsKey] = t
+		} else {
+			mmSplit.Timers[metricName] = map[string]Timer{tagsKey: t}
+		}
+	})
+	mm.Sets.Each(func(metricName string, tagsKey string, s Set) {
+		mmSplit := maps[Bucket(metricName, s.Hostname, count)]
+		if v, ok := mmSplit.Sets[metricName]; ok {
+			v[tagsKey] = s
+		} else {
+			mmSplit.Sets[metricName] = map[string]Set{tagsKey: s}
+		}
+	})
+
+	return maps
+}
+
 func (mm *MetricMap) receiveCounter(m *Metric, tagsKey string) {
 	value := int64(m.Value / m.Rate)
 	v, ok := mm.Counters[m.Name]
diff --git a/metrics.go b/metrics.go
index 2a81c97..4aa2fe5 100644
--- a/metrics.go
+++ b/metrics.go
@@ -65,11 +65,14 @@ func (m *Metric) Reset() {
 
 // Bucket will pick a distribution bucket for this metric to land in.  max is exclusive.
 func (m *Metric) Bucket(max int) int {
-	bucket := adler32.Checksum([]byte(m.Name))
-	bucket += adler32.Checksum([]byte(m.Hostname))
+	return Bucket(m.Name, m.Hostname, max)
+}
+
+func Bucket(metricName, hostname string, max int) int {
 	// Consider hashing the tags here too
-	bucket %= uint32(max)
-	return int(bucket)
+	bucket := adler32.Checksum([]byte(metricName))
+	bucket += adler32.Checksum([]byte(hostname))
+	return int(bucket % uint32(max))
 }
 
 func (m *Metric) String() string {
@@ -84,11 +87,18 @@ func (m *Metric) Done() {
 }
 
 func (m *Metric) FormatTagsKey() string {
-	t := m.Tags.SortedString()
-	if m.Hostname == "" {
+	if m.TagsKey == "" {
+		m.TagsKey = FormatTagsKey(m.Hostname, m.Tags)
+	}
+	return m.TagsKey
+}
+
+func FormatTagsKey(hostname string, tags Tags) string {
+	t := tags.SortedString()
+	if hostname == "" {
 		return t
 	}
-	return t + "," + StatsdSourceID + ":" + m.Hostname
+	return t + "," + StatsdSourceID + ":" + hostname
 }
 
 // AggregatedMetrics is an interface for aggregated metrics.
diff --git a/pkg/statsd/aggregator.go b/pkg/statsd/aggregator.go
index f729df8..d9a4144 100644
--- a/pkg/statsd/aggregator.go
+++ b/pkg/statsd/aggregator.go
@@ -23,13 +23,14 @@ type percentStruct struct {
 
 // MetricAggregator aggregates metrics.
 type MetricAggregator struct {
-	metricsReceived   uint64
-	expiryInterval    time.Duration // How often to expire metrics
-	percentThresholds map[float64]percentStruct
-	now               func() time.Time // Returns current time. Useful for testing.
-	statser           stats.Statser
-	disabledSubtypes  gostatsd.TimerSubtypes
-	metricMap         *gostatsd.MetricMap
+	metricsReceived    uint64
+	metricMapsReceived uint64
+	expiryInterval     time.Duration // How often to expire metrics
+	percentThresholds  map[float64]percentStruct
+	now                func() time.Time // Returns current time. Useful for testing.
+	statser            stats.Statser
+	disabledSubtypes   gostatsd.TimerSubtypes
+	metricMap          *gostatsd.MetricMap
 }
 
 // NewMetricAggregator creates a new MetricAggregator object.
@@ -65,6 +66,7 @@ func round(v float64) float64 {
 // Flush prepares the contents of a MetricAggregator for sending via the Sender.
 func (a *MetricAggregator) Flush(flushInterval time.Duration) {
 	a.statser.Gauge("aggregator.metrics_received", float64(a.metricsReceived), nil)
+	a.statser.Gauge("aggregator.metricmaps_received", float64(a.metricMapsReceived), nil)
 
 	flushInSeconds := float64(flushInterval) / float64(time.Second)
 
@@ -241,7 +243,14 @@ func (a *MetricAggregator) Reset() {
 }
 
 // Receive aggregates an incoming metric.
-func (a *MetricAggregator) Receive(m *gostatsd.Metric) {
-	a.metricsReceived++
-	a.metricMap.Receive(m)
+func (a *MetricAggregator) Receive(ms ...*gostatsd.Metric) {
+	a.metricsReceived += uint64(len(ms))
+	for _, m := range ms {
+		a.metricMap.Receive(m)
+	}
+}
+
+func (a *MetricAggregator) ReceiveMap(mm *gostatsd.MetricMap) {
+	a.metricMapsReceived++
+	a.metricMap.Merge(mm)
 }
diff --git a/pkg/statsd/handler_backend.go b/pkg/statsd/handler_backend.go
index 687ede3..9ce0746 100644
--- a/pkg/statsd/handler_backend.go
+++ b/pkg/statsd/handler_backend.go
@@ -43,10 +43,12 @@ func NewBackendHandler(backends []gostatsd.Backend, maxConcurrentEvents uint, nu
 
 	for i := 0; i < numWorkers; i++ {
 		workers[i] = &worker{
-			aggr:         af.Create(),
-			metricsQueue: make(chan *gostatsd.Metric, perWorkerBufferSize),
-			processChan:  make(chan *processCommand),
-			id:           i,
+			aggr: af.Create(),
+			// TODO: Reassess the defaults
+			metricsQueue:   make(chan []*gostatsd.Metric, perWorkerBufferSize),
+			metricMapQueue: make(chan *gostatsd.MetricMap, perWorkerBufferSize),
+			processChan:    make(chan *processCommand),
+			id:             i,
 		}
 	}
 
@@ -64,7 +66,8 @@ func (bh *BackendHandler) Run(ctx context.Context) {
 	var wg wait.Group
 	defer func() {
 		for _, worker := range bh.workers {
-			close(worker.metricsQueue) // Close channel to terminate worker
+			close(worker.metricsQueue)   // Close channel to terminate worker
+			close(worker.metricMapQueue) // Close channel to terminate worker
 		}
 		wg.Wait() // Wait for all workers to finish
 	}()
@@ -110,12 +113,34 @@ func (bh *BackendHandler) EstimatedTags() int {
 
 // DispatchMetrics dispatches metric to a corresponding Aggregator.
 func (bh *BackendHandler) DispatchMetrics(ctx context.Context, metrics []*gostatsd.Metric) {
+	metricsByAggr := make([][]*gostatsd.Metric, bh.numWorkers)
+
 	for _, m := range metrics {
 		m.TagsKey = m.FormatTagsKey() // this is expensive, so do it with no aggregator affinity
-		w := bh.workers[m.Bucket(bh.numWorkers)]
+		bucket := m.Bucket(bh.numWorkers)
+		metricsByAggr[bucket] = append(metricsByAggr[bucket], m)
+	}
+
+	for aggrIdx, bucketedMetrics := range metricsByAggr {
+		w := bh.workers[aggrIdx]
 		select {
 		case <-ctx.Done():
-		case w.metricsQueue <- m:
+		case w.metricsQueue <- bucketedMetrics:
+		}
+	}
+}
+
+// DispatchMetricMap re-dispatches a metric map through BackendHandler.DispatchMetrics
+func (bh *BackendHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
+	maps := mm.Split(bh.numWorkers)
+
+	for aggrIdx, mmSplit := range maps {
+		if !mm.IsEmpty() {
+			w := bh.workers[aggrIdx]
+			select {
+			case <-ctx.Done():
+			case w.metricMapQueue <- mmSplit:
+			}
 		}
 	}
 }
diff --git a/pkg/statsd/handler_cloud.go b/pkg/statsd/handler_cloud.go
index eb7ed50..9420c81 100644
--- a/pkg/statsd/handler_cloud.go
+++ b/pkg/statsd/handler_cloud.go
@@ -127,6 +127,14 @@ func (ch *CloudHandler) DispatchMetrics(ctx context.Context, metrics []*gostatsd
 	}
 }
 
+// DispatchMetricMap re-dispatches a metric map through CloudHandler.DispatchMetrics
+// TODO: This is inefficient, and should be handled first class, however that is a major re-factor of
+//  the CloudHandler.  It is also recommended to not use a CloudHandler in an http receiver based
+//  service, as the IP is not propagated.
+func (ch *CloudHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
+	mm.DispatchMetrics(ctx, ch)
+}
+
 func (ch *CloudHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	if ch.updateTagsAndHostname(e.SourceIP, &e.Tags, &e.Hostname) {
 		atomic.AddUint64(&ch.statsCacheHit, 1)
diff --git a/pkg/statsd/handler_http_forwarder_v2.go b/pkg/statsd/handler_http_forwarder_v2.go
index 8ba5ac9..23457eb 100644
--- a/pkg/statsd/handler_http_forwarder_v2.go
+++ b/pkg/statsd/handler_http_forwarder_v2.go
@@ -173,6 +173,11 @@ func (hfh *HttpForwarderHandlerV2) DispatchMetrics(ctx context.Context, metrics
 	hfh.consolidator.ReceiveMetrics(metrics)
 }
 
+// DispatchMetricMap re-dispatches a metric map through HttpForwarderHandlerV2.DispatchMetrics
+func (hfh *HttpForwarderHandlerV2) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
+	hfh.consolidator.ReceiveMetricMap(mm)
+}
+
 func (hfh *HttpForwarderHandlerV2) RunMetrics(ctx context.Context) {
 	statser := stats.FromContext(ctx)
 
diff --git a/pkg/statsd/handler_tags.go b/pkg/statsd/handler_tags.go
index b26929e..da1ce7c 100644
--- a/pkg/statsd/handler_tags.go
+++ b/pkg/statsd/handler_tags.go
@@ -67,7 +67,93 @@ func (th *TagHandler) DispatchMetrics(ctx context.Context, metrics []*gostatsd.M
 	}
 }
 
-// uniqueFilterMetricAndAddTags will perform 3 tasks:
+// DispatchMetricMap adds the unique tags from the TagHandler to each consolidated metric in the map and passes it to
+// the next stage in the pipeline
+//
+// There is potential to optimize here: if the tagsKey doesn't change, we don't need to re-calculate it.  But we're
+// keeping things simple for now.
+func (th *TagHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
+	mmNew := gostatsd.NewMetricMap()
+
+	mm.Counters.Each(func(metricName, x string, cOriginal gostatsd.Counter) {
+		if th.uniqueFilterAndAddTags(metricName, &cOriginal.Hostname, &cOriginal.Tags) {
+			newTagsKey := gostatsd.FormatTagsKey(cOriginal.Hostname, cOriginal.Tags)
+			if cs, ok := mmNew.Counters[metricName]; ok {
+				if cNew, ok := cs[newTagsKey]; ok {
+					cNew.Value += cOriginal.Value
+					cNew.Timestamp = gostatsd.NanoMax(cNew.Timestamp, cOriginal.Timestamp)
+					cs[newTagsKey] = cNew
+				} else {
+					cs[newTagsKey] = cOriginal
+				}
+			} else {
+				mmNew.Counters[metricName] = map[string]gostatsd.Counter{newTagsKey: cOriginal}
+			}
+		}
+	})
+
+	mm.Gauges.Each(func(metricName, _ string, gOriginal gostatsd.Gauge) {
+		if th.uniqueFilterAndAddTags(metricName, &gOriginal.Hostname, &gOriginal.Tags) {
+			newTagsKey := gostatsd.FormatTagsKey(gOriginal.Hostname, gOriginal.Tags)
+			if gs, ok := mmNew.Gauges[metricName]; ok {
+				if gNew, ok := gs[newTagsKey]; ok {
+					if gOriginal.Timestamp > gNew.Timestamp {
+						gNew.Value = gOriginal.Value
+						gNew.Timestamp = gOriginal.Timestamp
+						gs[newTagsKey] = gNew
+					}
+				} else {
+					gs[newTagsKey] = gOriginal
+				}
+			} else {
+				mmNew.Gauges[metricName] = map[string]gostatsd.Gauge{newTagsKey: gOriginal}
+			}
+		}
+	})
+
+	mm.Timers.Each(func(metricName, _ string, tOriginal gostatsd.Timer) {
+		if th.uniqueFilterAndAddTags(metricName, &tOriginal.Hostname, &tOriginal.Tags) {
+			newTagsKey := gostatsd.FormatTagsKey(tOriginal.Hostname, tOriginal.Tags)
+			if ts, ok := mmNew.Timers[metricName]; ok {
+				if tNew, ok := ts[newTagsKey]; ok {
+					tNew.Values = append(tNew.Values, tOriginal.Values...)
+					tNew.Timestamp = gostatsd.NanoMax(tNew.Timestamp, tOriginal.Timestamp)
+					tNew.SampledCount += tOriginal.SampledCount
+					ts[newTagsKey] = tNew
+				} else {
+					ts[newTagsKey] = tOriginal
+				}
+			} else {
+				mmNew.Timers[metricName] = map[string]gostatsd.Timer{newTagsKey: tOriginal}
+			}
+		}
+	})
+
+	mm.Sets.Each(func(metricName, _ string, sOriginal gostatsd.Set) {
+		if th.uniqueFilterAndAddTags(metricName, &sOriginal.Hostname, &sOriginal.Tags) {
+			newTagsKey := gostatsd.FormatTagsKey(sOriginal.Hostname, sOriginal.Tags)
+			if ss, ok := mmNew.Sets[metricName]; ok {
+				if sNew, ok := ss[newTagsKey]; ok {
+					for key := range sOriginal.Values {
+						sNew.Values[key] = struct{}{}
+					}
+					sNew.Timestamp = gostatsd.NanoMax(sNew.Timestamp, sOriginal.Timestamp)
+					ss[newTagsKey] = sNew
+				} else {
+					ss[newTagsKey] = sOriginal
+				}
+			} else {
+				mmNew.Sets[metricName] = map[string]gostatsd.Set{newTagsKey: sOriginal}
+			}
+		}
+	})
+
+	if !mmNew.IsEmpty() {
+		th.handler.DispatchMetricMap(ctx, mmNew)
+	}
+}
+
+// uniqueFilterAndAddTags will perform 3 tasks:
 // - Add static tags configured to the metric
 // - De-duplicate tags
 // - Perform rule based filtering
@@ -76,27 +162,27 @@ func (th *TagHandler) DispatchMetrics(ctx context.Context, metrics []*gostatsd.M
 // hot code path.
 //
 // Returns true if the metric should be processed further, or false to drop it.
-func (th *TagHandler) uniqueFilterMetricAndAddTags(m *gostatsd.Metric) bool {
+func (th *TagHandler) uniqueFilterAndAddTags(mName string, mHostname *string, mTags *gostatsd.Tags) bool {
 	if len(th.filters) == 0 {
-		m.Tags = uniqueTags(m.Tags, th.tags)
+		*mTags = uniqueTags(*mTags, th.tags)
 		return true
 	}
 
 	dropTags := map[string]struct{}{}
 
 	for _, filter := range th.filters {
-		if len(filter.MatchMetrics) > 0 && !filter.MatchMetrics.MatchAny(m.Name) { // returns false if nothing present
+		if len(filter.MatchMetrics) > 0 && !filter.MatchMetrics.MatchAny(mName) { // returns false if nothing present
 			// name doesn't match an include, stop
 			continue
 		}
 
 		// this list may be empty, and therefore return false
-		if filter.ExcludeMetrics.MatchAny(m.Name) { // returns false if nothing present
+		if filter.ExcludeMetrics.MatchAny(mName) { // returns false if nothing present
 			// name matches an exclude, stop
 			continue
 		}
 
-		if len(filter.MatchTags) > 0 && !filter.MatchTags.MatchAnyMultiple(m.Tags) { // returns false if either list is empty
+		if len(filter.MatchTags) > 0 && !filter.MatchTags.MatchAnyMultiple(*mTags) { // returns false if either list is empty
 			// no tags match
 			continue
 		}
@@ -106,7 +192,7 @@ func (th *TagHandler) uniqueFilterMetricAndAddTags(m *gostatsd.Metric) bool {
 		}
 
 		for _, dropFilter := range filter.DropTags {
-			for _, tag := range m.Tags {
+			for _, tag := range *mTags {
 				if dropFilter.Match(tag) {
 					dropTags[tag] = present
 				}
@@ -114,14 +200,18 @@ func (th *TagHandler) uniqueFilterMetricAndAddTags(m *gostatsd.Metric) bool {
 		}
 
 		if filter.DropHost {
-			m.Hostname = ""
+			*mHostname = ""
 		}
 	}
 
-	m.Tags = uniqueTagsWithSeen(dropTags, m.Tags, th.tags)
+	*mTags = uniqueTagsWithSeen(dropTags, *mTags, th.tags)
 	return true
 }
 
+func (th *TagHandler) uniqueFilterMetricAndAddTags(m *gostatsd.Metric) bool {
+	return th.uniqueFilterAndAddTags(m.Name, &m.Hostname, &m.Tags)
+}
+
 // DispatchEvent adds the unique tags from the TagHandler to the event and passes it to the next stage in the pipeline
 func (th *TagHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	if e.Hostname == "" {
diff --git a/pkg/statsd/types.go b/pkg/statsd/types.go
index af18be0..b23c888 100644
--- a/pkg/statsd/types.go
+++ b/pkg/statsd/types.go
@@ -25,7 +25,8 @@ type ProcessFunc func(*gostatsd.MetricMap)
 //
 // Incoming metrics should be passed via Receive function.
 type Aggregator interface {
-	Receive(*gostatsd.Metric)
+	Receive(metrics ...*gostatsd.Metric)
+	ReceiveMap(mm *gostatsd.MetricMap)
 	Flush(interval time.Duration)
 	Process(ProcessFunc)
 	Reset()
diff --git a/pkg/statsd/worker.go b/pkg/statsd/worker.go
index d29a8ee..3d3e060 100644
--- a/pkg/statsd/worker.go
+++ b/pkg/statsd/worker.go
@@ -15,20 +15,26 @@ type processCommand struct {
 }
 
 type worker struct {
-	aggr         Aggregator
-	metricsQueue chan *gostatsd.Metric
-	processChan  chan *processCommand
-	id           int
+	aggr           Aggregator
+	metricsQueue   chan []*gostatsd.Metric
+	metricMapQueue chan *gostatsd.MetricMap
+	processChan    chan *processCommand
+	id             int
 }
 
 func (w *worker) work() {
 	for {
 		select {
-		case metric, ok := <-w.metricsQueue:
+		case metrics, ok := <-w.metricsQueue:
 			if !ok {
 				return
 			}
-			w.aggr.Receive(metric)
+			w.aggr.Receive(metrics...)
+		case mm, ok := <-w.metricMapQueue:
+			if !ok {
+				return
+			}
+			w.aggr.ReceiveMap(mm)
 		case cmd := <-w.processChan:
 			w.executeProcess(cmd)
 		}
diff --git a/pkg/web/http_receiver_v2.go b/pkg/web/http_receiver_v2.go
index 455d528..42dfabc 100644
--- a/pkg/web/http_receiver_v2.go
+++ b/pkg/web/http_receiver_v2.go
@@ -121,7 +121,7 @@ func (rhh *rawHttpHandlerV2) MetricHandler(w http.ResponseWriter, req *http.Requ
 	}
 
 	mm := translateFromProtobufV2(&msg)
-	mm.DispatchMetrics(req.Context(), rhh.handler)
+	rhh.handler.DispatchMetricMap(req.Context(), mm)
 
 	atomic.AddUint64(&rhh.requestSuccess, 1)
 	w.WriteHeader(http.StatusAccepted)
diff --git a/types.go b/types.go
index 0945a1d..19ab3dd 100644
--- a/types.go
+++ b/types.go
@@ -13,6 +13,13 @@ func NanoNow() Nanotime {
 	return Nanotime(time.Now().UnixNano())
 }
 
+func NanoMax(t1, t2 Nanotime) Nanotime {
+	if t1 > t2 {
+		return t1
+	}
+	return t2
+}
+
 // IP is a v4/v6 IP address.
 // We do not use net.IP because it will involve conversion to string and back several times.
 type IP string
@@ -52,6 +59,7 @@ type Runner interface {
 // pre-consolidation.
 type RawMetricHandler interface {
 	DispatchMetrics(ctx context.Context, m []*Metric)
+	DispatchMetricMap(ctx context.Context, mm *MetricMap)
 }
 
 // PipelineHandler can be used to handle metrics and events, it provides an estimate of how many tags it may add.


Test patch:
diff --git a/fixtures_test.go b/fixtures_test.go
index da1c8d3..1730d35 100644
--- a/fixtures_test.go
+++ b/fixtures_test.go
@@ -44,6 +44,11 @@ func (ch *capturingHandler) DispatchMetrics(ctx context.Context, metrics []*Metr
 	}
 }
 
+// DispatchMetricMap re-dispatches a metric map through capturingHandler.DispatchMetrics
+func (ch *capturingHandler) DispatchMetricMap(ctx context.Context, mm *MetricMap) {
+	mm.DispatchMetrics(ctx, ch)
+}
+
 func (ch *capturingHandler) DispatchEvent(ctx context.Context, e *Event) {
 	ch.mu.Lock()
 	defer ch.mu.Unlock()
diff --git a/metric_map_test.go b/metric_map_test.go
index 242803d..1765b40 100644
--- a/metric_map_test.go
+++ b/metric_map_test.go
@@ -266,3 +266,77 @@ func TestMetricMapMerge(t *testing.T) {
 	require.Equal(t, expected.Gauges, merged.Gauges)
 	require.Equal(t, expected.Sets, merged.Sets)
 }
+
+func TestMetricMapSplit(t *testing.T) {
+	mmOriginal := NewMetricMap()
+	mmOriginal.Counters["m"] = map[string]Counter{
+		"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Value: 10},
+		"t.s.h2": {Tags: Tags{"t"}, Hostname: "h2", Value: 20},
+		"t.s.h3": {Tags: Tags{"t"}, Hostname: "h3", Value: 30},
+		"t.s.h4": {Tags: Tags{"t"}, Hostname: "h4", Value: 40},
+		"t.s.h5": {Tags: Tags{"t"}, Hostname: "h5", Value: 50},
+	}
+	mmOriginal.Gauges["m"] = map[string]Gauge{
+		"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Value: 10},
+		"t.s.h2": {Tags: Tags{"t"}, Hostname: "h2", Value: 20},
+		"t.s.h3": {Tags: Tags{"t"}, Hostname: "h3", Value: 30},
+		"t.s.h4": {Tags: Tags{"t"}, Hostname: "h4", Value: 40},
+		"t.s.h5": {Tags: Tags{"t"}, Hostname: "h5", Value: 50},
+	}
+	mmOriginal.Timers["m"] = map[string]Timer{
+		"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Values: []float64{10, 50}},
+		"t.s.h2": {Tags: Tags{"t"}, Hostname: "h2", Values: []float64{20, 40}},
+		"t.s.h3": {Tags: Tags{"t"}, Hostname: "h3", Values: []float64{30, 30}},
+		"t.s.h4": {Tags: Tags{"t"}, Hostname: "h4", Values: []float64{40, 20}},
+		"t.s.h5": {Tags: Tags{"t"}, Hostname: "h5", Values: []float64{50, 10}},
+	}
+	mmOriginal.Sets["m"] = map[string]Set{
+		"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Values: map[string]struct{}{"10": {}, "50": {}}},
+		"t.s.h2": {Tags: Tags{"t"}, Hostname: "h2", Values: map[string]struct{}{"20": {}, "40": {}}},
+		"t.s.h3": {Tags: Tags{"t"}, Hostname: "h3", Values: map[string]struct{}{"30": {}, "3.0": {}}},
+		"t.s.h4": {Tags: Tags{"t"}, Hostname: "h4", Values: map[string]struct{}{"40": {}, "20": {}}},
+		"t.s.h5": {Tags: Tags{"t"}, Hostname: "h5", Values: map[string]struct{}{"50": {}, "10": {}}},
+	}
+
+	mmMerged := NewMetricMap()
+	mms := mmOriginal.Split(2)
+	for _, mmSplit := range mms {
+		// Make sure something landed in each (don't use mmSplit.IsEmpty() to ensure that all types are split)
+		require.True(t, len(mmSplit.Counters) > 0)
+		require.True(t, len(mmSplit.Gauges) > 0)
+		require.True(t, len(mmSplit.Timers) > 0)
+		require.True(t, len(mmSplit.Sets) > 0)
+		mmMerged.Merge(mmSplit)
+	}
+	// Make sure when merge back they are the same
+	require.EqualValues(t, mmOriginal, mmMerged)
+}
+
+func TestMetricMapIsEmpty(t *testing.T) {
+	mm := NewMetricMap()
+	require.True(t, mm.IsEmpty())
+
+	// Counter
+	mm.Counters["m"] = map[string]Counter{"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Value: 10}}
+	require.False(t, mm.IsEmpty())
+	mm.Counters.Delete("m")
+	require.True(t, mm.IsEmpty())
+
+	// Gauge
+	mm.Gauges["m"] = map[string]Gauge{"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Value: 10}}
+	require.False(t, mm.IsEmpty())
+	mm.Gauges.Delete("m")
+	require.True(t, mm.IsEmpty())
+
+	// Timer
+	mm.Timers["m"] = map[string]Timer{"t.s.h1": {Tags: Tags{"t"}, Hostname: "h1", Values: []float64{10}}}
+	require.False(t, mm.IsEmpty())
+	mm.Timers.Delete("m")
+	require.True(t, mm.IsEmpty())
+
+	// Set
+	mm.Sets["m"] = map[string]Set{"t.s.h1": {Tags: Tags{"t"}, Hostname: "h5", Values: map[string]struct{}{"10": {}}}}
+	require.False(t, mm.IsEmpty())
+	mm.Sets.Delete("m")
+	require.True(t, mm.IsEmpty())
+}
diff --git a/pkg/statsd/handler_backend_test.go b/pkg/statsd/handler_backend_test.go
index 0243846..62149a8 100644
--- a/pkg/statsd/handler_backend_test.go
+++ b/pkg/statsd/handler_backend_test.go
@@ -25,10 +25,16 @@ type testAggregator struct {
 func (a *testAggregator) TrackMetrics(statser stats.Statser) {
 }
 
-func (a *testAggregator) Receive(m *gostatsd.Metric) {
+func (a *testAggregator) Receive(m ...*gostatsd.Metric) {
 	a.af.Mutex.Lock()
 	defer a.af.Mutex.Unlock()
-	a.af.receiveInvocations[a.agrNumber]++
+	a.af.receiveInvocations[a.agrNumber] += len(m)
+}
+
+func (a *testAggregator) ReceiveMap(mm *gostatsd.MetricMap) {
+	a.af.Mutex.Lock()
+	defer a.af.Mutex.Unlock()
+	a.af.receiveMapInvocations[a.agrNumber]++
 }
 
 func (a *testAggregator) Flush(interval time.Duration) {
@@ -52,17 +58,19 @@ func (a *testAggregator) Reset() {
 
 type testAggregatorFactory struct {
 	sync.Mutex
-	receiveInvocations map[int]int
-	flushInvocations   map[int]int
-	processInvocations map[int]int
-	resetInvocations   map[int]int
-	numAgrs            int
+	receiveInvocations    map[int]int
+	receiveMapInvocations map[int]int
+	flushInvocations      map[int]int
+	processInvocations    map[int]int
+	resetInvocations      map[int]int
+	numAgrs               int
 }
 
 func (af *testAggregatorFactory) Create() Aggregator {
 	agrNumber := af.numAgrs
 	af.numAgrs++
 	af.receiveInvocations[agrNumber] = 0
+	af.receiveMapInvocations[agrNumber] = 0
 	af.flushInvocations[agrNumber] = 0
 	af.processInvocations[agrNumber] = 0
 	af.resetInvocations[agrNumber] = 0
@@ -81,10 +89,11 @@ func (af *testAggregatorFactory) Create() Aggregator {
 
 func newTestFactory() *testAggregatorFactory {
 	return &testAggregatorFactory{
-		receiveInvocations: make(map[int]int),
-		flushInvocations:   make(map[int]int),
-		processInvocations: make(map[int]int),
-		resetInvocations:   make(map[int]int),
+		receiveInvocations:    make(map[int]int),
+		receiveMapInvocations: make(map[int]int),
+		flushInvocations:      make(map[int]int),
+		processInvocations:    make(map[int]int),
+		resetInvocations:      make(map[int]int),
 	}
 }
 
@@ -106,7 +115,7 @@ func TestRunShouldReturnWhenContextCancelled(t *testing.T) {
 	h.Run(ctx)
 }
 
-func TestDispatchMetricShouldDistributeMetrics(t *testing.T) {
+func TestDispatchMetricsShouldDistributeMetrics(t *testing.T) {
 	t.Parallel()
 	r := rand.New(rand.NewSource(time.Now().UnixNano()))
 	n := r.Intn(5) + 1
@@ -146,6 +155,48 @@ func TestDispatchMetricShouldDistributeMetrics(t *testing.T) {
 	}
 }
 
+func TestDispatchMetricMapShouldDistributeMetrics(t *testing.T) {
+	t.Parallel()
+	r := rand.New(rand.NewSource(time.Now().UnixNano()))
+	numAggregators := r.Intn(5) + 1
+	factory := newTestFactory()
+	// use a sync channel to force the workers to process events before the context is cancelled
+	h := NewBackendHandler(nil, 0, numAggregators, 0, factory)
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+	var wgFinish wait.Group
+	wgFinish.StartWithContext(ctx, h.Run)
+
+	mm := gostatsd.NewMetricMap()
+	for i := 0; i < numAggregators*100; i++ {
+		m := &gostatsd.Metric{
+			Type:  gostatsd.COUNTER,
+			Name:  fmt.Sprintf("counter.metric.%d", r.Int63()),
+			Tags:  nil,
+			Value: r.Float64(),
+		}
+		m.TagsKey = m.FormatTagsKey()
+		mm.Receive(m)
+	}
+
+	h.DispatchMetricMap(ctx, mm)
+
+	cancelFunc()    // After dispatch, we signal dispatcher to shut down
+	wgFinish.Wait() // Wait for dispatcher to shutdown
+
+	for agrNum, count := range factory.receiveMapInvocations {
+		assert.NotZerof(t, count, "aggregator=%d", agrNum)
+		if count == 0 {
+			t.Errorf("aggregator %d was never invoked", agrNum)
+			for idx, mmSplit := range mm.Split(numAggregators) {
+				fmt.Printf("aggr %d, names %d\n", idx, len(mmSplit.Counters))
+			}
+		} else {
+			t.Logf("aggregator %d was invoked %d time(s)", agrNum, count)
+		}
+	}
+}
+
 func getTotalInvocations(inv map[int]int) int {
 	var counter int
 	for _, i := range inv {
diff --git a/pkg/statsd/handler_fixtures_test.go b/pkg/statsd/handler_fixtures_test.go
index bddfbd1..0b79faf 100644
--- a/pkg/statsd/handler_fixtures_test.go
+++ b/pkg/statsd/handler_fixtures_test.go
@@ -8,8 +8,9 @@ import (
 )
 
 type capturingHandler struct {
-	m []*gostatsd.Metric
-	e []*gostatsd.Event
+	m  []*gostatsd.Metric
+	mm []*gostatsd.MetricMap
+	e  []*gostatsd.Event
 }
 
 func (tch *capturingHandler) EstimatedTags() int {
@@ -20,6 +21,10 @@ func (tch *capturingHandler) DispatchMetrics(ctx context.Context, metrics []*gos
 	tch.m = append(tch.m, metrics...)
 }
 
+func (tch *capturingHandler) DispatchMetricMap(ctx context.Context, metrics *gostatsd.MetricMap) {
+	tch.mm = append(tch.mm, metrics)
+}
+
 func (tch *capturingHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	tch.e = append(tch.e, e)
 }
@@ -36,6 +41,9 @@ func (nh *nopHandler) EstimatedTags() int {
 func (nh *nopHandler) DispatchMetrics(ctx context.Context, m []*gostatsd.Metric) {
 }
 
+func (nh *nopHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
+}
+
 func (nh *nopHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 }
 
@@ -61,6 +69,11 @@ func (ch *countingHandler) DispatchMetrics(ctx context.Context, metrics []*gosta
 	}
 }
 
+// DispatchMetricMap re-dispatches a metric map through BackendHandler.DispatchMetrics
+func (ch *countingHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
+	mm.DispatchMetrics(ctx, ch)
+}
+
 func (ch *countingHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	ch.mu.Lock()
 	defer ch.mu.Unlock()
diff --git a/pkg/statsd/handler_http_forwarder_v2_test.go b/pkg/statsd/handler_http_forwarder_v2_test.go
index 767001e..9aa6e08 100644
--- a/pkg/statsd/handler_http_forwarder_v2_test.go
+++ b/pkg/statsd/handler_http_forwarder_v2_test.go
@@ -80,7 +80,6 @@ func TestHttpForwarderV2Translation(t *testing.T) {
 
 	mm := gostatsd.NewMetricMap()
 	for _, metric := range metrics {
-		metric.TagsKey = metric.FormatTagsKey()
 		mm.Receive(metric)
 	}
 
diff --git a/pkg/statsd/handler_tags_test.go b/pkg/statsd/handler_tags_test.go
index 6bba1c8..2d2da4e 100644
--- a/pkg/statsd/handler_tags_test.go
+++ b/pkg/statsd/handler_tags_test.go
@@ -3,6 +3,7 @@ package statsd
 import (
 	"bytes"
 	"context"
+	"sort"
 	"strings"
 	"testing"
 
@@ -10,8 +11,80 @@ import (
 
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
+func TestTagStripMergesCounters(t *testing.T) {
+	tch := &capturingHandler{}
+	th := NewTagHandler(tch, gostatsd.Tags{}, []Filter{
+		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
+	})
+	mm := gostatsd.NewMetricMap()
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.COUNTER, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value", "key2:value2"}, Value: 1, Rate: 0.1})
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.COUNTER, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value"}, Value: 20, Rate: 1})
+	th.DispatchMetricMap(context.Background(), mm)
+
+	expected := gostatsd.NewMetricMap()
+	expected.Counters["metric"] = map[string]gostatsd.Counter{
+		"key:value": {Timestamp: 20, Value: 30, Tags: gostatsd.Tags{"key:value"}},
+	}
+	require.EqualValues(t, expected, tch.mm[0])
+}
+
+func TestTagStripMergesGauges(t *testing.T) {
+	tch := &capturingHandler{}
+	th := NewTagHandler(tch, gostatsd.Tags{}, []Filter{
+		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
+	})
+	mm := gostatsd.NewMetricMap()
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.GAUGE, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value", "key2:value2"}, Value: 10})
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.GAUGE, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value"}, Value: 20})
+	th.DispatchMetricMap(context.Background(), mm)
+
+	expected := gostatsd.NewMetricMap()
+	expected.Gauges["metric"] = map[string]gostatsd.Gauge{
+		"key:value": {Timestamp: 20, Value: 20, Tags: gostatsd.Tags{"key:value"}},
+	}
+	require.EqualValues(t, expected, tch.mm[0])
+}
+
+func TestTagStripMergesTimers(t *testing.T) {
+	tch := &capturingHandler{}
+	th := NewTagHandler(tch, gostatsd.Tags{}, []Filter{
+		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
+	})
+	mm := gostatsd.NewMetricMap()
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.TIMER, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value", "key2:value2"}, Value: 10, Rate: 1})
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.TIMER, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value"}, Value: 20, Rate: 1})
+	th.DispatchMetricMap(context.Background(), mm)
+
+	// Make sure the actual values are deterministic
+	sort.Float64s(tch.mm[0].Timers["metric"]["key:value"].Values)
+
+	expected := gostatsd.NewMetricMap()
+	expected.Timers["metric"] = map[string]gostatsd.Timer{
+		"key:value": {Timestamp: 20, Values: []float64{10, 20}, Tags: gostatsd.Tags{"key:value"}, SampledCount: 2},
+	}
+	require.EqualValues(t, expected, tch.mm[0])
+}
+
+func TestTagStripMergesSets(t *testing.T) {
+	tch := &capturingHandler{}
+	th := NewTagHandler(tch, gostatsd.Tags{}, []Filter{
+		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
+	})
+	mm := gostatsd.NewMetricMap()
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.SET, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value", "key2:value2"}, StringValue: "abc"})
+	mm.Receive(&gostatsd.Metric{Type: gostatsd.SET, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value"}, StringValue: "def"})
+	th.DispatchMetricMap(context.Background(), mm)
+
+	expected := gostatsd.NewMetricMap()
+	expected.Sets["metric"] = map[string]gostatsd.Set{
+		"key:value": {Timestamp: 20, Values: map[string]struct{}{"abc": {}, "def": {}}, Tags: gostatsd.Tags{"key:value"}},
+	}
+	require.EqualValues(t, expected, tch.mm[0])
+}
+
 func TestFilterPassesNoFilters(t *testing.T) {
 	tch := &capturingHandler{}
 	th := NewTagHandler(tch, gostatsd.Tags{}, nil)
diff --git a/pkg/web/fixtures_test.go b/pkg/web/fixtures_test.go
index 24bb1da..ae4b1ed 100644
--- a/pkg/web/fixtures_test.go
+++ b/pkg/web/fixtures_test.go
@@ -30,6 +30,10 @@ func (ch *capturingHandler) DispatchMetrics(ctx context.Context, metrics []*gost
 	}
 }
 
+func (ch *capturingHandler) DispatchMetricMap(ctx context.Context, metrics *gostatsd.MetricMap) {
+	metrics.DispatchMetrics(ctx, ch)
+}
+
 func (ch *capturingHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	ch.mu.Lock()
 	defer ch.mu.Unlock()