Fixes store duplicate label names and values. (#1790)

* Fixes store label name and values duplicate.

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

* Normalize string unicity control

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

* Adds benchmark to compare the new unique algorithm

Signed-off-by: Cyril Tovena <cyril.tovena@gmail.com>

Author: cyriltovena

pkg/chunk/chunk_store.go

@@ -204,17 +204,15 @@ func (c *store) LabelValuesForMetricName(ctx context.Context, userID string, fro
 		return nil, err
 	}
 
-	var result []string
+	var result UniqueStrings
 	for _, entry := range entries {
 		_, labelValue, _, _, err := parseChunkTimeRangeValue(entry.RangeValue, entry.Value)
 		if err != nil {
 			return nil, err
 		}
-		result = append(result, string(labelValue))
+		result.Add(string(labelValue))
 	}
-	sort.Strings(result)
-	result = uniqueStrings(result)
-	return result, nil
+	return result.Strings(), nil
 }
 
 // LabelNamesForMetricName retrieves all label names for a metric name.
@@ -462,7 +460,6 @@ func (c *store) lookupEntriesByQueries(ctx context.Context, queries []IndexQuery
 
 func (c *store) parseIndexEntries(ctx context.Context, entries []IndexEntry, matcher *labels.Matcher) ([]string, error) {
 	result := make([]string, 0, len(entries))
-
 	for _, entry := range entries {
 		chunkKey, labelValue, _, _, err := parseChunkTimeRangeValue(entry.RangeValue, entry.Value)
 		if err != nil {
@@ -474,7 +471,6 @@ func (c *store) parseIndexEntries(ctx context.Context, entries []IndexEntry, mat
 		}
 		result = append(result, chunkKey)
 	}
-
 	// Return ids sorted and deduped because they will be merged with other sets.
 	sort.Strings(result)
 	result = uniqueStrings(result)

pkg/chunk/chunk_store_test.go

@@ -851,3 +851,49 @@ func TestStoreMaxLookBack(t *testing.T) {
 	require.Equal(t, 1, len(chunks))
 	chunks[0].Through.Equal(now)
 }
+
+func benchmarkParseIndexEntries(i int64, b *testing.B) {
+	b.ReportAllocs()
+	b.StopTimer()
+	store := &store{}
+	ctx := context.Background()
+	entries := generateIndexEntries(i)
+	matcher, err := labels.NewMatcher(labels.MatchRegexp, "", ".*")
+	if err != nil {
+		b.Fatal(err)
+	}
+	b.StartTimer()
+	for n := 0; n < b.N; n++ {
+		keys, err := store.parseIndexEntries(ctx, entries, matcher)
+		if err != nil {
+			b.Fatal(err)
+		}
+		if len(keys) != len(entries)/2 {
+			b.Fatalf("expected keys:%d got:%d", len(entries)/2, len(keys))
+		}
+	}
+}
+
+func BenchmarkParseIndexEntries500(b *testing.B)   { benchmarkParseIndexEntries(500, b) }
+func BenchmarkParseIndexEntries2500(b *testing.B)  { benchmarkParseIndexEntries(2500, b) }
+func BenchmarkParseIndexEntries10000(b *testing.B) { benchmarkParseIndexEntries(10000, b) }
+func BenchmarkParseIndexEntries50000(b *testing.B) { benchmarkParseIndexEntries(50000, b) }
+
+func generateIndexEntries(n int64) []IndexEntry {
+	res := make([]IndexEntry, 0, n)
+	for i := int64(n - 1); i >= 0; i-- {
+		labelValue := fmt.Sprintf("labelvalue%d", i%(n/2))
+		chunkID := fmt.Sprintf("chunkid%d", i%(n/2))
+		rangeValue := []byte{}
+		rangeValue = append(rangeValue, []byte("component1")...)
+		rangeValue = append(rangeValue, 0)
+		rangeValue = append(rangeValue, []byte(labelValue)...)
+		rangeValue = append(rangeValue, 0)
+		rangeValue = append(rangeValue, []byte(chunkID)...)
+		rangeValue = append(rangeValue, 0)
+		res = append(res, IndexEntry{
+			RangeValue: rangeValue,
+		})
+	}
+	return res
+}

pkg/chunk/chunk_store_utils.go

@@ -2,7 +2,6 @@ package chunk
 
 import (
 	"context"
-	"sort"
 	"sync"
 
 	"github.com/go-kit/kit/log/level"
@@ -38,18 +37,13 @@ func keysFromChunks(chunks []Chunk) []string {
 }
 
 func labelNamesFromChunks(chunks []Chunk) []string {
-	keys := map[string]struct{}{}
-	var result []string
+	var result UniqueStrings
 	for _, c := range chunks {
 		for _, l := range c.Metric {
-			if _, ok := keys[string(l.Name)]; !ok {
-				keys[string(l.Name)] = struct{}{}
-				result = append(result, string(l.Name))
-			}
+			result.Add(string(l.Name))
 		}
 	}
-	sort.Strings(result)
-	return result
+	return result.Strings()
 }
 
 func filterChunksByUniqueFingerprint(chunks []Chunk) ([]Chunk, []string) {

pkg/chunk/composite_store.go

@@ -100,30 +100,30 @@ func (c compositeStore) Get(ctx context.Context, userID string, from, through mo
 
 // LabelValuesForMetricName retrieves all label values for a single label name and metric name.
 func (c compositeStore) LabelValuesForMetricName(ctx context.Context, userID string, from, through model.Time, metricName string, labelName string) ([]string, error) {
-	var result []string
+	var result UniqueStrings
 	err := c.forStores(from, through, func(from, through model.Time, store Store) error {
 		labelValues, err := store.LabelValuesForMetricName(ctx, userID, from, through, metricName, labelName)
 		if err != nil {
 			return err
 		}
-		result = append(result, labelValues...)
+		result.Add(labelValues...)
 		return nil
 	})
-	return result, err
+	return result.Strings(), err
 }
 
 // LabelNamesForMetricName retrieves all label names for a metric name.
 func (c compositeStore) LabelNamesForMetricName(ctx context.Context, userID string, from, through model.Time, metricName string) ([]string, error) {
-	var result []string
+	var result UniqueStrings
 	err := c.forStores(from, through, func(from, through model.Time, store Store) error {
 		labelNames, err := store.LabelNamesForMetricName(ctx, userID, from, through, metricName)
 		if err != nil {
 			return err
 		}
-		result = append(result, labelNames...)
+		result.Add(labelNames...)
 		return nil
 	})
-	return result, err
+	return result.Strings(), err
 }
 
 func (c compositeStore) GetChunkRefs(ctx context.Context, userID string, from, through model.Time, matchers ...*labels.Matcher) ([][]Chunk, []*Fetcher, error) {

pkg/chunk/composite_store_test.go

@@ -180,3 +180,64 @@ func TestCompositeStore(t *testing.T) {
 		})
 	}
 }
+
+type mockStoreLabel struct {
+	mockStore
+	values []string
+}
+
+func (m mockStoreLabel) LabelValuesForMetricName(ctx context.Context, userID string, from, through model.Time, metricName string, labelName string) ([]string, error) {
+	return m.values, nil
+}
+
+func (m mockStoreLabel) LabelNamesForMetricName(ctx context.Context, userID string, from, through model.Time, metricName string) ([]string, error) {
+	return m.values, nil
+}
+
+func TestCompositeStoreLabels(t *testing.T) {
+	t.Parallel()
+
+	cs := compositeStore{
+		stores: []compositeStoreEntry{
+			{model.TimeFromUnix(0), mockStore(1)},
+			{model.TimeFromUnix(20), mockStoreLabel{mockStore(1), []string{"b", "c", "e"}}},
+			{model.TimeFromUnix(40), mockStoreLabel{mockStore(1), []string{"a", "b", "c", "f"}}},
+		},
+	}
+
+	for i, tc := range []struct {
+		from, through int64
+		want          []string
+	}{
+		{
+			0, 10,
+			nil,
+		},
+		{
+			0, 30,
+			[]string{"b", "c", "e"},
+		},
+		{
+			0, 40,
+			[]string{"a", "b", "c", "e", "f"},
+		},
+	} {
+		t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
+			have, err := cs.LabelNamesForMetricName(context.Background(), "", model.TimeFromUnix(tc.from), model.TimeFromUnix(tc.through), "")
+			if err != nil {
+				t.Fatalf("err - %s", err)
+			}
+			if !reflect.DeepEqual(tc.want, have) {
+				t.Fatalf("wrong label names - %s", test.Diff(tc.want, have))
+			}
+			have, err = cs.LabelValuesForMetricName(context.Background(), "", model.TimeFromUnix(tc.from), model.TimeFromUnix(tc.through), "", "")
+			if err != nil {
+				t.Fatalf("err - %s", err)
+			}
+			if !reflect.DeepEqual(tc.want, have) {
+				t.Fatalf("wrong label values - %s", test.Diff(tc.want, have))
+			}
+		})
+	}
+
+}

pkg/chunk/series_store.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"fmt"
 	"net/http"
-	"sort"
 
 	"github.com/go-kit/kit/log/level"
 	jsoniter "github.com/json-iterator/go"
@@ -406,23 +405,18 @@ func (c *seriesStore) lookupLabelNamesBySeries(ctx context.Context, from, throug
 		return nil, err
 	}
 	level.Debug(log).Log("entries", len(entries))
-	result := []string{model.MetricNameLabel}
-	uniqueLabelNames := map[string]struct{}{model.MetricNameLabel: {}}
+
+	var result UniqueStrings
+	result.Add(model.MetricNameLabel)
 	for _, entry := range entries {
 		lbs := []string{}
 		err := jsoniter.ConfigFastest.Unmarshal(entry.Value, &lbs)
 		if err != nil {
 			return nil, err
 		}
-		for _, l := range lbs {
-			if _, ok := uniqueLabelNames[l]; !ok {
-				uniqueLabelNames[l] = struct{}{}
-				result = append(result, l)
-			}
-		}
+		result.Add(lbs...)
 	}
-	sort.Strings(result)
-	return result, nil
+	return result.Strings(), nil
 }
 
 // Put implements ChunkStore

pkg/chunk/strings.go

@@ -1,5 +1,7 @@
 package chunk
 
+import "sort"
+
 func uniqueStrings(cs []string) []string {
 	if len(cs) == 0 {
 		return []string{}
@@ -57,3 +59,29 @@ func nWayIntersectStrings(sets [][]string) []string {
 		return intersectStrings(left, right)
 	}
 }
+
+// UniqueStrings keeps a slice of unique strings.
+type UniqueStrings struct {
+	values map[string]struct{}
+	result []string
+}
+
+// Add adds a new string, dropping duplicates.
+func (us *UniqueStrings) Add(strings ...string) {
+	for _, s := range strings {
+		if _, ok := us.values[s]; ok {
+			continue
+		}
+		if us.values == nil {
+			us.values = map[string]struct{}{}
+		}
+		us.values[s] = struct{}{}
+		us.result = append(us.result, s)
+	}
+}
+
+// Strings returns the sorted sliced of unique strings.
+func (us UniqueStrings) Strings() []string {
+	sort.Strings(us.result)
+	return us.result
+}