distributor: Report correct size in "message too big" error

CHANGELOG.md

@@ -88,6 +88,7 @@
 * [BUGFIX] Ingester, Block-builder: silently ignore duplicate sample if it's due to zero sample from created timestamp. Created timestamp equal to the timestamp of the first sample of series is a common case if created timestamp comes from OTLP where start time equal to timestamp of the first sample simply means unknown start time. #12726
 * [BUGFIX] Distributor: Fix error when native histograms bucket limit is set then no NHCB passes validation. #12741
 * [BUGFIX] Ingester: Fix continous reload of active series counters when cost-attribution labels are above the max cardinality. #12822
+* [BUGFIX] Distributor: Report correct size in `err-mimir-distributor-max-write-message-size` error. #12799
 
 ### Mixin
 

pkg/distributor/otel.go

@@ -237,8 +237,9 @@ func newOTLPParser(
 				var tooLargeErr util.MsgSizeTooLargeErr
 				if errors.As(err, &tooLargeErr) {
 					return exportReq, 0, httpgrpc.Error(http.StatusRequestEntityTooLarge, distributorMaxOTLPRequestSizeErr{
-						actual: tooLargeErr.Actual,
-						limit:  tooLargeErr.Limit,
+						compressed: tooLargeErr.Compressed,
+						actual:     tooLargeErr.Actual,
+						limit:      tooLargeErr.Limit,
 					}.Error())
 				}
 				return exportReq, protoBodySize, err

pkg/distributor/otel_test.go

@@ -1164,9 +1164,9 @@ func TestHandlerOTLPPush(t *testing.T) {
 			},
 			responseCode:          http.StatusRequestEntityTooLarge,
 			responseContentType:   pbContentType,
-			responseContentLength: 292,
-			errMessage:            "the incoming OTLP request has been rejected because its message size of 89 bytes is larger",
-			expectedLogs:          []string{`level=warn user=test msg="detected an error while ingesting OTLP metrics request (the request may have been partially ingested)" httpCode=413 err="rpc error: code = Code(413) desc = the incoming OTLP request has been rejected because its message size of 89 bytes is larger than the allowed limit of 30 bytes (err-mimir-distributor-max-otlp-request-size). To adjust the related limit, configure -distributor.max-otlp-request-size, or contact your service administrator." insight=true`},
+			responseContentLength: 307,
+			errMessage:            "the incoming OTLP request has been rejected because its message size of 89 bytes (uncompressed) is larger",
+			expectedLogs:          []string{`level=warn user=test msg="detected an error while ingesting OTLP metrics request (the request may have been partially ingested)" httpCode=413 err="rpc error: code = Code(413) desc = the incoming OTLP request has been rejected because its message size of 89 bytes (uncompressed) is larger than the allowed limit of 30 bytes (err-mimir-distributor-max-otlp-request-size). To adjust the related limit, configure -distributor.max-otlp-request-size, or contact your service administrator." insight=true`},
 		},
 		{
 			name:        "Write samples. Unsupported compression",
@@ -1213,9 +1213,9 @@ func TestHandlerOTLPPush(t *testing.T) {
 			},
 			responseCode:          http.StatusRequestEntityTooLarge,
 			responseContentType:   pbContentType,
-			responseContentLength: 293,
-			errMessage:            "the incoming OTLP request has been rejected because its message size of 104 bytes is larger",
-			expectedLogs:          []string{`level=warn user=test msg="detected an error while ingesting OTLP metrics request (the request may have been partially ingested)" httpCode=413 err="rpc error: code = Code(413) desc = the incoming OTLP request has been rejected because its message size of 104 bytes is larger than the allowed limit of 30 bytes (err-mimir-distributor-max-otlp-request-size). To adjust the related limit, configure -distributor.max-otlp-request-size, or contact your service administrator." insight=true`},
+			responseContentLength: 308,
+			errMessage:            "the incoming OTLP request has been rejected because its message size of 104 bytes (uncompressed) is larger",
+			expectedLogs:          []string{`level=warn user=test msg="detected an error while ingesting OTLP metrics request (the request may have been partially ingested)" httpCode=413 err="rpc error: code = Code(413) desc = the incoming OTLP request has been rejected because its message size of 104 bytes (uncompressed) is larger than the allowed limit of 30 bytes (err-mimir-distributor-max-otlp-request-size). To adjust the related limit, configure -distributor.max-otlp-request-size, or contact your service administrator." insight=true`},
 		},
 		{
 			name:        "Write samples. With lz4 compression, request too big",
@@ -1229,9 +1229,9 @@ func TestHandlerOTLPPush(t *testing.T) {
 			},
 			responseCode:          http.StatusRequestEntityTooLarge,
 			responseContentType:   pbContentType,
-			responseContentLength: 293,
-			errMessage:            "the incoming OTLP request has been rejected because its message size of 106 bytes is larger",
-			expectedLogs:          []string{`level=warn user=test msg="detected an error while ingesting OTLP metrics request (the request may have been partially ingested)" httpCode=413 err="rpc error: code = Code(413) desc = the incoming OTLP request has been rejected because its message size of 106 bytes is larger than the allowed limit of 30 bytes (err-mimir-distributor-max-otlp-request-size). To adjust the related limit, configure -distributor.max-otlp-request-size, or contact your service administrator." insight=true`},
+			responseContentLength: 308,
+			errMessage:            "the incoming OTLP request has been rejected because its message size of 106 bytes (uncompressed) is larger",
+			expectedLogs:          []string{`level=warn user=test msg="detected an error while ingesting OTLP metrics request (the request may have been partially ingested)" httpCode=413 err="rpc error: code = Code(413) desc = the incoming OTLP request has been rejected because its message size of 106 bytes (uncompressed) is larger than the allowed limit of 30 bytes (err-mimir-distributor-max-otlp-request-size). To adjust the related limit, configure -distributor.max-otlp-request-size, or contact your service administrator." insight=true`},
 		},
 		{
 			name:       "Rate limited request",

pkg/distributor/push.go

@@ -98,8 +98,8 @@ func Handler(
 ) http.Handler {
 	return handler(maxRecvMsgSize, newRequestBuffers, sourceIPs, allowSkipLabelNameValidation, allowSkipLabelCountValidation, limits, retryCfg, push, logger, func(ctx context.Context, r *http.Request, maxRecvMsgSize int, buffers *util.RequestBuffers, req *mimirpb.PreallocWriteRequest, _ log.Logger) error {
 		protoBodySize, err := util.ParseProtoReader(ctx, r.Body, int(r.ContentLength), maxRecvMsgSize, buffers, req, util.RawSnappy)
-		if errors.Is(err, util.MsgSizeTooLargeErr{}) {
-			err = distributorMaxWriteMessageSizeErr{actual: int(r.ContentLength), limit: maxRecvMsgSize}
+		if e := (util.MsgSizeTooLargeErr{}); errors.As(err, &e) {
+			err = distributorMaxWriteMessageSizeErr{compressed: e.Compressed, actual: e.Actual, limit: e.Limit}
 		}
 		if err != nil {
 			return err
@@ -115,25 +115,29 @@ func Handler(
 }
 
 type distributorMaxWriteMessageSizeErr struct {
-	actual, limit int
+	compressed, actual, limit int
 }
 
 func (e distributorMaxWriteMessageSizeErr) Error() string {
-	msgSizeDesc := fmt.Sprintf(" of %d bytes", e.actual)
-	if e.actual < 0 {
-		msgSizeDesc = ""
+	msgSizeDesc := ""
+	if e.actual > 0 {
+		msgSizeDesc = fmt.Sprintf(" of %d bytes (uncompressed)", e.actual)
+	} else if e.compressed > 0 {
+		msgSizeDesc = fmt.Sprintf(" of %d bytes (compressed)", e.compressed)
 	}
 	return globalerror.DistributorMaxWriteMessageSize.MessageWithPerInstanceLimitConfig(fmt.Sprintf("the incoming push request has been rejected because its message size%s is larger than the allowed limit of %d bytes", msgSizeDesc, e.limit), "distributor.max-recv-msg-size")
 }
 
 type distributorMaxOTLPRequestSizeErr struct {
-	actual, limit int
+	compressed, actual, limit int
 }
 
 func (e distributorMaxOTLPRequestSizeErr) Error() string {
-	msgSizeDesc := fmt.Sprintf(" of %d bytes", e.actual)
-	if e.actual < 0 {
-		msgSizeDesc = ""
+	msgSizeDesc := ""
+	if e.actual > 0 {
+		msgSizeDesc = fmt.Sprintf(" of %d bytes (uncompressed)", e.actual)
+	} else if e.compressed > 0 {
+		msgSizeDesc = fmt.Sprintf(" of %d bytes (compressed)", e.compressed)
 	}
 	return globalerror.DistributorMaxOTLPRequestSize.MessageWithPerInstanceLimitConfig(fmt.Sprintf("the incoming OTLP request has been rejected because its message size%s is larger than the allowed limit of %d bytes", msgSizeDesc, e.limit), maxOTLPRequestSizeFlag)
 }

pkg/distributor/push_test.go

@@ -671,7 +671,7 @@ func (n bufCloser) BytesBuffer() *bytes.Buffer { return n.Buffer }
 
 func TestNewDistributorMaxWriteMessageSizeErr(t *testing.T) {
 	err := distributorMaxWriteMessageSizeErr{actual: 100, limit: 50}
-	msg := `the incoming push request has been rejected because its message size of 100 bytes is larger than the allowed limit of 50 bytes (err-mimir-distributor-max-write-message-size). To adjust the related limit, configure -distributor.max-recv-msg-size, or contact your service administrator.`
+	msg := `the incoming push request has been rejected because its message size of 100 bytes (uncompressed) is larger than the allowed limit of 50 bytes (err-mimir-distributor-max-write-message-size). To adjust the related limit, configure -distributor.max-recv-msg-size, or contact your service administrator.`
 
 	assert.Equal(t, msg, err.Error())
 }

pkg/util/http.go

@@ -179,11 +179,29 @@ func ParseProtoReader(ctx context.Context, reader io.Reader, expectedSize, maxSi
 }
 
 type MsgSizeTooLargeErr struct {
-	Actual, Limit int
+	Compressed, Actual, Limit int
+}
+
+func NewMsgCompressedSizeTooLargeErr(size, limit int) MsgSizeTooLargeErr {
+	return MsgSizeTooLargeErr{Compressed: size, Limit: limit}
+}
+
+func NewMsgUncompressedSizeTooLargeErr(size, limit int) MsgSizeTooLargeErr {
+	return MsgSizeTooLargeErr{Actual: size, Limit: limit}
+}
+
+func NewMsgUnknownSizeTooLargeErr(limit int) MsgSizeTooLargeErr {
+	return MsgSizeTooLargeErr{Limit: limit}
 }
 
 func (e MsgSizeTooLargeErr) Error() string {
-	return fmt.Sprintf("the request has been rejected because its size of %d bytes exceeds the limit of %d bytes", e.Actual, e.Limit)
+	msgSizeDesc := ""
+	if e.Actual > 0 {
+		msgSizeDesc = fmt.Sprintf(" of %d bytes (uncompressed)", e.Actual)
+	} else if e.Compressed > 0 {
+		msgSizeDesc = fmt.Sprintf(" of %d bytes (compressed)", e.Compressed)
+	}
+	return fmt.Sprintf("the request has been rejected because its size%s exceeds the limit of %d bytes", msgSizeDesc, e.Limit)
 }
 
 // Needed for errors.Is to work properly.
@@ -195,7 +213,10 @@ func (e MsgSizeTooLargeErr) Is(err error) bool {
 
 func decompressRequest(buffers *RequestBuffers, reader io.Reader, expectedSize, maxSize int, compression CompressionType, sp trace.Span) ([]byte, error) {
 	if expectedSize > maxSize {
-		return nil, MsgSizeTooLargeErr{Actual: expectedSize, Limit: maxSize}
+		if compression == NoCompression {
+			return nil, NewMsgUncompressedSizeTooLargeErr(expectedSize, maxSize)
+		}
+		return nil, NewMsgCompressedSizeTooLargeErr(expectedSize, maxSize)
 	}
 
 	switch compression {
@@ -265,7 +286,7 @@ func decompressRequest(buffers *RequestBuffers, reader io.Reader, expectedSize,
 	}
 
 	if buf.Len() > maxSize {
-		return nil, MsgSizeTooLargeErr{Actual: -1, Limit: maxSize}
+		return nil, NewMsgUnknownSizeTooLargeErr(maxSize)
 	}
 	return buf.Bytes(), nil
 }

pkg/util/http_test.go

@@ -246,8 +246,18 @@ func TestIsRequestBodyTooLargeRegression(t *testing.T) {
 }
 
 func TestNewMsgSizeTooLargeErr(t *testing.T) {
-	err := MsgSizeTooLargeErr{Actual: 100, Limit: 50}
-	msg := `the request has been rejected because its size of 100 bytes exceeds the limit of 50 bytes`
+	err := NewMsgUncompressedSizeTooLargeErr(100, 50)
+	msg := `the request has been rejected because its size of 100 bytes (uncompressed) exceeds the limit of 50 bytes`
+
+	assert.Equal(t, msg, err.Error())
+
+	err = NewMsgCompressedSizeTooLargeErr(100, 50)
+	msg = `the request has been rejected because its size of 100 bytes (compressed) exceeds the limit of 50 bytes`
+
+	assert.Equal(t, msg, err.Error())
+
+	err = NewMsgUnknownSizeTooLargeErr(50)
+	msg = `the request has been rejected because its size exceeds the limit of 50 bytes`
 
 	assert.Equal(t, msg, err.Error())
 }