Use prometheus NH validate

Signed-off-by: SungJin1212 <[email protected]>

Author: SungJin1212

CHANGELOG.md

@@ -4,7 +4,7 @@
 
 * [ENHANCEMENT] Ingester: Add `enable_matcher_optimization` config to apply low selectivity matchers lazily. #7063
 * [ENHANCEMENT] Distributor: Add a label references validation for remote write v2 request. #7074
-* [ENHANCEMENT] Distributor: Add count validations for native histogram. #7072
+* [ENHANCEMENT] Distributor: Add count, spans, and buckets validations for native histogram. #7072
 * [BUGFIX] Compactor: Avoid race condition which allow a grouper to not compact all partitions. #7082
 
 ## 1.20.0 in progress

pkg/util/validation/errors.go

@@ -282,60 +282,20 @@ 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))
 }
 
-// nativeHistogramMisMatchCountError is a ValidationError implementation for native histogram
-// where the Count does not match the sum of observations in the buckets.
-type nativeHistogramMisMatchCountError struct {
-	series       []cortexpb.LabelAdapter
-	observations uint64
-	count        uint64
-}
-
-func newNativeHistogramMisMatchedCountError(series []cortexpb.LabelAdapter, observations, count uint64) ValidationError {
-	return &nativeHistogramMisMatchCountError{
-		series:       series,
-		observations: observations,
-		count:        count,
-	}
-}
-
-func (e *nativeHistogramMisMatchCountError) Error() string {
-	return fmt.Sprintf("native histogram bucket count mismatch: count is %d, but observations found in buckets is %d, metric: %.200q", e.count, e.observations, formatLabelSet(e.series))
-}
-
-// nativeHistogramNegativeCountError is a ValidationError implementation for float native histogram
-// where the Count field is negative.
-type nativeHistogramNegativeCountError struct {
-	series []cortexpb.LabelAdapter
-	count  float64
-}
-
-func newNativeHistogramNegativeCountError(series []cortexpb.LabelAdapter, count float64) ValidationError {
-	return &nativeHistogramNegativeCountError{
-		series: series,
-		count:  count,
-	}
-}
-
-func (e *nativeHistogramNegativeCountError) Error() string {
-	return fmt.Sprintf("native histogram observation count %.2f is negative, metric: %.200q", e.count, formatLabelSet(e.series))
-}
-
-// nativeHistogramNegativeBucketCountError is a ValidationError implementation for float native histogram
-// where the count in buckets is negative.
-type nativeHistogramNegativeBucketCountError struct {
-	series []cortexpb.LabelAdapter
-	count  float64
+type nativeHistogramInvalidError struct {
+	nhValidationErr error
+	series          []cortexpb.LabelAdapter
 }
 
-func newNativeHistogramNegativeBucketCountError(series []cortexpb.LabelAdapter, count float64) ValidationError {
-	return &nativeHistogramNegativeBucketCountError{
-		series: series,
-		count:  count,
+func newNativeHistogramInvalidError(series []cortexpb.LabelAdapter, nhValidationErr error) ValidationError {
+	return &nativeHistogramInvalidError{
+		series:          series,
+		nhValidationErr: nhValidationErr,
 	}
 }
 
-func (e *nativeHistogramNegativeBucketCountError) Error() string {
-	return fmt.Sprintf("native histogram buckets have a negative count: %.2f, metric: %.200q", e.count, formatLabelSet(e.series))
+func (e *nativeHistogramInvalidError) Error() string {
+	return fmt.Sprintf("invalid native histogram, validation err: %v, metric: %.200q", e.nhValidationErr, formatLabelSet(e.series))
 }
 
 // formatLabelSet formats label adapters as a metric name with labels, while preserving

pkg/util/validation/validate.go

@@ -3,7 +3,6 @@ package validation
 import (
 	"errors"
 	"fmt"
-	"math"
 	"net/http"
 	"strings"
 	"time"
@@ -62,9 +61,7 @@ const (
 	nativeHistogramBucketCountLimitExceeded = "native_histogram_buckets_exceeded"
 	nativeHistogramInvalidSchema            = "native_histogram_invalid_schema"
 	nativeHistogramSampleSizeBytesExceeded  = "native_histogram_sample_size_bytes_exceeded"
-	nativeHistogramNegativeCount            = "native_histogram_negative_count"
-	nativeHistogramNegativeBucketCount      = "native_histogram_negative_bucket_count"
-	nativeHistogramMisMatchCount            = "native_histogram_mismatch_count"
+	nativeHistogramInvalid                  = "native_histogram_invalid"
 
 	// RateLimited is one of the values for the reason to discard samples.
 	// Declared here to avoid duplication in ingester and distributor.
@@ -385,77 +382,33 @@ func ValidateNativeHistogram(validateMetrics *ValidateMetrics, limits *Limits, u
 		return cortexpb.Histogram{}, newNativeHistogramSchemaInvalidError(ls, int(histogramSample.Schema))
 	}
 
-	var nCount, pCount uint64
-	if histogramSample.IsFloatHistogram() {
-		if c, err := checkHistogramBuckets(histogramSample.GetNegativeCounts(), &nCount, false); err != nil {
-			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramNegativeBucketCount, userID).Inc()
-			return cortexpb.Histogram{}, newNativeHistogramNegativeBucketCountError(ls, c)
-		}
-
-		if c, err := checkHistogramBuckets(histogramSample.GetPositiveCounts(), &pCount, false); err != nil {
-			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramNegativeBucketCount, userID).Inc()
-			return cortexpb.Histogram{}, newNativeHistogramNegativeBucketCountError(ls, c)
-		}
-
-		if histogramSample.GetZeroCountFloat() < 0 {
-			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramNegativeBucketCount, userID).Inc()
-			return cortexpb.Histogram{}, newNativeHistogramNegativeBucketCountError(ls, histogramSample.GetZeroCountFloat())
-		}
-
-		if histogramSample.GetCountFloat() < 0 {
-			// validate if float histogram has negative count
-			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramNegativeCount, userID).Inc()
-			return cortexpb.Histogram{}, newNativeHistogramNegativeCountError(ls, histogramSample.GetCountFloat())
-		}
-	} else {
-		if c, err := checkHistogramBuckets(histogramSample.GetNegativeDeltas(), &nCount, true); err != nil {
-			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramNegativeBucketCount, userID).Inc()
-			return cortexpb.Histogram{}, newNativeHistogramNegativeBucketCountError(ls, c)
-		}
-
-		if c, err := checkHistogramBuckets(histogramSample.GetPositiveDeltas(), &pCount, true); err != nil {
-			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramNegativeBucketCount, userID).Inc()
-			return cortexpb.Histogram{}, newNativeHistogramNegativeBucketCountError(ls, c)
-		}
-
-		// validate if there is mismatch between count with observations in buckets
-		observations := nCount + pCount + histogramSample.GetZeroCountInt()
-		count := histogramSample.GetCountInt()
-
-		if math.IsNaN(histogramSample.Sum) {
-			if observations > count {
-				validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramMisMatchCount, userID).Inc()
-				return cortexpb.Histogram{}, newNativeHistogramMisMatchedCountError(ls, observations, count)
-			}
-		} else {
-			if observations != count {
-				validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramMisMatchCount, userID).Inc()
-				return cortexpb.Histogram{}, newNativeHistogramMisMatchedCountError(ls, observations, count)
-			}
-		}
-	}
-
-	if limits.MaxNativeHistogramBuckets == 0 {
-		return histogramSample, nil
-	}
-
 	var (
 		exceedLimit bool
 	)
 
 	if histogramSample.IsFloatHistogram() {
-		// Initial check to see if the bucket limit is exceeded or not. If not, we can avoid type casting.
+		fh := cortexpb.FloatHistogramProtoToFloatHistogram(histogramSample)
+		if err := fh.Validate(); err != nil {
+			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramInvalid, userID).Inc()
+			return cortexpb.Histogram{}, newNativeHistogramInvalidError(ls, err)
+		}
+
+		// limit check
+		if limits.MaxNativeHistogramBuckets == 0 {
+			return histogramSample, nil
+		}
+
 		exceedLimit = len(histogramSample.PositiveCounts)+len(histogramSample.NegativeCounts) > limits.MaxNativeHistogramBuckets
 		if !exceedLimit {
 			return histogramSample, nil
 		}
+
 		// Exceed limit.
 		if histogramSample.Schema <= histogram.ExponentialSchemaMin {
 			validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramBucketCountLimitExceeded, userID).Inc()
 			return cortexpb.Histogram{}, newHistogramBucketLimitExceededError(ls, limits.MaxNativeHistogramBuckets)
 		}
 
-		fh := cortexpb.FloatHistogramProtoToFloatHistogram(histogramSample)
 		oBuckets := len(fh.PositiveBuckets) + len(fh.NegativeBuckets)
 		for len(fh.PositiveBuckets)+len(fh.NegativeBuckets) > limits.MaxNativeHistogramBuckets {
 			if fh.Schema <= histogram.ExponentialSchemaMin {
@@ -471,7 +424,17 @@ func ValidateNativeHistogram(validateMetrics *ValidateMetrics, limits *Limits, u
 		return cortexpb.FloatHistogramToHistogramProto(histogramSample.TimestampMs, fh), nil
 	}
 
-	// Initial check to see if bucket limit is exceeded or not. If not, we can avoid type casting.
+	h := cortexpb.HistogramProtoToHistogram(histogramSample)
+	if err := h.Validate(); err != nil {
+		validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramInvalid, userID).Inc()
+		return cortexpb.Histogram{}, newNativeHistogramInvalidError(ls, err)
+	}
+
+	// limit check
+	if limits.MaxNativeHistogramBuckets == 0 {
+		return histogramSample, nil
+	}
+
 	exceedLimit = len(histogramSample.PositiveDeltas)+len(histogramSample.NegativeDeltas) > limits.MaxNativeHistogramBuckets
 	if !exceedLimit {
 		return histogramSample, nil
@@ -481,7 +444,6 @@ func ValidateNativeHistogram(validateMetrics *ValidateMetrics, limits *Limits, u
 		validateMetrics.DiscardedSamples.WithLabelValues(nativeHistogramBucketCountLimitExceeded, userID).Inc()
 		return cortexpb.Histogram{}, newHistogramBucketLimitExceededError(ls, limits.MaxNativeHistogramBuckets)
 	}
-	h := cortexpb.HistogramProtoToHistogram(histogramSample)
 	oBuckets := len(h.PositiveBuckets) + len(h.NegativeBuckets)
 	for len(h.PositiveBuckets)+len(h.NegativeBuckets) > limits.MaxNativeHistogramBuckets {
 		if h.Schema <= histogram.ExponentialSchemaMin {