update metrics synchronously

yeya24 · yeya24 · commit 0cdf9e4e87c4 · 2024-12-18T16:53:00.000-08:00
Signed-off-by: Ben Ye &lt;benye@amazon.com&gt;
diff --git a/pkg/distributor/distributor.go b/pkg/distributor/distributor.go
@@ -129,8 +129,8 @@ type Distributor struct {
 
 	asyncExecutor util.AsyncExecutor
 
-	// Counter to track metrics per label set.
-	labelSetCounter *labelSetCounter
+	// Map to track label sets from user.
+	labelSetTracker *labelSetTracker
 }
 
 // Config contains the configuration required to
@@ -296,7 +296,6 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 		ingestionRateLimiter:   limiter.NewRateLimiter(ingestionRateStrategy, 10*time.Second),
 		HATracker:              haTracker,
 		ingestionRate:          util_math.NewEWMARate(0.2, instanceIngestionRateTickInterval),
-		labelSetCounter:        newLabelSetCounter(),
 
 		queryDuration: instrument.NewHistogramCollector(promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
 			Namespace: "cortex",
@@ -389,6 +388,8 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove
 		asyncExecutor:   util.NewNoOpExecutor(),
 	}
 
+	d.labelSetTracker = newLabelSetTracker(d.receivedSamplesPerLabelSet)
+
 	if cfg.NumPushWorkers > 0 {
 		util_log.WarnExperimentalUse("Distributor: using goroutine worker pool")
 		d.asyncExecutor = util.NewWorkerPool("distributor", cfg.NumPushWorkers, reg)
@@ -809,7 +810,7 @@ func (d *Distributor) updateLabelSetMetrics() {
 		}
 	}
 
-	d.labelSetCounter.updateMetrics(activeUserSet, d.receivedSamplesPerLabelSet)
+	d.labelSetTracker.updateMetrics(activeUserSet)
 }
 
 func (d *Distributor) cleanStaleIngesterMetrics() {
@@ -1062,7 +1063,7 @@ func (d *Distributor) prepareSeriesKeys(ctx context.Context, req *cortexpb.Write
 		validatedExemplars += len(ts.Exemplars)
 	}
 	for h, counter := range labelSetCounters {
-		d.labelSetCounter.increaseSamplesLabelSet(userID, h, counter.labels, counter.floatSamples, counter.histogramSamples)
+		d.labelSetTracker.increaseSamplesLabelSet(userID, h, counter.labels, counter.floatSamples, counter.histogramSamples)
 	}
 	return seriesKeys, validatedTimeseries, validatedFloatSamples, validatedHistogramSamples, validatedExemplars, firstPartialErr, nil
 }
diff --git a/pkg/distributor/distributor_test.go b/pkg/distributor/distributor_test.go
@@ -4135,16 +4135,12 @@ func TestDistributor_PushLabelSetMetrics(t *testing.T) {
 	_, err = ds[0].Push(ctx, req)
 	require.NoError(t, err)
 
-	ds[0].updateLabelSetMetrics()
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# HELP cortex_distributor_received_samples_per_labelset_total The total number of received samples per label set, excluding rejected and deduped samples.
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"one\"}",type="float",user="user"} 2
-		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"one\"}",type="histogram",user="user"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"two\"}",type="float",user="user"} 1
-		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"two\"}",type="histogram",user="user"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="user"} 1
-		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="user"} 0
 		`), "cortex_distributor_received_samples_per_labelset_total"))
 
 	// Push more series.
@@ -4166,20 +4162,14 @@ func TestDistributor_PushLabelSetMetrics(t *testing.T) {
 	req = mockWriteRequest(inputSeries, 1, 1, false)
 	_, err = ds[0].Push(ctx2, req)
 	require.NoError(t, err)
-	ds[0].updateLabelSetMetrics()
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# HELP cortex_distributor_received_samples_per_labelset_total The total number of received samples per label set, excluding rejected and deduped samples.
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"one\"}",type="float",user="user"} 2
-		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"one\"}",type="histogram",user="user"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"two\"}",type="float",user="user"} 2
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"two\"}",type="float",user="user2"} 1
-		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"two\"}",type="histogram",user="user"} 0
-		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"two\"}",type="histogram",user="user2"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="user"} 2
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="user2"} 1
-		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="user"} 0
-		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="user2"} 0
 		`), "cortex_distributor_received_samples_per_labelset_total"))
 
 	// Remove existing limits and add new limits
@@ -4198,8 +4188,6 @@ func TestDistributor_PushLabelSetMetrics(t *testing.T) {
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="user"} 2
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="user2"} 1
-		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="user"} 0
-		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="user2"} 0
 		`), "cortex_distributor_received_samples_per_labelset_total"))
 
 	// Metrics from `user` got removed but `user2` metric should remain.
@@ -4208,7 +4196,6 @@ func TestDistributor_PushLabelSetMetrics(t *testing.T) {
 		# HELP cortex_distributor_received_samples_per_labelset_total The total number of received samples per label set, excluding rejected and deduped samples.
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="user2"} 1
-		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="user2"} 0
 		`), "cortex_distributor_received_samples_per_labelset_total"))
 }
 
diff --git a/pkg/distributor/metrics.go b/pkg/distributor/metrics.go
@@ -11,27 +11,29 @@ import (
 )
 
 const (
-	numMetricCounterShards = 128
+	numMetricShards = 128
 )
 
-type labelSetCounter struct {
+type labelSetTracker struct {
+	receivedSamplesPerLabelSet *prometheus.CounterVec
+
 	shards []*labelSetCounterShard
 }
 
-func newLabelSetCounter() *labelSetCounter {
-	shards := make([]*labelSetCounterShard, 0, numMetricCounterShards)
-	for i := 0; i < numMetricCounterShards; i++ {
+func newLabelSetTracker(receivedSamplesPerLabelSet *prometheus.CounterVec) *labelSetTracker {
+	shards := make([]*labelSetCounterShard, 0, numMetricShards)
+	for i := 0; i < numMetricShards; i++ {
 		shards = append(shards, &labelSetCounterShard{
 			RWMutex:       &sync.RWMutex{},
-			valuesCounter: map[string]map[uint64]*samplesLabelSetEntry{},
+			userLabelSets: map[string]map[uint64]labels.Labels{},
 		})
 	}
-	return &labelSetCounter{shards: shards}
+	return &labelSetTracker{shards: shards, receivedSamplesPerLabelSet: receivedSamplesPerLabelSet}
 }
 
 type labelSetCounterShard struct {
 	*sync.RWMutex
-	valuesCounter map[string]map[uint64]*samplesLabelSetEntry
+	userLabelSets map[string]map[uint64]labels.Labels
 }
 
 type samplesLabelSetEntry struct {
@@ -40,63 +42,51 @@ type samplesLabelSetEntry struct {
 	labels           labels.Labels
 }
 
-func (s *samplesLabelSetEntry) reset() {
-	s.floatSamples = 0
-	s.histogramSamples = 0
-}
-
-func (m *labelSetCounter) increaseSamplesLabelSet(userId string, hash uint64, labelSet labels.Labels, floatSamples, histogramSamples int64) {
-	s := m.shards[util.HashFP(model.Fingerprint(hash))%numMetricCounterShards]
+func (m *labelSetTracker) increaseSamplesLabelSet(userId string, hash uint64, labelSet labels.Labels, floatSamples, histogramSamples int64) {
+	s := m.shards[util.HashFP(model.Fingerprint(hash))%numMetricShards]
 	s.Lock()
-	defer s.Unlock()
-	if userEntry, ok := s.valuesCounter[userId]; ok {
-		if e, ok2 := userEntry[hash]; ok2 {
-			e.floatSamples += floatSamples
-			e.histogramSamples += histogramSamples
-		} else {
-			userEntry[hash] = &samplesLabelSetEntry{
-				floatSamples:     floatSamples,
-				histogramSamples: histogramSamples,
-				labels:           labelSet,
-			}
+	if userEntry, ok := s.userLabelSets[userId]; ok {
+		if _, ok2 := userEntry[hash]; !ok2 {
+			userEntry[hash] = labelSet
 		}
 	} else {
-		s.valuesCounter[userId] = map[uint64]*samplesLabelSetEntry{
-			hash: {
-				floatSamples:     floatSamples,
-				histogramSamples: histogramSamples,
-				labels:           labelSet,
-			},
-		}
+		s.userLabelSets[userId] = map[uint64]labels.Labels{hash: labelSet}
+	}
+	// Unlock before we update metrics.
+	s.Unlock()
+
+	labelSetStr := labelSet.String()
+	if floatSamples > 0 {
+		m.receivedSamplesPerLabelSet.WithLabelValues(userId, sampleMetricTypeFloat, labelSetStr).Add(float64(floatSamples))
+	}
+	if histogramSamples > 0 {
+		m.receivedSamplesPerLabelSet.WithLabelValues(userId, sampleMetricTypeHistogram, labelSetStr).Add(float64(histogramSamples))
 	}
 }
 
-func (m *labelSetCounter) updateMetrics(userSet map[string]map[uint64]struct{}, receivedSamplesPerLabelSet *prometheus.CounterVec) {
-	for i := 0; i < numMetricCounterShards; i++ {
+// Clean up dangling user and label set from the tracker as well as metrics.
+func (m *labelSetTracker) updateMetrics(userSet map[string]map[uint64]struct{}) {
+	for i := 0; i < numMetricShards; i++ {
 		shard := m.shards[i]
 		shard.Lock()
 
-		for user, userEntry := range shard.valuesCounter {
+		for user, userEntry := range shard.userLabelSets {
 			limits, ok := userSet[user]
 			if !ok {
 				// If user is removed, we will delete user metrics in cleanupInactiveUser loop
 				// so skip deleting metrics here.
-				delete(shard.valuesCounter, user)
+				delete(shard.userLabelSets, user)
 				continue
 			}
-			for h, entry := range userEntry {
-				labelSetStr := entry.labels.String()
+			for h, lbls := range userEntry {
 				// This limit no longer exists.
 				if _, ok := limits[h]; !ok {
 					delete(userEntry, h)
-					receivedSamplesPerLabelSet.DeleteLabelValues(user, sampleMetricTypeFloat, labelSetStr)
-					receivedSamplesPerLabelSet.DeleteLabelValues(user, sampleMetricTypeHistogram, labelSetStr)
+					labelSetStr := lbls.String()
+					m.receivedSamplesPerLabelSet.DeleteLabelValues(user, sampleMetricTypeFloat, labelSetStr)
+					m.receivedSamplesPerLabelSet.DeleteLabelValues(user, sampleMetricTypeHistogram, labelSetStr)
 					continue
 				}
-				receivedSamplesPerLabelSet.WithLabelValues(user, sampleMetricTypeFloat, labelSetStr).Add(float64(entry.floatSamples))
-				receivedSamplesPerLabelSet.WithLabelValues(user, sampleMetricTypeHistogram, labelSetStr).Add(float64(entry.histogramSamples))
-				// Reset entry counter to 0. Delete it only if it is removed from the limit.
-				entry.reset()
 			}
 		}
 
diff --git a/pkg/distributor/metrics_test.go b/pkg/distributor/metrics_test.go
@@ -11,14 +11,13 @@ import (
 )
 
 func TestLabelSetCounter(t *testing.T) {
-	counter := newLabelSetCounter()
-
 	metricName := "cortex_distributor_received_samples_per_labelset_total"
 	reg := prometheus.NewPedanticRegistry()
 	dummyCounter := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: metricName,
 		Help: "",
 	}, []string{"user", "type", "labelset"})
+	counter := newLabelSetTracker(dummyCounter)
 	reg.MustRegister(dummyCounter)
 
 	userID := "1"
@@ -31,22 +30,12 @@ func TestLabelSetCounter(t *testing.T) {
 	counter.increaseSamplesLabelSet(userID2, 0, labels.FromStrings("foo", "bar"), 100, 5)
 	counter.increaseSamplesLabelSet(userID2, 2, labels.FromStrings("cluster", "us-west-2"), 0, 100)
 
-	userSet := map[string]map[uint64]struct {
-	}{
-		userID:  {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}},
-		userID2: {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}},
-	}
-	counter.updateMetrics(userSet, dummyCounter)
-
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
-		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-west-2\"}",type="float",user="2"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-west-2\"}",type="histogram",user="2"} 100
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="1"} 10
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="2"} 100
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="2"} 5
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"baz\"}",type="float",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"baz\"}",type="histogram",user="1"} 5
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="1"} 20
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="1"} 20
@@ -57,13 +46,6 @@ func TestLabelSetCounter(t *testing.T) {
 	counter.increaseSamplesLabelSet(userID2, 2, labels.FromStrings("cluster", "us-west-2"), 0, 100)
 	counter.increaseSamplesLabelSet(userID2, 4, labels.FromStrings("cluster", "us-west-2"), 10, 10)
 	counter.increaseSamplesLabelSet(userID3, 4, labels.FromStrings("cluster", "us-east-1"), 30, 30)
-	userSet = map[string]map[uint64]struct {
-	}{
-		userID:  {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}},
-		userID2: {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}, 4: struct{}{}},
-		userID3: {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}, 4: struct{}{}},
-	}
-	counter.updateMetrics(userSet, dummyCounter)
 
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
@@ -73,18 +55,20 @@ func TestLabelSetCounter(t *testing.T) {
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-west-2\"}",type="histogram",user="2"} 210
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="1"} 10
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="2"} 100
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="2"} 5
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"baz\"}",type="float",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"baz\"}",type="histogram",user="1"} 5
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="1"} 40
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="1"} 40
 		`), metricName))
 
 	// Remove user 2. But metrics for user 2 not cleaned up as it is expected to be cleaned up
 	// in cleanupInactiveUser loop. It is expected to have 3 minutes delay in this case.
-	delete(userSet, userID2)
-	counter.updateMetrics(userSet, dummyCounter)
+	userSet := map[string]map[uint64]struct {
+	}{
+		userID:  {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}},
+		userID3: {0: struct{}{}, 1: struct{}{}, 2: struct{}{}, 3: struct{}{}, 4: struct{}{}},
+	}
+	counter.updateMetrics(userSet)
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-east-1\"}",type="float",user="3"} 30
@@ -93,9 +77,7 @@ func TestLabelSetCounter(t *testing.T) {
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-west-2\"}",type="histogram",user="2"} 210
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="1"} 10
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="2"} 100
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="2"} 5
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"baz\"}",type="float",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"baz\"}",type="histogram",user="1"} 5
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="float",user="1"} 40
 		cortex_distributor_received_samples_per_labelset_total{labelset="{}",type="histogram",user="1"} 40
@@ -108,15 +90,14 @@ func TestLabelSetCounter(t *testing.T) {
 		userID2: {},
 		userID3: {},
 	}
-	counter.updateMetrics(userSet, dummyCounter)
+	counter.updateMetrics(userSet)
 
 	require.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(`
 		# TYPE cortex_distributor_received_samples_per_labelset_total counter
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-west-2\"}",type="float",user="2"} 10
 		cortex_distributor_received_samples_per_labelset_total{labelset="{cluster=\"us-west-2\"}",type="histogram",user="2"} 210
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="1"} 10
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="float",user="2"} 100
-		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="1"} 0
 		cortex_distributor_received_samples_per_labelset_total{labelset="{foo=\"bar\"}",type="histogram",user="2"} 5
 		`), metricName))
 }

Original file line number	Diff line number	Diff line change
`@@ -129,8 +129,8 @@ type Distributor struct {`
`129`	`129`
`130`	`130`	`asyncExecutor util.AsyncExecutor`
`131`	`131`
`132`		`- // Counter to track metrics per label set.`
`133`		`- labelSetCounter *labelSetCounter`
	`132`	`+ // Map to track label sets from user.`
	`133`	`+ labelSetTracker *labelSetTracker`
`134`	`134`	`}`
`135`	`135`
`136`	`136`	`// Config contains the configuration required to`
`@@ -296,7 +296,6 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove`
`296`	`296`	`ingestionRateLimiter: limiter.NewRateLimiter(ingestionRateStrategy, 10*time.Second),`
`297`	`297`	`HATracker: haTracker,`
`298`	`298`	`ingestionRate: util_math.NewEWMARate(0.2, instanceIngestionRateTickInterval),`
`299`		`- labelSetCounter: newLabelSetCounter(),`
`300`	`299`
`301`	`300`	`queryDuration: instrument.NewHistogramCollector(promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{`
`302`	`301`	`Namespace: "cortex",`
`@@ -389,6 +388,8 @@ func New(cfg Config, clientConfig ingester_client.Config, limits *validation.Ove`
`389`	`388`	`asyncExecutor: util.NewNoOpExecutor(),`
`390`	`389`	`}`
`391`	`390`
	`391`	`+ d.labelSetTracker = newLabelSetTracker(d.receivedSamplesPerLabelSet)`
	`392`	`+`
`392`	`393`	`if cfg.NumPushWorkers > 0 {`
`393`	`394`	`util_log.WarnExperimentalUse("Distributor: using goroutine worker pool")`
`394`	`395`	`d.asyncExecutor = util.NewWorkerPool("distributor", cfg.NumPushWorkers, reg)`
`@@ -809,7 +810,7 @@ func (d *Distributor) updateLabelSetMetrics() {`
`809`	`810`	`}`
`810`	`811`	`}`
`811`	`812`
`812`		`- d.labelSetCounter.updateMetrics(activeUserSet, d.receivedSamplesPerLabelSet)`
	`813`	`+ d.labelSetTracker.updateMetrics(activeUserSet)`
`813`	`814`	`}`
`814`	`815`
`815`	`816`	`func (d *Distributor) cleanStaleIngesterMetrics() {`
`@@ -1062,7 +1063,7 @@ func (d Distributor) prepareSeriesKeys(ctx context.Context, req cortexpb.Write`
`1062`	`1063`	`validatedExemplars += len(ts.Exemplars)`
`1063`	`1064`	`}`
`1064`	`1065`	`for h, counter := range labelSetCounters {`
`1065`		`- d.labelSetCounter.increaseSamplesLabelSet(userID, h, counter.labels, counter.floatSamples, counter.histogramSamples)`
	`1066`	`+ d.labelSetTracker.increaseSamplesLabelSet(userID, h, counter.labels, counter.floatSamples, counter.histogramSamples)`
`1066`	`1067`	`}`
`1067`	`1068`	`return seriesKeys, validatedTimeseries, validatedFloatSamples, validatedHistogramSamples, validatedExemplars, firstPartialErr, nil`
`1068`	`1069`	`}`