Adding testing

Signed-off-by: Paurush Garg <[email protected]>

Author: PaurushGarg

pkg/ingester/ingester.go

@@ -1224,21 +1224,22 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 	// Keep track of some stats which are tracked only if the samples will be
 	// successfully committed
 	var (
-		succeededSamplesCount         = 0
-		failedSamplesCount            = 0
-		succeededHistogramsCount      = 0
-		failedHistogramsCount         = 0
-		succeededExemplarsCount       = 0
-		failedExemplarsCount          = 0
-		startAppend                   = time.Now()
-		sampleOutOfBoundsCount        = 0
-		sampleOutOfOrderCount         = 0
-		sampleTooOldCount             = 0
-		newValueForTimestampCount     = 0
-		perUserSeriesLimitCount       = 0
-		perLabelSetSeriesLimitCount   = 0
-		perMetricSeriesLimitCount     = 0
-		discardedNativeHistogramCount = 0
+		succeededSamplesCount                   = 0
+		failedSamplesCount                      = 0
+		succeededHistogramsCount                = 0
+		failedHistogramsCount                   = 0
+		succeededExemplarsCount                 = 0
+		failedExemplarsCount                    = 0
+		startAppend                             = time.Now()
+		sampleOutOfBoundsCount                  = 0
+		sampleOutOfOrderCount                   = 0
+		sampleTooOldCount                       = 0
+		newValueForTimestampCount               = 0
+		perUserSeriesLimitCount                 = 0
+		perUserNativeHistogramsSeriesLimitCount = 0
+		perLabelSetSeriesLimitCount             = 0
+		perMetricSeriesLimitCount               = 0
+		discardedNativeHistogramCount           = 0
 
 		updateFirstPartial = func(errFn func() error) {
 			if firstPartialErr == nil {
@@ -1274,6 +1275,12 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 					return makeLimitError(perUserSeriesLimit, i.limiter.FormatError(userID, cause, copiedLabels))
 				})
 
+			case errors.Is(cause, errMaxNativeHistogramsSeriesPerUserLimitExceeded):
+				perUserNativeHistogramsSeriesLimitCount++
+				updateFirstPartial(func() error {
+					return makeLimitError(perUserSeriesLimit, i.limiter.FormatError(userID, cause, copiedLabels))
+				})
+
 			case errors.Is(cause, errMaxSeriesPerMetricLimitExceeded):
 				perMetricSeriesLimitCount++
 				updateFirstPartial(func() error {
@@ -1517,6 +1524,9 @@ func (i *Ingester) Push(ctx context.Context, req *cortexpb.WriteRequest) (*corte
 	if perUserSeriesLimitCount > 0 {
 		i.validateMetrics.DiscardedSamples.WithLabelValues(perUserSeriesLimit, userID).Add(float64(perUserSeriesLimitCount))
 	}
+	if perUserNativeHistogramsSeriesLimitCount > 0 {
+		i.validateMetrics.DiscardedSamples.WithLabelValues(perUserNativeHistogramsSeriesLimit, userID).Add(float64(perUserNativeHistogramsSeriesLimitCount))
+	}
 	if perMetricSeriesLimitCount > 0 {
 		i.validateMetrics.DiscardedSamples.WithLabelValues(perMetricSeriesLimit, userID).Add(float64(perMetricSeriesLimitCount))
 	}

pkg/ingester/ingester_test.go

@@ -868,6 +868,93 @@ func TestIngesterUserLimitExceeded(t *testing.T) {
 
 }
 
+func TestIngesterUserLimitExceededForNativeHistograms(t *testing.T) {
+	limits := defaultLimitsTestConfig()
+	limits.EnableNativeHistograms = true
+	limits.MaxLocalNativeHistogramsSeriesPerUser = 1
+	limits.MaxLocalSeriesPerUser = 1
+	limits.MaxLocalMetricsWithMetadataPerUser = 1
+
+	userID := "1"
+	// Series
+	labels1 := labels.Labels{{Name: labels.MetricName, Value: "testmetric"}, {Name: "foo", Value: "bar"}}
+	labels3 := labels.Labels{{Name: labels.MetricName, Value: "testmetric"}, {Name: "foo", Value: "biz"}}
+	sampleNativeHistogram1 := cortexpb.HistogramToHistogramProto(0, tsdbutil.GenerateTestHistogram(1))
+	sampleNativeHistogram2 := cortexpb.HistogramToHistogramProto(1, tsdbutil.GenerateTestHistogram(2))
+	sampleNativeHistogram3 := cortexpb.HistogramToHistogramProto(0, tsdbutil.GenerateTestHistogram(3))
+
+	// Metadata
+	metadata1 := &cortexpb.MetricMetadata{MetricFamilyName: "testmetric", Help: "a help for testmetric", Type: cortexpb.COUNTER}
+	metadata2 := &cortexpb.MetricMetadata{MetricFamilyName: "testmetric2", Help: "a help for testmetric2", Type: cortexpb.COUNTER}
+
+	dir := t.TempDir()
+
+	chunksDir := filepath.Join(dir, "chunks")
+	blocksDir := filepath.Join(dir, "blocks")
+	require.NoError(t, os.Mkdir(chunksDir, os.ModePerm))
+	require.NoError(t, os.Mkdir(blocksDir, os.ModePerm))
+
+	blocksIngesterGenerator := func(reg prometheus.Registerer) *Ingester {
+		ing, err := prepareIngesterWithBlocksStorageAndLimits(t, defaultIngesterTestConfig(t), limits, nil, blocksDir, reg)
+		require.NoError(t, err)
+		require.NoError(t, services.StartAndAwaitRunning(context.Background(), ing))
+		// Wait until it's ACTIVE
+		test.Poll(t, time.Second, ring.ACTIVE, func() interface{} {
+			return ing.lifecycler.GetState()
+		})
+
+		return ing
+	}
+
+	tests := []string{"blocks"}
+	for i, ingGenerator := range []func(reg prometheus.Registerer) *Ingester{blocksIngesterGenerator} {
+		t.Run(tests[i], func(t *testing.T) {
+			reg := prometheus.NewRegistry()
+			ing := ingGenerator(reg)
+
+			// Append only one series and one metadata first, expect no error.
+			ctx := user.InjectOrgID(context.Background(), userID)
+			_, err := ing.Push(ctx, cortexpb.ToWriteRequest([]labels.Labels{labels1}, nil, []*cortexpb.MetricMetadata{metadata1}, []cortexpb.Histogram{sampleNativeHistogram1}, cortexpb.API))
+			require.NoError(t, err)
+
+			testLimits := func(reg prometheus.Gatherer) {
+				// Append to two series, expect series-exceeded error.
+				_, err = ing.Push(ctx, cortexpb.ToWriteRequest([]labels.Labels{labels1, labels3}, nil, nil, []cortexpb.Histogram{sampleNativeHistogram2, sampleNativeHistogram3}, cortexpb.API))
+				httpResp, ok := httpgrpc.HTTPResponseFromError(err)
+				require.True(t, ok, "returned error is not an httpgrpc response")
+				assert.Equal(t, http.StatusBadRequest, int(httpResp.Code))
+				assert.Equal(t, wrapWithUser(makeLimitError(perUserNativeHistogramsSeriesLimit, ing.limiter.FormatError(userID, errMaxNativeHistogramsSeriesPerUserLimitExceeded, labels1)), userID).Error(), string(httpResp.Body))
+
+				// Append two metadata, expect no error since metadata is a best effort approach.
+				_, err = ing.Push(ctx, cortexpb.ToWriteRequest(nil, nil, []*cortexpb.MetricMetadata{metadata1, metadata2}, nil, cortexpb.API))
+				require.NoError(t, err)
+
+				// Read samples back via ingester queries.
+				res, _, err := runTestQuery(ctx, t, ing, labels.MatchEqual, model.MetricNameLabel, "testmetric")
+				require.NoError(t, err)
+				require.NotNil(t, res)
+
+				// Verify metadata
+				m, err := ing.MetricsMetadata(ctx, &client.MetricsMetadataRequest{Limit: -1, LimitPerMetric: -1, Metric: ""})
+				require.NoError(t, err)
+				assert.Equal(t, []*cortexpb.MetricMetadata{metadata1}, m.Metadata)
+			}
+
+			testLimits(reg)
+
+			// Limits should hold after restart.
+			services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
+			// Use new registry to prevent metrics registration panic.
+			reg = prometheus.NewRegistry()
+			ing = ingGenerator(reg)
+			defer services.StopAndAwaitTerminated(context.Background(), ing) //nolint:errcheck
+
+			testLimits(reg)
+		})
+	}
+
+}
+
 func benchmarkData(nSeries int) (allLabels []labels.Labels, allSamples []cortexpb.Sample) {
 	for j := 0; j < nSeries; j++ {
 		labels := chunk.BenchmarkLabels.Copy()

pkg/ingester/limiter_test.go

@@ -656,10 +656,11 @@ func TestLimiter_FormatError(t *testing.T) {
 
 	// Mock limits
 	limits, err := validation.NewOverrides(validation.Limits{
-		MaxGlobalSeriesPerUser:              100,
-		MaxGlobalSeriesPerMetric:            20,
-		MaxGlobalMetricsWithMetadataPerUser: 10,
-		MaxGlobalMetadataPerMetric:          3,
+		MaxGlobalSeriesPerUser:                 100,
+		MaxGlobalNativeHistogramsSeriesPerUser: 100,
+		MaxGlobalSeriesPerMetric:               20,
+		MaxGlobalMetricsWithMetadataPerUser:    10,
+		MaxGlobalMetadataPerMetric:             3,
 	}, nil)
 	require.NoError(t, err)
 
@@ -669,6 +670,9 @@ func TestLimiter_FormatError(t *testing.T) {
 	actual := limiter.FormatError("user-1", errMaxSeriesPerUserLimitExceeded, lbls)
 	assert.EqualError(t, actual, "per-user series limit of 100 exceeded, please contact administrator to raise it (local limit: 0 global limit: 100 actual local limit: 100)")
 
+	actual = limiter.FormatError("user-1", errMaxNativeHistogramsSeriesPerUserLimitExceeded, lbls)
+	assert.EqualError(t, actual, "per-user nativeHistograms series limit of 100 exceeded, please contact administrator to raise it (local limit: 0 global limit: 100 actual local limit: 100)")
+
 	actual = limiter.FormatError("user-1", errMaxSeriesPerMetricLimitExceeded, lbls)
 	assert.EqualError(t, actual, "per-metric series limit of 20 exceeded for metric testMetric, please contact administrator to raise it (local limit: 0 global limit: 20 actual local limit: 20)")
 

pkg/ingester/user_state.go

@@ -16,9 +16,10 @@ import (
 
 // DiscardedSamples metric labels
 const (
-	perUserSeriesLimit     = "per_user_series_limit"
-	perMetricSeriesLimit   = "per_metric_series_limit"
-	perLabelsetSeriesLimit = "per_labelset_series_limit"
+	perUserSeriesLimit                 = "per_user_series_limit"
+	perUserNativeHistogramsSeriesLimit = "per_user_native_histograms_series_limit"
+	perMetricSeriesLimit               = "per_metric_series_limit"
+	perLabelsetSeriesLimit             = "per_labelset_series_limit"
 )
 
 const numMetricCounterShards = 128