Use validationError instead of httpgrpc error to reduce memory allocation

It's more efficient to defer constructing the httpgrpc error until the
end, because we only report the last error out of possibly hundreds.

Signed-off-by: Bryan Boreham <[email protected]>

Author: bboreham

pkg/ingester/ingester.go

@@ -293,7 +293,7 @@ func (i *Ingester) Push(ctx old_ctx.Context, req *client.WriteRequest) (*client.
 	if err != nil {
 		return nil, fmt.Errorf("no user id")
 	}
-	var lastPartialErr error
+	var lastPartialErr *validationError
 
 	for _, ts := range req.Timeseries {
 		for _, s := range ts.Samples {
@@ -303,20 +303,20 @@ func (i *Ingester) Push(ctx old_ctx.Context, req *client.WriteRequest) (*client.
 			}
 
 			i.metrics.ingestedSamplesFail.Inc()
-			if httpResp, ok := httpgrpc.HTTPResponseFromError(err); ok {
-				switch httpResp.Code {
-				case http.StatusBadRequest, http.StatusTooManyRequests:
-					lastPartialErr = err
-					continue
-				}
+			if ve, ok := err.(*validationError); ok {
+				lastPartialErr = ve
+				continue
 			}
 
 			return nil, err
 		}
 	}
 	client.ReuseSlice(req.Timeseries)
 
-	return &client.WriteResponse{}, lastPartialErr
+	if lastPartialErr != nil {
+		return &client.WriteResponse{}, lastPartialErr.WrappedError()
+	}
+	return &client.WriteResponse{}, nil
 }
 
 func (i *Ingester) append(ctx context.Context, userID string, labels labelPairs, timestamp model.Time, value model.SampleValue, source client.WriteRequest_SourceEnum) error {
@@ -338,6 +338,9 @@ func (i *Ingester) append(ctx context.Context, userID string, labels labelPairs,
 	}
 	state, fp, series, err := i.userStates.getOrCreateSeries(ctx, userID, labels)
 	if err != nil {
+		if ve, ok := err.(*validationError); ok {
+			state.discardedSamples.WithLabelValues(ve.errorType).Inc()
+		}
 		state = nil // don't want to unlock the fp if there is an error
 		return err
 	}
@@ -357,13 +360,11 @@ func (i *Ingester) append(ctx context.Context, userID string, labels labelPairs,
 		Value:     value,
 		Timestamp: timestamp,
 	}); err != nil {
-		if mse, ok := err.(*memorySeriesError); ok {
-			state.discardedSamples.WithLabelValues(mse.errorType).Inc()
-			if mse.noReport {
+		if ve, ok := err.(*validationError); ok {
+			state.discardedSamples.WithLabelValues(ve.errorType).Inc()
+			if ve.noReport {
 				return nil
 			}
-			// Use a dumb string template to avoid the message being parsed as a template
-			err = httpgrpc.Errorf(http.StatusBadRequest, "%s", mse.message)
 		}
 		return err
 	}

pkg/ingester/ingester_test.go

@@ -317,16 +317,16 @@ func TestIngesterAppendOutOfOrderAndDuplicate(t *testing.T) {
 	// Earlier sample than previous one.
 	err = ing.append(ctx, userID, m, 0, 0, client.API)
 	require.Contains(t, err.Error(), "sample timestamp out of order")
-	errResp, ok := httpgrpc.HTTPResponseFromError(err)
+	errResp, ok := err.(*validationError)
 	require.True(t, ok)
-	require.Equal(t, errResp.Code, int32(400))
+	require.Equal(t, errResp.code, 400)
 
 	// Same timestamp as previous sample, but different value.
 	err = ing.append(ctx, userID, m, 1, 1, client.API)
 	require.Contains(t, err.Error(), "sample with repeated timestamp but different value")
-	errResp, ok = httpgrpc.HTTPResponseFromError(err)
+	errResp, ok = err.(*validationError)
 	require.True(t, ok)
-	require.Equal(t, errResp.Code, int32(400))
+	require.Equal(t, errResp.code, 400)
 }
 
 // Test that blank labels are removed by the ingester

pkg/ingester/ingester_v2.go

@@ -111,7 +111,7 @@ func (i *Ingester) v2Push(ctx old_ctx.Context, req *client.WriteRequest) (*clien
 			// 400 error to the client. The client (Prometheus) will not retry on 400, and
 			// we actually ingested all samples which haven't failed.
 			if err == tsdb.ErrOutOfBounds || err == tsdb.ErrOutOfOrderSample || err == tsdb.ErrAmendSample {
-				lastPartialErr = httpgrpc.Errorf(http.StatusBadRequest, err.Error())
+				lastPartialErr = err
 				continue
 			}
 
@@ -135,7 +135,10 @@ func (i *Ingester) v2Push(ctx old_ctx.Context, req *client.WriteRequest) (*clien
 
 	client.ReuseSlice(req.Timeseries)
 
-	return &client.WriteResponse{}, lastPartialErr
+	if lastPartialErr != nil {
+		return &client.WriteResponse{}, httpgrpc.Errorf(http.StatusBadRequest, lastPartialErr.Error())
+	}
+	return &client.WriteResponse{}, nil
 }
 
 func (i *Ingester) v2Query(ctx old_ctx.Context, req *client.QueryRequest) (*client.QueryResponse, error) {

pkg/ingester/user_state.go

@@ -195,8 +195,7 @@ func (u *userState) getSeries(metric labelPairs) (model.Fingerprint, *memorySeri
 	err := u.limiter.AssertMaxSeriesPerUser(u.userID, u.fpToSeries.length())
 	if err != nil {
 		u.fpLocker.Unlock(fp)
-		u.discardedSamples.WithLabelValues(perUserSeriesLimit).Inc()
-		return fp, nil, httpgrpc.Errorf(http.StatusTooManyRequests, err.Error())
+		return fp, nil, makeLimitError(perUserSeriesLimit, err)
 	}
 
 	metricName, err := extract.MetricNameFromLabelAdapters(metric)
@@ -209,8 +208,7 @@ func (u *userState) getSeries(metric labelPairs) (model.Fingerprint, *memorySeri
 	err = u.canAddSeriesFor(string(metricName))
 	if err != nil {
 		u.fpLocker.Unlock(fp)
-		u.discardedSamples.WithLabelValues(perMetricSeriesLimit).Inc()
-		return fp, nil, httpgrpc.Errorf(http.StatusTooManyRequests, "%s for: %s", err.Error(), metric)
+		return fp, nil, makeMetricLimitError(perMetricSeriesLimit, client.FromLabelAdaptersToLabels(metric), err)
 	}
 
 	u.memSeriesCreatedTotal.Inc()