Repository: atlassian/gostatsd
Pull request #459 (2022-02-07 02:32:29)

Track forwarded metrics in preparation for clustering
This is a very noisy change, so it lives on its own.

Note: there is currently no source of forwarded metrics, it's purely foundational.


Patch:
diff --git a/metric_consolidator.go b/metric_consolidator.go
index 8dfe306..60def65 100644
--- a/metric_consolidator.go
+++ b/metric_consolidator.go
@@ -27,14 +27,19 @@ type MetricConsolidator struct {
 	maps          chan *MetricMap
 	sink          chan<- []*MetricMap
 	flushInterval time.Duration
+	forwarded     bool
 }
 
-func NewMetricConsolidator(spots int, flushInterval time.Duration, sink chan<- []*MetricMap) *MetricConsolidator {
-	mc := &MetricConsolidator{}
-	mc.maps = make(chan *MetricMap, spots)
+// NewMetricConsolidator returns a new MetricConsolidator, all MetricMap's generated by it will propagate
+// the forwarded value from the provided forwarded parameter, and not from any source MetricMap.
+func NewMetricConsolidator(spots int, forwarded bool, flushInterval time.Duration, sink chan<- []*MetricMap) *MetricConsolidator {
+	mc := &MetricConsolidator{
+		maps:          make(chan *MetricMap, spots),
+		sink:          sink,
+		flushInterval: flushInterval,
+		forwarded:     forwarded,
+	}
 	mc.Fill()
-	mc.flushInterval = flushInterval
-	mc.sink = sink
 	return mc
 }
 
@@ -93,7 +98,7 @@ func (mc *MetricConsolidator) Flush() {
 // must be called after a successful Drain[WithContext], must not be called after a failed DrainWithContext.
 func (mc *MetricConsolidator) Fill() {
 	for i := 0; i < cap(mc.maps); i++ {
-		mc.maps <- NewMetricMap()
+		mc.maps <- NewMetricMap(mc.forwarded)
 	}
 }
 
diff --git a/metric_map.go b/metric_map.go
index 6b5dbe2..0fec28c 100644
--- a/metric_map.go
+++ b/metric_map.go
@@ -15,10 +15,14 @@ type MetricMap struct {
 	Timers   Timers
 	Gauges   Gauges
 	Sets     Sets
+
+	Forwarded bool
 }
 
-func NewMetricMap() *MetricMap {
+func NewMetricMap(forwarded bool) *MetricMap {
 	return &MetricMap{
+		Forwarded: forwarded,
+
 		Counters: Counters{},
 		Timers:   Timers{},
 		Gauges:   Gauges{},
@@ -46,7 +50,10 @@ func (mm *MetricMap) Receive(m *Metric) {
 }
 
 func MergeMaps(mms []*MetricMap) *MetricMap {
-	mm := NewMetricMap()
+	if len(mms) == 0 { // garbage in, garbage out
+		return nil
+	}
+	mm := NewMetricMap(mms[0].Forwarded)
 	for _, mmFrom := range mms {
 		mm.Merge(mmFrom)
 	}
@@ -151,7 +158,7 @@ func (mm *MetricMap) IsEmpty() bool {
 func (mm *MetricMap) Split(count int) []*MetricMap {
 	maps := make([]*MetricMap, count)
 	for i := 0; i < count; i++ {
-		maps[i] = NewMetricMap()
+		maps[i] = NewMetricMap(mm.Forwarded)
 	}
 
 	mm.Counters.Each(func(metricName string, tagsKey string, c Counter) {
@@ -215,7 +222,7 @@ func (mm *MetricMap) SplitByTags(tagNames []string) map[string]*MetricMap {
 	mm.Counters.Each(func(metricName string, tagsKey string, c Counter) {
 		key := tagsMatch(tagNames, tagsKey)
 		if _, ok := maps[key]; !ok {
-			maps[key] = NewMetricMap()
+			maps[key] = NewMetricMap(mm.Forwarded)
 		}
 		mmSplit := maps[key]
 		if v, ok := mmSplit.Counters[metricName]; ok {
@@ -228,7 +235,7 @@ func (mm *MetricMap) SplitByTags(tagNames []string) map[string]*MetricMap {
 	mm.Gauges.Each(func(metricName string, tagsKey string, g Gauge) {
 		key := tagsMatch(tagNames, tagsKey)
 		if _, ok := maps[key]; !ok {
-			maps[key] = NewMetricMap()
+			maps[key] = NewMetricMap(mm.Forwarded)
 		}
 		mmSplit := maps[key]
 		if v, ok := mmSplit.Gauges[metricName]; ok {
@@ -241,7 +248,7 @@ func (mm *MetricMap) SplitByTags(tagNames []string) map[string]*MetricMap {
 	mm.Timers.Each(func(metricName string, tagsKey string, t Timer) {
 		key := tagsMatch(tagNames, tagsKey)
 		if _, ok := maps[key]; !ok {
-			maps[key] = NewMetricMap()
+			maps[key] = NewMetricMap(mm.Forwarded)
 		}
 		mmSplit := maps[key]
 		if v, ok := mmSplit.Timers[metricName]; ok {
@@ -254,7 +261,7 @@ func (mm *MetricMap) SplitByTags(tagNames []string) map[string]*MetricMap {
 	mm.Sets.Each(func(metricName string, tagsKey string, s Set) {
 		key := tagsMatch(tagNames, tagsKey)
 		if _, ok := maps[key]; !ok {
-			maps[key] = NewMetricMap()
+			maps[key] = NewMetricMap(mm.Forwarded)
 		}
 		mmSplit := maps[key]
 		if v, ok := mmSplit.Sets[metricName]; ok {
diff --git a/pkg/stats/statser_internal.go b/pkg/stats/statser_internal.go
index 07f70c1..8316631 100644
--- a/pkg/stats/statser_internal.go
+++ b/pkg/stats/statser_internal.go
@@ -39,7 +39,7 @@ func NewInternalStatser(tags gostatsd.Tags, namespace string, hostname gostatsd.
 		handler:   handler,
 		// We can't just use a MetricMap because everything
 		// that writes to it is on its own goroutine.
-		consolidator: gostatsd.NewMetricConsolidator(10, 0, nil),
+		consolidator: gostatsd.NewMetricConsolidator(10, false, 0, nil),
 	}
 }
 
diff --git a/pkg/statsd/aggregator.go b/pkg/statsd/aggregator.go
index 5e10372..eb065a5 100644
--- a/pkg/statsd/aggregator.go
+++ b/pkg/statsd/aggregator.go
@@ -55,7 +55,7 @@ func NewMetricAggregator(
 		percentThresholds: make(map[float64]percentStruct, len(percentThresholds)),
 		now:               time.Now,
 		statser:           stats.NewNullStatser(), // Will probably be replaced via RunMetrics
-		metricMap:         gostatsd.NewMetricMap(),
+		metricMap:         gostatsd.NewMetricMap(false),
 		disabledSubtypes:  disabled,
 		histogramLimit:    histogramLimit,
 	}
diff --git a/pkg/statsd/handler_cloud.go b/pkg/statsd/handler_cloud.go
index d39cf4e..9af9cb6 100644
--- a/pkg/statsd/handler_cloud.go
+++ b/pkg/statsd/handler_cloud.go
@@ -57,8 +57,8 @@ func (ch *CloudHandler) EstimatedTags() int {
 }
 
 func (ch *CloudHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
-	mmToDispatch := gostatsd.NewMetricMap()
-	mmToHandle := gostatsd.NewMetricMap()
+	mmToDispatch := gostatsd.NewMetricMap(false)
+	mmToHandle := gostatsd.NewMetricMap(false)
 	mm.Counters.Each(func(metricName string, tagsKey string, c gostatsd.Counter) {
 		if ch.updateTagsAndHostname(&c, c.Source) {
 			mmToDispatch.MergeCounter(metricName, gostatsd.FormatTagsKey(c.Source, c.Tags), c)
@@ -229,7 +229,7 @@ func (ch *CloudHandler) prepareMetricQueue(source gostatsd.Source) *gostatsd.Met
 		ch.toLookupIPs = append(ch.toLookupIPs, source)
 		ch.statsMetricHostsQueued++
 	}
-	queue := gostatsd.NewMetricMap()
+	queue := gostatsd.NewMetricMap(false)
 	ch.awaitingMetrics[source] = queue
 	return queue
 }
@@ -261,7 +261,7 @@ func (ch *CloudHandler) handleIncomingEvent(e *gostatsd.Event) {
 }
 
 func (ch *CloudHandler) updateAndDispatchMetrics(ctx context.Context, instance *gostatsd.Instance, mmIn *gostatsd.MetricMap) {
-	mmOut := gostatsd.NewMetricMap()
+	mmOut := gostatsd.NewMetricMap(false)
 	mmIn.Counters.Each(func(metricName string, tagsKey string, c gostatsd.Counter) {
 		updateInplace(&c, instance)
 		mmOut.MergeCounter(metricName, gostatsd.FormatTagsKey(c.Source, c.Tags), c)
diff --git a/pkg/statsd/handler_http_forwarder_v2.go b/pkg/statsd/handler_http_forwarder_v2.go
index 89fb243..ab66298 100644
--- a/pkg/statsd/handler_http_forwarder_v2.go
+++ b/pkg/statsd/handler_http_forwarder_v2.go
@@ -170,7 +170,7 @@ func NewHttpForwarderHandlerV2(
 		maxRequestElapsedTime: maxRequestElapsedTime,
 		metricsSem:            metricsSem,
 		compress:              compress,
-		consolidator:          gostatsd.NewMetricConsolidator(consolidatorSlots, flushInterval, ch),
+		consolidator:          gostatsd.NewMetricConsolidator(consolidatorSlots, false, flushInterval, ch),
 		consolidatedMetrics:   ch,
 		client:                httpClient.Client,
 		headers:               headers,
@@ -229,7 +229,7 @@ func (hfh *HttpForwarderHandlerV2) Run(ctx context.Context) {
 	var wg wait.Group
 	wg.Start(func() {
 		for metricMaps := range hfh.consolidatedMetrics {
-			mergedMetricMap := mergeMaps(metricMaps)
+			mergedMetricMap := gostatsd.MergeMaps(metricMaps)
 			mms := mergedMetricMap.SplitByTags(hfh.dynHeaderNames)
 			for dynHeaderTags, mm := range mms {
 				if mm.IsEmpty() {
@@ -262,14 +262,6 @@ func (hfh *HttpForwarderHandlerV2) Close() {
 	close(hfh.consolidatedMetrics)
 }
 
-func mergeMaps(maps []*gostatsd.MetricMap) *gostatsd.MetricMap {
-	mm := gostatsd.NewMetricMap()
-	for _, m := range maps {
-		mm.Merge(m)
-	}
-	return mm
-}
-
 func (hfh *HttpForwarderHandlerV2) acquireSem() {
 	<-hfh.metricsSem // will potentially block
 }
diff --git a/pkg/statsd/handler_tags.go b/pkg/statsd/handler_tags.go
index 432ed69..7a6e0c7 100644
--- a/pkg/statsd/handler_tags.go
+++ b/pkg/statsd/handler_tags.go
@@ -56,7 +56,7 @@ func (th *TagHandler) EstimatedTags() int {
 // 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()
+	mmNew := gostatsd.NewMetricMap(mm.Forwarded)
 
 	mm.Counters.Each(func(metricName, _ string, cOriginal gostatsd.Counter) {
 		if th.uniqueFilterAndAddTags(metricName, &cOriginal.Source, &cOriginal.Tags) {
diff --git a/pkg/statsd/parser.go b/pkg/statsd/parser.go
index 4eccd18..792cf4a 100644
--- a/pkg/statsd/parser.go
+++ b/pkg/statsd/parser.go
@@ -116,7 +116,7 @@ func (dp *DatagramParser) Run(ctx context.Context) {
 				accumB += badLineCount
 			}
 			// TODO: Refactor this to use a MetricConsolidator
-			mm := gostatsd.NewMetricMap()
+			mm := gostatsd.NewMetricMap(false)
 			for _, m := range metrics {
 				mm.Receive(m)
 			}
diff --git a/pkg/web/http_receiver_v2.go b/pkg/web/http_receiver_v2.go
index 76fe3ce..9dddc47 100644
--- a/pkg/web/http_receiver_v2.go
+++ b/pkg/web/http_receiver_v2.go
@@ -185,7 +185,7 @@ func (rhh *rawHttpHandlerV2) EventHandler(w http.ResponseWriter, req *http.Reque
 
 func translateFromProtobufV2(pbMetricMap *pb.RawMessageV2) *gostatsd.MetricMap {
 	now := gostatsd.Nanotime(time.Now().UnixNano())
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 
 	for metricName, tagMap := range pbMetricMap.Gauges {
 		mm.Gauges[metricName] = map[string]gostatsd.Gauge{}


Test patch:
diff --git a/metric_consolidator_test.go b/metric_consolidator_test.go
index 4a32568..b32f854 100644
--- a/metric_consolidator_test.go
+++ b/metric_consolidator_test.go
@@ -16,7 +16,7 @@ func TestConsolidation(t *testing.T) {
 	defer cancel()
 
 	ch := make(chan []*MetricMap, 1)
-	mc := NewMetricConsolidator(2, 1*time.Second, ch)
+	mc := NewMetricConsolidator(2, false, 1*time.Second, ch)
 
 	m1 := &Metric{
 		Name:      "foo",
@@ -43,7 +43,7 @@ func TestConsolidation(t *testing.T) {
 	case mm = <-ch:
 	}
 
-	expected := []*MetricMap{NewMetricMap(), NewMetricMap()}
+	expected := []*MetricMap{NewMetricMap(false), NewMetricMap(false)}
 	expected[0].Counters["foo"] = map[string]Counter{
 		"": {
 			PerSecond: 0,
@@ -89,7 +89,7 @@ func benchmarkMetricConsolidator(b *testing.B, parallelism, variations int) {
 	var wgInfra sync.WaitGroup
 
 	ch := make(chan []*MetricMap)
-	mc := NewMetricConsolidator(3, 100*time.Millisecond, ch)
+	mc := NewMetricConsolidator(3, false, 100*time.Millisecond, ch)
 
 	ctx, cancel := context.WithCancel(context.Background())
 
diff --git a/metric_map_test.go b/metric_map_test.go
index 62e632a..f39815a 100644
--- a/metric_map_test.go
+++ b/metric_map_test.go
@@ -42,7 +42,7 @@ func TestReceive(t *testing.T) {
 	t.Parallel()
 	assrt := assert.New(t)
 
-	mm := NewMetricMap()
+	mm := NewMetricMap(false)
 
 	tests := metricsFixtures()
 	for _, metric := range tests {
@@ -105,7 +105,7 @@ func TestReceive(t *testing.T) {
 }
 
 func benchmarkReceive(metric Metric, b *testing.B) {
-	ma := NewMetricMap()
+	ma := NewMetricMap(false)
 	b.ReportAllocs()
 	b.ResetTimer()
 
@@ -131,7 +131,7 @@ func BenchmarkReceiveSet(b *testing.B) {
 }
 
 func BenchmarkReceives(b *testing.B) {
-	ma := NewMetricMap()
+	ma := NewMetricMap(false)
 	tests := metricsFixtures()
 	b.ReportAllocs()
 	b.ResetTimer()
@@ -143,7 +143,7 @@ func BenchmarkReceives(b *testing.B) {
 }
 
 func TestMetricMapDispatch(t *testing.T) {
-	mm := NewMetricMap()
+	mm := NewMetricMap(false)
 	metrics := metricsFixtures()
 	for _, metric := range metrics {
 		mm.Receive(metric)
@@ -208,21 +208,21 @@ func TestMetricMapMerge(t *testing.T) {
 		{Name: "TestMetricMapMerge.set", StringValue: "def", Type: SET, Timestamp: 20},
 	}
 
-	m1 := NewMetricMap()
+	m1 := NewMetricMap(false)
 	for _, m := range metrics1 {
 		m1.Receive(m)
 	}
 
-	m2 := NewMetricMap()
+	m2 := NewMetricMap(false)
 	for _, m := range metrics2 {
 		m2.Receive(m)
 	}
 
-	merged := NewMetricMap()
+	merged := NewMetricMap(false)
 	merged.Merge(m1)
 	merged.Merge(m2)
 
-	expected := NewMetricMap()
+	expected := NewMetricMap(false)
 	expected.Counters = Counters{
 		"TestMetricMapMerge.counter": map[string]Counter{
 			"": {
@@ -263,7 +263,7 @@ func TestMetricMapMerge(t *testing.T) {
 }
 
 func TestMetricMapSplit(t *testing.T) {
-	mmOriginal := NewMetricMap()
+	mmOriginal := NewMetricMap(false)
 	mmOriginal.Counters["m"] = map[string]Counter{
 		"t.s.h1": {Tags: Tags{"t"}, Source: "h1", Value: 10},
 		"t.s.h2": {Tags: Tags{"t"}, Source: "h2", Value: 20},
@@ -293,7 +293,7 @@ func TestMetricMapSplit(t *testing.T) {
 		"t.s.h5": {Tags: Tags{"t"}, Source: "h5", Values: map[string]struct{}{"50": {}, "10": {}}},
 	}
 
-	mmMerged := NewMetricMap()
+	mmMerged := NewMetricMap(false)
 	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)
@@ -308,7 +308,7 @@ func TestMetricMapSplit(t *testing.T) {
 }
 
 func TestMetricMapIsEmpty(t *testing.T) {
-	mm := NewMetricMap()
+	mm := NewMetricMap(false)
 	require.True(t, mm.IsEmpty())
 
 	// Counter
@@ -345,7 +345,7 @@ func TestTagsMatch(t *testing.T) {
 }
 
 func TestMetricMapSplitByTags(t *testing.T) {
-	mmOriginal := NewMetricMap()
+	mmOriginal := NewMetricMap(false)
 	mmOriginal.Counters["m"] = map[string]Counter{
 		"t:x,s:h1":     {Tags: Tags{"t:x"}, Source: "h1", Value: 10},
 		"t:x,s:h2":     {Tags: Tags{"t:x"}, Source: "h2", Value: 20},
diff --git a/pkg/backends/graphite/graphite_test.go b/pkg/backends/graphite/graphite_test.go
index 2cee3d4..31127b1 100644
--- a/pkg/backends/graphite/graphite_test.go
+++ b/pkg/backends/graphite/graphite_test.go
@@ -186,7 +186,7 @@ func TestSendMetricsAsync(t *testing.T) {
 func metrics() *gostatsd.MetricMap {
 	timestamp := gostatsd.Nanotime(time.Unix(123456, 0).UnixNano())
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Counters["stat1"] = map[string]gostatsd.Counter{}
 	mm.Counters["stat1"][""] = gostatsd.Counter{PerSecond: 1.1, Value: 5, Timestamp: timestamp}
 	mm.Timers["t1"] = map[string]gostatsd.Timer{}
@@ -201,7 +201,7 @@ func metrics() *gostatsd.MetricMap {
 func metricsWithHistogram() *gostatsd.MetricMap {
 	timestamp := gostatsd.Nanotime(time.Unix(123456, 0).UnixNano())
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Timers["t1"] = map[string]gostatsd.Timer{}
 	mm.Timers["t1"]["gsd_histogram:20_30_40_50_60"] = gostatsd.Timer{Values: []float64{10}, Timestamp: timestamp, Histogram: map[gostatsd.HistogramThreshold]int{
 		20:                                       5,
diff --git a/pkg/backends/newrelic/newrelic_test.go b/pkg/backends/newrelic/newrelic_test.go
index 4bd6c2a..44bdd0c 100644
--- a/pkg/backends/newrelic/newrelic_test.go
+++ b/pkg/backends/newrelic/newrelic_test.go
@@ -433,7 +433,7 @@ func metricsOneOfEach() *gostatsd.MetricMap {
 }
 
 func metricsWithHistogram() *gostatsd.MetricMap {
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Timers["t1"] = map[string]gostatsd.Timer{}
 	mm.Timers["t1"]["gsd_histogram:20_30_40_50_60"] = gostatsd.Timer{
 		Values:    []float64{10},
diff --git a/pkg/statsd/aggregator_test.go b/pkg/statsd/aggregator_test.go
index 5d59f63..de23381 100644
--- a/pkg/statsd/aggregator_test.go
+++ b/pkg/statsd/aggregator_test.go
@@ -399,7 +399,7 @@ func TestDisabledCount(t *testing.T) {
 	t.Parallel()
 	ma := newFakeAggregator()
 	ma.disabledSubtypes.CountPct = true
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
 	ma.ReceiveMap(mm)
 	ma.Flush(1 * time.Second)
@@ -414,7 +414,7 @@ func TestDisabledMean(t *testing.T) {
 	t.Parallel()
 	ma := newFakeAggregator()
 	ma.disabledSubtypes.MeanPct = true
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
 	ma.ReceiveMap(mm)
 	ma.Flush(1 * time.Second)
@@ -429,7 +429,7 @@ func TestDisabledSum(t *testing.T) {
 	t.Parallel()
 	ma := newFakeAggregator()
 	ma.disabledSubtypes.SumPct = true
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
 	ma.ReceiveMap(mm)
 	ma.Flush(1 * time.Second)
@@ -444,7 +444,7 @@ func TestDisabledSumSquares(t *testing.T) {
 	t.Parallel()
 	ma := newFakeAggregator()
 	ma.disabledSubtypes.SumSquaresPct = true
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
 	ma.ReceiveMap(mm)
 	ma.Flush(1 * time.Second)
@@ -459,7 +459,7 @@ func TestDisabledUpper(t *testing.T) {
 	t.Parallel()
 	ma := newFakeAggregator()
 	ma.disabledSubtypes.UpperPct = true
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
 	ma.ReceiveMap(mm)
 	ma.Flush(1 * time.Second)
@@ -482,7 +482,7 @@ func TestDisabledLower(t *testing.T) {
 		math.MaxUint32,
 	)
 	ma.disabledSubtypes.LowerPct = true
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Name: "x", Value: 1, Type: gostatsd.TIMER})
 	ma.ReceiveMap(mm)
 	ma.Flush(1 * time.Second)
diff --git a/pkg/statsd/handler_backend_test.go b/pkg/statsd/handler_backend_test.go
index 023bb0f..3fb4680 100644
--- a/pkg/statsd/handler_backend_test.go
+++ b/pkg/statsd/handler_backend_test.go
@@ -126,7 +126,7 @@ func TestDispatchMetricMapShouldDistributeMetrics(t *testing.T) {
 	var wgFinish wait.Group
 	wgFinish.StartWithContext(ctx, h.Run)
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	for i := 0; i < numAggregators*100; i++ {
 		m := &gostatsd.Metric{
 			Type:  gostatsd.COUNTER,
@@ -161,7 +161,7 @@ func TestBackendHandlerDispatchMetricMapTerminates(t *testing.T) {
 	h := NewBackendHandler(nil, 0, 1, 0, newTestFactory())
 	cancelledCtx, cancelFunc := context.WithCancel(context.Background())
 	cancelFunc()
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{
 		Name:      "metric",
 		Value:     1,
diff --git a/pkg/statsd/handler_cloud_test.go b/pkg/statsd/handler_cloud_test.go
index 33ad590..d897bf9 100644
--- a/pkg/statsd/handler_cloud_test.go
+++ b/pkg/statsd/handler_cloud_test.go
@@ -42,7 +42,7 @@ func BenchmarkCloudHandlerDispatchMetricMap(b *testing.B) {
 
 	ctxBackground := context.Background()
 	b.RunParallel(func(pb *testing.PB) {
-		mm := gostatsd.NewMetricMap()
+		mm := gostatsd.NewMetricMap(false)
 		mm.Receive(sm1())
 
 		for pb.Next() {
@@ -92,7 +92,7 @@ func TestTransientInstanceFailure(t *testing.T) {
 
 	// t+0: prime the cache
 	expecting.Expect(1, 0)
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(m1)
 	ch.DispatchMetricMap(ctx, mm)
 	expecting.WaitAll()
@@ -104,7 +104,7 @@ func TestTransientInstanceFailure(t *testing.T) {
 	// t+100ms: read from cache, must still be valid
 	expecting.Expect(1, 0)
 
-	mm = gostatsd.NewMetricMap()
+	mm = gostatsd.NewMetricMap(false)
 	mm.Receive(m2)
 	ch.DispatchMetricMap(ctx, mm)
 	expecting.WaitAll()
@@ -231,14 +231,14 @@ func doCheck(
 	wg.StartWithContext(ctx, ci.Run)
 
 	expecting.Expect(1, 1)
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(m1)
 	ch.DispatchMetricMap(ctx, mm)
 	ch.DispatchEvent(ctx, e1)
 	expecting.WaitAll()
 
 	expecting.Expect(1, 1)
-	mm = gostatsd.NewMetricMap()
+	mm = gostatsd.NewMetricMap(false)
 	mm.Receive(m2)
 	ch.DispatchMetricMap(ctx, mm)
 	ch.DispatchEvent(ctx, e2)
diff --git a/pkg/statsd/handler_http_forwarder_v2_test.go b/pkg/statsd/handler_http_forwarder_v2_test.go
index 9176645..1d7ebf4 100644
--- a/pkg/statsd/handler_http_forwarder_v2_test.go
+++ b/pkg/statsd/handler_http_forwarder_v2_test.go
@@ -93,7 +93,7 @@ func TestHttpForwarderV2Translation(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	for _, metric := range metrics {
 		mm.Receive(metric)
 	}
@@ -205,7 +205,7 @@ func BenchmarkHttpForwarderV2TranslateAll(b *testing.B) {
 			Type:        1 + gostatsd.MetricType(i%4), // Use all types
 		})
 	}
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	for _, metric := range metrics {
 		mm.Receive(metric)
 	}
diff --git a/pkg/statsd/handler_tags_test.go b/pkg/statsd/handler_tags_test.go
index ee7e32d..a7647c5 100644
--- a/pkg/statsd/handler_tags_test.go
+++ b/pkg/statsd/handler_tags_test.go
@@ -22,12 +22,12 @@ func TestTagStripMergesCounters(t *testing.T) {
 		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
 	})
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.COUNTER, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value"}, Value: 20, Rate: 1})                              // Will merge in to metric with TS 20
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.COUNTER, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value", "key2:value2"}, Value: 1, Rate: 0.1})              // key2:value2 will be dropped, and TS 10 will merge in to it
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.COUNTER, Name: "metric", Timestamp: 30, Tags: gostatsd.Tags{"key:value", "key2:value2", "key3:value3"}, Value: 1, Rate: 1}) // key2:value2 will be dropped, but it won't merge in to anything
 
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Counters["metric"] = map[string]gostatsd.Counter{
 		"key:value":             {Timestamp: 20, Value: 30, Tags: gostatsd.Tags{"key:value"}},
 		"key3:value3,key:value": {Timestamp: 30, Value: 1, Tags: gostatsd.Tags{"key3:value3", "key:value"}},
@@ -49,12 +49,12 @@ func TestTagStripMergesGauges(t *testing.T) {
 		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
 	})
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.GAUGE, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value"}, Value: 10})                               // Will merge in to metric with TS 20
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.GAUGE, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value", "key2:value2"}, Value: 20})                // key2:value2 will be dropped, and TS 10 will merge in to it
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.GAUGE, Name: "metric", Timestamp: 30, Tags: gostatsd.Tags{"key:value", "key2:value2", "key3:value3"}, Value: 30}) // key2:value2 will be dropped, but it won't merge in to anything
 
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Gauges["metric"] = map[string]gostatsd.Gauge{
 		"key:value":             {Timestamp: 20, Value: 20, Tags: gostatsd.Tags{"key:value"}},
 		"key3:value3,key:value": {Timestamp: 30, Value: 30, Tags: gostatsd.Tags{"key3:value3", "key:value"}},
@@ -76,12 +76,12 @@ func TestTagStripMergesTimers(t *testing.T) {
 		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
 	})
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.TIMER, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value"}, Value: 10, Rate: 1})                               // Will merge in to metric with TS 20
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.TIMER, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value", "key2:value2"}, Value: 20, Rate: 1})                // key2:value2 will be dropped, and TS 10 will merge in to it
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.TIMER, Name: "metric", Timestamp: 30, Tags: gostatsd.Tags{"key:value", "key2:value2", "key3:value3"}, Value: 30, Rate: 1}) // key2:value2 will be dropped, but it won't merge in to anything
 
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Timers["metric"] = map[string]gostatsd.Timer{
 		"key:value":             {Timestamp: 20, Values: []float64{10, 20}, Tags: gostatsd.Tags{"key:value"}, SampledCount: 2},
 		"key3:value3,key:value": {Timestamp: 30, Values: []float64{30}, Tags: gostatsd.Tags{"key3:value3", "key:value"}, SampledCount: 1},
@@ -105,12 +105,12 @@ func TestTagStripMergesSets(t *testing.T) {
 		{DropTags: gostatsd.StringMatchList{gostatsd.NewStringMatch("key2:*")}},
 	})
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.SET, Name: "metric", Timestamp: 10, Tags: gostatsd.Tags{"key:value"}, StringValue: "abc"})                               // Will merge in to metric with TS 20
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.SET, Name: "metric", Timestamp: 20, Tags: gostatsd.Tags{"key:value", "key2:value2"}, StringValue: "def"})                // key2:value2 will be dropped, and TS 10 will merge in to it
 	mm.Receive(&gostatsd.Metric{Type: gostatsd.SET, Name: "metric", Timestamp: 30, Tags: gostatsd.Tags{"key:value", "key2:value2", "key3:value3"}, StringValue: "ghi"}) // key2:value2 will be dropped, but it won't merge in to anything
 
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Sets["metric"] = map[string]gostatsd.Set{
 		"key:value":             {Timestamp: 20, Values: map[string]struct{}{"abc": {}, "def": {}}, Tags: gostatsd.Tags{"key:value"}},
 		"key3:value3,key:value": {Timestamp: 30, Values: map[string]struct{}{"ghi": {}}, Tags: gostatsd.Tags{"key3:value3", "key:value"}},
@@ -131,9 +131,9 @@ func TestFilterPassesNoFilters(t *testing.T) {
 	tch := &capturingHandler{}
 	th := NewTagHandler(tch, gostatsd.Tags{}, nil)
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric())
 	th.DispatchMetricMap(context.Background(), mm)
 	require.Len(t, tch.mm, 1)
@@ -147,9 +147,9 @@ func TestFilterPassesEmptyFilters(t *testing.T) {
 	th := NewTagHandler(tch, gostatsd.Tags{}, nil)
 	th.filters = []Filter{}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric())
 	th.DispatchMetricMap(context.Background(), mm)
 	require.Len(t, tch.mm, 1)
@@ -168,9 +168,9 @@ func TestFilterKeepNonMatch(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric())
 	th.DispatchMetricMap(context.Background(), mm)
 	require.Len(t, tch.mm, 1)
@@ -189,7 +189,7 @@ func TestFilterDropsBadName(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
 	th.DispatchMetricMap(context.Background(), mm)
 	require.Len(t, tch.mm, 0) // nothing is dispatched if the entire MetricMap is dropped
@@ -207,7 +207,7 @@ func TestFilterDropsBadPrefix(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
 	th.DispatchMetricMap(context.Background(), mm)
 	require.Len(t, tch.mm, 0) // nothing is dispatched if the entire MetricMap is dropped
@@ -226,11 +226,11 @@ func TestFilterKeepsWhitelist(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric(Name("name.bad")))
 	mm.Receive(MakeMetric(Name("name.good")))
 
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(Name("name.good")))
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -250,9 +250,9 @@ func TestFilterDropsTag(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric(Name("bad.name")))
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(Name("bad.name"), DropTag("host:baz")))
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -272,9 +272,9 @@ func TestFilterDropsHost(t *testing.T) {
 		},
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(DropSource))
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -380,9 +380,9 @@ func TestTagMetricHandlerAddsNoTags(t *testing.T) {
 	tch := &capturingHandler{}
 	th := NewTagHandler(tch, gostatsd.Tags{}, nil)
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric(DropSource))
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(DropSource)) // No hostname added
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -396,9 +396,9 @@ func TestTagMetricHandlerAddsSingleTag(t *testing.T) {
 	tch := &capturingHandler{}
 	th := NewTagHandler(tch, gostatsd.Tags{"tag1"}, nil)
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(AddTag("tag1")))
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -412,9 +412,9 @@ func TestTagMetricHandlerAddsMultipleTags(t *testing.T) {
 	tch := &capturingHandler{}
 	th := NewTagHandler(tch, gostatsd.Tags{"tag1", "tag2"}, nil)
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(AddTag("tag1", "tag2")))
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -428,9 +428,9 @@ func TestTagMetricHandlerAddsDuplicateTags(t *testing.T) {
 	tch := &capturingHandler{}
 	th := NewTagHandler(tch, gostatsd.Tags{"tag1", "tag2", "tag2", "tag3", "tag1"}, nil)
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 	mm.Receive(MakeMetric())
-	expected := gostatsd.NewMetricMap()
+	expected := gostatsd.NewMetricMap(false)
 	expected.Receive(MakeMetric(AddTag("tag1", "tag2", "tag3")))
 
 	th.DispatchMetricMap(context.Background(), mm)
@@ -529,7 +529,7 @@ func BenchmarkTagMetricHandlerAddsDuplicateTagsSmall(b *testing.B) {
 			Type: gostatsd.COUNTER,
 			Tags: metricTags,
 		}
-		mm := gostatsd.NewMetricMap()
+		mm := gostatsd.NewMetricMap(false)
 		mm.Receive(m)
 		th.DispatchMetricMap(context.Background(), mm)
 	}
@@ -568,7 +568,7 @@ func BenchmarkTagMetricHandlerAddsDuplicateTagsLarge(b *testing.B) {
 			Type: gostatsd.COUNTER,
 			Tags: metricTags,
 		}
-		mm := gostatsd.NewMetricMap()
+		mm := gostatsd.NewMetricMap(false)
 		mm.Receive(m)
 		th.DispatchMetricMap(context.Background(), mm)
 	}
diff --git a/pkg/statsd/parser_test.go b/pkg/statsd/parser_test.go
index 54896f6..056444d 100644
--- a/pkg/statsd/parser_test.go
+++ b/pkg/statsd/parser_test.go
@@ -104,7 +104,7 @@ func TestParseDatagram(t *testing.T) {
 			t.Parallel()
 			mr, ch := newTestParser(false)
 			metrics, _, _ := mr.handleDatagram(context.Background(), lex(), 0, fakeIP, []byte(datagram))
-			mm := gostatsd.NewMetricMap()
+			mm := gostatsd.NewMetricMap(false)
 			for _, m := range metrics {
 				mm.Receive(m)
 			}
@@ -191,7 +191,7 @@ func TestParseDatagramIgnoreHost(t *testing.T) {
 				}
 				ch.events[i].DateHappened = 0
 			}
-			mm := gostatsd.NewMetricMap()
+			mm := gostatsd.NewMetricMap(false)
 			for _, m := range metrics {
 				mm.Receive(m)
 			}
diff --git a/pkg/web/http_receiver_v2_test.go b/pkg/web/http_receiver_v2_test.go
index 0818b4e..636dcdb 100644
--- a/pkg/web/http_receiver_v2_test.go
+++ b/pkg/web/http_receiver_v2_test.go
@@ -117,7 +117,7 @@ func TestForwardingEndToEndV2(t *testing.T) {
 		Rate:        0.1,
 	}
 
-	mm := gostatsd.NewMetricMap()
+	mm := gostatsd.NewMetricMap(false)
 
 	for i := 0; i < 100; i++ {
 		mm.Receive(m1)