Repository: atlassian/gostatsd
Pull request #341 (2020-08-13 09:21:45)

Update CloudHandler.DispatchMetricMap to use MetricMap all the way through
This is the current "big" thing that I'm aware of blocking clustering, 'cause it interferes with loop detection.

As usual, commits are relatively stand-alone.

@mwhittington21 - if you could get a chance to test this at some point (I'll release it as 28.3.0-rc1), that would be much appreciated.  I generally soak changes locally, but I can only validate this one with tests.

mwhittington21: @tiedotguy I'll try and find some time to upgrade the clusters. A bit swamped right now, but I'll get to it asap.

Patch:
diff --git a/.travis.yml b/.travis.yml
index 1f37f90..07c524c 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,15 +1,38 @@
-sudo: required
-
 language: go
 
 services:
   - docker
 
 go:
-  - "1.13.6"  # Keep in sync with Makefile and README.md
-
-os:
-  - linux
+  - "1.15.6"  # Keep in sync with Makefile and README.md
+
+arch:
+  - amd64
+  - arm64
+
+
+dist: bionic
+os: linux
+
+jobs:
+  include:
+   - os: linux
+     arch: amd64
+   - os: linux
+     arch: arm64
+   - stage: Push Docker manifest
+     env:
+       - DOCKER_CLI_EXPERIMENTAL=enabled
+     script:
+       - curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo apt-key add -
+       - sudo add-apt-repository "deb [arch=$TRAVIS_CPU_ARCH] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable"
+       - sudo apt-get update
+       - sudo apt-get -y -o Dpkg::Options::="--force-confnew" install docker-ce
+       - if [ "$TRAVIS_OS_NAME" == "linux" -a ! -z "$TRAVIS_TAG" ]; then
+           echo "Executing release-manifest on tag build $TRAVIS_TAG";
+           docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
+           make release-manifest;
+         fi
 
 before_install:
   - go get -v github.com/mattn/goveralls
@@ -25,10 +48,10 @@ script:
   - make bench-race
 
 after_success:
-  - if [ "$TRAVIS_OS_NAME" == "linux" -a ! -z "$TRAVIS_TAG" ]; then
+  - if [ "$TRAVIS_BUILD_STAGE_NAME" == "test" -a "$TRAVIS_OS_NAME" == "linux" -a ! -z "$TRAVIS_TAG" ]; then
     echo "Executing release on tag build $TRAVIS_TAG";
     docker login -u="$DOCKER_USERNAME" -p="$DOCKER_PASSWORD";
-    ARCH=linux make release;
+    ARCH=linux CPU_ARCH="$TRAVIS_CPU_ARCH" make release;
     else
     echo "Not executing release on non-tag build";
     fi
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 19fa3c3..ce25aeb 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,10 @@
+29.0.0
+------
+- Support for ARM64 docker image
+- Test changes and fixes to improve coverage
+- Internal code refactoring to reduce surface area
+- Refactor lexer to improve memory usage
+
 28.3.0
 ------
 - Rather than dropping data points, the Datadog backend will coerce non-numeric values resulting from aggregation to numeric.
diff --git a/CLOUDPROVIDERS.md b/CLOUDPROVIDERS.md
index db2bd09..49564e7 100644
--- a/CLOUDPROVIDERS.md
+++ b/CLOUDPROVIDERS.md
@@ -12,9 +12,6 @@ There are currently two supported cloud providers:
 
 All configuration is in a stanza named after the backend, and takes simple key value pairs.
 
-**Cloud providers should be disabled on the aggregation server when using http forwarding, as the source IP isn't
-propagated, and that information should be collected on the ingestion server.**
-
 aws
 ---
 ### TODO
diff --git a/METRICS.md b/METRICS.md
index 1e40b8a..621fa42 100644
--- a/METRICS.md
+++ b/METRICS.md
@@ -70,7 +70,7 @@ Metrics:
 | version       | The git tag of the build
 | commit        | The short git commit of the build
 | backend       | The backend sending a particular metric
-| type          | Either metric or event
+| type          | Either metric or event for cloudprovider.hosts_queued, or event for cloudprovider.items_queued
 | result        | Success to indicate a batch of metrics was successfully processed, failure to indicate a batch of metrics was not processed, with additional failure tag for why)
 | failure       | The reason a batch of metrics was not processed
 | server-name   | The name of an http-server as specified in the config file
diff --git a/Makefile b/Makefile
index 8c106f7..7f7931a 100644
--- a/Makefile
+++ b/Makefile
@@ -8,9 +8,11 @@ GIT_HASH ?= $$(git rev-parse --short HEAD)
 GOBUILD_VERSION_ARGS := -ldflags "-s -X $(VERSION_VAR)=$(REPO_VERSION) -X $(GIT_VAR)=$(GIT_HASH) -X $(BUILD_DATE_VAR)=$(BUILD_DATE)"
 GOBUILD_VERSION_ARGS_WITH_SYMS := -ldflags "-X $(VERSION_VAR)=$(REPO_VERSION) -X $(GIT_VAR)=$(GIT_HASH) -X $(BUILD_DATE_VAR)=$(BUILD_DATE)"
 BINARY_NAME := gostatsd
-IMAGE_NAME := atlassianlabs/$(BINARY_NAME)
+CPU_ARCH ?= amd64
+MANIFEST_NAME := atlassianlabs/$(BINARY_NAME)
+IMAGE_NAME := $(MANIFEST_NAME)-$(CPU_ARCH)
 ARCH ?= $$(uname -s | tr A-Z a-z)
-GOVERSION := 1.13.6  # Keep in sync with .travis.yml and README.md
+GOVERSION := 1.15.6  # Keep in sync with .travis.yml and README.md
 GP := /gopath
 MAIN_PKG := github.com/atlassian/gostatsd/cmd/gostatsd
 CLUSTER_PKG := github.com/atlassian/gostatsd/cmd/cluster
@@ -160,6 +162,17 @@ release-symbols: docker-symbols
 
 release: release-normal release-race release-symbols
 
+release-manifest:
+	for tag in latest $(REPO_VERSION) $(GIT_HASH)-race $(REPO_VERSION)-race $(REPO_VERSION)-syms; do \
+	  for arch in amd64 arm64; do \
+		  docker pull $(MANIFEST_NAME)-$$arch:$$tag; \
+		done; \
+	  docker manifest create $(MANIFEST_NAME):$$tag --amend \
+		  $(MANIFEST_NAME)-amd64:$$tag \
+		  $(MANIFEST_NAME)-arm64:$$tag; \
+	  docker manifest push $(MANIFEST_NAME):$$tag; \
+	done
+
 run: build
 	./build/bin/$(ARCH)/$(BINARY_NAME) --backends=stdout --verbose --flush-interval=2s
 
diff --git a/README.md b/README.md
index 2f751d8..c86a0c4 100644
--- a/README.md
+++ b/README.md
@@ -24,7 +24,7 @@ server based on load.
 
 Building the server
 -------------------
-Gostatsd currently targets Go 1.13.6.  If you are compiling from source, please ensure you are running this version.
+Gostatsd currently targets Go 1.15.6.  If you are compiling from source, please ensure you are running this version.
 
 From the `gostatsd` directory run `make build`. The binary will be built in `build/bin/<arch>/gostatsd`.
 
@@ -216,9 +216,6 @@ Cloud providers are a way to automatically enrich metrics with metadata from a c
 
 Refer to [cloud providers](CLOUDPROVIDERS.md) for configuration options for the cloud providers.
 
-They should be disabled on the aggregation server when using http forwarding, as the source IP isn't propagated, and
-that information should be collected on the ingestion server.
-
 
 Configuring timer sub-metrics
 -----------------------------
diff --git a/counters.go b/counters.go
index 6ac842a..faceb79 100644
--- a/counters.go
+++ b/counters.go
@@ -14,6 +14,11 @@ func NewCounter(timestamp Nanotime, value int64, source Source, tags Tags) Count
 	return Counter{Value: value, Timestamp: timestamp, Source: source, Tags: tags.Copy()}
 }
 
+func (c *Counter) AddTagsSetSource(additionalTags Tags, newSource Source) {
+	c.Tags = c.Tags.Concat(additionalTags)
+	c.Source = newSource
+}
+
 // Counters stores a map of counters by tags.
 type Counters map[string]map[string]Counter
 
diff --git a/gauges.go b/gauges.go
index 68c567c..dbf013a 100644
--- a/gauges.go
+++ b/gauges.go
@@ -13,6 +13,11 @@ func NewGauge(timestamp Nanotime, value float64, source Source, tags Tags) Gauge
 	return Gauge{Value: value, Timestamp: timestamp, Source: source, Tags: tags.Copy()}
 }
 
+func (g *Gauge) AddTagsSetSource(additionalTags Tags, newSource Source) {
+	g.Tags = g.Tags.Concat(additionalTags)
+	g.Source = newSource
+}
+
 // Gauges stores a map of gauges by tags.
 type Gauges map[string]map[string]Gauge
 
diff --git a/metric_map.go b/metric_map.go
index e42e04b..b057e7e 100644
--- a/metric_map.go
+++ b/metric_map.go
@@ -54,87 +54,93 @@ func MergeMaps(mms []*MetricMap) *MetricMap {
 }
 
 func (mm *MetricMap) Merge(mmFrom *MetricMap) {
-	mmFrom.Counters.Each(func(metricName string, tagsKey string, counterFrom Counter) {
-		v, ok := mm.Counters[metricName]
+	mmFrom.Counters.Each(mm.MergeCounter)
+	mmFrom.Gauges.Each(mm.MergeGauge)
+	mmFrom.Sets.Each(mm.MergeSet)
+	mmFrom.Timers.Each(mm.MergeTimer)
+}
+
+func (mm *MetricMap) MergeCounter(metricName string, tagsKey string, counterFrom Counter) {
+	v, ok := mm.Counters[metricName]
+	if ok {
+		counterInto, ok := v[tagsKey]
 		if ok {
-			counterInto, ok := v[tagsKey]
-			if ok {
-				if counterInto.Timestamp < counterFrom.Timestamp {
-					counterInto.Timestamp = counterFrom.Timestamp
-				}
-				counterInto.Value += counterFrom.Value
-			} else {
-				counterInto = counterFrom
+			if counterInto.Timestamp < counterFrom.Timestamp {
+				counterInto.Timestamp = counterFrom.Timestamp
 			}
-			v[tagsKey] = counterInto
+			counterInto.Value += counterFrom.Value
 		} else {
-			mm.Counters[metricName] = map[string]Counter{
-				tagsKey: counterFrom,
-			}
+			counterInto = counterFrom
 		}
-	})
+		v[tagsKey] = counterInto
+	} else {
+		mm.Counters[metricName] = map[string]Counter{
+			tagsKey: counterFrom,
+		}
+	}
+}
 
-	mmFrom.Gauges.Each(func(metricName string, tagsKey string, gaugeFrom Gauge) {
-		v, ok := mm.Gauges[metricName]
+func (mm *MetricMap) MergeGauge(metricName string, tagsKey string, gaugeFrom Gauge) {
+	v, ok := mm.Gauges[metricName]
+	if ok {
+		gaugeInto, ok := v[tagsKey]
 		if ok {
-			gaugeInto, ok := v[tagsKey]
-			if ok {
-				if gaugeInto.Timestamp < gaugeFrom.Timestamp {
-					gaugeInto.Timestamp = gaugeFrom.Timestamp
-					gaugeInto.Value = gaugeFrom.Value
-				}
-			} else {
-				gaugeInto = gaugeFrom
+			if gaugeInto.Timestamp < gaugeFrom.Timestamp {
+				gaugeInto.Timestamp = gaugeFrom.Timestamp
+				gaugeInto.Value = gaugeFrom.Value
 			}
-			v[tagsKey] = gaugeInto
 		} else {
-			mm.Gauges[metricName] = map[string]Gauge{
-				tagsKey: gaugeFrom,
-			}
+			gaugeInto = gaugeFrom
 		}
-	})
+		v[tagsKey] = gaugeInto
+	} else {
+		mm.Gauges[metricName] = map[string]Gauge{
+			tagsKey: gaugeFrom,
+		}
+	}
+}
 
-	mmFrom.Timers.Each(func(metricName string, tagsKey string, timerFrom Timer) {
-		v, ok := mm.Timers[metricName]
+func (mm *MetricMap) MergeSet(metricName string, tagsKey string, setFrom Set) {
+	v, ok := mm.Sets[metricName]
+	if ok {
+		setInto, ok := v[tagsKey]
 		if ok {
-			timerInto, ok := v[tagsKey]
-			if ok {
-				if timerInto.Timestamp < timerFrom.Timestamp {
-					timerInto.Timestamp = timerFrom.Timestamp
-				}
-				timerInto.Values = append(timerInto.Values, timerFrom.Values...)
-				timerInto.SampledCount += timerFrom.SampledCount
-			} else {
-				timerInto = timerFrom
+			if setInto.Timestamp < setFrom.Timestamp {
+				setInto.Timestamp = setFrom.Timestamp
 			}
-			v[tagsKey] = timerInto
-		} else {
-			mm.Timers[metricName] = map[string]Timer{
-				tagsKey: timerFrom,
+			for setValue := range setFrom.Values {
+				setInto.Values[setValue] = struct{}{}
 			}
+		} else {
+			setInto = setFrom
 		}
-	})
-	mmFrom.Sets.Each(func(metricName string, tagsKey string, setFrom Set) {
-		v, ok := mm.Sets[metricName]
+		v[tagsKey] = setInto
+	} else {
+		mm.Sets[metricName] = map[string]Set{
+			tagsKey: setFrom,
+		}
+	}
+}
+
+func (mm *MetricMap) MergeTimer(metricName string, tagsKey string, timerFrom Timer) {
+	v, ok := mm.Timers[metricName]
+	if ok {
+		timerInto, ok := v[tagsKey]
 		if ok {
-			setInto, ok := v[tagsKey]
-			if ok {
-				if setInto.Timestamp < setFrom.Timestamp {
-					setInto.Timestamp = setFrom.Timestamp
-				}
-				for setValue := range setFrom.Values {
-					setInto.Values[setValue] = struct{}{}
-				}
-			} else {
-				setInto = setFrom
+			if timerInto.Timestamp < timerFrom.Timestamp {
+				timerInto.Timestamp = timerFrom.Timestamp
 			}
-			v[tagsKey] = setInto
+			timerInto.Values = append(timerInto.Values, timerFrom.Values...)
+			timerInto.SampledCount += timerFrom.SampledCount
 		} else {
-			mm.Sets[metricName] = map[string]Set{
-				tagsKey: setFrom,
-			}
+			timerInto = timerFrom
 		}
-	})
+		v[tagsKey] = timerInto
+	} else {
+		mm.Timers[metricName] = map[string]Timer{
+			tagsKey: timerFrom,
+		}
+	}
 }
 
 func (mm *MetricMap) IsEmpty() bool {
@@ -367,6 +373,8 @@ func (mm *MetricMap) String() string {
 }
 
 // AsMetrics will synthesize Metrics from the MetricMap and return them as a slice
+// TODO: Remove this function.
+// Deprecated: It only supports tests now
 func (mm *MetricMap) AsMetrics() []*Metric {
 	var metrics []*Metric
 
diff --git a/metrics.go b/metrics.go
index af37de6..374a256 100644
--- a/metrics.go
+++ b/metrics.go
@@ -53,17 +53,6 @@ type Metric struct {
 	DoneFunc  func()     // Returns the metric to the pool. May be nil. Call Metric.Done(), not this.
 }
 
-func (m *Metric) AddTagsSetSource(additionalTags Tags, newSource Source) {
-	if len(additionalTags) > 0 {
-		m.Tags = m.Tags.Concat(additionalTags)
-		m.TagsKey = ""
-	}
-	if newSource != m.Source {
-		m.Source = newSource
-		m.TagsKey = ""
-	}
-}
-
 // Reset is used to reset a metric to as clean state, called on re-use from the pool.
 func (m *Metric) Reset() {
 	m.Name = ""
diff --git a/pkg/statsd/handler_cloud.go b/pkg/statsd/handler_cloud.go
index 468b979..d39cf4e 100644
--- a/pkg/statsd/handler_cloud.go
+++ b/pkg/statsd/handler_cloud.go
@@ -18,20 +18,19 @@ type CloudHandler struct {
 	statsCacheMiss uint64 // Cumulative number of cache misses
 
 	// All other stats fields may only be read or written by the main CloudHandler.Run goroutine
-	statsMetricItemsQueued uint64 // Absolute number of metrics queued, waiting for a CP to respond
 	statsMetricHostsQueued uint64 // Absolute number of IPs waiting for a CP to respond for metrics
 	statsEventItemsQueued  uint64 // Absolute number of events queued, waiting for a CP to respond
 	statsEventHostsQueued  uint64 // Absolute number of IPs waiting for a CP to respond for events
 
 	cachedInstances gostatsd.CachedInstances
 	handler         gostatsd.PipelineHandler
-	incomingMetrics chan []*gostatsd.Metric
+	incomingMetrics chan *gostatsd.MetricMap
 	incomingEvents  chan *gostatsd.Event
 
 	// emitChan triggers a write of all the current stats when it is given a Statser
 	emitChan        chan stats.Statser
 	awaitingEvents  map[gostatsd.Source][]*gostatsd.Event
-	awaitingMetrics map[gostatsd.Source][]*gostatsd.Metric
+	awaitingMetrics map[gostatsd.Source]*gostatsd.MetricMap
 	toLookupIPs     []gostatsd.Source
 	wg              sync.WaitGroup
 
@@ -43,11 +42,11 @@ func NewCloudHandler(cachedInstances gostatsd.CachedInstances, handler gostatsd.
 	return &CloudHandler{
 		cachedInstances: cachedInstances,
 		handler:         handler,
-		incomingMetrics: make(chan []*gostatsd.Metric),
+		incomingMetrics: make(chan *gostatsd.MetricMap),
 		incomingEvents:  make(chan *gostatsd.Event),
 		emitChan:        make(chan stats.Statser),
 		awaitingEvents:  make(map[gostatsd.Source][]*gostatsd.Event),
-		awaitingMetrics: make(map[gostatsd.Source][]*gostatsd.Metric),
+		awaitingMetrics: make(map[gostatsd.Source]*gostatsd.MetricMap),
 		estimatedTags:   handler.EstimatedTags() + cachedInstances.EstimatedTags(),
 	}
 }
@@ -57,37 +56,50 @@ func (ch *CloudHandler) EstimatedTags() int {
 	return ch.estimatedTags
 }
 
-func (ch *CloudHandler) processMetrics(ctx context.Context, metrics []*gostatsd.Metric) {
+func (ch *CloudHandler) DispatchMetricMap(ctx context.Context, mm *gostatsd.MetricMap) {
 	mmToDispatch := gostatsd.NewMetricMap()
-	var toHandle []*gostatsd.Metric
-	for _, m := range metrics {
-		if ch.updateTagsAndHostname(m, m.Source) {
-			mmToDispatch.Receive(m)
+	mmToHandle := gostatsd.NewMetricMap()
+	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)
 		} else {
-			toHandle = append(toHandle, m)
+			mmToHandle.MergeCounter(metricName, tagsKey, c)
 		}
-	}
+	})
+	mm.Gauges.Each(func(metricName string, tagsKey string, g gostatsd.Gauge) {
+		if ch.updateTagsAndHostname(&g, g.Source) {
+			mmToDispatch.MergeGauge(metricName, gostatsd.FormatTagsKey(g.Source, g.Tags), g)
+		} else {
+			mmToHandle.MergeGauge(metricName, tagsKey, g)
+		}
+	})
+	mm.Timers.Each(func(metricName string, tagsKey string, t gostatsd.Timer) {
+		if ch.updateTagsAndHostname(&t, t.Source) {
+			mmToDispatch.MergeTimer(metricName, gostatsd.FormatTagsKey(t.Source, t.Tags), t)
+		} else {
+			mmToHandle.MergeTimer(metricName, tagsKey, t)
+		}
+	})
+	mm.Sets.Each(func(metricName string, tagsKey string, s gostatsd.Set) {
+		if ch.updateTagsAndHostname(&s, s.Source) {
+			mmToDispatch.MergeSet(metricName, gostatsd.FormatTagsKey(s.Source, s.Tags), s)
+		} else {
+			mmToHandle.MergeSet(metricName, tagsKey, s)
+		}
+	})
 
 	if !mmToDispatch.IsEmpty() {
 		ch.handler.DispatchMetricMap(ctx, mmToDispatch)
 	}
 
-	if len(toHandle) > 0 {
+	if !mmToHandle.IsEmpty() {
 		select {
 		case <-ctx.Done():
-		case ch.incomingMetrics <- toHandle:
+		case ch.incomingMetrics <- mmToHandle:
 		}
 	}
 }
 
-// DispatchMetricMap re-dispatches a MetricMap through CloudHandler.processMetrics
-// 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) {
-	ch.processMetrics(ctx, mm.AsMetrics())
-}
-
 func (ch *CloudHandler) DispatchEvent(ctx context.Context, e *gostatsd.Event) {
 	if ch.updateTagsAndHostname(e, e.Source) {
 		ch.handler.DispatchEvent(ctx, e)
@@ -150,7 +162,6 @@ func (ch *CloudHandler) emit(statser stats.Statser) {
 	statser.Gauge("cloudprovider.cache_miss", float64(atomic.LoadUint64(&ch.statsCacheMiss)), nil)
 	t := gostatsd.Tags{"type:metric"}
 	statser.Gauge("cloudprovider.hosts_queued", float64(ch.statsMetricHostsQueued), t)
-	statser.Gauge("cloudprovider.items_queued", float64(ch.statsMetricItemsQueued), t)
 	t = gostatsd.Tags{"type:event"}
 	statser.Gauge("cloudprovider.hosts_queued", float64(ch.statsEventHostsQueued), t)
 	statser.Gauge("cloudprovider.items_queued", float64(ch.statsEventItemsQueued), t)
@@ -192,12 +203,11 @@ func (ch *CloudHandler) Run(ctx context.Context) {
 }
 
 func (ch *CloudHandler) handleInstanceInfo(ctx context.Context, info gostatsd.InstanceInfo) {
-	metrics := ch.awaitingMetrics[info.IP]
-	if len(metrics) > 0 {
+	mm := ch.awaitingMetrics[info.IP]
+	if mm != nil {
 		delete(ch.awaitingMetrics, info.IP)
-		ch.statsMetricItemsQueued -= uint64(len(metrics))
 		ch.statsMetricHostsQueued--
-		go ch.updateAndDispatchMetrics(ctx, info.Instance, metrics)
+		go ch.updateAndDispatchMetrics(ctx, info.Instance, mm)
 	}
 	events := ch.awaitingEvents[info.IP]
 	if len(events) > 0 {
@@ -208,23 +218,41 @@ func (ch *CloudHandler) handleInstanceInfo(ctx context.Context, info gostatsd.In
 	}
 }
 
-func (ch *CloudHandler) handleIncomingMetrics(metrics []*gostatsd.Metric) {
-	for _, m := range metrics {
-		queue := ch.awaitingMetrics[m.Source]
-		ch.awaitingMetrics[m.Source] = append(queue, m)
-		if len(queue) == 0 && len(ch.awaitingEvents[m.Source]) == 0 {
-			// This is the first metric for that IP in the queue. Need to fetch an Instance for this IP.
-			ch.toLookupIPs = append(ch.toLookupIPs, m.Source)
-			ch.statsMetricHostsQueued++
-		}
+// prepareMetricQueue will ensure that ch.awaitingMetrics has a matching MetricMap for
+// source, and return it.  If it did not have one initially, it will also enqueue source
+// for lookup.  The functionality is overloaded to minimize code duplication.
+func (ch *CloudHandler) prepareMetricQueue(source gostatsd.Source) *gostatsd.MetricMap {
+	if queue, ok := ch.awaitingMetrics[source]; ok {
+		return queue
 	}
-	ch.statsMetricItemsQueued += uint64(len(metrics))
+	if len(ch.awaitingEvents[source]) == 0 {
+		ch.toLookupIPs = append(ch.toLookupIPs, source)
+		ch.statsMetricHostsQueued++
+	}
+	queue := gostatsd.NewMetricMap()
+	ch.awaitingMetrics[source] = queue
+	return queue
+}
+
+func (ch *CloudHandler) handleIncomingMetrics(mm *gostatsd.MetricMap) {
+	mm.Counters.Each(func(metricName string, tagsKey string, c gostatsd.Counter) {
+		ch.prepareMetricQueue(c.Source).MergeCounter(metricName, tagsKey, c)
+	})
+	mm.Gauges.Each(func(metricName string, tagsKey string, g gostatsd.Gauge) {
+		ch.prepareMetricQueue(g.Source).MergeGauge(metricName, tagsKey, g)
+	})
+	mm.Sets.Each(func(metricName string, tagsKey string, s gostatsd.Set) {
+		ch.prepareMetricQueue(s.Source).MergeSet(metricName, tagsKey, s)
+	})
+	mm.Timers.Each(func(metricName string, tagsKey string, t gostatsd.Timer) {
+		ch.prepareMetricQueue(t.Source).MergeTimer(metricName, tagsKey, t)
+	})
 }
 
 func (ch *CloudHandler) handleIncomingEvent(e *gostatsd.Event) {
 	queue := ch.awaitingEvents[e.Source]
 	ch.awaitingEvents[e.Source] = append(queue, e)
-	if len(queue) == 0 && len(ch.awaitingMetrics[e.Source]) == 0 {
+	if len(queue) == 0 && ch.awaitingMetrics[e.Source] == nil {
 		// This is the first event for that IP in the queue. Need to fetch an Instance for this IP.
 		ch.toLookupIPs = append(ch.toLookupIPs, e.Source)
 		ch.statsEventHostsQueued++
@@ -232,13 +260,25 @@ func (ch *CloudHandler) handleIncomingEvent(e *gostatsd.Event) {
 	ch.statsEventItemsQueued++
 }
 
-func (ch *CloudHandler) updateAndDispatchMetrics(ctx context.Context, instance *gostatsd.Instance, metrics []*gostatsd.Metric) {
-	mm := gostatsd.NewMetricMap()
-	for _, m := range metrics {
-		updateInplace(m, instance)
-		mm.Receive(m)
-	}
-	ch.handler.DispatchMetricMap(ctx, mm)
+func (ch *CloudHandler) updateAndDispatchMetrics(ctx context.Context, instance *gostatsd.Instance, mmIn *gostatsd.MetricMap) {
+	mmOut := gostatsd.NewMetricMap()
+	mmIn.Counters.Each(func(metricName string, tagsKey string, c gostatsd.Counter) {
+		updateInplace(&c, instance)
+		mmOut.MergeCounter(metricName, gostatsd.FormatTagsKey(c.Source, c.Tags), c)
+	})
+	mmIn.Gauges.Each(func(metricName string, tagsKey string, g gostatsd.Gauge) {
+		updateInplace(&g, instance)
+		mmOut.MergeGauge(metricName, gostatsd.FormatTagsKey(g.Source, g.Tags), g)
+	})
+	mmIn.Sets.Each(func(metricName string, tagsKey string, s gostatsd.Set) {
+		updateInplace(&s, instance)
+		mmOut.MergeSet(metricName, gostatsd.FormatTagsKey(s.Source, s.Tags), s)
+	})
+	mmIn.Timers.Each(func(metricName string, tagsKey string, t gostatsd.Timer) {
+		updateInplace(&t, instance)
+		mmOut.MergeTimer(metricName, gostatsd.FormatTagsKey(t.Source, t.Tags), t)
+	})
+	ch.handler.DispatchMetricMap(ctx, mmOut)
 }
 
 func (ch *CloudHandler) updateAndDispatchEvents(ctx context.Context, instance *gostatsd.Instance, events []*gostatsd.Event) {
diff --git a/sets.go b/sets.go
index b186840..2c82600 100644
--- a/sets.go
+++ b/sets.go
@@ -13,6 +13,11 @@ func NewSet(timestamp Nanotime, values map[string]struct{}, source Source, tags
 	return Set{Values: values, Timestamp: timestamp, Source: source, Tags: tags.Copy()}
 }
 
+func (s *Set) AddTagsSetSource(additionalTags Tags, newSource Source) {
+	s.Tags = s.Tags.Concat(additionalTags)
+	s.Source = newSource
+}
+
 // Sets stores a map of sets by tags.
 type Sets map[string]map[string]Set
 
diff --git a/timers.go b/timers.go
index fd78231..e8d52bd 100644
--- a/timers.go
+++ b/timers.go
@@ -1,6 +1,8 @@
 package gostatsd
 
-import "github.com/spf13/viper"
+import (
+	"github.com/spf13/viper"
+)
 
 // Timer is used for storing aggregated values for timers.
 type Timer struct {
@@ -37,6 +39,11 @@ func NewTimerValues(values []float64) Timer {
 	return NewTimer(Nanotime(0), values, "", nil)
 }
 
+func (t *Timer) AddTagsSetSource(additionalTags Tags, newSource Source) {
+	t.Tags = t.Tags.Concat(additionalTags)
+	t.Source = newSource
+}
+
 // Timers stores a map of timers by tags.
 type Timers map[string]map[string]Timer
 
@@ -108,5 +115,4 @@ func DisabledSubMetrics(viper *viper.Viper) TimerSubtypes {
 		SumSquares:     subViper.GetBool("sum-squares"),
 		SumSquaresPct:  subViper.GetBool("sum-squares-pct"),
 	}
-
 }


Test patch:
diff --git a/metrics_test.go b/metrics_test.go
index 651cd50..cb47477 100644
--- a/metrics_test.go
+++ b/metrics_test.go
@@ -36,20 +36,20 @@ func TestMetricString(t *testing.T) {
 	}
 }
 
-func TestUpdateTags(t *testing.T) {
-	m := &Metric{}
-	m.FormatTagsKey()
-	require.EqualValues(t, "", m.TagsKey)
-	m.AddTagsSetSource(Tags{"foo"}, "source")
-	require.EqualValues(t, Tags{"foo"}, m.Tags)
-	require.EqualValues(t, "source", m.Source)
-	require.EqualValues(t, "", m.TagsKey)
-	m.FormatTagsKey()
-	require.EqualValues(t, "foo,s:source", m.TagsKey) // It's set
-	m.AddTagsSetSource(Tags{"foo2"}, "source2")
-	require.EqualValues(t, Tags{"foo", "foo2"}, m.Tags)
-	require.EqualValues(t, "source2", m.Source)
-	require.EqualValues(t, "", m.TagsKey) // It's cleared
-	m.FormatTagsKey()
-	require.EqualValues(t, "foo,foo2,s:source2", m.TagsKey)
+func TestAddTagsSetSource(t *testing.T) {
+	mCounter := Counter{}
+	mCounter.AddTagsSetSource(Tags{"foo"}, "source")
+	require.Equal(t, Tags{"foo"}, mCounter.Tags)
+
+	mGauge := Gauge{}
+	mGauge.AddTagsSetSource(Tags{"foo"}, "source")
+	require.Equal(t, Tags{"foo"}, mGauge.Tags)
+
+	mSet := Set{}
+	mSet.AddTagsSetSource(Tags{"foo"}, "source")
+	require.Equal(t, Tags{"foo"}, mSet.Tags)
+
+	mTimer := Timer{}
+	mTimer.AddTagsSetSource(Tags{"foo"}, "source")
+	require.Equal(t, Tags{"foo"}, mTimer.Tags)
 }