Fix multi-metric aggregation

statsd/buffer.go

@@ -72,6 +72,7 @@ func (b *statsdBuffer) writeAggregated(metricSymbol []byte, namespace string, gl
 	}
 
 	originalBuffer := b.buffer
+	b.writeSeparator()
 	b.buffer = appendHeader(b.buffer, namespace, name)
 
 	// buffer already full

statsd/worker.go

@@ -90,7 +90,7 @@ func (w *worker) writeAggregatedMetricUnsafe(m metric, metricSymbol []byte) erro
 
 	// first check how much data we can write to the buffer:
 	//   +3 + len(metricSymbol) because the message will include '|<metricSymbol>|#' before the tags
-	//   +1 for the coma between the two set of tags
+	//   +1 for the potential line break at the start of the metric
 	tagsSize := len(m.stags) + 4 + len(metricSymbol)
 	for _, t := range m.globalTags {
 		tagsSize += len(t) + 1

statsd/worker_test.go

@@ -253,3 +253,37 @@ func TestWorkerDistributionAggregatedMultiple(t *testing.T) {
 	data = <-s.queue
 	assert.Equal(t, "namespace.test_distribution:3.3:4.4|d|#globalTags,globalTags2,tag1,tag2", string(data.buffer))
 }
+
+func TestWorkerMultipleDifferentDistributionAggregated(t *testing.T) {
+	// first metric will fit but not the second one
+	_, s, w := initWorker(160)
+
+	m := metric{
+		metricType: distributionAggregated,
+		namespace:  "namespace.",
+		globalTags: []string{"globalTags", "globalTags2"},
+		name:       "test_distribution",
+		fvalues:    []float64{1.1, 2.2, 3.3, 4.4},
+		stags:      "tag1,tag2",
+		rate:       1,
+	}
+	err := w.processMetric(m)
+	assert.Nil(t, err)
+	m = metric{
+		metricType: distributionAggregated,
+		namespace:  "namespace.",
+		globalTags: []string{"globalTags", "globalTags2"},
+		name:       "test_distribution_2",
+		fvalues:    []float64{1.1, 2.2, 3.3, 4.4},
+		stags:      "tag1,tag2",
+		rate:       1,
+	}
+	err = w.processMetric(m)
+	assert.Nil(t, err)
+
+	w.flush()
+	data := <-s.queue
+	assert.Equal(t, "namespace.test_distribution:1.1:2.2:3.3:4.4|d|#globalTags,globalTags2,tag1,tag2\nnamespace.test_distribution_2:1.1:2.2:3.3|d|#globalTags,globalTags2,tag1,tag2", string(data.buffer))
+	data = <-s.queue
+	assert.Equal(t, "namespace.test_distribution_2:4.4|d|#globalTags,globalTags2,tag1,tag2", string(data.buffer))
+}