Added per labelset support

Signed-off-by: Essam Eldaly <[email protected]>

Author: eeldaly

pkg/ingester/ingester.go

@@ -1251,35 +1251,35 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 			}
 		}
 
-		handleAppendFailure = func(err error, timestampMs int64, lbls []cortexpb.LabelAdapter, copiedLabels labels.Labels, matchedLabelSetLimits []validation.LimitsPerLabelSet, seriesHash uint64) (rollback bool) {
+		handleAppendFailure = func(err error, timestampMs int64, lbls []cortexpb.LabelAdapter, copiedLabels labels.Labels, matchedLabelSetLimits []validation.LimitsPerLabelSet) (rollback bool) {
 			// Check if the error is a soft error we can proceed on. If so, we keep track
 			// of it, so that we can return it back to the distributor, which will return a
 			// 400 error to the client. The client (Prometheus) will not retry on 400, and
 			// we actually ingested all samples which haven't failed.
 			switch cause := errors.Cause(err); {
 			case errors.Is(cause, storage.ErrOutOfBounds):
 				sampleOutOfBoundsCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("sample_out_of_bounds", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(sampleOutOfBounds, userID, &copiedLabels)
 				updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(timestampMs), lbls) })
 
 			case errors.Is(cause, storage.ErrOutOfOrderSample):
 				sampleOutOfOrderCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("sample_out_of_order", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(sampleOutOfOrder, userID, &copiedLabels)
 				updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(timestampMs), lbls) })
 
 			case errors.Is(cause, storage.ErrDuplicateSampleForTimestamp):
 				newValueForTimestampCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("new_value_for_timestamp", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(newValueForTimestamp, userID, &copiedLabels)
 				updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(timestampMs), lbls) })
 
 			case errors.Is(cause, storage.ErrTooOldSample):
 				sampleTooOldCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("sample_too_old", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(sampleTooOld, userID, &copiedLabels)
 				updateFirstPartial(func() error { return wrappedTSDBIngestErr(err, model.Time(timestampMs), lbls) })
 
 			case errors.Is(cause, errMaxSeriesPerUserLimitExceeded):
 				perUserSeriesLimitCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("per_user_series_limit", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(perUserSeriesLimit, userID, &copiedLabels)
 				updateFirstPartial(func() error {
 					return makeLimitError(perUserSeriesLimit, i.limiter.FormatError(userID, cause, copiedLabels))
 				})
@@ -1292,14 +1292,14 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 
 			case errors.Is(cause, errMaxSeriesPerMetricLimitExceeded):
 				perMetricSeriesLimitCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("per_metric_series_limit", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(perMetricSeriesLimit, userID, &copiedLabels)
 				updateFirstPartial(func() error {
 					return makeMetricLimitError(perMetricSeriesLimit, copiedLabels, i.limiter.FormatError(userID, cause, copiedLabels))
 				})
 
 			case errors.As(cause, &errMaxSeriesPerLabelSetLimitExceeded{}):
 				perLabelSetSeriesLimitCount++
-				i.validateMetrics.DiscardedSeriesTracker.Track("per_labelset_series_limit", userID, seriesHash)
+				i.validateMetrics.DiscardedSeriesTracker.Track(perLabelsetSeriesLimit, userID, &copiedLabels)
 				// We only track per labelset discarded samples for throttling by labelset limit.
 				reasonCounter.increment(matchedLabelSetLimits, perLabelsetSeriesLimit)
 				updateFirstPartial(func() error {
@@ -1381,7 +1381,7 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 
 			failedSamplesCount++
 
-			if rollback := handleAppendFailure(err, s.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits, tsLabelsHash); !rollback {
+			if rollback := handleAppendFailure(err, s.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits); !rollback {
 				continue
 			}
 			// The error looks an issue on our side, so we should rollback
@@ -1426,7 +1426,7 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 
 				failedHistogramsCount++
 
-				if rollback := handleAppendFailure(err, hp.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits, tsLabelsHash); !rollback {
+				if rollback := handleAppendFailure(err, hp.TimestampMs, ts.Labels, copiedLabels, matchedLabelSetLimits); !rollback {
 					continue
 				}
 				// The error looks an issue on our side, so we should rollback

pkg/util/discardedseries/tracker.go

@@ -5,15 +5,22 @@ import (
 	"time"
 
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/model/labels"
 )
 
 const (
 	vendMetricsInterval = 30 * time.Second
 )
 
+type labelCounterStruct struct {
+	*sync.RWMutex
+	*labels.Labels
+	inCurrentCycle bool
+}
+
 type seriesCounterStruct struct {
 	*sync.RWMutex
-	seriesCountMap map[uint64]struct{}
+	seriesCountMap map[uint64]*labelCounterStruct
 }
 
 type userCounterStruct struct {
@@ -23,32 +30,33 @@ type userCounterStruct struct {
 
 type DiscardedSeriesTracker struct {
 	*sync.RWMutex
-	labelUserMap         map[string]*userCounterStruct
+	reasonUserMap        map[string]*userCounterStruct
 	discardedSeriesGauge *prometheus.GaugeVec
 }
 
 func NewDiscardedSeriesTracker(discardedSeriesGauge *prometheus.GaugeVec) *DiscardedSeriesTracker {
 	tracker := &DiscardedSeriesTracker{
 		RWMutex:              &sync.RWMutex{},
-		labelUserMap:         make(map[string]*userCounterStruct),
+		reasonUserMap:        make(map[string]*userCounterStruct),
 		discardedSeriesGauge: discardedSeriesGauge,
 	}
 	return tracker
 }
 
-func (t *DiscardedSeriesTracker) Track(reason string, user string, series uint64) {
+func (t *DiscardedSeriesTracker) Track(reason string, user string, labels *labels.Labels) {
+	series := labels.Hash()
 	t.RLock()
-	userCounter, ok := t.labelUserMap[reason]
+	userCounter, ok := t.reasonUserMap[reason]
 	t.RUnlock()
 	if !ok {
 		t.Lock()
-		userCounter, ok = t.labelUserMap[reason]
+		userCounter, ok = t.reasonUserMap[reason]
 		if !ok {
 			userCounter = &userCounterStruct{
 				RWMutex:       &sync.RWMutex{},
 				userSeriesMap: make(map[string]*seriesCounterStruct),
 			}
-			t.labelUserMap[reason] = userCounter
+			t.reasonUserMap[reason] = userCounter
 		}
 		t.Unlock()
 	}
@@ -62,42 +70,66 @@ func (t *DiscardedSeriesTracker) Track(reason string, user string, series uint64
 		if !ok {
 			seriesCounter = &seriesCounterStruct{
 				RWMutex:        &sync.RWMutex{},
-				seriesCountMap: make(map[uint64]struct{}),
+				seriesCountMap: make(map[uint64]*labelCounterStruct),
 			}
 			userCounter.userSeriesMap[user] = seriesCounter
 		}
 		userCounter.Unlock()
 	}
 
 	seriesCounter.RLock()
-	_, ok = seriesCounter.seriesCountMap[series]
+	labelCounter, ok := seriesCounter.seriesCountMap[series]
 	seriesCounter.RUnlock()
 	if !ok {
 		seriesCounter.Lock()
-		seriesCounter.seriesCountMap[series] = struct{}{}
+		labelCounter, ok = seriesCounter.seriesCountMap[series]
+		if !ok {
+			labelCounter = &labelCounterStruct{
+				Labels:         labels,
+				RWMutex:        &sync.RWMutex{},
+				inCurrentCycle: true,
+			}
+			seriesCounter.seriesCountMap[series] = labelCounter
+		}
 		seriesCounter.Unlock()
 	}
+
+	labelCounter.Lock()
+	labelCounter.inCurrentCycle = true
+	labelCounter.Unlock()
 }
 
 func (t *DiscardedSeriesTracker) UpdateMetrics() {
 	usersToDelete := make([]string, 0)
 	t.RLock()
-	for label, userCounter := range t.labelUserMap {
+	for reason, userCounter := range t.reasonUserMap {
 		userCounter.RLock()
 		for user, seriesCounter := range userCounter.userSeriesMap {
 			seriesCounter.Lock()
-			seriesCount := len(seriesCounter.seriesCountMap)
-			t.discardedSeriesGauge.WithLabelValues(label, user).Set(float64(seriesCount))
-			clear(seriesCounter.seriesCountMap)
-			if seriesCount == 0 {
+			for hash, labelCounter := range seriesCounter.seriesCountMap {
+				labelCounter.Lock()
+				val := 0.0
+				if labelCounter.inCurrentCycle {
+					val = 1.0
+				}
+				t.discardedSeriesGauge.WithLabelValues(reason, user, labelCounter.Labels.String()).Set(val)
+				if !labelCounter.inCurrentCycle {
+					delete(seriesCounter.seriesCountMap, hash)
+				} else {
+					labelCounter.inCurrentCycle = false
+				}
+				labelCounter.Unlock()
+			}
+			if len(seriesCounter.seriesCountMap) == 0 {
 				usersToDelete = append(usersToDelete, user)
 			}
 			seriesCounter.Unlock()
 		}
 		userCounter.RUnlock()
 		userCounter.Lock()
 		for _, user := range usersToDelete {
-			if len(userCounter.userSeriesMap[user].seriesCountMap) == 0 {
+			_, ok := userCounter.userSeriesMap[user]
+			if ok && userCounter.userSeriesMap[user].seriesCountMap != nil {
 				delete(userCounter.userSeriesMap, user)
 			}
 		}
@@ -116,11 +148,17 @@ func (t *DiscardedSeriesTracker) StartDiscardedSeriesGoroutine() {
 }
 
 // only used in testing
-func (t *DiscardedSeriesTracker) getSeriesCount(label string, user string) int {
-	if userCounter, ok := t.labelUserMap[label]; ok {
+func (t *DiscardedSeriesTracker) getSeriesCount(reason string, user string) int {
+	count := -1
+	if userCounter, ok := t.reasonUserMap[reason]; ok {
 		if seriesCounter, ok := userCounter.userSeriesMap[user]; ok {
-			return len(seriesCounter.seriesCountMap)
+			count = 0
+			for _, label := range seriesCounter.seriesCountMap {
+				if label.inCurrentCycle {
+					count++
+				}
+			}
 		}
 	}
-	return -1
+	return count
 }