Add new limit: max_series_label_size_bytes

Signed-off-by: 🌲 Harry 🌊 John 🏔 <[email protected]>

Author: harry671003

CHANGELOG.md

@@ -49,6 +49,7 @@
 * [FEATURE] Compactor: Added configurations for Azure MSI in blocks-storage, ruler-storage and alertmanager-storage. #4818
 * [FEATURE] Ruler: Add support to pass custom implementations of queryable and pusher. #4782
 * [FEATURE] Create OpenTelemetry Bridge for Tracing. Now cortex can send traces to multiple destinations using OTEL Collectors. #4834
+* [FEATURE] Distributor: Added a new limit `-validation.max-series-label-size-bytes` allowing to limit the combined size of labels for each timeseries. #4848
 * [BUGFIX] Memberlist: Add join with no retrying when starting service. #4804
 * [BUGFIX] Ruler: Fix /ruler/rule_groups returns YAML with extra fields. #4767
 

docs/configuration/config-file-reference.md

@@ -2553,6 +2553,11 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s
 # CLI flag: -validation.max-label-names-per-series
 [max_label_names_per_series: <int> | default = 30]
 
+# Maximum combined size in bytes of all labels and label values accepted for a
+# series. 0 to disable the limit.
+# CLI flag: -validation.max-series-label-size-bytes
+[max_series_label_size_bytes: <int> | default = 0]
+
 # Maximum length accepted for metric metadata. Metadata refers to Metric Name,
 # HELP and UNIT.
 # CLI flag: -validation.max-metadata-length

pkg/distributor/distributor_test.go

@@ -1680,6 +1680,34 @@ func BenchmarkDistributor_Push(b *testing.B) {
 			},
 			expectedErr: "label value too long",
 		},
+		"max label size bytes per series limit reached": {
+			prepareConfig: func(limits *validation.Limits) {
+				limits.MaxSeriesLabelSizeBytes = 1024
+			},
+			prepareSeries: func() ([]labels.Labels, []cortexpb.Sample) {
+				metrics := make([]labels.Labels, numSeriesPerRequest)
+				samples := make([]cortexpb.Sample, numSeriesPerRequest)
+
+				for i := 0; i < numSeriesPerRequest; i++ {
+					lbls := labels.NewBuilder(labels.Labels{{Name: model.MetricNameLabel, Value: "foo"}})
+					for i := 0; i < 10; i++ {
+						lbls.Set(fmt.Sprintf("name_%d", i), fmt.Sprintf("value_%d", i))
+					}
+
+					// Add a label with a very long value.
+					lbls.Set("xxx", fmt.Sprintf("xxx_%0.2000d", 1))
+
+					metrics[i] = lbls.Labels()
+					samples[i] = cortexpb.Sample{
+						Value:       float64(i),
+						TimestampMs: time.Now().UnixNano() / int64(time.Millisecond),
+					}
+				}
+
+				return metrics, samples
+			},
+			expectedErr: "combined label length too long",
+		},
 		"timestamp too old": {
 			prepareConfig: func(limits *validation.Limits) {
 				limits.RejectOldSamples = true
@@ -1770,7 +1798,7 @@ func BenchmarkDistributor_Push(b *testing.B) {
 			limits := validation.Limits{}
 			flagext.DefaultValues(&distributorCfg, &clientConfig, &limits)
 
-			limits.IngestionRate = 0 // Unlimited.
+			limits.IngestionRate = 10000000 // Unlimited.
 			testData.prepareConfig(&limits)
 
 			distributorCfg.ShardByAllLabels = true

pkg/util/validation/errors.go

@@ -68,6 +68,26 @@ func newLabelValueTooLongError(series []cortexpb.LabelAdapter, labelValue string
 	}
 }
 
+// seriesLabelSizeBytesExceededError is a customized ValidationError, in that the cause and the series are
+// formatted in different order in Error.
+type seriesLabelSizeBytesExceededError struct {
+	seriesLabelSizeBytes int
+	series               []cortexpb.LabelAdapter
+	limit                int
+}
+
+func (e *seriesLabelSizeBytesExceededError) Error() string {
+	return fmt.Sprintf("series label size bytes exceeded for metric (actual: %d, limit: %d) metric: %.200q", e.seriesLabelSizeBytes, e.limit, formatLabelSet(e.series))
+}
+
+func newSeriesLabelSizeBytesExceededError(series []cortexpb.LabelAdapter, seriesLabelSizeBytes int, limit int) ValidationError {
+	return &seriesLabelSizeBytesExceededError{
+		seriesLabelSizeBytes: seriesLabelSizeBytes,
+		series:               series,
+		limit:                limit,
+	}
+}
+
 func newInvalidLabelError(series []cortexpb.LabelAdapter, labelName string) ValidationError {
 	return &genericValidationError{
 		message: "sample invalid label: %.200q metric %.200q",

pkg/util/validation/limits.go

@@ -46,6 +46,7 @@ type Limits struct {
 	MaxLabelNameLength        int                 `yaml:"max_label_name_length" json:"max_label_name_length"`
 	MaxLabelValueLength       int                 `yaml:"max_label_value_length" json:"max_label_value_length"`
 	MaxLabelNamesPerSeries    int                 `yaml:"max_label_names_per_series" json:"max_label_names_per_series"`
+	MaxSeriesLabelSizeBytes   int                 `yaml:"max_series_label_size_bytes" json:"max_series_label_size_bytes"`
 	MaxMetadataLength         int                 `yaml:"max_metadata_length" json:"max_metadata_length"`
 	RejectOldSamples          bool                `yaml:"reject_old_samples" json:"reject_old_samples"`
 	RejectOldSamplesMaxAge    model.Duration      `yaml:"reject_old_samples_max_age" json:"reject_old_samples_max_age"`
@@ -126,6 +127,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) {
 	f.IntVar(&l.MaxLabelNameLength, "validation.max-length-label-name", 1024, "Maximum length accepted for label names")
 	f.IntVar(&l.MaxLabelValueLength, "validation.max-length-label-value", 2048, "Maximum length accepted for label value. This setting also applies to the metric name")
 	f.IntVar(&l.MaxLabelNamesPerSeries, "validation.max-label-names-per-series", 30, "Maximum number of label names per series.")
+	f.IntVar(&l.MaxSeriesLabelSizeBytes, "validation.max-series-label-size-bytes", 0, "Maximum combined size in bytes of all labels and label values accepted for a series. 0 to disable the limit.")
 	f.IntVar(&l.MaxMetadataLength, "validation.max-metadata-length", 1024, "Maximum length accepted for metric metadata. Metadata refers to Metric Name, HELP and UNIT.")
 	f.BoolVar(&l.RejectOldSamples, "validation.reject-old-samples", false, "Reject old samples.")
 	_ = l.RejectOldSamplesMaxAge.Set("14d")
@@ -327,6 +329,11 @@ func (o *Overrides) MaxLabelNamesPerSeries(userID string) int {
 	return o.getOverridesForUser(userID).MaxLabelNamesPerSeries
 }
 
+// MaxSeriesLabelSizeBytes returns maximum number of label/value pairs timeseries.
+func (o *Overrides) MaxSeriesLabelSizeBytes(userID string) int {
+	return o.getOverridesForUser(userID).MaxSeriesLabelSizeBytes
+}
+
 // MaxMetadataLength returns maximum length metadata can be. Metadata refers
 // to the Metric Name, HELP and UNIT.
 func (o *Overrides) MaxMetadataLength(userID string) int {

pkg/util/validation/limits_test.go

@@ -92,21 +92,25 @@ func TestOverridesManager_GetOverrides(t *testing.T) {
 
 	require.Equal(t, 100, ov.MaxLabelNamesPerSeries("user1"))
 	require.Equal(t, 0, ov.MaxLabelValueLength("user1"))
+	require.Equal(t, 0, ov.MaxSeriesLabelSizeBytes("user1"))
 
 	// Update limits for tenant user1. We only update single field, the rest is copied from defaults.
 	// (That is how limits work when loaded from YAML)
 	l := defaults
 	l.MaxLabelValueLength = 150
+	l.MaxSeriesLabelSizeBytes = 10
 
 	tenantLimits["user1"] = &l
 
 	// Checking whether overrides were enforced
 	require.Equal(t, 100, ov.MaxLabelNamesPerSeries("user1"))
 	require.Equal(t, 150, ov.MaxLabelValueLength("user1"))
+	require.Equal(t, 10, ov.MaxSeriesLabelSizeBytes("user1"))
 
 	// Verifying user2 limits are not impacted by overrides
 	require.Equal(t, 100, ov.MaxLabelNamesPerSeries("user2"))
 	require.Equal(t, 0, ov.MaxLabelValueLength("user2"))
+	require.Equal(t, 0, ov.MaxSeriesLabelSizeBytes("user2"))
 }
 
 func TestLimitsLoadingFromYaml(t *testing.T) {

pkg/util/validation/validate.go

@@ -3,6 +3,7 @@ package validation
 import (
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 	"unicode/utf8"
 
@@ -34,16 +35,17 @@ const (
 	// ErrQueryTooLong is used in chunk store, querier and query frontend.
 	ErrQueryTooLong = "the query time range exceeds the limit (query length: %s, limit: %s)"
 
-	missingMetricName       = "missing_metric_name"
-	invalidMetricName       = "metric_name_invalid"
-	greaterThanMaxSampleAge = "greater_than_max_sample_age"
-	maxLabelNamesPerSeries  = "max_label_names_per_series"
-	tooFarInFuture          = "too_far_in_future"
-	invalidLabel            = "label_invalid"
-	labelNameTooLong        = "label_name_too_long"
-	duplicateLabelNames     = "duplicate_label_names"
-	labelsNotSorted         = "labels_not_sorted"
-	labelValueTooLong       = "label_value_too_long"
+	missingMetricName            = "missing_metric_name"
+	invalidMetricName            = "metric_name_invalid"
+	greaterThanMaxSampleAge      = "greater_than_max_sample_age"
+	maxLabelNamesPerSeries       = "max_label_names_per_series"
+	tooFarInFuture               = "too_far_in_future"
+	invalidLabel                 = "label_invalid"
+	labelNameTooLong             = "label_name_too_long"
+	duplicateLabelNames          = "duplicate_label_names"
+	labelsNotSorted              = "labels_not_sorted"
+	labelValueTooLong            = "label_value_too_long"
+	seriesLabelSizeBytesExceeded = "series_label_size_bytes_exceeded"
 
 	// Exemplar-specific validation reasons
 	exemplarLabelsMissing    = "exemplar_labels_missing"
@@ -76,6 +78,11 @@ var DiscardedSamples = prometheus.NewCounterVec(
 	[]string{discardReasonLabel, "user"},
 )
 
+var MaxSeriesLabelSizeBytesPerUser = prometheus.NewGaugeVec(prometheus.GaugeOpts{
+	Name: "cortex_distributor_max_series_label_size_bytes",
+	Help: "Maximum combined size in bytes of all labels of a series per user.",
+}, []string{"user"})
+
 // DiscardedExemplars is a metric of the number of discarded exemplars, by reason.
 var DiscardedExemplars = prometheus.NewCounterVec(
 	prometheus.CounterOpts{
@@ -94,6 +101,9 @@ var DiscardedMetadata = prometheus.NewCounterVec(
 	[]string{discardReasonLabel, "user"},
 )
 
+var maxSeriesLabelSizeBytesMap = make(map[string]int)
+var maxSeriesLabelSizeBytesMutex = &sync.RWMutex{}
+
 func init() {
 	prometheus.MustRegister(DiscardedSamples)
 	prometheus.MustRegister(DiscardedExemplars)
@@ -168,6 +178,7 @@ type LabelValidationConfig interface {
 	MaxLabelNamesPerSeries(userID string) int
 	MaxLabelNameLength(userID string) int
 	MaxLabelValueLength(userID string) int
+	MaxSeriesLabelSizeBytes(userID string) int
 }
 
 // ValidateLabels returns an err if the labels are invalid.
@@ -195,6 +206,9 @@ func ValidateLabels(cfg LabelValidationConfig, userID string, ls []cortexpb.Labe
 	maxLabelNameLength := cfg.MaxLabelNameLength(userID)
 	maxLabelValueLength := cfg.MaxLabelValueLength(userID)
 	lastLabelName := ""
+	maxSeriesLabelSizeBytes := cfg.MaxSeriesLabelSizeBytes(userID)
+	seriesLabelSizeBytes := 0
+
 	for _, l := range ls {
 		if !skipLabelNameValidation && !model.LabelName(l.Name).IsValid() {
 			DiscardedSamples.WithLabelValues(invalidLabel, userID).Inc()
@@ -216,6 +230,23 @@ func ValidateLabels(cfg LabelValidationConfig, userID string, ls []cortexpb.Labe
 		}
 
 		lastLabelName = l.Name
+		seriesLabelSizeBytes += l.Size()
+	}
+
+	if maxSeriesLabelSizeBytes > 0 && seriesLabelSizeBytes > maxSeriesLabelSizeBytes {
+		DiscardedSamples.WithLabelValues(seriesLabelSizeBytesExceeded, userID).Inc()
+		return newSeriesLabelSizeBytesExceededError(ls, seriesLabelSizeBytes, maxSeriesLabelSizeBytes)
+	}
+
+	maxSeriesLabelSizeBytesMutex.Lock()
+	defer maxSeriesLabelSizeBytesMutex.Unlock()
+	if prev, ok := maxSeriesLabelSizeBytesMap[userID]; ok {
+		if prev > maxSeriesLabelSizeBytes {
+			maxSeriesLabelSizeBytesMap[userID] = maxSeriesLabelSizeBytes
+			MaxSeriesLabelSizeBytesPerUser.WithLabelValues(userID).Set(float64(maxSeriesLabelSizeBytes))
+		}
+	} else {
+		maxSeriesLabelSizeBytesMap[userID] = maxSeriesLabelSizeBytes
 	}
 	return nil
 }
@@ -271,4 +302,8 @@ func DeletePerUserValidationMetrics(userID string, log log.Logger) {
 	if err := util.DeleteMatchingLabels(DiscardedMetadata, filter); err != nil {
 		level.Warn(log).Log("msg", "failed to remove cortex_discarded_metadata_total metric for user", "user", userID, "err", err)
 	}
+	if err := util.DeleteMatchingLabels(MaxSeriesLabelSizeBytesPerUser, filter); err != nil {
+		level.Warn(log).Log("msg", "failed to remove cortex_distributor_max_series_label_size_bytes metric for user", "user", userID, "err", err)
+	}
+	delete(maxSeriesLabelSizeBytesMap, userID)
 }

pkg/util/validation/validate_test.go

@@ -17,10 +17,11 @@ import (
 )
 
 type validateLabelsCfg struct {
-	enforceMetricName      bool
-	maxLabelNamesPerSeries int
-	maxLabelNameLength     int
-	maxLabelValueLength    int
+	enforceMetricName       bool
+	maxLabelNamesPerSeries  int
+	maxLabelNameLength      int
+	maxLabelValueLength     int
+	maxSeriesLabelSizeBytes int
 }
 
 func (v validateLabelsCfg) EnforceMetricName(userID string) bool {
@@ -39,6 +40,10 @@ func (v validateLabelsCfg) MaxLabelValueLength(userID string) int {
 	return v.maxLabelValueLength
 }
 
+func (v validateLabelsCfg) MaxSeriesLabelSizeBytes(userID string) int {
+	return v.maxSeriesLabelSizeBytes
+}
+
 type validateMetadataCfg struct {
 	enforceMetadataMetricName bool
 	maxMetadataLength         int
@@ -59,6 +64,7 @@ func TestValidateLabels(t *testing.T) {
 	cfg.maxLabelValueLength = 25
 	cfg.maxLabelNameLength = 25
 	cfg.maxLabelNamesPerSeries = 2
+	cfg.maxSeriesLabelSizeBytes = 90
 	cfg.enforceMetricName = true
 
 	for _, c := range []struct {
@@ -114,6 +120,14 @@ func TestValidateLabels(t *testing.T) {
 				{Name: "blip", Value: "blop"},
 			}, 2),
 		},
+		{
+			map[model.LabelName]model.LabelValue{model.MetricNameLabel: "exactly_twenty_five_chars", "exactly_twenty_five_chars": "exactly_twenty_five_chars"},
+			false,
+			newSeriesLabelSizeBytesExceededError([]cortexpb.LabelAdapter{
+				{Name: model.MetricNameLabel, Value: "exactly_twenty_five_chars"},
+				{Name: "exactly_twenty_five_chars", Value: "exactly_twenty_five_chars"},
+			}, 91, cfg.maxSeriesLabelSizeBytes),
+		},
 		{
 			map[model.LabelName]model.LabelValue{model.MetricNameLabel: "foo", "invalid%label&name": "bar"},
 			true,
@@ -135,6 +149,7 @@ func TestValidateLabels(t *testing.T) {
 			cortex_discarded_samples_total{reason="max_label_names_per_series",user="testUser"} 1
 			cortex_discarded_samples_total{reason="metric_name_invalid",user="testUser"} 1
 			cortex_discarded_samples_total{reason="missing_metric_name",user="testUser"} 1
+			cortex_discarded_samples_total{reason="series_label_size_bytes_exceeded",user="testUser"} 1
 
 			cortex_discarded_samples_total{reason="random reason",user="different user"} 1
 	`), "cortex_discarded_samples_total"))