add label checks

pkg/costattribution/caimpl/managerImpl.go

@@ -94,7 +94,11 @@ func (m *ManagerImpl) UpdateAttributionTimestamp(user string, calb string, lbs l
 }
 
 // SetActiveSeries adjust the input attribution and sets the active series gauge for the given user and attribution
-func (m *ManagerImpl) SetActiveSeries(userID, attribution string, value float64) {
+func (m *ManagerImpl) SetActiveSeries(userID, calb, attribution string, value float64) {
+	// if the input label is outdated, we skip the update
+	if calb != m.GetUserAttributionLabel(userID) {
+		return
+	}
 	attribution = m.adjustUserAttribution(userID, attribution)
 
 	m.attributionTracker.mu.Lock()
@@ -105,7 +109,12 @@ func (m *ManagerImpl) SetActiveSeries(userID, attribution string, value float64)
 }
 
 // IncrementDiscardedSamples increments the discarded samples counter for a given user and attribution
-func (m *ManagerImpl) IncrementDiscardedSamples(userID, attribution string, value float64) {
+func (m *ManagerImpl) IncrementDiscardedSamples(userID, calb, attribution string, value float64) {
+	// if the input label is outdated, we skip the update
+	if calb != m.GetUserAttributionLabel(userID) {
+		return
+	}
+
 	attribution = m.adjustUserAttribution(userID, attribution)
 	m.attributionTracker.mu.RLock()
 	defer m.attributionTracker.mu.RUnlock()
@@ -115,7 +124,11 @@ func (m *ManagerImpl) IncrementDiscardedSamples(userID, attribution string, valu
 }
 
 // IncrementReceivedSamples increments the received samples counter for a given user and attribution
-func (m *ManagerImpl) IncrementReceivedSamples(userID, attribution string, value float64) {
+func (m *ManagerImpl) IncrementReceivedSamples(userID, calb, attribution string, value float64) {
+	// if the input label is outdated, we skip the update
+	if calb != m.GetUserAttributionLabel(userID) {
+		return
+	}
 	attribution = m.adjustUserAttribution(userID, attribution)
 	m.attributionTracker.mu.RLock()
 	defer m.attributionTracker.mu.RUnlock()

pkg/costattribution/caimpl/managerImpl_test.go

@@ -118,7 +118,7 @@ func Test_SetActiveSeries(t *testing.T) {
 		lbls.Set("team", "foo")
 		isOutdated, val := manager.UpdateAttributionTimestamp(userID, "team", lbls.Labels(), time.Unix(0, 0))
 		assert.False(t, isOutdated)
-		manager.SetActiveSeries(userID, val, 1.0)
+		manager.SetActiveSeries(userID, "team", val, 1.0)
 		expectedMetrics := `
 		# HELP cortex_ingester_active_series_attribution The total number of active series per user and attribution.
 		# TYPE cortex_ingester_active_series_attribution gauge
@@ -135,12 +135,12 @@ func Test_SetActiveSeries(t *testing.T) {
 		lbls.Set("department", "bar")
 		isOutdated, val := manager.UpdateAttributionTimestamp(userID, "department", lbls.Labels(), time.Unix(0, 0))
 		assert.False(t, isOutdated)
-		manager.SetActiveSeries(userID, val, 2.0)
+		manager.SetActiveSeries(userID, "department", val, 2.0)
 
 		lbls.Set("department", "baz")
-		isOutdated, val = manager.UpdateAttributionTimestamp(userID, "team", lbls.Labels(), time.Unix(0, 0))
-		assert.True(t, isOutdated)
-		manager.SetActiveSeries(userID, val, 3.0)
+		isOutdated, val = manager.UpdateAttributionTimestamp(userID, "department", lbls.Labels(), time.Unix(1, 0))
+		assert.False(t, isOutdated)
+		manager.SetActiveSeries(userID, "department", val, 3.0)
 
 		expectedMetrics := `
 		# HELP cortex_ingester_active_series_attribution The total number of active series per user and attribution.
@@ -170,7 +170,7 @@ func Test_SetActiveSeries(t *testing.T) {
 		manager.attributionTracker.limits = overrides
 		isOutdated, val := manager.UpdateAttributionTimestamp(userID, "department", lbls.Labels(), time.Unix(5, 0))
 		assert.False(t, isOutdated)
-		manager.SetActiveSeries(userID, val, 3.0)
+		manager.SetActiveSeries(userID, val, "department", 3.0)
 
 		expectedMetrics := `
 		# HELP cortex_ingester_active_series_attribution The total number of active series per user and attribution.
@@ -188,7 +188,7 @@ func Test_SetActiveSeries(t *testing.T) {
 		lbls.Set("department", "bar")
 		isOutdated, val := manager.UpdateAttributionTimestamp(userID, "department", lbls.Labels(), time.Unix(0, 0))
 		assert.False(t, isOutdated)
-		manager.SetActiveSeries(userID, val, 4.0)
+		manager.SetActiveSeries(userID, val, "department", 4.0)
 
 		expectedMetrics := `
 		# HELP cortex_ingester_active_series_attribution The total number of active series per user and attribution.

pkg/costattribution/caimpl/tracker.go

@@ -36,14 +36,17 @@ func newTracker(trackedLabel string, limit int) (*Tracker, error) {
 		attributionLimit:      limit,
 		attributionTimestamps: map[string]*atomic.Int64{},
 		coolDownDeadline:      atomic.NewInt64(0),
+		//nolint:faillint // the metrics are registered in the mimir package
 		discardedSampleAttribution: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Name: "cortex_discarded_samples_attribution_total",
 			Help: "The total number of samples that were discarded per attribution.",
 		}, []string{"user", trackedLabel}),
+		//nolint:faillint
 		receivedSamplesAttribution: prometheus.NewCounterVec(prometheus.CounterOpts{
 			Name: "cortex_received_samples_attribution_total",
 			Help: "The total number of samples that were received per attribution.",
 		}, []string{"user", trackedLabel}),
+		//nolint:faillint
 		activeSeriesPerUserAttribution: prometheus.NewGaugeVec(prometheus.GaugeOpts{
 			Name: "cortex_ingester_active_series_attribution",
 			Help: "The total number of active series per user and attribution.",

pkg/costattribution/caimpl/tracker_test.go

@@ -48,8 +48,4 @@ func Test_NewTracker(t *testing.T) {
 
 	// Clean the tracker for the user attribution
 	tracker.cleanupTrackerAttribution(userID, attribution)
-
-	// Verify that metrics have been cleaned
-	expectedMetrics = ``
-	assert.NoError(t, testutil.GatherAndCompare(reg, strings.NewReader(expectedMetrics), metricNames...))
 }

pkg/costattribution/manager.go

@@ -15,9 +15,9 @@ type Manager interface {
 	GetUserAttributionLabel(userID string) string
 	GetUserAttributionLimit(userID string) int
 	UpdateAttributionTimestamp(user string, calb string, lbs labels.Labels, now time.Time) (bool, string)
-	SetActiveSeries(userID, attribution string, value float64)
-	IncrementDiscardedSamples(userID, attribution string, value float64)
-	IncrementReceivedSamples(userID, attribution string, value float64)
+	SetActiveSeries(userID, calb string, attribution string, value float64)
+	IncrementDiscardedSamples(userID, calb string, attribution string, value float64)
+	IncrementReceivedSamples(userID, calb string, attribution string, value float64)
 
 	Collect(out chan<- prometheus.Metric)
 	Describe(chan<- *prometheus.Desc)

pkg/distributor/distributor.go

@@ -1692,8 +1692,8 @@ func (d *Distributor) updateReceivedMetrics(req *mimirpb.WriteRequest, userID st
 	}
 	receivedMetadata = len(req.Metadata)
 	if caEnabled {
-		for lv, count := range costAttribution {
-			d.costAttributionMng.IncrementReceivedSamples(userID, lv, float64(count))
+		for value, count := range costAttribution {
+			d.costAttributionMng.IncrementReceivedSamples(userID, caLabel, value, float64(count))
 		}
 	}
 	d.receivedSamples.WithLabelValues(userID).Add(float64(receivedSamples))

pkg/ingester/activeseries/active_series.go

@@ -231,26 +231,20 @@ func (c *ActiveSeries) ActiveWithMatchers() (total int, totalMatching []int, tot
 	return
 }
 
-func (c *ActiveSeries) ActiveByAttributionValue() map[string]uint32 {
+func (c *ActiveSeries) ActiveByAttributionValue(calb string) map[string]uint32 {
 	total := make(map[string]uint32, c.costAttributionMng.GetUserAttributionLimit(c.userID))
 	for s := 0; s < numStripes; s++ {
 		c.stripes[s].mu.RLock()
-		for k, v := range c.stripes[s].costAttributionValues {
-			total[k] += v
+		if c.stripes[s].caLabel == calb {
+			for k, v := range c.stripes[s].costAttributionValues {
+				total[k] += v
+			}
 		}
 		c.stripes[s].mu.RUnlock()
 	}
 	return total
 }
 
-func (c *ActiveSeries) ResetAttribution() {
-	for s := 0; s < numStripes; s++ {
-		c.stripes[s].mu.Lock()
-		c.stripes[s].costAttributionValues = map[string]uint32{}
-		c.stripes[s].mu.Unlock()
-	}
-}
-
 func (c *ActiveSeries) Delete(ref chunks.HeadSeriesRef) {
 	stripeID := storage.SeriesRef(ref) % numStripes
 	c.stripes[stripeID].remove(storage.SeriesRef(ref))

pkg/ingester/ingester.go

@@ -331,10 +331,6 @@ type Ingester struct {
 	usersMetadataMtx sync.RWMutex
 	usersMetadata    map[string]*userMetricsMetadata
 
-	// For storing tenant current cost attribution labels.
-	costAttributionMtx sync.RWMutex
-	costAttributionlbs map[string]string
-
 	// Rate of pushed samples. Used to limit global samples push rate.
 	ingestionRate             *util_math.EwmaRate
 	inflightPushRequests      atomic.Int64
@@ -794,14 +790,15 @@ func (i *Ingester) updateActiveSeries(now time.Time) {
 			i.metrics.activeSeriesLoading.DeleteLabelValues(userID)
 			if allActive > 0 {
 				if i.isCostAttributionEnabledForUser(userID) {
-					labelAttributions := userDB.activeSeries.ActiveByAttributionValue()
-					for label, count := range labelAttributions {
-						i.costAttributionMng.SetActiveSeries(userID, label, float64(count))
+					calb := i.costAttributionMng.GetUserAttributionLabel(userID)
+					labelAttributions := userDB.activeSeries.ActiveByAttributionValue(calb)
+					for value, count := range labelAttributions {
+						i.costAttributionMng.SetActiveSeries(userID, calb, value, float64(count))
 					}
 				}
 				i.metrics.activeSeriesPerUser.WithLabelValues(userID).Set(float64(allActive))
 			} else {
-				i.metrics.activeSeriesPerUser.DeletePartialMatch(prometheus.Labels{"user": userID})
+				i.metrics.activeSeriesPerUser.DeleteLabelValues(userID)
 			}
 			if allActiveHistograms > 0 {
 				i.metrics.activeSeriesPerUserNativeHistograms.WithLabelValues(userID).Set(float64(allActiveHistograms))
@@ -1288,8 +1285,8 @@ func (i *Ingester) updateMetricsFromPushStats(userID string, group string, stats
 		}
 	}
 	if i.isCostAttributionEnabledForUser(userID) {
-		for label, count := range stats.failedSamplesAttribution {
-			i.costAttributionMng.IncrementDiscardedSamples(userID, label, float64(count))
+		for value, count := range stats.failedSamplesAttribution {
+			i.costAttributionMng.IncrementDiscardedSamples(userID, stats.attributionLabel, value, float64(count))
 		}
 	}
 }