Better use label reusing.

Author: mattdurham

go.mod

@@ -5,6 +5,7 @@ go 1.23
 toolchain go1.23.1
 
 require (
+	github.com/dghubble/trie v0.1.0
 	github.com/dolthub/swiss v0.2.1
 	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3
 	github.com/go-kit/log v0.2.1

go.sum

@@ -26,6 +26,8 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dennwc/varint v1.0.0 h1:kGNFFSSw8ToIy3obO/kKr8U9GZYUAxQEVuix4zfDWzE=
 github.com/dennwc/varint v1.0.0/go.mod h1:hnItb35rvZvJrbTALZtY/iQfDs48JKRG1RPpgziApxA=
+github.com/dghubble/trie v0.1.0 h1:kJnjBLFFElBwS60N4tkPvnLhnpcDxbBjIulgI8CpNGM=
+github.com/dghubble/trie v0.1.0/go.mod h1:sOmnzfBNH7H92ow2292dDFWNsVQuh/izuD7otCYb1ak=
 github.com/dolthub/maphash v0.1.0 h1:bsQ7JsF4FkkWyrP3oCnFJgrCUAFbFf3kOl4L/QxPDyQ=
 github.com/dolthub/maphash v0.1.0/go.mod h1:gkg4Ch4CdCDu5h6PMriVLawB7koZ+5ijb9puGMV50a4=
 github.com/dolthub/swiss v0.2.1 h1:gs2osYs5SJkAaH5/ggVJqXQxRXtWshF6uE0lgR/Y3Gw=

serialization/appender.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/grafana/walqueue/types/v2"
+	"sync"
 	"time"
 
 	"github.com/go-kit/log"
@@ -15,6 +16,34 @@ import (
 	"github.com/prometheus/prometheus/storage"
 )
 
+var metricSinglePool = sync.Pool{
+	New: func() any {
+		return &types.Metric{}
+	},
+}
+
+func GetMetricFromPool() *types.Metric {
+	return metricSinglePool.Get().(*types.Metric)
+}
+
+func PutMetricSliceIntoPool(m []*types.Metric) {
+	for _, mt := range m {
+		PutMetricIntoPool(mt)
+	}
+}
+
+func PutMetricIntoPool(m *types.Metric) {
+	m.Hash = 0
+	m.TS = 0
+	m.Value = 0
+	// We dont reuse these labels since they are owned by the scraper.
+	m.Labels = nil
+	m.Histogram = nil
+	m.FloatHistogram = nil
+
+	metricSinglePool.Put(m)
+}
+
 type appender struct {
 	ctx          context.Context
 	ttl          time.Duration
@@ -42,8 +71,8 @@ func NewAppender(ctx context.Context, ttl time.Duration, s types.Serializer, log
 
 // Append metric
 func (a *appender) Append(ref storage.SeriesRef, l labels.Labels, t int64, v float64) (storage.SeriesRef, error) {
-	ts := types.GetMetricFromPool()
-	ts.Labels = l.Copy()
+	ts := types.GetMetricFromPool(false)
+	ts.Labels = l
 	ts.TS = t
 	ts.Value = v
 	ts.Hash = l.Hash()
@@ -67,7 +96,7 @@ func (a *appender) AppendExemplar(ref storage.SeriesRef, _ labels.Labels, e exem
 	if e.HasTs && e.Ts < endTime {
 		return ref, nil
 	}
-	ts := types.GetMetricFromPool()
+	ts := types.GetMetricFromPool(false)
 	ts.Hash = e.Labels.Hash()
 	ts.TS = e.Ts
 	ts.Labels = e.Labels.Copy()
@@ -82,7 +111,7 @@ func (a *appender) AppendHistogram(ref storage.SeriesRef, l labels.Labels, t int
 	if t < endTime {
 		return ref, nil
 	}
-	ts := types.GetMetricFromPool()
+	ts := types.GetMetricFromPool(false)
 	ts.Labels = l.Copy()
 	ts.TS = t
 	if h != nil {

serialization/serializer_test.go

@@ -38,7 +38,7 @@ func TestRoundTripSerialization(t *testing.T) {
 	s.Start()
 	defer s.Stop()
 	for i := 0; i < 10; i++ {
-		tss := types.GetMetricFromPool()
+		tss := types.GetMetricFromPool(true)
 		tss.Labels = make(labels.Labels, 10)
 		for j := 0; j < 10; j++ {
 			tss.Labels[j] = labels.Label{
@@ -101,7 +101,7 @@ func (f *fqq) Store(ctx context.Context, meta map[string]string, value []byte) e
 	metrics, _, err := v2.DeserializeToSeriesGroup(sg, f.buf)
 	require.NoError(f.t, err)
 	require.Len(f.t, sg.Series, 10)
-	for _, series := range metrics {
+	for _, series := range metrics.M {
 		require.Len(f.t, series.Labels, 10)
 		for j := 0; j < 10; j++ {
 			series.Labels[j].Name = fmt.Sprintf("name_%d_%d", int(series.Value), j)

types/metric.go

@@ -21,20 +21,21 @@ type Metrics struct {
 	M []*Metric
 }
 
-func (m *Metrics) Resize(length int) {
+func (m *Metrics) Resize(length int, canLabelsBeReused bool) {
 	m.M = make([]*Metric, length)
 	for i := range m.M {
-		m.M[i] = GetMetricFromPool()
+		m.M[i] = GetMetricFromPool(canLabelsBeReused)
 	}
 }
 
 type Metric struct {
-	Labels         labels.Labels
-	TS             int64
-	Value          float64
-	Hash           uint64
-	Histogram      *histogram.Histogram
-	FloatHistogram *histogram.FloatHistogram
+	Labels            labels.Labels
+	TS                int64
+	Value             float64
+	Hash              uint64
+	Histogram         *histogram.Histogram
+	FloatHistogram    *histogram.FloatHistogram
+	canLabelsBeReused bool
 }
 
 // IsMetadata is used because it's easier to store metadata as a set of labels.
@@ -43,15 +44,28 @@ func (m Metric) IsMetadata() bool {
 }
 
 var OutstandingIndividualMetrics = atomic.Int32{}
-var metricSinglePool = sync.Pool{
+var metricReusableLabelPool = sync.Pool{
 	New: func() any {
 		return &Metric{}
 	},
 }
 
-func GetMetricFromPool() *Metric {
+var metricNonReusablePool = sync.Pool{
+	New: func() any {
+		return &Metric{}
+	},
+}
+
+func GetMetricFromPool(canLabelsBeReused bool) *Metric {
 	OutstandingIndividualMetrics.Inc()
-	return metricSinglePool.Get().(*Metric)
+	var m *Metric
+	if canLabelsBeReused {
+		m = metricReusableLabelPool.Get().(*Metric)
+	} else {
+		m = metricNonReusablePool.Get().(*Metric)
+	}
+	m.canLabelsBeReused = canLabelsBeReused
+	return m
 }
 
 func PutMetricSliceIntoPool(m []*Metric) {
@@ -66,9 +80,17 @@ func PutMetricIntoPool(m *Metric) {
 	m.Hash = 0
 	m.TS = 0
 	m.Value = 0
-	m.Labels = m.Labels[:0]
+	if m.canLabelsBeReused {
+		m.Labels = m.Labels[:0]
+	} else {
+		m.Labels = nil
+	}
 	m.Histogram = nil
 	m.FloatHistogram = nil
 
-	metricSinglePool.Put(m)
+	if m.canLabelsBeReused {
+		metricReusableLabelPool.Put(m)
+	} else {
+		metricNonReusablePool.Put(m)
+	}
 }

types/v1/serialization_func.go

@@ -268,7 +268,7 @@ func DeserializeToSeriesGroup(sg *SeriesGroup, buf []byte) (*types.Metrics, *typ
 	}
 	// Need to fill in the labels.
 	metrics := &types.Metrics{}
-	metrics.Resize(len(sg.Series))
+	metrics.Resize(len(sg.Series), true)
 	for seriesIndex, series := range sg.Series {
 		metric := metrics.M[seriesIndex]
 		if cap(metric.Labels) < len(series.LabelsNames) {
@@ -288,7 +288,7 @@ func DeserializeToSeriesGroup(sg *SeriesGroup, buf []byte) (*types.Metrics, *typ
 		series.LabelsValues = series.LabelsValues[:0]
 	}
 	meta := &types.Metrics{}
-	meta.Resize(len(sg.Metadata))
+	meta.Resize(len(sg.Metadata), true)
 	for seriesIndex, series := range sg.Metadata {
 		m := meta.M[seriesIndex]
 		if cap(m.Labels) < len(series.LabelsNames) {

types/v2/serialization_func.go

@@ -60,7 +60,7 @@ func DeserializeToSeriesGroup(sg *SeriesGroup, buf []byte) (*types.Metrics, *typ
 		return nil, nil, err
 	}
 	metrics := &types.Metrics{}
-	metrics.Resize(len(sg.Series))
+	metrics.Resize(len(sg.Series), true)
 
 	// Need to fill in the labels.
 	for seriesIndex, series := range sg.Series {
@@ -92,7 +92,7 @@ func DeserializeToSeriesGroup(sg *SeriesGroup, buf []byte) (*types.Metrics, *typ
 	}
 
 	metadata := &types.Metrics{}
-	metadata.Resize(len(sg.Metadata))
+	metadata.Resize(len(sg.Metadata), true)
 	for seriesIndex, series := range sg.Metadata {
 		meta := metadata.M[seriesIndex]