Reduce histogram resolution instead of rejecting

CHANGELOG.md

@@ -69,6 +69,7 @@
 * [ENHANCEMENT] Server: Add `-ingester.client.report-grpc-codes-in-instrumentation-label-enabled` CLI flag to specify whether gRPC status codes should be used in `status_code` label of `cortex_ingester_client_request_duration_seconds` metric. It defaults to false, meaning that successful and erroneous gRPC status codes are represented with `2xx` and `error` respectively. #6562
 * [ENHANCEMENT] Server: Add `-server.http-log-closed-connections-without-response-enabled` option to log details about connections to HTTP server that were closed before any data was sent back. This can happen if client doesn't manage to send complete HTTP headers before timeout. #6612
 * [ENHANCEMENT] Query-frontend: include length of query, time since the earliest and latest points of a query in "query stats" logs. Time parameters (start/end/time) are always formatted as RFC3339 now. #6473
+* [ENHANCEMENT] Distributor: added support for reducing the resolution of native histogram samples upon ingestion if the sample has too many buckets compared to `-validation.max-native-histogram-buckets`. This is enabled by default and can be turned off by setting `-validation.reduce-native-histogram-over-max-buckets` to `false`. #6535
 * [BUGFIX] Distributor: return server overload error in the event of exceeding the ingestion rate limit. #6549
 * [BUGFIX] Ring: Ensure network addresses used for component hash rings are formatted correctly when using IPv6. #6068
 * [BUGFIX] Query-scheduler: don't retain connections from queriers that have shut down, leading to gradually increasing enqueue latency over time. #6100 #6145

cmd/mimir/config-descriptor.json

@@ -3172,6 +3172,16 @@
           "fieldFlag": "validation.max-native-histogram-buckets",
           "fieldType": "int"
         },
+        {
+          "kind": "field",
+          "name": "reduce_native_histogram_over_max_buckets",
+          "required": false,
+          "desc": "Whether to reduce or reject native histogram samples with more buckets than the configured limit.",
+          "fieldValue": null,
+          "fieldDefaultValue": true,
+          "fieldFlag": "validation.reduce-native-histogram-over-max-buckets",
+          "fieldType": "boolean"
+        },
         {
           "kind": "field",
           "name": "creation_grace_period",

cmd/mimir/help-all.txt.tmpl

@@ -2731,6 +2731,8 @@ Usage of ./cmd/mimir/mimir:
     	Maximum length accepted for metric metadata. Metadata refers to Metric Name, HELP and UNIT. Longer metadata is dropped except for HELP which is truncated. (default 1024)
   -validation.max-native-histogram-buckets int
     	Maximum number of buckets per native histogram sample. 0 to disable the limit.
+  -validation.reduce-native-histogram-over-max-buckets
+    	Whether to reduce or reject native histogram samples with more buckets than the configured limit. (default true)
   -validation.separate-metrics-group-label string
     	[experimental] Label used to define the group label for metrics separation. For each write request, the group is obtained from the first non-empty group label from the first timeseries in the incoming list of timeseries. Specific distributor and ingester metrics will be further separated adding a 'group' label with group label's value. Currently applies to the following metrics: cortex_discarded_samples_total
   -vault.auth.approle.mount-path string

cmd/mimir/help.txt.tmpl

@@ -723,6 +723,8 @@ Usage of ./cmd/mimir/mimir:
     	Maximum length accepted for metric metadata. Metadata refers to Metric Name, HELP and UNIT. Longer metadata is dropped except for HELP which is truncated. (default 1024)
   -validation.max-native-histogram-buckets int
     	Maximum number of buckets per native histogram sample. 0 to disable the limit.
+  -validation.reduce-native-histogram-over-max-buckets
+    	Whether to reduce or reject native histogram samples with more buckets than the configured limit. (default true)
   -version
     	Print application version and exit.
 

docs/sources/mimir/configure/configure-native-histograms-ingestion.md

@@ -21,15 +21,15 @@ To enable ingesting Prometheus native histograms over the [remote write API]({{<
 
 To limit the number of native histogram buckets per sample, set the `-validation.max-native-histogram-buckets` flag on distributors.
 The recommended value is 160 which is the default in the [OpenTelemetry SDK](https://opentelemetry.io/docs/specs/otel/metrics/sdk/) for exponential histograms, which are a similar concept in OpenTelemetry.
-Samples with more buckets than the limit will be dropped.
+At the time of ingestion, samples with more buckets than the limit will be scaled down, meaning that the resolution will be reduced and buckets will be merged until either the number of buckets is under the limit or the minimal resolution is reached. The behavior can be changed to dropping such samples by setting the ``-validation.reduce-native-histogram-over-max-buckets` option to `false`.
 
 ## Configure native histograms per tenant
 
 To enable ingesting Prometheus native histograms over the [remote write API]({{< relref "../references/http-api#remote-write" >}}) for a tenant, set the `native_histograms_ingestion_enabled` runtime value to `true`.
 
 To limit the number of native histogram buckets per sample for a tenant, set the `max_native_histogram_buckets` runtime value.
 The recommended value is 160 which is the default in the [OpenTelemetry SDK](https://opentelemetry.io/docs/specs/otel/metrics/sdk/) for exponential histograms, which are a similar concept in OpenTelemetry.
-Samples with more buckets than the limit will be dropped.
+At the time of ingestion, samples with more buckets than the limit will be scaled down, meaning that the resolution will be reduced and buckets will be merged until either the number of buckets is under the limit or the minimal resolution is reached. The behavior can be changed to dropping such samples by setting the ``-validation.reduce-native-histogram-over-max-buckets` option to `false`.
 
 ```yaml
 overrides:

docs/sources/mimir/manage/mimir-runbooks/_index.md

@@ -1258,6 +1258,13 @@ The limit protects the system from using too much memory. To configure the limit
 
 > **Note:** The series containing such samples are skipped during ingestion, and valid series within the same request are ingested.
 
+### err-mimir-not-reducible-native-histogram
+
+This non-critical error occurs when Mimir receives a write request that contains a sample that is a native histogram that has too many observation buckets and it is not possible to reduce the buckets further. Since native buckets at the lowest resolution of -4 can cover all 64 bit float observations with a handful of buckets, this indicates that the
+`-validation.max-native-histogram-buckets` option is set too low (<20).
+
+> **Note:** The series containing such samples are skipped during ingestion, and valid series within the same request are ingested.
+
 ### err-mimir-label-invalid
 
 This non-critical error occurs when Mimir receives a write request that contains a series with an invalid label name.

docs/sources/mimir/references/configuration-parameters/index.md

@@ -2891,6 +2891,11 @@ The `limits` block configures default and per-tenant limits imposed by component
 # CLI flag: -validation.max-native-histogram-buckets
 [max_native_histogram_buckets: <int> | default = 0]
 
+# Whether to reduce or reject native histogram samples with more buckets than
+# the configured limit.
+# CLI flag: -validation.reduce-native-histogram-over-max-buckets
+[reduce_native_histogram_over_max_buckets: <boolean> | default = true]
+
 # (advanced) Controls how far into the future incoming samples and exemplars are
 # accepted compared to the wall clock. Any sample or exemplar will be rejected
 # if its timestamp is greater than '(now + grace_period)'. This configuration is

pkg/distributor/distributor.go

@@ -629,13 +629,13 @@ func (d *Distributor) validateSeries(nowt time.Time, ts *mimirpb.PreallocTimeser
 		}
 	}
 
-	for _, h := range ts.Histograms {
+	for i, h := range ts.Histograms {
 		delta := now - model.Time(h.Timestamp)
 		if delta > 0 {
 			d.sampleDelayHistogram.Observe(float64(delta) / 1000)
 		}
 
-		if err := validateSampleHistogram(d.sampleValidationMetrics, now, d.limits, userID, group, ts.Labels, h); err != nil {
+		if err := validateSampleHistogram(d.sampleValidationMetrics, now, d.limits, userID, group, ts.Labels, &ts.Histograms[i]); err != nil {
 			return err
 		}
 	}

pkg/distributor/distributor_test.go

@@ -1400,6 +1400,7 @@ func TestDistributor_Push_HistogramValidation(t *testing.T) {
 			flagext.DefaultValues(limits)
 			limits.CreationGracePeriod = model.Duration(time.Minute)
 			limits.MaxNativeHistogramBuckets = tc.bucketLimit
+			limits.ReduceNativeHistogramOverMaxBuckets = false
 
 			ds, _, _ := prepare(t, prepConfig{
 				numIngesters:     2,
@@ -1615,6 +1616,118 @@ func TestDistributor_ExemplarValidation(t *testing.T) {
 	}
 }
 
+func TestDistributor_HistogramReduction(t *testing.T) {
+	h := &histogram.Histogram{
+		Count:         12,
+		ZeroCount:     2,
+		ZeroThreshold: 0.001,
+		Sum:           18.4,
+		Schema:        0,
+		PositiveSpans: []histogram.Span{
+			{Offset: 0, Length: 2},
+			{Offset: 1, Length: 2},
+		},
+		PositiveBuckets: []int64{1, 1, -1, 0}, // 1 in 0(0.5, 1], 2 in 1(1, 2], 1 in 3(4, 8], 1 in 4(8, 16]
+		NegativeSpans: []histogram.Span{
+			{Offset: 0, Length: 2},
+			{Offset: 1, Length: 2},
+		},
+		NegativeBuckets: []int64{1, 1, -1, 0}, // 1 in -4[-16, -8), 1 in -3[-8, -4), 2 in -1[-2, -1), 1 in -0[-1, -0.5)
+	}
+
+	reducedH := &histogram.Histogram{
+		Count:         12,
+		ZeroCount:     2,
+		ZeroThreshold: 0.001,
+		Sum:           18.4,
+		Schema:        -1,
+		PositiveSpans: []histogram.Span{
+			{Offset: 0, Length: 3},
+		},
+		PositiveBuckets: []int64{1, 1, 0}, // 1 in 0(0.25, 1], 2 in 1(1, 4], 2 in 2(4, 16]
+		NegativeSpans: []histogram.Span{
+			{Offset: 0, Length: 3},
+		},
+		NegativeBuckets: []int64{1, 1, 0},
+	}
+
+	hugeH := &histogram.Histogram{
+		Count:         12,
+		ZeroCount:     2,
+		ZeroThreshold: 0.001,
+		Sum:           18.4,
+		Schema:        -3,
+		PositiveSpans: []histogram.Span{
+			{Offset: -1e6, Length: 1},
+			{Offset: 2e6, Length: 1}, // Further then min schema of -4, bucket with of 64K
+		},
+		PositiveBuckets: []int64{1, 1},
+	}
+
+	tests := map[string]struct {
+		prepareConfig      func(limits *validation.Limits)
+		req                *mimirpb.WriteRequest
+		expectedError      error
+		expectedTimeSeries []mimirpb.PreallocTimeseries
+	}{
+		"should not reduce histogram under bucket limit": {
+			prepareConfig: func(limits *validation.Limits) {
+				limits.MaxNativeHistogramBuckets = 8
+			},
+			req: makeWriteRequestHistogram([]string{model.MetricNameLabel, "test"}, 1000, h),
+			expectedTimeSeries: []mimirpb.PreallocTimeseries{
+				makeHistogramTimeseries([]string{model.MetricNameLabel, "test"}, 1000, h),
+			},
+		},
+		"should reduce histogram over bucket limit": {
+			prepareConfig: func(limits *validation.Limits) {
+				limits.MaxNativeHistogramBuckets = 7
+			},
+			req: makeWriteRequestHistogram([]string{model.MetricNameLabel, "test"}, 1000, h),
+			expectedTimeSeries: []mimirpb.PreallocTimeseries{
+				makeHistogramTimeseries([]string{model.MetricNameLabel, "test"}, 1000, reducedH),
+			},
+		},
+		"should fail if not possible to reduce": {
+			prepareConfig: func(limits *validation.Limits) {
+				limits.MaxNativeHistogramBuckets = 1
+			},
+			req:                makeWriteRequestHistogram([]string{model.MetricNameLabel, "test"}, 1000, hugeH),
+			expectedError:      fmt.Errorf("received a native histogram sample with too many buckets and cannot reduce, timestamp: 1000 series: {__name__=\"test\"}, buckets: 2, limit: 1 (not-reducible-native-histogram)"),
+			expectedTimeSeries: []mimirpb.PreallocTimeseries{},
+		},
+	}
+	now := mtime.Now()
+	for testName, tc := range tests {
+		t.Run(testName, func(t *testing.T) {
+			limits := &validation.Limits{}
+			flagext.DefaultValues(limits)
+			tc.prepareConfig(limits)
+			limits.ReduceNativeHistogramOverMaxBuckets = true
+			ds, _, regs := prepare(t, prepConfig{
+				limits:          limits,
+				numDistributors: 1,
+			})
+
+			// Pre-condition check.
+			require.Len(t, ds, 1)
+			require.Len(t, regs, 1)
+
+			for _, ts := range tc.req.Timeseries {
+				err := ds[0].validateSeries(now, &ts, "user", "test-group", false, 0, 0)
+				if tc.expectedError != nil {
+					require.ErrorAs(t, err, &tc.expectedError)
+				} else {
+					assert.NoError(t, err)
+				}
+			}
+			if tc.expectedError == nil {
+				assert.Equal(t, tc.expectedTimeSeries, tc.req.Timeseries)
+			}
+		})
+	}
+}
+
 func mkLabels(n int, extra ...string) []mimirpb.LabelAdapter {
 	ret := make([]mimirpb.LabelAdapter, 1+n+len(extra)/2)
 	ret[0] = mimirpb.LabelAdapter{Name: model.MetricNameLabel, Value: "foo"}

pkg/distributor/validate.go

@@ -78,6 +78,10 @@ var (
 		"received a native histogram sample with too many buckets, timestamp: %%d series: %%s, buckets: %%d, limit: %%d (%s)",
 		globalerror.MaxNativeHistogramBuckets,
 	)
+	notReducibleNativeHistogramMsgFormat = fmt.Sprintf(
+		"received a native histogram sample with too many buckets and cannot reduce, timestamp: %%d series: %%s, buckets: %%d, limit: %%d (%s)",
+		globalerror.NotReducibleNativeHistogram,
+	)
 	sampleTimestampTooNewMsgFormat = globalerror.SampleTooFarInFuture.MessageWithPerTenantLimitConfig(
 		"received a sample whose timestamp is too far in the future, timestamp: %d series: '%.200s'",
 		validation.CreationGracePeriodFlag,
@@ -107,6 +111,7 @@ var (
 type sampleValidationConfig interface {
 	CreationGracePeriod(userID string) time.Duration
 	MaxNativeHistogramBuckets(userID string) int
+	ReduceNativeHistogramOverMaxBuckets(userID string) bool
 }
 
 // sampleValidationMetrics is a collection of metrics used during sample validation.
@@ -207,7 +212,7 @@ func validateSample(m *sampleValidationMetrics, now model.Time, cfg sampleValida
 // validateSampleHistogram returns an err if the sample is invalid.
 // The returned error may retain the provided series labels.
 // It uses the passed 'now' time to measure the relative time of the sample.
-func validateSampleHistogram(m *sampleValidationMetrics, now model.Time, cfg sampleValidationConfig, userID, group string, ls []mimirpb.LabelAdapter, s mimirpb.Histogram) error {
+func validateSampleHistogram(m *sampleValidationMetrics, now model.Time, cfg sampleValidationConfig, userID, group string, ls []mimirpb.LabelAdapter, s *mimirpb.Histogram) error {
 	if model.Time(s.Timestamp) > now.Add(cfg.CreationGracePeriod(userID)) {
 		m.tooFarInFuture.WithLabelValues(userID, group).Inc()
 		unsafeMetricName, _ := extract.UnsafeMetricNameFromLabelAdapters(ls)
@@ -222,8 +227,20 @@ func validateSampleHistogram(m *sampleValidationMetrics, now model.Time, cfg sam
 			bucketCount = len(s.GetNegativeDeltas()) + len(s.GetPositiveDeltas())
 		}
 		if bucketCount > bucketLimit {
-			m.maxNativeHistogramBuckets.WithLabelValues(userID, group).Inc()
-			return fmt.Errorf(maxNativeHistogramBucketsMsgFormat, s.Timestamp, mimirpb.FromLabelAdaptersToLabels(ls).String(), bucketCount, bucketLimit)
+			if !cfg.ReduceNativeHistogramOverMaxBuckets(userID) {
+				m.maxNativeHistogramBuckets.WithLabelValues(userID, group).Inc()
+				return fmt.Errorf(maxNativeHistogramBucketsMsgFormat, s.Timestamp, mimirpb.FromLabelAdaptersToLabels(ls).String(), bucketCount, bucketLimit)
+			}
+			for {
+				bc, err := s.ReduceResolution()
+				if err != nil {
+					m.maxNativeHistogramBuckets.WithLabelValues(userID, group).Inc()
+					return fmt.Errorf(notReducibleNativeHistogramMsgFormat, s.Timestamp, mimirpb.FromLabelAdaptersToLabels(ls).String(), bucketCount, bucketLimit)
+				}
+				if bc < bucketLimit {
+					break
+				}
+			}
 		}
 	}
 

pkg/distributor/validate_test.go

@@ -384,7 +384,8 @@ func TestValidateLabelDuplication(t *testing.T) {
 }
 
 type sampleValidationCfg struct {
-	maxNativeHistogramBuckets int
+	maxNativeHistogramBuckets           int
+	reduceNativeHistogramOverMaxBuckets bool
 }
 
 func (c sampleValidationCfg) CreationGracePeriod(_ string) time.Duration {
@@ -395,6 +396,10 @@ func (c sampleValidationCfg) MaxNativeHistogramBuckets(_ string) int {
 	return c.maxNativeHistogramBuckets
 }
 
+func (c sampleValidationCfg) ReduceNativeHistogramOverMaxBuckets(_ string) bool {
+	return c.reduceNativeHistogramOverMaxBuckets
+}
+
 func TestMaxNativeHistorgramBuckets(t *testing.T) {
 	// All will have 2 buckets, one negative and one positive
 	testCases := map[string]mimirpb.Histogram{
@@ -515,7 +520,7 @@ func TestMaxNativeHistorgramBuckets(t *testing.T) {
 
 				err := validateSampleHistogram(metrics, model.Now(), cfg, "user-1", "group-1", []mimirpb.LabelAdapter{
 					{Name: model.MetricNameLabel, Value: "a"},
-					{Name: "a", Value: "a"}}, h)
+					{Name: "a", Value: "a"}}, &h)
 
 				if limit == 1 {
 					require.Error(t, err)