Add native histogram max sample size limit validation (#6834)

* Add sample size limit validation for native histograms samples

Signed-off-by: Paurush Garg <paurushg@amazon.com>

* Fixing Rebase

Signed-off-by: Paurush Garg <paurushg@amazon.com>

---------

Signed-off-by: Paurush Garg <paurushg@amazon.com>

Author: PaurushGarg

CHANGELOG.md

@@ -47,9 +47,10 @@
 * [ENHANCEMENT] Compactor: Optimize cleaner run time. #6815
 * [ENHANCEMENT] Parquet Storage: Allow percentage based dynamic shard size for Parquet Converter. #6817
 * [ENHANCEMENT] Query Frontend: Enhance the performance of the JSON codec. #6816
+* [ENHANCEMENT] Compactor: Emit partition metrics separate from cleaner job. #6827
 * [ENHANCEMENT] Metadata Cache: Support inmemory and multi level cache backend. #6829
 * [ENHANCEMENT] Store Gateway: Allow to ignore syncing blocks older than certain time using `ignore_blocks_before`. #6830
-* [ENHANCEMENT] Compactor: Emit partition metrics separate from cleaner job. #6827
+* [ENHANCEMENT] Distributor: Add native histograms max sample size bytes limit validation. #6834
 * [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517
 * [BUGFIX] Ingester: Fix labelset data race condition. #6573
 * [BUGFIX] Compactor: Cleaner should not put deletion marker for blocks with no-compact marker. #6576

docs/configuration/config-file-reference.md

@@ -3542,6 +3542,10 @@ The `limits_config` configures default and per-tenant limits imposed by Cortex s
 # CLI flag: -validation.max-labels-size-bytes
 [max_labels_size_bytes: <int> | default = 0]
 
+# Maximum size in bytes of a native histogram sample. 0 to disable the limit.
+# CLI flag: -validation.max-native-histogram-sample-size-bytes
+[max_native_histogram_sample_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/util/validation/errors.go

@@ -262,6 +262,26 @@ func (e *nativeHistogramSchemaInvalidError) Error() string {
 	return fmt.Sprintf("invalid native histogram schema %d for metric: %.200q. supported schema from %d to %d", e.receivedSchema, formatLabelSet(e.series), histogram.ExponentialSchemaMin, histogram.ExponentialSchemaMax)
 }
 
+// nativeHistogramSampleSizeBytesExceededError is a ValidationError implementation for samples with native histogram
+// exceeding the sample size bytes limit
+type nativeHistogramSampleSizeBytesExceededError struct {
+	nhSampleSizeBytes int
+	series            []cortexpb.LabelAdapter
+	limit             int
+}
+
+func newNativeHistogramSampleSizeBytesExceededError(series []cortexpb.LabelAdapter, nhSampleSizeBytes int, limit int) ValidationError {
+	return &nativeHistogramSampleSizeBytesExceededError{
+		nhSampleSizeBytes: nhSampleSizeBytes,
+		series:            series,
+		limit:             limit,
+	}
+}
+
+func (e *nativeHistogramSampleSizeBytesExceededError) Error() string {
+	return fmt.Sprintf("native histogram sample size bytes exceeded for metric (actual: %d, limit: %d) metric: %.200q", e.nhSampleSizeBytes, e.limit, formatLabelSet(e.series))
+}
+
 // formatLabelSet formats label adapters as a metric name with labels, while preserving
 // label order, and keeping duplicates. If there are multiple "__name__" labels, only
 // first one is used as metric name, other ones will be included as regular labels.

pkg/util/validation/limits.go

@@ -138,6 +138,7 @@ type Limits struct {
 	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"`
 	MaxLabelsSizeBytes                int                 `yaml:"max_labels_size_bytes" json:"max_labels_size_bytes"`
+	MaxNativeHistogramSampleSizeBytes int                 `yaml:"max_native_histogram_sample_size_bytes" json:"max_native_histogram_sample_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"`
@@ -257,6 +258,7 @@ func (l *Limits) RegisterFlags(f *flag.FlagSet) {
 	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.MaxLabelsSizeBytes, "validation.max-labels-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.MaxNativeHistogramSampleSizeBytes, "validation.max-native-histogram-sample-size-bytes", 0, "Maximum size in bytes of a native histogram sample. 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")

pkg/util/validation/validate.go

@@ -58,6 +58,7 @@ const (
 	// Native Histogram specific validation reasons
 	nativeHistogramBucketCountLimitExceeded = "native_histogram_buckets_exceeded"
 	nativeHistogramInvalidSchema            = "native_histogram_invalid_schema"
+	nativeHistogramSampleSizeBytesExceeded  = "native_histogram_sample_size_bytes_exceeded"
 
 	// RateLimited is one of the values for the reason to discard samples.
 	// Declared here to avoid duplication in ingester and distributor.
@@ -340,6 +341,12 @@ func ValidateMetadata(validateMetrics *ValidateMetrics, cfg *Limits, userID stri
 
 func ValidateNativeHistogram(validateMetrics *ValidateMetrics, limits *Limits, userID string, ls []cortexpb.LabelAdapter, histogramSample cortexpb.Histogram) (cortexpb.Histogram, error) {
 
+	// sample size validation for native histogram
+	if limits.MaxNativeHistogramSampleSizeBytes > 0 && histogramSample.Size() > limits.MaxNativeHistogramSampleSizeBytes {
+		validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramSampleSizeBytesExceeded, userID).Inc()
+		return cortexpb.Histogram{}, newNativeHistogramSampleSizeBytesExceededError(ls, histogramSample.Size(), limits.MaxNativeHistogramSampleSizeBytes)
+	}
+
 	// schema validation for native histogram
 	if histogramSample.Schema < histogram.ExponentialSchemaMin || histogramSample.Schema > histogram.ExponentialSchemaMax {
 		validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramInvalidSchema, userID).Inc()

pkg/util/validation/validate_test.go

@@ -349,15 +349,17 @@ func TestValidateNativeHistogram(t *testing.T) {
 	belowMinRangeSchemaHistogram.Schema = -5
 	exceedMaxRangeSchemaFloatHistogram := tsdbutil.GenerateTestFloatHistogram(0)
 	exceedMaxRangeSchemaFloatHistogram.Schema = 20
+	exceedMaxSampleSizeBytesLimitFloatHistogram := tsdbutil.GenerateTestFloatHistogram(100)
 
 	for _, tc := range []struct {
-		name                      string
-		bucketLimit               int
-		resolutionReduced         bool
-		histogram                 cortexpb.Histogram
-		expectedHistogram         cortexpb.Histogram
-		expectedErr               error
-		discardedSampleLabelValue string
+		name                                   string
+		bucketLimit                            int
+		resolutionReduced                      bool
+		histogram                              cortexpb.Histogram
+		expectedHistogram                      cortexpb.Histogram
+		expectedErr                            error
+		discardedSampleLabelValue              string
+		maxNativeHistogramSampleSizeBytesLimit int
 	}{
 		{
 			name:                      "no limit, histogram",
@@ -455,12 +457,20 @@ func TestValidateNativeHistogram(t *testing.T) {
 			expectedErr:               newNativeHistogramSchemaInvalidError(lbls, int(exceedMaxRangeSchemaFloatHistogram.Schema)),
 			discardedSampleLabelValue: nativeHistogramInvalidSchema,
 		},
+		{
+			name:                                   "exceed max sample size bytes limit",
+			histogram:                              cortexpb.FloatHistogramToHistogramProto(0, exceedMaxSampleSizeBytesLimitFloatHistogram.Copy()),
+			expectedErr:                            newNativeHistogramSampleSizeBytesExceededError(lbls, 126, 100),
+			discardedSampleLabelValue:              nativeHistogramSampleSizeBytesExceeded,
+			maxNativeHistogramSampleSizeBytesLimit: 100,
+		},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			reg := prometheus.NewRegistry()
 			validateMetrics := NewValidateMetrics(reg)
 			limits := new(Limits)
 			limits.MaxNativeHistogramBuckets = tc.bucketLimit
+			limits.MaxNativeHistogramSampleSizeBytes = tc.maxNativeHistogramSampleSizeBytesLimit
 			actualHistogram, actualErr := ValidateNativeHistogram(validateMetrics, limits, userID, lbls, tc.histogram)
 			if tc.expectedErr != nil {
 				require.Equal(t, tc.expectedErr, actualErr)