Wrap ingester Push errors to avoid retaining any reference to unsafe data

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

Author: bboreham

pkg/ingester/errors.go

@@ -4,7 +4,6 @@ import (
 	"fmt"
 	"net/http"
 
-	"github.com/pkg/errors"
 	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/weaveworks/common/httpgrpc"
 )
@@ -50,11 +49,6 @@ func makeMetricLimitError(errorType string, labels labels.Labels, err error) err
 	}
 }
 
-func (e *validationError) WrapWithUser(userID string) *validationError {
-	e.err = wrapWithUser(e.err, userID)
-	return e
-}
-
 func (e *validationError) Error() string {
 	if e.err == nil {
 		return e.errorType
@@ -65,14 +59,15 @@ func (e *validationError) Error() string {
 	return fmt.Sprintf("%s for series %s", e.err.Error(), e.labels.String())
 }
 
-// WrappedError returns a HTTP gRPC error than is correctly forwarded over gRPC.
-func (e *validationError) WrappedError() error {
+// returns a HTTP gRPC error than is correctly forwarded over gRPC, with no reference to `e` retained.
+func grpcForwardableError(userID string, code int, e error) error {
 	return httpgrpc.ErrorFromHTTPResponse(&httpgrpc.HTTPResponse{
-		Code: int32(e.code),
-		Body: []byte(e.Error()),
+		Code: int32(code),
+		Body: []byte(wrapWithUser(e, userID).Error()),
 	})
 }
 
+// Note: does not retain a reference to `err`
 func wrapWithUser(err error, userID string) error {
-	return errors.Wrapf(err, "user=%s", userID)
+	return fmt.Errorf("user=%s: %s", userID, err)
 }

pkg/ingester/ingester.go

@@ -277,6 +277,8 @@ func (i *Ingester) Push(ctx old_ctx.Context, req *client.WriteRequest) (*client.
 		return i.v2Push(ctx, req)
 	}
 
+	// NOTE: because we use `unsafe` in deserialisation, we must not
+	// retain anything from `req` past the call to ReuseSlice
 	defer client.ReuseSlice(req.Timeseries)
 
 	userID, err := user.ExtractOrgID(ctx)
@@ -299,8 +301,6 @@ func (i *Ingester) Push(ctx old_ctx.Context, req *client.WriteRequest) (*client.
 		}
 	}
 
-	// NOTE: because we use `unsafe` in deserialisation, we must not
-	// retain anything from `req` past the call to ReuseSlice
 	for _, ts := range req.Timeseries {
 		for _, s := range ts.Samples {
 			// append() copies the memory in `ts.Labels` except on the error path
@@ -316,13 +316,13 @@ func (i *Ingester) Push(ctx old_ctx.Context, req *client.WriteRequest) (*client.
 			}
 
 			// non-validation error: abandon this request
-			return nil, wrapWithUser(err, userID)
+			return nil, grpcForwardableError(userID, http.StatusInternalServerError, err)
 		}
 	}
 
 	if lastPartialErr != nil {
-		// WrappedError turns the error into a string so it no longer references `req`
-		return &client.WriteResponse{}, lastPartialErr.WrapWithUser(userID).WrappedError()
+		// grpcForwardableError turns the error into a string so it no longer references `req`
+		return &client.WriteResponse{}, grpcForwardableError(userID, lastPartialErr.code, lastPartialErr)
 	}
 
 	if record != nil {

pkg/ingester/ingester_v2.go

@@ -105,6 +105,8 @@ func NewV2(cfg Config, clientConfig client.Config, limits *validation.Overrides,
 func (i *Ingester) v2Push(ctx old_ctx.Context, req *client.WriteRequest) (*client.WriteResponse, error) {
 	var firstPartialErr error
 
+	// NOTE: because we use `unsafe` in deserialisation, we must not
+	// retain anything from `req` past the call to ReuseSlice
 	defer client.ReuseSlice(req.Timeseries)
 
 	userID, err := user.ExtractOrgID(ctx)