Avoid unnecessary conversions between model.Metric and labels.Labels

Signed-off-by: Xiaochao Dong (@damnever) <[email protected]>

Author: damnever

pkg/cortexpb/signature.go

@@ -0,0 +1,50 @@
+package cortexpb
+
+import (
+	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/model/labels"
+)
+
+// Inline and byte-free variant of hash/fnv's fnv64a.
+// Ref: https://github.com/prometheus/common/blob/main/model/fnv.go
+
+func LabelsToFingerprint(lset labels.Labels) model.Fingerprint {
+	if len(lset) == 0 {
+		return model.Fingerprint(hashNew())
+	}
+
+	sum := hashNew()
+	for _, l := range lset {
+		sum = hashAdd(sum, string(l.Name))
+		sum = hashAddByte(sum, model.SeparatorByte)
+		sum = hashAdd(sum, string(l.Value))
+		sum = hashAddByte(sum, model.SeparatorByte)
+	}
+	return model.Fingerprint(sum)
+}
+
+const (
+	offset64 = 14695981039346656037
+	prime64  = 1099511628211
+)
+
+// hashNew initializes a new fnv64a hash value.
+func hashNew() uint64 {
+	return offset64
+}
+
+// hashAdd adds a string to a fnv64a hash value, returning the updated hash.
+func hashAdd(h uint64, s string) uint64 {
+	for i := 0; i < len(s); i++ {
+		h ^= uint64(s[i])
+		h *= prime64
+	}
+	return h
+}
+
+// hashAddByte adds a byte to a fnv64a hash value, returning the updated hash.
+func hashAddByte(h uint64, b byte) uint64 {
+	h ^= uint64(b)
+	h *= prime64
+	return h
+}

pkg/distributor/distributor.go

@@ -1367,8 +1367,8 @@ func (d *Distributor) LabelNames(ctx context.Context, from, to model.Time, hint
 }
 
 // MetricsForLabelMatchers gets the metrics that match said matchers
-func (d *Distributor) MetricsForLabelMatchers(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]model.Metric, error) {
-	return d.metricsForLabelMatchersCommon(ctx, from, through, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.MetricsForLabelMatchersRequest, metrics *map[model.Fingerprint]model.Metric, mutex *sync.Mutex, queryLimiter *limiter.QueryLimiter) error {
+func (d *Distributor) MetricsForLabelMatchers(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error) {
+	return d.metricsForLabelMatchersCommon(ctx, from, through, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.MetricsForLabelMatchersRequest, metrics *map[model.Fingerprint]labels.Labels, mutex *sync.Mutex, queryLimiter *limiter.QueryLimiter) error {
 		_, err := d.ForReplicationSet(ctx, rs, false, partialDataEnabled, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
 			resp, err := client.MetricsForLabelMatchers(ctx, req)
 			if err != nil {
@@ -1380,8 +1380,8 @@ func (d *Distributor) MetricsForLabelMatchers(ctx context.Context, from, through
 			s := make([][]cortexpb.LabelAdapter, 0, len(resp.Metric))
 			for _, m := range resp.Metric {
 				s = append(s, m.Labels)
-				m := cortexpb.FromLabelAdaptersToMetric(m.Labels)
-				fingerprint := m.Fingerprint()
+				m := cortexpb.FromLabelAdaptersToLabels(m.Labels)
+				fingerprint := cortexpb.LabelsToFingerprint(m)
 				mutex.Lock()
 				(*metrics)[fingerprint] = m
 				mutex.Unlock()
@@ -1396,8 +1396,8 @@ func (d *Distributor) MetricsForLabelMatchers(ctx context.Context, from, through
 	}, matchers...)
 }
 
-func (d *Distributor) MetricsForLabelMatchersStream(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]model.Metric, error) {
-	return d.metricsForLabelMatchersCommon(ctx, from, through, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.MetricsForLabelMatchersRequest, metrics *map[model.Fingerprint]model.Metric, mutex *sync.Mutex, queryLimiter *limiter.QueryLimiter) error {
+func (d *Distributor) MetricsForLabelMatchersStream(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error) {
+	return d.metricsForLabelMatchersCommon(ctx, from, through, hint, func(ctx context.Context, rs ring.ReplicationSet, req *ingester_client.MetricsForLabelMatchersRequest, metrics *map[model.Fingerprint]labels.Labels, mutex *sync.Mutex, queryLimiter *limiter.QueryLimiter) error {
 		_, err := d.ForReplicationSet(ctx, rs, false, partialDataEnabled, func(ctx context.Context, client ingester_client.IngesterClient) (interface{}, error) {
 			stream, err := client.MetricsForLabelMatchersStream(ctx, req)
 			if err != nil {
@@ -1417,9 +1417,9 @@ func (d *Distributor) MetricsForLabelMatchersStream(ctx context.Context, from, t
 
 				s := make([][]cortexpb.LabelAdapter, 0, len(resp.Metric))
 				for _, metric := range resp.Metric {
-					m := cortexpb.FromLabelAdaptersToMetricWithCopy(metric.Labels)
+					m := cortexpb.FromLabelAdaptersToLabels(metric.Labels)
 					s = append(s, metric.Labels)
-					fingerprint := m.Fingerprint()
+					fingerprint := cortexpb.LabelsToFingerprint(m)
 					mutex.Lock()
 					(*metrics)[fingerprint] = m
 					mutex.Unlock()
@@ -1436,7 +1436,7 @@ func (d *Distributor) MetricsForLabelMatchersStream(ctx context.Context, from, t
 	}, matchers...)
 }
 
-func (d *Distributor) metricsForLabelMatchersCommon(ctx context.Context, from, through model.Time, hints *storage.SelectHints, f func(context.Context, ring.ReplicationSet, *ingester_client.MetricsForLabelMatchersRequest, *map[model.Fingerprint]model.Metric, *sync.Mutex, *limiter.QueryLimiter) error, matchers ...*labels.Matcher) ([]model.Metric, error) {
+func (d *Distributor) metricsForLabelMatchersCommon(ctx context.Context, from, through model.Time, hints *storage.SelectHints, f func(context.Context, ring.ReplicationSet, *ingester_client.MetricsForLabelMatchersRequest, *map[model.Fingerprint]labels.Labels, *sync.Mutex, *limiter.QueryLimiter) error, matchers ...*labels.Matcher) ([]labels.Labels, error) {
 	replicationSet, err := d.GetIngestersForMetadata(ctx)
 	queryLimiter := limiter.QueryLimiterFromContextWithFallback(ctx)
 	if err != nil {
@@ -1448,7 +1448,7 @@ func (d *Distributor) metricsForLabelMatchersCommon(ctx context.Context, from, t
 		return nil, err
 	}
 	mutex := sync.Mutex{}
-	metrics := map[model.Fingerprint]model.Metric{}
+	metrics := map[model.Fingerprint]labels.Labels{}
 
 	err = f(ctx, replicationSet, req, &metrics, &mutex, queryLimiter)
 
@@ -1457,7 +1457,7 @@ func (d *Distributor) metricsForLabelMatchersCommon(ctx context.Context, from, t
 	}
 
 	mutex.Lock()
-	result := make([]model.Metric, 0, len(metrics))
+	result := make([]labels.Labels, 0, len(metrics))
 	for _, m := range metrics {
 		result = append(result, m)
 	}

pkg/distributor/distributor_test.go

@@ -502,7 +502,7 @@ func TestDistributor_MetricsCleanup(t *testing.T) {
 		# HELP cortex_distributor_exemplars_in_total The total number of exemplars that have come in to the distributor, including rejected or deduped exemplars.
 		# TYPE cortex_distributor_exemplars_in_total counter
 		cortex_distributor_exemplars_in_total{user="userA"} 5
-		
+
 		# HELP cortex_distributor_ingester_append_failures_total The total number of failed batch appends sent to ingesters.
 		# TYPE cortex_distributor_ingester_append_failures_total counter
 		cortex_distributor_ingester_append_failures_total{ingester="ingester-0",status="2xx",type="metadata"} 1
@@ -2459,7 +2459,7 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 		shuffleShardEnabled bool
 		shuffleShardSize    int
 		matchers            []*labels.Matcher
-		expectedResult      []model.Metric
+		expectedResult      []labels.Labels
 		expectedIngesters   int
 		queryLimiter        *limiter.QueryLimiter
 		expectedErr         error
@@ -2468,7 +2468,7 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 			matchers: []*labels.Matcher{
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "unknown"),
 			},
-			expectedResult:    []model.Metric{},
+			expectedResult:    []labels.Labels{},
 			expectedIngesters: numIngesters,
 			queryLimiter:      limiter.NewQueryLimiter(0, 0, 0, 0),
 			expectedErr:       nil,
@@ -2477,9 +2477,9 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 			matchers: []*labels.Matcher{
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "test_1"),
 			},
-			expectedResult: []model.Metric{
-				util.LabelsToMetric(fixtures[0].lbls),
-				util.LabelsToMetric(fixtures[1].lbls),
+			expectedResult: []labels.Labels{
+				fixtures[0].lbls,
+				fixtures[1].lbls,
 			},
 			expectedIngesters: numIngesters,
 			queryLimiter:      limiter.NewQueryLimiter(0, 0, 0, 0),
@@ -2490,8 +2490,8 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 				mustNewMatcher(labels.MatchEqual, "status", "200"),
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "test_1"),
 			},
-			expectedResult: []model.Metric{
-				util.LabelsToMetric(fixtures[0].lbls),
+			expectedResult: []labels.Labels{
+				fixtures[0].lbls,
 			},
 			expectedIngesters: numIngesters,
 			queryLimiter:      limiter.NewQueryLimiter(0, 0, 0, 0),
@@ -2501,9 +2501,9 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 			matchers: []*labels.Matcher{
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "fast_fingerprint_collision"),
 			},
-			expectedResult: []model.Metric{
-				util.LabelsToMetric(fixtures[3].lbls),
-				util.LabelsToMetric(fixtures[4].lbls),
+			expectedResult: []labels.Labels{
+				fixtures[3].lbls,
+				fixtures[4].lbls,
 			},
 			expectedIngesters: numIngesters,
 			queryLimiter:      limiter.NewQueryLimiter(0, 0, 0, 0),
@@ -2515,9 +2515,9 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 			matchers: []*labels.Matcher{
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "test_1"),
 			},
-			expectedResult: []model.Metric{
-				util.LabelsToMetric(fixtures[0].lbls),
-				util.LabelsToMetric(fixtures[1].lbls),
+			expectedResult: []labels.Labels{
+				fixtures[0].lbls,
+				fixtures[1].lbls,
 			},
 			expectedIngesters: 3,
 			queryLimiter:      limiter.NewQueryLimiter(0, 0, 0, 0),
@@ -2529,9 +2529,9 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 			matchers: []*labels.Matcher{
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "test_1"),
 			},
-			expectedResult: []model.Metric{
-				util.LabelsToMetric(fixtures[0].lbls),
-				util.LabelsToMetric(fixtures[1].lbls),
+			expectedResult: []labels.Labels{
+				fixtures[0].lbls,
+				fixtures[1].lbls,
 			},
 			expectedIngesters: numIngesters,
 			queryLimiter:      limiter.NewQueryLimiter(0, 0, 0, 0),
@@ -2563,8 +2563,8 @@ func TestDistributor_MetricsForLabelMatchers(t *testing.T) {
 			matchers: []*labels.Matcher{
 				mustNewMatcher(labels.MatchEqual, model.MetricNameLabel, "test_2"),
 			},
-			expectedResult: []model.Metric{
-				util.LabelsToMetric(fixtures[2].lbls),
+			expectedResult: []labels.Labels{
+				fixtures[2].lbls,
 			},
 			expectedIngesters: numIngesters,
 			queryLimiter:      limiter.NewQueryLimiter(1, 0, 0, 0),

pkg/querier/distributor_queryable.go

@@ -33,8 +33,8 @@ type Distributor interface {
 	LabelValuesForLabelNameStream(ctx context.Context, from, to model.Time, label model.LabelName, hint *storage.LabelHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]string, error)
 	LabelNames(context.Context, model.Time, model.Time, *storage.LabelHints, bool, ...*labels.Matcher) ([]string, error)
 	LabelNamesStream(context.Context, model.Time, model.Time, *storage.LabelHints, bool, ...*labels.Matcher) ([]string, error)
-	MetricsForLabelMatchers(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]model.Metric, error)
-	MetricsForLabelMatchersStream(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]model.Metric, error)
+	MetricsForLabelMatchers(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error)
+	MetricsForLabelMatchersStream(ctx context.Context, from, through model.Time, hint *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error)
 	MetricsMetadata(ctx context.Context, req *client.MetricsMetadataRequest) ([]scrape.MetricMetadata, error)
 }
 
@@ -122,7 +122,7 @@ func (q *distributorQuerier) Select(ctx context.Context, sortSeries bool, sp *st
 	// See: https://github.com/prometheus/prometheus/pull/8050
 	if sp != nil && sp.Func == "series" {
 		var (
-			ms  []model.Metric
+			ms  []labels.Labels
 			err error
 		)
 
@@ -136,14 +136,14 @@ func (q *distributorQuerier) Select(ctx context.Context, sortSeries bool, sp *st
 			return storage.ErrSeriesSet(err)
 		}
 
-		seriesSet := series.MetricsToSeriesSet(ctx, sortSeries, ms)
+		seriesSet := series.LabelsSetToSeriesSet(sortSeries, ms)
 
 		if partialdata.IsPartialDataError(err) {
 			warning := seriesSet.Warnings()
 			return series.NewSeriesSetWithWarnings(seriesSet, warning.Add(err))
 		}
 
-		return series.MetricsToSeriesSet(ctx, sortSeries, ms)
+		return seriesSet
 	}
 
 	return q.streamingSelect(ctx, sortSeries, partialDataEnabled, minT, maxT, matchers)
@@ -249,7 +249,7 @@ func (q *distributorQuerier) labelNamesWithMatchers(ctx context.Context, hints *
 	defer log.Span.Finish()
 
 	var (
-		ms  []model.Metric
+		ms  []labels.Labels
 		err error
 	)
 
@@ -265,8 +265,8 @@ func (q *distributorQuerier) labelNamesWithMatchers(ctx context.Context, hints *
 	namesMap := make(map[string]struct{})
 
 	for _, m := range ms {
-		for name := range m {
-			namesMap[string(name)] = struct{}{}
+		for _, l := range m {
+			namesMap[l.Name] = struct{}{}
 		}
 	}
 

pkg/querier/distributor_queryable_test.go

@@ -224,10 +224,10 @@ func TestDistributorQuerier_LabelNames(t *testing.T) {
 				t.Run("with matchers", func(t *testing.T) {
 					t.Parallel()
 
-					metrics := []model.Metric{
-						{"foo": "bar"},
-						{"job": "baz"},
-						{"job": "baz", "foo": "boom"},
+					metrics := []labels.Labels{
+						labels.FromStrings("foo", "bar"),
+						labels.FromStrings("job", "baz"),
+						labels.FromStrings("job", "baz", "foo", "boom"),
 					}
 					d := &MockDistributor{}
 

pkg/querier/series/series_set.go

@@ -18,6 +18,7 @@ package series
 
 import (
 	"context"
+	"slices"
 	"sort"
 
 	"github.com/prometheus/common/model"
@@ -30,6 +31,44 @@ import (
 	"github.com/cortexproject/cortex/pkg/util"
 )
 
+type labelsSetSeriesSet struct {
+	cur       int
+	labelsSet []labels.Labels
+}
+
+// LabelsSetToSeriesSet creates a storage.SeriesSet from a []labels.Labels.
+func LabelsSetToSeriesSet(sortSeries bool, labelsSet []labels.Labels) storage.SeriesSet {
+	if sortSeries {
+		slices.SortFunc(labelsSet, func(a, b labels.Labels) int { return labels.Compare(a, b) })
+	}
+	return &labelsSetSeriesSet{
+		cur:       -1,
+		labelsSet: labelsSet,
+	}
+}
+
+// Next iterates through a series set and implements storage.SeriesSet.
+func (c *labelsSetSeriesSet) Next() bool {
+	c.cur++
+	return c.cur < len(c.labelsSet)
+}
+
+// At returns the current series and implements storage.SeriesSet.
+func (c *labelsSetSeriesSet) At() storage.Series {
+	lset := c.labelsSet[c.cur]
+	return &ConcreteSeries{labels: lset}
+}
+
+// Err implements storage.SeriesSet.
+func (c *labelsSetSeriesSet) Err() error {
+	return nil
+}
+
+// Warnings implements storage.SeriesSet.
+func (c *labelsSetSeriesSet) Warnings() annotations.Annotations {
+	return nil
+}
+
 // ConcreteSeriesSet implements storage.SeriesSet.
 type ConcreteSeriesSet struct {
 	cur    int

pkg/querier/testutils.go

@@ -49,13 +49,13 @@ func (m *MockDistributor) LabelNamesStream(ctx context.Context, from model.Time,
 	args := m.Called(ctx, from, to, hints, matchers)
 	return args.Get(0).([]string), args.Error(1)
 }
-func (m *MockDistributor) MetricsForLabelMatchers(ctx context.Context, from, to model.Time, hints *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]model.Metric, error) {
+func (m *MockDistributor) MetricsForLabelMatchers(ctx context.Context, from, to model.Time, hints *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error) {
 	args := m.Called(ctx, from, to, hints, matchers)
-	return args.Get(0).([]model.Metric), args.Error(1)
+	return args.Get(0).([]labels.Labels), args.Error(1)
 }
-func (m *MockDistributor) MetricsForLabelMatchersStream(ctx context.Context, from, to model.Time, hints *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]model.Metric, error) {
+func (m *MockDistributor) MetricsForLabelMatchersStream(ctx context.Context, from, to model.Time, hints *storage.SelectHints, partialDataEnabled bool, matchers ...*labels.Matcher) ([]labels.Labels, error) {
 	args := m.Called(ctx, from, to, hints, matchers)
-	return args.Get(0).([]model.Metric), args.Error(1)
+	return args.Get(0).([]labels.Labels), args.Error(1)
 }
 
 func (m *MockDistributor) MetricsMetadata(ctx context.Context, request *client.MetricsMetadataRequest) ([]scrape.MetricMetadata, error) {